feat(runtime): make async i/o test pass

src/runtime-prototype/src/StuntDouble/Actor.hs

@@ -30,7 +30,7 @@ data Cont a
   --   | ...
 
 data IOResult = Unit | String String
-  deriving Show
+  deriving (Eq, Ord, Show)
 
 data ActorF x
   = Call LocalRef Message (Message -> x)

src/runtime-prototype/src/StuntDouble/EventLoop.hs

@@ -132,7 +132,7 @@ handleReceive e ls = case envelopeKind e of
             installContinuation async (envelopeReceiver e) (envelopeCorrelationId e) k ls
           LaterIO async k -> do
             say ls "installing i/o continuation"
-            -- installIOContinuation async ...
+            installIOContinuation async (envelopeReceiver e) (envelopeCorrelationId e) k ls
 
   ResponseKind -> do
     emit ls (LogReceive (envelopeSender e) (envelopeReceiver e) (envelopeMessage e)
@@ -158,7 +158,17 @@ handleReceive e ls = case envelopeKind e of
             undefined
 
 handleAsyncIODone :: Async IOResult -> IOResult -> LoopState -> IO ()
-handleAsyncIODone a r ls = undefined
+handleAsyncIODone a r ls = do
+  (self, corrId, k) <- recallIOContinuation a ls
+  emit ls (LogAsyncIOFinish corrId r)
+  cont <- runActor ls self (k r)
+  case cont of
+    Now replyMsg -> do
+      resps <- readTVarIO (loopStateResponses ls)
+      let respVar = resps Map.! corrId
+      atomically (putTMVar respVar replyMsg)
+    Later {} -> do
+      undefined
 
 runActor :: LoopState -> RemoteRef -> Free ActorF a -> IO a
 runActor ls self = iterM go return

src/runtime-prototype/src/StuntDouble/EventLoop/Event.hs

@@ -65,6 +65,7 @@ data LogEntry
   | LogRequestStart RemoteRef RemoteRef Message CorrelationId EventLoopName
   | LogRequestFinish CorrelationId Message EventLoopName
   | LogComment String EventLoopName
+  | LogAsyncIOFinish CorrelationId IOResult EventLoopName
   deriving (Eq, Show)
 
 isComment :: LogEntry -> Bool

src/runtime-prototype/src/StuntDouble/EventLoop/State.hs

@@ -20,7 +20,8 @@ data LoopState = LoopState
   , loopStateQueue :: TBQueue Event
   , loopStateActors :: TVar (Map LocalRef (Message -> Actor)) -- XXX: Only changed by main loop, so no need for STM?
   , loopStateIOAsyncs :: TVar [Async IOResult]
-  , loopStateIOHandlers :: TVar (Map (Async IOResult) (LocalRef, IOResult -> Actor))
+  , loopStateIOContinuations :: TVar (Map (Async IOResult)
+                                          (RemoteRef, CorrelationId, IOResult -> Actor))
   , loopStateTransport :: Transport IO -- Will not change once created, so doesn't need STM?
   , loopStateNextCorrelationId :: TVar CorrelationId
   , loopStateResponses     :: TVar (Map CorrelationId (TMVar Message))
@@ -48,6 +49,20 @@ recallContinuation a ls = do
     return ks
   return (Map.lookup a ks)
 
+installIOContinuation :: Async IOResult -> RemoteRef -> CorrelationId
+                      -> (IOResult -> Actor) -> LoopState -> IO ()
+installIOContinuation a self corrId k ls = atomically $
+    modifyTVar' (loopStateIOContinuations ls) (Map.insert a (self, corrId, k))
+
+recallIOContinuation :: Async IOResult -> LoopState
+                     -> IO (RemoteRef, CorrelationId, IOResult -> Actor)
+recallIOContinuation a ls = do
+  ks <- atomically $ do
+    ks <- readTVar (loopStateIOContinuations ls)
+    writeTVar (loopStateIOContinuations ls) (Map.delete a ks)
+    return ks
+  return (ks Map.! a)
+
 correlateAsync :: CorrelationId -> Async Message -> LoopState -> IO ()
 correlateAsync cid a ls = atomically $
   modifyTVar' (loopStateWaitingAsyncs ls) (Map.insert cid a)

src/runtime-prototype/test/StuntDouble/EventLoopTest.hs

@@ -21,7 +21,7 @@ testActor2 rref (Message "init") = do
 
 testActor3 :: Message -> Actor
 testActor3 (Message "init") = do
-  a <- asyncIO (threadDelay 1000000 >> return (String "result"))
+  a <- asyncIO (threadDelay 300000 >> return (String "result"))
   return (LaterIO a (\(String result) -> return (Now (Message ("Got: " ++ result)))))
 
 eventLoopA :: String -> EventLoopName
@@ -88,10 +88,9 @@ unit_sendLater = do
         reply <- wait a
         reply @?= Message "Got: bye!")
     (\(e :: SomeException) -> dump el1 >> dump el2 >> eventLog el1 >>= mapM_ print)
-  -}
-
   dump el1
   dump el2
+  -}
 
   quit el1
   quit el2
@@ -101,8 +100,15 @@ unit_asyncIO = do
   elog <- emptyEventLog
   let ev = eventLoopA "asyncIO"
   el <- makeEventLoop "/tmp" ev elog
+  lref <- spawn el testActor3
+  a <- send el (localToRemoteRef ev lref) (Message "init")
+  reply <- wait a
+  reply @?= Message "Got: result"
+  quit el
+  {-
   catch (do lref <- spawn el testActor3
             a <- send el (localToRemoteRef ev lref) (Message "init")
             reply <- wait a
             reply @?= Message "Got: result")
     (\(e :: SomeException) -> dump el >> eventLog el >>= mapM_ print >> print e)
+-}