Merge branch 'dev-support-concurrent-workers' into 'dev'

[cli] add support to spawn multiple workers of the same type

See merge request !437

bin/gargantext-cli/CLI/Types.hs

@@ -105,6 +105,7 @@ data ServerArgs = ServerArgs
 
 data CLIWorker
   = CLIW_run WorkerArgs
+  | CLIW_runSingleJob WorkerSingleArgs
   | CLIW_runAll WorkerAllArgs
   | CLIW_stats WorkerStatsArgs
   deriving (Show, Eq)
@@ -112,7 +113,12 @@ data CLIWorker
 data WorkerArgs = WorkerArgs
   { worker_toml       :: !SettingsFile
   , worker_name       :: !Text
-  , worker_run_single :: !Bool
+  , worker_count      :: !Int
+  } deriving (Show, Eq)
+
+data WorkerSingleArgs = WorkerSingleArgs
+  { worker_toml       :: !SettingsFile
+  , worker_name       :: !Text
   } deriving (Show, Eq)
 
 data WorkerAllArgs = WorkerAllArgs

bin/gargantext-cli/CLI/Worker.hs

@@ -42,11 +42,40 @@ workerCLI (CLIW_run (WorkerArgs { .. })) = do
                        $ List.take 72
                        $ List.cycle ["_"]) :: Prelude.String)
 
+  forConcurrently_ [1..worker_count] $ \idx -> do
+    withWorkerEnv worker_toml $ \env -> do
+      let log_cfg = env ^. hasConfig . gc_logging
+      withLogger log_cfg $ \ioLogger -> do
+        ___
+        logMsg ioLogger INFO $ "GarganText worker #" <> show idx
+        logMsg ioLogger INFO $ "worker_name: " <> T.unpack worker_name
+        logMsg ioLogger INFO $ "worker toml: " <> _SettingsFile worker_toml
+        ___
+        
+        let ws = env ^. hasConfig . gc_worker
+        case findDefinitionByName ws worker_name of
+          Nothing -> do
+            let workerNames = _wdName <$> (_wsDefinitions ws)
+            let availableWorkers = T.intercalate ", " workerNames
+            putText $ "Worker definition not found! Available workers: " <> availableWorkers
+          Just wd -> do
+            logMsg ioLogger INFO $ "Starting worker '" <> T.unpack worker_name <> "' #" <> show idx
+            logMsg ioLogger DEBUG $ "gc config: " <> show (env ^. hasConfig)
+            logMsg ioLogger DEBUG $ "Worker settings: " <> show ws
+            ___
+            withPGMQWorkerCtrlC env wd $ \a _state -> do
+              -- _ <- runReaderT (sendJob Ping) env
+              wait a
+workerCLI (CLIW_runSingleJob (WorkerSingleArgs { .. })) = do
+  let ___ = putStrLn ((List.concat
+                       $ List.take 72
+                       $ List.cycle ["_"]) :: Prelude.String)
+
   withWorkerEnv worker_toml $ \env -> do
     let log_cfg = env ^. hasConfig . gc_logging
     withLogger log_cfg $ \ioLogger -> do
       ___
-      logMsg ioLogger INFO "GarganText worker"
+      logMsg ioLogger INFO "Running single GarganText job"
       logMsg ioLogger INFO $ "worker_name: " <> T.unpack worker_name
       logMsg ioLogger INFO $ "worker toml: " <> _SettingsFile worker_toml
       ___
@@ -62,13 +91,8 @@ workerCLI (CLIW_run (WorkerArgs { .. })) = do
           logMsg ioLogger DEBUG $ "gc config: " <> show (env ^. hasConfig)
           logMsg ioLogger DEBUG $ "Worker settings: " <> show ws
           ___
-          if worker_run_single then
-            withPGMQWorkerSingleCtrlC env wd $ \a _state -> do
-              wait a
-          else
-            withPGMQWorkerCtrlC env wd $ \a _state -> do
-              -- _ <- runReaderT (sendJob Ping) env
-              wait a
+          withPGMQWorkerSingleCtrlC env wd $ \a _state -> do
+            wait a
 workerCLI (CLIW_runAll (WorkerAllArgs { .. })) = withWorkerEnv worker_toml $ \env -> do
   let log_cfg = env ^. hasConfig . gc_logging
   withLogger log_cfg $ \ioLogger -> runAllWorkers ioLogger worker_toml
@@ -101,7 +125,9 @@ workerCmd = command "worker" (info (helper <*> (fmap CLISub $ fmap CCMD_worker w
 workerParser :: Parser CLIWorker
 workerParser = hsubparser (
   command "run" (info (helper <*> worker_p)
-                  (progDesc "Run a single worker")) <>
+                  (progDesc "Run a worker")) <>
+  command "run-single-job" (info (helper <*> worker_single_p)
+                  (progDesc "Run a single job, then quit")) <>
   command "run-all" (info (helper <*> worker_all_p)
                      (progDesc "Run all worker definitions")) <>
   command "stats" (info (helper <*> stats_p)
@@ -114,8 +140,17 @@ worker_p = fmap CLIW_run $ WorkerArgs
   <*> strOption ( long "name"
                 <> metavar "STRING"
                 <> help "Worker name, as defined in the .toml file" )
-  <*> flag False True ( long "run-single"
-                     <> help "Whether to loop or run a single job from queue" )
+  <*> option auto ( long "count"
+                  <> metavar "INT"
+                  <> help "How many concurrent workers to start"
+                  <> value 1 )
+
+worker_single_p :: Parser CLIWorker
+worker_single_p = fmap CLIW_runSingleJob $ WorkerSingleArgs
+  <$> settings_p
+  <*> strOption ( long "name"
+                <> metavar "STRING"
+                <> help "Worker name, as defined in the .toml file" )
 
 worker_all_p :: Parser CLIWorker
 worker_all_p = fmap CLIW_runAll $ WorkerAllArgs