Better design of the Emitter #26

ScottSallinen · 2024-06-30T12:43:07Z

This is the updated design of the Emitter, where it emits to both source and destination asynchronously. The source and destination then remit to the other. In this design, performance with directed graphs is still quite good, and undirected graphs can see a significant boost in performance.

Further, this design offers far more stable guarantees for vertex creation: vertexes are now always created in temporal order within a graph thread (i.e., they are sorted by first observation, regardless of if the first observation was from some other graph thread creating a new edge to a vertex that didn't exist yet.)

The resultant graph structure is now fully deterministic, and everything is sorted by total global ordering 🎉

ScottSallinen · 2024-06-30T12:52:56Z

graph/stream-enact-events.go

 		closed = gt.FromEmitQueue.IsClosed()
 	}
-	return closed, count
+	return closed, count, largest, false, false
 }

 // Injects expired edges into the topology event buffer. To be done after the topology event buffer has been remapped with internal source ids.
 func InjectExpired[EP EPP[E], V VPI[V], E EPI[E], M MVI[M], N any](g *Graph[V, E, M, N], gt *GraphThread[V, E, M, N], changeCount uint64, uniqueCount uint64, delOnExpire uint64) (newUniqueCount uint64) {


Because of the way the TopologyEventBuff is now indexed (via InputEvent ThreadOrder and topologyFlushedUpTo), this will need to be updated to inject deletes in a different way. (They can't go into the main TopologyEventBuff unless they actually came from the Emitter, or this would mess with the FlushedUpTo index.)

One way could be to have the Emitter track them and inject them, but this would be terrible for performance.

Possibly better to keep a separate buffer of the expired events and have perhaps a flag in the VertexPendingBuff that signifies which buffer to pull from for the next change...

A third possibility (and probably best for performance) is to indeed use the ThreadOrder and use additional offset calculations (incremented with each injected delete into the buffer), but this might be complicated...

I think we can turn TopologyEventBuff into a FIFO queue with one head pointer (for inserting new adds) and two tail pointers (next event to add and next event to delete). With this queue, we don't need to re-insert the same event into TopologyEventBuff to delete the event.

If the Emitter doesn't track deletes, can deletes still have proper global event IDs or ThreadOrder? This might be a problem for enforcing ordering between messages and events.

Maybe we should consider inserting deletes directly in the input file, which should make things much more straightforward...

LuuO

Great! The new design is really cool, as it provides both higher performance and better ordering guarantees. It also gets rid of one extra thread!

I think we can further improve the performance on directed graphs by skipping the remit messages from src to dst? Those messages don't seem to do anything.

LuuO · 2024-07-27T17:15:24Z

utils/ring-buffer.go

@@ -557,3 +561,150 @@ func (rb *GrowableRingBuff[T]) GetSlow(pos uint64) (item T, closed bool, fails i
 		}
 	}
 }
+
+// ---------------------------- MPSC Token Ring Buffer ----------------------------


These ring buffers are really cool!

LuuO · 2024-07-28T15:51:27Z

graph/run-dynamic.go

+			gt.Response <- ACK
+			wantsAsyncQueryNow = false
+			blockTop = true // We will block further topology until we receive the resume command.
+			// TODO: Check for g.Options.AllowAsyncVertexProps , need to adjust for this.


Is g.Options.AllowAsyncVertexProps assumed to be true here?

LuuO · 2024-07-28T16:23:00Z

graph/stream-enact-events.go

-			pos = vs.InEventPos
-			if onInEdgeAddFunc != nil {
-				onInEdgeAddFunc(g, gt, gt.Vertex(didx), gt.VertexProperty(didx), didx, pos, &event)
+		if gt.InputEvent.EventType() == ADD {


Are deletes from Emitter supported?

LuuO · 2024-07-28T16:32:43Z

graph/stream-enact-events.go

+			gt.FailedRemit = true
+			return false, count, largest, true, false
+		}
+		gt.TopologyEventBuff[gt.InputEvent.ThreadOrderDst-topologyFlushedUpTo] = RawEdgeEvent[E]{TypeAndEventIdx: gt.InputEvent.TypeAndEventIdx, Sidx: vidx, Edge: gt.RemitEvent.Edge}


Why do we need to save the event here? (Same for Line 94)

LuuO · 2024-07-30T15:31:04Z

graph/stream-enact-events.go

 		closed = gt.FromEmitQueue.IsClosed()
 	}
