From 2d2e7cee9e05ff82bc586ee10a7d659438609d5d Mon Sep 17 00:00:00 2001 From: spaced4ndy <8711996+spaced4ndy@users.noreply.github.com> Date: Tue, 12 Dec 2023 13:28:10 +0400 Subject: [PATCH] implement batching --- src/Simplex/Chat.hs | 82 +++++++++++++++++++++++++--------- src/Simplex/Chat/Controller.hs | 1 + src/Simplex/Chat/View.hs | 1 + 3 files changed, 62 insertions(+), 22 deletions(-) diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index 15c70ed530..6631f1af3a 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -39,7 +39,7 @@ import Data.Fixed (div') import Data.Functor (($>)) import Data.Int (Int64) import Data.List (find, foldl', isSuffixOf, partition, sortBy, sortOn) -import Data.List.NonEmpty (NonEmpty, nonEmpty) +import Data.List.NonEmpty (NonEmpty (..), nonEmpty, toList) import qualified Data.List.NonEmpty as L import Data.Map.Strict (Map) import qualified Data.Map.Strict as M @@ -120,7 +120,7 @@ import UnliftIO.STM -- this limit reserves space for metadata in forwarded messages -- 15780 (limit used for fileChunkSize) - 161 (x.grp.msg.forward overhead) = 15619, round to 15610 -maxChatMsgSize :: Int64 +maxChatMsgSize :: Int maxChatMsgSize = 15610 defaultChatConfig :: ChatConfig @@ -5536,28 +5536,66 @@ createSndMessage chatMsgEvent connOrGroupId = do let msgBody = encodeChatMessage ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent} in NewMessage {chatMsgEvent, msgBody} -sendBatchedDirectMessages :: (MsgEncodingI e, ChatMonad m) => Connection -> [ChatMsgEvent e] -> ConnOrGroupId -> m (SndMessage, Int64) -sendBatchedDirectMessages conn events connOrGroupId = do +sendBatchedDirectMessages :: forall e m. (MsgEncodingI e, ChatMonad m) => Connection -> NonEmpty (ChatMsgEvent e) -> ConnOrGroupId -> m () +sendBatchedDirectMessages conn@Connection {connId} events connOrGroupId = do when (connDisabled conn) $ throwChatError (CEConnectionDisabled conn) - msg@SndMessage {msgId, msgBody} <- createBatchedSndMessage events connOrGroupId - (msg,) <$> deliverMessage' conn MsgFlags {notification = True} msgBody msgId + (errs, msgs) <- partitionEithers <$> createSndMessages + toView $ CRChatErrors Nothing (map ChatErrorStore errs) + forM_ (L.nonEmpty msgs) $ \msgs' -> do + let (largeMsgs, msgBatches) = partitionBatches $ batchChatMessages msgs' + errs' = map (\SndMessage {msgId} -> ChatError $ CELargeChatMsg msgId) largeMsgs + toView $ CRChatErrors Nothing errs' + forM_ msgBatches $ \(MessagesBatch batchMsgBody sndMsgs) -> do + agentMsgId <- withAgent $ \a -> sendMessage a (aConnId conn) MsgFlags {notification = True} batchMsgBody + let sndMsgDelivery = SndMsgDelivery {connId, agentMsgId} + withStore' $ \db -> forM_ sndMsgs $ \SndMessage {msgId} -> + createSndMsgDelivery db sndMsgDelivery msgId + where + createSndMessages :: m [Either StoreError SndMessage] + createSndMessages = do + gVar <- asks idsDrg + ChatConfig {chatVRange} <- asks config + withStore' $ \db -> forM (toList events) $ \event -> + runExceptT $ createNewSndMessage db gVar connOrGroupId (newMsg chatVRange event) + newMsg chatVRange chatMsgEvent sharedMsgId = + let msgBody = encodeChatMessage ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent} + in NewMessage {chatMsgEvent, msgBody} + partitionBatches :: [ChatMessageBatch] -> ([SndMessage], [MessagesBatch]) + partitionBatches = foldr partition' ([], []) + where + partition' :: ChatMessageBatch -> ([SndMessage], [MessagesBatch]) -> ([SndMessage], [MessagesBatch]) + partition' (CMBMessages msgBatch) (largeMsgs, msgBatches) = (largeMsgs, msgBatch : msgBatches) + partition' (CMBLargeMessage largeMsg) (largeMsgs, msgBatches) = (largeMsg : largeMsgs, msgBatches) -createBatchedSndMessage :: (MsgEncodingI e, ChatMonad m) => [ChatMsgEvent e] -> ConnOrGroupId -> m SndMessage -createBatchedSndMessage events connOrGroupId = do - gVar <- asks idsDrg - ChatConfig {chatVRange} <- asks config - -- TODO [batch send] save batched messages - -- - use tag of first event for 'chat_msg_event'? - -- - add flag 'batched'? - -- - loop: - -- - add events to batch while it fits max message size - -- - save batch as SndMessage, deliver it (deliverMessage') - -- - ChatMessage encoding should support list of ChatMsgEvents - -- - * return list of SndMessages? it's not necessary for current use cases - withStore $ \db -> createNewSndMessage db gVar connOrGroupId $ \sharedMsgId -> - let chatMsgEvent = XOk -- dummy - msgBody = encodeChatMessage ChatMessage {chatVRange, msgId = Just sharedMsgId, chatMsgEvent} - in NewMessage {chatMsgEvent, msgBody} +data MessagesBatch = MessagesBatch ByteString [SndMessage] + +data ChatMessageBatch + = CMBMessages MessagesBatch + | CMBLargeMessage SndMessage + +batchChatMessages :: NonEmpty SndMessage -> [ChatMessageBatch] +batchChatMessages = mkBatch [] + where + mkBatch :: [ChatMessageBatch] -> NonEmpty SndMessage -> [ChatMessageBatch] + mkBatch bs msgs = + let (b, msgs_) = encodeBatch "[" 0 [] msgs + bs' = b : bs + in maybe bs' (mkBatch bs') msgs_ + encodeBatch :: ByteString -> Int -> [SndMessage] -> NonEmpty SndMessage -> (ChatMessageBatch, Maybe (NonEmpty SndMessage)) + encodeBatch s n batchedMsgs remainingMsgs@(msg :| msgs_) + | B.length s' <= maxChatMsgSize - 1 = + case L.nonEmpty msgs_ of + Just msgs' -> encodeBatch s' n' batchedMsgs' msgs' + Nothing -> (CMBMessages $ MessagesBatch (s' <> "]") (reverse batchedMsgs'), Nothing) + | n == 0 = (CMBLargeMessage msg, L.nonEmpty msgs_) + | otherwise = (CMBMessages $ MessagesBatch (s <> "]") (reverse batchedMsgs), Just remainingMsgs) + where + SndMessage {msgBody} = msg + s' + | n == 0 = s <> msgBody + | otherwise = s <> "," <> msgBody + n' = n + 1 + batchedMsgs' = msg : batchedMsgs directMessage :: (MsgEncodingI e, ChatMonad m) => ChatMsgEvent e -> m ByteString directMessage chatMsgEvent = do diff --git a/src/Simplex/Chat/Controller.hs b/src/Simplex/Chat/Controller.hs index 04c47a6462..d73dba25e9 100644 --- a/src/Simplex/Chat/Controller.hs +++ b/src/Simplex/Chat/Controller.hs @@ -1027,6 +1027,7 @@ data ChatErrorType | CEContactNotActive {contact :: Contact} | CEContactDisabled {contact :: Contact} | CEConnectionDisabled {connection :: Connection} + | CELargeChatMsg {messageId :: Int64} | CEGroupUserRole {groupInfo :: GroupInfo, requiredRole :: GroupMemberRole} | CEGroupMemberInitialRole {groupInfo :: GroupInfo, initialRole :: GroupMemberRole} | CEContactIncognitoCantInvite diff --git a/src/Simplex/Chat/View.hs b/src/Simplex/Chat/View.hs index f801e30753..6cb5de47cc 100644 --- a/src/Simplex/Chat/View.hs +++ b/src/Simplex/Chat/View.hs @@ -1791,6 +1791,7 @@ viewChatError logLevel testView = \case CEContactDisabled ct -> [ttyContact' ct <> ": disabled, to enable: " <> highlight ("/enable " <> viewContactName ct) <> ", to delete: " <> highlight ("/d " <> viewContactName ct)] CEContactNotActive c -> [ttyContact' c <> ": not active"] CEConnectionDisabled Connection {connId, connType} -> [plain $ "connection " <> textEncode connType <> " (" <> tshow connId <> ") is disabled" | logLevel <= CLLWarning] + CELargeChatMsg msgId -> ["large chat message, message_id=" <> sShow msgId] CEGroupDuplicateMember c -> ["contact " <> ttyContact c <> " is already in the group"] CEGroupDuplicateMemberId -> ["cannot add member - duplicate member ID"] CEGroupUserRole g role ->