Method sendDeltasToSubscribers

dgraph/cmd/zero/oracle.go:185–248

sendDeltasToSubscribers reads updates from the o.updates constructs a delta object containing transactions from one or more updates and sends the delta object to each subscriber's channel

()

// constructs a delta object containing transactions from one or more updates
// and sends the delta object to each subscriber's channel
func (o *Oracle) sendDeltasToSubscribers() {
	delta := &pb.OracleDelta{}
	ticker := time.Tick(time.Second)

	// waitFor calculates the maximum value of delta.MaxAssigned and all the CommitTs of delta.Txns
	waitFor := func() uint64 {
		w := delta.MaxAssigned
		for _, txn := range delta.Txns {
			w = x.Max(w, txn.CommitTs)
		}
		return w
	}

	for {
	get_update:
		var update *pb.OracleDelta
		select {
		case update = <-o.updates:
		case <-ticker:
			wait := waitFor()
			if wait == 0 || o.doneUntil.DoneUntil() < wait {
				goto get_update
			}
			// Send empty update.
			update = &pb.OracleDelta{}
		}
	slurp_loop:
		for {
			delta.MaxAssigned = x.Max(delta.MaxAssigned, update.MaxAssigned)
			delta.Txns = append(delta.Txns, update.Txns...)
			select {
			case update = <-o.updates:
			default:
				break slurp_loop
			}
		}
		// No need to sort the txn updates here. Alpha would sort them before
		// applying.

		// Let's ensure that we have all the commits up until the max here.
		// Otherwise, we'll be sending commit timestamps out of order, which
		// would cause Alphas to drop some of them, during writes to Badger.
		if o.doneUntil.DoneUntil() < waitFor() {
			continue // The for loop doing blocking reads from o.updates.
			// We need at least one entry from the updates channel to pick up a missing update.
			// Don't goto slurp_loop, because it would break from select immediately.
		}

		if glog.V(3) {
			glog.Infof("DoneUntil: %d. Sending delta: %+v\n", o.doneUntil.DoneUntil(), delta)
		}
		o.Lock()
		for id, ch := range o.subscribers {
			select {
			case ch <- *delta:
			default:
				close(ch)
				delete(o.subscribers, id)