Fix: some specs rely on Fiber.yield behavior

Some specs are relying on the `Fiber.yield` behavior is actually the
event loop behavior (libevent2). They start failing whenever the
`Fiber.yield` algorithm changes —for example to push the current
fiber to the runnables queue, instead of adding a resume event.

This patch proposes changes to make the fiber synchronization
expectations explicit. Either by looping until somethig is ready, or
using Channel::Unbuffered's sync ability, or with an explicit
enqueue/resume for specs testing Channel itself.

spec/std/channel_spec.cr

@@ -1,5 +1,10 @@
 require "spec"
 
+private def yield_to(fiber)
+  Crystal::Scheduler.enqueue(Fiber.current)
+  Crystal::Scheduler.resume(fiber)
+end
+
 describe Channel do
   it "creates unbuffered with no arguments" do
     Channel(Int32).new.should be_a(Channel::Unbuffered(Int32))
@@ -40,17 +45,22 @@ describe Channel::Unbuffered do
   it "blocks if there is no receiver" do
     ch = Channel::Unbuffered(Int32).new
     state = 0
-    spawn do
+    main = Fiber.current
+
+    sender = Fiber.new do
       state = 1
       ch.send 123
       state = 2
+    ensure
+      yield_to(main)
     end
 
-    Fiber.yield
+    yield_to(sender)
     state.should eq(1)
     ch.receive.should eq(123)
     state.should eq(1)
-    Fiber.yield
+
+    sleep
     state.should eq(2)
   end
 
@@ -67,8 +77,8 @@ describe Channel::Unbuffered do
     ch = Channel::Unbuffered(Int32).new
     ch.full?.should be_true
     ch.empty?.should be_true
-    spawn { ch.send 123 }
-    Fiber.yield
+    sender = Fiber.new { ch.send 123 }
+    yield_to(sender)
     ch.empty?.should be_false
     ch.full?.should be_true
     ch.receive.should eq(123)
@@ -88,8 +98,8 @@ describe Channel::Unbuffered do
 
   it "can send and receive nil" do
     ch = Channel::Unbuffered(Nil).new
-    spawn { ch.send nil }
-    Fiber.yield
+    sender = Fiber.new { ch.send nil }
+    yield_to(sender)
     ch.empty?.should be_false
     ch.receive.should be_nil
     ch.empty?.should be_true
@@ -113,10 +123,19 @@ describe Channel::Unbuffered do
   it "can be closed from different fiber" do
     ch = Channel::Unbuffered(Int32).new
     received = false
-    spawn { expect_raises(Channel::ClosedError) { ch.receive }; received = true }
-    Fiber.yield
+    main = Fiber.current
+
+    receiver = Fiber.new do
+      expect_raises(Channel::ClosedError) { ch.receive }
+      received = true
+    ensure
+      yield_to(main)
+    end
+
+    yield_to(receiver)
     ch.close
-    Fiber.yield
+
+    sleep
     received.should be_true
   end
 
@@ -138,27 +157,44 @@ describe Channel::Unbuffered do
     ch.receive?.should eq(123)
   end
 
-  it "wakes up the sender fiber when channel is closed" do
+  it "wakes up sender fiber when channel is closed" do
     ch = Channel::Unbuffered(Nil).new
-    sender_closed = false
-    spawn do
-      ch.send nil
-      ch.send nil
-    rescue Channel::ClosedError
-      sender_closed = true
+    closed = false
+    main = Fiber.current
+
+    sender = Fiber.new do
+      ch.send(nil)
+      closed = ch.closed?
+      yield_to(main)
     end
-    receiver_closed = false
-    spawn do
-      Fiber.yield
+
+    yield_to(sender)
+
+    ch.close
+    sleep
+
+    closed.should be_true
+  end
+
+  it "wakes up receiver fibers when channel is closed" do
+    ch = Channel::Unbuffered(Nil).new
+    closed = false
+    main = Fiber.current
+
+    receiver = Fiber.new do
       ch.receive
     rescue Channel::ClosedError
