From: Anderson A. Giacomolli Date: Fri, 6 Nov 2026 19:21:48 +0000 (-0300) Subject: Limit the probe calls at once and sort by start X-Git-Url: https://git.giacomolli.uk/onechat/commitdiff_plain/fe39a670d7af062af089bdba4dd07c3f0aee4048?ds=inline;hp=d1c01b959fb703e604e7cce4e49af411420c5300 Limit the probe calls at once and sort by start onechat probe takes an optional BUILDS argument from 1 to 8, which limits the builds that call the model at once in that run. A summarizer with that many slots wraps the compactor, so maxBuilds in the core stays 8. A lower limit lowers the rate of calls on one cache key, which tests whether the rate is the cause of the low cached share. The report adds the calls per minute. The probe reads the usage records in the order of the start of their calls. The file holds them in the order of their end, so with calls at once the quota stop compared the first record written with the last one, not the earliest call with the latest, and the report showed the same records as the first and the last call. Issue: b14d537 --- diff --git a/README.rst b/README.rst index 46de601..04dcafb 100644 --- a/README.rst +++ b/README.rst @@ -67,14 +67,16 @@ logs the other out. Probe ******* -``onechat probe LOG COUNT`` measures the compactor on a log. It appends -the first COUNT messages of LOG, one JSON object ``{"kind", "text"}`` on -each line, to a new chat in ``probe/`` in the data directory, and builds -its tree with real model calls. The calls use the quota of the account. -The run stops at the first failed call, or when the used percent of a -quota window grows by more than 15 points. At the end it reports the -calls per message, the cached share of the input tokens, the retries, -the line sizes, and the quota at the first and the last call. +``onechat probe LOG COUNT [BUILDS]`` measures the compactor on a log. It +appends the first COUNT messages of LOG, one JSON object ``{"kind", +"text"}`` on each line, to a new chat in ``probe/`` in the data +directory, and builds its tree with real model calls. The calls use the +quota of the account. The run stops at the first failed call, after 660 +calls, or when the used percent of a quota window grows by more than 15 +points. At the end it reports the calls per message, the cached share of +the input tokens, the retries, the line sizes, and the quota at the +first and the last call. BUILDS, from 1 to 8, limits the model calls at +once in the run. A test writes a generated log of readable made-up messages: diff --git a/probe.go b/probe.go index a1af30f..bbce24f 100644 --- a/probe.go +++ b/probe.go @@ -33,17 +33,19 @@ const ( 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. 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. +// 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 { - 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]) @@ -51,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]) } + 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 @@ -80,12 +90,17 @@ func probe(ctx context.Context, args []string, out io.Writer) error { usagePath := filepath.Join(root, "usage.jsonl") client := newClient(address, auth, "onechat-"+probeName, usagePath) client.LimitCalls(probeCallLimit) - memory, err := openMemory(ctx, root, time.Now, newCompactor(client, probeName, config.Compactor)) + 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 } - _, 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()) } @@ -198,6 +213,28 @@ func (r *probeRun) check() string { 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. type probeEntry struct { Kind string `json:"kind"` @@ -319,7 +356,9 @@ func builtByModel(memory *Memory, key nodeKey) (bool, error) { 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) { @@ -351,9 +390,24 @@ func readUsageFile(path string) (records []usageRecord, err error) { 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 } +// 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 { @@ -476,7 +530,7 @@ func timeText(records []usageRecord) string { wall := events[len(events)-1].at.Sub(events[0].at).Seconds() slices.Sort(seconds) - return fmt.Sprintf("time: %.0f s in all, %.1f s per call (median %.1f, max %.1f), calls at once: at most %d, %.1f on average", wall, total/float64(len(seconds)), seconds[len(seconds)/2], seconds[len(seconds)-1], most, total/max(wall, 1e-9)) + 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. diff --git a/probe_test.go b/probe_test.go index c1d08df..0651281 100644 --- a/probe_test.go +++ b/probe_test.go @@ -130,7 +130,7 @@ func TestProbeRunStopsWhenTheQuotaGrows(t *testing.T) { require.NoError(t, err) require.NoError(t, run.memory.Close()) - require.Contains(t, stop, "x-codex-primary-used-percent grew from 10 to") + require.Regexp(t, `^x-codex-primary-used-percent grew from \d+ to \d+$`, stop) require.Less(t, appended, probeMessages) require.NotEmpty(t, sizes) @@ -144,7 +144,7 @@ func TestProbeRunStopsWhenTheQuotaGrows(t *testing.T) { require.Contains(t, report.String(), "cached (80.0%)") require.Contains(t, report.String(), "line bytes: min 400, median 400, mean 400, max 400, 0 over 512") require.Contains(t, report.String(), "calls at once: at most ") - require.Contains(t, report.String(), "quota at the first call: primary-used-percent=10") + require.Contains(t, report.String(), "quota at the first call: primary-used-percent=") } func TestProbeRunStopsAtTheCallLimit(t *testing.T) { @@ -233,6 +233,52 @@ func TestProbeRunWaitsOnlyBeforeUserMessages(t *testing.T) { require.Equal(t, 3, server.Calls()) } +func TestSlotSummarizerLimitsTheBuildsAtOnce(t *testing.T) { + inner := newFakeSummarizer() + inner.gate = make(chan struct{}) + summarizer := slotSummarizer{ + inner: inner, + slots: make(chan struct{}, 2), + } + + var group sync.WaitGroup + for n := range 6 { + group.Go(func() { + _, err := summarizer.Summarize(t.Context(), Task{ + Level: 0, + Index: n, + Input: []string{"user: " + strings.Repeat("x", 600)}, + }) + require.NoError(t, err) + }) + } + + require.Eventually(t, func() bool { + return inner.Active() == 2 + }, 10*time.Second, time.Millisecond) + + close(inner.gate) + group.Wait() + require.Equal(t, 2, inner.peak) +} + +func TestReadUsageFileSortsByCallStart(t *testing.T) { + path := filepath.Join(t.TempDir(), "usage.jsonl") + usage := usageLog{ + path: path, + } + + for _, date := range []string{"2026-01-01T12:00:02Z", "2026-01-01T12:00:00.5Z", "2026-01-01T12:00:01.25Z"} { + require.NoError(t, usage.Write(usageRecord{ + Date: date, + })) + } + + records, err := readUsageFile(path) + require.NoError(t, err) + require.Equal(t, []string{"2026-01-01T12:00:00.5Z", "2026-01-01T12:00:01.25Z", "2026-01-01T12:00:02Z"}, []string{records[0].Date, records[1].Date, records[2].Date}) +} + func TestTimeTextCountsTheCallsAtOnce(t *testing.T) { start := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) record := func(offset, seconds float64) usageRecord { @@ -243,7 +289,7 @@ func TestTimeTextCountsTheCallsAtOnce(t *testing.T) { } text := timeText([]usageRecord{record(0, 10), record(2, 4), record(3, 2), record(10, 10)}) - require.Equal(t, "time: 20 s in all, 6.5 s per call (median 10.0, max 10.0), calls at once: at most 3, 1.3 on average", text) + require.Equal(t, "time: 20 s in all, 12.0 calls per minute, 6.5 s per call (median 10.0, max 10.0), calls at once: at most 3, 1.3 on average", text) } // probeLog returns count messages of made-up work on a small bakery order