Use RWMutex instead of Mutex for the job queue

Should be a safe optimisation for the job queue code.  Also renamed
a variable for better clarity.

api/main.go

@@ -99,7 +99,7 @@ func main() {
 
 	// Start background goroutines to handle job queue responses
 	if !com.UseAMQP {
-		com.ResponseWaiters = com.NewResponseReceiver()
+		com.ResponseQueue = com.NewResponseQueue()
 		com.CheckResponsesQueue = make(chan struct{})
 		com.SubmitterInstance = com.RandomString(3)
 		go com.ResponseQueueCheck()

common/live.go

@@ -691,13 +691,13 @@ func RemoveLiveDB(dbOwner, dbName string) (err error) {
 func WaitForResponse[T any](jobID int, resp *T) (err error) {
 	// Add the response receiver
 	responseChan := make(chan ResponseInfo)
-	ResponseWaiters.AddReceiver(jobID, &responseChan)
+	ResponseQueue.AddReceiver(jobID, &responseChan)
 
 	// Wait for a response
 	response := <-responseChan
 
 	// Remove the response receiver
-	ResponseWaiters.RemoveReceiver(jobID)
+	ResponseQueue.RemoveReceiver(jobID)
 
 	// Update the response status to 'processed' (should be fine done async)
 	go ResponseComplete(response.responseID)

common/live_types.go

@@ -98,14 +98,14 @@ type ResponseInfo struct {
 
 // ResponseReceivers is a simple structure used for matching up job queue responses to the caller who submitted the job
 type ResponseReceivers struct {
-	sync.Mutex
+	sync.RWMutex
 	receivers map[int]*chan ResponseInfo
 }
 
-// NewResponseReceiver is the constructor function for creating new ResponseReceivers
-func NewResponseReceiver() *ResponseReceivers {
+// NewResponseQueue is the constructor function for creating a new ResponseReceivers
+func NewResponseQueue() *ResponseReceivers {
 	z := ResponseReceivers{
-		Mutex:     sync.Mutex{},
+		RWMutex:   sync.RWMutex{},
 		receivers: nil,
 	}
 	z.receivers = make(map[int]*chan ResponseInfo)

common/postgresql_live.go

@@ -21,8 +21,8 @@ var (
 	// CheckResponsesQueue is used by the non-live daemons for triggering a check of the job responses queue
 	CheckResponsesQueue chan struct{}
 
-	// ResponseWaiters is used to direct job queue responses back to the appropriate callers
-	ResponseWaiters *ResponseReceivers
+	// ResponseQueue is used to direct job queue responses back to the appropriate callers
+	ResponseQueue *ResponseReceivers
 
 	// SubmitterInstance is a random string generated at server start for identification purposes
 	SubmitterInstance string
@@ -49,8 +49,7 @@ func JobQueueCheck() {
 			continue
 		}
 
-		// TODO: a) we're in a loop so can't use defer, and
-		//       b) we'll likely need to update the job state to 'error' on failure instead of rolling back
+		// TODO: should we update the job state to 'error' on failure?
 
 		dbQuery := `
 			SELECT job_id, operation, submitter_node, details
@@ -523,12 +522,12 @@ func ResponseQueueCheck() {
 				log.Printf("%s: picked up response %d for jobID %d", Conf.Live.Nodename, responseID, jobID)
 			}
 
-			ResponseWaiters.Lock()
-			receiverChan, ok := ResponseWaiters.receivers[jobID]
+			ResponseQueue.RLock()
+			receiverChan, ok := ResponseQueue.receivers[jobID]
 			if ok {
 				*receiverChan <- ResponseInfo{jobID: jobID, responseID: responseID, payload: details}
 			}
-			ResponseWaiters.Unlock()
+			ResponseQueue.RUnlock()
 			return nil
 		})
 		if err != nil {

webui/main.go

@@ -3189,7 +3189,7 @@ func main() {
 
 	// Start background goroutines to handle job queue responses
 	if !com.UseAMQP {
-		com.ResponseWaiters = com.NewResponseReceiver()
+		com.ResponseQueue = com.NewResponseQueue()
 		com.CheckResponsesQueue = make(chan struct{})
 		com.SubmitterInstance = com.RandomString(3)
 		go com.ResponseQueueCheck()