Merge pull request #3 from gosuda/perf/conflict-engine-integration

perf(epaxos): integrate conflict engine; remove O(n) index

Author: metaphorics

epaxos/bootstrap.go

@@ -2098,7 +2098,6 @@ func (n *RawNode) proposeMembershipAt(ref InstanceRef, command Command) error {
 		return ErrInvalidConfig
 	}
 	n.installInstance(inst)
-	n.indexConflicts(record)
 	n.enqueueRecord(record)
 	if n.slowQuorumForConf(ref.Conf) == 1 {
 		n.commit(inst, record.Attributes())

epaxos/conflict_engine.go

@@ -278,6 +278,58 @@ func walkRadixDesc(node *radixNode, prefix, from InstanceNum, limited bool, yiel
 	return true
 }
 
+// walkGlobalRadixDesc descends only subtrees that contain a global-eligible instance
+// (node.globalMax != 0 and within the from prefix), so unrelated residents are not visited.
+func walkGlobalRadixDesc(node *radixNode, prefix, from InstanceNum, limited bool, yield func(InstanceNum, laneSlot) bool) bool {
+	if node == nil || node.globalMax == 0 {
+		return true
+	}
+	if node.level == 0 {
+		last := uint(radixFanout - 1)
+		if limited {
+			last = uint(from) & (radixFanout - 1)
+		}
+		for idx := int(last); idx >= 0; idx-- {
+			if node.leaf.present&(uint64(1)<<uint(idx)) == 0 {
+				continue
+			}
+			instance := prefix | InstanceNum(idx)
+			if limited && instance > from {
+				continue
+			}
+			slot := node.leaf.slots[idx]
+			if !slot.global() || !slot.eligible() {
+				continue
+			}
+			if !yield(instance, slot) {
+				return false
+			}
+		}
+		return true
+	}
+	shift := radixBits * node.level
+	last := uint(radixFanout - 1)
+	if limited {
+		last = uint(from>>shift) & (radixFanout - 1)
+	}
+	for idx := int(last); idx >= 0; idx-- {
+		child := node.children[idx]
+		if child == nil || child.globalMax == 0 {
+			continue
+		}
+		childPrefix := prefix | InstanceNum(idx)<<shift
+		// Prune children whose entire key range is > from when limited.
+		if limited && childPrefix > from {
+			continue
+		}
+		childLimited := limited && uint(idx) == last
+		if !walkGlobalRadixDesc(child, childPrefix, from, childLimited, yield) {
+			return false
+		}
+	}
+	return true
+}
+
 func (t *laneTree) slot(instance InstanceNum) (laneSlot, bool) {
 	node := t.root
 	if node == nil || radixLevel(instance) > node.level {
@@ -641,6 +693,50 @@ func (e *conflictEngine) keyMax(conf ConfID, key []byte, lane instanceLane) (res
 	return entry.postings.max(), entry.retiredFloor
 }
 
+// walkKeyDesc yields resident instances for one (conf,key,lane) in descending Instance order,
+// only visiting that key's posting tree (not unrelated lane residents).
+func (e *conflictEngine) walkKeyDesc(conf ConfID, key []byte, lane instanceLane, from InstanceNum, yield func(InstanceNum, laneSlot) bool) {
+	keys := e.byKey[conf]
+	if keys == nil {
+		return
+	}
+	lanes := keys[string(key)]
+	if lanes == nil {
+		return
+	}
+	entry := (*lanes)[lane]
+	if entry == nil || entry.postings.root == nil {
+		return
+	}
+	index := e.laneIndex[lane]
+	if index == nil {
+		return
+	}
+	if from == 0 {
+		from = entry.postings.max()
+	}
+	if from == 0 {
+		return
+	}
+	walkPostingDesc(entry.postings.root, 0, from, true, func(instance InstanceNum) bool {
+		slot, ok := index.resident.slot(instance)
+		if !ok {
+			return true
+		}
+		return yield(instance, slot)
+	})
+}
+
+// walkGlobalDesc yields global-scope eligible residents for a lane descending from from.
+// Descent prunes radix subtrees with globalMax==0 so unrelated residents are not visited.
+func (e *conflictEngine) walkGlobalDesc(lane instanceLane, from InstanceNum, yield func(InstanceNum, laneSlot) bool) {
+	index := e.laneIndex[lane]
+	if index == nil || index.resident.root == nil || from == 0 {
+		return
+	}
+	walkGlobalRadixDesc(index.resident.root, 0, from, radixLevel(from) <= index.resident.root.level, yield)
+}
+
 func (e *conflictEngine) keyLaneSet(conf ConfID, keys [][]byte, yield func(instanceLane) bool) {
 	for lane := range e.laneIndex {
 		if lane.conf != conf {
@@ -671,28 +767,26 @@ func (e *conflictEngine) lanes(conf ConfID, yield func(instanceLane) bool) {
 	}
 }
 
-func (e *conflictEngine) maxEligibleAny(lane instanceLane) InstanceNum {
+func (e *conflictEngine) maxEligibleAny(lane instanceLane) (resident, retired InstanceNum) {
 	index := e.laneIndex[lane]
 	if index == nil {
-		return 0
+		return 0, 0
 	}
-	resident := InstanceNum(0)
 	if index.resident.root != nil {
 		resident = index.resident.root.maxEligibleAny
 	}
-	return max(resident, index.retiredEligibleAny)
+	return resident, index.retiredEligibleAny
 }
 
-func (e *conflictEngine) globalMax(lane instanceLane) InstanceNum {
+func (e *conflictEngine) globalMax(lane instanceLane) (resident, retired InstanceNum) {
 	index := e.laneIndex[lane]
 	if index == nil {
-		return 0
+		return 0, 0
 	}
-	resident := InstanceNum(0)
 	if index.resident.root != nil {
 		resident = index.resident.root.globalMax
 	}
-	return max(resident, index.retiredGlobal)
+	return resident, index.retiredGlobal
 }
 
 func (e *conflictEngine) foldRecord(rec InstanceRecord) {

epaxos/conflict_engine_test.go

@@ -191,11 +191,11 @@ func assertConflictEngineMatchesModel(t *testing.T, engine *conflictEngine, mode
 				wantGlobal = max(wantGlobal, ref.Instance)
 			}
 		}
-		if got := engine.maxEligibleAny(lane); got != wantMax {
-			t.Fatalf("step %d lane %v: max eligible=%d, want %d", step, lane, got, wantMax)
+		if resident, retired := engine.maxEligibleAny(lane); max(resident, retired) != wantMax {
+			t.Fatalf("step %d lane %v: max eligible=%d/%d, want %d", step, lane, resident, retired, wantMax)
 		}
-		if got := engine.globalMax(lane); got != wantGlobal {
-			t.Fatalf("step %d lane %v: global max=%d, want %d", step, lane, got, wantGlobal)
+		if resident, retired := engine.globalMax(lane); max(resident, retired) != wantGlobal {
+			t.Fatalf("step %d lane %v: global max=%d/%d, want %d", step, lane, resident, retired, wantGlobal)
 		}
 		for _, through := range []InstanceNum{0, 1, 63, 64, 4_096, InstanceNum(1)<<60 + 63, ^InstanceNum(0)} {
 			want := modelPrefixMaxSeq(model, lane, through)
@@ -265,9 +265,9 @@ func engineModelAttrs(engine *conflictEngine, conf ConfID, cmd Command) modelAtt
 	engine.lanes(conf, func(lane instanceLane) bool {
 		var dep InstanceNum
 		if commandHasGlobalConflictScope(cmd.Kind) {
-			dep = engine.maxEligibleAny(lane)
+			r, ret := engine.maxEligibleAny(lane); dep = max(r, ret)
 		} else {
-			dep = engine.globalMax(lane)
+			r, ret := engine.globalMax(lane); dep = max(r, ret)
 			for _, key := range cmd.ConflictKeys {
 				resident, retired := engine.keyMax(conf, key, lane)
 				dep = max(dep, resident, retired)
@@ -417,11 +417,11 @@ func TestConflictEnginePostFoldDomination(t *testing.T) {
 			t.Fatalf("query %+v decreased across fold: before=%+v after=%+v", query, beforeAttrs, afterAttrs)
 		}
 	}
-	if got := engine.globalMax(lane); got != 3 {
-		t.Fatalf("retired global max=%d, want 3", got)
+	if resident, retired := engine.globalMax(lane); max(resident, retired) != 3 {
+		t.Fatalf("retired global max=%d, want 3", max(resident, retired))
 	}
-	if got := engine.maxEligibleAny(lane); got != 4 {
-		t.Fatalf("retired eligible max=%d, want 4", got)
+	if resident, retired := engine.maxEligibleAny(lane); max(resident, retired) != 4 {
+		t.Fatalf("retired eligible max=%d, want 4", max(resident, retired))
 	}
 	if err := engine.verify(); err != nil {
 		t.Fatal(err)
@@ -471,8 +471,8 @@ func TestConflictEngineNoopMutationDropsMax(t *testing.T) {
 	if resident != lower.Ref.Instance || retired != 0 {
 		t.Fatalf("key max after noop mutation=(%d,%d), want (%d,0)", resident, retired, lower.Ref.Instance)
 	}
-	if got := engine.maxEligibleAny(lane); got != lower.Ref.Instance {
-		t.Fatalf("max eligible after noop mutation=%d, want %d", got, lower.Ref.Instance)
+	if resident, retired := engine.maxEligibleAny(lane); max(resident, retired) != lower.Ref.Instance {
+		t.Fatalf("max eligible after noop mutation=%d/%d, want %d", resident, retired, lower.Ref.Instance)
 	}
 	if err := engine.verify(); err != nil {
 		t.Fatal(err)
@@ -641,3 +641,38 @@ func assertExactKeyPostings(t *testing.T, e *conflictEngine, records map[Instanc
 		}
 	}
 }
+
+
+func TestWalkGlobalDescSkipsUnrelatedResidents(t *testing.T) {
+	t.Parallel()
+	var engine conflictEngine
+	lane := instanceLane{conf: 1, replica: 1}
+	// One old global at instance 1, then many ordinary residents.
+	global := InstanceRecord{
+		Ref: InstanceRef{Conf: 1, Replica: 1, Instance: 1},
+		Status: StatusCommitted, Seq: 1,
+		Command: Command{Kind: CommandConfChange, Payload: []byte("cfg")},
+	}
+	engine.apply(nil, global)
+	for i := InstanceNum(2); i <= 200; i++ {
+		rec := InstanceRecord{
+			Ref: InstanceRef{Conf: 1, Replica: 1, Instance: i},
+			Status: StatusCommitted, Seq: uint64(i),
+			Command: Command{Kind: CommandUser, Payload: []byte("u"), ConflictKeys: [][]byte{[]byte("k")}},
+		}
+		engine.apply(nil, rec)
+	}
+	visits := 0
+	var seen []InstanceNum
+	engine.walkGlobalDesc(lane, 200, func(instance InstanceNum, slot laneSlot) bool {
+		visits++
+		seen = append(seen, instance)
+		if !slot.global() {
+			t.Fatalf("yielded non-global instance %d", instance)
+		}
+		return true
+	})
+	if visits != 1 || len(seen) != 1 || seen[0] != 1 {
+		t.Fatalf("walkGlobalDesc visits=%d seen=%v, want only global instance 1", visits, seen)
+	}
+}

epaxos/conflict_ordering_test.go

@@ -98,13 +98,15 @@ func TestConflictIndexScopesConfigurationsAndExcludedLaneMaximum(t *testing.T) {
 	}
 	for _, rec := range records {
 		rn.instances[rec.Ref] = &instance{rec: rec, phase: phasePreAccept}
-		rn.indexConflicts(rec)
+		rn.engine.apply(nil, rec)
 	}
-	if got := rn.conflictIndex(1, key)[instanceLane{conf: 1, replica: 2}]; got != records[0].Ref {
-		t.Fatalf("configuration 1 index=%s, want %s", got, records[0].Ref)
+	lane2c1 := instanceLane{conf: 1, replica: 2}
+	if resident, _ := rn.engine.keyMax(1, key, lane2c1); resident != records[0].Ref.Instance {
+		t.Fatalf("configuration 1 index=%d, want %s", resident, records[0].Ref)
 	}
-	if got := rn.conflictIndex(2, key)[instanceLane{conf: 2, replica: 2}]; got != records[2].Ref {
-		t.Fatalf("configuration 2 index=%s, want latest %s", got, records[2].Ref)
+	lane2c2 := instanceLane{conf: 2, replica: 2}
+	if resident, _ := rn.engine.keyMax(2, key, lane2c2); resident != records[2].Ref.Instance {
+		t.Fatalf("configuration 2 index=%d, want latest %s", resident, records[2].Ref)
 	}
 
 	newCommand := Command{Payload: []byte("new"), ConflictKeys: [][]byte{key}}