[config] notifications: configurable CE timeout & Dispatcher throttle

bin/gargantext-cli/CLI/Ini.hs

@@ -141,5 +141,7 @@ defaultNotificationsConfig :: CTypes.NotificationsConfig
 defaultNotificationsConfig =
   CTypes.NotificationsConfig { _nc_central_exchange_bind = "tcp://*:5560"
                              , _nc_central_exchange_connect = "tcp://localhost:5560"
+                             , _nc_ce_send_timeout_ms = 200
                              , _nc_dispatcher_bind = "tcp://*:5561"
-                             , _nc_dispatcher_connect = "tcp://localhost:5561" }
+                             , _nc_dispatcher_connect = "tcp://localhost:5561"
+                             , _nc_dispatcher_throttle_ms = 500 }

gargantext-settings.toml_toModify

@@ -123,9 +123,17 @@ smtp_host = "localhost"
 # HOST_password = password
 
 
-[notifications]
-central-exchange = { bind = "tcp://*:5560", connect = "tcp://127.0.0.1:5560" }
-dispatcher = { bind = "tcp://*:5561", connect = "tcp://127.0.0.1:5561" }
+[notifications.central-exchange]
+bind = "tcp://:5560"
+connect = "tcp://127.0.0.1:5560"
+# see https://gitlab.iscpif.fr/gargantext/haskell-gargantext/commit/77a687ea1483441675320fd2413fac52bb112a4c
+send_timeout_ms = 200
+
+[notifications.dispatcher]
+bind = "tcp://:5561"
+connect = "tcp://127.0.0.1:5561"
+# Same dispatcher messages are throttled, this is the throttle delay
+throttle_ms = 500
 
 
 [nlp]

src/Gargantext/Core/Config/Types.hs

@@ -323,21 +323,25 @@ makeLenses ''APIsConfig
 data NotificationsConfig =
   NotificationsConfig { _nc_central_exchange_bind    :: ~T.Text
                       , _nc_central_exchange_connect :: ~T.Text
+                      , _nc_ce_send_timeout_ms       :: ~Int
                       , _nc_dispatcher_bind          :: ~T.Text
-                      , _nc_dispatcher_connect       :: ~T.Text }
+                      , _nc_dispatcher_connect       :: ~T.Text
+                      , _nc_dispatcher_throttle_ms   :: ~Int }
   deriving (Show, Eq)
 instance FromValue NotificationsConfig where
   fromValue = parseTableFromValue $ do
-    (_nc_central_exchange_bind, _nc_central_exchange_connect) <-
+    (_nc_central_exchange_bind, _nc_central_exchange_connect, _nc_ce_send_timeout_ms) <-
       reqKeyOf "central-exchange" $ parseTableFromValue $ do
         b <- reqKey "bind"
         c <- reqKey "connect"
-        pure (b, c)
-    (_nc_dispatcher_bind, _nc_dispatcher_connect) <-
+        t <- reqKey "send_timeout_ms"
+        pure (b, c, t)
+    (_nc_dispatcher_bind, _nc_dispatcher_connect, _nc_dispatcher_throttle_ms) <-
       reqKeyOf "dispatcher" $ parseTableFromValue $ do
         b <- reqKey "bind"
         c <- reqKey "connect"
-        pure (b, c)
+        t <- reqKey "throttle_ms"
+        pure (b, c, t)
     return $ NotificationsConfig { .. }
 instance ToValue NotificationsConfig where
   toValue = defaultTableToValue
@@ -345,8 +349,10 @@ instance ToTable NotificationsConfig where
   toTable (NotificationsConfig { .. }) =
     table [ "central-exchange" .=
             table [ "bind" .= _nc_central_exchange_bind
-                  , "connect" .= _nc_central_exchange_connect ]
+                  , "connect" .= _nc_central_exchange_connect
+                  , "send_timeout_ms" .= _nc_ce_send_timeout_ms ]
           , "dispatcher" .=
             table [ "bind" .= _nc_dispatcher_bind
-                  , "connect" .= _nc_dispatcher_connect ]
+                  , "connect" .= _nc_dispatcher_connect
+                  , "throttle" .= _nc_dispatcher_throttle_ms ]
           ]

src/Gargantext/Core/Notifications/CentralExchange.hs

@@ -74,7 +74,7 @@ gServer cfg = do
               -- C.putStrLn $ "[central_exchange] " <> r
               atomically $ TChan.writeTChan tChan r
   where
-    NotificationsConfig{..} = cfg ^. gc_notifications_config
+    nc@NotificationsConfig{..} = cfg ^. gc_notifications_config
     log_cfg = cfg ^. gc_logging
     worker s_dispatcher tChan = do
       withLogger log_cfg $ \ioLogger -> do
@@ -99,24 +99,20 @@ gServer cfg = do
               -- process, independent of the server.
               -- send the same message that we received
               -- void $ sendNonblocking s_dispatcher r
-              sendTimeout ioLogger s_dispatcher r
+              sendTimeout nc ioLogger s_dispatcher r
             Just (UpdateWorkerProgress _ji _jl) -> do
               -- $(logLoc) ioLogger DEBUG $ "[central_exchange] update worker progress: " <> show ji <> ", " <> show jl
