Name and order the build queue code

The queue of the node builds is BuildQueue in build_queue.go, and its
other names start with build: buildMax and buildLeafWindow are its
limits, and BuildMergeHeap holds its merges. The Telegram bot has a
queue of events too, so a bare queue does not tell which one it is.

The declarations of the queue file follow the order of the go-code
skill. The behavior does not change.

Issue: 947bddb

diff --git a/queue.go b/build_queue.go
similarity index 64%
rename from queue.go
rename to build_queue.go
index c128d6173fd408e669332e0d1106bb2d8019314a..2458f5a185100cfd43a8a33c0db42e0716030f28 100644 (file)
--- a/queue.go
+++ b/build_queue.go
@@ -6,16 +6,24 @@ import (
 )
 
 const (
 )
 
 const (

-       // maxBuilds is the number of model builds that can run at once.
-       maxBuilds = 8
+       // buildMax is the number of model builds that can run at once.
+       buildMax = 8

 
 

-       // leafWindow is the number of unbuilt leaves that can be before a leaf
-       // whose build starts.
-       leafWindow = 8
+       // buildLeafWindow is the number of unbuilt leaves that can be before a
+       // leaf whose build starts.
+       buildLeafWindow = 8

 )
 
 )
 

-// queue holds the nodes that wait for a model build. A leaf joins it when its
-// message is too long to be its own node. A merge joins it when both of its
+func newBuildQueue() *BuildQueue {
+       return &BuildQueue{
+               queued:  make(map[nodeKey]bool),
+               running: make(map[nodeKey]bool),
+               failed:  make(map[nodeKey]bool),
+       }
+}
+
+// BuildQueue holds the nodes that wait for a model build. A leaf joins it when
+// its message is too long to be its own node. A merge joins it when both of its

 // 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.
 //


@@ -23,36 +31,25 @@ const (
 // 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
 // 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 {
+// shrink. A leaf can start only when fewer than buildLeafWindow leaves before
+// it are unbuilt. A failed build waits until the next message.
+type BuildQueue struct {

        leaves  []int
        leaves  []int

-       merges  mergeHeap
+       merges  BuildMergeHeap

        queued  map[nodeKey]bool
        running map[nodeKey]bool
        failed  map[nodeKey]bool
        retry   []nodeKey
 }
 
        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),
-               running: make(map[nodeKey]bool),
-               failed:  make(map[nodeKey]bool),
-       }
-}
-

 // AddLeaf adds the leaf of message id. The IDs must come in increasing order.
 // AddLeaf adds the leaf of message id. The IDs must come in increasing order.

-func (q *queue) AddLeaf(id int) {
+func (q *BuildQueue) AddLeaf(id int) {

        q.leaves = append(q.leaves, id)
 }
 
 // AddMerge adds a merge whose children are built. A merge that is in the queue
 // already stays as it is.
        q.leaves = append(q.leaves, id)
 }
 
 // AddMerge adds a merge whose children are built. A merge that is in the queue
 // already stays as it is.