-      receiver_closed = true
+      closed = ch.closed?
+    ensure
+      yield_to(main)
     end
-    Fiber.yield
+
+    yield_to(receiver)
+
     ch.close
-    Fiber.yield
-    sender_closed.should be_true
-    receiver_closed.should be_true
+    sleep
+
+    closed.should be_true
   end
 end
 
@@ -191,9 +227,8 @@ describe Channel::Buffered do
   it "doesn't block when not full" do
     ch = Channel::Buffered(Int32).new
     done = false
-    spawn { ch.send 123; done = true }
-    done.should be_false
-    Fiber.yield
+    sender = Fiber.new { ch.send 123; done = true }
+    yield_to(sender)
     done.should be_true
   end
 
@@ -213,8 +248,8 @@ describe Channel::Buffered do
 
   it "can send and receive nil" do
     ch = Channel::Buffered(Nil).new
-    spawn { ch.send nil }
-    Fiber.yield
+    sender = Fiber.new { ch.send nil }
+    yield_to(sender)
     ch.empty?.should be_false
     ch.receive.should be_nil
     ch.empty?.should be_true
@@ -238,10 +273,18 @@ describe Channel::Buffered do
   it "can be closed from different fiber" do
     ch = Channel::Buffered(Int32).new
     received = false
-    spawn { expect_raises(Channel::ClosedError) { ch.receive }; received = true }
-    Fiber.yield
+    main = Fiber.current
+
+    receiver = Fiber.new do
+      expect_raises(Channel::ClosedError) { ch.receive }
+      received = true
+    ensure
+      yield_to(main)
+    end
+
+    yield_to(receiver)
     ch.close
-    Fiber.yield
+    sleep
     received.should be_true
   end
 

spec/std/concurrent/future_spec.cr

@@ -34,13 +34,12 @@ describe Concurrent::Future do
 
   describe "future" do
     it "computes a value" do
-      chan = Channel(Int32).new(1)
+      chan = Channel::Unbuffered(Int32).new
 
       f = future { chan.receive }
       f.running?.should be_true
 
       chan.send 42
-      Fiber.yield
       f.completed?.should be_true
 
       f.get.should eq(42)
@@ -59,10 +58,16 @@ describe Concurrent::Future do
     end
 
     it "raises" do
-      f = future { raise IndexError.new("test error") }
+      # we rely on the channel to sync fibers:
+      chan = Channel::Unbuffered(Int32).new
+
+      f = future do
+        chan.receive
+        raise IndexError.new("test error")
+      end
       f.running?.should be_true
 
-      Fiber.yield
+      chan.send(0)
       f.completed?.should be_true
 
       expect_raises(IndexError) { f.get }

spec/std/concurrent/select_spec.cr

@@ -1,5 +1,10 @@
 require "spec"
 
+private def yield_to(fiber)
+  Crystal::Scheduler.enqueue(Fiber.current)
+  Crystal::Scheduler.resume(fiber)
+end
+
 describe "select" do
   it "select many receviers" do
     ch1 = Channel(Int32).new
@@ -70,6 +75,7 @@ describe "select" do
   it "select should work with send which started before receive, fixed #3862" do
     ch1 = Channel(Int32).new
     ch2 = Channel(Int32).new
+    main = Fiber.current
 
     spawn do
       select
@@ -87,10 +93,11 @@ describe "select" do
       when b = ch2.receive
         x = b
       end
+    ensure
+      yield_to(main)
     end
 
-    Fiber.yield
-
+    sleep
     x.should eq 1
   end
 end

spec/std/concurrent_spec.cr

@@ -4,8 +4,9 @@ private def method_with_named_args(chan, x = 1, y = 2)
   chan.send(x + y)
 end
 
-private def method_named(expected_named)
+private def method_named(expected_named, chan)
   Fiber.current.name.should eq(expected_named)
+  chan.close
 end
 
 class SomeParallelJobException < Exception
