feat(ludicrous): Run mutations from the same predicate concurrently in ludicrous mode

dgraph/cmd/alpha/run.go

@@ -201,6 +201,7 @@ they form a Raft group and provide synchronous replication.
 
 	flag.Bool("graphql_introspection", true, "Set to false for no GraphQL schema introspection")
 	flag.Bool("ludicrous_mode", false, "Run alpha in ludicrous mode")
+	flag.Int("ludicrous_concurrency", 2000, "Number of concurrent threads in ludicrous mode")
 	flag.Bool("graphql_extensions", true, "Set to false if extensions not required in GraphQL response body")
 	flag.Duration("graphql_poll_interval", time.Second, "polling interval for graphql subscription.")
 
@@ -674,20 +675,21 @@ func run() {
 	x.Check(err)
 
 	x.WorkerConfig = x.WorkerOptions{
-		ExportPath:          Alpha.Conf.GetString("export"),
-		NumPendingProposals: Alpha.Conf.GetInt("pending_proposals"),
-		Tracing:             Alpha.Conf.GetFloat64("trace"),
-		MyAddr:              Alpha.Conf.GetString("my"),
-		ZeroAddr:            strings.Split(Alpha.Conf.GetString("zero"), ","),
-		RaftId:              cast.ToUint64(Alpha.Conf.GetString("idx")),
-		WhiteListedIPRanges: ips,
-		MaxRetries:          Alpha.Conf.GetInt("max_retries"),
-		StrictMutations:     opts.MutationsMode == worker.StrictMutations,
-		AclEnabled:          secretFile != "",
-		SnapshotAfter:       Alpha.Conf.GetInt("snapshot_after"),
-		AbortOlderThan:      abortDur,
-		StartTime:           startTime,
-		LudicrousMode:       Alpha.Conf.GetBool("ludicrous_mode"),
+		ExportPath:           Alpha.Conf.GetString("export"),
+		NumPendingProposals:  Alpha.Conf.GetInt("pending_proposals"),
+		Tracing:              Alpha.Conf.GetFloat64("trace"),
+		MyAddr:               Alpha.Conf.GetString("my"),
+		ZeroAddr:             strings.Split(Alpha.Conf.GetString("zero"), ","),
+		RaftId:               cast.ToUint64(Alpha.Conf.GetString("idx")),
+		WhiteListedIPRanges:  ips,
+		MaxRetries:           Alpha.Conf.GetInt("max_retries"),
+		StrictMutations:      opts.MutationsMode == worker.StrictMutations,
+		AclEnabled:           secretFile != "",
+		SnapshotAfter:        Alpha.Conf.GetInt("snapshot_after"),
+		AbortOlderThan:       abortDur,
+		StartTime:            startTime,
+		LudicrousMode:        Alpha.Conf.GetBool("ludicrous_mode"),
+		LudicrousConcurrency: Alpha.Conf.GetInt("ludicrous_concurrency"),
 	}
 	if x.WorkerConfig.EncryptionKey, err = enc.ReadKey(Alpha.Conf); err != nil {
 		glog.Infof("unable to read key %v", err)

posting/list.go

@@ -434,50 +434,14 @@ func (l *List) addMutation(ctx context.Context, txn *Txn, t *pb.DirectedEdge) er
 	return l.addMutationInternal(ctx, txn, t)
 }
 
-func (l *List) addMutationInternal(ctx context.Context, txn *Txn, t *pb.DirectedEdge) error {
-	l.AssertLock()
-
-	if txn.ShouldAbort() {
-		return zero.ErrConflict
-	}
-
+func GetConflictKey(pk x.ParsedKey, key []byte, t *pb.DirectedEdge) uint64 {
 	getKey := func(key []byte, uid uint64) uint64 {
 		// Instead of creating a string first and then doing a fingerprint, let's do a fingerprint
 		// here to save memory allocations.
 		// Not entirely sure about effect on collision chances due to this simple XOR with uid.
 		return farm.Fingerprint64(key) ^ uid
 	}
 
-	mpost := NewPosting(t)
-	mpost.StartTs = txn.StartTs
-	if mpost.PostingType != pb.Posting_REF {
-		t.ValueId = fingerprintEdge(t)
-		mpost.Uid = t.ValueId
-	}
-
-	// Check whether this mutation is an update for a predicate of type uid.
-	pk, err := x.Parse(l.key)
-	if err != nil {
-		return errors.Wrapf(err, "cannot parse key when adding mutation to list with key %s",
-			hex.EncodeToString(l.key))
-	}
-	pred, ok := schema.State().Get(ctx, t.Attr)
-	isSingleUidUpdate := ok && !pred.GetList() && pred.GetValueType() == pb.Posting_UID &&
-		pk.IsData() && mpost.Op == Set && mpost.PostingType == pb.Posting_REF
-
-	if err != l.updateMutationLayer(mpost, isSingleUidUpdate) {
-		return errors.Wrapf(err, "cannot update mutation layer of key %s with value %+v",
-			hex.EncodeToString(l.key), mpost)
-	}
-
-	if x.WorkerConfig.LudicrousMode {
-		// Conflict detection is not required for ludicrous mode.
-		return nil
-	}
-
-	// We ensure that commit marks are applied to posting lists in the right
-	// order. We can do so by proposing them in the same order as received by the Oracle delta
-	// stream from Zero, instead of in goroutines.
 	var conflictKey uint64
 	switch {
 	case schema.State().HasNoConflict(t.Attr):
@@ -490,7 +454,7 @@ func (l *List) addMutationInternal(ctx context.Context, txn *Txn, t *pb.Directed
 		// The first key won't conflict, because two different UIDs can try to
 		// get the same email id. But, the second key would. Thus, we ensure
 		// that two users don't set the same email id.
-		conflictKey = getKey(l.key, 0)
+		conflictKey = getKey(key, 0)
 
 	case pk.IsData() && schema.State().IsList(t.Attr):
 		// Data keys, irrespective of whether they are UID or values, should be judged based on
@@ -508,19 +472,60 @@ func (l *List) addMutationInternal(ctx context.Context, txn *Txn, t *pb.Directed
 		// a -> "y"
 		// This should definitely have a conflict.
 		// But, if name: [string], then they can both succeed.
-		conflictKey = getKey(l.key, t.ValueId)
+		conflictKey = getKey(key, t.ValueId)
 
 	case pk.IsData(): // NOT a list. This case must happen after the above case.
-		conflictKey = getKey(l.key, 0)
+		conflictKey = getKey(key, 0)
 
 	case pk.IsIndex() || pk.IsCountOrCountRev():
 		// Index keys are by default of type [uid].
-		conflictKey = getKey(l.key, t.ValueId)
+		conflictKey = getKey(key, t.ValueId)
 
 	default:
 		// Don't assign a conflictKey.
 	}
-	txn.addConflictKey(conflictKey)
+
+	return conflictKey
+}
+
+func (l *List) addMutationInternal(ctx context.Context, txn *Txn, t *pb.DirectedEdge) error {
+	l.AssertLock()
+
+	if txn.ShouldAbort() {
+		return zero.ErrConflict
+	}
+
+	mpost := NewPosting(t)
+	mpost.StartTs = txn.StartTs
+	if mpost.PostingType != pb.Posting_REF {
+		t.ValueId = fingerprintEdge(t)
+		mpost.Uid = t.ValueId
+	}
+
+	// Check whether this mutation is an update for a predicate of type uid.
+	pk, err := x.Parse(l.key)
+	if err != nil {
+		return errors.Wrapf(err, "cannot parse key when adding mutation to list with key %s",
+			hex.EncodeToString(l.key))
+	}
+	pred, ok := schema.State().Get(ctx, t.Attr)
+	isSingleUidUpdate := ok && !pred.GetList() && pred.GetValueType() == pb.Posting_UID &&
+		pk.IsData() && mpost.Op == Set && mpost.PostingType == pb.Posting_REF
+
+	if err != l.updateMutationLayer(mpost, isSingleUidUpdate) {
+		return errors.Wrapf(err, "cannot update mutation layer of key %s with value %+v",
+			hex.EncodeToString(l.key), mpost)
+	}
+
+	if x.WorkerConfig.LudicrousMode {
+		// Conflict detection is not required for ludicrous mode.
+		return nil
+	}
+
+	// We ensure that commit marks are applied to posting lists in the right
+	// order. We can do so by proposing them in the same order as received by the Oracle delta
+	// stream from Zero, instead of in goroutines.
+	txn.addConflictKey(GetConflictKey(pk, l.key, t))
 	return nil
 }
 

worker/draft.go

@@ -243,7 +243,7 @@ func newNode(store *raftwal.DiskStorage, gid uint32, id uint64, myAddr string) *
 		ops:     make(map[op]*y.Closer),
 	}
 	if x.WorkerConfig.LudicrousMode {
-		n.ex = newExecutor(&m.Applied)
+		n.ex = newExecutor(&m.Applied, x.WorkerConfig.LudicrousConcurrency)
 	}
 	return n
 }