smp server: do not open/read journal message queues that are known to be empty (#1406)

* smp server: do not open/read journal message queues that are known to be empty

* cleanup

* version

* close empty queues on first subscription

* revert version

src/Simplex/Messaging/Server.hs

@@ -1821,9 +1821,9 @@ importMessages tty ms f old_ = do
               mergeQuotaMsgs >> writeMsg ms rId q False msg $> (stored, expired, M.insert rId q overQuota)
               where
                 -- if the first message in queue head is "quota", remove it.
-                mergeQuotaMsgs = withMsgQueue ms rId q "mergeQuotaMsgs" $ \mq ->
-                  tryPeekMsg_ mq >>= \case
-                    Just MessageQuota {} -> tryDeleteMsg_ q mq False
+                mergeQuotaMsgs = withPeekMsgQueue ms rId q "mergeQuotaMsgs" $ maybe (pure ()) $ \(mq, msg) ->
+                  case msg of
+                    MessageQuota {} -> tryDeleteMsg_ q mq False
                     _ -> pure ()
         msgErr :: Show e => String -> e -> String
         msgErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)

src/Simplex/Messaging/Server/MsgStore/Journal.hs

@@ -11,8 +11,8 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
-{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.Journal
   ( JournalMsgStore (queues, senders, notifiers, random),
@@ -20,7 +20,6 @@ module Simplex.Messaging.Server.MsgStore.Journal
     JournalMsgQueue (queue, state),
     JMQueue (queueDirectory, statePath),
     JournalStoreConfig (..),
-    getQueueMessages,
     closeMsgQueue,
     closeMsgQueueHandles,
     -- below are exported for tests
@@ -50,7 +49,7 @@ import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (intercalate)
-import Data.Maybe (catMaybes, fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isNothing)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -105,7 +104,9 @@ data JournalQueue = JournalQueue
     queueRec :: TVar (Maybe QueueRec),
     msgQueue_ :: TVar (Maybe JournalMsgQueue),
     -- system time in seconds since epoch
-    activeAt :: TVar Int64
+    activeAt :: TVar Int64,
+    -- Just True - empty, Just False - non-empty, Nothing - unknown
+    isEmpty :: TVar (Maybe Bool)
   }
 
 data JMQueue = JMQueue
@@ -224,10 +225,11 @@ instance STMQueueStore JournalMsgStore where
   storeLog' = storeLog
   mkQueue st qr = do
     lock <- getMapLock (queueLocks st) $ recipientId qr
-    q <- newTVar $! Just qr
+    q <- newTVar $ Just qr
     mq <- newTVar Nothing
     activeAt <- newTVar 0
-    pure $ JournalQueue lock q mq activeAt
+    isEmpty <- newTVar Nothing
+    pure $ JournalQueue lock q mq activeAt isEmpty
   msgQueue_' = msgQueue_
 
 instance MsgStoreClass JournalMsgStore where
@@ -322,7 +324,7 @@ instance MsgStoreClass JournalMsgStore where
             statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
             queue = JMQueue {queueDirectory = dir, statePath}
         q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
-        atomically $ writeTVar msgQueue_ $! Just q
+        atomically $ writeTVar msgQueue_ $ Just q
         pure q
         where
           createQ :: JMQueue -> IO JournalMsgQueue
@@ -332,14 +334,39 @@ instance MsgStoreClass JournalMsgStore where
             journalId <- newJournalId random
             mkJournalQueue queue (newMsgQueueState journalId) Nothing
 
+  getPeekMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
+  getPeekMsgQueue ms rId q@JournalQueue {isEmpty} =
+    StoreIO (readTVarIO isEmpty) >>= \case
+      Just True -> pure Nothing
+      Just False -> peek
+      Nothing -> do
+        -- We only close the queue if we just learnt it's empty.
+        -- This is needed to reduce file descriptors and memory usage
+        -- after the server just started and many clients subscribe.
+        -- In case the queue became non-empty on write and then again empty on read
+        -- we won't be closing it, to avoid frequent open/close on active queues.
+        r <- peek
+        when (isNothing r) $ StoreIO $ closeMsgQueue q
+        pure r
+    where
+      peek = do
+        mq <- getMsgQueue ms rId q
+        (mq,) <$$> tryPeekMsg_ q mq
+
+  -- only runs action if queue is not empty
   withIdleMsgQueue :: Int64 -> JournalMsgStore -> RecipientId -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
   withIdleMsgQueue now ms@JournalMsgStore {config} rId q action =
     StoreIO $ readTVarIO (msgQueue_ q) >>= \case
       Nothing ->
-        E.bracket (unStoreIO $ getMsgQueue ms rId q) (\_ -> closeMsgQueue q) $ \mq -> unStoreIO $ do
-          r <- action mq
-          sz <- getQueueSize_ mq
-          pure (Just r, sz)
+        E.bracket
+          (unStoreIO $ getPeekMsgQueue ms rId q)
+          (mapM_ $ \_ -> closeMsgQueue q)
+          (maybe (pure (Nothing, 0)) (unStoreIO . run))
+        where
+          run (mq, _) = do
+            r <- action mq
+            sz <- getQueueSize_ mq
+            pure (Just r, sz)
       Just mq -> do
         ts <- readTVarIO $ activeAt q
         r <- if now - ts >= idleInterval config
@@ -378,6 +405,7 @@ instance MsgStoreClass JournalMsgStore where
       let empty = size == 0
       if canWrite || empty
         then do
+          atomically $ writeTVar (isEmpty q') (Just False)
           let canWrt' = quota > size
           if canWrt'
             then writeToJournal q st canWrt' msg $> Just (msg, empty)
@@ -426,9 +454,9 @@ instance MsgStoreClass JournalMsgStore where
   getQueueSize_ :: JournalMsgQueue -> StoreIO Int
   getQueueSize_ JournalMsgQueue {state} = StoreIO $ size <$> readTVarIO state
 
-  tryPeekMsg_ :: JournalMsgQueue -> StoreIO (Maybe Message)
-  tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} =
-    StoreIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
+  tryPeekMsg_ :: JournalQueue -> JournalMsgQueue -> StoreIO (Maybe Message)
+  tryPeekMsg_ q mq@JournalMsgQueue {tipMsg, handles} =
+    StoreIO $ (readTVarIO handles $>>= chooseReadJournal mq True $>>= peekMsg) >>= setEmpty
     where
       peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
         where
@@ -436,6 +464,9 @@ instance MsgStoreClass JournalMsgStore where
             ml@(msg, _) <- hGetMsgAt h $ bytePos rs
             atomically $ writeTVar tipMsg $ Just (Just ml)
             pure $ Just msg
+      setEmpty msg = do
+        atomically $ writeTVar (isEmpty q) (Just $ isNothing msg)
+        pure msg
 
   tryDeleteMsg_ :: JournalQueue -> JournalMsgQueue -> Bool -> StoreIO ()
   tryDeleteMsg_ q mq@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` when logState (updateActiveAt q)) $

src/Simplex/Messaging/Server/MsgStore/STM.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
@@ -8,8 +7,8 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.STM
   ( STMMsgStore (..),
@@ -29,7 +28,7 @@ import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util ((<$$>))
+import Simplex.Messaging.Util ((<$$>), ($>>=))
 import System.IO (IOMode (..))
 
 data STMMsgStore = STMMsgStore
@@ -63,7 +62,7 @@ instance STMQueueStore STMMsgStore where
   senders' = senders
   notifiers' = notifiers
   storeLog' = storeLog
-  mkQueue _ qr = STMQueue <$> (newTVar $! Just qr) <*> newTVar Nothing
+  mkQueue _ qr = STMQueue <$> newTVar (Just qr) <*> newTVar Nothing
   msgQueue_' = msgQueue_
 
 instance MsgStoreClass STMMsgStore where
@@ -106,9 +105,12 @@ instance MsgStoreClass STMMsgStore where
         canWrite <- newTVar True
         size <- newTVar 0
         let q = STMMsgQueue {msgQueue, canWrite, size}
-        writeTVar msgQueue_ $! Just q
+        writeTVar msgQueue_ (Just q)
         pure q
 
+  getPeekMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
+  getPeekMsgQueue _ _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
+
   -- does not create queue if it does not exist, does not delete it if it does (can't just close in-memory queue)
   withIdleMsgQueue :: Int64 -> STMMsgStore -> RecipientId -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
   withIdleMsgQueue _ _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= \case
@@ -159,8 +161,8 @@ instance MsgStoreClass STMMsgStore where
   getQueueSize_ :: STMMsgQueue -> STM Int
   getQueueSize_ STMMsgQueue {size} = readTVar size
 
-  tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message)
-  tryPeekMsg_ = tryPeekTQueue . msgQueue
+  tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message)
+  tryPeekMsg_ _ = tryPeekTQueue . msgQueue
   {-# INLINE tryPeekMsg_ #-}
 
   tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM ()

src/Simplex/Messaging/Server/MsgStore/Types.hs

@@ -4,15 +4,20 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiWayIf #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}
+
+{-# HLINT ignore "Redundant multi-way if" #-}
 
 module Simplex.Messaging.Server.MsgStore.Types where
 
 import Control.Concurrent.STM
 import Control.Monad (foldM)
 import Control.Monad.Trans.Except
+import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Kind
 import qualified Data.Map.Strict as M
@@ -21,6 +26,7 @@ import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.StoreLog.Types
 import Simplex.Messaging.TMap (TMap)
+import Simplex.Messaging.Util ((<$$>))
 import System.IO (IOMode (..))
 
 class MsgStoreClass s => STMQueueStore s where
@@ -44,7 +50,9 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   logQueueStates :: s -> IO ()
   logQueueState :: StoreQueue s -> StoreMonad s ()
   queueRec' :: StoreQueue s -> TVar (Maybe QueueRec)
+  getPeekMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
   getMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (MsgQueue s)
+
   -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
   withIdleMsgQueue :: Int64 -> s -> RecipientId -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
   deleteQueue :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType QueueRec)
@@ -53,7 +61,7 @@ class Monad (StoreMonad s) => MsgStoreClass s where
   writeMsg :: s -> RecipientId -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
   setOverQuota_ :: StoreQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
   getQueueSize_ :: MsgQueue s -> StoreMonad s Int
-  tryPeekMsg_ :: MsgQueue s -> StoreMonad s (Maybe Message)
+  tryPeekMsg_ :: StoreQueue s -> MsgQueue s -> StoreMonad s (Maybe Message)
   tryDeleteMsg_ :: StoreQueue s -> MsgQueue s -> Bool -> StoreMonad s ()
   isolateQueue :: RecipientId -> StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
 
@@ -73,39 +81,39 @@ withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= foldM run mempty
       pure $! acc <> r
 
 getQueueMessages :: MsgStoreClass s => Bool -> s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO [Message]
-getQueueMessages drainMsgs st rId q = withMsgQueue st rId q "getQueueSize" $ getQueueMessages_ drainMsgs
+getQueueMessages drainMsgs st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
 {-# INLINE getQueueMessages #-}
 
 getQueueSize :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO Int
-getQueueSize st rId q = withMsgQueue st rId q "getQueueSize" $ getQueueSize_
+getQueueSize st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
 {-# INLINE getQueueSize #-}
 
 tryPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
-tryPeekMsg st rId q = withMsgQueue st rId q "tryPeekMsg" $ tryPeekMsg_
+tryPeekMsg st rId q = snd <$$> withPeekMsgQueue st rId q "tryPeekMsg" pure
 {-# INLINE tryPeekMsg #-}
 
 tryDelMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
 tryDelMsg st rId q msgId' =
-  withMsgQueue st rId q "tryDelMsg" $ \mq ->
-    tryPeekMsg_ mq >>= \case
-      msg_@(Just msg)
+  withPeekMsgQueue st rId q "tryDelMsg" $
+    maybe (pure Nothing) $ \(mq, msg) ->
+      if
         | messageId msg == msgId' ->
-            tryDeleteMsg_ q mq True >> pure msg_
-      _ -> pure Nothing
+            tryDeleteMsg_ q mq True $> Just msg
+        | otherwise -> pure Nothing
 
 -- atomic delete (== read) last and peek next message if available
 tryDelPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
 tryDelPeekMsg st rId q msgId' =
-  withMsgQueue st rId q "tryDelPeekMsg" $ \mq ->
-    tryPeekMsg_ mq >>= \case
-      msg_@(Just msg)
-        | messageId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ mq)
-        | otherwise -> pure (Nothing, msg_)
-      _ -> pure (Nothing, Nothing)
+  withPeekMsgQueue st rId q "tryDelPeekMsg" $
+    maybe (pure (Nothing, Nothing)) $ \(mq, msg) ->
+      if
+        | messageId msg == msgId' -> (Just msg,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ q mq)
+        | otherwise -> pure (Nothing, Just msg)
 
-withMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (MsgQueue s -> StoreMonad s a) -> ExceptT ErrorType IO a
-withMsgQueue st rId q op a = isolateQueue rId q op $ getMsgQueue st rId q >>= a
-{-# INLINE withMsgQueue #-}
+-- The action is called with Nothing when it is known that the queue is empty
+withPeekMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
+withPeekMsgQueue st rId q op a = isolateQueue rId q op $ getPeekMsgQueue st rId q >>= a
+{-# INLINE withPeekMsgQueue #-}
 
 deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
 deleteExpiredMsgs st rId q old =
@@ -126,7 +134,7 @@ deleteExpireMsgs_ old q mq = do
   pure n
   where
     loop dc =
-      tryPeekMsg_ mq >>= \case
+      tryPeekMsg_ q mq >>= \case
         Just Message {msgTs}
           | systemSeconds msgTs < old ->
               tryDeleteMsg_ q mq False >> loop (dc + 1)

src/Simplex/Messaging/Server/QueueStore/STM.hs

@@ -84,7 +84,7 @@ secureQueue st sq sKey =
     secure q@QueueRec {recipientId = rId} = case senderKey q of
       Just k -> pure $ if sKey == k then Right rId else Left AUTH
       Nothing -> do
-        writeTVar qr $! Just q {senderKey = Just sKey}
+        writeTVar qr $ Just q {senderKey = Just sKey}
         pure $ Right rId
 
 addQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
@@ -96,7 +96,7 @@ addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId} =
     add q@QueueRec {recipientId = rId} = ifM (TM.member nId (notifiers' st)) (pure $ Left DUPLICATE_) $ do
       nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId (notifiers' st) $> notifierId
       let !q' = q {notifier = Just ntfCreds}
-      writeTVar qr $! Just q'
+      writeTVar qr $ Just q'
       TM.insert nId rId $ notifiers' st
       pure $ Right (rId, nId_)