-func (q *queue) AddMerge(key nodeKey) {
+func (q *BuildQueue) AddMerge(key nodeKey) {

        if q.queued[key] {
                return
        }
        if q.queued[key] {
                return
        }


@@ -62,8 +59,8 @@ func (q *queue) AddMerge(key nodeKey) {
 }
 
 // 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.

-func (q *queue) Next() (nodeKey, bool) {
-       if len(q.running) >= maxBuilds {
+func (q *BuildQueue) Next() (nodeKey, bool) {
+       if len(q.running) >= buildMax {

                return nodeKey{}, false
        }
 
                return nodeKey{}, false
        }
 


@@ -82,20 +79,8 @@ func (q *queue) Next() (nodeKey, bool) {
        return leaf, found
 }
 
        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 nodeKey{}, false
-}
-

 // Done removes a built node from the queue.
 // Done removes a built node from the queue.

-func (q *queue) Done(key nodeKey) {
+func (q *BuildQueue) Done(key nodeKey) {

        delete(q.running, key)
        if key.level > 0 {
                delete(q.queued, key)
        delete(q.running, key)
        if key.level > 0 {
                delete(q.queued, key)


@@ -109,7 +94,7 @@ func (q *queue) Done(key nodeKey) {
 }
 
 // Fail keeps a node in the queue, but it cannot start until Retry.
 }
 
 // Fail keeps a node in the queue, but it cannot start until Retry.

-func (q *queue) Fail(key nodeKey) {
+func (q *BuildQueue) Fail(key nodeKey) {

        delete(q.running, key)
        if key.level > 0 {
                q.retry = append(q.retry, key)
        delete(q.running, key)
        if key.level > 0 {
                q.retry = append(q.retry, key)


@@ -120,7 +105,7 @@ func (q *queue) Fail(key nodeKey) {
 }
 
 // Retry lets the failed nodes start again.
 }
 
 // Retry lets the failed nodes start again.

-func (q *queue) Retry() {
+func (q *BuildQueue) Retry() {

        clear(q.failed)
        for _, key := range q.retry {
                heap.Push(&q.merges, key)
        clear(q.failed)
        for _, key := range q.retry {
                heap.Push(&q.merges, key)


@@ -130,27 +115,42 @@ func (q *queue) Retry() {
 }
 
 // Running returns the number of builds that run.
 }
 
 // Running returns the number of builds that run.

-func (q *queue) Running() int {
+func (q *BuildQueue) Running() int {

        return len(q.running)
 }
 
        return len(q.running)
 }
 

-func (h mergeHeap) Len() int {
+// nextLeaf returns the oldest leaf in the window that can start.
+func (q *BuildQueue) nextLeaf() (nodeKey, bool) {
+       for _, id := range q.leaves[:min(len(q.leaves), buildLeafWindow)] {
+               key := nodeKey{0, id}
+               if !q.running[key] && !q.failed[key] {
+                       return key, true
+               }
+       }
+
+       return nodeKey{}, false
+}
+
+// BuildMergeHeap holds the merges that can start, by the end of their context.
+type BuildMergeHeap []nodeKey
+
+func (h BuildMergeHeap) Len() int {

        return len(h)
 }
 
        return len(h)
 }
 

-func (h mergeHeap) Less(a, b int) bool {
+func (h BuildMergeHeap) Less(a, b int) bool {

        return nodeEnd(h[a].level, h[a].index) < nodeEnd(h[b].level, h[b].index)
 }
 
        return nodeEnd(h[a].level, h[a].index) < nodeEnd(h[b].level, h[b].index)
 }
 

-func (h mergeHeap) Swap(a, b int) {
+func (h BuildMergeHeap) Swap(a, b int) {

        h[a], h[b] = h[b], h[a]
 }
 
        h[a], h[b] = h[b], h[a]
 }
 

-func (h *mergeHeap) Push(value any) {
+func (h *BuildMergeHeap) Push(value any) {

        *h = append(*h, value.(nodeKey))
 }
 
        *h = append(*h, value.(nodeKey))
 }
 

-func (h *mergeHeap) Pop() any {
+func (h *BuildMergeHeap) Pop() any {

        old := *h
        key := old[len(old)-1]
        *h = old[:len(old)-1]
        old := *h
        key := old[len(old)-1]
        *h = old[:len(old)-1]

diff --git a/queue_test.go b/build_queue_test.go
similarity index 94%
rename from queue_test.go
rename to build_queue_test.go
index 6bafb5df840e3a2e1ed7017448a2c895b6ce029d..76d9211b0ffeb0fd6c075c090258e058c755fc46 100644 (file)
--- a/queue_test.go
+++ b/build_queue_test.go
@@ -7,7 +7,7 @@ import (
 )
 
 func TestQueueStartsInMessageOrder(t *testing.T) {
 )
 
 func TestQueueStartsInMessageOrder(t *testing.T) {

-       queue := newQueue()
+       queue := newBuildQueue()

        queue.AddLeaf(4)
        queue.AddLeaf(5)
        queue.AddMerge(nodeKey{1, 2})
        queue.AddLeaf(4)
        queue.AddLeaf(5)
        queue.AddMerge(nodeKey{1, 2})


@@ -27,7 +27,7 @@ func TestQueueStartsInMessageOrder(t *testing.T) {
 }
 
 func TestQueueRetriesAFailedMergeAtTheNextMessage(t *testing.T) {
 }
 
 func TestQueueRetriesAFailedMergeAtTheNextMessage(t *testing.T) {

-       queue := newQueue()
+       queue := newBuildQueue()

        queue.AddMerge(nodeKey{1, 0})
 
        key, ready := queue.Next()
        queue.AddMerge(nodeKey{1, 0})
 
        key, ready := queue.Next()

diff --git a/memory.go b/memory.go
index f8171b2239b9b502b422447cf81ae4f73fe0ae33..6937726b36de7998e18617d964ccc8fe77157e8f 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -56,7 +56,7 @@ type Memory struct {
        tree       *Tree
        chat       *View
        cview      *View
        tree       *Tree
        chat       *View
        cview      *View

-       queue      *queue
+       queue      *BuildQueue

        summarizer Summarizer
        ctx        context.Context
        cancel     context.CancelFunc
        summarizer Summarizer
        ctx        context.Context
        cancel     context.CancelFunc


@@ -82,7 +82,7 @@ func openMemory(ctx context.Context, root string, clock func() time.Time, summar
        m := &Memory{
                root:       root,
                log:        log,
        m := &Memory{
                root:       root,
                log:        log,

-               queue:      newQueue(),
+               queue:      newBuildQueue(),

                summarizer: summarizer,
                changed:    make(chan struct{}),
        }
                summarizer: summarizer,
                changed:    make(chan struct{}),
        }

diff --git a/memory_test.go b/memory_test.go
index b2076011913912453f32bc98b02096aa5a6f7c65..8b418d860d376f55f672fcb030215c0c34a1fba4 100644 (file)
--- a/memory_test.go
+++ b/memory_test.go
@@ -134,12 +134,12 @@ func TestMemoryRunsAtMostEightBuilds(t *testing.T) {
        require.Equal(t, []nodeKey{{0, 0}, {0, 1}, {0, 2}, {0, 3}, {0, 4}, {0, 5}, {0, 6}, {0, 7}}, running)
 
        require.Eventually(t, func() bool {
        require.Equal(t, []nodeKey{{0, 0}, {0, 1}, {0, 2}, {0, 3}, {0, 4}, {0, 5}, {0, 6}, {0, 7}}, running)
 
        require.Eventually(t, func() bool {

-               return summarizer.Active() == maxBuilds
+               return summarizer.Active() == buildMax

        }, 10*time.Second, time.Millisecond)
 
        close(summarizer.gate)
        require.NoError(t, memory.Wait(t.Context()))
        }, 10*time.Second, time.Millisecond)
 
        close(summarizer.gate)
        require.NoError(t, memory.Wait(t.Context()))

-       require.Equal(t, maxBuilds, summarizer.peak)
+       require.Equal(t, buildMax, summarizer.peak)

 
        for level, count := range []int{12, 6, 3, 1} {
                for index := range count {
 
        for level, count := range []int{12, 6, 3, 1} {
                for index := range count {

diff --git a/probe.go b/probe.go
index 32a3ba99b27f98997bfe113839f931cca61882a5..e12516d12b19559e532c067a59ea10ccf73e914e 100644 (file)
--- a/probe.go
+++ b/probe.go
@@ -40,10 +40,10 @@ const (
 // the first failed call, after probeCallLimit calls, or when the used percent
 // of a quota window grows by more than probeQuotaLimit points. At the end it
 // reports the calls, the time per call, the calls at once, the cached share,
 // the first failed call, after probeCallLimit calls, or when the used percent
 // of a quota window grows by more than probeQuotaLimit points. At the end it
 // reports the calls, the time per call, the calls at once, the cached share,

-// the quota, the retries, and the line sizes. BUILDS, from 1 to maxBuilds,
+// the quota, the retries, and the line sizes. BUILDS, from 1 to buildMax,

 // limits the builds that call the model at once in this run only. The core
 // limits the builds that call the model at once in this run only. The core

-// still queues up to maxBuilds builds. The run holds the lock of the token
-// file, so it does not run while the daemon runs.
+// still queues up to buildMax builds. The run holds the lock of the token file,
+// so it does not run while the daemon runs.

 func probe(ctx context.Context, args []string, out io.Writer) (err error) {
        if len(args) != 2 && len(args) != 3 {
                return errors.New("usage: onechat probe LOG COUNT [BUILDS]")
 func probe(ctx context.Context, args []string, out io.Writer) (err error) {
        if len(args) != 2 && len(args) != 3 {
                return errors.New("usage: onechat probe LOG COUNT [BUILDS]")


@@ -54,11 +54,11 @@ func probe(ctx context.Context, args []string, out io.Writer) (err error) {
                return fmt.Errorf("COUNT %q is not a positive number", args[1])
        }
 
                return fmt.Errorf("COUNT %q is not a positive number", args[1])
        }
 

-       builds := maxBuilds
+       builds := buildMax

        if len(args) == 3 {
                builds, err = strconv.Atoi(args[2])
        if len(args) == 3 {
                builds, err = strconv.Atoi(args[2])

-               if err != nil || builds < 1 || builds > maxBuilds {
-                       return fmt.Errorf("BUILDS %q is not a number from 1 to %d", args[2], maxBuilds)
+               if err != nil || builds < 1 || builds > buildMax {
+                       return fmt.Errorf("BUILDS %q is not a number from 1 to %d", args[2], buildMax)

                }
        }
 
                }
        }
 


@@ -145,10 +145,10 @@ type probeRun struct {
 }
 
 // Run appends the entries. It waits for the builds only before a user message,
 }
 
 // Run appends the entries. It waits for the builds only before a user message,

-// as a turn does, so up to maxBuilds builds run at once in between, and it
-// waits for all builds at the end. It returns the number of messages appended
-// and why the run stopped early, or "" when it did not. After a stop it waits
-// for the calls in flight, so no sent call is cancelled.
+// as a turn does, so up to buildMax builds run at once in between, and it waits
+// for all builds at the end. It returns the number of messages appended and why
+// the run stopped early, or "" when it did not. After a stop it waits for the
+// calls in flight, so no sent call is cancelled.

 func (r *probeRun) Run(ctx context.Context, entries []probeEntry) (int, string) {
        appended, reason := r.append(ctx, entries)
        if reason == "" {
 func (r *probeRun) Run(ctx context.Context, entries []probeEntry) (int, string) {
        appended, reason := r.append(ctx, entries)
        if reason == "" {