-              sendTimeout ioLogger s_dispatcher r
+              sendTimeout nc ioLogger s_dispatcher r
             Just Ping -> do
-              sendTimeout ioLogger s_dispatcher r
+              sendTimeout nc ioLogger s_dispatcher r
             Nothing ->
               $(logLoc) ioLogger ERROR $ "[central_exchange] cannot decode message: " <> show r
 
--- | A static send timeout in microseconds.
-send_timeout_us :: Int
-send_timeout_us = 50_000
-
--- | Sends the given payload ensure the send doesn't take more than the static
--- 'send_timeout_ns', logging a message if the timeouts kicks in.
-sendTimeout :: Sender a => Logger IO -> Socket a -> ByteString -> IO ()
-sendTimeout ioLogger sock payload = withFrozenCallStack $ do
-  timeoutKickedIn <- timeout send_timeout_us $ send sock $ payload
+-- | Sends the given payload ensure the send doesn't take more than the
+-- 'nc_ce_send_timeout_ms', logging a message if the timeouts kicks in.
+sendTimeout :: Sender a => NotificationsConfig -> Logger IO -> Socket a -> ByteString -> IO ()
+sendTimeout (NotificationsConfig { _nc_ce_send_timeout_ms }) ioLogger sock payload = withFrozenCallStack $ do
+  timeoutKickedIn <- timeout (_nc_ce_send_timeout_ms * 1000) $ send sock $ payload
   case timeoutKickedIn of
     Nothing ->
       $(logLoc) ioLogger ERROR $ "[central_exchange] couldn't send msg in timely fashion."
@@ -132,8 +128,8 @@ notify cfg ceMessage = withLogger log_cfg $ \ioLogger -> do
             $(logLoc) ioLogger DEBUG $ "[central_exchange] sending: " <> (TE.decodeUtf8 $ BSL.toStrict str)
             -- err <- sendNonblocking s $ BSL.toStrict str
             -- putText $ "[notify] err: " <> show err
-            sendTimeout ioLogger s (BSL.toStrict str)
+            sendTimeout nc ioLogger s (BSL.toStrict str)
       do_work `finally` shutdown s connectEndpoint
   where
-    NotificationsConfig { _nc_central_exchange_connect } = cfg ^. gc_notifications_config
+    nc@NotificationsConfig { _nc_central_exchange_connect } = cfg ^. gc_notifications_config
     log_cfg = cfg ^. gc_logging

src/Gargantext/Core/Notifications/Dispatcher.hs

@@ -87,7 +87,8 @@ dispatcherListener config subscriptions = do
     -- NOTE I'm not sure that we need more than 1 worker here, but in
     -- theory, the worker can perform things like user authentication,
     -- DB queries etc so it can be slow sometimes.
-    Async.withAsync (throttle 500_000 throttleTChan (sendDataMessageThrottled log_cfg)) $ \_ -> do
+    Async.withAsync (throttle (_nc_dispatcher_throttle_ms * 1000) throttleTChan
+                     (sendDataMessageThrottled log_cfg)) $ \_ -> do
       void $ Async.concurrently (Async.replicateConcurrently 5 $ worker tChan throttleTChan) $ do
         forever $ do
           -- putText "[dispatcher_listener] receiving"
@@ -95,7 +96,7 @@ dispatcherListener config subscriptions = do
           -- C.putStrLn $ "[dispatcher_listener] " <> r
           atomically $ TChan.writeTChan tChan r
   where
-    NotificationsConfig { _nc_dispatcher_bind } = config ^. gc_notifications_config
+    NotificationsConfig { _nc_dispatcher_bind, _nc_dispatcher_throttle_ms } = config ^. gc_notifications_config
     log_cfg = config ^. gc_logging
     worker tChan throttleTChan = withLogger log_cfg $ \ioL -> do
       tId <- myThreadId

test-data/test_config.toml

@@ -66,11 +66,17 @@ from = ""
 login_type = "Normal"
 
 
-[notifications]
 # We do not hardcode the bind and connect here, because the test infrastructure
 # will randomize the connection endpoints via IPC.
-central-exchange = { bind = "", connect = "" }
-dispatcher = { bind = "", connect = "" }
+[notifications.central-exchange]
+bind = ""
+connect = ""
+send_timeout_ms = 200
+
+[notifications.dispatcher]
+bind = ""
+connect = ""
+throttle_ms = 500
 
 
 [nlp]

test/Test/Database/Setup.hs

@@ -87,8 +87,10 @@ withTestNotificationConfig cfg action = do
       action $ cfg & gc_notifications_config
                    .~ NotificationsConfig { _nc_central_exchange_bind = "ipc://" <> ce_fp
                                           , _nc_central_exchange_connect = "ipc://" <> ce_fp
+                                          , _nc_ce_send_timeout_ms = 200
                                           , _nc_dispatcher_bind = "ipc://" <> ds_fp
                                           , _nc_dispatcher_connect = "ipc://" <> ds_fp
+                                          , _nc_dispatcher_throttle_ms = 500
                                           }
 
 setup :: IO TestEnv