Start the builds in the order of the messages

The queue starts the node whose context ends first: a leaf ends before
its message and a merge after its last message, and at a tie the merge
starts first. With leaves before all merges, a backlog of long messages
kept back the merges that a batch needs, so the view could not shrink.
The merges wait in a heap by the end of their context, so the next build
is found without a scan. The leaf window of 8 unbuilt leaves and the
limit of 8 builds stay as they are.

Issue: beb9a70

diff --git a/queue.go b/queue.go
index 5f2fcb2eb6a31a5d094d072d895aad69baf5b349..c128d6173fd408e669332e0d1106bb2d8019314a 100644 (file)
--- a/queue.go
+++ b/queue.go
@@ -1,6 +1,7 @@
 package main
 
 import (
 package main
 
 import (

+       "container/heap"

        "slices"
 )
 
        "slices"
 )
 


@@ -18,18 +19,24 @@ const (
 // children are built and their texts do not fit in one node. So the core never
 // scans the tree for work.
 //
 // children are built and their texts do not fit in one node. So the core never
 // scans the tree for work.
 //

-// A leaf can start when fewer than leafWindow leaves before it are unbuilt.
-// Leaves start before merges, oldest first, because a turn waits for the leaves
-// of all earlier messages. Merges start in the order in which they became
-// ready. A failed build waits until the next message.
+// The builds start in the order of the messages: the node whose context ends
+// first starts first. The context of leaf i ends before message i, and the
+// context of a merge ends after its last message. At a tie the merge starts
+// first. So a backlog of leaves does not stop the merges that the view needs to
+// shrink. A leaf can start only when fewer than leafWindow leaves before it are
+// unbuilt. A failed build waits until the next message.

 type queue struct {
        leaves  []int
 type queue struct {
        leaves  []int

-       merges  []nodeKey
+       merges  mergeHeap

        queued  map[nodeKey]bool
        running map[nodeKey]bool
        failed  map[nodeKey]bool
        queued  map[nodeKey]bool
        running map[nodeKey]bool
        failed  map[nodeKey]bool

+       retry   []nodeKey

 }
 
 }
 

+// mergeHeap holds the merges that can start, by the end of their context.
+type mergeHeap []nodeKey
+

 func newQueue() *queue {
        return &queue{
                queued:  make(map[nodeKey]bool),
 func newQueue() *queue {
        return &queue{
                queued:  make(map[nodeKey]bool),


@@ -51,7 +58,7 @@ func (q *queue) AddMerge(key nodeKey) {
        }
 
        q.queued[key] = true
        }
 
        q.queued[key] = true

-       q.merges = append(q.merges, key)
+       heap.Push(&q.merges, key)

 }
 
 // Next returns the next node that can start, and marks it as running.
 }
 
 // Next returns the next node that can start, and marks it as running.


@@ -60,17 +67,26 @@ func (q *queue) Next() (nodeKey, bool) {
                return nodeKey{}, false
        }
 
                return nodeKey{}, false
        }
 

-       for _, id := range q.leaves[:min(len(q.leaves), leafWindow)] {
-               key := nodeKey{0, id}
-               if q.startable(key) {
-                       q.running[key] = true
-                       return key, true
-               }
+       leaf, found := q.nextLeaf()
+       if len(q.merges) > 0 && (!found || nodeEnd(q.merges[0].level, q.merges[0].index) <= leaf.index) {
+               key := heap.Pop(&q.merges).(nodeKey)
+               q.running[key] = true
+
+               return key, true
+       }
+
+       if found {
+               q.running[leaf] = true

        }
 
        }
 

-       for _, key := range q.merges {
-               if q.startable(key) {
-                       q.running[key] = true
+       return leaf, found
+}
+
+// nextLeaf returns the oldest leaf in the window that can start.
+func (q *queue) nextLeaf() (nodeKey, bool) {
+       for _, id := range q.leaves[:min(len(q.leaves), leafWindow)] {
+               key := nodeKey{0, id}
+               if !q.running[key] && !q.failed[key] {

                        return key, true
                }
        }
                        return key, true
                }
        }


@@ -78,39 +94,66 @@ func (q *queue) Next() (nodeKey, bool) {
        return nodeKey{}, false
 }
 
        return nodeKey{}, false
 }
 

