From 7762b4d364e9e1b0217fbdccd994197f6f60898a Mon Sep 17 00:00:00 2001 From: Evgeny Date: Fri, 9 Aug 2024 16:38:46 +0100 Subject: [PATCH] reduce the number of STM transactions (#1255) * reduce the number of STM transactions * reduce STM more * refactor * remove more --- src/Simplex/FileTransfer/Agent.hs | 44 ++-- src/Simplex/FileTransfer/Client/Agent.hs | 4 +- src/Simplex/FileTransfer/Client/Main.hs | 6 +- src/Simplex/FileTransfer/Server.hs | 4 +- src/Simplex/FileTransfer/Server/Env.hs | 13 +- src/Simplex/FileTransfer/Server/Stats.hs | 44 ++-- src/Simplex/FileTransfer/Server/Store.hs | 8 +- src/Simplex/Messaging/Agent.hs | 32 +-- src/Simplex/Messaging/Agent/Client.hs | 204 +++++++++--------- src/Simplex/Messaging/Agent/Env/SQLite.hs | 24 +-- .../Messaging/Agent/NtfSubSupervisor.hs | 6 +- src/Simplex/Messaging/Agent/Store/SQLite.hs | 9 +- .../Messaging/Agent/Store/SQLite/Common.hs | 2 +- .../Messaging/Agent/Store/SQLite/DB.hs | 2 +- src/Simplex/Messaging/Agent/TRcvQueues.hs | 12 +- src/Simplex/Messaging/Client.hs | 50 ++--- src/Simplex/Messaging/Client/Agent.hs | 33 +-- src/Simplex/Messaging/Notifications/Server.hs | 10 +- .../Messaging/Notifications/Server/Env.hs | 51 +++-- .../Messaging/Notifications/Server/Stats.hs | 36 ++-- .../Messaging/Notifications/Server/Store.hs | 15 +- src/Simplex/Messaging/Server.hs | 16 +- src/Simplex/Messaging/Server/Env/STM.hs | 62 +++--- src/Simplex/Messaging/Server/MsgStore/STM.hs | 4 +- .../Messaging/Server/QueueStore/STM.hs | 8 +- src/Simplex/Messaging/Server/Stats.hs | 188 ++++++++-------- src/Simplex/Messaging/TMap.hs | 18 +- src/Simplex/Messaging/Transport.hs | 2 +- src/Simplex/Messaging/Transport/Buffer.hs | 6 +- src/Simplex/Messaging/Transport/HTTP2.hs | 2 +- .../Messaging/Transport/HTTP2/Client.hs | 8 +- src/Simplex/Messaging/Transport/Server.hs | 10 +- tests/CoreTests/BatchingTests.hs | 4 +- tests/CoreTests/TRcvQueuesTests.hs | 32 +-- 34 files changed, 495 insertions(+), 474 deletions(-) diff --git a/src/Simplex/FileTransfer/Agent.hs b/src/Simplex/FileTransfer/Agent.hs index c82316c82..d6ee75ae9 100644 --- a/src/Simplex/FileTransfer/Agent.hs +++ b/src/Simplex/FileTransfer/Agent.hs @@ -184,7 +184,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do cfg <- asks config forever $ do lift $ waitForWork doWork - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c runXFTPOperation cfg where runXFTPOperation :: AgentConfig -> AM () @@ -194,7 +194,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do (fc@RcvFileChunk {userId, rcvFileId, rcvFileEntityId, digest, fileTmpPath, replicas = replica@RcvFileChunkReplica {rcvChunkReplicaId, server, delay} : _}, approvedRelays) -> do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv downloadAttempts downloadFileChunk fc replica approvedRelays @@ -205,7 +205,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do when (serverHostError e) $ notify c rcvFileEntityId $ RFWARN e liftIO $ closeXFTPServerClient c userId server digest withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c loop retryDone e = do atomically . incXFTPServerStat c userId srv $ case e of @@ -221,7 +221,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do chunkSpec = XFTPRcvChunkSpec chunkPath chSize (unFileDigest digest) relChunkPath = fileTmpPath takeFileName chunkPath agentXFTPDownloadChunk c userId digest replica chunkSpec - atomically $ waitUntilForeground c + liftIO $ waitUntilForeground c (entityId, complete, progress) <- withStore c $ \db -> runExceptT $ do liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath RcvFile {size = FileSize currentSize, chunks, redirect} <- ExceptT $ getRcvFile db rcvFileId @@ -240,7 +240,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do where ipAddressProtected' :: AM Bool ipAddressProtected' = do - cfg <- liftIO $ getNetworkConfig' c + cfg <- liftIO $ getFastNetworkConfig c pure $ ipAddressProtected cfg srv receivedSize :: [RcvFileChunk] -> Int64 receivedSize = foldl' (\sz ch -> sz + receivedChunkSize ch) 0 @@ -273,7 +273,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do cfg <- asks config forever $ do lift $ waitForWork doWork - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c runXFTPOperation cfg where runXFTPOperation :: AgentConfig -> AM () @@ -299,12 +299,12 @@ runXFTPRcvLocalWorker c Worker {doWork} = do Nothing -> do notify c rcvFileEntityId $ RFDONE fsSavePath lift $ forM_ tmpPath (removePath <=< toFSFilePath) - atomically $ waitUntilForeground c + liftIO $ waitUntilForeground c withStore' c (`updateRcvFileComplete` rcvFileId) Just RcvFileRedirect {redirectFileInfo, redirectDbId} -> do let RedirectFileInfo {size = redirectSize, digest = redirectDigest} = redirectFileInfo lift $ forM_ tmpPath (removePath <=< toFSFilePath) - atomically $ waitUntilForeground c + liftIO $ waitUntilForeground c withStore' c (`updateRcvFileComplete` rcvFileId) -- proceed with redirect yaml <- liftError (FILE . FILE_IO . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath) @@ -392,7 +392,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do cfg <- asks config forever $ do lift $ waitForWork doWork - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c runXFTPOperation cfg where runXFTPOperation :: AgentConfig -> AM () @@ -454,7 +454,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do SndFileChunkReplica {server} : _ -> Right server createChunk :: Int -> SndFileChunk -> AM (ProtocolServer 'PXFTP) createChunk numRecipients' ch = do - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c (replica, ProtoServerWithAuth srv _) <- tryCreate withStore' c $ \db -> createSndFileReplica db ch replica pure srv @@ -462,9 +462,9 @@ runXFTPSndPrepareWorker c Worker {doWork} = do tryCreate = do usedSrvs <- newTVarIO ([] :: [XFTPServer]) let AgentClient {xftpServers} = c - userSrvCount <- length <$> atomically (TM.lookup userId xftpServers) + userSrvCount <- liftIO $ length <$> TM.lookupIO userId xftpServers withRetryIntervalCount (riFast ri) $ \n _ loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c let triedAllSrvs = n > userSrvCount createWithNextSrv usedSrvs @@ -474,7 +474,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do retryLoop loop triedAllSrvs e = do flip catchAgentError (\_ -> pure ()) $ do when (triedAllSrvs && serverHostError e) $ notify c sndFileEntityId $ SFWARN e - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c loop createWithNextSrv usedSrvs = do deleted <- withStore' c $ \db -> getSndFileDeleted db sndFileId @@ -494,7 +494,7 @@ runXFTPSndWorker c srv Worker {doWork} = do cfg <- asks config forever $ do lift $ waitForWork doWork - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c runXFTPOperation cfg where runXFTPOperation :: AgentConfig -> AM () @@ -504,7 +504,7 @@ runXFTPSndWorker c srv Worker {doWork} = do fc@SndFileChunk {userId, sndFileId, sndFileEntityId, filePrefixPath, digest, replicas = replica@SndFileChunkReplica {sndChunkReplicaId, server, delay} : _} -> do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv uploadAttempts uploadFileChunk cfg fc replica @@ -515,7 +515,7 @@ runXFTPSndWorker c srv Worker {doWork} = do when (serverHostError e) $ notify c sndFileEntityId $ SFWARN e liftIO $ closeXFTPServerClient c userId server digest withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c loop retryDone e = do atomically $ incXFTPServerStat c userId srv uploadErrs @@ -526,9 +526,9 @@ runXFTPSndWorker c srv Worker {doWork} = do fsFilePath <- lift $ toFSFilePath filePath unlessM (doesFileExist fsFilePath) $ throwE $ FILE NO_FILE let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c agentXFTPUploadChunk c userId chunkDigest replica' chunkSpec' - atomically $ waitUntilForeground c + liftIO $ waitUntilForeground c sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded getSndFile db sndFileId @@ -666,7 +666,7 @@ runXFTPDelWorker c srv Worker {doWork} = do cfg <- asks config forever $ do lift $ waitForWork doWork - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c runXFTPOperation cfg where runXFTPOperation :: AgentConfig -> AM () @@ -677,7 +677,7 @@ runXFTPDelWorker c srv Worker {doWork} = do processDeletedReplica replica@DeletedSndChunkReplica {deletedSndChunkReplicaId, userId, server, chunkDigest, delay} = do let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay withRetryIntervalLimit xftpConsecutiveRetries ri' $ \delay' loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c atomically $ incXFTPServerStat c userId srv deleteAttempts deleteChunkReplica @@ -688,7 +688,7 @@ runXFTPDelWorker c srv Worker {doWork} = do when (serverHostError e) $ notify c "" $ SFWARN e liftIO $ closeXFTPServerClient c userId server chunkDigest withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay - atomically $ assertAgentForeground c + liftIO $ assertAgentForeground c loop retryDone e = do atomically $ incXFTPServerStat c userId srv deleteErrs @@ -703,7 +703,7 @@ delWorkerInternalError c deletedSndChunkReplicaId e = do withStore' c $ \db -> deleteDeletedSndChunkReplica db deletedSndChunkReplicaId notify c "" $ SFERR e -assertAgentForeground :: AgentClient -> STM () +assertAgentForeground :: AgentClient -> IO () assertAgentForeground c = do throwWhenInactive c waitUntilForeground c diff --git a/src/Simplex/FileTransfer/Client/Agent.hs b/src/Simplex/FileTransfer/Client/Agent.hs index 86b093ee7..863a91ce1 100644 --- a/src/Simplex/FileTransfer/Client/Agent.hs +++ b/src/Simplex/FileTransfer/Client/Agent.hs @@ -53,9 +53,9 @@ defaultXFTPClientAgentConfig = data XFTPClientAgentError = XFTPClientAgentError XFTPServer XFTPClientError deriving (Show, Exception) -newXFTPAgent :: XFTPClientAgentConfig -> STM XFTPClientAgent +newXFTPAgent :: XFTPClientAgentConfig -> IO XFTPClientAgent newXFTPAgent config = do - xftpClients <- TM.empty + xftpClients <- TM.emptyIO pure XFTPClientAgent {xftpClients, config} type ME a = ExceptT XFTPClientAgentError IO a diff --git a/src/Simplex/FileTransfer/Client/Main.hs b/src/Simplex/FileTransfer/Client/Main.hs index fee44832a..1eea6ef5a 100644 --- a/src/Simplex/FileTransfer/Client/Main.hs +++ b/src/Simplex/FileTransfer/Client/Main.hs @@ -313,7 +313,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re pure (encPath, fdRcv, fdSnd, chunkSpecs, encSize) uploadFile :: TVar ChaChaDRG -> [XFTPChunkSpec] -> TVar [Int64] -> Int64 -> ExceptT CLIError IO [SentFileChunk] uploadFile g chunks uploadedChunks encSize = do - a <- atomically $ newXFTPAgent defaultXFTPClientAgentConfig + a <- liftIO $ newXFTPAgent defaultXFTPClientAgentConfig gen <- newTVarIO =<< liftIO newStdGen let xftpSrvs = fromMaybe defaultXFTPServers (nonEmpty xftpServers) srvs <- liftIO $ replicateM (length chunks) $ getXFTPServer gen xftpSrvs @@ -429,7 +429,7 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath, receive (ValidFileDescription FileDescription {size, digest, key, nonce, chunks}) = do encPath <- getEncPath tempPath "xftp" createDirectory encPath - a <- atomically $ newXFTPAgent defaultXFTPClientAgentConfig + a <- liftIO $ newXFTPAgent defaultXFTPClientAgentConfig liftIO $ printNoNewLine "Downloading file..." downloadedChunks <- newTVarIO [] let srv FileChunk {replicas} = case replicas of @@ -494,7 +494,7 @@ cliDeleteFile DeleteOptions {fileDescription, retryCount, yes} = do where deleteFile :: ValidFileDescription 'FSender -> ExceptT CLIError IO () deleteFile (ValidFileDescription FileDescription {chunks}) = do - a <- atomically $ newXFTPAgent defaultXFTPClientAgentConfig + a <- liftIO $ newXFTPAgent defaultXFTPClientAgentConfig forM_ chunks $ deleteFileChunk a liftIO $ do printNoNewLine "File deleted!" diff --git a/src/Simplex/FileTransfer/Server.hs b/src/Simplex/FileTransfer/Server.hs index 24dcc5e38..819be9a81 100644 --- a/src/Simplex/FileTransfer/Server.hs +++ b/src/Simplex/FileTransfer/Server.hs @@ -112,7 +112,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira Right pk' -> pure pk' Left e -> putStrLn ("servers has no valid key: " <> show e) >> exitFailure env <- ask - sessions <- atomically TM.empty + sessions <- liftIO TM.emptyIO let cleanup sessionId = atomically $ TM.delete sessionId sessions liftIO . runHTTP2Server started xftpPort defaultHTTP2BufferSize serverParams transportConfig inactiveClientExpiration cleanup $ \sessionId sessionALPN r sendResponse -> do reqBody <- getHTTP2Body r xftpBlockSize @@ -576,7 +576,7 @@ incFileStat statSel = do saveServerStats :: M () saveServerStats = asks (serverStatsBackupFile . config) - >>= mapM_ (\f -> asks serverStats >>= atomically . getFileServerStatsData >>= liftIO . saveStats f) + >>= mapM_ (\f -> asks serverStats >>= liftIO . getFileServerStatsData >>= liftIO . saveStats f) where saveStats f stats = do logInfo $ "saving server stats to file " <> T.pack f diff --git a/src/Simplex/FileTransfer/Server/Env.hs b/src/Simplex/FileTransfer/Server/Env.hs index f8a6bc996..1fa399a2a 100644 --- a/src/Simplex/FileTransfer/Server/Env.hs +++ b/src/Simplex/FileTransfer/Server/Env.hs @@ -11,7 +11,6 @@ module Simplex.FileTransfer.Server.Env where import Control.Logger.Simple import Control.Monad -import Control.Monad.IO.Unlift import Crypto.Random import Data.Int (Int64) import Data.List.NonEmpty (NonEmpty) @@ -105,17 +104,17 @@ supportedXFTPhandshakes = ["xftp/1"] newXFTPServerEnv :: XFTPServerConfig -> IO XFTPEnv newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do - random <- liftIO C.newRandom - store <- atomically newFileStore - storeLog <- liftIO $ mapM (`readWriteFileStore` store) storeLogFile + random <- C.newRandom + store <- newFileStore + storeLog <- mapM (`readWriteFileStore` store) storeLogFile used <- countUsedStorage <$> readTVarIO (files store) atomically $ writeTVar (usedStorage store) used forM_ fileSizeQuota $ \quota -> do logInfo $ "Total / available storage: " <> tshow quota <> " / " <> tshow (quota - used) when (quota < used) $ logInfo "WARNING: storage quota is less than used storage, no files can be uploaded!" - tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig) - Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile - serverStats <- atomically . newFileServerStats =<< liftIO getCurrentTime + tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig) + Fingerprint fp <- loadFingerprint caCertificateFile + serverStats <- newFileServerStats =<< getCurrentTime pure XFTPEnv {config, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats} countUsedStorage :: M.Map k FileRec -> Int64 diff --git a/src/Simplex/FileTransfer/Server/Stats.hs b/src/Simplex/FileTransfer/Server/Stats.hs index 08813dc2a..1178dd5f6 100644 --- a/src/Simplex/FileTransfer/Server/Stats.hs +++ b/src/Simplex/FileTransfer/Server/Stats.hs @@ -43,34 +43,34 @@ data FileServerStatsData = FileServerStatsData } deriving (Show) -newFileServerStats :: UTCTime -> STM FileServerStats +newFileServerStats :: UTCTime -> IO FileServerStats newFileServerStats ts = do - fromTime <- newTVar ts - filesCreated <- newTVar 0 - fileRecipients <- newTVar 0 - filesUploaded <- newTVar 0 - filesExpired <- newTVar 0 - filesDeleted <- newTVar 0 + fromTime <- newTVarIO ts + filesCreated <- newTVarIO 0 + fileRecipients <- newTVarIO 0 + filesUploaded <- newTVarIO 0 + filesExpired <- newTVarIO 0 + filesDeleted <- newTVarIO 0 filesDownloaded <- newPeriodStats - fileDownloads <- newTVar 0 - fileDownloadAcks <- newTVar 0 - filesCount <- newTVar 0 - filesSize <- newTVar 0 + fileDownloads <- newTVarIO 0 + fileDownloadAcks <- newTVarIO 0 + filesCount <- newTVarIO 0 + filesSize <- newTVarIO 0 pure FileServerStats {fromTime, filesCreated, fileRecipients, filesUploaded, filesExpired, filesDeleted, filesDownloaded, fileDownloads, fileDownloadAcks, filesCount, filesSize} -getFileServerStatsData :: FileServerStats -> STM FileServerStatsData +getFileServerStatsData :: FileServerStats -> IO FileServerStatsData getFileServerStatsData s = do - _fromTime <- readTVar $ fromTime (s :: FileServerStats) - _filesCreated <- readTVar $ filesCreated s - _fileRecipients <- readTVar $ fileRecipients s - _filesUploaded <- readTVar $ filesUploaded s - _filesExpired <- readTVar $ filesExpired s - _filesDeleted <- readTVar $ filesDeleted s + _fromTime <- readTVarIO $ fromTime (s :: FileServerStats) + _filesCreated <- readTVarIO $ filesCreated s + _fileRecipients <- readTVarIO $ fileRecipients s + _filesUploaded <- readTVarIO $ filesUploaded s + _filesExpired <- readTVarIO $ filesExpired s + _filesDeleted <- readTVarIO $ filesDeleted s _filesDownloaded <- getPeriodStatsData $ filesDownloaded s - _fileDownloads <- readTVar $ fileDownloads s - _fileDownloadAcks <- readTVar $ fileDownloadAcks s - _filesCount <- readTVar $ filesCount s - _filesSize <- readTVar $ filesSize s + _fileDownloads <- readTVarIO $ fileDownloads s + _fileDownloadAcks <- readTVarIO $ fileDownloadAcks s + _filesCount <- readTVarIO $ filesCount s + _filesSize <- readTVarIO $ filesSize s pure FileServerStatsData {_fromTime, _filesCreated, _fileRecipients, _filesUploaded, _filesExpired, _filesDeleted, _filesDownloaded, _fileDownloads, _fileDownloadAcks, _filesCount, _filesSize} setFileServerStats :: FileServerStats -> FileServerStatsData -> STM () diff --git a/src/Simplex/FileTransfer/Server/Store.hs b/src/Simplex/FileTransfer/Server/Store.hs index aa8eaa932..b56b516aa 100644 --- a/src/Simplex/FileTransfer/Server/Store.hs +++ b/src/Simplex/FileTransfer/Server/Store.hs @@ -55,11 +55,11 @@ instance StrEncoding FileRecipient where strEncode (FileRecipient rId rKey) = strEncode rId <> ":" <> strEncode rKey strP = FileRecipient <$> strP <* A.char ':' <*> strP -newFileStore :: STM FileStore +newFileStore :: IO FileStore newFileStore = do - files <- TM.empty - recipients <- TM.empty - usedStorage <- newTVar 0 + files <- TM.emptyIO + recipients <- TM.emptyIO + usedStorage <- newTVarIO 0 pure FileStore {files, recipients, usedStorage} addFile :: FileStore -> SenderId -> FileInfo -> SystemTime -> STM (Either XFTPErrorType ()) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index f7e70b721..bdf83c8d6 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -207,7 +207,7 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp} stor runAgent = do liftIO $ checkServers "SMP" smp >> checkServers "XFTP" xftp currentTs <- liftIO getCurrentTime - c@AgentClient {acThread} <- atomically . newAgentClient clientId initServers currentTs =<< ask + c@AgentClient {acThread} <- liftIO . newAgentClient clientId initServers currentTs =<< ask t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c) atomically . writeTVar acThread . Just =<< mkWeakThreadId t pure c @@ -235,15 +235,15 @@ logServersStats c = do liftIO $ threadDelay' delay int <- asks (logStatsInterval . config) forever $ do - atomically $ waitUntilActive c + liftIO $ waitUntilActive c saveServersStats c liftIO $ threadDelay' int saveServersStats :: AgentClient -> AM' () saveServersStats c@AgentClient {subQ, smpServersStats, xftpServersStats, ntfServersStats} = do - sss <- mapM (lift . getAgentSMPServerStats) =<< readTVarIO smpServersStats - xss <- mapM (lift . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats - nss <- mapM (lift . getAgentNtfServerStats) =<< readTVarIO ntfServersStats + sss <- mapM (liftIO . getAgentSMPServerStats) =<< readTVarIO smpServersStats + xss <- mapM (liftIO . getAgentXFTPServerStats) =<< readTVarIO xftpServersStats + nss <- mapM (liftIO . getAgentNtfServerStats) =<< readTVarIO ntfServersStats let stats = AgentPersistedServerStats {smpServersStats = sss, xftpServersStats = xss, ntfServersStats = OptionalMap nss} tryAgentError' (withStore' c (`updateServersStats` stats)) >>= \case Left e -> atomically $ writeTBQueue subQ ("", "", AEvt SAEConn $ ERR $ INTERNAL $ show e) @@ -1084,7 +1084,7 @@ sendMessagesB_ c reqs connIds = withConnLocks c connIds "sendMessages" $ do getConn_ db prev req@(connId, _, _, _) = (req,) <$$> if B.null connId - then fromMaybe (Left $ INTERNAL "sendMessagesB_: empty prev connId") <$> atomically (readTVar prev) + then fromMaybe (Left $ INTERNAL "sendMessagesB_: empty prev connId") <$> readTVarIO prev else do conn <- first storeError <$> getConn db connId conn <$ atomically (writeTVar prev $ Just conn) @@ -1136,7 +1136,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do forever $ do atomically $ endAgentOperation c AOSndNetwork lift $ waitForWork doWork - atomically $ throwWhenInactive c + liftIO $ throwWhenInactive c atomically $ beginAgentOperation c AOSndNetwork withWork c doWork (`getPendingServerCommand` server_) $ runProcessCmd (riFast ri) where @@ -1254,7 +1254,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do SomeConn _ conn@DuplexConnection {} -> a conn _ -> internalErr "command requires duplex connection" tryCommand action = withRetryInterval ri $ \_ loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c tryError action >>= \case Left e @@ -1363,8 +1363,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq@SndQueue {userI forever $ do atomically $ endAgentOperation c AOSndNetwork lift $ waitForWork doWork - atomically $ throwWhenInactive c - atomically $ throwWhenNoDelivery c sq + liftIO $ throwWhenInactive c + liftIO $ throwWhenNoDelivery c sq atomically $ beginAgentOperation c AOSndNetwork withWork c doWork (\db -> getPendingQueueMsg db connId sq) $ \(rq_, PendingMsgData {msgId, msgType, msgBody, pqEncryption, msgFlags, msgRetryState, internalTs}) -> do @@ -1372,7 +1372,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq@SndQueue {userI let mId = unId msgId ri' = maybe id updateRetryInterval2 msgRetryState ri withRetryLock2 ri' qLock $ \riState loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c resp <- tryError $ case msgType of AM_CONN_INFO -> sendConfirmation c sq msgBody @@ -1525,7 +1525,7 @@ retrySndOp :: AgentClient -> AM () -> AM () retrySndOp c loop = do -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent atomically $ endAgentOperation c AOSndNetwork - atomically $ throwWhenInactive c + liftIO $ throwWhenInactive c atomically $ beginAgentOperation c AOSndNetwork loop @@ -2030,7 +2030,7 @@ deleteNtfSubs c deleteCmd = do sendNtfConnCommands :: AgentClient -> NtfSupervisorCommand -> AM () sendNtfConnCommands c cmd = do ns <- asks ntfSupervisor - connIds <- atomically $ getSubscriptions c + connIds <- liftIO $ getSubscriptions c forM_ connIds $ \connId -> do withStore' c (`getConnData` connId) >>= \case Just (ConnData {enableNtfs}, _) -> @@ -2113,7 +2113,7 @@ cleanupManager c@AgentClient {subQ} = do int <- asks (cleanupInterval . config) ttl <- asks $ storedMsgDataTTL . config forever $ do - atomically $ waitUntilActive c + liftIO $ waitUntilActive c run ERR deleteConns run ERR $ withStore' c (`deleteRcvMsgHashesExpired` ttl) run ERR $ withStore' c (`deleteSndMsgsExpired` ttl) @@ -2133,7 +2133,7 @@ cleanupManager c@AgentClient {subQ} = do step <- asks $ cleanupStepInterval . config liftIO $ threadDelay step -- we are catching it to avoid CRITICAL errors in tests when this is the only remaining handle to active - waitActive a = liftIO (E.tryAny . atomically $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a) + waitActive a = liftIO (E.tryAny $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a) deleteConns = withLock (deleteLock c) "cleanupManager" $ do void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c @@ -2258,7 +2258,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId ack' <- handleNotifyAck $ case msg' of SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody SMP.ClientRcvMsgQuota {} -> queueDrained >> ack - whenM (atomically $ hasGetLock c rq) $ + whenM (liftIO $ hasGetLock c rq) $ notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg') pure ack' where diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index d3e1f2eca..d5262d2eb 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -118,7 +118,7 @@ module Simplex.Messaging.Agent.Client waitUntilActive, UserNetworkInfo (..), UserNetworkType (..), - getNetworkConfig', + getFastNetworkConfig, waitForUserNetwork, isNetworkOnline, isOnline, @@ -165,7 +165,7 @@ where import Control.Applicative ((<|>)) import Control.Concurrent (ThreadId, forkIO) import Control.Concurrent.Async (Async, uninterruptibleCancel) -import Control.Concurrent.STM (retry, throwSTM) +import Control.Concurrent.STM (retry) import Control.Exception (AsyncException (..), BlockedIndefinitelyOnSTM (..)) import Control.Logger.Simple import Control.Monad @@ -453,47 +453,47 @@ data UserNetworkType = UNNone | UNCellular | UNWifi | UNEthernet | UNOther deriving (Eq, Show) -- | Creates an SMP agent client instance that receives commands and sends responses via 'TBQueue's. -newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Env -> STM AgentClient +newAgentClient :: Int -> InitialAgentServers -> UTCTime -> Env -> IO AgentClient newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs agentEnv = do let cfg = config agentEnv qSize = tbqSize cfg - acThread <- newTVar Nothing - active <- newTVar True - subQ <- newTBQueue qSize - msgQ <- newTBQueue qSize - smpServers <- newTVar $ M.map mkUserServers smp - smpClients <- TM.empty - smpProxiedRelays <- TM.empty - ntfServers <- newTVar ntf - ntfClients <- TM.empty - xftpServers <- newTVar $ M.map mkUserServers xftp - xftpClients <- TM.empty - useNetworkConfig <- newTVar (slowNetworkConfig netCfg, netCfg) - userNetworkInfo <- newTVar $ UserNetworkInfo UNOther True - userNetworkUpdated <- newTVar Nothing - subscrConns <- newTVar S.empty + acThread <- newTVarIO Nothing + active <- newTVarIO True + subQ <- newTBQueueIO qSize + msgQ <- newTBQueueIO qSize + smpServers <- newTVarIO $ M.map mkUserServers smp + smpClients <- TM.emptyIO + smpProxiedRelays <- TM.emptyIO + ntfServers <- newTVarIO ntf + ntfClients <- TM.emptyIO + xftpServers <- newTVarIO $ M.map mkUserServers xftp + xftpClients <- TM.emptyIO + useNetworkConfig <- newTVarIO (slowNetworkConfig netCfg, netCfg) + userNetworkInfo <- newTVarIO $ UserNetworkInfo UNOther True + userNetworkUpdated <- newTVarIO Nothing + subscrConns <- newTVarIO S.empty activeSubs <- RQ.empty pendingSubs <- RQ.empty - removedSubs <- TM.empty - workerSeq <- newTVar 0 - smpDeliveryWorkers <- TM.empty - asyncCmdWorkers <- TM.empty - connCmdsQueued <- TM.empty - ntfNetworkOp <- newTVar $ AgentOpState False 0 - rcvNetworkOp <- newTVar $ AgentOpState False 0 - msgDeliveryOp <- newTVar $ AgentOpState False 0 - sndNetworkOp <- newTVar $ AgentOpState False 0 - databaseOp <- newTVar $ AgentOpState False 0 - agentState <- newTVar ASForeground - getMsgLocks <- TM.empty - connLocks <- TM.empty - invLocks <- TM.empty - deleteLock <- createLock - smpSubWorkers <- TM.empty - smpServersStats <- TM.empty - xftpServersStats <- TM.empty - ntfServersStats <- TM.empty - srvStatsStartedAt <- newTVar currentTs + removedSubs <- TM.emptyIO + workerSeq <- newTVarIO 0 + smpDeliveryWorkers <- TM.emptyIO + asyncCmdWorkers <- TM.emptyIO + connCmdsQueued <- TM.emptyIO + ntfNetworkOp <- newTVarIO $ AgentOpState False 0 + rcvNetworkOp <- newTVarIO $ AgentOpState False 0 + msgDeliveryOp <- newTVarIO $ AgentOpState False 0 + sndNetworkOp <- newTVarIO $ AgentOpState False 0 + databaseOp <- newTVarIO $ AgentOpState False 0 + agentState <- newTVarIO ASForeground + getMsgLocks <- TM.emptyIO + connLocks <- TM.emptyIO + invLocks <- TM.emptyIO + deleteLock <- atomically createLock + smpSubWorkers <- TM.emptyIO + smpServersStats <- TM.emptyIO + xftpServersStats <- TM.emptyIO + ntfServersStats <- TM.emptyIO + srvStatsStartedAt <- newTVarIO currentTs return AgentClient { acThread, @@ -600,7 +600,7 @@ getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do >>= either newClient (waitForProtocolClient c tSess smpClients) where newClient v = do - prs <- atomically TM.empty + prs <- liftIO TM.emptyIO smpConnectClient c tSess prs v getSMPProxyClient :: AgentClient -> Maybe SMPServerWithAuth -> SMPTransportSession -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay) @@ -618,11 +618,10 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq (tSess,auth,) <$> getSessVar workerSeq tSess smpClients ts newProxyClient :: SMPTransportSession -> Maybe SMP.BasicAuth -> UTCTime -> SMPClientVar -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay) newProxyClient tSess auth ts v = do - (prs, rv) <- atomically $ do - prs <- TM.empty - -- we do not need to check if it is a new proxied relay session, - -- as the client is just created and there are no sessions yet - (prs,) . either id id <$> getSessVar workerSeq destSrv prs ts + prs <- liftIO TM.emptyIO + -- we do not need to check if it is a new proxied relay session, + -- as the client is just created and there are no sessions yet + rv <- atomically $ either id id <$> getSessVar workerSeq destSrv prs ts clnt <- smpConnectClient c tSess prs v (clnt,) <$> newProxiedRelay clnt auth rv waitForProxyClient :: SMPTransportSession -> Maybe SMP.BasicAuth -> SMPClientVar -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay) @@ -648,7 +647,7 @@ getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq pure $ Left e waitForProxiedRelay :: SMPTransportSession -> ProxiedRelayVar -> AM (Either AgentErrorType ProxiedRelay) waitForProxiedRelay (_, srv, _) rv = do - NetworkConfig {tcpConnectTimeout} <- atomically $ getNetworkConfig c + NetworkConfig {tcpConnectTimeout} <- getNetworkConfig c sess_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar rv) pure $ case sess_ of Just (Right sess) -> Right sess @@ -709,7 +708,7 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do where getWorkerVar ts = ifM - (null <$> getPending) + (not <$> RQ.hasSessQueues tSess (pendingSubs c)) (pure Nothing) -- prevent race with cleanup and adding pending queues in another call (Just <$> getSessVar workerSeq tSess smpSubWorkers ts) newSubWorker v = do @@ -718,14 +717,13 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers, workerSeq} tSess = do runSubWorker = do ri <- asks $ reconnectInterval . config withRetryForeground ri isForeground (isNetworkOnline c) $ \_ loop -> do - pending <- atomically getPending + pending <- liftIO $ RQ.getSessQueues tSess $ pendingSubs c forM_ (L.nonEmpty pending) $ \qs -> do - atomically $ waitUntilForeground c + liftIO $ waitUntilForeground c liftIO $ waitForUserNetwork c reconnectSMPClient c tSess qs loop isForeground = (ASForeground ==) <$> readTVar (agentState c) - getPending = RQ.getSessQueues tSess $ pendingSubs c cleanup :: SessionVar (Async ()) -> STM () cleanup v = do -- Here we wait until TMVar is not empty to prevent worker cleanup happening before worker is added to TMVar. @@ -790,7 +788,7 @@ getXFTPServerClient c@AgentClient {active, xftpClients, workerSeq} tSess@(_, srv connectClient :: XFTPClientVar -> AM XFTPClient connectClient v = do cfg <- asks $ xftpCfg . config - xftpNetworkConfig <- atomically $ getNetworkConfig c + xftpNetworkConfig <- getNetworkConfig c liftError' (protocolClientError XFTP $ B.unpack $ strEncode srv) $ X.getXFTPClient tSess cfg {xftpNetworkConfig} $ clientDisconnected v @@ -809,7 +807,7 @@ waitForProtocolClient :: ClientVar msg -> AM (Client msg) waitForProtocolClient c tSess@(_, srv, _) clients v = do - NetworkConfig {tcpConnectTimeout} <- atomically $ getNetworkConfig c + NetworkConfig {tcpConnectTimeout} <- getNetworkConfig c client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v) case client_ of Just (Right smpClient) -> pure smpClient @@ -860,26 +858,26 @@ hostEvent' event = event (AProtocolType $ protocolTypeI @(ProtoType msg)) . clie getClientConfig :: AgentClient -> (AgentConfig -> ProtocolClientConfig v) -> AM' (ProtocolClientConfig v) getClientConfig c cfgSel = do cfg <- asks $ cfgSel . config - networkConfig <- atomically $ getNetworkConfig c + networkConfig <- getNetworkConfig c pure cfg {networkConfig} -getNetworkConfig :: AgentClient -> STM NetworkConfig +getNetworkConfig :: MonadIO m => AgentClient -> m NetworkConfig getNetworkConfig c = do - (slowCfg, fastCfg) <- readTVar (useNetworkConfig c) - UserNetworkInfo {networkType} <- readTVar $ userNetworkInfo c + (slowCfg, fastCfg) <- readTVarIO $ useNetworkConfig c + UserNetworkInfo {networkType} <- readTVarIO $ userNetworkInfo c pure $ case networkType of UNCellular -> slowCfg UNNone -> slowCfg _ -> fastCfg -- returns fast network config -getNetworkConfig' :: AgentClient -> IO NetworkConfig -getNetworkConfig' = fmap snd . readTVarIO . useNetworkConfig -{-# INLINE getNetworkConfig' #-} +getFastNetworkConfig :: AgentClient -> IO NetworkConfig +getFastNetworkConfig = fmap snd . readTVarIO . useNetworkConfig +{-# INLINE getFastNetworkConfig #-} waitForUserNetwork :: AgentClient -> IO () waitForUserNetwork c = - unlessM (atomically $ isNetworkOnline c) $ do + unlessM (isOnline <$> readTVarIO (userNetworkInfo c)) $ do delay <- registerDelay $ userNetworkInterval $ config $ agentEnv c atomically $ unlessM (isNetworkOnline c) $ unlessM (readTVar delay) retry @@ -911,19 +909,18 @@ cancelWorker Worker {doWork, action} = do noWorkToDo doWork atomically (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel) -waitUntilActive :: AgentClient -> STM () -waitUntilActive c = unlessM (readTVar $ active c) retry -{-# INLINE waitUntilActive #-} +waitUntilActive :: AgentClient -> IO () +waitUntilActive AgentClient {active} = unlessM (readTVarIO active) $ atomically $ unlessM (readTVar active) retry -throwWhenInactive :: AgentClient -> STM () -throwWhenInactive c = unlessM (readTVar $ active c) $ throwSTM ThreadKilled +throwWhenInactive :: AgentClient -> IO () +throwWhenInactive c = unlessM (readTVarIO $ active c) $ E.throwIO ThreadKilled {-# INLINE throwWhenInactive #-} -- this function is used to remove workers once delivery is complete, not when it is removed from the map -throwWhenNoDelivery :: AgentClient -> SndQueue -> STM () +throwWhenNoDelivery :: AgentClient -> SndQueue -> IO () throwWhenNoDelivery c sq = - unlessM (TM.member (qAddress sq) $ smpDeliveryWorkers c) $ - throwSTM ThreadKilled + unlessM (TM.memberIO (qAddress sq) $ smpDeliveryWorkers c) $ + E.throwIO ThreadKilled closeProtocolServerClients :: ProtocolServerClient v err msg => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO () closeProtocolServerClients c clientsSel = @@ -949,7 +946,7 @@ closeClient c clientSel tSess = closeClient_ :: ProtocolServerClient v err msg => AgentClient -> ClientVar msg -> IO () closeClient_ c v = do - NetworkConfig {tcpConnectTimeout} <- atomically $ getNetworkConfig c + NetworkConfig {tcpConnectTimeout} <- getNetworkConfig c E.handle (\BlockedIndefinitelyOnSTM -> pure ()) $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v) >>= \case Just (Right client) -> closeProtocolServerClient (protocolClient client) `catchAll_` pure () @@ -1037,7 +1034,7 @@ withLogClient c tSess entId cmdStr action = withLogClient_ c tSess entId cmdStr withSMPClient :: SMPQueueRec q => AgentClient -> q -> ByteString -> (SMPClient -> ExceptT SMPClientError IO a) -> AM a withSMPClient c q cmdStr action = do - tSess <- liftIO $ mkSMPTransportSession c q + tSess <- mkSMPTransportSession c q withLogClient c tSess (queueId q) cmdStr $ action . connectedClient sendOrProxySMPMessage :: AgentClient -> UserId -> SMPServer -> ByteString -> Maybe SMP.SndPrivateAuthKey -> SMP.SenderId -> MsgFlags -> SMP.MsgBody -> AM (Maybe SMPServer) @@ -1062,8 +1059,8 @@ sendOrProxySMPCommand :: (SMPClient -> ExceptT SMPClientError IO ()) -> AM (Maybe SMPServer) sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDirectly = do - sess <- liftIO $ mkTransportSession c userId destSrv senderId - ifM (atomically shouldUseProxy) (sendViaProxy Nothing sess) (sendDirectly sess $> Nothing) + sess <- mkTransportSession c userId destSrv senderId + ifM shouldUseProxy (sendViaProxy Nothing sess) (sendDirectly sess $> Nothing) where shouldUseProxy = do cfg <- getNetworkConfig c @@ -1080,7 +1077,7 @@ sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDi SPFAllow -> True SPFAllowProtected -> ipAddressProtected cfg destSrv SPFProhibit -> False - unknownServer = maybe True (notElem destSrv . knownSrvs) <$> TM.lookup userId (smpServers c) + unknownServer = liftIO $ maybe True (notElem destSrv . knownSrvs) <$> TM.lookupIO userId (smpServers c) sendViaProxy :: Maybe SMPServerWithAuth -> SMPTransportSession -> AM (Maybe SMPServer) sendViaProxy proxySrv_ destSess@(_, _, qId) = do r <- tryAgentError . withProxySession c proxySrv_ destSess senderId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do @@ -1126,7 +1123,7 @@ sendOrProxySMPCommand c userId destSrv cmdStr senderId sendCmdViaProxy sendCmdDi forM_ r' $ \proxySrv -> atomically $ incSMPServerStat c userId proxySrv sentProxied pure r' Left e - | serverHostError e -> ifM (atomically directAllowed) (sendDirectly destSess $> Nothing) (throwE e) + | serverHostError e -> ifM directAllowed (sendDirectly destSess $> Nothing) (throwE e) | otherwise -> throwE e sendDirectly tSess = withLogClient_ c tSess senderId ("SEND " <> cmdStr) $ \(SMPConnectedClient smp _) -> do @@ -1152,7 +1149,7 @@ withXFTPClient :: (Client msg -> ExceptT (ProtocolClientError err) IO b) -> AM b withXFTPClient c (userId, srv, entityId) cmdStr action = do - tSess <- liftIO $ mkTransportSession c userId srv entityId + tSess <- mkTransportSession c userId srv entityId withLogClient c tSess entityId cmdStr action liftClient :: (Show err, Encoding err) => (HostName -> err -> AgentErrorType) -> HostName -> ExceptT (ProtocolClientError err) IO a -> AM a @@ -1224,7 +1221,7 @@ runXFTPServerTest :: AgentClient -> UserId -> XFTPServerWithAuth -> AM' (Maybe P runXFTPServerTest c userId (ProtoServerWithAuth srv auth) = do cfg <- asks $ xftpCfg . config g <- asks random - xftpNetworkConfig <- atomically $ getNetworkConfig c + xftpNetworkConfig <- getNetworkConfig c workDir <- getXFTPWorkPath filePath <- getTempFilePath workDir rcvPath <- getTempFilePath workDir @@ -1295,7 +1292,7 @@ getXFTPWorkPath = do workDir <- readTVarIO =<< asks (xftpWorkDir . xftpAgent) maybe getTemporaryDirectory pure workDir -mkTransportSession :: AgentClient -> UserId -> ProtoServer msg -> EntityId -> IO (TransportSession msg) +mkTransportSession :: MonadIO m => AgentClient -> UserId -> ProtoServer msg -> EntityId -> m (TransportSession msg) mkTransportSession c userId srv entityId = mkTSession userId srv entityId <$> getSessionMode c {-# INLINE mkTransportSession #-} @@ -1303,7 +1300,7 @@ mkTSession :: UserId -> ProtoServer msg -> EntityId -> TransportSessionMode -> T mkTSession userId srv entityId mode = (userId, srv, if mode == TSMEntity then Just entityId else Nothing) {-# INLINE mkTSession #-} -mkSMPTransportSession :: SMPQueueRec q => AgentClient -> q -> IO SMPTransportSession +mkSMPTransportSession :: (SMPQueueRec q, MonadIO m) => AgentClient -> q -> m SMPTransportSession mkSMPTransportSession c q = mkSMPTSession q <$> getSessionMode c {-# INLINE mkSMPTransportSession #-} @@ -1311,8 +1308,8 @@ mkSMPTSession :: SMPQueueRec q => q -> TransportSessionMode -> SMPTransportSessi mkSMPTSession q = mkTSession (qUserId q) (qServer q) (qConnId q) {-# INLINE mkSMPTSession #-} -getSessionMode :: AgentClient -> IO TransportSessionMode -getSessionMode = atomically . fmap sessionMode . getNetworkConfig +getSessionMode :: MonadIO m => AgentClient -> m TransportSessionMode +getSessionMode = fmap sessionMode . getNetworkConfig {-# INLINE getSessionMode #-} newRcvQueue :: AgentClient -> UserId -> ConnId -> SMPServerWithAuth -> VersionRangeSMPC -> SubscriptionMode -> SenderCanSecure -> AM (NewRcvQueue, SMPQueueUri, SMPTransportSession, SessionId) @@ -1323,7 +1320,7 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange subMode sender (dhKey, privDhKey) <- atomically $ C.generateKeyPair g (e2eDhKey, e2ePrivKey) <- atomically $ C.generateKeyPair g logServer "-->" c srv "" "NEW" - tSess <- liftIO $ mkTransportSession c userId srv connId + tSess <- mkTransportSession c userId srv connId (sessId, QIK {rcvId, sndId, rcvPublicDhKey, sndSecure}) <- withClient c tSess $ \(SMPConnectedClient smp _) -> (sessionId $ thParams smp,) <$> createSMPQueue smp rKeys dhKey auth subMode senderCanSecure @@ -1409,7 +1406,7 @@ subscribeQueues c qs = do (errs <> rs,) <$> readTVarIO session where checkQueue rq = do - prohibited <- atomically $ hasGetLock c rq + prohibited <- liftIO $ hasGetLock c rq pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED "subscribeQueues") else Right rq subscribeQueues_ :: Env -> TVar (Maybe SessionId) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ()) subscribeQueues_ env session smp qs' = do @@ -1450,7 +1447,7 @@ sendTSessionBatches statCmd toRQ action c qs = where batchQueues :: AM' [(SMPTransportSession, NonEmpty q)] batchQueues = do - mode <- atomically $ sessionMode <$> getNetworkConfig c + mode <- getSessionMode c pure . M.assocs $ foldl' (batch mode) M.empty qs where batch mode m q = @@ -1511,8 +1508,8 @@ removeSubscription c connId = do RQ.deleteConn connId $ activeSubs c RQ.deleteConn connId $ pendingSubs c -getSubscriptions :: AgentClient -> STM (Set ConnId) -getSubscriptions = readTVar . subscrConns +getSubscriptions :: AgentClient -> IO (Set ConnId) +getSubscriptions = readTVarIO . subscrConns {-# INLINE getSubscriptions #-} logServer :: MonadIO m => ByteString -> AgentClient -> ProtocolServer s -> QueueId -> ByteString -> m () @@ -1611,9 +1608,9 @@ sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do ackSMPMessage smp rcvPrivateKey rcvId msgId atomically $ releaseGetLock c rq -hasGetLock :: AgentClient -> RcvQueue -> STM Bool +hasGetLock :: AgentClient -> RcvQueue -> IO Bool hasGetLock c RcvQueue {server, rcvId} = - TM.member (server, rcvId) $ getMsgLocks c + TM.memberIO (server, rcvId) $ getMsgLocks c releaseGetLock :: AgentClient -> RcvQueue -> STM () releaseGetLock c RcvQueue {server, rcvId} = @@ -1712,7 +1709,7 @@ agentXFTPNewChunk c SndFileChunk {userId, chunkSpec = XFTPChunkSpec {chunkSize}, (sndKey, replicaKey) <- atomically . C.generateAuthKeyPair C.SEd25519 =<< asks random let fileInfo = FileInfo {sndKey, size = chunkSize, digest = chunkDigest} logServer "-->" c srv "" "FNEW" - tSess <- liftIO $ mkTransportSession c userId srv chunkDigest + tSess <- mkTransportSession c userId srv chunkDigest (sndId, rIds) <- withClient c tSess $ \xftp -> X.createXFTPChunk xftp replicaKey fileInfo (L.map fst rKeys) auth logServer "<--" c srv "" $ B.unwords ["SIDS", logSecret sndId] pure NewSndChunkReplica {server = srv, replicaId = ChunkReplicaId sndId, replicaKey, rcvIdsKeys = L.toList $ xftpRcvIdsKeys rIds rKeys} @@ -1865,22 +1862,28 @@ beginAgentOperation c op = do -- unsafeIOToSTM $ putStrLn $ "beginOperation! " <> show op <> " " <> show (opsInProgress s + 1) writeTVar opVar $! s {opsInProgress = opsInProgress s + 1} -agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (AgentClient -> STM ()) -> m a -> m a +agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (AgentClient -> IO ()) -> m a -> m a agentOperationBracket c op check action = E.bracket - (atomically (check c) >> atomically (beginAgentOperation c op)) + (liftIO (check c) >> atomically (beginAgentOperation c op)) (\_ -> atomically $ endAgentOperation c op) (const action) -waitUntilForeground :: AgentClient -> STM () -waitUntilForeground c = unlessM ((ASForeground ==) <$> readTVar (agentState c)) retry -{-# INLINE waitUntilForeground #-} +waitUntilForeground :: AgentClient -> IO () +waitUntilForeground c = + unlessM (foreground readTVarIO) $ atomically $ unlessM (foreground readTVar) retry + where + foreground :: Monad m => (TVar AgentState -> m AgentState) -> m Bool + foreground rd = (ASForeground ==) <$> rd (agentState c) -- This function waits while agent is suspended, but will proceed while it is suspending, -- to allow completing in-flight operations. -waitWhileSuspended :: AgentClient -> STM () -waitWhileSuspended c = unlessM ((ASSuspended /=) <$> readTVar (agentState c)) retry -{-# INLINE waitWhileSuspended #-} +waitWhileSuspended :: AgentClient -> IO () +waitWhileSuspended c = + whenM (suspended readTVarIO) $ atomically $ whenM (suspended readTVar) retry + where + suspended :: Monad m => (TVar AgentState -> m AgentState) -> m Bool + suspended rd = (ASSuspended ==) <$> rd (agentState c) withStore' :: AgentClient -> (DB.Connection -> IO a) -> AM a withStore' c action = withStore c $ fmap Right . action @@ -1951,7 +1954,7 @@ getNextServer c userId usedSrvs = withUserServers c userId $ \srvs -> withUserServers :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> (NonEmpty (ProtoServerWithAuth p) -> AM a) -> AM a withUserServers c userId action = - atomically (TM.lookup userId $ userServers c) >>= \case + liftIO (TM.lookupIO userId $ userServers c) >>= \case Just srvs -> action $ enabledSrvs srvs _ -> throwE $ INTERNAL "unknown userId - no user servers" @@ -1959,11 +1962,10 @@ withNextSrv :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> U withNextSrv c userId usedSrvs initUsed action = do used <- readTVarIO usedSrvs srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId used - atomically $ do - srvs_ <- TM.lookup userId $ userServers c - let unused = maybe [] ((\\ used) . map protoServer . L.toList . enabledSrvs) srvs_ - used' = if null unused then initUsed else srv : used - writeTVar usedSrvs $! used' + srvs_ <- liftIO $ TM.lookupIO userId $ userServers c + let unused = maybe [] ((\\ used) . map protoServer . L.toList . enabledSrvs) srvs_ + used' = if null unused then initUsed else srv : used + atomically $ writeTVar usedSrvs $! used' action srvAuth incSMPServerStat :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats -> TVar Int) -> STM () diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 86203266b..f57cf91e9 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -242,8 +242,8 @@ newSMPAgentEnv :: AgentConfig -> SQLiteStore -> IO Env newSMPAgentEnv config store = do random <- C.newRandom randomServer <- newTVarIO =<< liftIO newStdGen - ntfSupervisor <- atomically . newNtfSubSupervisor $ tbqSize config - xftpAgent <- atomically newXFTPAgent + ntfSupervisor <- newNtfSubSupervisor $ tbqSize config + xftpAgent <- newXFTPAgent multicastSubscribers <- newTMVarIO 0 pure Env {config, store, random, randomServer, ntfSupervisor, xftpAgent, multicastSubscribers} @@ -260,12 +260,12 @@ data NtfSupervisor = NtfSupervisor data NtfSupervisorCommand = NSCCreate | NSCDelete | NSCSmpDelete | NSCNtfWorker NtfServer | NSCNtfSMPWorker SMPServer deriving (Show) -newNtfSubSupervisor :: Natural -> STM NtfSupervisor +newNtfSubSupervisor :: Natural -> IO NtfSupervisor newNtfSubSupervisor qSize = do - ntfTkn <- newTVar Nothing - ntfSubQ <- newTBQueue qSize - ntfWorkers <- TM.empty - ntfSMPWorkers <- TM.empty + ntfTkn <- newTVarIO Nothing + ntfSubQ <- newTBQueueIO qSize + ntfWorkers <- TM.emptyIO + ntfSMPWorkers <- TM.emptyIO pure NtfSupervisor {ntfTkn, ntfSubQ, ntfWorkers, ntfSMPWorkers} data XFTPAgent = XFTPAgent @@ -276,12 +276,12 @@ data XFTPAgent = XFTPAgent xftpDelWorkers :: TMap XFTPServer Worker } -newXFTPAgent :: STM XFTPAgent +newXFTPAgent :: IO XFTPAgent newXFTPAgent = do - xftpWorkDir <- newTVar Nothing - xftpRcvWorkers <- TM.empty - xftpSndWorkers <- TM.empty - xftpDelWorkers <- TM.empty + xftpWorkDir <- newTVarIO Nothing + xftpRcvWorkers <- TM.emptyIO + xftpSndWorkers <- TM.emptyIO + xftpDelWorkers <- TM.emptyIO pure XFTPAgent {xftpWorkDir, xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers} tryAgentError :: AM a -> AM (Either AgentErrorType a) diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index 249aad942..23a88ea70 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -159,7 +159,7 @@ runNtfWorker c srv Worker {doWork} = logInfo $ "runNtfWorker, nextSub " <> tshow nextSub ri <- asks $ reconnectInterval . config withRetryInterval ri $ \_ loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c processSub nextSub `catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show) @@ -244,7 +244,7 @@ runNtfSMPWorker c srv Worker {doWork} = do logInfo $ "runNtfSMPWorker, nextSub " <> tshow nextSub ri <- asks $ reconnectInterval . config withRetryInterval ri $ \_ loop -> do - atomically $ waitWhileSuspended c + liftIO $ waitWhileSuspended c liftIO $ waitForUserNetwork c processSub nextSub `catchAgentError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show) @@ -297,7 +297,7 @@ retryOnError c name loop done e = do where retryLoop = do atomically $ endAgentOperation c AONtfNetwork - atomically $ throwWhenInactive c + liftIO $ throwWhenInactive c atomically $ beginAgentOperation c AONtfNetwork loop diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 0bb80069c..20f382d40 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -392,11 +392,10 @@ connectSQLiteStore dbFilePath key keepKey = do dbNew <- not <$> doesFileExist dbFilePath dbConn <- dbBusyLoop (connectDB dbFilePath key) dbConnection <- newMVar dbConn - atomically $ do - dbKey <- newTVar $! storeKey key keepKey - dbClosed <- newTVar False - dbSem <- newTVar 0 - pure SQLiteStore {dbFilePath, dbKey, dbSem, dbConnection, dbNew, dbClosed} + dbKey <- newTVarIO $! storeKey key keepKey + dbClosed <- newTVarIO False + dbSem <- newTVarIO 0 + pure SQLiteStore {dbFilePath, dbKey, dbSem, dbConnection, dbNew, dbClosed} connectDB :: FilePath -> ScrubbedBytes -> IO DB.Connection connectDB path key = do diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs index 6b121b0b4..a7ad47f37 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Common.hs @@ -47,7 +47,7 @@ withConnectionPriority SQLiteStore {dbSem, dbConnection} priority action lowPriority = wait >> withMVar dbConnection (\db -> ifM free (Just <$> action db) (pure Nothing)) >>= maybe lowPriority pure signal = atomically $ modifyTVar' dbSem (+ 1) release = atomically $ modifyTVar' dbSem $ \sem -> if sem > 0 then sem - 1 else 0 - wait = atomically $ unlessM ((0 ==) <$> readTVar dbSem) retry + wait = unlessM free $ atomically $ unlessM ((0 ==) <$> readTVar dbSem) retry free = (0 ==) <$> readTVarIO dbSem withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs b/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs index 2ae4eb731..b356b3f87 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/DB.hs @@ -64,7 +64,7 @@ timeIt slow sql a = do open :: String -> IO Connection open f = do conn <- SQL.open f - slow <- atomically $ TM.empty + slow <- TM.emptyIO pure Connection {conn, slow} close :: Connection -> IO () diff --git a/src/Simplex/Messaging/Agent/TRcvQueues.hs b/src/Simplex/Messaging/Agent/TRcvQueues.hs index 10e4574cb..3b02f64ae 100644 --- a/src/Simplex/Messaging/Agent/TRcvQueues.hs +++ b/src/Simplex/Messaging/Agent/TRcvQueues.hs @@ -11,6 +11,7 @@ module Simplex.Messaging.Agent.TRcvQueues addQueue, batchAddQueues, deleteQueue, + hasSessQueues, getSessQueues, getDelSessQueues, ) @@ -39,8 +40,8 @@ data TRcvQueues q = TRcvQueues getConnections :: TMap ConnId (NonEmpty (UserId, SMPServer, RecipientId)) } -empty :: STM (TRcvQueues q) -empty = TRcvQueues <$> TM.empty <*> TM.empty +empty :: IO (TRcvQueues q) +empty = TRcvQueues <$> TM.emptyIO <*> TM.emptyIO clear :: TRcvQueues q -> STM () clear (TRcvQueues qs cs) = TM.clear qs >> TM.clear cs @@ -78,8 +79,11 @@ deleteQueue rq (TRcvQueues qs cs) = do delQ = L.nonEmpty . L.filter (/= k) k = qKey rq -getSessQueues :: (UserId, SMPServer, Maybe ConnId) -> TRcvQueues RcvQueue -> STM [RcvQueue] -getSessQueues tSess (TRcvQueues qs _) = M.foldl' addQ [] <$> readTVar qs +hasSessQueues :: (UserId, SMPServer, Maybe ConnId) -> TRcvQueues RcvQueue -> STM Bool +hasSessQueues tSess (TRcvQueues qs _) = any (`isSession` tSess) <$> readTVar qs + +getSessQueues :: (UserId, SMPServer, Maybe ConnId) -> TRcvQueues RcvQueue -> IO [RcvQueue] +getSessQueues tSess (TRcvQueues qs _) = M.foldl' addQ [] <$> readTVarIO qs where addQ qs' rq = if rq `isSession` tSess then rq : qs' else qs' diff --git a/src/Simplex/Messaging/Client.hs b/src/Simplex/Messaging/Client.hs index 80fd65ffc..b4567c62e 100644 --- a/src/Simplex/Messaging/Client.hs +++ b/src/Simplex/Messaging/Client.hs @@ -170,17 +170,17 @@ data PClient v err msg = PClient msgQ :: Maybe (TBQueue (ServerTransmissionBatch v err msg)) } -smpClientStub :: TVar ChaChaDRG -> ByteString -> VersionSMP -> Maybe (THandleAuth 'TClient) -> STM SMPClient +smpClientStub :: TVar ChaChaDRG -> ByteString -> VersionSMP -> Maybe (THandleAuth 'TClient) -> IO SMPClient smpClientStub g sessionId thVersion thAuth = do let ts = UTCTime (read "2024-03-31") 0 - connected <- newTVar False - clientCorrId <- C.newRandomDRG g - sentCommands <- TM.empty - sendPings <- newTVar False - lastReceived <- newTVar ts - timeoutErrorCount <- newTVar 0 - sndQ <- newTBQueue 100 - rcvQ <- newTBQueue 100 + connected <- newTVarIO False + clientCorrId <- atomically $ C.newRandomDRG g + sentCommands <- TM.emptyIO + sendPings <- newTVarIO False + lastReceived <- newTVarIO ts + timeoutErrorCount <- newTVarIO 0 + sndQ <- newTBQueueIO 100 + rcvQ <- newTBQueueIO 100 return ProtocolClient { action = Nothing, @@ -452,21 +452,21 @@ getProtocolClient :: forall v err msg. Protocol v err msg => TVar ChaChaDRG -> T getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, clientALPN, serverVRange, agreeSecret} msgQ disconnected = do case chooseTransportHost networkConfig (host srv) of Right useHost -> - (getCurrentTime >>= atomically . mkProtocolClient useHost >>= runClient useTransport useHost) + (getCurrentTime >>= mkProtocolClient useHost >>= runClient useTransport useHost) `catch` \(e :: IOException) -> pure . Left $ PCEIOError e Left e -> pure $ Left e where NetworkConfig {tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig - mkProtocolClient :: TransportHost -> UTCTime -> STM (PClient v err msg) + mkProtocolClient :: TransportHost -> UTCTime -> IO (PClient v err msg) mkProtocolClient transportHost ts = do - connected <- newTVar False - sendPings <- newTVar False - lastReceived <- newTVar ts - timeoutErrorCount <- newTVar 0 - clientCorrId <- C.newRandomDRG g - sentCommands <- TM.empty - sndQ <- newTBQueue qSize - rcvQ <- newTBQueue qSize + connected <- newTVarIO False + sendPings <- newTVarIO False + lastReceived <- newTVarIO ts + timeoutErrorCount <- newTVarIO 0 + clientCorrId <- atomically $ C.newRandomDRG g + sentCommands <- TM.emptyIO + sndQ <- newTBQueueIO qSize + rcvQ <- newTBQueueIO qSize return PClient { connected, @@ -565,7 +565,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize processMsg ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr)) | B.null $ bs corrId = sendMsg $ STEvent clientResp | otherwise = - atomically (TM.lookup corrId sentCommands) >>= \case + TM.lookupIO corrId sentCommands >>= \case Nothing -> sendMsg $ STUnexpectedError unexpected Just Request {entityId, command, pending, responseVar} -> do wasPending <- @@ -1089,13 +1089,13 @@ mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentC nonce@(C.CbNonce corrId) <- maybe (atomically $ C.randomCbNonce clientCorrId) pure nonce_ let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (CorrId corrId, entityId, command) auth = authTransmission (thAuth thParams) pKey_ nonce tForAuth - r <- atomically $ mkRequest (CorrId corrId) + r <- mkRequest (CorrId corrId) pure ((,tToSend) <$> auth, r) where - mkRequest :: CorrId -> STM (Request err msg) + mkRequest :: CorrId -> IO (Request err msg) mkRequest corrId = do - pending <- newTVar True - responseVar <- newEmptyTMVar + pending <- newTVarIO True + responseVar <- newEmptyTMVarIO let r = Request { corrId, @@ -1104,7 +1104,7 @@ mkTransmission_ ProtocolClient {thParams, client_ = PClient {clientCorrId, sentC pending, responseVar } - TM.insert corrId r sentCommands + atomically $ TM.insert corrId r sentCommands pure r authTransmission :: Maybe (THandleAuth 'TClient) -> Maybe C.APrivateAuthKey -> C.CbNonce -> ByteString -> Either TransportError (Maybe TransmissionAuth) diff --git a/src/Simplex/Messaging/Client/Agent.hs b/src/Simplex/Messaging/Client/Agent.hs index a95f706bf..8073f1d48 100644 --- a/src/Simplex/Messaging/Client/Agent.hs +++ b/src/Simplex/Messaging/Client/Agent.hs @@ -108,17 +108,17 @@ data SMPClientAgent = SMPClientAgent type OwnServer = Bool -newSMPClientAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> STM SMPClientAgent +newSMPClientAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO SMPClientAgent newSMPClientAgent agentCfg@SMPClientAgentConfig {msgQSize, agentQSize} randomDrg = do - active <- newTVar True - msgQ <- newTBQueue msgQSize - agentQ <- newTBQueue agentQSize - smpClients <- TM.empty - smpSessions <- TM.empty - srvSubs <- TM.empty - pendingSrvSubs <- TM.empty - smpSubWorkers <- TM.empty - workerSeq <- newTVar 0 + active <- newTVarIO True + msgQ <- newTBQueueIO msgQSize + agentQ <- newTBQueueIO agentQSize + smpClients <- TM.emptyIO + smpSessions <- TM.emptyIO + srvSubs <- TM.emptyIO + pendingSrvSubs <- TM.emptyIO + smpSubWorkers <- TM.emptyIO + workerSeq <- newTVarIO 0 pure SMPClientAgent { agentCfg, @@ -229,7 +229,7 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s where getWorkerVar ts = ifM - (null <$> getPending) + (noPending) (pure Nothing) -- prevent race with cleanup and adding pending queues in another call (Just <$> getSessVar workerSeq srv smpSubWorkers ts) newSubWorker :: SessionVar (Async ()) -> IO () @@ -238,12 +238,13 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s atomically $ putTMVar (sessionVar v) a runSubWorker = withRetryInterval (reconnectInterval agentCfg) $ \_ loop -> do - pending <- atomically getPending + pending <- liftIO getPending unless (null pending) $ whenM (readTVarIO active) $ do void $ tcpConnectTimeout `timeout` runExceptT (reconnectSMPClient ca srv pending) loop ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout}} = smpCfg agentCfg - getPending = maybe (pure M.empty) readTVar =<< TM.lookup srv (pendingSrvSubs ca) + noPending = maybe (pure True) (fmap M.null . readTVar) =<< TM.lookup srv (pendingSrvSubs ca) + getPending = maybe (pure M.empty) readTVarIO =<< TM.lookupIO srv (pendingSrvSubs ca) cleanup :: SessionVar (Async ()) -> STM () cleanup v = do -- Here we wait until TMVar is not empty to prevent worker cleanup happening before worker is added to TMVar. @@ -254,7 +255,7 @@ reconnectClient ca@SMPClientAgent {active, agentCfg, smpSubWorkers, workerSeq} s reconnectSMPClient :: SMPClientAgent -> SMPServer -> Map SMPSub C.APrivateAuthKey -> ExceptT SMPClientError IO () reconnectSMPClient ca@SMPClientAgent {agentCfg} srv cs = withSMP ca srv $ \smp -> liftIO $ do - currSubs <- atomically $ maybe (pure M.empty) readTVar =<< TM.lookup srv (srvSubs ca) + currSubs <- maybe (pure M.empty) readTVarIO =<< TM.lookupIO srv (srvSubs ca) let (nSubs, rSubs) = foldr (groupSub currSubs) ([], []) $ M.assocs cs subscribe_ smp SPNotifier nSubs subscribe_ smp SPRecipient rSubs @@ -289,8 +290,8 @@ getConnectedSMPServerClient SMPClientAgent {smpClients} srv = (Nothing <$ atomically (removeSessVar v srv smpClients)) -- proxy will create a new connection (pure $ Just $ Left e) -- not expired, returning error -lookupSMPServerClient :: SMPClientAgent -> SessionId -> STM (Maybe (OwnServer, SMPClient)) -lookupSMPServerClient SMPClientAgent {smpSessions} sessId = TM.lookup sessId smpSessions +lookupSMPServerClient :: SMPClientAgent -> SessionId -> IO (Maybe (OwnServer, SMPClient)) +lookupSMPServerClient SMPClientAgent {smpSessions} sessId = TM.lookupIO sessId smpSessions closeSMPClientAgent :: SMPClientAgent -> IO () closeSMPClientAgent c = do diff --git a/src/Simplex/Messaging/Notifications/Server.hs b/src/Simplex/Messaging/Notifications/Server.hs index 2bf8dbcbf..1192148ac 100644 --- a/src/Simplex/Messaging/Notifications/Server.hs +++ b/src/Simplex/Messaging/Notifications/Server.hs @@ -176,10 +176,10 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge getSMPSubscriber :: SMPServer -> M SMPSubscriber getSMPSubscriber smpServer = - atomically (TM.lookup smpServer smpSubscribers) >>= maybe createSMPSubscriber pure + liftIO (TM.lookupIO smpServer smpSubscribers) >>= maybe createSMPSubscriber pure where createSMPSubscriber = do - sub@SMPSubscriber {subThreadId} <- atomically newSMPSubscriber + sub@SMPSubscriber {subThreadId} <- liftIO newSMPSubscriber atomically $ TM.insert smpServer sub smpSubscribers tId <- mkWeakThreadId =<< forkIO (runSMPSubscriber sub) atomically . writeTVar subThreadId $ Just tId @@ -333,7 +333,7 @@ runNtfClientTransport :: Transport c => THandleNTF c 'TServer -> M () runNtfClientTransport th@THandle {params} = do qSize <- asks $ clientQSize . config ts <- liftIO getSystemTime - c <- atomically $ newNtfServerClient qSize params ts + c <- liftIO $ newNtfServerClient qSize params ts s <- asks subscriber ps <- asks pushServer expCfg <- asks $ inactiveClientExpiration . config @@ -507,7 +507,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu | otherwise -> do logDebug "TCRN" atomically $ writeTVar tknCronInterval int - atomically (TM.lookup tknId intervalNotifiers) >>= \case + liftIO (TM.lookupIO tknId intervalNotifiers) >>= \case Nothing -> runIntervalNotifier int Just IntervalNotifier {interval, action} -> unless (interval == int) $ do @@ -585,7 +585,7 @@ incNtfStat statSel = do saveServerStats :: M () saveServerStats = asks (serverStatsBackupFile . config) - >>= mapM_ (\f -> asks serverStats >>= atomically . getNtfServerStatsData >>= liftIO . saveStats f) + >>= mapM_ (\f -> asks serverStats >>= liftIO . getNtfServerStatsData >>= liftIO . saveStats f) where saveStats f stats = do logInfo $ "saving server stats to file " <> T.pack f diff --git a/src/Simplex/Messaging/Notifications/Server/Env.hs b/src/Simplex/Messaging/Notifications/Server/Env.hs index 5ebd5230e..dc0cb0a73 100644 --- a/src/Simplex/Messaging/Notifications/Server/Env.hs +++ b/src/Simplex/Messaging/Notifications/Server/Env.hs @@ -10,7 +10,6 @@ module Simplex.Messaging.Notifications.Server.Env where import Control.Concurrent (ThreadId) import Control.Concurrent.Async (Async) import Control.Logger.Simple -import Control.Monad.IO.Unlift import Crypto.Random import Data.Int (Int64) import Data.List.NonEmpty (NonEmpty) @@ -85,16 +84,16 @@ data NtfEnv = NtfEnv newNtfServerEnv :: NtfServerConfig -> IO NtfEnv newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsConfig, storeLogFile, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do - random <- liftIO C.newRandom - store <- atomically newNtfStore + random <- C.newRandom + store <- newNtfStore logInfo "restoring subscriptions..." - storeLog <- liftIO $ mapM (`readWriteNtfStore` store) storeLogFile + storeLog <- mapM (`readWriteNtfStore` store) storeLogFile logInfo "restored subscriptions" - subscriber <- atomically $ newNtfSubscriber subQSize smpAgentCfg random - pushServer <- atomically $ newNtfPushServer pushQSize apnsConfig - tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig) - Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile - serverStats <- atomically . newNtfServerStats =<< liftIO getCurrentTime + subscriber <- newNtfSubscriber subQSize smpAgentCfg random + pushServer <- newNtfPushServer pushQSize apnsConfig + tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig) + Fingerprint fp <- loadFingerprint caCertificateFile + serverStats <- newNtfServerStats =<< getCurrentTime pure NtfEnv {config, subscriber, pushServer, store, storeLog, random, tlsServerParams, serverIdentity = C.KeyHash fp, serverStats} data NtfSubscriber = NtfSubscriber @@ -103,10 +102,10 @@ data NtfSubscriber = NtfSubscriber smpAgent :: SMPClientAgent } -newNtfSubscriber :: Natural -> SMPClientAgentConfig -> TVar ChaChaDRG -> STM NtfSubscriber +newNtfSubscriber :: Natural -> SMPClientAgentConfig -> TVar ChaChaDRG -> IO NtfSubscriber newNtfSubscriber qSize smpAgentCfg random = do - smpSubscribers <- TM.empty - newSubQ <- newTBQueue qSize + smpSubscribers <- TM.emptyIO + newSubQ <- newTBQueueIO qSize smpAgent <- newSMPClientAgent smpAgentCfg random pure NtfSubscriber {smpSubscribers, newSubQ, smpAgent} @@ -115,10 +114,10 @@ data SMPSubscriber = SMPSubscriber subThreadId :: TVar (Maybe (Weak ThreadId)) } -newSMPSubscriber :: STM SMPSubscriber +newSMPSubscriber :: IO SMPSubscriber newSMPSubscriber = do - newSubQ <- newTQueue - subThreadId <- newTVar Nothing + newSubQ <- newTQueueIO + subThreadId <- newTVarIO Nothing pure SMPSubscriber {newSubQ, subThreadId} data NtfPushServer = NtfPushServer @@ -134,11 +133,11 @@ data IntervalNotifier = IntervalNotifier interval :: Word16 } -newNtfPushServer :: Natural -> APNSPushClientConfig -> STM NtfPushServer +newNtfPushServer :: Natural -> APNSPushClientConfig -> IO NtfPushServer newNtfPushServer qSize apnsConfig = do - pushQ <- newTBQueue qSize - pushClients <- TM.empty - intervalNotifiers <- TM.empty + pushQ <- newTBQueueIO qSize + pushClients <- TM.emptyIO + intervalNotifiers <- TM.emptyIO pure NtfPushServer {pushQ, pushClients, intervalNotifiers, apnsConfig} newPushClient :: NtfPushServer -> PushProvider -> IO PushProviderClient @@ -151,7 +150,7 @@ newPushClient NtfPushServer {apnsConfig, pushClients} pp = do getPushClient :: NtfPushServer -> PushProvider -> IO PushProviderClient getPushClient s@NtfPushServer {pushClients} pp = - atomically (TM.lookup pp pushClients) >>= maybe (newPushClient s pp) pure + TM.lookupIO pp pushClients >>= maybe (newPushClient s pp) pure data NtfRequest = NtfReqNew CorrId ANewNtfEntity @@ -167,11 +166,11 @@ data NtfServerClient = NtfServerClient sndActiveAt :: TVar SystemTime } -newNtfServerClient :: Natural -> THandleParams NTFVersion 'TServer -> SystemTime -> STM NtfServerClient +newNtfServerClient :: Natural -> THandleParams NTFVersion 'TServer -> SystemTime -> IO NtfServerClient newNtfServerClient qSize ntfThParams ts = do - rcvQ <- newTBQueue qSize - sndQ <- newTBQueue qSize - connected <- newTVar True - rcvActiveAt <- newTVar ts - sndActiveAt <- newTVar ts + rcvQ <- newTBQueueIO qSize + sndQ <- newTBQueueIO qSize + connected <- newTVarIO True + rcvActiveAt <- newTVarIO ts + sndActiveAt <- newTVarIO ts return NtfServerClient {rcvQ, sndQ, ntfThParams, connected, rcvActiveAt, sndActiveAt} diff --git a/src/Simplex/Messaging/Notifications/Server/Stats.hs b/src/Simplex/Messaging/Notifications/Server/Stats.hs index 7debc1ac9..b73e6098f 100644 --- a/src/Simplex/Messaging/Notifications/Server/Stats.hs +++ b/src/Simplex/Messaging/Notifications/Server/Stats.hs @@ -40,30 +40,30 @@ data NtfServerStatsData = NtfServerStatsData _activeSubs :: PeriodStatsData NotifierId } -newNtfServerStats :: UTCTime -> STM NtfServerStats +newNtfServerStats :: UTCTime -> IO NtfServerStats newNtfServerStats ts = do - fromTime <- newTVar ts - tknCreated <- newTVar 0 - tknVerified <- newTVar 0 - tknDeleted <- newTVar 0 - subCreated <- newTVar 0 - subDeleted <- newTVar 0 - ntfReceived <- newTVar 0 - ntfDelivered <- newTVar 0 + fromTime <- newTVarIO ts + tknCreated <- newTVarIO 0 + tknVerified <- newTVarIO 0 + tknDeleted <- newTVarIO 0 + subCreated <- newTVarIO 0 + subDeleted <- newTVarIO 0 + ntfReceived <- newTVarIO 0 + ntfDelivered <- newTVarIO 0 activeTokens <- newPeriodStats activeSubs <- newPeriodStats pure NtfServerStats {fromTime, tknCreated, tknVerified, tknDeleted, subCreated, subDeleted, ntfReceived, ntfDelivered, activeTokens, activeSubs} -getNtfServerStatsData :: NtfServerStats -> STM NtfServerStatsData +getNtfServerStatsData :: NtfServerStats -> IO NtfServerStatsData getNtfServerStatsData s@NtfServerStats {fromTime} = do - _fromTime <- readTVar fromTime - _tknCreated <- readTVar $ tknCreated s - _tknVerified <- readTVar $ tknVerified s - _tknDeleted <- readTVar $ tknDeleted s - _subCreated <- readTVar $ subCreated s - _subDeleted <- readTVar $ subDeleted s - _ntfReceived <- readTVar $ ntfReceived s - _ntfDelivered <- readTVar $ ntfDelivered s + _fromTime <- readTVarIO fromTime + _tknCreated <- readTVarIO $ tknCreated s + _tknVerified <- readTVarIO $ tknVerified s + _tknDeleted <- readTVarIO $ tknDeleted s + _subCreated <- readTVarIO $ subCreated s + _subDeleted <- readTVarIO $ subDeleted s + _ntfReceived <- readTVarIO $ ntfReceived s + _ntfDelivered <- readTVarIO $ ntfDelivered s _activeTokens <- getPeriodStatsData $ activeTokens s _activeSubs <- getPeriodStatsData $ activeSubs s pure NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs} diff --git a/src/Simplex/Messaging/Notifications/Server/Store.hs b/src/Simplex/Messaging/Notifications/Server/Store.hs index 83dc1a4c2..b4d91dc88 100644 --- a/src/Simplex/Messaging/Notifications/Server/Store.hs +++ b/src/Simplex/Messaging/Notifications/Server/Store.hs @@ -33,13 +33,13 @@ data NtfStore = NtfStore subscriptionLookup :: TMap SMPQueueNtf NtfSubscriptionId } -newNtfStore :: STM NtfStore +newNtfStore :: IO NtfStore newNtfStore = do - tokens <- TM.empty - tokenRegistrations <- TM.empty - subscriptions <- TM.empty - tokenSubscriptions <- TM.empty - subscriptionLookup <- TM.empty + tokens <- TM.emptyIO + tokenRegistrations <- TM.emptyIO + subscriptions <- TM.emptyIO + tokenSubscriptions <- TM.emptyIO + subscriptionLookup <- TM.emptyIO pure NtfStore {tokens, tokenRegistrations, subscriptions, tokenSubscriptions, subscriptionLookup} data NtfTknData = NtfTknData @@ -77,6 +77,9 @@ data NtfEntityRec (e :: NtfEntity) where getNtfToken :: NtfStore -> NtfTokenId -> STM (Maybe NtfTknData) getNtfToken st tknId = TM.lookup tknId (tokens st) +getNtfTokenIO :: NtfStore -> NtfTokenId -> IO (Maybe NtfTknData) +getNtfTokenIO st tknId = TM.lookupIO tknId (tokens st) + addNtfToken :: NtfStore -> NtfTokenId -> NtfTknData -> STM () addNtfToken st tknId tkn@NtfTknData {token, tknVerifyKey} = do TM.insert tknId tkn $ tokens st diff --git a/src/Simplex/Messaging/Server.hs b/src/Simplex/Messaging/Server.hs index fb39c3240..c5d067475 100644 --- a/src/Simplex/Messaging/Server.hs +++ b/src/Simplex/Messaging/Server.hs @@ -428,7 +428,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do putStat label var = getStat var >>= \v -> hPutStrLn h $ label <> ": " <> show v putProxyStat :: String -> (ServerStats -> ProxyStats) -> IO () putProxyStat label var = do - ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} <- atomically $ getProxyStatsData $ var ss + ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} <- getProxyStatsData $ var ss hPutStrLn h $ label <> ": requests=" <> show _pRequests <> ", successes=" <> show _pSuccesses <> ", errorsConnect=" <> show _pErrorsConnect <> ", errorsCompat=" <> show _pErrorsCompat <> ", errorsOther=" <> show _pErrorsOther putStat "fromTime" fromTime putStat "qCreated" qCreated @@ -472,7 +472,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do #endif CPSockets -> withUserRole $ do (accepted', closed', active') <- unliftIO u $ asks sockets - (accepted, closed, active) <- atomically $ (,,) <$> readTVar accepted' <*> readTVar closed' <*> readTVar active' + (accepted, closed, active) <- (,,) <$> readTVarIO accepted' <*> readTVarIO closed' <*> readTVarIO active' hPutStrLn h "Sockets: " hPutStrLn h $ "accepted: " <> show accepted hPutStrLn h $ "closed: " <> show closed @@ -619,10 +619,8 @@ runClientTransport h@THandle {params = thParams@THandleParams {thVersion, sessio ts <- liftIO getSystemTime active <- asks clients nextClientId <- asks clientSeq - c <- atomically $ do - new@Client {clientId} <- newClient nextClientId q thVersion sessionId ts - modifyTVar' active $ IM.insert clientId new - pure new + c@Client {clientId} <- liftIO $ newClient nextClientId q thVersion sessionId ts + atomically $ modifyTVar' active $ IM.insert clientId c s <- asks server expCfg <- asks $ inactiveClientExpiration . config th <- newMVar h -- put TH under a fair lock to interleave messages and command responses @@ -887,7 +885,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi ProxyAgent {smpAgent = a} <- asks proxyAgent ServerStats {pMsgFwds, pMsgFwdsOwn} <- asks serverStats let inc = mkIncProxyStats pMsgFwds pMsgFwdsOwn - atomically (lookupSMPServerClient a sessId) >>= \case + liftIO (lookupSMPServerClient a sessId) >>= \case Just (own, smp) -> do inc own pRequests if v >= sendingProxySMPVersion @@ -1138,7 +1136,7 @@ client thParams' clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessi acknowledgeMsg :: QueueRec -> MsgId -> M (Transmission BrokerMsg) acknowledgeMsg qr msgId = time "ACK" $ do - atomically (TM.lookup entId subscriptions) >>= \case + liftIO (TM.lookupIO entId subscriptions) >>= \case Nothing -> pure $ err NO_MSG Just sub -> atomically (getDelivered sub) >>= \case @@ -1524,7 +1522,7 @@ restoreServerMessages = saveServerStats :: M () saveServerStats = asks (serverStatsBackupFile . config) - >>= mapM_ (\f -> asks serverStats >>= atomically . getServerStatsData >>= liftIO . saveStats f) + >>= mapM_ (\f -> asks serverStats >>= liftIO . getServerStatsData >>= liftIO . saveStats f) where saveStats f stats = do logInfo $ "saving server stats to file " <> T.pack f diff --git a/src/Simplex/Messaging/Server/Env/STM.hs b/src/Simplex/Messaging/Server/Env/STM.hs index 7a3bcc362..84e664607 100644 --- a/src/Simplex/Messaging/Server/Env/STM.hs +++ b/src/Simplex/Messaging/Server/Env/STM.hs @@ -10,7 +10,6 @@ module Simplex.Messaging.Server.Env.STM where import Control.Concurrent (ThreadId) import Control.Logger.Simple import Control.Monad -import Control.Monad.IO.Unlift import Crypto.Random import Data.ByteString.Char8 (ByteString) import Data.Int (Int64) @@ -175,29 +174,29 @@ data Sub = Sub delivered :: TMVar MsgId } -newServer :: STM Server +newServer :: IO Server newServer = do - subscribedQ <- newTQueue - subscribers <- TM.empty - ntfSubscribedQ <- newTQueue - notifiers <- TM.empty - savingLock <- createLock + subscribedQ <- newTQueueIO + subscribers <- TM.emptyIO + ntfSubscribedQ <- newTQueueIO + notifiers <- TM.emptyIO + savingLock <- atomically createLock return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, savingLock} -newClient :: TVar ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> STM Client +newClient :: TVar ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client newClient nextClientId qSize thVersion sessionId createdAt = do - clientId <- stateTVar nextClientId $ \next -> (next, next + 1) - subscriptions <- TM.empty - ntfSubscriptions <- TM.empty - rcvQ <- newTBQueue qSize - sndQ <- newTBQueue qSize - msgQ <- newTBQueue qSize - procThreads <- newTVar 0 - endThreads <- newTVar IM.empty - endThreadSeq <- newTVar 0 - connected <- newTVar True - rcvActiveAt <- newTVar createdAt - sndActiveAt <- newTVar createdAt + clientId <- atomically $ stateTVar nextClientId $ \next -> (next, next + 1) + subscriptions <- TM.emptyIO + ntfSubscriptions <- TM.emptyIO + rcvQ <- newTBQueueIO qSize + sndQ <- newTBQueueIO qSize + msgQ <- newTBQueueIO qSize + procThreads <- newTVarIO 0 + endThreads <- newTVarIO IM.empty + endThreadSeq <- newTVarIO 0 + connected <- newTVarIO True + rcvActiveAt <- newTVarIO createdAt + sndActiveAt <- newTVarIO createdAt return Client {clientId, subscriptions, ntfSubscriptions, rcvQ, sndQ, msgQ, procThreads, endThreads, endThreadSeq, thVersion, sessionId, connected, createdAt, rcvActiveAt, sndActiveAt} newSubscription :: SubscriptionThread -> STM Sub @@ -213,10 +212,10 @@ newProhibitedSub = do newEnv :: ServerConfig -> IO Env newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile, smpAgentCfg, transportConfig, information, messageExpiration} = do - server <- atomically newServer - queueStore <- atomically newQueueStore - msgStore <- atomically newMsgStore - random <- liftIO C.newRandom + server <- newServer + queueStore <- newQueueStore + msgStore <- newMsgStore + random <- C.newRandom storeLog <- forM storeLogFile $ \f -> do logInfo $ "restoring queues from file " <> T.pack f @@ -224,20 +223,19 @@ newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, tlsServerParams <- loadTLSServerParams caCertificateFile certificateFile privateKeyFile (alpn transportConfig) Fingerprint fp <- loadFingerprint caCertificateFile let serverIdentity = KeyHash fp - serverStats <- atomically . newServerStats =<< getCurrentTime - sockets <- atomically newSocketState + serverStats <- newServerStats =<< getCurrentTime + sockets <- newSocketState clientSeq <- newTVarIO 0 clients <- newTVarIO mempty - proxyAgent <- atomically $ newSMPProxyAgent smpAgentCfg random + proxyAgent <- newSMPProxyAgent smpAgentCfg random pure Env {config, serverInfo, server, serverIdentity, queueStore, msgStore, random, storeLog, tlsServerParams, serverStats, sockets, clientSeq, clients, proxyAgent} where restoreQueues :: QueueStore -> FilePath -> IO (StoreLog 'WriteMode) restoreQueues QueueStore {queues, senders, notifiers} f = do (qs, s) <- readWriteStoreLog f - atomically $ do - writeTVar queues =<< mapM newTVar qs - writeTVar senders $! M.foldr' addSender M.empty qs - writeTVar notifiers $! M.foldr' addNotifier M.empty qs + atomically . writeTVar queues =<< mapM newTVarIO qs + atomically $ writeTVar senders $! M.foldr' addSender M.empty qs + atomically $ writeTVar notifiers $! M.foldr' addNotifier M.empty qs pure s addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId addSender q = M.insert (senderId q) (recipientId q) @@ -263,7 +261,7 @@ newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, | isJust (storeMsgsFile config) = SPMMessages | otherwise = SPMQueues -newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> STM ProxyAgent +newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent newSMPProxyAgent smpAgentCfg random = do smpAgent <- newSMPClientAgent smpAgentCfg random pure ProxyAgent {smpAgent} diff --git a/src/Simplex/Messaging/Server/MsgStore/STM.hs b/src/Simplex/Messaging/Server/MsgStore/STM.hs index 6f9c40e58..e0a5c8b45 100644 --- a/src/Simplex/Messaging/Server/MsgStore/STM.hs +++ b/src/Simplex/Messaging/Server/MsgStore/STM.hs @@ -41,8 +41,8 @@ data MsgQueue = MsgQueue type STMMsgStore = TMap RecipientId MsgQueue -newMsgStore :: STM STMMsgStore -newMsgStore = TM.empty +newMsgStore :: IO STMMsgStore +newMsgStore = TM.emptyIO getMsgQueue :: STMMsgStore -> RecipientId -> Int -> STM MsgQueue getMsgQueue st rId quota = maybe newQ pure =<< TM.lookup rId st diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index d6cdaf10a..50907cf9a 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/src/Simplex/Messaging/Server/QueueStore/STM.hs @@ -38,11 +38,11 @@ data QueueStore = QueueStore notifiers :: TMap NotifierId RecipientId } -newQueueStore :: STM QueueStore +newQueueStore :: IO QueueStore newQueueStore = do - queues <- TM.empty - senders <- TM.empty - notifiers <- TM.empty + queues <- TM.emptyIO + senders <- TM.emptyIO + notifiers <- TM.emptyIO pure QueueStore {queues, senders, notifiers} addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ()) diff --git a/src/Simplex/Messaging/Server/Stats.hs b/src/Simplex/Messaging/Server/Stats.hs index 377fddd1a..f5b430bb6 100644 --- a/src/Simplex/Messaging/Server/Stats.hs +++ b/src/Simplex/Messaging/Server/Stats.hs @@ -112,51 +112,51 @@ data ServerStatsData = ServerStatsData } deriving (Show) -newServerStats :: UTCTime -> STM ServerStats +newServerStats :: UTCTime -> IO ServerStats newServerStats ts = do - fromTime <- newTVar ts - qCreated <- newTVar 0 - qSecured <- newTVar 0 - qDeletedAll <- newTVar 0 - qDeletedNew <- newTVar 0 - qDeletedSecured <- newTVar 0 - qSub <- newTVar 0 - qSubNoMsg <- newTVar 0 - qSubAuth <- newTVar 0 - qSubDuplicate <- newTVar 0 - qSubProhibited <- newTVar 0 - ntfCreated <- newTVar 0 - ntfDeleted <- newTVar 0 - ntfSub <- newTVar 0 - ntfSubAuth <- newTVar 0 - ntfSubDuplicate <- newTVar 0 - msgSent <- newTVar 0 - msgSentAuth <- newTVar 0 - msgSentQuota <- newTVar 0 - msgSentLarge <- newTVar 0 - msgRecv <- newTVar 0 - msgRecvGet <- newTVar 0 - msgGet <- newTVar 0 - msgGetNoMsg <- newTVar 0 - msgGetAuth <- newTVar 0 - msgGetDuplicate <- newTVar 0 - msgGetProhibited <- newTVar 0 - msgExpired <- newTVar 0 + fromTime <- newTVarIO ts + qCreated <- newTVarIO 0 + qSecured <- newTVarIO 0 + qDeletedAll <- newTVarIO 0 + qDeletedNew <- newTVarIO 0 + qDeletedSecured <- newTVarIO 0 + qSub <- newTVarIO 0 + qSubNoMsg <- newTVarIO 0 + qSubAuth <- newTVarIO 0 + qSubDuplicate <- newTVarIO 0 + qSubProhibited <- newTVarIO 0 + ntfCreated <- newTVarIO 0 + ntfDeleted <- newTVarIO 0 + ntfSub <- newTVarIO 0 + ntfSubAuth <- newTVarIO 0 + ntfSubDuplicate <- newTVarIO 0 + msgSent <- newTVarIO 0 + msgSentAuth <- newTVarIO 0 + msgSentQuota <- newTVarIO 0 + msgSentLarge <- newTVarIO 0 + msgRecv <- newTVarIO 0 + msgRecvGet <- newTVarIO 0 + msgGet <- newTVarIO 0 + msgGetNoMsg <- newTVarIO 0 + msgGetAuth <- newTVarIO 0 + msgGetDuplicate <- newTVarIO 0 + msgGetProhibited <- newTVarIO 0 + msgExpired <- newTVarIO 0 activeQueues <- newPeriodStats subscribedQueues <- newPeriodStats - msgSentNtf <- newTVar 0 - msgRecvNtf <- newTVar 0 + msgSentNtf <- newTVarIO 0 + msgRecvNtf <- newTVarIO 0 activeQueuesNtf <- newPeriodStats - msgNtfs <- newTVar 0 - msgNtfNoSub <- newTVar 0 - msgNtfLost <- newTVar 0 + msgNtfs <- newTVarIO 0 + msgNtfNoSub <- newTVarIO 0 + msgNtfLost <- newTVarIO 0 pRelays <- newProxyStats pRelaysOwn <- newProxyStats pMsgFwds <- newProxyStats pMsgFwdsOwn <- newProxyStats - pMsgFwdsRecv <- newTVar 0 - qCount <- newTVar 0 - msgCount <- newTVar 0 + pMsgFwdsRecv <- newTVarIO 0 + qCount <- newTVarIO 0 + msgCount <- newTVarIO 0 pure ServerStats { fromTime, @@ -204,51 +204,51 @@ newServerStats ts = do msgCount } -getServerStatsData :: ServerStats -> STM ServerStatsData +getServerStatsData :: ServerStats -> IO ServerStatsData getServerStatsData s = do - _fromTime <- readTVar $ fromTime s - _qCreated <- readTVar $ qCreated s - _qSecured <- readTVar $ qSecured s - _qDeletedAll <- readTVar $ qDeletedAll s - _qDeletedNew <- readTVar $ qDeletedNew s - _qDeletedSecured <- readTVar $ qDeletedSecured s - _qSub <- readTVar $ qSub s - _qSubNoMsg <- readTVar $ qSubNoMsg s - _qSubAuth <- readTVar $ qSubAuth s - _qSubDuplicate <- readTVar $ qSubDuplicate s - _qSubProhibited <- readTVar $ qSubProhibited s - _ntfCreated <- readTVar $ ntfCreated s - _ntfDeleted <- readTVar $ ntfDeleted s - _ntfSub <- readTVar $ ntfSub s - _ntfSubAuth <- readTVar $ ntfSubAuth s - _ntfSubDuplicate <- readTVar $ ntfSubDuplicate s - _msgSent <- readTVar $ msgSent s - _msgSentAuth <- readTVar $ msgSentAuth s - _msgSentQuota <- readTVar $ msgSentQuota s - _msgSentLarge <- readTVar $ msgSentLarge s - _msgRecv <- readTVar $ msgRecv s - _msgRecvGet <- readTVar $ msgRecvGet s - _msgGet <- readTVar $ msgGet s - _msgGetNoMsg <- readTVar $ msgGetNoMsg s - _msgGetAuth <- readTVar $ msgGetAuth s - _msgGetDuplicate <- readTVar $ msgGetDuplicate s - _msgGetProhibited <- readTVar $ msgGetProhibited s - _msgExpired <- readTVar $ msgExpired s + _fromTime <- readTVarIO $ fromTime s + _qCreated <- readTVarIO $ qCreated s + _qSecured <- readTVarIO $ qSecured s + _qDeletedAll <- readTVarIO $ qDeletedAll s + _qDeletedNew <- readTVarIO $ qDeletedNew s + _qDeletedSecured <- readTVarIO $ qDeletedSecured s + _qSub <- readTVarIO $ qSub s + _qSubNoMsg <- readTVarIO $ qSubNoMsg s + _qSubAuth <- readTVarIO $ qSubAuth s + _qSubDuplicate <- readTVarIO $ qSubDuplicate s + _qSubProhibited <- readTVarIO $ qSubProhibited s + _ntfCreated <- readTVarIO $ ntfCreated s + _ntfDeleted <- readTVarIO $ ntfDeleted s + _ntfSub <- readTVarIO $ ntfSub s + _ntfSubAuth <- readTVarIO $ ntfSubAuth s + _ntfSubDuplicate <- readTVarIO $ ntfSubDuplicate s + _msgSent <- readTVarIO $ msgSent s + _msgSentAuth <- readTVarIO $ msgSentAuth s + _msgSentQuota <- readTVarIO $ msgSentQuota s + _msgSentLarge <- readTVarIO $ msgSentLarge s + _msgRecv <- readTVarIO $ msgRecv s + _msgRecvGet <- readTVarIO $ msgRecvGet s + _msgGet <- readTVarIO $ msgGet s + _msgGetNoMsg <- readTVarIO $ msgGetNoMsg s + _msgGetAuth <- readTVarIO $ msgGetAuth s + _msgGetDuplicate <- readTVarIO $ msgGetDuplicate s + _msgGetProhibited <- readTVarIO $ msgGetProhibited s + _msgExpired <- readTVarIO $ msgExpired s _activeQueues <- getPeriodStatsData $ activeQueues s _subscribedQueues <- getPeriodStatsData $ subscribedQueues s - _msgSentNtf <- readTVar $ msgSentNtf s - _msgRecvNtf <- readTVar $ msgRecvNtf s + _msgSentNtf <- readTVarIO $ msgSentNtf s + _msgRecvNtf <- readTVarIO $ msgRecvNtf s _activeQueuesNtf <- getPeriodStatsData $ activeQueuesNtf s - _msgNtfs <- readTVar $ msgNtfs s - _msgNtfNoSub <- readTVar $ msgNtfNoSub s - _msgNtfLost <- readTVar $ msgNtfLost s + _msgNtfs <- readTVarIO $ msgNtfs s + _msgNtfNoSub <- readTVarIO $ msgNtfNoSub s + _msgNtfLost <- readTVarIO $ msgNtfLost s _pRelays <- getProxyStatsData $ pRelays s _pRelaysOwn <- getProxyStatsData $ pRelaysOwn s _pMsgFwds <- getProxyStatsData $ pMsgFwds s _pMsgFwdsOwn <- getProxyStatsData $ pMsgFwdsOwn s - _pMsgFwdsRecv <- readTVar $ pMsgFwdsRecv s - _qCount <- readTVar $ qCount s - _msgCount <- readTVar $ msgCount s + _pMsgFwdsRecv <- readTVarIO $ pMsgFwdsRecv s + _qCount <- readTVarIO $ qCount s + _msgCount <- readTVarIO $ msgCount s pure ServerStatsData { _fromTime, @@ -510,11 +510,11 @@ data PeriodStats a = PeriodStats month :: TVar (Set a) } -newPeriodStats :: STM (PeriodStats a) +newPeriodStats :: IO (PeriodStats a) newPeriodStats = do - day <- newTVar S.empty - week <- newTVar S.empty - month <- newTVar S.empty + day <- newTVarIO S.empty + week <- newTVarIO S.empty + month <- newTVarIO S.empty pure PeriodStats {day, week, month} data PeriodStatsData a = PeriodStatsData @@ -527,11 +527,11 @@ data PeriodStatsData a = PeriodStatsData newPeriodStatsData :: PeriodStatsData a newPeriodStatsData = PeriodStatsData {_day = S.empty, _week = S.empty, _month = S.empty} -getPeriodStatsData :: PeriodStats a -> STM (PeriodStatsData a) +getPeriodStatsData :: PeriodStats a -> IO (PeriodStatsData a) getPeriodStatsData s = do - _day <- readTVar $ day s - _week <- readTVar $ week s - _month <- readTVar $ month s + _day <- readTVarIO $ day s + _week <- readTVarIO $ week s + _month <- readTVarIO $ month s pure PeriodStatsData {_day, _week, _month} setPeriodStats :: PeriodStats a -> PeriodStatsData a -> STM () @@ -585,13 +585,13 @@ data ProxyStats = ProxyStats pErrorsOther :: TVar Int } -newProxyStats :: STM ProxyStats +newProxyStats :: IO ProxyStats newProxyStats = do - pRequests <- newTVar 0 - pSuccesses <- newTVar 0 - pErrorsConnect <- newTVar 0 - pErrorsCompat <- newTVar 0 - pErrorsOther <- newTVar 0 + pRequests <- newTVarIO 0 + pSuccesses <- newTVarIO 0 + pErrorsConnect <- newTVarIO 0 + pErrorsCompat <- newTVarIO 0 + pErrorsOther <- newTVarIO 0 pure ProxyStats {pRequests, pSuccesses, pErrorsConnect, pErrorsCompat, pErrorsOther} data ProxyStatsData = ProxyStatsData @@ -606,13 +606,13 @@ data ProxyStatsData = ProxyStatsData newProxyStatsData :: ProxyStatsData newProxyStatsData = ProxyStatsData {_pRequests = 0, _pSuccesses = 0, _pErrorsConnect = 0, _pErrorsCompat = 0, _pErrorsOther = 0} -getProxyStatsData :: ProxyStats -> STM ProxyStatsData +getProxyStatsData :: ProxyStats -> IO ProxyStatsData getProxyStatsData s = do - _pRequests <- readTVar $ pRequests s - _pSuccesses <- readTVar $ pSuccesses s - _pErrorsConnect <- readTVar $ pErrorsConnect s - _pErrorsCompat <- readTVar $ pErrorsCompat s - _pErrorsOther <- readTVar $ pErrorsOther s + _pRequests <- readTVarIO $ pRequests s + _pSuccesses <- readTVarIO $ pSuccesses s + _pErrorsConnect <- readTVarIO $ pErrorsConnect s + _pErrorsCompat <- readTVarIO $ pErrorsCompat s + _pErrorsOther <- readTVarIO $ pErrorsOther s pure ProxyStatsData {_pRequests, _pSuccesses, _pErrorsConnect, _pErrorsCompat, _pErrorsOther} getResetProxyStatsData :: ProxyStats -> STM ProxyStatsData diff --git a/src/Simplex/Messaging/TMap.hs b/src/Simplex/Messaging/TMap.hs index 2f6e0cf8a..1bc9bcb60 100644 --- a/src/Simplex/Messaging/TMap.hs +++ b/src/Simplex/Messaging/TMap.hs @@ -1,11 +1,13 @@ module Simplex.Messaging.TMap ( TMap, - empty, + emptyIO, singleton, clear, Simplex.Messaging.TMap.null, Simplex.Messaging.TMap.lookup, + lookupIO, member, + memberIO, insert, delete, lookupInsert, @@ -24,9 +26,9 @@ import qualified Data.Map.Strict as M type TMap k a = TVar (Map k a) -empty :: STM (TMap k a) -empty = newTVar M.empty -{-# INLINE empty #-} +emptyIO :: IO (TMap k a) +emptyIO = newTVarIO M.empty +{-# INLINE emptyIO #-} singleton :: k -> a -> STM (TMap k a) singleton k v = newTVar $ M.singleton k v @@ -44,10 +46,18 @@ lookup :: Ord k => k -> TMap k a -> STM (Maybe a) lookup k m = M.lookup k <$> readTVar m {-# INLINE lookup #-} +lookupIO :: Ord k => k -> TMap k a -> IO (Maybe a) +lookupIO k m = M.lookup k <$> readTVarIO m +{-# INLINE lookupIO #-} + member :: Ord k => k -> TMap k a -> STM Bool member k m = M.member k <$> readTVar m {-# INLINE member #-} +memberIO :: Ord k => k -> TMap k a -> IO Bool +memberIO k m = M.member k <$> readTVarIO m +{-# INLINE memberIO #-} + insert :: Ord k => k -> a -> TMap k a -> STM () insert k v m = modifyTVar' m $ M.insert k v {-# INLINE insert #-} diff --git a/src/Simplex/Messaging/Transport.hs b/src/Simplex/Messaging/Transport.hs index d7f81f563..58843b7f5 100644 --- a/src/Simplex/Messaging/Transport.hs +++ b/src/Simplex/Messaging/Transport.hs @@ -285,7 +285,7 @@ getTLS :: TransportPeer -> TransportConfig -> X.CertificateChain -> T.Context -> getTLS tlsPeer cfg tlsServerCerts cxt = withTlsUnique tlsPeer cxt newTLS where newTLS tlsUniq = do - tlsBuffer <- atomically newTBuffer + tlsBuffer <- newTBuffer tlsALPN <- T.getNegotiatedProtocol cxt pure TLS {tlsContext = cxt, tlsALPN, tlsTransportConfig = cfg, tlsServerCerts, tlsPeer, tlsUniq, tlsBuffer} diff --git a/src/Simplex/Messaging/Transport/Buffer.hs b/src/Simplex/Messaging/Transport/Buffer.hs index 6de9326f8..a612afafc 100644 --- a/src/Simplex/Messaging/Transport/Buffer.hs +++ b/src/Simplex/Messaging/Transport/Buffer.hs @@ -17,10 +17,10 @@ data TBuffer = TBuffer getLock :: TMVar () } -newTBuffer :: STM TBuffer +newTBuffer :: IO TBuffer newTBuffer = do - buffer <- newTVar "" - getLock <- newTMVar () + buffer <- newTVarIO "" + getLock <- newTMVarIO () pure TBuffer {buffer, getLock} withBufferLock :: TBuffer -> IO a -> IO a diff --git a/src/Simplex/Messaging/Transport/HTTP2.hs b/src/Simplex/Messaging/Transport/HTTP2.hs index 9c6cd7abc..3b741e6ce 100644 --- a/src/Simplex/Messaging/Transport/HTTP2.hs +++ b/src/Simplex/Messaging/Transport/HTTP2.hs @@ -75,7 +75,7 @@ instance HTTP2BodyChunk HS.Request where getHTTP2Body :: HTTP2BodyChunk a => a -> Int -> IO HTTP2Body getHTTP2Body r n = do - bodyBuffer <- atomically newTBuffer + bodyBuffer <- newTBuffer let getPart n' = getBuffered bodyBuffer n' Nothing $ getBodyChunk r bodyHead <- getPart n let bodySize = fromMaybe 0 $ getBodySize r diff --git a/src/Simplex/Messaging/Transport/HTTP2/Client.hs b/src/Simplex/Messaging/Transport/HTTP2/Client.hs index 71757ca6d..d8d3d495d 100644 --- a/src/Simplex/Messaging/Transport/HTTP2/Client.hs +++ b/src/Simplex/Messaging/Transport/HTTP2/Client.hs @@ -104,13 +104,13 @@ attachHTTP2Client config host port disconnected bufferSize tls = getVerifiedHTTP getVerifiedHTTP2ClientWith :: HTTP2ClientConfig -> TransportHost -> ServiceName -> IO () -> ((TLS -> H.Client HTTP2Response) -> IO HTTP2Response) -> IO (Either HTTP2ClientError HTTP2Client) getVerifiedHTTP2ClientWith config host port disconnected setup = - (atomically mkHTTPS2Client >>= runClient) + (mkHTTPS2Client >>= runClient) `E.catch` \(e :: IOException) -> pure . Left $ HCIOError e where - mkHTTPS2Client :: STM HClient + mkHTTPS2Client :: IO HClient mkHTTPS2Client = do - connected <- newTVar False - reqQ <- newTBQueue $ qSize config + connected <- newTVarIO False + reqQ <- newTBQueueIO $ qSize config pure HClient {connected, disconnected, host, port, config, reqQ} runClient :: HClient -> IO (Either HTTP2ClientError HTTP2Client) diff --git a/src/Simplex/Messaging/Transport/Server.hs b/src/Simplex/Messaging/Transport/Server.hs index ffde39991..0b4da7833 100644 --- a/src/Simplex/Messaging/Transport/Server.hs +++ b/src/Simplex/Messaging/Transport/Server.hs @@ -76,7 +76,7 @@ serverTransportConfig TransportServerConfig {logTLSErrors} = -- All accepted connections are passed to the passed function. runTransportServer :: forall c. Transport c => TMVar Bool -> ServiceName -> T.ServerParams -> TransportServerConfig -> (c -> IO ()) -> IO () runTransportServer started port params cfg server = do - ss <- atomically newSocketState + ss <- newSocketState runTransportServerState ss started port params cfg server runTransportServerState :: forall c . Transport c => SocketState -> TMVar Bool -> ServiceName -> T.ServerParams -> TransportServerConfig -> (c -> IO ()) -> IO () @@ -85,7 +85,7 @@ runTransportServerState ss started port = runTransportServerSocketState ss start -- | Run a transport server with provided connection setup and handler. runTransportServerSocket :: Transport a => TMVar Bool -> IO Socket -> String -> T.ServerParams -> TransportServerConfig -> (a -> IO ()) -> IO () runTransportServerSocket started getSocket threadLabel serverParams cfg server = do - ss <- atomically newSocketState + ss <- newSocketState runTransportServerSocketState ss started getSocket threadLabel serverParams cfg server -- | Run a transport server with provided connection setup and handler. @@ -109,7 +109,7 @@ tlsServerCredentials serverParams = case T.sharedCredentials $ T.serverShared se -- | Run TCP server without TLS runTCPServer :: TMVar Bool -> ServiceName -> (Socket -> IO ()) -> IO () runTCPServer started port server = do - ss <- atomically newSocketState + ss <- newSocketState runTCPServerSocket ss started (startTCPServer started port) server -- | Wrap socket provider in a TCP server bracket. @@ -148,8 +148,8 @@ safeAccept sock = type SocketState = (TVar Int, TVar Int, TVar (IntMap (Weak ThreadId))) -newSocketState :: STM SocketState -newSocketState = (,,) <$> newTVar 0 <*> newTVar 0 <*> newTVar mempty +newSocketState :: IO SocketState +newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO () closeServer started clients sock = do diff --git a/tests/CoreTests/BatchingTests.hs b/tests/CoreTests/BatchingTests.hs index caab0637a..5f6beb034 100644 --- a/tests/CoreTests/BatchingTests.hs +++ b/tests/CoreTests/BatchingTests.hs @@ -261,7 +261,7 @@ testClientStub :: IO (ProtocolClient SMPVersion ErrorType BrokerMsg) testClientStub = do g <- C.newRandom sessId <- atomically $ C.randomBytes 32 g - atomically $ smpClientStub g sessId subModeSMPVersion Nothing + smpClientStub g sessId subModeSMPVersion Nothing clientStubV7 :: IO (ProtocolClient SMPVersion ErrorType BrokerMsg) clientStubV7 = do @@ -269,7 +269,7 @@ clientStubV7 = do sessId <- atomically $ C.randomBytes 32 g (rKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g thAuth_ <- testTHandleAuth authCmdsSMPVersion g rKey - atomically $ smpClientStub g sessId authCmdsSMPVersion thAuth_ + smpClientStub g sessId authCmdsSMPVersion thAuth_ randomSUB :: ByteString -> IO (Either TransportError (Maybe TransmissionAuth, ByteString)) randomSUB = randomSUB_ C.SEd25519 subModeSMPVersion diff --git a/tests/CoreTests/TRcvQueuesTests.hs b/tests/CoreTests/TRcvQueuesTests.hs index 7e39d7fd9..24d54fc8e 100644 --- a/tests/CoreTests/TRcvQueuesTests.hs +++ b/tests/CoreTests/TRcvQueuesTests.hs @@ -43,7 +43,7 @@ checkDataInvariant trq = atomically $ do hasConnTest :: IO () hasConnTest = do - trq <- atomically RQ.empty + trq <- RQ.empty atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq checkDataInvariant trq `shouldReturn` True atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq @@ -57,7 +57,7 @@ hasConnTest = do hasConnTestBatch :: IO () hasConnTestBatch = do - trq <- atomically RQ.empty + trq <- RQ.empty let qs = [dummyRQ 0 "smp://1234-w==@alpha" "c1", dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@beta" "c3"] atomically $ RQ.batchAddQueues trq qs checkDataInvariant trq `shouldReturn` True @@ -68,7 +68,7 @@ hasConnTestBatch = do batchIdempotentTest :: IO () batchIdempotentTest = do - trq <- atomically RQ.empty + trq <- RQ.empty let qs = [dummyRQ 0 "smp://1234-w==@alpha" "c1", dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@beta" "c3"] atomically $ RQ.batchAddQueues trq qs checkDataInvariant trq `shouldReturn` True @@ -81,7 +81,7 @@ batchIdempotentTest = do deleteConnTest :: IO () deleteConnTest = do - trq <- atomically RQ.empty + trq <- RQ.empty atomically $ do RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq @@ -95,7 +95,7 @@ deleteConnTest = do getSessQueuesTest :: IO () getSessQueuesTest = do - trq <- atomically RQ.empty + trq <- RQ.empty atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq checkDataInvariant trq `shouldReturn` True atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq @@ -104,14 +104,22 @@ getSessQueuesTest = do checkDataInvariant trq `shouldReturn` True atomically $ RQ.addQueue (dummyRQ 1 "smp://1234-w==@beta" "c4") trq checkDataInvariant trq `shouldReturn` True - atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c1"] - atomically (RQ.getSessQueues (1, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` [] - atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "nope") trq) `shouldReturn` [] - atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"] + let tSess1 = (0, "smp://1234-w==@alpha", Just "c1") + RQ.getSessQueues tSess1 trq `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c1"] + atomically (RQ.hasSessQueues tSess1 trq) `shouldReturn` True + let tSess2 = (1, "smp://1234-w==@alpha", Just "c1") + RQ.getSessQueues tSess2 trq `shouldReturn` [] + atomically (RQ.hasSessQueues tSess2 trq) `shouldReturn` False + let tSess3 = (0, "smp://1234-w==@alpha", Just "nope") + RQ.getSessQueues tSess3 trq `shouldReturn` [] + atomically (RQ.hasSessQueues tSess3 trq) `shouldReturn` False + let tSess4 = (0, "smp://1234-w==@alpha", Nothing) + RQ.getSessQueues tSess4 trq `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"] + atomically (RQ.hasSessQueues tSess4 trq) `shouldReturn`True getDelSessQueuesTest :: IO () getDelSessQueuesTest = do - trq <- atomically RQ.empty + trq <- RQ.empty let qs = [ ("1", dummyRQ 0 "smp://1234-w==@alpha" "c1"), ("1", dummyRQ 0 "smp://1234-w==@alpha" "c2"), @@ -140,7 +148,7 @@ getDelSessQueuesTest = do removeSubsTest :: IO () removeSubsTest = do - aq <- atomically RQ.empty + aq <- RQ.empty let qs = [ ("1", dummyRQ 0 "smp://1234-w==@alpha" "c1"), ("1", dummyRQ 0 "smp://1234-w==@alpha" "c2"), @@ -149,7 +157,7 @@ removeSubsTest = do ] atomically $ RQ.batchAddQueues aq qs - pq <- atomically RQ.empty + pq <- RQ.empty atomically (totalSize aq pq) `shouldReturn` (4, 4) atomically $ RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) "1" aq >>= RQ.batchAddQueues pq . map ("1",) . fst