@@ -57,15 +58,18 @@ describe "concurrent" do
     end
 
     it "spawns named" do
+      chan = Channel::Unbuffered(Nil).new
       spawn(name: "sub") do
         Fiber.current.name.should eq("sub")
+        chan.close
       end
-      Fiber.yield
+      chan.receive?
     end
 
     it "spawns named with macro" do
-      spawn method_named("foo"), name: "foo"
-      Fiber.yield
+      chan = Channel::Unbuffered(Nil).new
+      spawn method_named("foo", chan), name: "foo"
+      chan.receive?
     end
   end
 

spec/std/http/server/server_spec.cr

@@ -3,6 +3,18 @@ require "http/server"
 require "http/client/response"
 require "../../../support/ssl"
 
+private def wait_for(timeout = 5.seconds)
+  now = Time.monotonic
+
+  until yield
+    Fiber.yield
+
+    if (Time.monotonic - now) > timeout
+      raise "server failed to start within 5 seconds"
+    end
+  end
+end
+
 private class RaiseErrno < IO
   def initialize(@value : Int32)
   end
@@ -268,8 +280,6 @@ module HTTP
 
       spawn { server.listen }
 
-      Fiber.yield
-
       HTTP::Client.get("http://#{address2}/").body.should eq "Test Server (#{address2})"
       HTTP::Client.get("http://#{address1}/").body.should eq "Test Server (#{address1})"
       HTTP::Client.get("http://#{address1}/").body.should eq "Test Server (#{address1})"
@@ -290,7 +300,7 @@ module HTTP
         server = Server.new { }
         server.bind_unused_port
         spawn { server.listen }
-        Fiber.yield
+        wait_for { server.listening? }
         expect_raises(Exception, "Can't add socket to running server") do
           server.bind_unused_port
         end
@@ -301,7 +311,7 @@ module HTTP
         server = Server.new { }
         server.bind_unused_port
         spawn { server.listen }
-        Fiber.yield
+        wait_for { server.listening? }
         server.close
         expect_raises(Exception, "Can't add socket to closed server") do
           server.bind_unused_port
@@ -413,7 +423,6 @@ module HTTP
         ip_address2 = socket.local_address
 
         spawn server.listen
-        Fiber.yield
 
         HTTP::Client.get("https://#{ip_address1}", tls: client_context).body.should eq "Test Server (#{ip_address1})\n"
         HTTP::Client.get("https://#{ip_address2}", tls: client_context).body.should eq "Test Server (#{ip_address2})\n"
@@ -427,8 +436,7 @@ module HTTP
         server = Server.new { }
         server.bind_unused_port
         spawn { server.listen }
-        Fiber.yield
-        server.listening?.should be_true
+        wait_for { server.listening? }
         expect_raises(Exception, "Can't start running server") do
           server.listen
         end
@@ -439,8 +447,7 @@ module HTTP
         server = Server.new { }
         server.bind_unused_port
         spawn { server.listen }
-        Fiber.yield
-        server.listening?.should be_true
+        wait_for { server.listening? }
         server.close
         server.listening?.should be_false
         expect_raises(Exception, "Can't re-start closed server") do
@@ -467,8 +474,7 @@ module HTTP
             socket2 = server.bind_unix path2
 
             spawn server.listen
-
-            Fiber.yield
+            wait_for { server.listening? }
 
             unix_request(path1).should eq "Test Server (#{path1})"
             unix_request(path2).should eq "Test Server (#{path2})"
@@ -497,6 +503,7 @@ module HTTP
       server_done = false
       spawn do
         server.listen
+      ensure
         server_done = true
       end
 
@@ -513,8 +520,6 @@ module HTTP
         HTTP::Client.get("https://#{address}/", tls: client_context).body.should eq "ok"
       end
 
-      Fiber.yield
-
       server_done.should be_false
     end
 
@@ -525,8 +530,6 @@ module HTTP
           context.response.puts "foo"
           context.response.flush
 
-          Fiber.yield
-
           context.response.puts "bar"
         end