Control.Funflow.External.Coordinator

A Funflow coordinator is used to distribute tasks amongst multiple executors. It provides a functionality to submit tasks, to fetch them for execution, and to check on their status.

There are multiple possible instantiations of the Coordinator class.

Documentation

newtype Executor

Information about an executor capable of running tasks. Currently this is just a newtype wrapper around hostname.

Constructors

Executor HostName

Instances

Show Executor

Defined in Control.Funflow.External.Coordinator

Methods

showsPrec :: Int -> Executor -> ShowS

show :: Executor -> String

showList :: [Executor] -> ShowS

Store Executor

Defined in Control.Funflow.External.Coordinator

Methods

size :: Size Executor

poke :: Executor -> Poke ()

peek :: Peek Executor

data TaskStatus

Constructors

Pending

Task is in the queue and has not begun executing

Running ExecutionInfo

Completed ExecutionInfo

Failed ExecutionInfo Int

Task has failed with failure count

Instances

Show TaskStatus

Defined in Control.Funflow.External.Coordinator

Methods

showsPrec :: Int -> TaskStatus -> ShowS

show :: TaskStatus -> String

showList :: [TaskStatus] -> ShowS

Store TaskStatus

Defined in Control.Funflow.External.Coordinator

Methods

size :: Size TaskStatus

poke :: TaskStatus -> Poke ()

peek :: Peek TaskStatus

data TaskInfo

Constructors

KnownTask TaskStatus

UnknownTask

Instances

Show TaskInfo

Defined in Control.Funflow.External.Coordinator

Methods

showsPrec :: Int -> TaskInfo -> ShowS

show :: TaskInfo -> String

showList :: [TaskInfo] -> ShowS

Store TaskInfo

Defined in Control.Funflow.External.Coordinator

Methods

size :: Size TaskInfo

poke :: TaskInfo -> Poke ()

peek :: Peek TaskInfo

data ExecutionInfo

Constructors

ExecutionInfo

Fields

• _eiExecutor :: Executor
• _eiElapsed :: TimeSpec

Instances

Show ExecutionInfo

Defined in Control.Funflow.External.Coordinator

Methods

showsPrec :: Int -> ExecutionInfo -> ShowS

show :: ExecutionInfo -> String

showList :: [ExecutionInfo] -> ShowS

Store ExecutionInfo

Defined in Control.Funflow.External.Coordinator

Methods

size :: Size ExecutionInfo

poke :: ExecutionInfo -> Poke ()

peek :: Peek ExecutionInfo

data TaskError

Constructors

ExternalTaskFailed TaskDescription TaskInfo (Maybe (Path Abs File)) (Maybe (Path Abs File))

Instances

Show TaskError

Defined in Control.Funflow.External.Coordinator

Methods

showsPrec :: Int -> TaskError -> ShowS

show :: TaskError -> String

showList :: [TaskError] -> ShowS

Exception TaskError

Defined in Control.Funflow.External.Coordinator

Methods

toException :: TaskError -> SomeException

fromException :: SomeException -> Maybe TaskError

displayException :: TaskError -> String

class Coordinator c where

Associated Types

type Config c

type Hook c = h | h -> c

Methods

initialise :: MonadIO m => Config c -> m (Hook c)

Perform any necessary initialisation to connect to the coordinator.

submitTask :: MonadIO m => Hook c -> TaskDescription -> m ()

Submit a task to the task queue. It is allowed to overwrite a known task.

queueSize :: MonadIO m => Hook c -> m Int

View the size of the current task queue

taskInfo :: MonadIO m => Hook c -> ContentHash -> m TaskInfo

Fetch information on the current task

popTask :: MonadIO m => Hook c -> Executor -> m (Maybe TaskDescription)

Pop a task off of the queue for execution. The popped task should be added to the execution queue

awaitTask :: MonadIO m => Hook c -> ContentHash -> m TaskInfo

Await task completion.

If the task is complete, this will return 'KnownTask Completed'. If the task is failed, this will return 'KnownTask Failed'. If the task is not known to the system, this will return UnknownTask. Otherwise (if the task is pending or running), this will block until the task either completes or fails.

updateTaskStatus :: MonadIO m => Hook c -> ContentHash -> TaskStatus -> m ()

Update execution status for a running task. This should error for a task which is not running.

dropTasks :: MonadIO m => Hook c -> m ()

Remove all pending tasks from the queue.

Instances

Coordinator SQLite

Defined in Control.Funflow.External.Coordinator.SQLite

Associated Types

type Config SQLite :: Type

type Hook SQLite = (h :: Type)

Methods

