| ... |
... |
@@ -314,9 +314,9 @@ handleAsyncReq pipe getit mii k = mask $ \restore -> do |
|
314
|
314
|
Nothing -> tryReadTQueue $ requestsQueue am
|
|
315
|
315
|
Just ii -> tryReadTQueue =<< getMailbox am ii
|
|
316
|
316
|
case taken of
|
|
317
|
|
- Just req@(uq,_) -> fetchedRequest am uq >> pure (Just req)
|
|
318
|
|
- Nothing | isClosed -> pure Nothing -- empty and closed
|
|
319
|
|
- | otherwise -> retry -- wait for msg
|
|
|
317
|
+ Just req@(uq,_) -> pure (Just req)
|
|
|
318
|
+ Nothing | isClosed -> pure Nothing -- empty and closed
|
|
|
319
|
+ | otherwise -> retry -- wait for msg
|
|
320
|
320
|
|
|
321
|
321
|
case mbreq of
|
|
322
|
322
|
Nothing -> throwIO PipeReadClosedException -- can't expect reqs
|
| ... |
... |
@@ -394,8 +394,8 @@ getMailbox am (InboxId ii) = do |
|
394
|
394
|
-- | Tracks handling status of a received 'Request'. The values are states of a
|
|
395
|
395
|
-- tiny state machine used to coordinate cancelations.
|
|
396
|
396
|
--
|
|
397
|
|
--- 'handleAsyncReq' will pop a request from the appropriate queue, setting the
|
|
398
|
|
--- status to `Fetched`. Then one of two things can happen first:
|
|
|
397
|
+-- 'drainPipe' will push a request to the appropriate queue, and set the
|
|
|
398
|
+-- status to `Received`. Then one of two things can happen first:
|
|
399
|
399
|
-- - we receive a 'Cancel', then status is set to `Canceling` so the cancel
|
|
400
|
400
|
-- action will run when set.
|
|
401
|
401
|
-- - handler sets a cancel action, then reception of a `Cancel` runs that
|
| ... |
... |
@@ -404,14 +404,14 @@ getMailbox am (InboxId ii) = do |
|
404
|
404
|
-- If RequestStatus is `Canceling` or absent, then a `Reply` won't be sent.
|
|
405
|
405
|
--
|
|
406
|
406
|
-- The two possible transition paths:
|
|
407
|
|
--- requestsQueue -> Fetched -> Active cancel -> gone
|
|
408
|
|
--- requestsQueue -> Fetched -> Canceling -> gone
|
|
|
407
|
+-- readPipe -> Received -> Active cancel -> gone
|
|
|
408
|
+-- readPipe -> Received -> Canceling -> gone
|
|
409
|
409
|
|
|
410
|
410
|
data RequestStatus
|
|
411
|
|
- = Fetched -- ^ 'Fetched' from a request queue, to pass to handler.
|
|
412
|
|
- | Active (IO ()) -- ^ Handler set a cancel action, and has not sent a reply
|
|
|
411
|
+ = Received -- ^ 'Received' from the 'Pipe'.
|
|
|
412
|
+ | Active (IO ()) -- ^ Handler has set a cancel action, and has not sent a reply
|
|
413
|
413
|
-- yet. The cancel action must not block.
|
|
414
|
|
- | Canceling -- ^ `Cancel` received while status is still `Fetched`.
|
|
|
414
|
+ | Canceling -- ^ `Cancel` received while status is still `Received`.
|
|
415
|
415
|
|
|
416
|
416
|
data AsyncManager = AsyncManager
|
|
417
|
417
|
{ pendingReply :: !(TVar (IntMap (TMVar (Either SomeException LB.ByteString))))
|
| ... |
... |
@@ -530,7 +530,7 @@ drainPipe p unmask = catch (unmask $ loop `finally` pipeCloseRead p) replyToPend |
|
530
|
530
|
putTMVar wv (bimap (toException . RemoteProcessException) id bs)
|
|
531
|
531
|
-- Receive a request and queue it
|
|
532
|
532
|
Request uq mbi bs -> do
|
|
533
|
|
- atomically $ case mbi of
|
|
|
533
|
+ atomically $ receivedRequest am uq >> case mbi of
|
|
534
|
534
|
Nothing ->
|
|
535
|
535
|
-- "Global" request
|
|
536
|
536
|
writeTQueue (requestsQueue am) (uq,bs)
|
| ... |
... |
@@ -548,10 +548,10 @@ cancelRequest :: AsyncManager -> CorrelationId -> IO () |
|
548
|
548
|
cancelRequest am uq = join . atomically $ do
|
|
549
|
549
|
cs <- readTVar (cancelStatus am)
|
|
550
|
550
|
case IntMap.lookup uq cs of
|
|
551
|
|
- Nothing -> pure (pure ())
|
|
552
|
|
- Just Fetched -> writeTVar (cancelStatus am) (IntMap.insert uq Canceling cs) >> pure (pure ())
|
|
|
551
|
+ Nothing -> pure (pure ())
|
|
|
552
|
+ Just Received -> writeTVar (cancelStatus am) (IntMap.insert uq Canceling cs) >> pure (pure ())
|
|
553
|
553
|
Just (Active cancel) -> writeTVar (cancelStatus am) (IntMap.delete uq cs) >> pure cancel
|
|
554
|
|
- Just Canceling -> pure (pure ())
|
|
|
554
|
+ Just Canceling -> pure (pure ())
|
|
555
|
555
|
|
|
556
|
556
|
-- | Returns @True@ when the handler should write its 'Reply'.
|
|
557
|
557
|
doneHandling :: AsyncManager -> CorrelationId -> STM Bool
|
| ... |
... |
@@ -569,19 +569,19 @@ setActive :: AsyncManager -> CorrelationId -> IO () -> IO () |
|
569
|
569
|
setActive am uq cancel = join . atomically $ do
|
|
570
|
570
|
cs <- readTVar (cancelStatus am)
|
|
571
|
571
|
case IntMap.lookup uq cs of
|
|
572
|
|
- Nothing -> pure (pure ()) -- reply already received.
|
|
573
|
|
- Just Fetched -> writeTVar (cancelStatus am) (IntMap.insert uq (Active cancel) cs) >> pure (pure ())
|
|
|
572
|
+ Nothing -> pure (pure ()) -- reply already sent.
|
|
|
573
|
+ Just Received -> writeTVar (cancelStatus am) (IntMap.insert uq (Active cancel) cs) >> pure (pure ())
|
|
574
|
574
|
Just (Active _) -> writeTVar (cancelStatus am) (IntMap.insert uq (Active cancel) cs) >> pure (pure ())
|
|
575
|
|
- Just Canceling -> pure cancel
|
|
|
575
|
+ Just Canceling -> pure cancel
|
|
576
|
576
|
|
|
577
|
|
-fetchedRequest :: AsyncManager -> CorrelationId -> STM ()
|
|
578
|
|
-fetchedRequest am uq = do
|
|
|
577
|
+receivedRequest :: AsyncManager -> CorrelationId -> STM ()
|
|
|
578
|
+receivedRequest am uq = do
|
|
579
|
579
|
cs <- readTVar (cancelStatus am)
|
|
580
|
580
|
case IntMap.lookup uq cs of
|
|
581
|
|
- Nothing -> writeTVar (cancelStatus am) (IntMap.insert uq Fetched cs)
|
|
582
|
|
- Just Fetched -> error "impossible"
|
|
|
581
|
+ Nothing -> writeTVar (cancelStatus am) (IntMap.insert uq Received cs)
|
|
|
582
|
+ Just Received -> error "impossible"
|
|
583
|
583
|
Just (Active _) -> error "impossible"
|
|
584
|
|
- Just Canceling -> error "impossible"
|
|
|
584
|
+ Just Canceling -> error "impossible"
|
|
585
|
585
|
|
|
586
|
586
|
|
|
587
|
587
|
|