code review 110580043: runtime: add fast paths to non-blocking channel operations

src/pkg/runtime/chan.goc

 // Copyright 2009 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
 package runtime
 #include "runtime.h"
 #include "arch_GOARCH.h"
 #include "type.h"
 #include "race.h"
 #include "malloc.h"
@@ skipping 78 matching lines @@
                 runtime·park(nil, nil, "chan send (nil chan)");
                 return false;  // not reached
         }
 
         if(debug) {
                 runtime·printf("chansend: chan=%p; elem=", c);
                 c->elemtype->alg->print(c->elemsize, ep);
                 runtime·prints("\n");
         }
 
+        if(raceenabled)
+                runtime·racereadpc(c, pc, chansend);
+
+        // Fast path: if it's a non-blocking operation that can't proceed,
+        // then we can return w/o acquiring the lock.
+        if(!block && !c->closed && ((c->dataqsiz == 0 && c->recvq.first == nil) ||
+                (c->dataqsiz > 0 && c->qcount == c->dataqsiz)))
+                return false;
+
         t0 = 0;
         mysg.releasetime = 0;
         if(runtime·blockprofilerate > 0) {
                 t0 = runtime·cputicks();
                 mysg.releasetime = -1;
         }
 
         runtime·lock(c);
-        if(raceenabled)
-                runtime·racereadpc(c, pc, chansend);
         if(c->closed)
                 goto closed;
 
         if(c->dataqsiz > 0)
                 goto asynch;
 
         sg = dequeue(&c->recvq);
         if(sg != nil) {
                 if(raceenabled)
                         racesync(c, sg);
@@ skipping 96 matching lines @@
                 runtime·printf("chanrecv: chan=%p\n", c);
 
         if(c == nil) {
                 USED(t);
                 if(!block)
                         return false;
                 runtime·park(nil, nil, "chan receive (nil chan)");
                 return false;  // not reached
         }
 
+        // Fast path: if it's a non-blocking operation that can't proceed,
+        // then we can return w/o acquiring the lock.
+        if(!block && !c->closed && ((c->dataqsiz == 0 && c->sendq.first == nil) ||

[rsc, 2014/07/16 11:40:51]

The check for data in the channel is a single read of mutable state
(c->recvq.first or c->qcount, depending on the immutable c->dataqsiz). However,
this line also contains a read of c->closed. There is nothing to say that the
two reads will be self-consistent. 

If you have a program that does:

c := make(chan int, 1)
c <- 1

go func() {
    select {
    case <-c:
        println("recv from c")
    default:
        println("c is not ready - BUG!")
    }
}()

close(c)
<-c

I believe c is always ready and the select should never end up in the default
case. Either the select happens before the close, in which case it receives the
1, or it happens after the close but before the <-c, in which case it receives
the 1, or it happens after both close(c) and <-c, in which case it receives a 0
(from being closed and empty).

In the fast path, it could happen that the select sees c->closed == false, then
the original goroutine does close(c), <-c, then the select sees c->qcount == 0,
which causes it to return false incorrectly, falling into the 'c is not ready -
BUG!' case of the program.

Maybe moving the c->closed check to the end of the condition fixes the problem,
or maybe checking c->closed twice fixes the problem. Either way the change seems
a bit scarier and more subtle (less straightforward) than advertised. If moving
or repeating the closed check fixes the problem, do we need to use the atomic
operations to make sure the ordering is respected on ARM or Power or (maybe
later) MIPS?

I am not 100% sure, but I think that for send the checks are already in the
correct order. Reversing them there would introduce a similar race.

+                (c->dataqsiz > 0 && c->qcount == 0)))
+                return false;
+
         t0 = 0;
         mysg.releasetime = 0;
         if(runtime·blockprofilerate > 0) {
                 t0 = runtime·cputicks();
                 mysg.releasetime = -1;
         }
 
         runtime·lock(c);
         if(c->dataqsiz > 0)
                 goto asynch;
@@ skipping 911 matching lines @@
 }
 
 static void
 racesync(Hchan *c, SudoG *sg)
 {
         runtime·racerelease(chanbuf(c, 0));
         runtime·raceacquireg(sg->g, chanbuf(c, 0));
         runtime·racereleaseg(sg->g, chanbuf(c, 0));
         runtime·raceacquire(chanbuf(c, 0));
 }