Kariel myrr/#4 run requests in parallel

app/cli/Main.hs

@@ -22,19 +22,21 @@ import Error
 import Fmt
 import Options.Applicative (execParser)
 import Polysemy
+import Polysemy.Async (Async, asyncToIOFinal)
 import Polysemy.Error (Error, errorToIOFinal)
 import System.Environment (lookupEnv)
 import System.Exit (die, exitFailure)
 import Text.Interpolation.Nyan
 import Toml qualified
 
 runBackendIO
-  :: Sem '[BackendEffect, Error CofferError, Embed IO, Final IO ] a
+  :: Sem '[BackendEffect, Error CofferError, Embed IO, Async, Final IO] a
   -> IO a
 runBackendIO action =
   runBackend action
     & errorToIOFinal @CofferError
     & embedToFinal @IO
+    & asyncToIOFinal
     & runFinal
     >>= \case
       Right a -> pure a

lib/Backend/Commands.hs

@@ -17,16 +17,18 @@ import Coffer.Path
 import Coffer.Path qualified as Path
 import Coffer.Util (catchAndReturn)
 import Config (Config(backends, mainBackend))
+import Control.Concurrent
 import Control.Lens (view)
 import Control.Lens hiding (view)
 import Control.Monad.Extra (whenM)
 import Control.Monad.State
 import Data.Bifunctor (Bifunctor(first))
 import Data.Either (rights)
