[flow] PoC for threaded flow

gargantext.cabal

@@ -510,6 +510,8 @@ library
     , ansi-terminal
     , array ^>= 0.5.4.0
     , async ^>= 2.2.4
+    , lifted-async
+    , lifted-base
     , attoparsec ^>= 0.14.4
     , base64-bytestring ^>= 1.2.1.0
     , bimap >= 0.5.0

src/Gargantext/Database/Action/Flow.hs

@@ -55,6 +55,9 @@ module Gargantext.Database.Action.Flow -- (flowDatabase, ngrams2list)
     where
 
 import Conduit
+import Control.Concurrent.Async.Lifted qualified as AsyncL
+import Control.Concurrent.QSem.Lifted qualified as QSemL
+import Control.Exception.Lifted qualified as CEL
 import Control.Exception.Safe qualified as CES
 import Control.Lens ( to, view )
 import Control.Exception.Safe (catch, MonadCatch)
@@ -300,7 +303,7 @@ flow c mkCorpusUser la mfslw (count, docsC) jobHandle = do
   forM_ msgs ce_notify
   -- TODO if public insertMasterDocs else insertUserDocs
   runConduit (zipSources (yieldMany ([1..] :: [Int])) docsC
-    .| CList.chunksOf 5
+    .| CList.chunksOf 10
     .| mapM_C (addDocumentsWithProgress userCorpusId)
     .| sinkNull) `CES.catches`
       [ CES.Handler $ \(e :: ClientError) -> do
@@ -544,13 +547,19 @@ extractNgramsFromDocuments :: forall doc env err m.
                            -> TermType Lang
                            -> [doc]
                            -> m (UncommittedNgrams doc)
-extractNgramsFromDocuments nlpServer lang docs =
-  foldlM go mempty docs
-  where
-    go :: UncommittedNgrams doc -> doc -> m (UncommittedNgrams doc)
-    go !acc inputDoc = do
-      ngrams <- extractNgramsFromDocument nlpServer lang inputDoc
-      pure $ acc <> ngrams
+extractNgramsFromDocuments nlpServer lang docs = do
+  sem <- QSemL.newQSem 10
+  let f = extractNgramsFromDocument nlpServer lang
+  ret <- AsyncL.mapConcurrently (\doc ->
+                                   CEL.bracket_ (QSemL.waitQSem sem) (QSemL.signalQSem sem) (f doc)
+                                ) docs
+  pure $ foldl (<>) mempty ret
+  -- foldlM go mempty docs
+  -- where
+  --   go :: UncommittedNgrams doc -> doc -> m (UncommittedNgrams doc)
+  --   go !acc inputDoc = do
+  --     ngrams <- extractNgramsFromDocument nlpServer lang inputDoc
+  --     pure $ acc <> ngrams
 
 commitNgramsForDocuments :: UniqParameters doc
                          => UncommittedNgrams doc