-	return closed, count
+	return closed, count, largest, false, false
 }

 // Injects expired edges into the topology event buffer. To be done after the topology event buffer has been remapped with internal source ids.
 func InjectExpired[EP EPP[E], V VPI[V], E EPI[E], M MVI[M], N any](g *Graph[V, E, M, N], gt *GraphThread[V, E, M, N], changeCount uint64, uniqueCount uint64, delOnExpire uint64) (newUniqueCount uint64) {


I think we can turn TopologyEventBuff into a FIFO queue with one head pointer (for inserting new adds) and two tail pointers (next event to add and next event to delete). With this queue, we don't need to re-insert the same event into TopologyEventBuff to delete the event.

LuuO · 2024-07-30T16:07:10Z

graph/run-dynamic.go

+	eventsHeight := uint64(0)
+	eventsApplyCountNow := uint64(0)
+	eventsTotalCount := uint64(0)
+	eventsFlushedUpTo := uint64(0)


Suggested change

eventsFlushedUpTo := uint64(0)

eventsFlushedUpTo := uint64(0) // ThreadOrder of the last event flushed - 1 (or the number of events flushed)

LuuO · 2024-07-30T16:11:27Z

graph/stream.go

+	ThreadOrderSrc  uint64  // For src thread, the emmiter tracked index.
+	ThreadOrderDst  uint64  // For dst thread, the emmiter tracked index.


Suggested change

ThreadOrderSrc uint64 // For src thread, the emmiter tracked index.

ThreadOrderDst uint64 // For dst thread, the emmiter tracked index.

ThreadOrderSrc uint64 // For src thread, the emitter tracked index.

ThreadOrderDst uint64 // For dst thread, the emitter tracked index.

LuuO · 2024-07-30T16:16:43Z

graph/run-dynamic.go

 			gt.Status = REMIT
-			remitClosed, remitCount = checkToRemit(alg, g, gt, onInEdgeAddFunc)
+			remitClosed, remitCount, currLargest, _, wantsQueryNow = checkToRemit(alg, g, gt, onInEdgeAddFunc, eventsFlushedUpTo)
+			largest = utils.Max(largest, currLargest)


Suggested change

largest = utils.Max(largest, currLargest)

largest = utils.Max(largest, currLargest) // checkToRemit might return 0 for currLargest

LuuO · 2024-07-30T16:19:46Z

graph/run-dynamic.go

 	remitCount := uint64(0)
+	remitTotalCount := uint64(0)
+
+	largest := uint64(0)


Suggested change

largest := uint64(0)

largest := uint64(0) // ThreadOrder index of the last event remitted (i.e., the last event buffered in TopologyEventBuff).

LuuO · 2024-08-02T18:05:01Z

graph/stream-enact-events.go

-		if sidx, ok = gt.VertexMap[gt.TopologyEventBuff[i].SrcRaw]; !ok {
-			sidx = NewVertex(alg, g, gt, gt.TopologyEventBuff[i].SrcRaw, gt.TopologyEventBuff[i].EventIdx())
+		if !undirected && gt.TopologyEventBuff[i].EventIdx()%2 == 0 {
+			continue // Skip even events for directed graphs.


On direct graphs, why do both src and dst still need to remit to each other? It seems like the remit events from src to dst are just ignored.

ScottSallinen added 2 commits June 30, 2024 05:32

Update Emitter (removes prior remitter design)

d78b0b6

Update run-go-test.yml

d51b497

ScottSallinen requested a review from LuuO June 30, 2024 12:43

ScottSallinen assigned ScottSallinen and LuuO Jun 30, 2024

ScottSallinen commented Jun 30, 2024

View reviewed changes

LuuO reviewed Aug 2, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Better design of the Emitter #26

Better design of the Emitter #26

ScottSallinen commented Jun 30, 2024 •

edited

Loading

ScottSallinen Jun 30, 2024 •

edited

Loading

LuuO Jul 30, 2024

LuuO Jul 30, 2024

LuuO Jul 30, 2024

LuuO left a comment •

edited

Loading

LuuO Jul 27, 2024

LuuO Jul 28, 2024

LuuO Jul 28, 2024

LuuO Jul 28, 2024

LuuO Jul 30, 2024

LuuO Jul 30, 2024

LuuO Jul 30, 2024

LuuO Jul 30, 2024

LuuO Jul 30, 2024

LuuO Aug 2, 2024

	eventsFlushedUpTo := uint64(0)
	eventsFlushedUpTo := uint64(0) // ThreadOrder of the last event flushed - 1 (or the number of events flushed)

		ThreadOrderSrc uint64 // For src thread, the emmiter tracked index.
		ThreadOrderDst uint64 // For dst thread, the emmiter tracked index.

	largest = utils.Max(largest, currLargest)
	largest = utils.Max(largest, currLargest) // checkToRemit might return 0 for currLargest

	largest := uint64(0)
	largest := uint64(0) // ThreadOrder index of the last event remitted (i.e., the last event buffered in TopologyEventBuff).

Better design of the Emitter #26

Are you sure you want to change the base?

Better design of the Emitter #26

Conversation

ScottSallinen commented Jun 30, 2024 • edited Loading

ScottSallinen Jun 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuuO left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ScottSallinen commented Jun 30, 2024 •

edited

Loading

ScottSallinen Jun 30, 2024 •

edited

Loading

LuuO left a comment •

edited

Loading