-func (q *queue) startable(key nodeKey) bool {
-       return !q.running[key] && !q.failed[key]
-}
-

 // Done removes a built node from the queue.
 func (q *queue) Done(key nodeKey) {
        delete(q.running, key)
 // Done removes a built node from the queue.
 func (q *queue) Done(key nodeKey) {
        delete(q.running, key)

-       if key.level == 0 {
-               q.leaves = slices.DeleteFunc(q.leaves, func(id int) bool {
-                       return id == key.index
-               })
-
+       if key.level > 0 {
+               delete(q.queued, key)

                return
        }
 
                return
        }
 

-       delete(q.queued, key)
-       q.merges = slices.DeleteFunc(q.merges, func(other nodeKey) bool {
-               return other == key
-       })
+       position := slices.Index(q.leaves, key.index)
+       if position >= 0 {
+               q.leaves = slices.Delete(q.leaves, position, position+1)
+       }

 }
 
 // Fail keeps a node in the queue, but it cannot start until Retry.
 func (q *queue) Fail(key nodeKey) {
        delete(q.running, key)
 }
 
 // Fail keeps a node in the queue, but it cannot start until Retry.
 func (q *queue) Fail(key nodeKey) {
        delete(q.running, key)

+       if key.level > 0 {
+               q.retry = append(q.retry, key)
+               return
+       }
+

        q.failed[key] = true
 }
 
 // Retry lets the failed nodes start again.
 func (q *queue) Retry() {
        clear(q.failed)
        q.failed[key] = true
 }
 
 // Retry lets the failed nodes start again.
 func (q *queue) Retry() {
        clear(q.failed)

+       for _, key := range q.retry {
+               heap.Push(&q.merges, key)
+       }
+
+       q.retry = q.retry[:0]

 }
 
 // Running returns the number of builds that run.
 func (q *queue) Running() int {
        return len(q.running)
 }
 }
 
 // Running returns the number of builds that run.
 func (q *queue) Running() int {
        return len(q.running)
 }

+
+func (h mergeHeap) Len() int {
+       return len(h)
+}
+
+func (h mergeHeap) Less(a, b int) bool {
+       return nodeEnd(h[a].level, h[a].index) < nodeEnd(h[b].level, h[b].index)
+}
+
+func (h mergeHeap) Swap(a, b int) {
+       h[a], h[b] = h[b], h[a]
+}
+
+func (h *mergeHeap) Push(value any) {
+       *h = append(*h, value.(nodeKey))
+}
+
+func (h *mergeHeap) Pop() any {
+       old := *h
+       key := old[len(old)-1]
+       *h = old[:len(old)-1]
+
+       return key
+}

diff --git a/queue_test.go b/queue_test.go
new file mode 100644 (file)
index 0000000..6bafb5d
--- /dev/null
+++ b/queue_test.go
@@ -0,0 +1,49 @@
+package main
+
+import (
+       "testing"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestQueueStartsInMessageOrder(t *testing.T) {
+       queue := newQueue()
+       queue.AddLeaf(4)
+       queue.AddLeaf(5)
+       queue.AddMerge(nodeKey{1, 2})
+       queue.AddMerge(nodeKey{1, 1})
+
+       var order []nodeKey
+       for {
+               key, ready := queue.Next()
+               if !ready {
+                       break
+               }
+
+               order = append(order, key)
+       }
+
+       require.Equal(t, []nodeKey{{1, 1}, {0, 4}, {0, 5}, {1, 2}}, order)
+}
+
+func TestQueueRetriesAFailedMergeAtTheNextMessage(t *testing.T) {
+       queue := newQueue()
+       queue.AddMerge(nodeKey{1, 0})
+
+       key, ready := queue.Next()
+       require.True(t, ready)
+
+       queue.Fail(key)
+       _, ready = queue.Next()
+       require.False(t, ready)
+
+       queue.Retry()
+       key, ready = queue.Next()
+       require.True(t, ready)
+       require.Equal(t, nodeKey{1, 0}, key)
+
+       queue.Done(key)
+       queue.AddMerge(nodeKey{1, 0})
+       _, ready = queue.Next()
+       require.True(t, ready)
+}