perf(zero): update zero's delta proposal frequency

Author: darkcoderrises

dgraph/cmd/zero/oracle.go

@@ -225,19 +225,47 @@ func (o *Oracle) sendDeltasToSubscribers() {
 		// 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.
+
+		newDelta := &pb.OracleDelta{}
+		useNewDelta := false
 		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 len(delta.Txns) > 5 {
+				replacementTxn := []*pb.TxnStatus{}
+
+				ts := o.doneUntil.DoneUntil()
+				for _, txn := range delta.Txns {
+					if txn.CommitTs > ts {
+						replacementTxn = append(replacementTxn, txn)
+					} else {
+						newDelta.Txns = append(newDelta.Txns, txn)
+						newDelta.MaxAssigned = x.Max(newDelta.MaxAssigned, txn.CommitTs)
+					}
+				}
+
+				if len(newDelta.Txns) == 0 {
+					continue
+				}
+
+				useNewDelta = true
+				delta.Txns = replacementTxn
+			} else {
+				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()
+		k := *delta
+		if useNewDelta {
+			k = *newDelta
+		}
 		for id, ch := range o.subscribers {
 			select {
-			case ch <- *delta:
+			case ch <- k:
 			default:
 				close(ch)
 				delete(o.subscribers, id)

worker/draft.go

@@ -68,6 +68,8 @@ type node struct {
 	opsLock     sync.Mutex
 	cdcTracker  *CDC
 	canCampaign bool
+
+	startTsKey map[uint64]uint64
 }
 
 type op int
@@ -876,6 +878,12 @@ func (n *node) processApplyCh() {
 		glog.V(3).Infof("handling element in applyCh with #entries %v", len(entries))
 		defer glog.V(3).Infof("done handling element in applyCh")
 
+		_, spanHandler := otel.Tracer("applyCh").Start(context.Background(), "Alpha.processApplyCh")
+		defer spanHandler.End()
+
+		spanHandler.AddEvent("handling element in applyCh with #entries %v", trace.WithAttributes(
+			attribute.Int64("numEntries", int64(len(entries)))))
+
 		var totalSize int64
 		for _, entry := range entries {
 			x.AssertTrue(len(entry.Data) > 0)
@@ -909,7 +917,7 @@ func (n *node) processApplyCh() {
 					p := &P{err: perr, size: psz, seen: time.Now()}
 					previous[key] = p
 				}
-				span := trace.SpanFromContext(n.ctx)
+				span := trace.SpanFromContext(n.Ctx(key))
 				if perr != nil {
 					glog.Errorf("Applying proposal. Error: %v. Proposal: %q.", perr, &proposal)
 					span.AddEvent(fmt.Sprintf("Applying proposal failed. Error: %v Proposal: %q", perr, &proposal))
@@ -919,11 +927,18 @@ func (n *node) processApplyCh() {
 						attribute.Int64("key", int64(key)),
 						attribute.Int64("index", int64(proposal.Index)),
 					))
+				spanHandler.AddEvent("Applied proposal with key: %d, index: %d. Err: %v",
+					trace.WithAttributes(
+						attribute.Int64("key", int64(key)),
+						attribute.Int64("index", int64(proposal.Index)),
+					))
 
 				var tags []tag.Mutator
 				switch {
 				case proposal.Mutations != nil:
 					tags = append(tags, tag.Upsert(x.KeyMethod, "apply.Mutations"))
+					span.SetAttributes(attribute.Int64("start_ts", int64(proposal.Mutations.StartTs)))
+					n.startTsKey[proposal.Mutations.StartTs] = key
 				case proposal.Delta != nil:
 					tags = append(tags, tag.Upsert(x.KeyMethod, "apply.Delta"))
 				}
@@ -966,6 +981,9 @@ func (n *node) processApplyCh() {
 
 // TODO(Anurag - 4 May 2020): Are we using pkey? Remove if unused.
 func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
+	_, span := otel.Tracer("alpha.CommitLoop").Start(context.Background(), "commitOrAbort")
+	defer span.End()
+
 	x.PrintOracleDelta(delta)
 	// First let's commit all mutations to disk.
 	writer := posting.NewTxnWriter(pstore)
@@ -995,6 +1013,18 @@ func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
 				start, commit, err)
 			panic(err)
 		}
+
+		span.AddEvent("Committed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
+			attribute.Int64("start_ts", int64(start)),
+			attribute.Int64("commit_ts", int64(commit)),
+		))
+
+		spani := trace.SpanFromContext(n.Ctx(n.startTsKey[start]))
+		fmt.Println("FIRST", spani)
+		spani.AddEvent("Committed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
+			attribute.Int64("start_ts", int64(start)),
+			attribute.Int64("commit_ts", int64(commit)),
+		))
 	}
 
 	for _, status := range delta.Txns {
@@ -1004,6 +1034,16 @@ func (n *node) commitOrAbort(pkey uint64, delta *pb.OracleDelta) error {
 		return errors.Wrapf(err, "while flushing to disk")
 	}
 
+	span.AddEvent("Flushed to disk")
+	for _, status := range delta.Txns {
+		spani := trace.SpanFromContext(n.Ctx(n.startTsKey[status.StartTs]))
+		fmt.Println("Second", spani)
+		spani.AddEvent("Flushed txn with start_ts: %d, commit_ts: %d", trace.WithAttributes(
+			attribute.Int64("start_ts", int64(status.StartTs)),
+			attribute.Int64("commit_ts", int64(status.CommitTs)),
+		))
+	}
+
 	if x.WorkerConfig.HardSync {
 		if err := pstore.Sync(); err != nil {
 			glog.Errorf("Error while calling Sync while commitOrAbort: %v", err)