Repository navigation
fix(task): stop locking Base.SendData so concurrent workers run in parallel - #132
Conversation
File read workers shared one task mutex, so context evaluation for every record ran one at a time. Each worker now owns a task copy and a disjoint set of paths.
![argus-pattern[bot]](https://avatars.githubusercontent.com/in/4870434?s=60&v=4){kind=small}
There was a problem hiding this comment.
Reviewed the change to concurrent file reads (one task copy per worker, paths split round-robin) and checked it against task.Base.SendData/SendRecord and pipeline.runTaskConcurrently. I left three inline comments: static partitioning can leave workers idle when file sizes vary, record IDs now repeat across workers, and the hand-copied fields in readWorker can fall out of sync with the struct.
Testing: the PR adds no unit tests. A small table test for partitionPaths (workers == len(paths), workers < len(paths), a single path) and a race-enabled test showing every path is emitted exactly once with task_concurrency > 1 would protect this path. Note that pipeline.runTaskConcurrently still forces source file tasks to concurrency 1, so this in-task fan-out is the only concurrency the reader gets.
Round-robin lists stall when file sizes differ, and each worker copy restarted record ids at 1. Workers now take the next path from a shared queue, and one counter keeps ids unique without a shared lock.
SendData held the task mutex while it marshalled the record, evaluated every context query and blocked on the output send, so concurrent workers of any task waited on each other. Only the record id needed guarding; it is now an atomic counter and the mutex is gone. This replaces the file-reader-only workaround (per-worker task copies, a separate id counter and a SendData override), so file.go is back to main. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Mahesh Kamble (ma-gk)
changed the title
Summary
Base.SendDataheld the task mutex while it marshalled the record, evaluated everycontextquery, and blocked on the output send. So concurrent workers of one task waited on each other, even after their input was ready.recordIndexis now anatomic.Int64, and the mutex is gone fromBase. Nothing else used it.task_concurrency > 1, not just the file reader. The earlier file-only workaround (per-worker task copies, a separate id counter and aSendDataoverride) is gone, sofile.gomatchesmain.Benchmark
Local Apple M3 (
darwin/arm64). Five runs after one warmup. Each time includes a similar fixed overhead from the timing script, so compare the columns rather than reading them as absolute times.Eight context fields, concurrency 10
64 local JSON files (60 items each), eight
.data | fromjson | .items | map(...) | lengthcontext fields,task_concurrency: 10.Sample pipeline, no context fields
test/pipelines/file_concurrency_test.yaml(100 tiny.txtfiles, nocontextblock) holds the lock for only microseconds per record, so there is no measurable change here. It still reads and writes 100 files and exits 0.Test plan
go build ./...go test -race ./...Executeon one sharedjq.Query(context queries can now run concurrently)go vet ./internal/pkg/pipeline/...shows the same three copylocks warnings asmain. They come from therawstruct copies during unmarshalling.maintest/pipelines/file_concurrency_test.yaml: 100 files written, exit 0🤖 Generated with Claude Code