jesseduffield.lazygit/pkg/tasks/read_request_queue.go

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
}