Make inmemPipeline check that it's not closed before sending RPCs

This change fixes a race inside between the shutdownCh and the
consumerCh in the select statement of inmempipeLine.AppendEntries,
which could otherwise forward an RPC request to the consumer even if
the transport was closed with InmemTransport.Close().

See also hashicorp#275 for a sample program that reproduces the race if ran
long enough. The same program doesn't fail anymore with this change
applied.

inmem_transport.go

@@ -24,7 +24,7 @@ type inmemPipeline struct {
 
 	shutdown     bool
 	shutdownCh   chan struct{}
-	shutdownLock sync.Mutex
+	shutdownLock sync.RWMutex
 }
 
 type inmemPipelineInflight struct {
@@ -288,6 +288,17 @@ func (i *inmemPipeline) AppendEntries(args *AppendEntriesRequest, resp *AppendEn
 		Command:  args,
 		RespChan: respCh,
 	}
+
+	// Check if we have been already shutdown, otherwise the random choose
+	// made by select statement below might pick consumerCh even if
+	// shutdownCh was closed.
+	i.shutdownLock.RLock()
+	shutdown := i.shutdown
+	i.shutdownLock.RUnlock()
+	if shutdown {
+		return nil, ErrPipelineShutdown
+	}
+
 	select {
 	case i.peer.consumerCh <- rpc:
 	case <-timeout: