如何与time.After进行惯用同步？

Question

I'm writing an application that queues incoming requests. If a request has been on the queue for more than a certain amount of time, I'd like to throw a timeout. I'm doing that with time.After:

timeoutCh := time.After(5 * time.Second)
select {
    case <-timeoutCh:
         //throw timeout 504
    case <-processing:
         //process request
}

The processing channel (along with the request) is put on the queue, and when a request is taken off to be processed, I send a signal to the channel to hit the case statement:

processing <- true

The problem with this is that if timeoutCh has already been selected, the processing channel will block, so I need some way to check whether the request has timed out.

I considered using a shared atomic boolean, but if I do something like this:

case <-timeoutCh:
     requestTimedOut = true

and then check the boolean before sending to the processing channel, there's still a race condition, because the timeoutCh case may have been selected, but the bool not yet set to true!

Is there an idiomatic way of dealing with this sort of synchronization problem in Go?

Answer 1

Use a mutex coordinate processing of the data and timeout.

Define a type to hold the mutex, input, result, a channel to signal completion of the work and a flag indicating that the work, if any, is complete.

type work struct {
    sync.Mutex
    input    InputType
    result   ResultType
    signal   chan struct {}
    done     bool
}

The request handler creates and enqueues a work item and waits for a timeout or a signal from the queue processor. Either way, the request handler checks to see if the queue processor did the work and responds as appropriate.

func handler(resp http.ResponseWriter, req *http.Request) {
    w := &queueElement{
        input: computeInputFromRequest(req)
        signal:  make(chan struct{})
    }
    enqueue(w)

    // Wait for timeout or for queue processor to signal that the work is complete.
    select {
    case <-time.After(5 * time.Second):
    case <-w.signal:
    }

    w.Lock()
    done := w.done  // Record state of the work item.
    w.done = true   // Mark the work item as complete.
    w.Unlock()

    if !done {
        http.Error(w, "Timeout", http.StatusGatewayTimeout)
    }  else {
        respondWithResult(resp, w.result)
    }
}

The queue processor will look something like this:

 for {
   w := dequeue()
   w.Lock()
   if !w.done {
      w.done = true
      w.result = computeResultFromInput(w.input)
      close(w.signal)
   }
   w.Unlock()
}

To ensure that the request handler waits on the result, the queue processor holds the lock while processing the work item.