+import Data.Foldable (foldr')
 import Data.HashMap.Strict ((!?))
 import Data.HashMap.Strict qualified as HashMap
 import Data.List.NonEmpty qualified as NE
-import Data.Maybe (fromMaybe)
+import Data.Maybe (catMaybes, fromMaybe)
 import Data.Set (Set)
 import Data.Set qualified as Set
 import Data.Text (Text)
@@ -41,11 +43,12 @@ import Entry qualified as E
 import Error (CofferError(..), InternalCommandsError(EntryPathDoesntHavePrefix))
 import GHC.Exts (Down(..), sortWith)
 import Polysemy
+import Polysemy.Async (Async, sequenceConcurrently)
 import Polysemy.Error (Error, throw)
 import Validation (Validation(Failure, Success))
 
 runCommand
-  :: (Members '[BackendEffect, Embed IO, Error CofferError] r)
+  :: (Members '[BackendEffect, Embed IO, Error CofferError, Async] r)
   => Config -> Command res -> Sem r res
 runCommand config = \case
   CmdView opts -> catchAndReturn $ viewCmd config opts
@@ -59,7 +62,7 @@ runCommand config = \case
   CmdTag opts -> catchAndReturn $ tagCmd config opts
 
 viewCmd
-  :: (Members '[BackendEffect, Error CofferError, Error ViewResult] r)
+  :: (Members '[BackendEffect, Error CofferError, Error ViewResult, Embed IO, Async] r)
   => Config -> ViewOptions -> Sem r ViewResult
 viewCmd config (ViewOptions qPath@(QualifiedPath backendNameMb _) fieldNameMb) = do
   backend <- getBackend config backendNameMb
@@ -176,7 +179,7 @@ deleteFieldCmd config (DeleteFieldOptions qPath@(QualifiedPath backendNameMb pat
           pure $ DFRSuccess fieldName qPath { qpPath = newEntry }
 
 findCmd
-  :: (Members '[BackendEffect, Error CofferError] r)
+  :: (Members '[BackendEffect, Error CofferError, Embed IO, Async] r)
   => Config -> FindOptions -> Sem r (Maybe Directory)
 findCmd config (FindOptions qPathMb textMb sortMb filters) = do
   let backendNameMb = qPathMb >>= qpBackendName
@@ -270,7 +273,7 @@ findCmd config (FindOptions qPathMb textMb sortMb filters) = do
 
 renameCmd
   :: forall r
-   . (Members '[BackendEffect, Embed IO, Error CofferError, Error RenameResult] r)
+   . (Members '[BackendEffect, Embed IO, Error CofferError, Error RenameResult, Async] r)
   => Config -> RenameOptions -> Sem r RenameResult
 renameCmd
   config
@@ -296,11 +299,14 @@ renameCmd
   -- If directory/entry was successfully copied,
   -- then we can delete old directory/entry without delete errors.
   unless dryRun do
-    forM_ pathsToDelete \(CopyOperation old _) -> do
-      deleteEntry oldBackend (qpPath old ^. path)
+    void $ sequenceConcurrently $ map (deleteOldEntry oldBackend) pathsToDelete
 
   pure $ CPRSuccess dryRun $ getOperationPaths <$> operations
 
+  where
+    deleteOldEntry :: SomeBackend -> CopyOperation -> Sem r ()
+    deleteOldEntry oldBackend (CopyOperation old _) = deleteEntry oldBackend (qpPath old ^. path)
+
 data CopyOperation = CopyOperation
   { coQOld :: QualifiedPath Entry
   , coQNew :: QualifiedPath Entry
@@ -314,7 +320,7 @@ getOperationPaths (CopyOperation old new) =
 {-# ANN buildCopyOperations ("HLint: ignore Redundant <$>" :: Text) #-}
 buildCopyOperations
   :: forall r
-   . (Members '[BackendEffect, Embed IO, Error CofferError, Error CopyResult] r)
+   . (Members '[BackendEffect, Embed IO, Error CofferError, Error CopyResult, Async] r)
   => SomeBackend -> SomeBackend -> QualifiedPath Path -> QualifiedPath Path -> Bool -> Sem r [CopyOperation]
 buildCopyOperations
   oldBackend
@@ -383,13 +389,13 @@ buildCopyOperations
         oldBackend = qpBackendName old
         newBackend = qpBackendName new
 
-runCopyOperations :: (Member BackendEffect r) => SomeBackend -> [CopyOperation] -> Sem r ()
+runCopyOperations :: (Members '[BackendEffect, Async] r) => SomeBackend -> [CopyOperation] -> Sem r ()
 runCopyOperations backend operations = do
   let newEntries = qpPath . coQNew <$> operations
-  forM_ newEntries (writeEntry backend)
+  void $ sequenceConcurrently $ map (writeEntry backend) newEntries
 
 copyCmd
-  :: (Members '[BackendEffect, Embed IO, Error CofferError, Error CopyResult] r)
+  :: (Members '[BackendEffect, Embed IO, Error CofferError, Error CopyResult, Async] r)
   => Config -> CopyOptions -> Sem r CopyResult
 copyCmd
   config
@@ -410,7 +416,7 @@ copyCmd
   pure $ CPRSuccess dryRun $ getOperationPaths <$> operations
 
 deleteCmd
-  :: (Members '[BackendEffect, Embed IO, Error CofferError, Error DeleteResult] r)
+  :: (Members '[BackendEffect, Embed IO, Error CofferError, Error DeleteResult, Async] r)
   => Config -> DeleteOptions -> Sem r DeleteResult
 deleteCmd config (DeleteOptions dryRun qPath@(QualifiedPath backendNameMb _) recursive) = do
   backend <- getBackend config backendNameMb
@@ -424,9 +430,11 @@ deleteCmd config (DeleteOptions dryRun qPath@(QualifiedPath backendNameMb _) rec
     Right dir
       | recursive -> do
           let entries = Dir.allEntries dir
+          let entryPaths = entries ^.. each . E.path
           unless dryRun do
-            forM_ entries \entry -> deleteEntry backend (entry ^. E.path)
-          let qEntryPaths = entries ^.. each . E.path <&> QualifiedPath backendNameMb
+            void $ sequenceConcurrently $ map (deleteEntry backend) entryPaths
+            --forM_ entries \entry -> deleteEntry backend (entry ^. E.path)
+          let qEntryPaths = map (QualifiedPath backendNameMb) entryPaths
           pure $ DRSuccess dryRun qEntryPaths
       | otherwise -> pure $ DRDirectoryFound qPath
 
@@ -483,7 +491,7 @@ pathIsEntry backend entryPath =
 -- | Returns the entry or directory that the path points to.
 -- If the path doesn't exist at all, throws an error.
 getEntryOrDirThrow
-  :: (Members '[BackendEffect, Error CofferError, Error e] r)
+  :: (Members '[BackendEffect, Error CofferError, Error e, Embed IO, Async] r)
   => SomeBackend -> (QualifiedPath Path -> e) -> QualifiedPath Path -> Sem r (Either Entry Directory)
 getEntryOrDirThrow backend mkError qPath@(QualifiedPath _ path) = do
   getEntryOrDir backend path >>= \case
@@ -494,7 +502,7 @@ getEntryOrDirThrow backend mkError qPath@(QualifiedPath _ path) = do
 -- If the path doesn't exist at all, returns `Nothing`.
 getEntryOrDir
   :: forall r
-   . (Members '[BackendEffect, Error CofferError] r)
+   . (Members '[BackendEffect, Error CofferError, Embed IO, Async] r)
   => SomeBackend -> Path -> Sem r (Maybe (Either Entry Directory))
 getEntryOrDir backend path =
   tryGetEntry path >>= \case
@@ -513,26 +521,30 @@ getEntryOrDir backend path =
     -- If the path doesn't exist OR is an entry, returns `Nothing`.
     tryGetDir :: Path -> Sem r (Maybe Directory)
     tryGetDir rootPath = do
-      dir <- execStateT (go rootPath) Dir.emptyDir
+      mDir <- embed $ newMVar Dir.emptyDir
+      go rootPath mDir
+      dir <- embed $ sortDirectoryAlphabetic <$> takeMVar mDir
       if dir == Dir.emptyDir
         then pure Nothing
         else pure $ Just dir
       where
-        go :: Path -> StateT Directory (Sem r) ()
-        go rootPath = do
-          contents <- lift $ fromMaybe (DirectoryContents [] []) <$> listDirectoryContents backend rootPath
-          -- TODO: run in parallel
-          forM_ (contents ^. entryNames) \entryName -> do
-            entry <- lift $ readEntry backend (Path.appendEntryName rootPath entryName)
-            case entry of
-              Just entry -> modify' (Dir.insertEntry entry)
-              -- This entry has been concurrently deleted (e.g. by some other user) _while_ we're traversing the directory.
-              -- We should just ignore it.
-              Nothing -> pure ()
-
-          forM_ (contents ^. directoryNames) \directoryName -> do
-            let subdir = Path [directoryName]
-            go (rootPath <> subdir)
+        go :: Path -> MVar Directory -> Sem r ()
+        go rootPath mDir = do
+          contents <- fromMaybe (DirectoryContents [] []) <$> listDirectoryContents backend rootPath
+          entries <- fmap (catMaybes . catMaybes) $ sequenceConcurrently $ map
+            (\entryName -> readEntry backend $ Path.appendEntryName rootPath entryName)
+            (contents ^. entryNames)
+          embed $ modifyMVar_ mDir (\dir -> return (foldr' Dir.insertEntry dir entries))
+
+          void $ sequenceConcurrently $ map
+            (\directoryName -> do
+              let subdir = Path [directoryName]
+              go (rootPath <> subdir) mDir
+            )
+            (contents ^. directoryNames)
+
+        sortDirectoryAlphabetic :: Directory -> Directory
+        sortDirectoryAlphabetic = Dir.mapDir $ sortWith  (view $ E.path . to entryPathName)
 
 -- | This function gets all entries, that are exist in given entry path.
 --

lib/Web/Server.hs

@@ -36,6 +36,7 @@ import Data.Set qualified as Set
 import Data.Text qualified as T
 import Fmt (Builder, pretty, unlinesF)
 import GHC.Generics (Generic)
+import Polysemy.Async (Async, asyncToIOFinal)
 import Servant.API
 import Servant.Server
 import Web.Types (NewEntry(NewEntry), NewField(NewField))
@@ -103,11 +104,12 @@ handleCopyResult = handleCopyOrRenameResult False
 handleRenameResult :: RenameResult -> Handler [(EntryPath, EntryPath)]
 handleRenameResult = handleCopyOrRenameResult True
 
-runBackendIO' :: Sem '[BackendEffect, Error CofferError, Embed IO, Final IO] a -> IO (Either CofferError a)
+runBackendIO' :: Sem '[BackendEffect, Error CofferError, Embed IO, Async, Final IO] a -> IO (Either CofferError a)
 runBackendIO' action =
   runBackend action
     & errorToIOFinal @CofferError
     & embedToFinal @IO
+    & asyncToIOFinal
     & runFinal
 
 reportErrors :: IO (Either CofferError a) -> Handler a