iframe-proxy | Sunbelt Computer Software

178 lines (150 loc) · 3.38 KB

 * Copyright 2018 The CovenantSQL Authors.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *     http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
package kayak
	"sync/atomic"
	kt "github.com/CovenantSQL/CovenantSQL/kayak/types"
	"github.com/CovenantSQL/CovenantSQL/proto"
	rpc "github.com/CovenantSQL/CovenantSQL/rpc/mux"
	"github.com/CovenantSQL/CovenantSQL/utils/trace"
// rpcTracker defines the rpc call tracker
// support tracking the rpc result.
type rpcTracker struct {
	// related runtime
	r *Runtime
	// target nodes, a copy of current followers
	nodes []proto.NodeID
	// rpc method
	method string
	// rpc request
	req interface{}
	// minimum response count
	minCount int
	// responses
	errLock sync.RWMutex
	errors  map[proto.NodeID]error
	// scoreboard
	complete int
	sent     uint32
	doneOnce sync.Once
	doneCh   chan struct{}
	wg       sync.WaitGroup
	closed   uint32
func newTracker(r *Runtime, req interface{}, minCount int) (t *rpcTracker) {
	// copy nodes
	nodes := append([]proto.NodeID(nil), r.followers...)
	if minCount > len(nodes) {
		minCount = len(nodes)
	if minCount < 0 {
		minCount = 0
	t = &rpcTracker{
		r:        r,
		nodes:    nodes,
		method:   r.applyRPCMethod,
		req:      req,
		minCount: minCount,
		errors:   make(map[proto.NodeID]error, len(nodes)),
		doneCh:   make(chan struct{}),
func (t *rpcTracker) send() {
	if !atomic.CompareAndSwapUint32(&t.sent, 0, 1) {
	for i := range t.nodes {
		t.wg.Add(1)
		go t.callSingle(i)
	if t.minCount == 0 {
func (t *rpcTracker) callSingle(idx int) {
	caller := t.r.TrackerNewCallerFunc(t.nodes[idx])
	if pcaller, ok := caller.(*rpc.PersistentCaller); ok && pcaller != nil {
		defer pcaller.Close()
	err := caller.Call(t.method, t.req, nil)
	defer t.wg.Done()
	t.errLock.Lock()
	defer t.errLock.Unlock()
	t.errors[t.nodes[idx]] = err
	t.complete++
	if t.complete >= t.minCount {
func (t *rpcTracker) done() {
	t.doneOnce.Do(func() {
		if t.doneCh != nil {
			select {
			case <-t.doneCh:
			default:
				close(t.doneCh)
func (t *rpcTracker) get(ctx context.Context) (errors map[proto.NodeID]error, meets bool, finished bool) {
	if trace.IsEnabled() {
		// get request log type
		traceType := "rpcCall"
		if rawReq, ok := t.req.(*kt.ApplyRequest); ok {
			traceType += rawReq.Log.Type.String()
		defer trace.StartRegion(ctx, traceType).End()
		case <-t.doneCh:
			meets = true
		case <-ctx.Done():
		case <-t.doneCh:
			meets = true
	t.errLock.RLock()
	defer t.errLock.RUnlock()
	errors = make(map[proto.NodeID]error)
	for s, e := range t.errors {
		errors[s] = e
	if !meets && len(errors) >= t.minCount {
		meets = true
	if len(errors) == len(t.nodes) {
		finished = true
func (t *rpcTracker) close() {
	if !atomic.CompareAndSwapUint32(&t.closed, 0, 1) {
	t.wg.Wait()

Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

tracker.go

Latest commit

History

tracker.go

File metadata and controls