| ... |
... |
@@ -164,13 +164,15 @@ writePipe p msg = do |
|
164
|
164
|
var <- atomically $ writePipeAsync p msg
|
|
165
|
165
|
either throwIO pure =<< atomically (takeTMVar var)
|
|
166
|
166
|
|
|
|
167
|
+-- | Enqueues the message for writing.
|
|
|
168
|
+--
|
|
|
169
|
+-- The returned TMVar can be checked for the outcome.
|
|
167
|
170
|
writePipeAsync :: Pipe -> IddMsg -> STM (TMVar (Either SomeException ()))
|
|
168
|
171
|
writePipeAsync Pipe{..} msg = do
|
|
169
|
172
|
var <- newEmptyTMVar
|
|
170
|
173
|
readTVar pipeWriteClosed >>= \case
|
|
171
|
|
- True -> throwSTM PipeWriteClosedException
|
|
172
|
|
- False -> pure ()
|
|
173
|
|
- writeTQueue (writeQueue pipeAsyncMngr) (msg, var)
|
|
|
174
|
+ True -> putTMVar var (Left $ toException PipeWriteClosedException)
|
|
|
175
|
+ False -> writeTQueue (writeQueue pipeAsyncMngr) (msg, var)
|
|
174
|
176
|
pure var
|
|
175
|
177
|
|
|
176
|
178
|
|
| ... |
... |
@@ -217,11 +219,12 @@ withLock Pipe{..} = bracket (takeMVar pipeLock) (putMVar pipeLock) . const |
|
217
|
219
|
-- * Matching requests with async replies
|
|
218
|
220
|
--------------------------------------------------------------------------------
|
|
219
|
221
|
|
|
220
|
|
--- | Send an asynchronous request and block waiting for a matching reply
|
|
|
222
|
+-- | Allocates a new 'pendingReply' and takes care of deallocation if the
|
|
|
223
|
+-- continuation throws an exception.
|
|
221
|
224
|
--
|
|
222
|
|
--- See Note [Exceptions during serialization] about exceptions the 'Put' argument might raise.
|
|
223
|
|
-remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString)
|
|
224
|
|
-remoteAsyncReq pipe msg ii = do
|
|
|
225
|
+-- Will throw 'PipeReadClosedException' if that is the case.
|
|
|
226
|
+newPendingReply :: Pipe -> (Int -> TMVar (Either SomeException LB.ByteString) -> IO b) -> (IO b)
|
|
|
227
|
+newPendingReply pipe k = do
|
|
225
|
228
|
|
|
226
|
229
|
let pendingReplies = pendingReply (pipeAsyncMngr pipe)
|
|
227
|
230
|
|
| ... |
... |
@@ -229,33 +232,39 @@ remoteAsyncReq pipe msg ii = do |
|
229
|
232
|
uq <- freshId (pipeAsyncMngr pipe)
|
|
230
|
233
|
wv <- newEmptyTMVarIO
|
|
231
|
234
|
|
|
232
|
|
- isClosed <- atomically $ do
|
|
233
|
|
- isClosed <- readTVar (pipeReadClosed pipe)
|
|
234
|
|
- -- don't send the req if we won't get a reply.
|
|
235
|
|
- if isClosed
|
|
236
|
|
- then pure True
|
|
237
|
|
- else do
|
|
238
|
|
- modifyTVar pendingReplies (IntMap.insert uq wv)
|
|
239
|
|
- pure False
|
|
|
235
|
+ bracketOnError
|
|
|
236
|
+ (atomically $ do
|
|
|
237
|
+ readTVar (pipeReadClosed pipe) >>= \case
|
|
|
238
|
+ -- don't set a reply as pending if we won't receive it.
|
|
|
239
|
+ True -> throwSTM PipeReadClosedException
|
|
|
240
|
+ False -> modifyTVar pendingReplies (IntMap.insert uq wv))
|
|
240
|
241
|
|
|
241
|
|
- when isClosed $ do
|
|
242
|
|
- throwIO PipeReadClosedException
|
|
|
242
|
+ (const $ atomically $ modifyTVar pendingReplies (IntMap.delete uq))
|
|
|
243
|
+
|
|
|
244
|
+ (const $ k uq wv)
|
|
|
245
|
+
|
|
|
246
|
+-- | Send an asynchronous request and block waiting for a matching reply
|
|
|
247
|
+--
|
|
|
248
|
+-- See Note [Exceptions during serialization] about exceptions the 'Put' argument might raise.
|
|
|
249
|
+remoteAsyncReq :: Pipe -> Put -> (Maybe InboxId) -> IO (Either SomeException LB.ByteString)
|
|
|
250
|
+remoteAsyncReq pipe msg ii = newPendingReply pipe $ \ uq wv ->
|
|
|
251
|
+ -- We might send a Cancel even if the request never got sent, but it will just
|
|
|
252
|
+ -- get ignored in that case.
|
|
|
253
|
+ handle (\e -> sendCancelMaybe uq e >> throwIO e) $ do
|
|
243
|
254
|
|
|
244
|
255
|
-- Write request
|
|
245
|
256
|
writePipe pipe (Request uq ii (runPut msg))
|
|
246
|
|
- `onException`
|
|
247
|
|
- -- the req wasn't done, or no one will be reading the reply:
|
|
248
|
|
- -- in either case, drop it from the pending map.
|
|
249
|
|
- atomically (modifyTVar pendingReplies (IntMap.delete uq))
|
|
250
|
257
|
|
|
251
|
258
|
-- Block waiting for reply
|
|
252
|
259
|
atomically (takeTMVar wv)
|
|
253
|
|
- `onException`
|
|
254
|
|
- atomically (do
|
|
255
|
|
- modifyTVar pendingReplies (IntMap.delete uq)
|
|
256
|
|
- -- we don't wait for Cancel to be written on the pipe,
|
|
257
|
|
- -- and won't get a reply: nowhere to report it anyway.
|
|
258
|
|
- writePipeAsync pipe (Cancel uq))
|
|
|
260
|
+ where
|
|
|
261
|
+ sendCancelMaybe uq e
|
|
|
262
|
+ | Just PipeWriteClosedException <- fromException e = pure ()
|
|
|
263
|
+ | otherwise = sendCancel uq
|
|
|
264
|
+ sendCancel uq = void . atomically $ do
|
|
|
265
|
+ -- we don't wait for Cancel to be written on the pipe,
|
|
|
266
|
+ -- and won't get a reply: nowhere to report it anyway.
|
|
|
267
|
+ writePipeAsync pipe (Cancel uq)
|
|
259
|
268
|
|
|
260
|
269
|
-- | Receive a single asynchronous request and send a matching reply
|
|
261
|
270
|
-- asynchronously (the handler may fork a computation that eventually replies
|