dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-25 16:10:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

SMP client: dont block on writing to sending queues (#1454)

Browse Source 
* SMP client: dont block on writing to sending queues

* only fork if full
This commit is contained in:
Evgeny
2025-02-15 11:31:05 +00:00
committed by GitHub
parent 560b257af7
commit 55ff581655
1 changed files with 9 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Client.hs
+9 -4
View File
@@ -105,7 +105,7 @@ module Simplex.Messaging.Client
where
import Control.Applicative ((<|>))
import Control.Concurrent (ThreadId, forkFinally, killThread, mkWeakThreadId)
import Control.Concurrent (ThreadId, forkFinally, forkIO, killThread, mkWeakThreadId)
import Control.Concurrent.Async
import Control.Concurrent.STM
import Control.Exception
@@ -1086,11 +1086,11 @@ sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
pure [Response entityId $ Left $ PCETransportError e]
TBTransmissions s n rs
| n > 0 -> do
atomically $ writeTBQueue sndQ (Nothing, s) -- do not expire batched responses
nonBlockingWriteTBQueue sndQ (Nothing, s) -- do not expire batched responses
mapConcurrently (getResponse c Nothing) rs
| otherwise -> pure []
TBTransmission s r -> do
atomically $ writeTBQueue sndQ (Nothing, s)
nonBlockingWriteTBQueue sndQ (Nothing, s)
(: []) <$> getResponse c Nothing r
-- | Send Protocol command
@@ -1112,13 +1112,18 @@ sendProtocolCommand_ c@ProtocolClient {client_ = PClient {sndQ}, thParams = THan
Right t
| B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
| otherwise -> do
atomically $ writeTBQueue sndQ (Just r, s)
nonBlockingWriteTBQueue sndQ (Just r, s)
response <$> getResponse c tOut r
where
s
| batch = tEncodeBatch1 t
| otherwise = tEncode t
nonBlockingWriteTBQueue :: TBQueue a -> a -> IO ()
nonBlockingWriteTBQueue q x = do
sent <- atomically $ ifM (isFullTBQueue q) (pure False) (writeTBQueue q x $> True)
unless sent $ void $ forkIO $ atomically $ writeTBQueue q x
getResponse :: ProtocolClient v err msg -> Maybe Int -> Request err msg -> IO (Response err msg)
getResponse ProtocolClient {client_ = PClient {tcpTimeout, timeoutErrorCount}} tOut Request {entityId, pending, responseVar} = do
r <- fromMaybe tcpTimeout tOut `timeout` atomically (takeTMVar responseVar)