task-distribution 0.1.0.2 → 0.1.0.3

raw patch · 7 files changed

+55/−10 lines, 7 files

Files

• src/Control/Distributed/Task/Distribution/LogConfiguration.hs +1/−0
• src/Control/Distributed/Task/Distribution/RunComputation.hs +22/−5
• src/Control/Distributed/Task/Distribution/TaskDistribution.hs +19/−1
• src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs +3/−0
• src/Control/Distributed/Task/Types/TaskTypes.hs +2/−1
• src/Control/Distributed/Task/Util/Logging.hs +1/−0
• task-distribution.cabal +7/−3

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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)

@@ -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