From dd2a1a634f34d736e3480388f8f283e57ff9cb60 Mon Sep 17 00:00:00 2001 From: Stefan Haller Date: Sat, 5 Sep 2026 13:27:03 +0200 Subject: [PATCH] Answer every read request, whether or not a task is still serving A caller of ReadLines or ReadToEnd is told that the content it asked for has been read by the request's Then being called. A task that reaches the end of its input answers the requests still queued behind the one it was serving, but a task that is stopped drops them, and their callers wait for a callback that never comes. Pressing "/" in the focused main view opens the search prompt from such a callback, so if a re-render replaces the task at that moment the prompt never opens. Answering them as the read loop ends would leave a request handed over after that point unanswered, and there is a window for one. A caller reads the channel to send on, and can reach the send itself only once the loop has gone. So hand requests over through a queue instead. Asking whether a task is there and giving it the request are one step, as are taking the task away and handing back what it never answered; a request made in between goes back to the caller to answer. The queue is unbounded rather than a fixed-size channel, for the reasons gocui's userEventQueue is. Requests are handed over from the UI thread, where a blocking send would deadlock against the task waiting to be let go, and a fixed channel that fills up leaves only blocking, dropping, reordering or panicking to choose between. Co-authored-by: Claude Opus 5 (1M context) --- pkg/tasks/read_request_queue.go | 101 +++++++++++++++++++++++++++ pkg/tasks/read_request_queue_test.go | 58 +++++++++++++++ pkg/tasks/tasks.go | 87 ++++++++++++----------- pkg/tasks/tasks_test.go | 10 +-- 4 files changed, 206 insertions(+), 50 deletions(-) create mode 100644 pkg/tasks/read_request_queue.go create mode 100644 pkg/tasks/read_request_queue_test.go diff --git a/pkg/tasks/read_request_queue.go b/pkg/tasks/read_request_queue.go new file mode 100644 index 000000000..444e514fd --- /dev/null +++ b/pkg/tasks/read_request_queue.go @@ -0,0 +1,101 @@ +package tasks + +import "sync" + +// readRequestQueue is an unbounded, order-preserving FIFO of the read requests a +// view's running command task serves (see LinesToRead), with a reader that comes +// and goes. +// +// It's unbounded, rather than a fixed-size channel, for the same reasons as the +// user-event queue in gocui. Requests are handed over from the UI thread, where a +// blocking send would deadlock against the task that is waiting to be let go, and +// a fixed channel that fills up leaves only bad choices: blocking, dropping, +// reordering, or panicking on overflow. Appending to a slice does none of those. +// +// The reader coming and going is the other half of what it's for. A request is +// how a caller asks for content to be read and hears, through the request's Then, +// that it has been; a request nobody answers leaves that caller waiting for good. +// So asking whether a task is there and handing it the request are one step, and +// so are taking the task away and handing back what it never answered. A request +// made in between finds no task and goes back to its caller to answer. +// +// enqueue appends under the mutex and rings the doorbell; the task selects on the +// doorbell to wake, then takes requests until there are none left. The doorbell is +// buffered(1) and rung with a non-blocking send, so it's a coalescing "work +// pending" flag rather than a per-request signal: a burst of appends leaves at +// most one token, and the task takes everything the token stands for on a single +// wake. A token left over after the queue empties causes one harmless empty wake. +type readRequestQueue struct { + mutex sync.Mutex + requests []LinesToRead + doorbell chan struct{} + + // Whether a task is there to serve the requests. False before the first task + // starts, and between one task ending and the next starting. + serving bool +} + +func newReadRequestQueue() *readRequestQueue { + return &readRequestQueue{doorbell: make(chan struct{}, 1)} +} + +// beginServing says that a task is now there to serve the queue, and returns the +// doorbell that tells it when there is something to serve. +func (self *readRequestQueue) beginServing() <-chan struct{} { + self.mutex.Lock() + defer self.mutex.Unlock() + + self.serving = true + return self.doorbell +} + +// stopServing takes the task away and hands back the requests it never answered, +// for the caller to answer in its place. +func (self *readRequestQueue) stopServing() []LinesToRead { + self.mutex.Lock() + defer self.mutex.Unlock() + + self.serving = false + unanswered := self.requests + self.requests = nil + return unanswered +} + +// enqueue gives a request to the task serving the queue, and reports whether +// there was one to give it to. When there wasn't, the request is the caller's to +// answer. +func (self *readRequestQueue) enqueue(request LinesToRead) bool { + self.mutex.Lock() + if !self.serving { + self.mutex.Unlock() + return false + } + self.requests = append(self.requests, request) + self.mutex.Unlock() + + select { + case self.doorbell <- struct{}{}: + default: + } + return true +} + +// dequeue takes the oldest request, reporting false when there are none. +func (self *readRequestQueue) dequeue() (LinesToRead, bool) { + self.mutex.Lock() + defer self.mutex.Unlock() + + if len(self.requests) == 0 { + return LinesToRead{}, false + } + request := self.requests[0] + if len(self.requests) == 1 { + // Release the backing array whenever the queue drains, so a one-off burst + // doesn't pin its peak size for the rest of the session. + self.requests = nil + } else { + self.requests[0] = LinesToRead{} + self.requests = self.requests[1:] + } + return request, true +} diff --git a/pkg/tasks/read_request_queue_test.go b/pkg/tasks/read_request_queue_test.go new file mode 100644 index 000000000..423b89d09 --- /dev/null +++ b/pkg/tasks/read_request_queue_test.go @@ -0,0 +1,58 @@ +package tasks + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestReadRequestQueueHandsBackWhatNoTaskWillServe(t *testing.T) { + queue := newReadRequestQueue() + + // Nothing has begun serving, so the request comes straight back to its caller. + assert.False(t, queue.enqueue(LinesToRead{Total: 1})) + + queue.beginServing() + assert.True(t, queue.enqueue(LinesToRead{Total: 1})) + assert.True(t, queue.enqueue(LinesToRead{Total: 2})) + + request, ok := queue.dequeue() + assert.True(t, ok) + assert.Equal(t, 1, request.Total) + + // What the task never got to comes back when it stops, and nothing is taken + // from a caller after that. + unanswered := queue.stopServing() + assert.Len(t, unanswered, 1) + assert.Equal(t, 2, unanswered[0].Total) + + assert.False(t, queue.enqueue(LinesToRead{Total: 3})) + _, ok = queue.dequeue() + assert.False(t, ok) +} + +func TestReadRequestQueueRingsTheDoorbell(t *testing.T) { + queue := newReadRequestQueue() + doorbell := queue.beginServing() + + select { + case <-doorbell: + assert.Fail(t, "the doorbell rang before anything was queued") + default: + } + + // A burst leaves one token, which stands for everything queued. + queue.enqueue(LinesToRead{Total: 1}) + queue.enqueue(LinesToRead{Total: 2}) + + select { + case <-doorbell: + default: + assert.Fail(t, "the doorbell didn't ring") + } + select { + case <-doorbell: + assert.Fail(t, "the doorbell rang twice for one wake") + default: + } +} diff --git a/pkg/tasks/tasks.go b/pkg/tasks/tasks.go index 17cfabb5f..377e74c0a 100644 --- a/pkg/tasks/tasks.go +++ b/pkg/tasks/tasks.go @@ -70,12 +70,11 @@ type ViewBufferManager struct { taskIDMutex deadlock.Mutex Log *logrus.Entry newTaskID int - // The channel by which the currently-running task is told to read more - // lines (e.g. as the user scrolls). Held in an atomic because it's swapped - // out as tasks come and go while ReadLines/ReadToEnd read it from the UI - // thread; nil when no task is running. - readLines atomic.Pointer[chan LinesToRead] - taskKey string + // The requests by which the currently-running task is told to read more lines + // (e.g. as the user scrolls), and which it answers once it has. The task + // serving them comes and goes; see readRequestQueue. + readRequests *readRequestQueue + taskKey string // Resets the view's scroll position to the top. A render whose content is // different from what the view last showed (a different command key) calls @@ -173,6 +172,7 @@ func NewViewBufferManager( onUIThread func(f func()) error, ) *ViewBufferManager { return &ViewBufferManager{ + readRequests: newReadRequestQueue(), Log: log, writer: writer, beforeStart: beforeStart, @@ -191,12 +191,9 @@ func NewViewBufferManager( // (e.g. as the user scrolls down, back up, and down again) don't re-read lines // that have already been read: the task only ever reads the shortfall. func (self *ViewBufferManager) ReadLines(totalLines int) { - if ch := self.readLines.Load(); ch != nil { - readLines := *ch - go utils.Safe(func() { - readLines <- LinesToRead{Total: totalLines, InitialRefreshAfter: -1} - }) - } + // A request with no Then needs no answer, so there is nothing to do when no + // task is there to take it. + self.readRequests.enqueue(LinesToRead{Total: totalLines, InitialRefreshAfter: -1}) } // IsLoading reports whether a command task is currently reading content into the @@ -215,16 +212,24 @@ func (self *ViewBufferManager) StartLoading() { } func (self *ViewBufferManager) ReadToEnd(then func()) { - if ch := self.readLines.Load(); ch != nil { - readLines := *ch - go utils.Safe(func() { - readLines <- LinesToRead{Total: -1, InitialRefreshAfter: -1, Then: then} - }) - } else if then != nil { + request := LinesToRead{Total: -1, InitialRefreshAfter: -1, Then: then} + if !self.readRequests.enqueue(request) && then != nil { + // With no task reading, everything there is to read has been read. then() } } +// stopServingReadRequests takes the task away from the read-request queue and +// answers whatever it never got to, so that nobody is left waiting for a callback +// that isn't coming. +func (self *ViewBufferManager) stopServingReadRequests() { + for _, request := range self.readRequests.stopServing() { + if request.Then != nil { + request.Then() + } + } +} + func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix string, linesToRead LinesToRead, onDoneFn func()) func(TaskOpts) error { return func(opts TaskOpts) error { var onDoneOnce sync.Once @@ -289,8 +294,9 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix loadingMutex := deadlock.Mutex{} - readLines := make(chan LinesToRead, 1024) - self.readLines.Store(&readLines) + // Begin serving before any goroutine starts, so that the first request below + // can't arrive before there is a task to take it. + readRequests := self.readRequests.beginServing() scanner := bufio.NewScanner(r) scanner.Split(utils.ScanLinesAndTruncateWhenLongerThanBuffer(bufio.MaxScanTokenSize)) @@ -423,10 +429,17 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix if stopped() { break outer } - select { - case <-opts.Stop: - break outer - case linesToRead := <-readLines: + linesToRead, ok := self.readRequests.dequeue() + if !ok { + // Nothing to read yet: wait to be told there is, or to be stopped. + select { + case <-opts.Stop: + break outer + case <-readRequests: + } + continue + } + { callThen := func() { if linesToRead.Then != nil { linesToRead.Then() @@ -493,21 +506,6 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix // means a newer task is taking over and is still loading. self.loading.Store(false) callThen() - // Any read requests that were queued while we were reading are - // now trivially satisfied, since we've read everything. Fire - // their callbacks instead of dropping them when we break out of - // the loop below (and nil out readLines). - drain: - for { - select { - case queued := <-readLines: - if queued.Then != nil { - queued.Then() - } - default: - break drain - } - } break outer } writeToView(append(line, '\n')) @@ -532,7 +530,11 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix } } - self.readLines.Store(nil) + // Whoever made a request the loop never got to is waiting to hear that the + // content it asked for has been read, and there is nothing here to read it + // any more: at end of input it has all been read already, and a task that + // was stopped is handing the view over to the one replacing it. + self.stopServingReadRequests() refreshViewIfStale() @@ -556,7 +558,7 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix close(lineWrittenChan) }) - readLines <- linesToRead + self.readRequests.enqueue(linesToRead) <-done @@ -670,7 +672,8 @@ func (self *ViewBufferManager) NewTask(f func(TaskOpts) error, key string) error self.stopCurrentTask() } - self.readLines.Store(nil) + // Nothing serves read requests between one task and the next. + self.stopServingReadRequests() stop := make(chan struct{}) notifyStopped := make(chan struct{}) diff --git a/pkg/tasks/tasks_test.go b/pkg/tasks/tasks_test.go index 4020fb09d..c50a54cac 100644 --- a/pkg/tasks/tasks_test.go +++ b/pkg/tasks/tasks_test.go @@ -253,12 +253,9 @@ func TestNewCmdTaskQueuedReadAtEndOfInput(t *testing.T) { }) <-reader.blocked + // The request is queued by the time this returns, so it is outstanding when we + // let the task reach EOF below. manager.ReadToEnd(func() { thenCalled = true }) - // ReadToEnd queues its request from a goroutine; wait for it to land so that - // it is definitely outstanding by the time we let the task reach EOF. - for len(*manager.readLines.Load()) == 0 { - time.Sleep(time.Millisecond) - } close(reader.unblock) wg.Wait() @@ -523,8 +520,5 @@ func TestQueuedReadRequestsAreAnsweredWhenTheTaskStops(t *testing.T) { close(stop) time.Sleep(50 * time.Millisecond) - /* EXPECTED: assert.EqualValues(t, 2, answered.Load()) - ACTUAL: */ - assert.EqualValues(t, 1, answered.Load()) }