Process NATS messages concurrently
@@ -240,9 +240,10 @@ func (n *NATS) Subscribe(instID flux.InstanceID, remote platform.Platform, done
myID := guid.New()
n.raw.Publish(string(instID)+methodKick, []byte(myID))
- go func() {
+ errc := make(chan error)
+
+ processRequest := func(request *nats.Msg) {
var err error
- for request := range requests {
switch {
case strings.HasSuffix(request.Subject, methodKick):
id := string(request.Data)
@@ -304,12 +305,46 @@ func (n *NATS) Subscribe(instID flux.InstanceID, remote platform.Platform, done
default:
err = errors.New("unknown message: " + request.Subject)
}
+ if err != nil {
+ select {
+ case errc <- err:
+ default:
+ // If the error channel is closed, it means that a
+ // different RPC goroutine had a fatal error that
+ // triggered the clean up and return of the parent
+ // goroutine. It is likely that the error we have
+ // encountered is due to the closure of the RPC
+ // client whilst our request was still in progress
+ // - don't panic.
+ }
+ }
+ }
+
+ go func() {
+ for {
+ select {
+ // If both an error and a request are available, the runtime may
+ // chose (by uniform pseudo-random selection) to process the
+ // request first. This may seem like a problem, but even if we were
+ // guaranteed to prefer the error channel there would still be a
+ // race between selecting a request here and a failing goroutine
+ // putting an error into the channel - it's an unavoidable
+ // consequence of asynchronous request handling. The error will get
+ // selected and handled soon enough.
+ case err := <-errc:
if _, ok := err.(platform.FatalError); ok && err != nil {
+ close(errc)
sub.Unsubscribe()
close(requests)
done <- err
return
}
+ case request := <-requests:
+ // Some of these operations (Apply in particular) can block for a long time;
+ // dispatch in a goroutine and deliver any errors back to us so that we can
+ // clean up on any hard failures.
+ go processRequest(request)
+ }
}
}()
}
