diff --git a/src/Control/Distributed/Task/Distribution/LogConfiguration.hs b/src/Control/Distributed/Task/Distribution/LogConfiguration.hs --- a/src/Control/Distributed/Task/Distribution/LogConfiguration.hs +++ b/src/Control/Distributed/Task/Distribution/LogConfiguration.hs @@ -7,6 +7,7 @@ import Control.Distributed.Task.Util.Logging (initLogging) +-- | Sets up hslogger with a part logfile, part stdout configuration. initDefaultLogging :: String -> IO () initDefaultLogging suffix = do -- progName <- getExecutablePath diff --git a/src/Control/Distributed/Task/Distribution/RunComputation.hs b/src/Control/Distributed/Task/Distribution/RunComputation.hs --- a/src/Control/Distributed/Task/Distribution/RunComputation.hs +++ b/src/Control/Distributed/Task/Distribution/RunComputation.hs @@ -1,3 +1,6 @@ +{-| + Defines a higher level interface to running calculations. Resolves HDFS input paths. +|-} module Control.Distributed.Task.Distribution.RunComputation ( MasterOptions(..), TaskSpec(..), @@ -19,32 +22,46 @@ import Control.Distributed.Task.Util.Configuration import Control.Distributed.Task.Util.Logging +-- | The definition of a distributed calculation. data MasterOptions = MasterOptions { + -- | the master hostname _host :: String, + -- | the master port _port :: Int, + -- | the task logic _taskSpec :: TaskSpec, + -- | which data to process _dataSpecs :: DataSpec, + -- | how to process the result _resultSpec :: ResultSpec } -{- - ObjectCodeModuleDeployment: - - the function here is ignored, it only forces the compilation of the contained module - - this could contain configurations for the object code file path etc. in the future --} +-- | Task logic definition, most modes expect task mode support, see RemoteExecutionSupport. data TaskSpec + -- | build the given string as module remotely (restrictions apply) = SourceCodeSpec String + -- | run this binary as task | FullBinaryDeployment + -- | serialize the given function in the context of the given program, run both as task (restrictions apply) | SerializedThunk (TaskInput -> TaskResult) + -- | only transport some of the generated object code and relink remotely (restrictions apply) - the function here is ignored, it only forces the compilation of the contained module | ObjectCodeModuleDeployment (TaskInput -> TaskResult) +-- | definition of input data data DataSpec + -- | simple test data, the path is configured, amount of files can be limited = SimpleDataSpec Int + -- | use given HDFS as starting directory, descend a number of directories from there and take all files starting with the filter prefix (if any given) | HdfsDataSpec HdfsPath Int (Maybe String) +-- | what to do with the result data ResultSpec + -- | process all results with the given method = CollectOnMaster ([TaskResult] -> IO ()) + -- | store the results in HDFS, in the given directory(1), with the given suffix (2), based on the input path. | StoreInHdfs String String + -- | do nothing, for testing purposes only | Discard +-- | Run a computation. runMaster :: MasterOptions -> IO () runMaster (MasterOptions masterHost masterPort taskSpec dataSpec resultSpec) = do taskDef <- buildTaskDef taskSpec diff --git a/src/Control/Distributed/Task/Distribution/TaskDistribution.hs b/src/Control/Distributed/Task/Distribution/TaskDistribution.hs --- a/src/Control/Distributed/Task/Distribution/TaskDistribution.hs +++ b/src/Control/Distributed/Task/Distribution/TaskDistribution.hs @@ -1,3 +1,6 @@ +{-| + Contains all node communication (using Cloud Haskell). This includes distribution logic. +-} module Control.Distributed.Task.Distribution.TaskDistribution ( startSlaveNode, executeDistributed, @@ -88,7 +91,7 @@ slaveTaskClosure :: TaskTransport -> S.Closure (Process ()) slaveTaskClosure = - -- $(mkClosure 'slaveTask) + -- \$(mkClosure 'slaveTask) -- ======> ((S.closure (slaveTask__static @@ -121,6 +124,9 @@ type NodeConfig = (String, Int) +{-| +Start a slave listening on given hostname, port. +-} startSlaveNode :: NodeConfig -> IO () startSlaveNode (host, port) = do initDefaultLogging (show port) @@ -128,6 +134,9 @@ putStrLn "initializing slave" startSlave backend +{-| +Run a calculation on all accessible slaves. This is a low-level method, look at the RunComputaiton module for a nicer interface. +-} executeDistributed :: NodeConfig -> TaskDef -> [DataDef] -> ResultDef -> ([TaskResult] -> IO ())-> IO () executeDistributed (host, port) taskDef dataDefs resultDef resultProcessor = do backend <- initializeBackend host (show port) rtable @@ -325,10 +334,16 @@ handlePrepareSlave hash content = liftIO (RemoteStore.put hash content) >> return PreparationFinished -- simple tasks +{-| +List all accessible slaves. +-} showSlaveNodes :: NodeConfig -> IO () showSlaveNodes config = withSlaveNodes config ( \slaveNodes -> putStrLn ("Slave nodes: " ++ show slaveNodes)) +{-| +List all accessible slaves that have at least a single block of the specified path stored physically. +-} showSlaveNodesWithData :: NodeConfig -> String -> IO () showSlaveNodesWithData slaveConfig hdfsFilePath = withSlaveNodes slaveConfig ( \slaveNodes -> do @@ -340,6 +355,9 @@ backend <- initializeBackend host (show port) initRemoteTable startMaster backend (\slaveNodes -> liftIO (action slaveNodes)) +{-| +Convenience method to stop all accessible slaves remotely. +-} shutdownSlaveNodes :: NodeConfig -> IO () shutdownSlaveNodes (host, port) = do backend <- initializeBackend host (show port) rtable diff --git a/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs b/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs --- a/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs +++ b/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs @@ -17,9 +17,11 @@ import Control.Distributed.Task.Util.Configuration import Control.Distributed.Task.Util.Logging +-- | Combines all defined task mode hooks. withRemoteExecutionSupport :: (TaskInput -> TaskResult) -> IO () -> IO () withRemoteExecutionSupport fn = withSerializedThunkRemoteExecutionSupport . withFullBinaryRemoteExecutionSupport fn +-- | Provides support for fullbinary task mode. withFullBinaryRemoteExecutionSupport :: (TaskInput -> TaskResult) -> IO () -> IO () withFullBinaryRemoteExecutionSupport fn mainAction = do args <- getArgs @@ -32,6 +34,7 @@ else mainAction _ -> mainAction +-- | Provides support for serialized thunk task mode. withSerializedThunkRemoteExecutionSupport :: IO () -> IO () withSerializedThunkRemoteExecutionSupport mainAction = do args <- getArgs diff --git a/src/Control/Distributed/Task/Types/TaskTypes.hs b/src/Control/Distributed/Task/Types/TaskTypes.hs --- a/src/Control/Distributed/Task/Types/TaskTypes.hs +++ b/src/Control/Distributed/Task/Types/TaskTypes.hs @@ -2,6 +2,7 @@ import qualified Data.ByteString.Lazy as BLC +-- | This is a small common ground for some assumptions: input is a list of sth., deserialization is the responsibility of the task itself. type TaskInput = [BLC.ByteString] -type TaskResult = [BLC.ByteString] +type TaskResult = TaskInput type Task = TaskInput -> TaskResult diff --git a/src/Control/Distributed/Task/Util/Logging.hs b/src/Control/Distributed/Task/Util/Logging.hs --- a/src/Control/Distributed/Task/Util/Logging.hs +++ b/src/Control/Distributed/Task/Util/Logging.hs @@ -20,6 +20,7 @@ simpleLog :: (String -> String -> IO ()) -> String -> IO () simpleLog levelLogger = levelLogger L.rootLoggerName +-- | Sets up hslogger. initLogging :: L.Priority -> L.Priority -> FilePath -> IO () initLogging stdoutLogLevel fileLogLevel logfile = do L.updateGlobalLogger L.rootLoggerName (L.removeHandler) diff --git a/task-distribution.cabal b/task-distribution.cabal --- a/task-distribution.cabal +++ b/task-distribution.cabal @@ -1,7 +1,11 @@ name: task-distribution -version: 0.1.0.2 -synopsis: Initial project template from stack -description: Please see README.md + +version: 0.1.0.3 +synopsis: Distributed processing of changing tasks +description: A framework for distributing tasks running on HDFS data using Cloud Haskell. + The goal is speedup through distribution on clusters using regular hardware. + This framework provides different, simple workarounds to transport new code to other cluster nodes. + See project home / README.md for more information. homepage: http://github.com/michaxm/task-distribution#readme license: BSD3 license-file: LICENSE