diff --git a/dify-agent-runtime/Makefile b/dify-agent-runtime/Makefile index 533248f5e1d..3cbe4d54b07 100644 --- a/dify-agent-runtime/Makefile +++ b/dify-agent-runtime/Makefile @@ -143,3 +143,6 @@ sync-e2b-template-dev: sync-e2b-template-prod: E2B_API_KEY=$(E2B_API_KEY_PROD) ./docker/sync-e2b-template.sh + +benchmark: + go test ./benchmarks/ -bench . -run '^$$' -benchtime 40x diff --git a/dify-agent-runtime/benchmarks/e2e_test.go b/dify-agent-runtime/benchmarks/e2e_test.go new file mode 100644 index 00000000000..5b1ab4dd392 --- /dev/null +++ b/dify-agent-runtime/benchmarks/e2e_test.go @@ -0,0 +1,212 @@ +// End-to-end benchmarks that drive the real shellctl HTTP API in-process. +// +// Unlike perf_test.go (which isolates single techniques), this boots an actual +// *server.Service + Handler over httptest and fires real POST /v1/jobs/run +// requests, so it exercises the full path: job-dir mkdir, script/env writes, +// SQLite commits, tmux fork/exec, the WaitJob poll loop, and the background +// pipe monitor. It only uses stable exported APIs so it compiles unchanged +// against both the optimized and pre-optimization ("git stash") server code. +// +// Because it lives in an untracked file, `git stash` does NOT remove it — so +// the before/after workflow is: +// +// go test ./benchmarks/ -bench E2E -run '^$' -benchtime 200x | tee after.txt +// git stash +// go test ./benchmarks/ -bench E2E -run '^$' -benchtime 200x | tee before.txt +// git stash pop +// benchstat before.txt after.txt # optional +// +// Requires tmux on PATH. The shellctl helper binaries are built automatically +// into a temp dir the first time a benchmark runs. +package benchmarks + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "sync" + "testing" + + "github.com/langgenius/dify/dify-agent-runtime/internal/server" +) + +const modulePath = "github.com/langgenius/dify/dify-agent-runtime" + +var ( + buildOnce sync.Once + buildErr error + binDir string +) + +// ensureBinaries builds the shellctl helper binaries the server shells out to +// (runner, sanitize-pty, runner-exit) once per test process and prepends their +// directory to PATH so both installRunner's LookPath and the tmux-invoked +// sanitize/runner-exit commands resolve. +func ensureBinaries(tb testing.TB) { + tb.Helper() + buildOnce.Do(func() { + // Silence the server's request/lifecycle logging so it neither skews + // timings with I/O nor interleaves with benchmark output. + log.SetOutput(io.Discard) + if _, err := exec.LookPath("tmux"); err != nil { + buildErr = fmt.Errorf("tmux not found in PATH") + return + } + dir, err := os.MkdirTemp("/tmp", "agbench-bin") + if err != nil { + buildErr = err + return + } + binDir = dir + targets := map[string]string{ + "shellctl-runner": modulePath + "/cmd/runner", + "shellctl-sanitize-pty": modulePath + "/cmd/sanitize-pty", + "shellctl-runner-exit": modulePath + "/cmd/runner-exit", + } + for name, pkg := range targets { + out := filepath.Join(dir, name) + cmd := exec.Command("go", "build", "-o", out, pkg) + if b, err := cmd.CombinedOutput(); err != nil { + buildErr = fmt.Errorf("build %s: %v: %s", name, err, b) + return + } + } + _ = os.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) + }) + if buildErr != nil { + tb.Skipf("cannot build shellctl binaries: %v", buildErr) + } +} + +// startServer boots a fresh in-process shellctl server with background GC and +// the pipe monitor running (matching production wiring). Returns the base URL, +// auth token, and a cleanup func. +func startServer(tb testing.TB) (string, string, func()) { + tb.Helper() + ensureBinaries(tb) + + stateDir, err := os.MkdirTemp("/tmp", "agbench-state") + if err != nil { + tb.Fatalf("mkdtemp state: %v", err) + } + cwd, err := os.MkdirTemp("/tmp", "agbench-cwd") + if err != nil { + tb.Fatalf("mkdtemp cwd: %v", err) + } + + cfg, err := server.DefaultConfig() + if err != nil { + tb.Fatalf("default config: %v", err) + } + cfg.StateDir = stateDir + cfg.RuntimeDir = filepath.Join(stateDir, "runtime") + cfg.DefaultCwd = cwd + cfg.AuthToken = "bench-token" + + svc := server.NewService(cfg) + if err := svc.Initialize(); err != nil { + tb.Fatalf("service initialize: %v", err) + } + svc.StartBackgroundGC() + svc.StartBackgroundPipeMonitor() + + ts := httptest.NewServer(server.Handler(svc, cfg)) + + cleanup := func() { + ts.Close() + svc.Shutdown() + // Kill the dedicated tmux server so background sleep jobs don't leak. + kill := exec.Command("tmux", "-S", cfg.TmuxSocket(), "kill-server") + _ = kill.Run() + _ = os.RemoveAll(stateDir) + _ = os.RemoveAll(cwd) + } + return ts.URL, cfg.AuthToken, cleanup +} + +func runJob(tb testing.TB, client *http.Client, baseURL, token, script string, timeout float64) { + tb.Helper() + payload, _ := json.Marshal(map[string]any{"script": script, "timeout": timeout}) + req, _ := http.NewRequest("POST", baseURL+"/v1/jobs/run", bytes.NewReader(payload)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + resp, err := client.Do(req) + if err != nil { + tb.Fatalf("run job: %v", err) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + if resp.StatusCode != 200 { + tb.Fatalf("run job status = %d", resp.StatusCode) + } +} + +// BenchmarkE2E_RunEcho measures full request latency for a trivial job that +// creates, starts, runs, and completes: the RunJob write path (3 SQLite +// commits) plus the WaitJob poll loop's per-poll runtime-state tmux calls. +func BenchmarkE2E_RunEcho(b *testing.B) { + baseURL, token, cleanup := startServer(b) + defer cleanup() + client := &http.Client{} + + // Prime tmux/server lazy init. + runJob(b, client, baseURL, token, "echo warmup", 5) + + b.ResetTimer() + for b.Loop() { + runJob(b, client, baseURL, token, "echo hi", 5) + } +} + +// BenchmarkE2E_RunEchoParallel drives many concurrent run requests. This is +// where the optimizations bite: the single-writer SQLite connection serializes +// commits (so per-commit fsync cost is on the critical path) and concurrent +// tmux fork/exec competes for CPU, so halving forks + dropping fsync raises +// sustained throughput even though single-request latency is gated by fixed +// poll/handshake waits. +func BenchmarkE2E_RunEchoParallel(b *testing.B) { + baseURL, token, cleanup := startServer(b) + defer cleanup() + + runJob(b, &http.Client{}, baseURL, token, "echo warmup", 5) + + b.ResetTimer() + b.RunParallel(func(pb *testing.PB) { + client := &http.Client{} + for pb.Next() { + runJob(b, client, baseURL, token, "echo hi", 5) + } + }) +} + +// BenchmarkE2E_RunEchoUnderMonitorLoad keeps N long-running jobs alive so the +// 1s background pipe monitor reconciles them every tick (2N tmux forks before, +// 1 after), then measures echo-job latency under that contention on the shared +// single-writer SQLite connection and CPU. +func BenchmarkE2E_RunEchoUnderMonitorLoad(b *testing.B) { + for _, n := range []int{8, 32} { + b.Run(fmt.Sprintf("running=%d", n), func(b *testing.B) { + baseURL, token, cleanup := startServer(b) + defer cleanup() + client := &http.Client{} + + // Seed N background jobs: sleep long, but return the run call + // quickly via a short wait timeout so the job stays running. + for i := 0; i < n; i++ { + runJob(b, client, baseURL, token, "sleep 600", 0.3) + } + + b.ResetTimer() + for b.Loop() { + runJob(b, client, baseURL, token, "echo hi", 5) + } + }) + } +} diff --git a/dify-agent-runtime/cmd/runner/main.go b/dify-agent-runtime/cmd/runner/main.go index 9d9a669f885..945a20f01ed 100644 --- a/dify-agent-runtime/cmd/runner/main.go +++ b/dify-agent-runtime/cmd/runner/main.go @@ -33,6 +33,12 @@ import ( "github.com/langgenius/dify/dify-agent-runtime/internal/landlock" ) +// startGatePollInterval is how often the parent runner checks for the +// start-gate handshake file. It is a cheap local stat on the same path the +// server has just written, so a fine interval keeps per-job startup latency +// low without meaningful cost. +const startGatePollInterval = 5 * time.Millisecond + func main() { if len(os.Args) > 1 && os.Args[1] == "--exec" { childMode() @@ -67,7 +73,7 @@ func parentMode() { if _, err := os.Stat(startGate); err == nil { break } - time.Sleep(50 * time.Millisecond) + time.Sleep(startGatePollInterval) } // Load environment overlay from JSON. diff --git a/dify-agent-runtime/internal/server/config.go b/dify-agent-runtime/internal/server/config.go index a329bd512f4..c30ed34df70 100644 --- a/dify-agent-runtime/internal/server/config.go +++ b/dify-agent-runtime/internal/server/config.go @@ -23,7 +23,7 @@ const ( DefaultTerminateGraceSeconds = 10.0 DefaultGCIntervalSeconds = 60.0 DefaultGCFinishedJobRetentionSeconds = 300.0 - DefaultPollInterval = 50 * time.Millisecond + DefaultPollInterval = 5 * time.Millisecond DefaultPipeMonitorInterval = 1 * time.Second DefaultPipeReadyTimeout = 10 * time.Second DefaultSQLiteBusyTimeoutMs = 5000 diff --git a/dify-agent-runtime/internal/server/db.go b/dify-agent-runtime/internal/server/db.go index e6353e77e97..e64c01cbaf9 100644 --- a/dify-agent-runtime/internal/server/db.go +++ b/dify-agent-runtime/internal/server/db.go @@ -68,7 +68,14 @@ type DB struct { // OpenDB opens (or creates) the shellctl SQLite database. func OpenDB(dbPath string, busyTimeoutMs int) (*DB, error) { - dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(%d)&_pragma=journal_mode(WAL)", dbPath, busyTimeoutMs) + // synchronous=NORMAL is durable and crash-safe under WAL: it only fsyncs on + // checkpoint rather than on every commit, which removes the per-write fsync + // on the job-creation path. Worst case on power loss is losing the last few + // transactions of ephemeral job state, never DB corruption. + dsn := fmt.Sprintf( + "file:%s?_pragma=busy_timeout(%d)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)", + dbPath, busyTimeoutMs, + ) db, err := sql.Open("sqlite", dsn) if err != nil { return nil, fmt.Errorf("open sqlite: %w", err)