feat(concurrency): provider-wide lens budget, drop the model cap
Concurrency was two multiplicative gates in two processes: entrypoint.sh capped MODELS-at-once per provider (GADFLY_PROVIDER_CONCURRENCY) while each model's binary separately capped its own lenses (GADFLY_LENS_CONCURRENCY). A model therefore held its whole model-slot until its LAST lens finished, stalling the next model even with idle lens capacity. Collapse to one throttle: a provider-wide lens budget shared across all of that provider's models. entrypoint now runs every model in a lane at once and seeds a single cross-process permit pool per lane (a dir of N flock files, sized by GADFLY_PROVIDER_LENS_CONCURRENCY -> GADFLY_LENS_CONCURRENCY). Each lens pass (review+recheck) acquires a permit before it runs and releases it after, so a model winding down immediately yields its freed permits to another model's queued lenses. flock auto-releases on process death, so a killed/crashed model can't leak budget. - cmd/gadfly/lenssem.go: the flock permit pool (+ lenssem_test.go). - main.go: runSpecialists holds a shared permit per lens; fanout sized to the budget so a lone model can use all of it. Falls back to the in-process limit when no pool is set (local runs, tests). - entrypoint.sh: drop provider_cap/DEFAULT_CONC; run_lane runs all models and seeds the per-lane pool. - GADFLY_PROVIDER_CONCURRENCY / GADFLY_CONCURRENCY are now ignored; the reusable workflow marks provider_concurrency deprecated and stops forwarding it. Docs (README, CLAUDE.md, examples) updated per the maintenance rule. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
@@ -44,7 +44,7 @@ on:
|
|||||||
#
|
#
|
||||||
# Owner-set user-scope variables (see README "Central config via variables"):
|
# Owner-set user-scope variables (see README "Central config via variables"):
|
||||||
# GADFLY_DEFAULT_MODELS, GADFLY_DEFAULT_SPECIALISTS,
|
# GADFLY_DEFAULT_MODELS, GADFLY_DEFAULT_SPECIALISTS,
|
||||||
# GADFLY_DEFAULT_PROVIDER_CONCURRENCY, GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY,
|
# GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY (the provider-wide lens budget),
|
||||||
# GADFLY_ENDPOINT_NETHERSTORM (the local GPU box endpoint).
|
# GADFLY_ENDPOINT_NETHERSTORM (the local GPU box endpoint).
|
||||||
# An unset variable + no input → the image default (one model, default suite),
|
# An unset variable + no input → the image default (one model, default suite),
|
||||||
# so a public consumer with neither still gets a sane minimal review.
|
# so a public consumer with neither still gets a sane minimal review.
|
||||||
@@ -53,8 +53,8 @@ on:
|
|||||||
specialists: { type: string, default: "" } # GADFLY_SPECIALISTS — empty falls back to user var GADFLY_DEFAULT_SPECIALISTS
|
specialists: { type: string, default: "" } # GADFLY_SPECIALISTS — empty falls back to user var GADFLY_DEFAULT_SPECIALISTS
|
||||||
provider: { type: string, default: "" } # GADFLY_PROVIDER
|
provider: { type: string, default: "" } # GADFLY_PROVIDER
|
||||||
base_url: { type: string, default: "" } # GADFLY_BASE_URL
|
base_url: { type: string, default: "" } # GADFLY_BASE_URL
|
||||||
provider_concurrency: { type: string, default: "" } # GADFLY_PROVIDER_CONCURRENCY — empty falls back to user var GADFLY_DEFAULT_PROVIDER_CONCURRENCY
|
provider_concurrency: { type: string, default: "" } # DEPRECATED / ignored — the per-provider MODEL cap was removed; the lens budget below is the single throttle. Kept so existing callers don't error.
|
||||||
provider_lens_concurrency: { type: string, default: "" } # GADFLY_PROVIDER_LENS_CONCURRENCY — empty falls back to user var GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY
|
provider_lens_concurrency: { type: string, default: "" } # GADFLY_PROVIDER_LENS_CONCURRENCY — the per-provider lens budget (shared across the provider's models); empty falls back to user var GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY
|
||||||
timeout_secs: { type: string, default: "600" } # GADFLY_TIMEOUT_SECS (per lens)
|
timeout_secs: { type: string, default: "600" } # GADFLY_TIMEOUT_SECS (per lens)
|
||||||
max_steps: { type: string, default: "14" } # GADFLY_MAX_STEPS
|
max_steps: { type: string, default: "14" } # GADFLY_MAX_STEPS
|
||||||
worker_model: { type: string, default: "" } # GADFLY_WORKER_MODEL
|
worker_model: { type: string, default: "" } # GADFLY_WORKER_MODEL
|
||||||
@@ -151,7 +151,9 @@ jobs:
|
|||||||
GADFLY_SPECIALISTS: ${{ inputs.specialists || vars.GADFLY_DEFAULT_SPECIALISTS }}
|
GADFLY_SPECIALISTS: ${{ inputs.specialists || vars.GADFLY_DEFAULT_SPECIALISTS }}
|
||||||
GADFLY_PROVIDER: ${{ inputs.provider }}
|
GADFLY_PROVIDER: ${{ inputs.provider }}
|
||||||
GADFLY_BASE_URL: ${{ inputs.base_url }}
|
GADFLY_BASE_URL: ${{ inputs.base_url }}
|
||||||
GADFLY_PROVIDER_CONCURRENCY: ${{ inputs.provider_concurrency || vars.GADFLY_DEFAULT_PROVIDER_CONCURRENCY }}
|
# NB: GADFLY_PROVIDER_CONCURRENCY (the old model cap) is intentionally no
|
||||||
|
# longer forwarded — entrypoint.sh ignores it. The lens budget is the one
|
||||||
|
# throttle now, shared across a provider's models.
|
||||||
GADFLY_PROVIDER_LENS_CONCURRENCY: ${{ inputs.provider_lens_concurrency || vars.GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY }}
|
GADFLY_PROVIDER_LENS_CONCURRENCY: ${{ inputs.provider_lens_concurrency || vars.GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY }}
|
||||||
GADFLY_TIMEOUT_SECS: ${{ inputs.timeout_secs }}
|
GADFLY_TIMEOUT_SECS: ${{ inputs.timeout_secs }}
|
||||||
GADFLY_MAX_STEPS: ${{ inputs.max_steps }}
|
GADFLY_MAX_STEPS: ${{ inputs.max_steps }}
|
||||||
|
|||||||
@@ -57,7 +57,7 @@ Dockerfile multi-stage; private-module creds via BuildKit secrets ne
|
|||||||
.gitea/workflows/build-image.yml push main → :latest; tag v* → :<tag>+:latest; PR → build-only
|
.gitea/workflows/build-image.yml push main → :latest; tag v* → :<tag>+:latest; PR → build-only
|
||||||
.gitea/workflows/review-reusable.yml reusable (workflow_call) review job; resolves swarm config at
|
.gitea/workflows/review-reusable.yml reusable (workflow_call) review job; resolves swarm config at
|
||||||
RUNTIME: consumer `with:` input → owner user-scope var (GADFLY_DEFAULT_MODELS /
|
RUNTIME: consumer `with:` input → owner user-scope var (GADFLY_DEFAULT_MODELS /
|
||||||
_SPECIALISTS / _PROVIDER_CONCURRENCY / _PROVIDER_LENS_CONCURRENCY, +
|
_SPECIALISTS / _PROVIDER_LENS_CONCURRENCY, +
|
||||||
GADFLY_ENDPOINT_RAGNAROS) → image default. Vars are injected per-run, so editing
|
GADFLY_ENDPOINT_RAGNAROS) → image default. Vars are injected per-run, so editing
|
||||||
one var retunes the whole fleet even though long-lived act_runners CACHE this file
|
one var retunes the whole fleet even though long-lived act_runners CACHE this file
|
||||||
by ref (a moved tag is NOT re-fetched — only a runtime value or a fresh @<sha>
|
by ref (a moved tag is NOT re-fetched — only a runtime value or a fresh @<sha>
|
||||||
@@ -154,10 +154,19 @@ are actually exercised. OpenAI/Anthropic/Google come from majordomo's abstractio
|
|||||||
NOT re-pulled, so the job silently runs the previous image. For a run that must use a specific
|
NOT re-pulled, so the job silently runs the previous image. For a run that must use a specific
|
||||||
build (e.g. validating a just-pushed fix), pin the consumer stub to the immutable
|
build (e.g. validating a just-pushed fix), pin the consumer stub to the immutable
|
||||||
`:sha-<short>` tag the build publishes, not `:latest`.
|
`:sha-<short>` tag the build publishes, not `:latest`.
|
||||||
- **Concurrency is per-provider** (`entrypoint.sh`): each provider is a lane, lanes run in
|
- **Concurrency is one per-provider lens budget** (`entrypoint.sh` + `cmd/gadfly/lenssem.go`):
|
||||||
parallel, `cap` (from `GADFLY_PROVIDER_CONCURRENCY` else `GADFLY_CONCURRENCY`, default 1) bounds
|
each provider is a lane, lanes run in parallel, and within a lane ALL of the provider's models
|
||||||
models-at-once within a lane. The review timeout (`GADFLY_TIMEOUT_SECS`) is **per-lens**, not
|
run at once — the only throttle is a **provider-wide lens budget** (max lens passes in flight,
|
||||||
shared across the suite — a slow model can't starve later lenses (the original timeout bug).
|
a lens = one specialist's review+recheck). The budget comes from
|
||||||
|
`GADFLY_PROVIDER_LENS_CONCURRENCY` (`provider=N` map) else `GADFLY_LENS_CONCURRENCY` (default 1).
|
||||||
|
Because models are separate processes, the budget is a **cross-process permit pool**: entrypoint
|
||||||
|
seeds a per-lane dir of N flock files; each model's binary acquires one before a lens pass and
|
||||||
|
releases it after (flock auto-drops on process death). This replaced the old two-level
|
||||||
|
`GADFLY_PROVIDER_CONCURRENCY` MODEL cap × per-model `GADFLY_LENS_CONCURRENCY`, which multiplied
|
||||||
|
and let a model hold its slot through its last lens — stalling the next model with idle lens
|
||||||
|
capacity. Those two model-cap vars are now **ignored**. The review timeout
|
||||||
|
(`GADFLY_TIMEOUT_SECS`) is **per-lens**, not shared across the suite — a slow lens can't starve
|
||||||
|
the others (the original timeout bug).
|
||||||
- **Large-PR token burn**: the agent loop re-sends the whole transcript every step, so a giant
|
- **Large-PR token burn**: the agent loop re-sends the whole transcript every step, so a giant
|
||||||
diff (the old `get_diff` dumped it untruncated, and it was embedded in both the review and
|
diff (the old `get_diff` dumped it untruncated, and it was embedded in both the review and
|
||||||
recheck task) was re-transmitted ~steps × lenses × passes × models times — a ~250 K-token PR
|
recheck task) was re-transmitted ~steps × lenses × passes × models times — a ~250 K-token PR
|
||||||
|
|||||||
@@ -272,38 +272,32 @@ Unset = no delegation (current behavior).
|
|||||||
### Concurrency (per-provider lanes)
|
### Concurrency (per-provider lanes)
|
||||||
|
|
||||||
With multiple models, each **provider** is its own lane and lanes run in **parallel**, so a fast
|
With multiple models, each **provider** is its own lane and lanes run in **parallel**, so a fast
|
||||||
cloud provider isn't stuck behind a slow local box. Within a lane, at most `cap` models run at
|
cloud provider isn't stuck behind a slow local box. There is **one throttle**: a per-provider
|
||||||
once — `cap` comes from `GADFLY_PROVIDER_CONCURRENCY` (a `provider=N` map) else `GADFLY_CONCURRENCY`
|
**lens budget** — the max number of lens passes (a lens = one specialist's review+recheck) in
|
||||||
(default `1`). The timeout is **per-lens** (`GADFLY_TIMEOUT_SECS`), so a slow model on one lens
|
flight at once for that provider. Every model in the lane runs concurrently and its lenses draw
|
||||||
can't starve the others.
|
from that single shared budget, so nothing else caps how many models run. The budget comes from
|
||||||
|
`GADFLY_PROVIDER_LENS_CONCURRENCY` (a `provider=N` map) else the `GADFLY_LENS_CONCURRENCY` scalar
|
||||||
|
(default `1`). The timeout is **per-lens** (`GADFLY_TIMEOUT_SECS`), so a slow lens can't starve
|
||||||
|
the others.
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
# One local box (serial — it serves one model at a time) + 3 cloud reviews at once,
|
# The local box gets 1 lens at a time (serial); the cloud lane runs up to 3 lens passes at once,
|
||||||
# both lanes running concurrently:
|
# shared across ALL its models. Both lanes run concurrently.
|
||||||
GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=3,m1pro=1"
|
GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1pro=1"
|
||||||
GADFLY_MODELS: "m1pro/qwen3:14b,qwen3-coder:480b-cloud,gpt-oss:120b-cloud"
|
GADFLY_MODELS: "m1pro/qwen3:14b,qwen3-coder:480b-cloud,gpt-oss:120b-cloud"
|
||||||
```
|
```
|
||||||
|
|
||||||
A model's provider is the spec's first segment (`m1pro/…` → `m1pro`), or `GADFLY_PROVIDER`/
|
A model's provider is the spec's first segment (`m1pro/…` → `m1pro`), or `GADFLY_PROVIDER`/
|
||||||
`ollama-cloud` for a bare id. Default (`cap 1`) keeps a single-provider pool fully sequential.
|
`ollama-cloud` for a bare id. The budget is **shared across the provider's models**: with a
|
||||||
|
budget of 3 and two cloud models, you get 3 lens passes in flight in any mix — as one model
|
||||||
|
finishes a lens, the freed slot immediately goes to another model's next lens, so a model
|
||||||
|
winding down to its last lens never stalls the others (the pre-2026-07 design capped *models*
|
||||||
|
separately and did stall — that `GADFLY_PROVIDER_CONCURRENCY`/`GADFLY_CONCURRENCY` model cap is
|
||||||
|
**gone**; those vars are now ignored). Default (budget `1`) keeps a provider fully sequential.
|
||||||
|
|
||||||
**Lens fan-out (within a model).** By default the specialist lenses run **sequentially** inside
|
> Under the hood the shared budget is a small cross-process permit pool (flock files, seeded per
|
||||||
each model (`GADFLY_LENS_CONCURRENCY=1`). Raise it to overlap the independent per-lens
|
> lane by `entrypoint.sh`); permits release automatically if a model process dies, so a crashed
|
||||||
review+recheck passes — the model then posts its consolidated comment as soon as its lenses
|
> lens can't leak budget.
|
||||||
finish (so with sequential models, results stream in per model and per-model timings stay
|
|
||||||
clean). Like the model cap, it's **per-provider configurable**: `GADFLY_PROVIDER_LENS_CONCURRENCY`
|
|
||||||
takes a `provider=N` map keyed by the **same provider lanes** as `GADFLY_PROVIDER_CONCURRENCY`,
|
|
||||||
falling back to the `GADFLY_LENS_CONCURRENCY` scalar (default `1`). **It multiplies with the
|
|
||||||
model cap:** total in-flight requests ≈ *models-at-once × lenses-at-once*, so to fan lenses out
|
|
||||||
without oversubscribing a backend, keep its model cap low and raise its lens cap:
|
|
||||||
|
|
||||||
```yaml
|
|
||||||
# Per provider: cloud runs one model at a time but fans its 3 lenses out (3 concurrent requests);
|
|
||||||
# the slow local box stays fully serial. Both provider lanes still run in parallel.
|
|
||||||
GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=1,m1=1"
|
|
||||||
GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1=1"
|
|
||||||
GADFLY_SPECIALISTS: "security,correctness,error-handling"
|
|
||||||
```
|
|
||||||
|
|
||||||
### Live status board
|
### Live status board
|
||||||
|
|
||||||
@@ -418,8 +412,7 @@ on its next review **without** a re-pin or a tag move:
|
|||||||
|---|---|
|
|---|---|
|
||||||
| `GADFLY_DEFAULT_MODELS` | `GADFLY_MODELS` (csv) |
|
| `GADFLY_DEFAULT_MODELS` | `GADFLY_MODELS` (csv) |
|
||||||
| `GADFLY_DEFAULT_SPECIALISTS` | the lens suite |
|
| `GADFLY_DEFAULT_SPECIALISTS` | the lens suite |
|
||||||
| `GADFLY_DEFAULT_PROVIDER_CONCURRENCY` | models-at-once per provider |
|
| `GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY` | the per-provider lens budget (lens passes in flight per provider, shared across its models) |
|
||||||
| `GADFLY_DEFAULT_PROVIDER_LENS_CONCURRENCY` | lenses-at-once per provider |
|
|
||||||
| `GADFLY_ENDPOINT_RAGNAROS` | a named endpoint, e.g. `llamaswap\|https://host` |
|
| `GADFLY_ENDPOINT_RAGNAROS` | a named endpoint, e.g. `llamaswap\|https://host` |
|
||||||
|
|
||||||
Adding a *new* named endpoint still needs a one-line reusable edit (Gitea can't auto-expose arbitrary
|
Adding a *new* named endpoint still needs a one-line reusable edit (Gitea can't auto-expose arbitrary
|
||||||
@@ -441,10 +434,9 @@ The reviewer binary reads these (the stub/entrypoint set sane defaults):
|
|||||||
| `GADFLY_SELECTOR_MODEL` | review model | model that picks lenses in `auto` mode |
|
| `GADFLY_SELECTOR_MODEL` | review model | model that picks lenses in `auto` mode |
|
||||||
| `GADFLY_WORKER_MODEL` | — | cheap model for `delegate_investigation`; unset = no delegation |
|
| `GADFLY_WORKER_MODEL` | — | cheap model for `delegate_investigation`; unset = no delegation |
|
||||||
| `GADFLY_WORKER_MAX_STEPS` | 8 | tool-step cap for a delegated worker run |
|
| `GADFLY_WORKER_MAX_STEPS` | 8 | tool-step cap for a delegated worker run |
|
||||||
| `GADFLY_CONCURRENCY` | 1 | default max models run at once **per provider** |
|
| `GADFLY_LENS_CONCURRENCY` | 1 | **per-provider lens budget** — lens passes in flight per provider, shared across all its models (all a provider's models run at once; this is the only throttle) |
|
||||||
| `GADFLY_PROVIDER_CONCURRENCY` | — | per-provider overrides, e.g. `ollama-cloud=3,m1pro=1` |
|
| `GADFLY_PROVIDER_LENS_CONCURRENCY` | — | per-provider lens-budget overrides, a `provider=N` map, e.g. `ollama-cloud=3,m1=1` |
|
||||||
| `GADFLY_LENS_CONCURRENCY` | 1 | specialist lenses run at once **within a model** (× model cap = total in-flight) |
|
| `GADFLY_CONCURRENCY` / `GADFLY_PROVIDER_CONCURRENCY` | — | **removed** (was the per-provider models-at-once cap; now ignored — the lens budget is the single throttle) |
|
||||||
| `GADFLY_PROVIDER_LENS_CONCURRENCY` | — | per-provider lens overrides, same lanes as `GADFLY_PROVIDER_CONCURRENCY`, e.g. `ollama-cloud=3,m1=1` |
|
|
||||||
| `GADFLY_MAX_STEPS` | 24 | review-pass tool-step cap |
|
| `GADFLY_MAX_STEPS` | 24 | review-pass tool-step cap |
|
||||||
| `GADFLY_TIMEOUT_SECS` | 300 | deadline **per specialist lens** (review+recheck) |
|
| `GADFLY_TIMEOUT_SECS` | 300 | deadline **per specialist lens** (review+recheck) |
|
||||||
| `GADFLY_RECHECK` | on | set `0`/`false` to skip the recheck pass |
|
| `GADFLY_RECHECK` | on | set `0`/`false` to skip the recheck pass |
|
||||||
|
|||||||
@@ -0,0 +1,86 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"syscall"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// lensSem is the PROVIDER-WIDE lens-permit pool. Historically each model's binary
|
||||||
|
// throttled its own lenses in-process (GADFLY_LENS_CONCURRENCY) while entrypoint.sh
|
||||||
|
// separately capped how many MODELS ran at once — two multiplicative gates in
|
||||||
|
// different processes. That let a model hold its whole slot until its last lens
|
||||||
|
// finished, stalling the next model even with idle lens capacity.
|
||||||
|
//
|
||||||
|
// Instead, entrypoint.sh now runs all of a provider's models concurrently and
|
||||||
|
// seeds ONE directory of N permit files per provider (N = the provider's lens
|
||||||
|
// budget). Every model process in that lane draws from the same pool: a lens pass
|
||||||
|
// (review + recheck) acquires a permit before it runs and releases it after, so a
|
||||||
|
// model winding down immediately yields its freed permits to another model's
|
||||||
|
// queued lenses. Permits are held with flock, which the kernel drops when the
|
||||||
|
// holding process exits — so a killed/crashed model frees its permits for free.
|
||||||
|
type lensSem struct {
|
||||||
|
dir string
|
||||||
|
size int
|
||||||
|
}
|
||||||
|
|
||||||
|
// lensSemPollInterval is how often a blocked acquirer re-sweeps the permit files.
|
||||||
|
// Lens passes run for many seconds to minutes, so a coarse poll adds negligible
|
||||||
|
// latency while keeping the mechanism a few lines of stdlib (no IPC primitives).
|
||||||
|
const lensSemPollInterval = 150 * time.Millisecond
|
||||||
|
|
||||||
|
// activeLensSem returns the shared semaphore configured by entrypoint.sh, or nil
|
||||||
|
// when it isn't set (standalone/local runs, tests, or an older entrypoint) — in
|
||||||
|
// which case runSpecialists falls back to the in-process fanout limit alone, i.e.
|
||||||
|
// the pre-existing per-model behavior.
|
||||||
|
func activeLensSem() *lensSem {
|
||||||
|
dir := os.Getenv("GADFLY_LENS_SEM_DIR")
|
||||||
|
if dir == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
n := envInt("GADFLY_LENS_SEM_SIZE", 0)
|
||||||
|
if n < 1 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return &lensSem{dir: dir, size: n}
|
||||||
|
}
|
||||||
|
|
||||||
|
// acquire blocks until a permit is free (or ctx is done) and returns a release
|
||||||
|
// func. On cancellation it returns a no-op release plus ctx.Err(), so callers can
|
||||||
|
// surface the lens as "did not run" rather than leaking a permit.
|
||||||
|
func (s *lensSem) acquire(ctx context.Context) (func(), error) {
|
||||||
|
for {
|
||||||
|
if release, ok := s.tryAcquire(); ok {
|
||||||
|
return release, nil
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return func() {}, ctx.Err()
|
||||||
|
case <-time.After(lensSemPollInterval):
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// tryAcquire makes one non-blocking sweep over the permit files, returning a
|
||||||
|
// release func for the first one it locks. The permit files are created lazily
|
||||||
|
// (entrypoint.sh only guarantees the directory exists), so a fresh pool needs no
|
||||||
|
// seeding step beyond mkdir. Closing the *os.File in the returned func drops the
|
||||||
|
// flock (the lock lives on the open file description).
|
||||||
|
func (s *lensSem) tryAcquire() (func(), bool) {
|
||||||
|
for i := 0; i < s.size; i++ {
|
||||||
|
path := filepath.Join(s.dir, fmt.Sprintf("permit.%d", i))
|
||||||
|
f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
|
||||||
|
f.Close()
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
return func() { f.Close() }, true
|
||||||
|
}
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
@@ -0,0 +1,110 @@
|
|||||||
|
package main
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// flock permits are per open-file-description, so two separate opens of the same
|
||||||
|
// permit file conflict even within one process — the exhaustion, blocking, and
|
||||||
|
// max-in-flight tests below therefore exercise the same semantics a real
|
||||||
|
// multi-process lane would see.
|
||||||
|
|
||||||
|
func TestActiveLensSemUnset(t *testing.T) {
|
||||||
|
t.Setenv("GADFLY_LENS_SEM_DIR", "")
|
||||||
|
if s := activeLensSem(); s != nil {
|
||||||
|
t.Fatalf("expected nil sem when GADFLY_LENS_SEM_DIR unset, got %+v", s)
|
||||||
|
}
|
||||||
|
// A dir with a zero/blank size is inert too (must not divide the pool by 0).
|
||||||
|
t.Setenv("GADFLY_LENS_SEM_DIR", t.TempDir())
|
||||||
|
t.Setenv("GADFLY_LENS_SEM_SIZE", "0")
|
||||||
|
if s := activeLensSem(); s != nil {
|
||||||
|
t.Fatalf("expected nil sem when size < 1, got %+v", s)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLensSemExhaustionAndRelease(t *testing.T) {
|
||||||
|
s := &lensSem{dir: t.TempDir(), size: 2}
|
||||||
|
|
||||||
|
r1, ok := s.tryAcquire()
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("first acquire should succeed")
|
||||||
|
}
|
||||||
|
r2, ok := s.tryAcquire()
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("second acquire should succeed")
|
||||||
|
}
|
||||||
|
if _, ok := s.tryAcquire(); ok {
|
||||||
|
t.Fatal("third acquire should fail: pool of 2 is exhausted")
|
||||||
|
}
|
||||||
|
|
||||||
|
r1() // free one permit
|
||||||
|
r3, ok := s.tryAcquire()
|
||||||
|
if !ok {
|
||||||
|
t.Fatal("acquire should succeed after a release")
|
||||||
|
}
|
||||||
|
r2()
|
||||||
|
r3()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLensSemAcquireBlocksThenCancels(t *testing.T) {
|
||||||
|
s := &lensSem{dir: t.TempDir(), size: 1}
|
||||||
|
|
||||||
|
// Immediate success while a permit is free.
|
||||||
|
release, err := s.acquire(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("acquire on a free pool: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// With the only permit held, acquire must block until ctx expires.
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||||
|
defer cancel()
|
||||||
|
start := time.Now()
|
||||||
|
if _, err := s.acquire(ctx); err == nil {
|
||||||
|
t.Fatal("acquire on an exhausted pool should return ctx error, not a permit")
|
||||||
|
}
|
||||||
|
if waited := time.Since(start); waited < 50*time.Millisecond {
|
||||||
|
t.Fatalf("acquire returned too fast (%v); it should have blocked on the full pool", waited)
|
||||||
|
}
|
||||||
|
release()
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLensSemNeverExceedsSize(t *testing.T) {
|
||||||
|
const size = 3
|
||||||
|
s := &lensSem{dir: t.TempDir(), size: size}
|
||||||
|
|
||||||
|
var inFlight, peak int64
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
for i := 0; i < 12; i++ {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
release, err := s.acquire(context.Background())
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("acquire: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer release()
|
||||||
|
n := atomic.AddInt64(&inFlight, 1)
|
||||||
|
for {
|
||||||
|
p := atomic.LoadInt64(&peak)
|
||||||
|
if n <= p || atomic.CompareAndSwapInt64(&peak, p, n) {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
atomic.AddInt64(&inFlight, -1)
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
|
if peak > size {
|
||||||
|
t.Fatalf("max in-flight lens permits = %d, exceeds budget %d", peak, size)
|
||||||
|
}
|
||||||
|
if peak == 0 {
|
||||||
|
t.Fatal("no permits were ever acquired")
|
||||||
|
}
|
||||||
|
}
|
||||||
+46
-23
@@ -49,15 +49,18 @@
|
|||||||
// GADFLY_RECHECK set to 0/false to skip the recheck pass (optional, default on).
|
// GADFLY_RECHECK set to 0/false to skip the recheck pass (optional, default on).
|
||||||
// GADFLY_RECHECK_MAX_STEPS recheck-pass step cap (optional, default 16).
|
// GADFLY_RECHECK_MAX_STEPS recheck-pass step cap (optional, default 16).
|
||||||
// GADFLY_TIMEOUT_SECS overall deadline in seconds, shared by both passes (optional, default 300).
|
// GADFLY_TIMEOUT_SECS overall deadline in seconds, shared by both passes (optional, default 300).
|
||||||
// GADFLY_LENS_CONCURRENCY how many specialist lenses run concurrently within this
|
// GADFLY_LENS_CONCURRENCY how many specialist lenses run concurrently (optional,
|
||||||
// model (optional, default 1 = sequential). Total in-flight
|
// default 1 = sequential). Under entrypoint.sh this is the
|
||||||
// model requests ≈ this × entrypoint.sh's per-provider model
|
// PROVIDER-WIDE lens budget, shared across all of that
|
||||||
// concurrency, so keep the product within the backend's budget.
|
// provider's models via a permit pool (GADFLY_LENS_SEM_DIR),
|
||||||
|
// so it bounds total lens passes in flight per provider.
|
||||||
// GADFLY_PROVIDER_LENS_CONCURRENCY per-provider override for the above, as a
|
// GADFLY_PROVIDER_LENS_CONCURRENCY per-provider override for the above, as a
|
||||||
// "provider=N,provider=N" map keyed by the SAME provider
|
// "provider=N,provider=N" map keyed by the provider lanes
|
||||||
// lanes as GADFLY_PROVIDER_CONCURRENCY (e.g.
|
// (e.g. "ollama-cloud=3,m1=1"). Wins over
|
||||||
// "ollama-cloud=3,m1=1"). Wins over GADFLY_LENS_CONCURRENCY
|
// GADFLY_LENS_CONCURRENCY for the model's provider.
|
||||||
// for the model's provider; falls back to it otherwise.
|
// GADFLY_LENS_SEM_DIR / GADFLY_LENS_SEM_SIZE set by entrypoint.sh: the shared
|
||||||
|
// cross-process lens-permit pool (dir of N flock files)
|
||||||
|
// and its size. Unset => in-process lensConcurrency only.
|
||||||
// GADFLY_MAX_DIFF_CHARS diff chars embedded in the review prompt (optional, default 60000;
|
// GADFLY_MAX_DIFF_CHARS diff chars embedded in the review prompt (optional, default 60000;
|
||||||
// the full diff is reachable via the paginated get_diff tool).
|
// the full diff is reachable via the paginated get_diff tool).
|
||||||
//
|
//
|
||||||
@@ -248,19 +251,29 @@ func run() error {
|
|||||||
// own per-lens timeout (reviewWithSpecialist) and the lenses only read the
|
// own per-lens timeout (reviewWithSpecialist) and the lenses only read the
|
||||||
// immutable repoFS, so concurrency simply overlaps independent passes.
|
// immutable repoFS, so concurrency simply overlaps independent passes.
|
||||||
//
|
//
|
||||||
// Caution: this fans out WITHIN one model. It multiplies with entrypoint.sh's
|
// Provider-wide throttling: when entrypoint.sh runs several of a provider's
|
||||||
// per-provider model concurrency, so total concurrent backend requests ≈
|
// models at once it seeds a shared lens-permit pool (activeLensSem) that every
|
||||||
// (models at once) × (lenses at once). To fan lenses out without oversubscribing
|
// model's lenses draw from, so the real cap is total lens passes in flight per
|
||||||
// the backend, run models one at a time (provider lane cap 1) and raise this.
|
// provider — not (models at once) × (lenses at once). Absent that pool (local
|
||||||
|
// runs, tests) this falls back to the in-process lensConcurrency() limit alone.
|
||||||
func runSpecialists(eng reviewEngine, base string, specialists []Specialist, task, diff string) []specialistResult {
|
func runSpecialists(eng reviewEngine, base string, specialists []Specialist, task, diff string) []specialistResult {
|
||||||
// Optional live status board: publishes this model's per-lens progress to a
|
// Optional live status board: publishes this model's per-lens progress to a
|
||||||
// file the entrypoint board renders. Inert (no-op) unless GADFLY_STATUS_FILE
|
// file the entrypoint board renders. Inert (no-op) unless GADFLY_STATUS_FILE
|
||||||
// is set, so plain runs are unaffected.
|
// is set, so plain runs are unaffected.
|
||||||
sw := newStatusWriter(os.Getenv("GADFLY_MODEL"), modelProvider(), specialists)
|
sw := newStatusWriter(os.Getenv("GADFLY_MODEL"), modelProvider(), specialists)
|
||||||
|
|
||||||
|
// The cross-process pool (if any) is the real ceiling; size the in-process
|
||||||
|
// fanout to it so a lone model in its lane can use the whole provider budget,
|
||||||
|
// while extra goroutines simply block in sem.acquire until a permit frees.
|
||||||
|
sem := activeLensSem()
|
||||||
|
maxConcurrent := lensConcurrency()
|
||||||
|
if sem != nil {
|
||||||
|
maxConcurrent = sem.size
|
||||||
|
}
|
||||||
|
|
||||||
fanResults := fanout.Run(context.Background(), specialists, fanout.Options[Specialist]{
|
fanResults := fanout.Run(context.Background(), specialists, fanout.Options[Specialist]{
|
||||||
MaxConcurrent: lensConcurrency(),
|
MaxConcurrent: maxConcurrent,
|
||||||
}, func(_ context.Context, sp Specialist) (res specialistResult, _ error) {
|
}, func(ctx context.Context, sp Specialist) (res specialistResult, _ error) {
|
||||||
// A panic in one lens must not crash the whole binary (which would kill
|
// A panic in one lens must not crash the whole binary (which would kill
|
||||||
// every other lens's output) or leave this lens stuck at "running" on the
|
// every other lens's output) or leave this lens stuck at "running" on the
|
||||||
// status board. fanout does not recover fn panics, so we do it here:
|
// status board. fanout does not recover fn panics, so we do it here:
|
||||||
@@ -271,6 +284,17 @@ func runSpecialists(eng reviewEngine, base string, specialists []Specialist, tas
|
|||||||
sw.set(sp.Name, lensFinished, "", true)
|
sw.set(sp.Name, lensFinished, "", true)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
// Hold a provider-wide permit for this lens's whole review+recheck pass.
|
||||||
|
// While waiting the lens stays "queued" on the board; cancellation before
|
||||||
|
// a permit frees surfaces as a did-not-run lens rather than a leaked slot.
|
||||||
|
if sem != nil {
|
||||||
|
release, err := sem.acquire(ctx)
|
||||||
|
if err != nil {
|
||||||
|
sw.set(sp.Name, lensFinished, "", true)
|
||||||
|
return specialistResult{spec: sp, out: fmt.Sprintf("⚠️ This reviewer did not run: %v", err), verdict: verdictUnknown, errored: true}, nil
|
||||||
|
}
|
||||||
|
defer release()
|
||||||
|
}
|
||||||
sw.set(sp.Name, lensRunning, "", false)
|
sw.set(sp.Name, lensRunning, "", false)
|
||||||
out, errored := reviewWithSpecialist(eng, base, sp, task, diff)
|
out, errored := reviewWithSpecialist(eng, base, sp, task, diff)
|
||||||
v := parseVerdict(out)
|
v := parseVerdict(out)
|
||||||
@@ -293,14 +317,13 @@ func runSpecialists(eng reviewEngine, base string, specialists []Specialist, tas
|
|||||||
return results
|
return results
|
||||||
}
|
}
|
||||||
|
|
||||||
// lensConcurrency resolves how many specialist lenses run at once for THIS run's
|
// lensConcurrency resolves the lens budget for THIS run's provider: a per-provider
|
||||||
// model. It mirrors entrypoint.sh's per-provider MODEL concurrency: a
|
// override in GADFLY_PROVIDER_LENS_CONCURRENCY ("provider=N,...") wins for the
|
||||||
// per-provider override in GADFLY_PROVIDER_LENS_CONCURRENCY ("provider=N,...")
|
// model's provider (resolved by modelProvider()), otherwise the
|
||||||
// wins for the model's provider, otherwise the GADFLY_LENS_CONCURRENCY scalar
|
// GADFLY_LENS_CONCURRENCY scalar (default 1). Under entrypoint.sh the SAME value
|
||||||
// (default 1). The provider is resolved by modelProvider() — the SAME lane rule
|
// seeds the shared cross-process permit pool (activeLensSem), so it is the
|
||||||
// entrypoint uses for GADFLY_PROVIDER_CONCURRENCY — so e.g.
|
// provider-wide budget rather than a per-model one; standalone it caps the single
|
||||||
// "ollama-cloud=3,m1=1" fans cloud lenses out while keeping a slow local box
|
// model's in-process fanout.
|
||||||
// serial, exactly the way the model map does for whole models.
|
|
||||||
func lensConcurrency() int {
|
func lensConcurrency() int {
|
||||||
if n, ok := providerOverride("GADFLY_PROVIDER_LENS_CONCURRENCY", modelProvider()); ok {
|
if n, ok := providerOverride("GADFLY_PROVIDER_LENS_CONCURRENCY", modelProvider()); ok {
|
||||||
return n
|
return n
|
||||||
@@ -310,7 +333,7 @@ func lensConcurrency() int {
|
|||||||
|
|
||||||
// providerOverride parses a "provider=N,provider=N" env map and returns the
|
// providerOverride parses a "provider=N,provider=N" env map and returns the
|
||||||
// value for provider when present and valid (>0). Mirrors entrypoint.sh's
|
// value for provider when present and valid (>0). Mirrors entrypoint.sh's
|
||||||
// provider_cap lookup so the two concurrency maps share one syntax.
|
// provider_lens_cap lookup so the two share one syntax.
|
||||||
func providerOverride(envName, provider string) (int, bool) {
|
func providerOverride(envName, provider string) (int, bool) {
|
||||||
for _, item := range strings.Split(os.Getenv(envName), ",") {
|
for _, item := range strings.Split(os.Getenv(envName), ",") {
|
||||||
k, v, ok := strings.Cut(item, "=")
|
k, v, ok := strings.Cut(item, "=")
|
||||||
|
|||||||
+2
-2
@@ -162,8 +162,8 @@ func buildSpec(provider, model string) string {
|
|||||||
// entrypoint.sh's provider_of: the segment before the first "/" in GADFLY_MODEL,
|
// entrypoint.sh's provider_of: the segment before the first "/" in GADFLY_MODEL,
|
||||||
// else GADFLY_PROVIDER, else the default (ollama-cloud). The binary reviews one
|
// else GADFLY_PROVIDER, else the default (ollama-cloud). The binary reviews one
|
||||||
// model per invocation, so this is that model's provider — used to resolve
|
// model per invocation, so this is that model's provider — used to resolve
|
||||||
// per-provider policy (e.g. lens concurrency) against the SAME provider keys
|
// per-provider policy (e.g. the lens budget) against the SAME provider keys
|
||||||
// entrypoint uses for GADFLY_PROVIDER_CONCURRENCY.
|
// entrypoint uses for GADFLY_PROVIDER_LENS_CONCURRENCY.
|
||||||
func modelProvider() string {
|
func modelProvider() string {
|
||||||
model := strings.TrimSpace(os.Getenv("GADFLY_MODEL"))
|
model := strings.TrimSpace(os.Getenv("GADFLY_MODEL"))
|
||||||
if pfx, _, ok := strings.Cut(model, "/"); ok {
|
if pfx, _, ok := strings.Cut(model, "/"); ok {
|
||||||
|
|||||||
+39
-23
@@ -198,14 +198,18 @@ export GADFLY_FINDINGS_TOKEN="${GADFLY_FINDINGS_TOKEN:-}"
|
|||||||
# provider key envs (OPENAI_API_KEY, …) are inherited by run.sh and the binary.
|
# provider key envs (OPENAI_API_KEY, …) are inherited by run.sh and the binary.
|
||||||
#
|
#
|
||||||
# Concurrency: each PROVIDER is its own lane and lanes run in PARALLEL, so a fast
|
# Concurrency: each PROVIDER is its own lane and lanes run in PARALLEL, so a fast
|
||||||
# cloud provider isn't stuck behind a slow local box. Within a lane, at most
|
# cloud provider isn't stuck behind a slow local box. Within a lane ALL of the
|
||||||
# `cap` models run at once. cap = GADFLY_PROVIDER_CONCURRENCY's "provider=N"
|
# provider's models run at once; the real throttle is a single PROVIDER-WIDE lens
|
||||||
# entry, else GADFLY_CONCURRENCY (default 1). A model's provider is the spec's
|
# budget (a shared permit pool, seeded per lane) that every model's lenses draw
|
||||||
# first path segment ("m1pro/qwen3.6:35b-mlx" -> m1pro), or GADFLY_PROVIDER /
|
# from — so total lens passes in flight per provider is bounded, but a model
|
||||||
# ollama-cloud for a bare id. Default (cap 1) keeps a single-provider pool fully
|
# winding down to its last lens immediately yields its freed permits to another
|
||||||
# sequential, exactly as before.
|
# model's queued lenses instead of holding a whole "model slot" (the old
|
||||||
|
# GADFLY_PROVIDER_CONCURRENCY model cap, now removed, caused that tail stall). The
|
||||||
|
# budget = GADFLY_PROVIDER_LENS_CONCURRENCY's "provider=N" entry, else
|
||||||
|
# GADFLY_LENS_CONCURRENCY (default 1). A model's provider is the spec's first path
|
||||||
|
# segment ("m1pro/qwen3.6:35b-mlx" -> m1pro), or GADFLY_PROVIDER / ollama-cloud
|
||||||
|
# for a bare id.
|
||||||
MODELS="${GADFLY_MODELS:-${OLLAMA_REVIEW_MODELS:-$DEFAULT_MODELS}}"
|
MODELS="${GADFLY_MODELS:-${OLLAMA_REVIEW_MODELS:-$DEFAULT_MODELS}}"
|
||||||
DEFAULT_CONC="${GADFLY_CONCURRENCY:-1}"
|
|
||||||
|
|
||||||
# --- huge-PR downshift ------------------------------------------------------
|
# --- huge-PR downshift ------------------------------------------------------
|
||||||
# A very large diff is what burns the model budget: every review step re-sends
|
# A very large diff is what burns the model budget: every review step re-sends
|
||||||
@@ -245,24 +249,33 @@ provider_of() { case "$1" in */*) echo "${1%%/*}";; *) echo "${GADFLY_PROVIDER:-
|
|||||||
STATUS_DIR="${WORKDIR}/status"
|
STATUS_DIR="${WORKDIR}/status"
|
||||||
status_file_for() { echo "${STATUS_DIR}/$(echo "$1" | tr -c '[:alnum:]._-' '_').json"; }
|
status_file_for() { echo "${STATUS_DIR}/$(echo "$1" | tr -c '[:alnum:]._-' '_').json"; }
|
||||||
|
|
||||||
provider_cap() { # provider -> concurrency (override map "p=N,...", else default)
|
# Root of the per-provider lens permit pools (one subdir per lane, seeded by
|
||||||
|
# run_lane). Cleared up front so a reused WORKDIR can't leak stale permit files.
|
||||||
|
LENS_SEM_ROOT="${WORKDIR}/lenssem"
|
||||||
|
rm -rf "$LENS_SEM_ROOT" 2>/dev/null || true
|
||||||
|
|
||||||
|
provider_lens_cap() { # provider -> provider-wide lens budget (permit-pool size)
|
||||||
local p="$1" item k v
|
local p="$1" item k v
|
||||||
IFS=',' read -ra _caps <<< "${GADFLY_PROVIDER_CONCURRENCY:-}"
|
IFS=',' read -ra _lcaps <<< "${GADFLY_PROVIDER_LENS_CONCURRENCY:-}"
|
||||||
for item in "${_caps[@]}"; do
|
for item in "${_lcaps[@]}"; do
|
||||||
k="$(echo "${item%%=*}" | tr -d '[:space:]')"
|
k="$(echo "${item%%=*}" | tr -d '[:space:]')"
|
||||||
v="$(echo "${item#*=}" | tr -d '[:space:]')"
|
v="$(echo "${item#*=}" | tr -d '[:space:]')"
|
||||||
if [ "$k" = "$p" ] && [ -n "$v" ]; then echo "$v"; return; fi
|
if [ "$k" = "$p" ] && [ -n "$v" ]; then echo "$v"; return; fi
|
||||||
done
|
done
|
||||||
echo "$DEFAULT_CONC"
|
echo "${GADFLY_LENS_CONCURRENCY:-1}"
|
||||||
}
|
}
|
||||||
|
|
||||||
review_one() {
|
review_one() {
|
||||||
local sf="" ff=""
|
local m="$1" sem_dir="${2:-}" sem_size="${3:-}" sf="" ff=""
|
||||||
[ "${GADFLY_STATUS_BOARD:-1}" != "0" ] && sf="$(status_file_for "$1")"
|
[ "${GADFLY_STATUS_BOARD:-1}" != "0" ] && sf="$(status_file_for "$m")"
|
||||||
[ "$CONSOLIDATE" = "1" ] && ff="$(findings_file_for "$1")"
|
[ "$CONSOLIDATE" = "1" ] && ff="$(findings_file_for "$m")"
|
||||||
PROVIDER=ollama MODEL="$1" GADFLY_BIN="/usr/local/bin/gadfly" GADFLY_REPO_DIR="$REPO_DIR" \
|
# GADFLY_LENS_SEM_DIR/_SIZE point the binary at this provider's shared lens
|
||||||
|
# permit pool (empty => the binary just uses its in-process lens limit). These
|
||||||
|
# are inherited by the binary through run.sh's environment.
|
||||||
|
PROVIDER=ollama MODEL="$m" GADFLY_BIN="/usr/local/bin/gadfly" GADFLY_REPO_DIR="$REPO_DIR" \
|
||||||
GADFLY_STATUS_FILE="$sf" GADFLY_FINDINGS_OUT="$ff" GADFLY_CONSOLIDATE="$CONSOLIDATE" \
|
GADFLY_STATUS_FILE="$sf" GADFLY_FINDINGS_OUT="$ff" GADFLY_CONSOLIDATE="$CONSOLIDATE" \
|
||||||
bash "${SCRIPTS_DIR}/run.sh" || log "model $1 failed (continuing)"
|
GADFLY_LENS_SEM_DIR="$sem_dir" GADFLY_LENS_SEM_SIZE="$sem_size" \
|
||||||
|
bash "${SCRIPTS_DIR}/run.sh" || log "model $m failed (continuing)"
|
||||||
# If the binary never wrote real status (run.sh skipped it: empty diff, no key,
|
# If the binary never wrote real status (run.sh skipped it: empty diff, no key,
|
||||||
# binary missing), the pre-seed stays {started:0, done:false} and the board
|
# binary missing), the pre-seed stays {started:0, done:false} and the board
|
||||||
# would show this model "waiting to start" forever and never reach N/N. Mark
|
# would show this model "waiting to start" forever and never reach N/N. Mark
|
||||||
@@ -313,16 +326,19 @@ for m in "${MODEL_LIST[@]}"; do
|
|||||||
case " $PROVIDERS " in *" $p "*) ;; *) PROVIDERS="${PROVIDERS}${PROVIDERS:+ }$p" ;; esac
|
case " $PROVIDERS " in *" $p "*) ;; *) PROVIDERS="${PROVIDERS}${PROVIDERS:+ }$p" ;; esac
|
||||||
done
|
done
|
||||||
|
|
||||||
run_lane() { # $1=provider: run its models, at most `cap` at a time
|
run_lane() { # $1=provider: run ALL its models at once, throttled only by a shared
|
||||||
local p="$1" cap inflight=0 m
|
# provider-wide lens permit pool (no per-model cap).
|
||||||
cap="$(provider_cap "$p")"; [ "$cap" -ge 1 ] 2>/dev/null || cap=1
|
local p="$1" budget sem_dir m
|
||||||
|
budget="$(provider_lens_cap "$p")"; [ "$budget" -ge 1 ] 2>/dev/null || budget=1
|
||||||
local mine=()
|
local mine=()
|
||||||
for m in "${MODEL_LIST[@]}"; do [ "$(provider_of "$m")" = "$p" ] && mine+=("$m"); done
|
for m in "${MODEL_LIST[@]}"; do [ "$(provider_of "$m")" = "$p" ] && mine+=("$m"); done
|
||||||
log "lane ${p}: cap ${cap}; models: ${mine[*]}"
|
# Seed this provider's lens permit pool: a directory the binary flocks N permit
|
||||||
|
# files in (created lazily), one shared budget across every model in the lane.
|
||||||
|
sem_dir="${LENS_SEM_ROOT}/$(echo "$p" | tr -c '[:alnum:]._-' '_')"
|
||||||
|
mkdir -p "$sem_dir"
|
||||||
|
log "lane ${p}: lens budget ${budget} shared across ${#mine[@]} model(s): ${mine[*]}"
|
||||||
for m in "${mine[@]}"; do
|
for m in "${mine[@]}"; do
|
||||||
review_one "$m" &
|
review_one "$m" "$sem_dir" "$budget" &
|
||||||
inflight=$((inflight+1))
|
|
||||||
if [ "$inflight" -ge "$cap" ]; then wait -n 2>/dev/null || wait; inflight=$((inflight-1)); fi
|
|
||||||
done
|
done
|
||||||
wait
|
wait
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -55,14 +55,14 @@ jobs:
|
|||||||
# csv to choose; "all" for everything; or define custom ones via a repo
|
# csv to choose; "all" for everything; or define custom ones via a repo
|
||||||
# .gadfly.yml / GADFLY_SPECIALIST_<NAME>. See README "Specialists".
|
# .gadfly.yml / GADFLY_SPECIALIST_<NAME>. See README "Specialists".
|
||||||
GADFLY_SPECIALISTS: ${{ vars.GADFLY_SPECIALISTS }}
|
GADFLY_SPECIALISTS: ${{ vars.GADFLY_SPECIALISTS }}
|
||||||
# Lens fan-out (optional; default 1 = lenses run sequentially within a
|
# Concurrency (optional; default 1 = fully sequential per provider). The
|
||||||
# model). Raise it to run a model's lenses concurrently so each model
|
# ONE throttle is a per-provider LENS BUDGET: the max lens passes (a lens =
|
||||||
# posts its comment sooner. Total in-flight requests = (models at once)
|
# one specialist's review+recheck) in flight at once for a provider, shared
|
||||||
# × (lenses at once), so to fan out without oversubscribing a backend,
|
# across ALL that provider's models — every model in a lane runs at once and
|
||||||
# keep its model cap low and raise its lens cap. Per-provider configurable
|
# its lenses draw from the shared budget. Raise it to overlap lenses; set it
|
||||||
# via GADFLY_PROVIDER_LENS_CONCURRENCY (same lanes as the model map):
|
# per provider with GADFLY_PROVIDER_LENS_CONCURRENCY:
|
||||||
# GADFLY_PROVIDER_CONCURRENCY: "ollama-cloud=1,m1=1"
|
|
||||||
# GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1=1"
|
# GADFLY_PROVIDER_LENS_CONCURRENCY: "ollama-cloud=3,m1=1"
|
||||||
|
# (The old GADFLY_PROVIDER_CONCURRENCY model cap was removed and is ignored.)
|
||||||
# GADFLY_LENS_CONCURRENCY: ${{ vars.GADFLY_LENS_CONCURRENCY }}
|
# GADFLY_LENS_CONCURRENCY: ${{ vars.GADFLY_LENS_CONCURRENCY }}
|
||||||
# GADFLY_PROVIDER_LENS_CONCURRENCY: ${{ vars.GADFLY_PROVIDER_LENS_CONCURRENCY }}
|
# GADFLY_PROVIDER_LENS_CONCURRENCY: ${{ vars.GADFLY_PROVIDER_LENS_CONCURRENCY }}
|
||||||
# Live status board (optional; ON by default): one consolidated comment
|
# Live status board (optional; ON by default): one consolidated comment
|
||||||
|
|||||||
Reference in New Issue
Block a user