Send the compaction view in blocks of four lines

[onechat] / probe.go

diff --git a/probe.go b/probe.go
index acd4f6b5a20d12706eaec58763cee78a01fc4e2c..bbce24f65541d455b05796a4d30c58fdf5903e6d 100644 (file)
--- a/probe.go
+++ b/probe.go
@@ -2,6 +2,7 @@ package main
 
 import (
        "bufio"
 
 import (
        "bufio"

+       "cmp"

        "context"
        "encoding/json"
        "errors"
        "context"
        "encoding/json"
        "errors"


@@ -25,20 +26,26 @@ const (
        // can grow in a probe run before the run stops.
        probeQuotaLimit = 15.0
 
        // can grow in a probe run before the run stops.
        probeQuotaLimit = 15.0
 

+       // probeCallLimit is the most model calls of a probe run.
+       probeCallLimit = 660
+

        // probeLineBytes is the most bytes of one line of a probe log.
        probeLineBytes = 1 << 20
 )
 
        // probeLineBytes is the most bytes of one line of a probe log.
        probeLineBytes = 1 << 20
 )
 

-// probe runs "onechat probe LOG COUNT". It appends the first COUNT messages of
-// LOG, one JSON object {kind, text} per line, to a new chat in probe/ in the
-// data directory, and builds its tree with the real compactor. After each
-// message it waits for the builds, as a turn does. It stops at the first
-// failed call, or when the used percent of a quota window grows by more than
-// probeQuotaLimit points. At the end it reports the calls, the cached share,
-// the quota, the retries, and the line sizes.
+// probe runs "onechat probe LOG COUNT [BUILDS]". It appends the first COUNT
+// messages of LOG, one JSON object {kind, text} per line, to a new chat in
+// probe/ in the data directory, and builds its tree with the real compactor.
+// Before each user message it waits for the builds, as a turn does. It stops at
+// 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,
+// limits the builds that call the model at once in this run only. The core
+// still queues up to maxBuilds builds.

 func probe(ctx context.Context, args []string, out io.Writer) error {
 func probe(ctx context.Context, args []string, out io.Writer) error {

-       if len(args) != 2 {
-               return errors.New("usage: onechat probe LOG COUNT")
+       if len(args) != 2 && len(args) != 3 {
+               return errors.New("usage: onechat probe LOG COUNT [BUILDS]")

        }
 
        count, err := strconv.Atoi(args[1])
        }
 
        count, err := strconv.Atoi(args[1])


@@ -46,6 +53,14 @@ func probe(ctx context.Context, args []string, out io.Writer) 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
+       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)
+               }
+       }
+

        entries, err := readProbeLog(args[0], count)
        if err != nil {
                return err
        entries, err := readProbeLog(args[0], count)
        if err != nil {
                return err


@@ -74,17 +89,29 @@ func probe(ctx context.Context, args []string, out io.Writer) error {
 
        usagePath := filepath.Join(root, "usage.jsonl")
        client := newClient(address, auth, "onechat-"+probeName, usagePath)
 
        usagePath := filepath.Join(root, "usage.jsonl")
        client := newClient(address, auth, "onechat-"+probeName, usagePath)

-       memory, err := openMemory(ctx, root, time.Now, newCompactor(client, probeName, config.Compactor))
+       client.LimitCalls(probeCallLimit)
+       summarizer := slotSummarizer{
+               inner: newCompactor(client, probeName, config.Compactor),
+               slots: make(chan struct{}, builds),
+       }
+
+       memory, err := openMemory(ctx, root, time.Now, summarizer)

        if err != nil {
                return err
        }
 
        if err != nil {
                return err
        }
 

-       _, err = fmt.Fprintf(out, "probe: %d messages into %s with %s at %s effort\n", len(entries), root, config.Compactor.Model, config.Compactor.Effort)
+       _, err = fmt.Fprintf(out, "probe: %d messages into %s with %s at %s effort, %d builds at once\n", len(entries), root, config.Compactor.Model, config.Compactor.Effort, builds)

        if err != nil {
                return errors.Join(err, memory.Close())
        }
 
        if err != nil {
                return errors.Join(err, memory.Close())
        }
 

-       appended, stop := runProbe(ctx, memory, entries, usagePath)
+       run := probeRun{
+               memory:    memory,
+               client:    client,
+               usagePath: usagePath,
+       }
+
+       appended, stop := run.Run(ctx, entries)

        sizes, err := modelLines(memory)
        err = errors.Join(err, memory.Close())
        if err != nil {
        sizes, err := modelLines(memory)
        err = errors.Join(err, memory.Close())
        if err != nil {


@@ -99,32 +126,113 @@ func probe(ctx context.Context, args []string, out io.Writer) error {
        return writeProbeReport(out, appended, stop, records, sizes)
 }
 
        return writeProbeReport(out, appended, stop, records, sizes)
 }
 

-// runProbe appends the entries and waits for the builds after each one. It
-// returns the number of messages appended and why the run stopped early, or
-// "" when it did not.
-func runProbe(ctx context.Context, memory *Memory, entries []probeEntry, usagePath string) (int, string) {
+// probeRun appends the messages of a probe run to its chat and stops the run
+// by the rules of probeStop.
+type probeRun struct {
+       memory    *Memory
+       client    *Client
+       usagePath string
+}
+
+// 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.
+func (r *probeRun) Run(ctx context.Context, entries []probeEntry) (int, string) {
+       appended, reason := r.append(ctx, entries)
+       if reason == "" {
+               return appended, ""
+       }
+
+       err := r.memory.Wait(ctx)
+       if err != nil {
+               reason += "; " + err.Error()
+       }
+
+       return appended, reason
+}
+
+// append appends the entries and stops at the first stop rule.
+func (r *probeRun) append(ctx context.Context, entries []probeEntry) (int, string) {

        for n, entry := range entries {
        for n, entry := range entries {

-               _, err := memory.Append(entry.Kind, entry.Text)
-               if err == nil {
-                       err = memory.Wait(ctx)
+               if entry.Kind == "user" {
+                       reason := r.wait(ctx)
+                       if reason != "" {
+                               return n, reason
+                       }

                }
 
                }
 

+               _, err := r.memory.Append(entry.Kind, entry.Text)

                if err != nil {
                        return n, err.Error()
                }
 
                if err != nil {
                        return n, err.Error()
                }
 

-               records, err := readUsageFile(usagePath)
-               if err != nil {
-                       return n + 1, err.Error()
+               reason := r.check()
+               if reason != "" {
+                       return n + 1, reason

                }
                }

+       }

 
 

-               reason := probeStop(records)
+       return len(entries), r.wait(ctx)
+}
+
+// wait returns when no build runs, or when a stop rule ends the run. It checks
+// the rules each second, so a run stops soon also during a long wait.
+func (r *probeRun) wait(ctx context.Context) string {
+       for {
+               reason := r.check()

                if reason != "" {
                if reason != "" {

-                       return n + 1, reason
+                       return reason
+               }
+
+               waitCtx, cancel := context.WithTimeout(ctx, time.Second)
+               err := r.memory.Wait(waitCtx)
+               cancel()
+               if err == nil {
+                       return r.check()
+               }
+
+               if ctx.Err() != nil {
+                       return ctx.Err().Error()

                }
        }
                }
        }

+}
+
+// check returns why the run must stop, or "" when it can go on.
+func (r *probeRun) check() string {
+       if r.client.LimitReached() {
+               return fmt.Sprintf("call limit of %d reached", r.client.limit)
+       }
+
+       records, err := readUsageFile(r.usagePath)
+       if err != nil {
+               return err.Error()
+       }

 
 

-       return len(entries), ""
+       return probeStop(records)
+}
+
+// slotSummarizer lets at most cap(slots) builds call inner at once. A build
+// past the limit waits for a slot.
+type slotSummarizer struct {
+       inner Summarizer
+       slots chan struct{}
+}
+
+// Summarize waits for a slot and builds the node with inner.
+func (s slotSummarizer) Summarize(ctx context.Context, task Task) (string, error) {
+       select {
+       case s.slots <- struct{}{}:
+       case <-ctx.Done():
+               return "", ctx.Err()
+       }
+
+       defer func() {
+               <-s.slots
+       }()
+
+       return s.inner.Summarize(ctx, task)

 }
 
 // probeEntry is one message of a probe log.
 }
 
 // probeEntry is one message of a probe log.


@@ -248,7 +356,9 @@ func builtByModel(memory *Memory, key nodeKey) (bool, error) {
        return !free, nil
 }
 
        return !free, nil
 }
 

-// readUsageFile reads the records of usage.jsonl at path.
+// readUsageFile reads the records of usage.jsonl at path, in the order of the
+// start of their calls. The file has them in the order of the end of their
+// calls, which differs when calls run at once.

 func readUsageFile(path string) (records []usageRecord, err error) {
        file, err := os.Open(path)
        if errors.Is(err, os.ErrNotExist) {
 func readUsageFile(path string) (records []usageRecord, err error) {
        file, err := os.Open(path)
        if errors.Is(err, os.ErrNotExist) {


@@ -280,9 +390,24 @@ func readUsageFile(path string) (records []usageRecord, err error) {
                return nil, fmt.Errorf("read usage file: %w", err)
        }
 
                return nil, fmt.Errorf("read usage file: %w", err)
        }
 

+       slices.SortStableFunc(records, func(a, b usageRecord) int {
+               return callStart(a).Compare(callStart(b))
+       })
+

        return records, nil
 }
 
        return records, nil
 }
 

+// callStart returns the start of the call of record, or the zero time when its
+// date is not valid.
+func callStart(record usageRecord) time.Time {
+       start, err := time.Parse(time.RFC3339Nano, record.Date)
+       if err != nil {
+               return time.Time{}
+       }
+
+       return start
+}
+

 // usedPercents returns the used percent of each quota window of record, by
 // header name.
 func usedPercents(record usageRecord) map[string]float64 {
 // usedPercents returns the used percent of each quota window of record, by
 // header name.
 func usedPercents(record usageRecord) map[string]float64 {


@@ -321,7 +446,7 @@ func writeProbeReport(out io.Writer, appended int, stop string, records []usageR
        lines = append(lines,
                fmt.Sprintf("messages: %d", appended),
                fmt.Sprintf("calls: %d (%d failed), %.2f per message", len(records), failed, ratio(len(records), appended)),
        lines = append(lines,
                fmt.Sprintf("messages: %d", appended),
                fmt.Sprintf("calls: %d (%d failed), %.2f per message", len(records), failed, ratio(len(records), appended)),

-               fmt.Sprintf("model lines: %d, retries: %d", len(sizes), len(records)-failed-len(sizes)),
+               fmt.Sprintf("model lines: %d, retries: %d, %.2f calls per model line", len(sizes), len(records)-failed-len(sizes), ratio(len(records), len(sizes))),

                fmt.Sprintf("tokens: %d input, %d cached (%.1f%%), %d output", input, cached, 100*ratio(cached, input), output),
        )
 
                fmt.Sprintf("tokens: %d input, %d cached (%.1f%%), %d output", input, cached, 100*ratio(cached, input), output),
        )
 


@@ -329,6 +454,10 @@ func writeProbeReport(out io.Writer, appended int, stop string, records []usageR
                lines = append(lines, sizeText(sizes))
        }
 
                lines = append(lines, sizeText(sizes))
        }
 

+       if len(records) > 0 {
+               lines = append(lines, timeText(records))
+       }
+

        if len(records) > 0 {
                lines = append(lines,
                        "quota at the first call: "+quotaText(records[0]),
        if len(records) > 0 {
                lines = append(lines,
                        "quota at the first call: "+quotaText(records[0]),


@@ -355,6 +484,55 @@ func sizeText(sizes []int) string {
        return fmt.Sprintf("line bytes: min %d, median %d, mean %.0f, max %d, %d over %d", sorted[0], sorted[len(sorted)/2], ratio(total, len(sorted)), sorted[len(sorted)-1], over, nodeBytes)
 }
 
        return fmt.Sprintf("line bytes: min %d, median %d, mean %.0f, max %d, %d over %d", sorted[0], sorted[len(sorted)/2], ratio(total, len(sorted)), sorted[len(sorted)-1], over, nodeBytes)
 }
 

+// timeText describes the time per call and how many calls ran at once. A call
+// runs from its date for its seconds.
+func timeText(records []usageRecord) string {
+       type event struct {
+               at    time.Time
+               delta int
+       }
+
+       var events []event
+       seconds := make([]float64, 0, len(records))
+       total := 0.0
+       for _, record := range records {
+               start, err := time.Parse(time.RFC3339Nano, record.Date)
+               if err != nil {
+                       continue
+               }
+
+               length := time.Duration(record.Seconds * float64(time.Second))
+               events = append(events, event{
+                       at:    start,
+                       delta: 1,
+               }, event{
+                       at:    start.Add(length),
+                       delta: -1,
+               })
+               seconds = append(seconds, record.Seconds)
+               total += record.Seconds
+       }
+
+       if len(events) == 0 {
+               return "time: no dated calls"
+       }
+
+       slices.SortFunc(events, func(a, b event) int {
+               return cmp.Or(a.at.Compare(b.at), cmp.Compare(a.delta, b.delta))
+       })
+
+       running, most := 0, 0
+       for _, e := range events {
+               running += e.delta
+               most = max(most, running)
+       }
+
+       wall := events[len(events)-1].at.Sub(events[0].at).Seconds()
+       slices.Sort(seconds)
+
+       return fmt.Sprintf("time: %.0f s in all, %.1f calls per minute, %.1f s per call (median %.1f, max %.1f), calls at once: at most %d, %.1f on average", wall, 60*float64(len(seconds))/max(wall, 1e-9), total/float64(len(seconds)), seconds[len(seconds)/2], seconds[len(seconds)-1], most, total/max(wall, 1e-9))
+}
+

 // quotaText returns the quota headers of record, sorted by name.
 func quotaText(record usageRecord) string {
        names := slices.Sorted(maps.Keys(record.Quota))
 // quotaText returns the quota headers of record, sorted by name.
 func quotaText(record usageRecord) string {
        names := slices.Sorted(maps.Keys(record.Quota))