initialise :: MonadIO m => Config SQLite -> m (Hook SQLite)

submitTask :: MonadIO m => Hook SQLite -> TaskDescription -> m ()

queueSize :: MonadIO m => Hook SQLite -> m Int

taskInfo :: MonadIO m => Hook SQLite -> ContentHash -> m TaskInfo

popTask :: MonadIO m => Hook SQLite -> Executor -> m (Maybe TaskDescription)

awaitTask :: MonadIO m => Hook SQLite -> ContentHash -> m TaskInfo

updateTaskStatus :: MonadIO m => Hook SQLite -> ContentHash -> TaskStatus -> m ()

dropTasks :: MonadIO m => Hook SQLite -> m ()

Coordinator RedisPreconnected

Allow a preestablished redis connection to be used.

Defined in Control.Funflow.External.Coordinator.Redis

Associated Types

type Config RedisPreconnected :: Type

type Hook RedisPreconnected = (h :: Type)

Methods

initialise :: MonadIO m => Config RedisPreconnected -> m (Hook RedisPreconnected)

submitTask :: MonadIO m => Hook RedisPreconnected -> TaskDescription -> m ()

queueSize :: MonadIO m => Hook RedisPreconnected -> m Int

taskInfo :: MonadIO m => Hook RedisPreconnected -> ContentHash -> m TaskInfo

popTask :: MonadIO m => Hook RedisPreconnected -> Executor -> m (Maybe TaskDescription)

awaitTask :: MonadIO m => Hook RedisPreconnected -> ContentHash -> m TaskInfo

updateTaskStatus :: MonadIO m => Hook RedisPreconnected -> ContentHash -> TaskStatus -> m ()

dropTasks :: MonadIO m => Hook RedisPreconnected -> m ()

Coordinator Redis

Defined in Control.Funflow.External.Coordinator.Redis

Associated Types

type Config Redis :: Type

type Hook Redis = (h :: Type)

Methods

initialise :: MonadIO m => Config Redis -> m (Hook Redis)

submitTask :: MonadIO m => Hook Redis -> TaskDescription -> m ()

queueSize :: MonadIO m => Hook Redis -> m Int

taskInfo :: MonadIO m => Hook Redis -> ContentHash -> m TaskInfo

popTask :: MonadIO m => Hook Redis -> Executor -> m (Maybe TaskDescription)

awaitTask :: MonadIO m => Hook Redis -> ContentHash -> m TaskInfo

updateTaskStatus :: MonadIO m => Hook Redis -> ContentHash -> TaskStatus -> m ()

dropTasks :: MonadIO m => Hook Redis -> m ()

Coordinator MemoryCoordinator

Defined in Control.Funflow.External.Coordinator.Memory

Associated Types

type Config MemoryCoordinator :: Type

type Hook MemoryCoordinator = (h :: Type)

Methods

initialise :: MonadIO m => Config MemoryCoordinator -> m (Hook MemoryCoordinator)

submitTask :: MonadIO m => Hook MemoryCoordinator -> TaskDescription -> m ()

queueSize :: MonadIO m => Hook MemoryCoordinator -> m Int

taskInfo :: MonadIO m => Hook MemoryCoordinator -> ContentHash -> m TaskInfo

popTask :: MonadIO m => Hook MemoryCoordinator -> Executor -> m (Maybe TaskDescription)

awaitTask :: MonadIO m => Hook MemoryCoordinator -> ContentHash -> m TaskInfo

updateTaskStatus :: MonadIO m => Hook MemoryCoordinator -> ContentHash -> TaskStatus -> m ()

dropTasks :: MonadIO m => Hook MemoryCoordinator -> m ()

eiExecutor :: Lens' ExecutionInfo Executor

eiElapsed :: Lens' ExecutionInfo TimeSpec

startTask :: (Coordinator c, MonadIO m) => Hook c -> m (Maybe TaskDescription)

isInProgress :: (Coordinator c, MonadIO m) => Hook c -> ContentHash -> m Bool

Check if a task is currently 'in progress' - e.g. pending or running.

withPopTask :: (Coordinator c, MonadIO m, MonadMask m, KatipContext m) => Hook c -> Executor -> (TaskDescription -> m (TimeSpec, Either Int ())) -> m (Maybe ())

Pop a task off of the queue for execution. Passes the popped task to the given function for execution. If the function returns success (Right), then the task will be marked as completed in the given time. If the function returns failure (Left), then the task will be marked as failed. If the function raises an exception or is interrupted by an asynchronous exception, then the task will be placed back on the task queue and the exception propagated. Returns Nothing if no task is available and Just () on task completion or regular failure.