{-# LANGUAGE Arrows #-}
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE QuasiQuotes #-}
{-# LANGUAGE Rank2Types #-}
{-# LANGUAGE ScopedTypeVariables #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE TypeOperators #-}
module Control.Funflow.Exec.Simple
( runFlow
, runFlowEx
, runSimpleFlow
, withSimpleLocalRunner
) where
import Control.Arrow (returnA)
import Control.Arrow.Async
import Control.Arrow.Free (type (~>), eval)
import Control.Concurrent.Async (withAsync)
import Control.Exception.Safe (Exception,
SomeException,
bracket,
onException,
throwM, try)
import Control.Funflow.Base
import Control.Funflow.ContentHashable
import qualified Control.Funflow.ContentStore as CS
import Control.Funflow.External
import Control.Funflow.External.Coordinator
import Control.Funflow.External.Coordinator.Memory
import Control.Funflow.External.Executor (executeLoop)
import Control.Monad.IO.Class (liftIO)
import Control.Monad.Trans.Class (lift)
import qualified Data.ByteString as BS
import Data.Foldable (traverse_)
import Data.Int (Int64)
import Data.Monoid ((<>))
import Data.Void
import Katip
import Path
import System.IO (stderr)
import System.Random (randomIO)
runFlowEx :: forall c eff ex a b. (Coordinator c, Exception ex)
=> c
-> Config c
-> CS.ContentStore
-> (eff ~> AsyncA (KatipContextT IO))
-> Int
-> Flow eff ex a b
-> a
-> KatipContextT IO b
runFlowEx _ cfg store runWrapped confIdent flow input = do
hook <- initialise cfg
runAsyncA (eval (runFlow' hook) flow) input
where
simpleOutPath item = toFilePath
$ CS.itemPath store item </> [relfile|out|]
withStoreCache :: forall i o. Cacher i o
-> AsyncA (KatipContextT IO) i o -> AsyncA (KatipContextT IO) i o
withStoreCache NoCache f = f
withStoreCache c f = let
chashOf i = cacherKey c confIdent i
checkStore = AsyncA $ \chash ->
CS.constructOrWait store chash >>= \case
CS.Pending void -> absurd void
CS.Complete item -> do
bs <- liftIO . BS.readFile $ simpleOutPath item
return . Right . cacherReadValue c $ bs
CS.Missing fp -> return $ Left fp
writeStore = AsyncA $ \(chash, fp, res) ->
do
liftIO $ BS.writeFile (toFilePath $ fp </> [relfile|out|])
. cacherStoreValue c $ res
_ <- CS.markComplete store chash
return res
`onException`
CS.removeFailed store chash
in proc i -> do
let chash = chashOf i
mcontents <- checkStore -< chash
case mcontents of
Right contents -> returnA -< contents
Left fp -> do
res <- f -< i
writeStore -< (chash, fp, res)
writeMd :: forall i o. ContentHash
-> i
-> o
-> MDWriter i o
-> KatipContextT IO ()
writeMd _ _ _ Nothing = return ()
writeMd chash i o (Just writer) =
let kvs = writer i o
in traverse_ (uncurry $ CS.setMetadata store chash) kvs
runFlow' :: Hook c -> Flow' eff a1 b1 -> AsyncA (KatipContextT IO) a1 b1
runFlow' _ (Step props f) = withStoreCache (cache props)
. AsyncA $ \x -> do
let out = f x
case cache props of
NoCache -> return ()
Cache key _ _ -> writeMd (key confIdent x) x out $ mdpolicy props
return out
runFlow' _ (StepIO props f) = withStoreCache (cache props)
. liftAsyncA $ AsyncA f
runFlow' po (External props toTask) = AsyncA $ \x -> do
chash <- liftIO $ if (ep_impure props)
then do
salt <- randomIO :: IO Int64
contentHash (toTask x, salt)
else contentHash (toTask x)
CS.lookup store chash >>= \case
CS.Complete item -> return item
CS.Pending _ -> taskInfo po chash >>= \case
UnknownTask -> do
CS.removeFailed store chash
writeMd chash x () $ ep_mdpolicy props
submitAndWait chash (TaskDescription chash (toTask x))
KnownTask _ -> wait chash (TaskDescription chash (toTask x))
CS.Missing _ -> do
writeMd chash x () $ ep_mdpolicy props
submitAndWait chash (TaskDescription chash (toTask x))
where
submitAndWait chash td = do
submitTask po td
wait chash td
wait chash td = do
KnownTask _ <- awaitTask po chash
CS.waitUntilComplete store chash >>= \case
Just item -> return item
Nothing -> do
ti <- taskInfo po chash
mbStdout <- CS.getMetadataFile store chash [relfile|stdout|]
mbStderr <- CS.getMetadataFile store chash [relfile|stderr|]
throwM $ ExternalTaskFailed td ti mbStdout mbStderr
runFlow' _ (PutInStore f) = AsyncA $ \x -> katipAddNamespace "putInStore" $ do
chash <- liftIO $ contentHash x
CS.constructOrWait store chash >>= \case
CS.Pending void -> absurd void
CS.Complete item -> return item
CS.Missing fp ->
do
liftIO $ f fp x
CS.markComplete store chash
`onException`
(do $(logTM) WarningS . ls $ "Exception in construction: removing " <> show chash
CS.removeFailed store chash
)
runFlow' _ (GetFromStore f) = AsyncA $ \case
CS.All item -> lift . f $ CS.itemPath store item
item CS.:</> path -> lift . f $ CS.itemPath store item </> path
runFlow' _ (InternalManipulateStore f) = AsyncA $ \i ->lift $ f store i
runFlow' _ (Wrapped props w) = withStoreCache (cache props)
$ runWrapped w
runFlowLog :: forall c eff ex a b. (Coordinator c, Exception ex)
=> c
-> Config c
-> CS.ContentStore
-> (eff ~> AsyncA (KatipContextT IO))
-> Int
-> Flow eff ex a b
-> a
-> KatipContextT IO (Either ex b)
runFlowLog c cfg store runWrapped confIdent flow input =
try $ runFlowEx c cfg store runWrapped confIdent flow input
runFlow :: forall c eff ex a b. (Coordinator c, Exception ex)
=> c
-> Config c
-> CS.ContentStore
-> (eff ~> AsyncA IO)
-> Int
-> Flow eff ex a b
-> a
-> IO (Either ex b)
runFlow c cfg store runWrapped confIdent flow input = do
le <- initLogEnv "funflow" "production"
runKatipContextT le () "runFlow"
$ runFlowLog c cfg store (liftAsyncA . runWrapped) confIdent flow input
runSimpleFlow :: forall c a b. (Coordinator c)
=> c
-> Config c
-> CS.ContentStore
-> SimpleFlow a b
-> a
-> IO (Either SomeException b)
runSimpleFlow c ccfg store flow input = do
handleScribe <- mkHandleScribe ColorIfTerminal stderr InfoS V2
let mkLogEnv = registerScribe "stderr" handleScribe defaultScribeSettings =<< initLogEnv "funflow" "production"
bracket mkLogEnv closeScribes $ \le -> do
let initialContext = ()
initialNamespace = "executeLoop"
runKatipContextT le initialContext initialNamespace
$ runFlowLog c ccfg store runNoEffect 12345 flow input
withSimpleLocalRunner :: Path Abs Dir
-> ((SimpleFlow a b -> a -> IO (Either SomeException b))
-> IO c)
-> IO c
withSimpleLocalRunner storePath action =
CS.withStore storePath $ \store -> do
memHook <- createMemoryCoordinator
withAsync (executeLoop MemoryCoordinator memHook store) $ \_ ->
action $ runSimpleFlow MemoryCoordinator memHook store