stream trades through a sink instead of allocating them in the report

book/book.go

@@ -29,21 +29,39 @@ func New(ordersCap, levelsCap int) *Book {
 		asks:   side{bid: false},
 	}
 }
+func (b *Book) Reserve(seq uint64) levels.Handle {
+	return b.levels.Reserve(seq)
+}
 
-func (b *Book) Add(s Side, seq, price, qty uint64) (levels.Handle, error) {
+func (b *Book) Rest(s Side, h levels.Handle, price, qty uint64) error {
 	if qty == 0 {
-		return levels.NoHandle, levels.ErrZeroQty
+		b.levels.Release(h)
+		return levels.ErrZeroQty
 	}
 	sd := b.side(s)
 
 	i, found := sd.find(b.levels, price)
 	if found {
-		return b.levels.Push(sd.levels[i], seq, qty)
+		b.levels.Rest(h, sd.levels[i], qty)
+		return nil
 	}
 
 	lvl := b.levels.NewLevel(price)
 	sd.insertAt(i, lvl)
-	return b.levels.Push(lvl, seq, qty)
+	b.levels.Rest(h, lvl, qty)
+	return nil
+}
+
+func (b *Book) Release(h levels.Handle) {
+	b.levels.Release(h)
+}
+
+func (b *Book) Add(s Side, seq, price, qty uint64) (levels.Handle, error) {
+	h := b.Reserve(seq)
+	if err := b.Rest(s, h, price, qty); err != nil {
+		return levels.NoHandle, err
+	}
+	return h, nil
 }
 
 func (b *Book) LevelAt(s Side, depth int) (price, totalQty uint64, ok bool) {

book/book_test.go

@@ -73,6 +73,19 @@ func TestBook(t *testing.T) {
 	require.Equal(t, uint64(1), totalQty)
 
 	require.PanicsWithValue(t, "book: invalid side", func() { _, _ = book.Add(3, 7, 1, 1) })
+
+	h := book.Reserve(9)
+	require.Equal(t, uint64(9), h.Seq())
+	require.ErrorIs(t, book.Rest(Buy, h, 100, 0), levels.ErrZeroQty)
+	_, _, ok = book.Best(Buy)
+	require.False(t, ok)
+	_, err = book.Cancel(h)
+	require.ErrorIs(t, err, levels.ErrNotFound)
+
+	h = book.Reserve(10)
+	book.Release(h)
+	_, err = book.Cancel(h)
+	require.ErrorIs(t, err, levels.ErrNotFound)
 }
 
 func TestFrontFill(t *testing.T) {

levels/levels.go

@@ -23,15 +23,16 @@ func makeHandle(index int32, seq uint64) Handle {
 	return Handle(seq<<indexBits | uint64(index))
 }
 
-func SeqHandle(seq uint64) Handle {
-	return makeHandle(indexMask, seq)
-}
-
 func (h Handle) index() int32 { return int32(uint64(h) & indexMask) }
 func (h Handle) Seq() uint64  { return uint64(h) >> indexBits }
 
 type LevelIndex int32
 
+const (
+	unlinked LevelIndex = -1
+	reserved LevelIndex = -2
+)
+
 type level struct {
 	price    uint64
 	totalQty uint64
@@ -68,13 +69,13 @@ func (l *Levels) alloc() int32 {
 		panic("levels: alloc: order capacity exhausted")
 	}
 	index := int32(len(l.orders))
-	l.orders = append(l.orders, order{qty: 0, prev: -1, next: -1})
+	l.orders = append(l.orders, order{qty: 0, prev: -1, next: -1, level: unlinked})
 	return index
 }
 
 func (l *Levels) free(index int32) {
 	o := &l.orders[index]
-	o.level = -1
+	o.level = unlinked
 	o.qty = 0
 	o.prev = -1
 	o.next = l.freeOrder

levels/levels_test.go

@@ -45,9 +45,16 @@ func checkInvariants(t *testing.T, l *Levels) {
 		levelsLen--
 	}
 
+	for _, o := range l.orders {
+		if o.level == reserved {
+			require.Greater(t, ordersLen, 0)
+			ordersLen--
+		}
+	}
+
 	n := l.freeOrder
 	for n != -1 {
-		require.Equal(t, LevelIndex(-1), l.orders[n].level)
+		require.Equal(t, unlinked, l.orders[n].level)
 		require.Equal(t, int32(-1), l.orders[n].prev)
 		require.Greater(t, ordersLen, 0)
 		n = l.orders[n].next
@@ -277,3 +284,55 @@ func TestHandlePacking(t *testing.T) {
 	require.Equal(t, int32(0), NoHandle.index())
 	require.Equal(t, uint64(0), NoHandle.Seq())
 }
+
+func TestReserve(t *testing.T) {
+	l := New(10, 10)
+	lvl := l.NewLevel(5)
+
+	h := l.Reserve(7)
+	require.Equal(t, uint64(7), h.Seq())
+	require.Equal(t, int32(0), h.index())
+	require.Equal(t, reserved, l.orders[0].level)
+	checkInvariants(t, l)
+
+	_, _, _, err := l.Cancel(h)
+	require.ErrorIs(t, err, ErrNotFound)
+	_, err = l.Qty(h)
+	require.ErrorIs(t, err, ErrNotFound)
+	_, _, ok := l.Front(lvl)
+	require.False(t, ok)
+
+	require.PanicsWithValue(t, "levels: Rest: invalid qty", func() { l.Rest(h, lvl, 0) })
+
+	l.Rest(h, lvl, 10)
+	require.Equal(t, lvl, l.orders[0].level)
+	require.Equal(t, uint64(10), l.levels[lvl].totalQty)
+	fh, qty, ok := l.Front(lvl)
+	require.True(t, ok)
+	require.Equal(t, h, fh)
+	require.Equal(t, uint64(10), qty)
+	checkInvariants(t, l)
+
+	require.PanicsWithValue(t, "levels: Rest: order is not reserved", func() { l.Rest(h, lvl, 10) })
+	require.PanicsWithValue(t, "levels: Release: order is not reserved", func() { l.Release(h) })
+
+	h2 := l.Reserve(8)
+	require.Equal(t, int32(1), h2.index())
+	checkInvariants(t, l)
+
+	l.Release(h2)
+	require.Equal(t, unlinked, l.orders[1].level)
+	require.Equal(t, int32(1), l.freeOrder)
+	checkInvariants(t, l)
+
+	require.PanicsWithValue(t, "levels: Release: order is not reserved", func() { l.Release(h2) })
+	require.PanicsWithValue(t, "levels: Rest: order is not reserved", func() { l.Rest(h2, lvl, 10) })
+
+	h3 := l.Reserve(9)
+	require.Equal(t, h2.index(), h3.index())
+	require.PanicsWithValue(t, "levels: Release: order is not reserved", func() { l.Release(h2) })
+	l.Release(h3)
+	checkInvariants(t, l)
+
+	require.PanicsWithValue(t, "levels: Reserve: invalid seq", func() { l.Reserve(0) })
+}

levels/order.go

@@ -8,37 +8,65 @@ type order struct {
 	next  int32
 }
 
-func (l *Levels) Push(level LevelIndex, seq, qty uint64) (Handle, error) {
-	if qty == 0 {
-		return NoHandle, ErrZeroQty
+func (l *Levels) Reserve(seq uint64) Handle {
+	if seq == 0 || seq > seqMax {
+		panic("levels: Reserve: invalid seq")
 	}
 
-	lvl := l.getLevel(level)
+	index := l.alloc()
+	o := &l.orders[index]
+	o.seq = seq
+	o.level = reserved
+	return makeHandle(index, seq)
+}
 
-	if seq == 0 || seq > seqMax {
-		panic("levels: Push: invalid seq")
+func (l *Levels) Rest(h Handle, lvl LevelIndex, qty uint64) {
+	if qty == 0 {
+		panic("levels: Rest: invalid qty")
 	}
 
-	index := l.alloc()
+	index, seq := h.index(), h.Seq()
+
 	o := &l.orders[index]
+	if o.level != reserved || o.seq != seq {
+		panic("levels: Rest: order is not reserved")
+	}
+	lv := l.getLevel(lvl)
 
-	o.level = level
+	o.level = lvl
 	o.qty = qty
-	o.seq = seq
 
-	lvl.totalQty += qty
+	lv.totalQty += qty
+
+	if lv.head == -1 {
+		lv.head = index
+		lv.tail = index
+		return
+	}
 
-	if lvl.head == -1 {
-		lvl.head = index
-		lvl.tail = index
+	o.prev = lv.tail
+	l.orders[lv.tail].next = index
+	lv.tail = index
+}
 
-		return makeHandle(index, seq), nil
+func (l *Levels) Release(h Handle) {
+	index := h.index()
+	o := &l.orders[index]
+	if o.level != reserved || o.seq != h.Seq() {
+		panic("levels: Release: order is not reserved")
+	}
+	l.free(index)
+}
+
+func (l *Levels) Push(level LevelIndex, seq, qty uint64) (Handle, error) {
+	if qty == 0 {
+		return NoHandle, ErrZeroQty
 	}
+	l.getLevel(level)
 
-	o.prev = lvl.tail
-	l.orders[lvl.tail].next = index
-	lvl.tail = index
-	return makeHandle(index, seq), nil
+	h := l.Reserve(seq)
+	l.Rest(h, level, qty)
+	return h, nil
 }
 
 func (l *Levels) Cancel(h Handle) (LevelIndex, uint64, bool, error) {

levels/random_test.go

@@ -96,6 +96,7 @@ func runRandom(t *testing.T, seed int64, steps int) {
 	model := map[LevelIndex]*modelLevel{}
 	var liveLevels []LevelIndex // keys of model, for picking a random level
 	var stale []Handle          // handles of cancelled and fully filled orders
+	var pending []Handle        // reserved handles not yet rested or released
 	var seq uint64              // order sequence, owned by the caller of Push
 
 	pickLevel := func() LevelIndex {
@@ -137,7 +138,7 @@ func runRandom(t *testing.T, seed int64, steps int) {
 			model[id] = &modelLevel{price: price}
 			liveLevels = append(liveLevels, id)
 
-		case op < 50: // push
+		case op < 40: // push
 			if len(liveLevels) == 0 {
 				continue
 			}
@@ -148,6 +149,30 @@ func runRandom(t *testing.T, seed int64, steps int) {
 			require.NoError(t, err, "step %d", step)
 			model[id].orders = append(model[id].orders, modelOrder{h: h, qty: qty})
 
+		case op < 45: // reserve a slot, settle it on a later step
+			seq++
+			pending = append(pending, l.Reserve(seq))
+
+		case op < 50: // rest or release a pending reservation
+			if len(pending) == 0 {
+				continue
+			}
+			i := rng.Intn(len(pending))
+			h := pending[i]
+			pending[i] = pending[len(pending)-1]
+			pending = pending[:len(pending)-1]
+			_, _, _, err := l.Cancel(h)
+			require.ErrorIs(t, err, ErrNotFound, "step %d: cancel of a reserved order", step)
+			if len(liveLevels) == 0 || rng.Intn(2) == 0 {
+				l.Release(h)
+				stale = append(stale, h)
+				continue
+			}
+			id := pickLevel()
+			qty := uint64(rng.Intn(100) + 1)
+			l.Rest(h, id, qty)
+			model[id].orders = append(model[id].orders, modelOrder{h: h, qty: qty})
+
 		case op < 70: // cancel a live order
 			id, i, ok := pickOrder()
 			if !ok {
@@ -221,4 +246,10 @@ func runRandom(t *testing.T, seed int64, steps int) {
 			compareModel(t, l, model)
 		}
 	}
+
+	for _, h := range pending {
+		l.Release(h)
+	}
+	checkInvariants(t, l)
+	compareModel(t, l, model)
 }