diff --git a/pkg/tasks/read_request_queue.go b/pkg/tasks/read_request_queue.go index 444e514fd..630cf651d 100644 --- a/pkg/tasks/read_request_queue.go +++ b/pkg/tasks/read_request_queue.go @@ -15,9 +15,12 @@ import "sync" // 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. +// A caller acts on the content the view shows once it is answered, so a request +// is held for the task asked for last, from the moment it is asked for. When a +// task is replaced, the requests it hasn't answered stay for the task replacing it. +// Only once the task asked for last has given the view all of its content does the +// queue stop holding requests, and hand back the ones it still holds to answer. A +// request made after that 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 @@ -30,54 +33,82 @@ type readRequestQueue struct { 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 + // Whether requests are held for a task to serve: the one reading the content, or + // one that has been asked for and is going to replace it. False before the first + // task is asked for, and once the task asked for last has read all of its + // content. + holding 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{} { +// hold says that a task has been asked for, so that requests are held for it until +// it serves them. +func (self *readRequestQueue) hold() { self.mutex.Lock() defer self.mutex.Unlock() - self.serving = true + self.holding = true +} + +// 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. The task's own first +// request is served first, ahead of those it takes over from the task it replaces, +// so that it shows its content as soon as it would have without them. +func (self *readRequestQueue) beginServing(firstRequest LinesToRead) <-chan struct{} { + self.mutex.Lock() + self.holding = true + self.requests = append([]LinesToRead{firstRequest}, self.requests...) + self.mutex.Unlock() + + self.ringDoorbell() return self.doorbell } -// stopServing takes the task away and hands back the requests it never answered, -// for the caller to answer in its place. +// stopServing stops holding requests and hands back the ones it still holds, for +// the caller to answer in place of a task. func (self *readRequestQueue) stopServing() []LinesToRead { self.mutex.Lock() defer self.mutex.Unlock() - self.serving = false + self.holding = 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. +// enqueue gives a request to the task that is going to serve it, and reports +// whether there is one. When there isn't, the request is the caller's to answer. func (self *readRequestQueue) enqueue(request LinesToRead) bool { self.mutex.Lock() - if !self.serving { + if !self.holding { self.mutex.Unlock() return false } self.requests = append(self.requests, request) self.mutex.Unlock() + self.ringDoorbell() + return true +} + +// requeue puts back a request that a task was serving when it was stopped, ahead of +// the ones made after it, for the task replacing it to serve. +func (self *readRequestQueue) requeue(request LinesToRead) { + self.mutex.Lock() + self.requests = append([]LinesToRead{request}, self.requests...) + self.mutex.Unlock() + + self.ringDoorbell() +} + +func (self *readRequestQueue) ringDoorbell() { select { case self.doorbell <- struct{}{}: default: } - return true } // dequeue takes the oldest request, reporting false when there are none. diff --git a/pkg/tasks/read_request_queue_test.go b/pkg/tasks/read_request_queue_test.go index 423b89d09..0f9c14876 100644 --- a/pkg/tasks/read_request_queue_test.go +++ b/pkg/tasks/read_request_queue_test.go @@ -9,41 +9,60 @@ import ( func TestReadRequestQueueHandsBackWhatNoTaskWillServe(t *testing.T) { queue := newReadRequestQueue() - // Nothing has begun serving, so the request comes straight back to its caller. + // No task has been asked for, 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})) + // Once one has, requests are held for it, and the task serves its own first + // request ahead of them. + queue.hold() assert.True(t, queue.enqueue(LinesToRead{Total: 2})) + queue.beginServing(LinesToRead{Total: 1}) + assert.True(t, queue.enqueue(LinesToRead{Total: 3})) 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. + // A request that a task was serving when it was stopped goes back ahead of the + // ones made after it. + queue.requeue(request) + request, _ = queue.dequeue() + assert.Equal(t, 1, request.Total) + request, _ = queue.dequeue() + assert.Equal(t, 2, request.Total) + + // What no task got to comes back when the queue stops holding requests, 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.Equal(t, 3, unanswered[0].Total) - assert.False(t, queue.enqueue(LinesToRead{Total: 3})) + assert.False(t, queue.enqueue(LinesToRead{Total: 4})) _, ok = queue.dequeue() assert.False(t, ok) } func TestReadRequestQueueRingsTheDoorbell(t *testing.T) { queue := newReadRequestQueue() - doorbell := queue.beginServing() + doorbell := queue.beginServing(LinesToRead{Total: 1}) + + // The task's first request is there to be served right away. + select { + case <-doorbell: + default: + assert.Fail(t, "the doorbell didn't ring for the first request") + } + _, _ = queue.dequeue() select { case <-doorbell: - assert.Fail(t, "the doorbell rang before anything was queued") + assert.Fail(t, "the doorbell rang with nothing queued") default: } // A burst leaves one token, which stands for everything queued. - queue.enqueue(LinesToRead{Total: 1}) queue.enqueue(LinesToRead{Total: 2}) + queue.enqueue(LinesToRead{Total: 3}) select { case <-doorbell: diff --git a/pkg/tasks/tasks.go b/pkg/tasks/tasks.go index a7d0fafd9..5b70d2ca6 100644 --- a/pkg/tasks/tasks.go +++ b/pkg/tasks/tasks.go @@ -435,11 +435,30 @@ func (self *ViewBufferManager) readHoldingATask(totalLines int, then func()) { } } -// 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. +// stopServingReadRequests stops holding read requests for a task and answers the +// ones still waiting, so that nobody is left waiting for a callback that isn't +// coming. func (self *ViewBufferManager) stopServingReadRequests() { - for _, request := range self.readRequests.stopServing() { + answerReadRequests(self.readRequests.stopServing()) +} + +// answerReadRequestsIfLatest answers the read requests still waiting once the given +// task has given the view all of its content, unless another task has been asked for +// since. The requests are waiting for that task's content then, so they are left for +// it to serve. +func (self *ViewBufferManager) answerReadRequestsIfLatest(taskID int) { + self.taskIDMutex.Lock() + var unanswered []LinesToRead + if taskID == self.newTaskID { + unanswered = self.readRequests.stopServing() + } + self.taskIDMutex.Unlock() + + answerReadRequests(unanswered) +} + +func answerReadRequests(requests []LinesToRead) { + for _, request := range requests { if request.Then != nil { request.Then() } @@ -545,9 +564,7 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix loadingMutex := deadlock.Mutex{} - // 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() + readRequests := self.readRequests.beginServing(initialRequest) scanner := bufio.NewScanner(r) scanner.Split(utils.ScanLinesAndTruncateWhenLongerThanBuffer(bufio.MaxScanTokenSize)) @@ -705,6 +722,15 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix linesToRead.Then() } } + // A task that is stopped is being replaced by one that renders the + // content the caller is going to act on, so the request is left for + // that one to serve. A request without a Then has nobody waiting on + // it, and the task replacing this one reads what it needs by itself. + handOver := func() { + if linesToRead.Then != nil { + self.readRequests.requeue(linesToRead) + } + } // A restore that hasn't painted yet keeps us reading past the lines // asked for, all the way to the end of the input if need be. What it // is looking for may be anywhere in the new content, and a rendering @@ -713,14 +739,14 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix // and the view somewhere the user didn't put it. for linesToRead.Total == -1 || linesRead < linesToRead.Total || (restore != nil && !painted) { if stopped() { - callThen() + handOver() break outer } var ok bool var line []byte select { case <-opts.Stop: - callThen() + handOver() break outer case line, ok = <-lineChan: // process line below @@ -752,7 +778,7 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix // case above. select { case <-opts.Stop: - callThen() + handOver() break outer default: } @@ -805,10 +831,10 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix } // 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() + // content it asked for has been read. At end of input it has all been read + // already. A task that was stopped leaves the requests to the one replacing + // it. + self.answerReadRequestsIfLatest(opts.taskID) refreshViewIfStale() @@ -832,8 +858,6 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix close(lineWrittenChan) }) - self.readRequests.enqueue(initialRequest) - <-done return nil @@ -842,6 +866,10 @@ func (self *ViewBufferManager) NewCmdTask(start func() (Cmd, io.Reader), prefix // Close closes the task manager, killing whatever task may currently be running func (self *ViewBufferManager) Close() { + // No task is going to read anything after this, so nobody may be left waiting + // for content to be read. + defer self.stopServingReadRequests() + // stopCurrentTask is written by NewTask's goroutine under waitingMutex (and // so is the sync.Once it closes over), so read it under the lock and call // the captured value; a task starting on shutdown must not race us here. @@ -900,6 +928,8 @@ func (self *ViewBufferManager) ReserveTask() TaskReservation { defer self.taskIDMutex.Unlock() self.newTaskID++ + // Whoever asks for content to be read from now on is asking for this task's. + self.readRequests.hold() return TaskReservation{taskID: self.newTaskID} } @@ -986,9 +1016,6 @@ func (self *ViewBufferManager) NewReservedTask(reservation TaskReservation, f fu self.stopCurrentTask() } - // Nothing serves read requests between one task and the next. - self.stopServingReadRequests() - stop := make(chan struct{}) notifyStopped := make(chan struct{}) @@ -1006,6 +1033,10 @@ func (self *ViewBufferManager) NewReservedTask(reservation TaskReservation, f fu self.Log.Error(err) // might need an onError callback } + // A task that reads no command's output, such as one showing a message, has + // given the view all of its content once it returns. + self.answerReadRequestsIfLatest(taskID) + close(notifyStopped) }) diff --git a/pkg/tasks/tasks_test.go b/pkg/tasks/tasks_test.go index c391757cb..76bfa6aa7 100644 --- a/pkg/tasks/tasks_test.go +++ b/pkg/tasks/tasks_test.go @@ -842,45 +842,6 @@ func TestNewCmdTaskRefresh(t *testing.T) { } } -// A read request is answered by the task that serves it calling the request's Then. -// This checks that requests still waiting when the task is stopped are answered too, -// which is what happens when a re-render replaces the task. -func TestQueuedReadRequestsAreAnsweredWhenTheTaskStops(t *testing.T) { - noop := func() {} - task := gocui.NewFakeTask() - - // A pipe the task blocks on, so that the requests are still waiting when it stops. - pipeReader, pipeWriter := io.Pipe() - defer pipeWriter.Close() - - manager := NewViewBufferManager( - utils.NewDummyLog(), bytes.NewBuffer(nil), noop, noop, noop, noop, noop, noop, - func() gocui.Task { return task }, - func(f func()) error { f(); return nil }, - ) - - stop := make(chan struct{}) - fn := manager.NewCmdTask( - func() (Cmd, io.Reader) { return ExecCmd{Cmd: exec.Command("true")}, pipeReader }, - nil, LinesToRead{Total: 1, InitialRefreshAfter: -1}, noop) - go func() { _, _ = pipeWriter.Write([]byte("first line\n")) }() - go func() { _ = fn(TaskOpts{Stop: stop, InitialContentLoaded: noop}) }() - // Let the task start and read the line it was asked for, so that the requests - // below are handed to a task that is waiting for them. - time.Sleep(50 * time.Millisecond) - - answered := atomic.Int32{} - manager.ReadToEnd(func() { answered.Add(1) }) - manager.ReadToEnd(func() { answered.Add(1) }) - - // Let the first request be picked up and block on the pipe, then stop the task. - time.Sleep(50 * time.Millisecond) - close(stop) - time.Sleep(50 * time.Millisecond) - - assert.EqualValues(t, 2, answered.Load()) -} - // A caller of ReadToEnd acts on the whole of what the view shows once it is answered, // such as listing the files of the diff in a menu. If a re-render replaces the task // that is reading the content, the content is the re-render's, so whoever is waiting @@ -942,10 +903,107 @@ func TestReadToEndWaitingWhenARerenderReplacesTheTask(t *testing.T) { startTask(&BlankLineReader{totalLinesToYield: 50}) answered.Wait() - /* EXPECTED: assert.Equal(t, []int{50, 50}, linesWrittenWhenAnswered) - ACTUAL: */ - assert.Equal(t, []int{3, 3}, linesWrittenWhenAnswered) +} + +// A task that reads no command's output, such as one showing a message, has given the +// view all of its content once it returns. So when it replaces a task that was reading +// for someone, it answers the requests that task leaves it, after showing its content. +func TestReadRequestsLeftToAMessageAreAnsweredOnceItIsShown(t *testing.T) { + noop := func() {} + manager := NewViewBufferManager( + utils.NewDummyLog(), bytes.NewBuffer(nil), noop, noop, noop, noop, noop, noop, + func() gocui.Task { return gocui.NewFakeTask() }, + // no UI thread in the test; run the view mutations inline + func(f func()) error { f(); return nil }, + ) + defer manager.Close() + + stalled := BlockingLineReader{linesToYield: 1, blocked: make(chan struct{}), unblock: make(chan struct{})} + defer close(stalled.unblock) + start := func() (Cmd, io.Reader) { + // not actually starting this because it's not necessary + return ExecCmd{Cmd: exec.Command("blah")}, &stalled + } + _ = manager.NewTask(manager.NewCmdTask(start, nil, LinesToRead{Total: 1, InitialRefreshAfter: 1}, nil), "cmd") + <-stalled.blocked + + var messageShown atomic.Bool + answeredAfterTheMessage := make(chan bool, 2) + for range 2 { + manager.ReadToEnd(func() { answeredAfterTheMessage <- messageShown.Load() }) + } + + _ = manager.NewTask(func(TaskOpts) error { + messageShown.Store(true) + return nil + }, "message") + + assert.True(t, <-answeredAfterTheMessage) + assert.True(t, <-answeredAfterTheMessage) +} + +// A render whose task is created after the layout is the one whose content a request +// made in the meantime is about, so the request waits for that task to read it, even +// though the task before it had read all of its own. +func TestReadToEndWaitsForATaskAskedForButNotCreatedYet(t *testing.T) { + writer := bytes.NewBuffer(nil) + linesWritten := func() int { return strings.Count(writer.String(), "\n") } + noop := func() {} + manager := NewViewBufferManager( + utils.NewDummyLog(), writer, noop, noop, noop, noop, writer.Reset, noop, + func() gocui.Task { return gocui.NewFakeTask() }, + // no UI thread in the test; run the view mutations inline + func(f func()) error { f(); return nil }, + ) + defer manager.Close() + + newCmdTask := func(lines int, onDone func()) func(TaskOpts) error { + start := func() (Cmd, io.Reader) { + // not actually starting this because it's not necessary + return ExecCmd{Cmd: exec.Command("blah")}, &BlankLineReader{totalLinesToYield: lines} + } + return manager.NewCmdTask(start, nil, LinesToRead{Total: 100, InitialRefreshAfter: 1}, onDone) + } + + done := make(chan struct{}) + _ = manager.NewTask(newCmdTask(3, func() { close(done) }), "cmd1") + <-done + + reservation := manager.ReserveTask() + linesWrittenWhenAnswered := make(chan int, 1) + manager.ReadToEnd(func() { linesWrittenWhenAnswered <- linesWritten() }) + + _ = manager.NewReservedTask(reservation, newCmdTask(50, nil), "cmd2") + assert.Equal(t, 50, <-linesWrittenWhenAnswered) +} + +// Once the view's tasks are closed, no task is going to read anything, so whoever is +// waiting for content to be read is answered. +func TestCloseAnswersWaitingReadRequests(t *testing.T) { + noop := func() {} + manager := NewViewBufferManager( + utils.NewDummyLog(), bytes.NewBuffer(nil), noop, noop, noop, noop, noop, noop, + func() gocui.Task { return gocui.NewFakeTask() }, + // no UI thread in the test; run the view mutations inline + func(f func()) error { f(); return nil }, + ) + + stalled := BlockingLineReader{linesToYield: 1, blocked: make(chan struct{}), unblock: make(chan struct{})} + defer close(stalled.unblock) + start := func() (Cmd, io.Reader) { + // not actually starting this because it's not necessary + return ExecCmd{Cmd: exec.Command("blah")}, &stalled + } + _ = manager.NewTask(manager.NewCmdTask(start, nil, LinesToRead{Total: 1, InitialRefreshAfter: 1}, nil), "cmd") + <-stalled.blocked + + answered := atomic.Int32{} + manager.ReadToEnd(func() { answered.Add(1) }) + manager.ReadToEnd(func() { answered.Add(1) }) + + manager.Close() + assert.EqualValues(t, 2, answered.Load()) } // The callers of ReadToEnd hand their follow-up work to the UI thread from then, and