Fixed issue with sleeping and I/O

aiosched/kernel.py

@@ -489,7 +489,7 @@ class FIFOKernel:
         if not internal:
             self.reschedule_running()
 
-    def io_release_task(self, task: Task):
+    def io_release_task(self, task: Task, closing: bool = True):
         """
         Calls self.io_release in a loop
         for each I/O resource the given task owns
@@ -502,7 +502,8 @@ class FIFOKernel:
                 a, b = key.data.values()
                 if a is not task or b is not task:
                     continue
-            self.notify_closing(key.fileobj, broken=True, owner=task)
+            if closing:
+                self.notify_closing(key.fileobj, broken=True, owner=task)
             self.io_release(key.fileobj, internal=True)
         task.last_io = None
 
@@ -627,6 +628,7 @@ class FIFOKernel:
 
         if seconds:
             self.debugger.before_sleep(self.current_task, seconds)
+            self.io_release_task(self.current_task, closing=False)
             self.paused.put(self.current_task, seconds)
         else:
             # When we're called with a timeout of 0, this method acts as a checkpoint

aiosched/sync.py

@@ -321,7 +321,7 @@ class NetworkChannel(Channel):
 
         if self.closed:
             return False
-        elif self.reader.fileno() == -1:
+        elif await self.reader.fileno() == -1:
             return False
         else:
             try:

tests/network_channel.py

@@ -1,7 +1,6 @@
 import aiosched
 
 
-
 async def producer(c: aiosched.NetworkChannel, n: int):
     print("[producer] Started")
     for i in range(n):
@@ -18,7 +17,9 @@ async def consumer(c: aiosched.NetworkChannel):
         while await c.pending():
             item = await c.read(1)
             print(f"[consumer] Received {item.decode()}")
-            # await aiosched.sleep(2)    # If you uncomment this, the except block will be triggered
+            # In this example, the consumer is twice as slow
+            # as the producer
+            await aiosched.sleep(1)
     except aiosched.errors.ResourceClosed:
         print("[consumer] Stream has been closed early!")
     print("[consumer] Done")
@@ -35,4 +36,4 @@ async def main(channel: aiosched.NetworkChannel, n: int):
 
 aiosched.run(
     main, aiosched.NetworkChannel(), 5, debugger=()
-)  # 2 is the max size of the channel
+)