| ... |
... |
@@ -835,6 +835,8 @@ remoteAsyncReq pipe msg = do |
|
835
|
835
|
|
|
836
|
836
|
-- Block waiting for reply
|
|
837
|
837
|
atomically (takeTMVar wv)
|
|
|
838
|
+ `onException`
|
|
|
839
|
+ atomically (modifyTVar pendingReplies (IntMap.delete uq))
|
|
838
|
840
|
|
|
839
|
841
|
-- | Receive a single asynchronous request and send a matching reply
|
|
840
|
842
|
-- asynchronously (the handler may fork a computation that eventually replies
|
| ... |
... |
@@ -845,7 +847,7 @@ remoteAsyncReq pipe msg = do |
|
845
|
847
|
-- reply more than once). Linear types enforce the handler replies in every
|
|
846
|
848
|
-- branch, and if the handler otherwise throws, it is handled by 'withAsyncHandler'.
|
|
847
|
849
|
handleAsyncReq :: Pipe -> Get a -> (WriteReply Put %1 -> a -> IO r) -> IO r
|
|
848
|
|
-handleAsyncReq pipe getit k = do
|
|
|
850
|
+handleAsyncReq pipe getit k = mask $ \restore -> do
|
|
849
|
851
|
ack <- newEmptyMVar
|
|
850
|
852
|
|
|
851
|
853
|
-- Block waiting for request
|
| ... |
... |
@@ -870,7 +872,7 @@ handleAsyncReq pipe getit k = do |
|
870
|
872
|
reply (Left (toException exc))
|
|
871
|
873
|
throwIO exc
|
|
872
|
874
|
Right (_, _, it) ->
|
|
873
|
|
- k (WriteReply reply) it
|
|
|
875
|
+ restore (k (WriteReply reply) it)
|
|
874
|
876
|
|
|
875
|
877
|
-- If there's an exception before the reply is sent, reply with an error instead.
|
|
876
|
878
|
`catch` \(e :: SomeException) -> do
|
| ... |
... |
@@ -934,7 +936,7 @@ instance Binary IddMsg where |
|
934
|
936
|
case b of
|
|
935
|
937
|
0 -> Request <$> get <*> get
|
|
936
|
938
|
1 -> Reply <$> get <*> get
|
|
937
|
|
- _ -> error $ "impossible: Binary IddMsg - " ++ show b
|
|
|
939
|
+ _ -> fail $ "impossible: Binary IddMsg - " ++ show b
|
|
938
|
940
|
|
|
939
|
941
|
newAsyncManager :: Pipe -> IO AsyncManager
|
|
940
|
942
|
newAsyncManager pipe = do
|
| ... |
... |
@@ -944,7 +946,7 @@ newAsyncManager pipe = do |
|
944
|
946
|
|
|
945
|
947
|
-- Fork thread to drain read end.
|
|
946
|
948
|
-- (cleaned up on 'closePipe')
|
|
947
|
|
- asyncDrainTid <- forkIO (drainPipe pipe)
|
|
|
949
|
+ asyncDrainTid <- forkIOWithUnmask (drainPipe pipe)
|
|
948
|
950
|
|
|
949
|
951
|
let am = AsyncManager{..}
|
|
950
|
952
|
pure am
|
| ... |
... |
@@ -954,8 +956,8 @@ newAsyncManager pipe = do |
|
954
|
956
|
-- In pratice, we'll have both a thread draining the read end of the host's
|
|
955
|
957
|
-- 'Pipe' and a thread draining the read end of the interpreter's 'Pipe', since
|
|
956
|
958
|
-- both processes construct a 'Pipe' and send and receive messages
|
|
957
|
|
-drainPipe :: Pipe -> IO ()
|
|
958
|
|
-drainPipe p = catch loop replyToPendings where
|
|
|
959
|
+drainPipe :: Pipe -> (forall x. IO x -> IO x) -> IO ()
|
|
|
960
|
+drainPipe p unmask = catch (unmask loop) replyToPendings where
|
|
959
|
961
|
loop = do
|
|
960
|
962
|
readPipe p >>= \case
|
|
961
|
963
|
-- Receive a reply to a pending notification
|