golang: Deflake channel tests

Channel tests are passing on an idle machine. However if there is
another load - e.g. in the presence of simultaneous Firefox start,
channel tests break, for example:

	    def test_select():
	        ...

	        # non-blocking try recv: ok
	        ch = chan()
	        done = chan()
	        def _():
	            for i in range(N):
	                ch.send(i)
	            done.close()
	        go(_)

	        for i in range(N):
	            tdelay()
	            if i % 2:
	                _, _rx = select(
	                        ch.recv,
	                        default,
	                )
	                assert (_, _rx) == (0, i)
	            else:
	                _, _rx = select(
	                        ch.recv_,
	                        default,
	                )
	>               assert (_, _rx) == (0, (i, True))
	E               assert (1, None) == (0, (320, True))
	E                 At index 0 diff: 1 != 0
	E                 Use -v to get the full diff

	golang_test.py:209: AssertionError

The failure here is that it was default case selected, not ch.recv_. The
default was selected because the sending goroutine was not fast to
enqueue next send before we tried to receive. We were trying to make
sure that the sender will be enqueued first via adding 1ms time delay
before trying to receive, but in the presence of concurrent load spikes
that turns out to be not enough.

We could try to fix the test by increasing the time to sleep in tdelay,
make the tests more slow and still not 100% reliable. However we can
change the tests to instead actually wait for the condition that is
semantically required: a sender enqueued on the channel.

Do that everywhere where tdelay was used.

Now tests are faster (it was ~3s total, now it is ~ 0.5s total) and pass
irregardless of whether the machine is idle or otherwise loaded.

golang/golang_test.py

@@ -23,13 +23,8 @@ from pytest import raises
 from os.path import dirname
 import os, sys, time, threading, inspect, subprocess
 
-# tdelay delays a bit.
-#
-# XXX needed in situations when we need to start with known ordering but do not
-# have a way to wait properly for ordering event.
-def tdelay():
-    time.sleep(1E-3)    # 1ms
-
+from golang import _chan_recv, _chan_send
+from golang._pycompat import im_class
 
 def test_go():
     # leaked goroutine behaviour check: done in separate process because we need
@@ -50,6 +45,34 @@ def test_go():
             env=env)
 
 
+# waitBlocked waits till a receive or send channel operation blocks waiting on the channel.
+#
+# For example `waitBlocked(ch.send)` waits till sender blocks waiting on ch.
+def waitBlocked(chanop):
+    if im_class(chanop) is not chan:
+        panic("wait blocked: %r is method of a non-chan: %r" % (chanop, im_class(chanop)))
+    ch = chanop.__self__
+    recv = send = False
+    if chanop.__func__ is _chan_recv:
+        recv = True
+    elif chanop.__func__ is _chan_send:
+        send = True
+    else:
+        panic("wait blocked: unexpected chan method: %r" % (chanop,))
+
+    t0 = time.time()
+    while 1:
+        with ch._mu:
+            if recv and len(ch._recvq) > 0:
+                return
+            if send and len(ch._sendq) > 0:
+                return
+        now = time.time()
+        if now-t0 > 10: # waited > 10 seconds - likely deadlock
+            panic("deadlock")
+        time.sleep(0)   # yield to another thread / coroutine
+
+
 def test_chan():
     # sync: pre-close vs send/recv
     ch = chan()
@@ -75,7 +98,7 @@ def test_chan():
     # sync: close vs send
     ch = chan()
     def _():
-        tdelay()
+        waitBlocked(ch.send)
         ch.close()
     go(_)
     with raises(_PanicError): ch.send(0)
@@ -83,7 +106,7 @@ def test_chan():
     # close vs recv
     ch = chan()
     def _():
-        tdelay()
+        waitBlocked(ch.recv)
         ch.close()
     go(_)
     assert ch.recv_() == (None, False)
@@ -122,7 +145,7 @@ def test_chan():
         ch.send(i)
     assert len(ch) == 3
     def _():
-        tdelay()
+        waitBlocked(ch.send)
         assert ch.recv_() == (0, True)
         done.send('a')
         for i in range(1,4):
@@ -175,7 +198,7 @@ def test_select():
     go(_)
 
     for i in range(N):
-        tdelay()
+        waitBlocked(ch.recv)
         _, _rx = select(
                 (ch.send, i),
                 default,
@@ -194,7 +217,7 @@ def test_select():
     go(_)
 
     for i in range(N):
-        tdelay()
+        waitBlocked(ch.send)
         if i % 2:
             _, _rx = select(
                     ch.recv,
@@ -215,7 +238,7 @@ def test_select():
     ch2 = chan()
     done = chan()
     def _():
-        tdelay()
+        waitBlocked(ch1.send)
         assert ch1.recv() == 'a'
         done.close()
     go(_)
@@ -235,7 +258,7 @@ def test_select():
     ch2 = chan()
     done = chan()
     def _():
-        tdelay()
+        waitBlocked(ch1.recv)
         ch1.send('a')
         done.close()
     go(_)
@@ -255,7 +278,7 @@ def test_select():
     ch2 = chan()
     done = chan()
     def _():
-        tdelay()
+        waitBlocked(ch1.send)
         assert ch1.recv() == 'a'
         done.close()
     go(_)
@@ -275,7 +298,7 @@ def test_select():
     ch2 = chan()
     done = chan()
     def _():
-        tdelay()
+        waitBlocked(ch1.recv)
         ch1.send('a')
         done.close()
     go(_)