catfile: Fix deadlock between reading a new object and accessing itpks-catfile-request-queue-isdirty-deadlock

When we read in a new object via `ReadObject()`, then we must perform an
I/O operation to learn about its header. This operation is currently
done under the `currentObjectLock()`, but this can easily cause us to
deadlock when the I/O operation stalls. Consequentially, it's impossible
to access this object now anymore by other callers. Most importantly,
this means that both `close()` and `isDirty()` are blocked from making
any progress. But because those are used to determine whether processes
of the request queue should be killed in the first place we are now in a
deadlock-state.

Fix this issue by splitting up the lock into a read-lock for the current
object whose scope is a lot more fine-grained and only used when we
access the field or when we write to it. The second atomic lock is used
to only lock writing access across the whole lifetime of `ReadObject()`
so that no other caller can call it at the same time.

Changelog: fixed

1 files changed, 40 insertions, 4 deletions

diff --git a/internal/git/catfile/request_queue.go b/internal/git/catfile/request_queue.go
index d41ed38dc..a31a8b029 100644
--- a/internal/git/catfile/request_queue.go
+++ b/internal/git/catfile/request_queue.go
@@ -33,6 +33,10 @@ type requestQueue struct {
 	// closed indicates whether the queue is closed for additional requests.
 	closed int32
 
+	// isReadingObject indicates whether the queue is currently in `ReadObject()` to
+	// read in a new object's header.
+	isReadingObject int32
+
 	// isObjectQueue is set to `true` when this is a request queue which can be used for reading
 	// objects. If set to `false`, then this can only be used to read object info.
 	isObjectQueue bool
@@ -51,6 +55,8 @@ type requestQueue struct {
 // isDirty returns true either if there are outstanding requests for objects or if the current
 // object hasn't yet been fully consumed.
 func (q *requestQueue) isDirty() bool {
+	// We need to take this lock early. Like this we can ensure that `ReadObject()`
+	// cannot have it at the same time while `isReadingObject == 0`.
 	q.currentObjectLock.Lock()
 	defer q.currentObjectLock.Unlock()
 
@@ -60,6 +66,10 @@ func (q *requestQueue) isDirty() bool {
 		return true
 	}
 
+	if atomic.LoadInt32(&q.isReadingObject) != 0 {
+		return true
+	}
+
 	if atomic.LoadInt64(&q.outstandingRequests) != 0 {
 		return true
 	}
@@ -127,17 +137,39 @@ func (q *requestQueue) ReadObject() (*Object, error) {
 		panic("object queue used to read object info")
 	}
 
+	// We need to ensure that only a single call to `ReadObject()` can happen at the
+	// same point in time. While the easy solution would be to just lock
+	// `currentObjectLock()`, that would mean we hold the lock while performing I/O
+	// operations. That's a no-go though because the command may stall, and would
+	// deadlock with `isDirty()` and `close()`.
+	//
+	// Note that this must happen before we read the current object: otherwise, a
+	// concurrent caller might've already swapped it out under our feet.
+	if !atomic.CompareAndSwapInt32(&q.isReadingObject, 0, 1) {
+		return nil, fmt.Errorf("concurrent read of the object")
+	}
+	defer func() {
+		atomic.StoreInt32(&q.isReadingObject, 0)
+	}()
+
+	// We know no other user can change `currentObject` right now anymore, but we still
+	// need to take `currentObjectLock()` given that concurrent callers may try to access the
+	// object.
 	q.currentObjectLock.Lock()
-	defer q.currentObjectLock.Unlock()
+	currentObject := q.currentObject
+	q.currentObjectLock.Unlock()
 
-	if q.currentObject != nil {
+	if currentObject != nil {
 		// If the current object is still dirty, then we must not try to read a new object.
-		if q.currentObject.isDirty() {
+		if currentObject.isDirty() {
 			return nil, fmt.Errorf("current object has not been fully read")
 		}
 
-		q.currentObject.close()
+		currentObject.close()
+
+		q.currentObjectLock.Lock()
 		q.currentObject = nil
+		q.currentObjectLock.Unlock()
 
 		// If we have already read an object before, then we must consume the trailing
 		// newline after the object's data.
@@ -152,6 +184,10 @@ func (q *requestQueue) ReadObject() (*Object, error) {
 	}
 	q.trace.recordRequest(objectInfo.Type)
 
+	// Synchronize with readers of the object.
+	q.currentObjectLock.Lock()
+	defer q.currentObjectLock.Unlock()
+
 	q.currentObject = &Object{
 		ObjectInfo: *objectInfo,
 		dataReader: io.LimitedReader{