task-distribution (empty) → 0.1.0.0

raw patch · 42 files changed

+2254/−0 lines, 42 filesdep +asyncdep +basedep +binarysetup-changed

Dependencies added: async, base, binary, bytestring, containers, directory, distributed-process, distributed-process-simplelocalnet, distributed-static, filepath, hadoop-rpc, hashable, hint, hslogger, hspec, json, packman, process, rank1dynamic, split, strings, task-distribution, temporary, text, time, transformers, vector, zlib

Files

• LICENSE +30/−0
• Setup.hs +2/−0
• app/DemoTask.hs +37/−0
• app/FullBinaryExamples.hs +12/−0
• app/Main.hs +91/−0
• app/RemoteExecutable.hs +43/−0
• app/RunDemoTask.hs +29/−0
• app/Slave.hs +10/−0
• app/TestVisitCalculation.hs +18/−0
• app/VisitCalculation.hs +47/−0
• object-code-app/RemoteExecutable.hs +6/−0
• object-code-app/RemoteExecutor.hs +32/−0
• src/Control/Distributed/Task/DataAccess/DataSource.hs +10/−0
• src/Control/Distributed/Task/DataAccess/HdfsDataSource.hs +78/−0
• src/Control/Distributed/Task/DataAccess/HdfsListing.hs +59/−0
• src/Control/Distributed/Task/DataAccess/SimpleDataSource.hs +15/−0
• src/Control/Distributed/Task/Distribution/DataLocality.hs +89/−0
• src/Control/Distributed/Task/Distribution/LogConfiguration.hs +15/−0
• src/Control/Distributed/Task/Distribution/RunComputation.hs +109/−0
• src/Control/Distributed/Task/Distribution/TaskDistribution.hs +351/−0
• src/Control/Distributed/Task/Distribution/TaskTransport.hs +55/−0
• src/Control/Distributed/Task/TaskSpawning/BinaryStorage.hs +48/−0
• src/Control/Distributed/Task/TaskSpawning/DeployFullBinary.hs +89/−0
• src/Control/Distributed/Task/TaskSpawning/DeployObjectCodeRelinked.hs +76/−0
• src/Control/Distributed/Task/TaskSpawning/DeploySerializedThunk.hs +29/−0
• src/Control/Distributed/Task/TaskSpawning/ExecutionUtil.hs +111/−0
• src/Control/Distributed/Task/TaskSpawning/FunctionSerialization.hs +20/−0
• src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs +52/−0
• src/Control/Distributed/Task/TaskSpawning/SourceCodeExecution.hs +91/−0
• src/Control/Distributed/Task/TaskSpawning/TaskDefinition.hs +72/−0
• src/Control/Distributed/Task/TaskSpawning/TaskDescription.hs +15/−0
• src/Control/Distributed/Task/TaskSpawning/TaskSpawning.hs +87/−0
• src/Control/Distributed/Task/TaskSpawning/TaskSpawningTypes.hs +54/−0
• src/Control/Distributed/Task/Types/HdfsConfigTypes.hs +5/−0
• src/Control/Distributed/Task/Types/TaskTypes.hs +7/−0
• src/Control/Distributed/Task/Util/Configuration.hs +65/−0
• src/Control/Distributed/Task/Util/ErrorHandling.hs +17/−0
• src/Control/Distributed/Task/Util/FileUtil.hs +15/−0
• src/Control/Distributed/Task/Util/Logging.hs +35/−0
• src/Control/Distributed/Task/Util/SerializationUtil.hs +19/−0
• task-distribution.cabal +188/−0
• test/NodeMatchingTests.hs +21/−0

@@ -0,0 +1,30 @@+Copyright Axel Mannhardt (c) 2016++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Author name here nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,37 @@+{-# LANGUAGE BangPatterns #-}+module DemoTask where++import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BLC+import Data.List (foldl')+import Data.Maybe++calculateRatio :: [BL.ByteString] -> [BL.ByteString]+calculateRatio = (:[]) . BLC.pack . show . divideRatio . foldl' (flip countHits) (0, 0) . parseEntries+  where+    divideRatio :: (Int, Int) -> Double+    divideRatio (hits, count) = fromIntegral hits / (fromIntegral count)+    countHits :: Entry -> (Int, Int) -> (Int, Int)+    -- especially hits are important to evaluate strict here to avoid building up of costly to store thunks, containing entries (count is affected to a lesser degree)+    countHits e (!hits, !count) = (hits + if isHit e then 1 else 0, count+1)++isHit :: Entry -> Bool+isHit (_, _, vs) = maybe False notEmpty $ lookup (BLC.pack "searchCrit") vs+  where+    notEmpty :: BL.ByteString -> Bool+    notEmpty = (>0) . (read :: String -> Int) . BLC.unpack++parseEntries :: [BL.ByteString] -> [Entry]+parseEntries = catMaybes . map parseEntry++type Entry = (BL.ByteString, BL.ByteString, [(BL.ByteString, BL.ByteString)])++--nlrkwldbji|3101692061|xlerb:uubqpnce|searchCrit:0|longerAttr:nspixnozidyqweu|other:lju|otherRandomNam:tndhdno|otherRandomName2:dgkkj|otherRandomName3:yrdwgww|last:xvofbj+parseEntry :: BL.ByteString -> Maybe Entry+parseEntry str =+  let cells = BLC.split '|' str+  in if length cells < 2+     then Nothing+     else Just (cells !! 0, cells !! 1, catMaybes $ map parseValue $ drop 2 cells)+  where+    parseValue val = let kv = BLC.split ':' val in if length kv /= 2 then Nothing else Just (kv !! 0, kv !! 1)

@@ -0,0 +1,12 @@+module FullBinaryExamples where++import qualified Data.ByteString.Char8 as BC+import qualified Data.ByteString.Lazy.Char8 as BLC++import Control.Distributed.Task.Types.TaskTypes++appendDemo :: String -> TaskInput -> TaskResult+appendDemo  demoArg = map (\str -> str `BLC.append` (BLC.pack $ " "++demoArg))++filterDemo :: String -> TaskInput -> TaskResult+filterDemo demoArg = filter (((BC.pack demoArg) `BC.isInfixOf`) . BC.concat . BLC.toChunks )

@@ -0,0 +1,91 @@+module Main where++import Data.List (intersperse)+import Data.List.Split (splitOn)+import System.Environment (getArgs)++import Control.Distributed.Task.Distribution.LogConfiguration (initDefaultLogging)+import Control.Distributed.Task.Distribution.RunComputation+import Control.Distributed.Task.Distribution.TaskDistribution (startSlaveNode, showSlaveNodes, showSlaveNodesWithData, shutdownSlaveNodes)+import Control.Distributed.Task.TaskSpawning.RemoteExecutionSupport+import Control.Distributed.Task.Types.HdfsConfigTypes+import Control.Distributed.Task.Types.TaskTypes++import FullBinaryExamples+import RemoteExecutable+import DemoTask++main :: IO ()+main = withRemoteExecutionSupport calculateRatio $ do+  initDefaultLogging ""+  args <- getArgs+  case args of+   ("master" : masterArgs) -> runMaster (parseMasterOpts masterArgs)+   ["slave", slaveHost, slavePort] -> startSlaveNode (slaveHost, (read slavePort))+   ["showslaves"] -> showSlaveNodes localConfig+   ["slaveswithhdfsdata", host, port, hdfsFilePath] -> showSlaveNodesWithData (host, read port) hdfsFilePath+   ["shutdown"] -> shutdownSlaveNodes localConfig+   _ -> userSyntaxError "unknown mode"++-- note: assumes no nodes with that configuration, should be read as parameters+localConfig :: HdfsConfig+localConfig = ("localhost", 44440)++userSyntaxError :: String -> undefined+userSyntaxError reason = error $ usageInfo ++ reason ++ "\n"++usageInfo :: String+usageInfo = "Syntax: master"+            ++ " <host>"+            ++ " <port>"+            ++ " <module:<module path>|fullbinary|serializethunkdemo:<demo function>:<demo arg>|objectcodedemo>"+            ++ " <simpledata:numDBs|hdfs:<file path>[:<subdir depth>[:<filename filter prefix>]]"+            ++ " <collectonmaster|discard|storeinhdfs:<outputprefix>[:<outputsuffix>]>\n"+            ++ "| slave <slave host> <slave port>\n"+            ++ "| showslaves\n"+            ++ "| slaveswithhdfsdata <thrift host> <thrift port> <hdfs file path>\n"+            ++ "| shutdown\n"+            ++ "\n"+            ++ "demo functions (with demo arg description): append:<suffix> | filter:<infixfilter> | visitcalc:unused \n"++parseMasterOpts :: [String] -> MasterOptions+parseMasterOpts args =+  case args of+   [masterHost, port, taskSpec, dataSpec, resultSpec] -> MasterOptions masterHost (read port) (parseTaskSpec taskSpec) (parseDataSpec dataSpec) (parseResultSpec resultSpec)+   _ -> userSyntaxError "wrong number of master options"+  where+    parseTaskSpec :: String -> TaskSpec+    parseTaskSpec taskArgs =+      case splitOn ":" taskArgs of+       ["module", modulePath] -> SourceCodeSpec modulePath+       ["fullbinary"] -> FullBinaryDeployment+       ["serializethunkdemo", demoFunction, demoArg] -> mkSerializeThunkDemoArgs demoFunction demoArg+       ["objectcodedemo"] -> ObjectCodeModuleDeployment remoteExecutable+       _ -> userSyntaxError $ "unknown task specification: " ++ taskArgs+       where+        mkSerializeThunkDemoArgs :: String -> String -> TaskSpec+        mkSerializeThunkDemoArgs "append" demoArg = SerializedThunk (appendDemo demoArg)+        mkSerializeThunkDemoArgs "filter" demoArg = SerializedThunk (filterDemo demoArg)+        mkSerializeThunkDemoArgs "visitcalc" _ = SerializedThunk calculateRatio+        mkSerializeThunkDemoArgs d a = userSyntaxError $ "unknown demo: " ++ d ++ ":" ++ a+    parseDataSpec :: String -> DataSpec+    parseDataSpec inputArgs =+      case splitOn ":" inputArgs of+       ["simpledata", numDBs] -> SimpleDataSpec $ read numDBs+       ("hdfs": hdfsPath: rest) -> HdfsDataSpec hdfsPath depth filter'+         where depth = if length rest > 0 then read (rest !! 0) else 0; filter' = if length rest > 1 then Just (rest !! 1) else Nothing+       _ -> userSyntaxError $ "unknown data specification: " ++ inputArgs+    parseResultSpec resultArgs =+      case splitOn ":" resultArgs of+       ["collectonmaster"] -> CollectOnMaster resultProcessor+       ("storeinhdfs":outputPrefix:rest) -> StoreInHdfs outputPrefix $ if null rest then "" else head rest+       ["discard"] -> Discard+       _ -> userSyntaxError $ "unknown result specification: " ++ resultArgs++resultProcessor :: [TaskResult] -> IO ()+resultProcessor res = do+ putStrLn $ joinStrings "\n" $ map show $ concat res+ putStrLn $ "got " ++ (show $ length $ concat res) ++ " results in total"++joinStrings :: String -> [String] -> String+joinStrings separator = concat . intersperse separator

@@ -0,0 +1,43 @@+{-# LANGUAGE BangPatterns #-}+module RemoteExecutable (remoteExecutable) where++import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BLC+import Data.List (foldl')+import Data.Maybe++import Control.Distributed.Task.Types.TaskTypes++remoteExecutable :: Task+remoteExecutable = calculateRatio+--remoteExecutable = map $ flip BLC.append $ BLC.pack " appended"++calculateRatio :: [BL.ByteString] -> [BL.ByteString]+calculateRatio = (:[]) . BLC.pack . show . divideRatio . foldl' (flip countHits) (0, 0) . parseEntries+  where+    divideRatio :: (Int, Int) -> Double+    divideRatio (hits, count) = fromIntegral hits / (fromIntegral count)+    countHits :: Entry -> (Int, Int) -> (Int, Int)+    -- especially hits are important to evaluate strict here to avoid building up of costly to store thunks, containing entries (count is affected to a lesser degree)+    countHits e (!hits, !count) = (hits + if isHit e then 1 else 0, count+1)++isHit :: Entry -> Bool+isHit (_, _, vs) = maybe False notEmpty $ lookup (BLC.pack "searchCrit") vs+  where+    notEmpty :: BL.ByteString -> Bool+    notEmpty = (>0) . (read :: String -> Int) . BLC.unpack++parseEntries :: [BL.ByteString] -> [Entry]+parseEntries = catMaybes . map parseEntry++type Entry = (BL.ByteString, BL.ByteString, [(BL.ByteString, BL.ByteString)])++--nlrkwldbji|3101692061|xlerb:uubqpnce|searchCrit:0|longerAttr:nspixnozidyqweu|other:lju|otherRandomNam:tndhdno|otherRandomName2:dgkkj|otherRandomName3:yrdwgww|last:xvofbj+parseEntry :: BL.ByteString -> Maybe Entry+parseEntry str =+  let cells = BLC.split '|' str+  in if length cells < 2+     then Nothing+     else Just (cells !! 0, cells !! 1, catMaybes $ map parseValue $ drop 2 cells)+  where+    parseValue val = let kv = BLC.split ':' val in if length kv /= 2 then Nothing else Just (kv !! 0, kv !! 1)

@@ -0,0 +1,29 @@+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BLC+import Control.Concurrent.Async+import Data.List+import qualified Codec.Compression.GZip as GZip+import System.Environment (getArgs)++import DemoTask++main :: IO ()+main = do+  args <- getArgs+  case args of+   [] -> error $ "usage: <filename|[filename]>"+   [filename] -> runSequential filename+   filenames -> runParallel filenames++runParallel :: [FilePath] -> IO ()+runParallel filenames = do+  results <- mapM (async . runSequential) filenames+  mapM_ wait results++runSequential :: FilePath -> IO ()+runSequential filename = BL.readFile filename+                         >>= return . calculateRatio . BLC.lines . unzipIfNecessary+                         >>= print+  where+    isZipped = ".gz" `isSuffixOf` filename+    unzipIfNecessary = if isZipped then GZip.decompress else id

@@ -0,0 +1,10 @@+import System.Environment (getArgs)++import Control.Distributed.Task.Distribution.TaskDistribution (startSlaveNode)++main :: IO ()+main = do+  args <- getArgs+  case args of+   [slaveHost, slavePort] -> startSlaveNode (slaveHost, (read slavePort))+   _ -> error "syntax: <host> <port>"

@@ -0,0 +1,18 @@+import Data.List (intersperse)+import System.Environment (getArgs)++import qualified Data.ByteString.Lazy.Char8 as BLC+import VisitCalculation++main :: IO ()+main = do+  args <- getArgs+  case args of+   [filePath] -> calculateVisits' filePath+   _ -> error "missing input file"++calculateVisits' :: String -> IO ()+calculateVisits' filePath = do+  contents <- BLC.readFile filePath+  visits <- return $ calculateVisits $ BLC.lines contents+  BLC.writeFile "tmp/out" $ BLC.concat $ intersperse (BLC.pack "\n") visits

@@ -0,0 +1,47 @@+module VisitCalculation (calculateVisits) where++import qualified Data.ByteString.Lazy.Char8 as BLC+import Data.Foldable (foldl')+import Data.List.Split (splitOn)+import qualified Data.Map as Map+import Data.Map (Map)+import Text.JSON++type ParsedEvent = (VisitId, RequestTime, Event)+type VisitData = (VisitId, (RequestTime, [Event])) -- events sorted in reverse order+type VisitId = String+type RequestTime = String+type Event = String++calculateVisits :: [BLC.ByteString] -> [BLC.ByteString]+calculateVisits = map BLC.pack . map formatVisit . partitionByVisitorId . parseEvents . map BLC.unpack -- TODO probably highly imperformant++parseEvents :: [String] -> [ParsedEvent]+parseEvents = foldl' (\p -> maybe p (:p) . parseEvent) []+  where+    parseEvent :: String -> Maybe ParsedEvent+    parseEvent e =+      let first5 = take 5 (splitOn "|" e) in+       if length first5 < 5+       then Nothing+       else Just (first5 !! 4, first5 !! 0, e)++partitionByVisitorId :: [ParsedEvent] -> [VisitData]+partitionByVisitorId = Map.assocs . foldr insertEvent Map.empty+  where+    insertEvent :: ParsedEvent -> Map VisitId (RequestTime, [Event]) -> Map VisitId (RequestTime, [Event])+    insertEvent (k, t, e) = Map.insertWith mergeData k (t, [e])+      where+        mergeData :: (RequestTime, [Event]) -> (RequestTime, [Event]) -> (RequestTime, [Event])+        mergeData (newT, newEs) (oldT, oldEs) = (min newT oldT, reverse newEs ++ oldEs)++formatVisit :: VisitData -> String+formatVisit (i, (t, es)) = encode $ toJSObject [visitId, events]+  where+    events, visitId :: (String, JSValue)+    events = ("events", JSArray $ map (JSString . toJSString) es)+    visitId = ("visitId", JSObject $ toJSObject [visitorId, lastRequestTime])+      where+        visitorId, lastRequestTime :: (String, JSValue)+        visitorId = ("visitorId", JSObject $ toJSObject [("value", JSString $ toJSString i)])+        lastRequestTime = ("lastRequestTime", JSString $ toJSString t)

@@ -0,0 +1,6 @@+module RemoteExecutable (remoteExecutable) where++import Control.Distributed.Task.Types.TaskTypes++remoteExecutable :: Task+remoteExecutable = error "example signature for remote executable, not a real implementation"

@@ -0,0 +1,32 @@+import qualified Data.ByteString.Char8 as BLC+import System.Environment (getArgs, getExecutablePath)+import qualified System.Log.Logger as L++import Control.Distributed.Task.TaskSpawning.DeployFullBinary+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging++import RemoteExecutable++main :: IO ()+main = do+  args <- getArgs+  case args of+   [ioHandling] -> executeFunction (unpackIOHandling ioHandling)+   _ -> error "Syntax for relinked task: <ioHandling>\n"++executeFunction :: IOHandling -> IO ()+executeFunction ioHandling = do+  initTaskLogging+  fullBinaryExecution ioHandling executeF+  where+    executeF :: Task+    executeF = remoteExecutable++initTaskLogging :: IO ()+initTaskLogging = do+  conf <- getConfiguration+  initLogging L.ERROR L.INFO (_taskLogFile conf)+  self <- getExecutablePath+  logInfo $ "started task execution for: "++self

@@ -0,0 +1,10 @@+module Control.Distributed.Task.DataAccess.DataSource (loadData) where++import qualified Control.Distributed.Task.DataAccess.HdfsDataSource as HDS+import qualified Control.Distributed.Task.DataAccess.SimpleDataSource as SDS+import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.Types.TaskTypes++loadData :: DataDef -> IO (TaskInput)+loadData (PseudoDB p) = SDS.loadEntries p+loadData (HdfsData p) = HDS.supplyPathWithConfig p >>= HDS.loadEntries

@@ -0,0 +1,78 @@+module Control.Distributed.Task.DataAccess.HdfsDataSource (supplyPathWithConfig, loadEntries, copyToLocal) where++import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as BC+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BLC+import qualified Codec.Compression.GZip as GZip+import qualified Data.Hadoop.Configuration as HDFS+import qualified Data.Hadoop.Types as HDFS+import Data.List (isSuffixOf)+import qualified Data.Text as T+import Network.Hadoop.Hdfs+import Network.Hadoop.Read+import System.Directory (doesFileExist)+import System.IO (IOMode(..), withBinaryFile)++import Control.Distributed.Task.Types.TaskTypes (TaskInput)+import Control.Distributed.Task.Types.HdfsConfigTypes+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.ErrorHandling+import Control.Distributed.Task.Util.Logging++supplyPathWithConfig :: HdfsPath -> IO HdfsLocation+supplyPathWithConfig p = do+  config <- getConfiguration+  return (_hdfsConfig config, p)++loadEntries :: HdfsLocation -> IO TaskInput+loadEntries hdfsLocation = do+  logInfo $ "loading: " ++ targetDescription+  withErrorPrefix ("Error accessing "++ targetDescription) doLoad >>= return . BLC.lines . unzipIfNecessary+  where+    targetDescription = show hdfsLocation+    doLoad = readHdfsFile hdfsLocation+    unzipIfNecessary = if ".gz" `isSuffixOf` (snd hdfsLocation) then GZip.decompress else id++readHdfsFile :: HdfsLocation -> IO BL.ByteString+readHdfsFile (hdfsConfig, path) = do+  logInfo $ "now loading" ++ show (hdfsConfig, path)+  config <- buildConfig hdfsConfig+  chunks <- readAllChunks readerAction config path+  logInfo "loading chunks finished"+  return $ BLC.fromChunks $ reverse chunks+  where+    readerAction :: [B.ByteString] -> BC.ByteString -> IO [B.ByteString]+    readerAction collected nextChunk = return $ nextChunk:collected++buildConfig :: HdfsConfig -> IO HDFS.HadoopConfig+buildConfig (host, port) = do+  user <- HDFS.getHadoopUser+  return $ HDFS.HadoopConfig user [(HDFS.Endpoint (T.pack host) port)] Nothing++readAllChunks :: ([B.ByteString] -> BC.ByteString -> IO [B.ByteString]) -> HDFS.HadoopConfig -> String -> IO [B.ByteString]+readAllChunks action config path =  do+  readHandle_ <- runHdfs' config $ openRead $ BC.pack path+  case readHandle_ of+   (Just readHandle) -> hdfsFoldM action [] readHandle+   Nothing -> error "no read handle"++copyToLocal :: HdfsLocation -> FilePath -> IO ()+copyToLocal (hdfsConfig, path) destFile = do+  logInfo $ "now copying from " ++ show hdfsConfig+  config <- buildConfig hdfsConfig+  copyHdfsFileToLocal destFile (Just config) path++copyHdfsFileToLocal :: FilePath -> Maybe HDFS.HadoopConfig -> String -> IO ()+copyHdfsFileToLocal destFile config path = do+  destFileExists <- doesFileExist destFile+  if destFileExists+    then error $ destFile++" exists"+    else withBinaryFile destFile WriteMode $ \h -> withHdfsReader (BC.hPut h) config path++withHdfsReader :: (BC.ByteString -> IO ()) -> Maybe HDFS.HadoopConfig -> String -> IO ()+withHdfsReader action config path =  do+  readHandle_ <- maybe runHdfs runHdfs' config $ openRead $ BC.pack path+  case readHandle_ of+   (Just readHandle) -> hdfsMapM_ action readHandle+   Nothing -> error "no read handle"

@@ -0,0 +1,59 @@+module Control.Distributed.Task.DataAccess.HdfsListing (+  listFiles, listBlockDistribution+  ) where++import Control.Exception (catch, SomeException)+import qualified Data.ByteString.Char8 as BC+import qualified Data.Hadoop.Configuration as HRPC+import qualified Data.Hadoop.Types as HRPC+import qualified Network.Hadoop.Hdfs as HRPC+import qualified Data.Text as T+import qualified Data.Vector as V++import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging++listFiles :: String -> IO [String]+listFiles path = do+  logInfo $ "listing files at: "++path+  files <- runHdfsWithConfig $ HRPC.getListing' $ BC.pack path+  return $ V.toList $ V.map convert $ files+  where+    convert :: HRPC.FileStatus -> String+    convert = BC.unpack . HRPC.fsPath++listBlockDistribution :: String -> IO [(String, Int)]+listBlockDistribution path = do+  logInfo $ "listing file blocks for: "++path+  locs <- runHdfsWithConfig $ HRPC.getBlockLocations $ BC.pack path+  case V.length locs of+   0 -> error $ "no such file: "++path+   1 -> return $ convert $ HRPC.fsLocations $ locs V.! 0+   _ -> error $ "is not a regular file: "++path+  where+    convert :: HRPC.FsLocations -> [(String, Int)]+    convert HRPC.NotRequested = error "block locations not requested?"+    convert (HRPC.FsLocations ls) =+      let blockLocations = V.toList $ V.map V.toList $ V.map snd ls :: [[HRPC.BlockLocation]]+          hostnames = map (T.unpack . fst) $ concat blockLocations :: [String]+      in groupCount hostnames++groupCount :: (Eq key) => [key] -> [(key, Int)]+groupCount = foldr groupCount' []+  where+    groupCount' :: (Eq key) => key -> [(key, Int)] -> [(key, Int)]+    groupCount' key [] = [(key, 1)]+    groupCount' key (next:collected) = if (fst next == key) then (fst next, snd next +1):collected else next:(groupCount' key collected)++runHdfsWithConfig :: HRPC.Hdfs a -> IO a+runHdfsWithConfig action = do+  hdfsConfig <- mkHdfsConfig `catch` warnAndUseDefault+  (maybe HRPC.runHdfs HRPC.runHdfs' hdfsConfig) action+  where+    warnAndUseDefault :: SomeException -> IO (Maybe HRPC.HadoopConfig)+    warnAndUseDefault e = logWarn ("failed to load config, trying system defaults: "++show e) >> return Nothing+    mkHdfsConfig :: IO (Maybe HRPC.HadoopConfig)+    mkHdfsConfig = do+      user <- HRPC.getHadoopUser+      (host, port) <- getConfiguration >>= return . _hdfsConfig+      return $ Just $ HRPC.HadoopConfig user [HRPC.Endpoint (T.pack host) port] Nothing

@@ -0,0 +1,15 @@+module Control.Distributed.Task.DataAccess.SimpleDataSource where++import qualified Data.ByteString.Lazy.Char8 as BLC++import Control.Distributed.Task.Types.TaskTypes (TaskInput)+import Control.Distributed.Task.Util.Logging+import Control.Distributed.Task.Util.Configuration++loadEntries :: FilePath -> IO TaskInput+loadEntries filePath = do+  pseudoDBPath <- getConfiguration >>= return . _pseudoDBPath+  let path = pseudoDBPath++"/"++filePath+    in do+    logInfo $ "accessing pseudo db at: "++path+    BLC.readFile path >>= return . BLC.lines

@@ -0,0 +1,89 @@+module Control.Distributed.Task.Distribution.DataLocality (+  findNodesWithData,+  -- visible for testing:+  nodeMatcher+  ) where++import Control.Distributed.Process (NodeId)+import Data.List (sortBy)+import Data.List.Split (splitOn)+import Data.Ord (comparing)+import Prelude hiding (log)++import qualified Control.Distributed.Task.DataAccess.HdfsListing as HDFS+import Control.Distributed.Task.Util.Logging++{-+ Filters the given nodes to those with any of the file blocks, ordered by the number of file blocks (not regarding individual file block length).+-}+findNodesWithData :: String -> [NodeId] -> IO [NodeId]+findNodesWithData hdfsFilePath nodes = do+  logInfo ("All nodes for : " ++ hdfsFilePath ++": "++ (show nodes))+  hostsWithData <- HDFS.listBlockDistribution hdfsFilePath+  (if null hostsWithData then logError else logInfo) ("Hdfs hosts with data: " ++ (show hostsWithData))+  hosts <- readHostNames+  logDebug (show hosts)+  mergedNodeIds <- return $ map fst $ reverse $ sortOn snd $ merge (matcher hosts) merger nodes hostsWithData+  logInfo ("Merged nodes: " ++ (show mergedNodeIds))+  return mergedNodeIds+    where+      matcher hosts node (hdfsName, _) = nodeMatcher hosts (show node) hdfsName -- HACK uses show to access nodeId data+      merger :: NodeId -> HdfsHit -> (NodeId, Int)+      merger nid (_, n) = (nid, n)++type HdfsHit = (String, BlockCount)+type BlockCount = Int++readHostNames :: IO [(String, String)]+readHostNames = do+  allHosts <- readFile "/etc/hosts" >>= return . parseHostFile+  extraHosts <- readFile "etc/hostconfig" >>= return . parseHostFile+  return $ allHosts ++ extraHosts+    where+      parseHostFile :: String -> [(String, String)]+      parseHostFile = concat . map parseHosts . filter comments . lines+        where+          comments [] = False+          comments ('#':_) = False+          comments _ = True+          parseHosts :: String -> [(String, String)]+          parseHosts = parseHosts' . splitOn " " . collapseWhites . map replaceTabs+            where+              replaceTabs :: Char -> Char+              replaceTabs '\t' = ' '+              replaceTabs c = c+              collapseWhites :: String -> String+              collapseWhites (' ':' ':rest) = ' ':(collapseWhites rest)+              collapseWhites (c:rest) = c:(collapseWhites rest)+              collapseWhites r = r+              parseHosts' :: [String] -> [(String, String)]+              parseHosts' es = if length es < 2 then [] else map (\v -> (head es,v)) (tail es)++nodeMatcher ::[(String, String)] -> String -> String -> Bool+nodeMatcher hosts node hdfsName = (extractHdfsHost hdfsName) == (extractNodeIdHost node)+  where+    -- HACK: extracts the host name from "nid://localhost:44441:0"+    extractNodeIdHost = lookupHostname . dropWhile (=='/') . head . drop 1 . splitOn ":"+    -- HACK: extracts the host name from "127.0.0.1:50010"+    extractHdfsHost = lookupHostname . head . splitOn ":"+    lookupHostname :: String -> String+    lookupHostname k = maybe k id (lookup k hosts)++{-+ Merges left with right: for each left, take the first match in right, ignoring other possible matches.+-}+merge :: (a -> b -> Bool) -> (a -> b -> c) -> [a] -> [b] -> [c]+merge matcher merger = merge'+  where+    merge' _ [] = []+    merge' [] _ = []+    merge' (a:as) bs = maybe restMerge (:restMerge) (merge'' bs)+      where+        restMerge = merge' as bs+        merge'' [] = Nothing+        merge'' (b:bs') = if matcher a b then Just (merger a b) else merge'' bs'++{- since 4.8 in Data.List -}+sortOn :: Ord b => (a -> b) -> [a] -> [a] +sortOn f =+  map snd . sortBy (comparing fst) . map (\x -> let y = f x in y `seq` (y, x))

@@ -0,0 +1,15 @@+module Control.Distributed.Task.Distribution.LogConfiguration (+  initLogging,+  initDefaultLogging -- TODO take loglevels from configuration+  ) where++import qualified System.Log.Logger as L++import Control.Distributed.Task.Util.Logging (initLogging)++initDefaultLogging :: String -> IO ()+initDefaultLogging suffix = do+--  progName <- getExecutablePath+  initLogging L.WARNING L.INFO logfile --TODO logging relative to $CLUSTER_COMPUTING_HOME+    where+      logfile = ("log/task-distribution" ++ (if null suffix then "" else "-"++suffix) ++".log")

@@ -0,0 +1,109 @@+module Control.Distributed.Task.Distribution.RunComputation (+  MasterOptions(..),+  TaskSpec(..),+  DataSpec(..),+  ResultSpec(..),+  runMaster) where++import Data.List (isPrefixOf, sort)+import Data.List.Split (splitOn)+import System.Directory (getDirectoryContents) -- being renamed to listDirectory+import System.Environment (getExecutablePath)++import qualified Control.Distributed.Task.DataAccess.HdfsListing as HDFS+import Control.Distributed.Task.Distribution.TaskDistribution+import Control.Distributed.Task.TaskSpawning.TaskSpawning (fullBinarySerializationOnMaster, serializedThunkSerializationOnMaster, objectCodeSerializationOnMaster)+import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.Types.HdfsConfigTypes+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging++data MasterOptions = MasterOptions {+  _host :: String,+  _port :: Int,+  _taskSpec :: TaskSpec,+  _dataSpecs :: DataSpec,+  _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+-}+data TaskSpec+ = SourceCodeSpec String+ | FullBinaryDeployment+ | SerializedThunk (TaskInput -> TaskResult)+ | ObjectCodeModuleDeployment (TaskInput -> TaskResult)+data DataSpec+  = SimpleDataSpec Int+  | HdfsDataSpec HdfsPath Int (Maybe String)+data ResultSpec+  = CollectOnMaster ([TaskResult] -> IO ())+  | StoreInHdfs String String+  | Discard++runMaster :: MasterOptions -> IO ()+runMaster (MasterOptions masterHost masterPort taskSpec dataSpec resultSpec) = do+  taskDef <- buildTaskDef taskSpec+  dataDefs <- expandDataSpec dataSpec+  (resultDef, resultProcessor) <- return $ buildResultDef resultSpec+  executeDistributed (masterHost, masterPort) taskDef dataDefs resultDef resultProcessor+    where+      buildResultDef :: ResultSpec -> (ResultDef, [TaskResult] -> IO ())+      buildResultDef (CollectOnMaster resultProcessor) = (ReturnAsMessage, resultProcessor)+      buildResultDef (StoreInHdfs outputPrefix outputSuffix) = (HdfsResult outputPrefix outputSuffix True, \_ -> putStrLn "result stored in hdfs")+      buildResultDef Discard = (ReturnOnlyNumResults, \num -> putStrLn $ (show num) ++ " results discarded")++buildTaskDef :: TaskSpec -> IO TaskDef+buildTaskDef (SourceCodeSpec modulePath) = do+  moduleContent <- readFile modulePath+  return $ mkSourceCodeModule modulePath moduleContent+buildTaskDef FullBinaryDeployment =+  getExecutablePath >>= fullBinarySerializationOnMaster+buildTaskDef (SerializedThunk function) = do+  selfPath <- getExecutablePath+  serializedThunkSerializationOnMaster selfPath function+buildTaskDef (ObjectCodeModuleDeployment _) = objectCodeSerializationOnMaster++expandDataSpec :: DataSpec -> IO [DataDef]+expandDataSpec (HdfsDataSpec path depth filterPrefix) = do+  putStrLn $ "looking for files at " ++ path+  paths <- hdfsListFilesInSubdirsFiltering depth filterPrefix path+  putStrLn $ "found these input files: " ++ (show paths)+  return $ map HdfsData paths+expandDataSpec (SimpleDataSpec numTasks) = do+  config <- getConfiguration+  files <- getDirectoryContents (_pseudoDBPath config)+  return $ take numTasks $ map PseudoDB $ filter (not . ("." `isPrefixOf`)) $ sort files++mkSourceCodeModule :: String -> String -> TaskDef+mkSourceCodeModule modulePath moduleContent = SourceCodeModule (strippedModuleName modulePath) moduleContent+  where+    strippedModuleName = reverse . takeWhile (/= '/') . drop 1 . dropWhile (/= '.') . reverse++{-|+ Like hdfsListFiles, but descending into subdirectories and filtering the file names. Note that for now this is a rather a quick hack+ for special needs than a full fledged shell expansion+|-}+hdfsListFilesInSubdirsFiltering :: Int -> Maybe String -> String -> IO [String]+hdfsListFilesInSubdirsFiltering descendDepth fileNamePrefixFilter path = do+  initialFilePaths <- HDFS.listFiles path+  recursiveFiles <- recursiveDescent descendDepth path initialFilePaths+  logDebug $ "found: " ++ (show recursiveFiles)+  return $ map trimSlashes $ maybe recursiveFiles (\prefix -> filter ((prefix `isPrefixOf`) . getFileNamePart) recursiveFiles) fileNamePrefixFilter+  where+    getFileNamePart path' = let parts = splitOn "/" path' in if null parts then "" else parts !! (length parts -1)+    recursiveDescent :: Int -> String -> [String] -> IO [String]+    recursiveDescent 0 prefix paths = return (map (\p -> prefix++"/"++p) paths)+    recursiveDescent n prefix paths = do+      absolute <- return $ map (prefix++) paths :: IO [String]+      pathsWithChildren <- mapM (\p -> (HDFS.listFiles p >>= \cs -> return (p, cs))) absolute :: IO [(String, [String])]+      descended <- mapM (\(p, cs) -> if null cs then return [p] else recursiveDescent (n-1) p cs) pathsWithChildren :: IO [[String]]+      return $ concat descended+    trimSlashes :: String -> String+    trimSlashes [] = [] -- hadoop-rpc does not work on /paths//with/double//slashes+    trimSlashes ('/':'/':rest) = trimSlashes $ '/':rest+    trimSlashes (x:rest) = x:(trimSlashes rest)

@@ -0,0 +1,351 @@+module Control.Distributed.Task.Distribution.TaskDistribution (+  startSlaveNode,+  executeDistributed,+  showSlaveNodes,+  showSlaveNodesWithData,+  shutdownSlaveNodes) where++import Control.Distributed.Process (Process, ProcessId, NodeId,+                                    say, getSelfPid, spawn, send, expect, receiveWait, match, matchAny, catch, +                                    RemoteTable)+import Control.Distributed.Process.Backend.SimpleLocalnet (initializeBackend, startMaster, startSlave, terminateSlave)+import qualified Control.Distributed.Process.Serializable as PS+import qualified Control.Distributed.Static as S+import Control.Distributed.Process.Closure (staticDecode)+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Exception.Base (SomeException)+import Control.Monad (forM_)+import Control.Monad.IO.Class+import qualified Data.Binary as B (encode)+import Data.Bool (bool)+import Data.ByteString.Lazy (ByteString)+import Data.List (intersperse)+import qualified Data.Map.Strict as Map+import qualified Data.Rank1Dynamic as R1 (toDynamic)+import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime)++import Control.Distributed.Task.Distribution.DataLocality (findNodesWithData)+import Control.Distributed.Task.Distribution.LogConfiguration+import Control.Distributed.Task.Distribution.TaskTransport+import qualified Control.Distributed.Task.TaskSpawning.BinaryStorage as RemoteStore+import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.TaskSpawning.TaskDescription+import Control.Distributed.Task.TaskSpawning.TaskSpawning (processTasks, TasksExecutionResult)+import Control.Distributed.Task.TaskSpawning.TaskSpawningTypes+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging+import Control.Distributed.Task.Util.SerializationUtil++{-+ The bits in this file are arranged so that the less verbose Template Haskell version would work. That version is not used due to incompability+ issues between packman and Template Haskell though, but is retained in the commens for documentational purposes.+-}++-- BEGIN bindings for node communication+{-+ This is the final building block of the slave task execution, calling TaskSpawning.processTasks.+-}+slaveTask :: TaskTransport -> Process () -- TODO: have a node local config?+slaveTask = handleSlaveTask++querySlavePreparation :: QuerySlavePreparationRequest -> Process ()+querySlavePreparation (masterProcess, hash) = handleQuerySlavePreparation hash >>= send masterProcess++prepareSlave :: PrepareSlaveRequest -> Process ()+prepareSlave (masterProcess, hash, content) = handlePrepareSlave hash content >>= send masterProcess++-- template haskell vs. its result+-- needs: {-# LANGUAGE TemplateHaskell, DeriveDataTypeable, DeriveGeneric #-}+--remotable ['slaveTask] +-- ======>+slaveTask__static :: S.Static (TaskTransport -> Process ())+slaveTask__static = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.slaveTask"+slaveTask__sdict :: S.Static (PS.SerializableDict TaskTransport)+slaveTask__sdict = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.slaveTask__sdict"+-- not used for now, removed due to warnings as errors+--slaveTask__tdict :: S.Static (PS.SerializableDict ())+--slaveTask__tdict = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.slaveTask__tdict"+querySlavePreparation__static :: S.Static (QuerySlavePreparationRequest -> Process())+querySlavePreparation__static = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.querySlavePreparation"+querySlavePreparation__sdict :: S.Static (PS.SerializableDict QuerySlavePreparationRequest)+querySlavePreparation__sdict = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.querySlavePreparation__sdict"+prepareSlave__static :: S.Static (PrepareSlaveRequest -> Process())+prepareSlave__static = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.prepareSlave"+prepareSlave__sdict :: S.Static (PS.SerializableDict PrepareSlaveRequest)+prepareSlave__sdict = S.staticLabel "Control.Distributed.Task.Distribution.TaskDistribution.prepareSlave__sdict"+__remoteTable :: S.RemoteTable -> S.RemoteTable+__remoteTable =+  (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.slaveTask" (R1.toDynamic slaveTask))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.slaveTask__sdict" (R1.toDynamic (PS.SerializableDict :: PS.SerializableDict TaskTransport)))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.slaveTask__tdict" (R1.toDynamic (PS.SerializableDict :: PS.SerializableDict ())))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.querySlavePreparation" (R1.toDynamic querySlavePreparation))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.querySlavePreparation__sdict" (R1.toDynamic (PS.SerializableDict :: PS.SerializableDict QuerySlavePreparationRequest)))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.querySlavePreparation__tdict" (R1.toDynamic (PS.SerializableDict :: PS.SerializableDict ())))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.prepareSlave" (R1.toDynamic prepareSlave))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.prepareSlave__sdict" (R1.toDynamic (PS.SerializableDict :: PS.SerializableDict PrepareSlaveRequest)))+  . (S.registerStatic "Control.Distributed.Task.Distribution.TaskDistribution.prepareSlave__tdict" (R1.toDynamic (PS.SerializableDict :: PS.SerializableDict ())))++slaveTaskClosure :: TaskTransport -> S.Closure (Process ())+slaveTaskClosure =+  -- $(mkClosure 'slaveTask)+-- ======>+   ((S.closure+    (slaveTask__static+     `S.staticCompose`+     (staticDecode+      slaveTask__sdict)))+   . B.encode)++querySlavePreparationClosure :: QuerySlavePreparationRequest -> S.Closure (Process ())+querySlavePreparationClosure =+   ((S.closure+    (querySlavePreparation__static+     `S.staticCompose`+     (staticDecode+      querySlavePreparation__sdict)))+   . B.encode)+   +prepareSlaveClosure :: PrepareSlaveRequest -> S.Closure (Process ())+prepareSlaveClosure =+   ((S.closure+    (prepareSlave__static+     `S.staticCompose`+     (staticDecode+      prepareSlave__sdict)))+   . B.encode)++rtable :: RemoteTable+rtable = __remoteTable $ initRemoteTable+-- END bindings for node communication++type NodeConfig = (String, Int)++startSlaveNode :: NodeConfig -> IO ()+startSlaveNode (host, port) = do+  initDefaultLogging (show port)+  backend <- initializeBackend host (show port) rtable+  putStrLn "initializing slave"+  startSlave backend++executeDistributed :: NodeConfig -> TaskDef -> [DataDef] -> ResultDef -> ([TaskResult] -> IO ())-> IO ()+executeDistributed (host, port) taskDef dataDefs resultDef resultProcessor = do+  backend <- initializeBackend host (show port) rtable+  startMaster backend $ \slaveNodes -> do+    result <- executeOnNodes taskDef dataDefs resultDef slaveNodes+    liftIO $ resultProcessor result++executeOnNodes :: TaskDef -> [DataDef] -> ResultDef -> [NodeId] -> Process [TaskResult]+executeOnNodes taskDef dataDefs resultDef slaveNodes = do+  if null slaveNodes+    then say "no slaves => no results (ports open?)" >> return [] +    else executeOnNodes' taskDef dataDefs resultDef slaveNodes++executeOnNodes' :: TaskDef -> [DataDef] -> ResultDef -> [NodeId] -> Process [TaskResult]+executeOnNodes' taskDef dataDefs resultDef slaveNodes = do+  masterProcess <- getSelfPid+  config <- liftIO getConfiguration+  before <- liftIO getCurrentTime+  collectedResults <- distributeWorkForNodes masterProcess (_maxTasksPerNode config) (_distributionStrategy config) taskDef dataDefs resultDef slaveNodes+  after <- liftIO getCurrentTime+  say $ "total time: " ++ show (diffUTCTime after before)+  say $ "remote times:" ++ (printStatistics collectedResults)+  return $ concat $ map _collectedResults collectedResults++printStatistics :: [CollectedResult] -> String+printStatistics = concat . map printStat+  where+    printStat r = "\ntotal: " ++total++ ", remote total: " ++remoteTotal++ ", task execution total real time: " ++taskTotal++" for remote run: "++taskname+      where+        taskname = _taskName $ _collectedMetaData r+        total = show $ _totalDistributedRuntime r+        remoteTotal = show $ _totalRemoteRuntime r+        taskTotal = show $ _totalTasksRuntime r++type NodeOccupancy = Map.Map NodeId Bool++occupyNode :: NodeId -> NodeOccupancy -> NodeOccupancy+occupyNode = Map.adjust (bool True $ error "node already occupied")++unoccupyNode :: NodeId -> NodeOccupancy -> NodeOccupancy+unoccupyNode = Map.adjust (flip bool False $ error "node already vacant")++initializeOccupancy :: [NodeId] -> NodeOccupancy+initializeOccupancy = foldr (flip Map.insert False) Map.empty++{-|+ Tries to find work for every worker node, looking at all tasks, forgetting the node if no task is found.++ Note (obsolete, but may be reactivated): tries to be open to other result types but assumes a list of results, as these can be concatenated over multiple tasks. Requires result entries to be serializable, not the+  complete result - confusing this can cause devastatingly misleading compiler complaints about Serializable.+|-}+distributeWorkForNodes :: ProcessId -> Int -> DistributionStrategy -> TaskDef -> [DataDef] -> ResultDef -> [NodeId] -> Process [CollectedResult]+distributeWorkForNodes masterProcess maxTasksPerNode strategy taskDef dataDefs resultDef allNodes = doItFor ([], 0, initializeOccupancy allNodes, dataDefs)+  where+    doItFor :: ([CollectedResult], Int, NodeOccupancy, [DataDef]) -> Process [CollectedResult]+    doItFor (collected, 0, _, []) = return collected -- everything processed, results mainly in reversed order (a bit garbled)+    doItFor (collected, resultsWaitingOn, nodeOccupancy, []) = collectNextResult collected nodeOccupancy [] resultsWaitingOn -- everything distributed, but still results to be collected+    doItFor (collected, resultsWaitingOn, nodeOccupancy, undistributedTasks)+      | noFreeNodes nodeOccupancy = liftIO (logInfo $ "no unoccupied workers, wait results to come back: "++show nodeOccupancy)+                                    >> collectNextResult collected nodeOccupancy undistributedTasks resultsWaitingOn -- +      | otherwise = let freeNodes = nextFreeNodes nodeOccupancy+                        freeNode = if null freeNodes then error "no free nodes" else head freeNodes :: NodeId+                    in do+       liftIO $ logInfo $ "finding suitable tasks for the next unoccupied node: "++show freeNode++" - occupations: "++show nodeOccupancy+       (suitableTasks, remainingTasks) <- findSuitableTasks strategy freeNode+       if null suitableTasks+         then error $ "nothing found to distribute, should not have tried in the first place then: "++(show (map describe undistributedTasks, map describe suitableTasks, map describe remainingTasks))+         else do -- regular distribution+           say $ "spawning on: " ++ (show $ freeNode)+           spawnSlaveProcess masterProcess taskDef suitableTasks resultDef freeNode+           doItFor (collected, resultsWaitingOn+1, occupyNode freeNode nodeOccupancy, remainingTasks)+      where+        noFreeNodes = null . nextFreeNodes+        nextFreeNodes :: NodeOccupancy -> [NodeId]+        nextFreeNodes =  Map.keys . Map.filter not+        findSuitableTasks :: DistributionStrategy -> NodeId -> Process ([DataDef], [DataDef])+        findSuitableTasks AnywhereIsFine _ = return (take maxTasksPerNode undistributedTasks, drop maxTasksPerNode undistributedTasks)+        findSuitableTasks FirstTaskWithData freeNode = if length undistributedTasks <= maxTasksPerNode then findSuitableTasks AnywhereIsFine freeNode else takeLocalTasks maxTasksPerNode [] [] undistributedTasks+          where+            takeLocalTasks :: Int -> [DataDef] -> [DataDef] -> [DataDef] -> Process ([DataDef], [DataDef])+            takeLocalTasks 0 found notSuitable rest = return (found, notSuitable++rest) -- enough found+            takeLocalTasks n found notSuitable [] = return (found++(take n notSuitable), drop n notSuitable) -- everything searched, fill up+            takeLocalTasks n found notSuitable (t:rest) = do+              allNodesSuitableForTask <- findNodesWithData' t+              if any (==freeNode) allNodesSuitableForTask+                then takeLocalTasks (n-1) (t:found) notSuitable rest+                else takeLocalTasks n found (t:notSuitable) rest+              where+                findNodesWithData' :: DataDef -> Process [NodeId]+                findNodesWithData' (HdfsData loc) = liftIO $ findNodesWithData loc allNodes -- TODO this listing is not really efficient for this approach, caching necessary?+                findNodesWithData' (PseudoDB _) = return allNodes -- no data locality emulation support planned for simple pseudo db yet+    collectNextResult :: [CollectedResult] -> NodeOccupancy -> [DataDef] -> Int -> Process [CollectedResult]+    collectNextResult collected nodeOccupancy undistributedTasks resultsWaitingOn = do+      say $ "waiting for a result"+      collectedResult <- collectSingle+      let taskMetaData = _collectedMetaData collectedResult+          updatedResults = (collectedResult:collected) -- note: effectively scrambles result order a bit, but that should already be so due to parallel processing+        in do+        say $ "got result from: " ++ (show $ _slaveNodeId taskMetaData)+        doItFor (updatedResults, resultsWaitingOn-1, unoccupyNode (_slaveNodeId taskMetaData) nodeOccupancy, undistributedTasks)++{-|+ Represents the results of a single remote execution, including multiple results from parallel execution.+-}+data CollectedResult = CollectedResult {+  _collectedMetaData :: TaskMetaData,+  _totalDistributedRuntime :: NominalDiffTime,+  _totalRemoteRuntime :: NominalDiffTime,+  _totalTasksRuntime :: NominalDiffTime,+  _collectedResults :: [TaskResult]+}++collectSingle :: Process CollectedResult+collectSingle = receiveWait [+  match $ handleTransportedResult,+  matchAny $ \msg -> liftIO $ putStr " received unhandled  message : " >> print msg >> error "aborting due to unknown message type"+  ]+  where+    handleTransportedResult :: TransportedResult -> Process CollectedResult+    handleTransportedResult (TransportedResult taskMetaData remoteRuntime tasksRuntime serializedResults) = do+      say $ "got task processing response for: "++_taskName taskMetaData+      now <- liftIO getCurrentTime+      let+        taskName = _taskName taskMetaData+        distributedRuntime' = diffUTCTime now (deserializeTime $ _taskDistributionStartTime taskMetaData)+        remoteRuntime' = deserializeTimeDiff remoteRuntime+        tasksRuntime' = deserializeTimeDiff tasksRuntime+        in case serializedResults of+            (Left  msg) -> say (msg ++ " failure not handled for "++taskName++"...") >> return (CollectedResult taskMetaData distributedRuntime' remoteRuntime' tasksRuntime' []) -- note: no restarts take place here+            (Right results) -> return $ CollectedResult taskMetaData distributedRuntime' remoteRuntime' tasksRuntime' results++spawnSlaveProcess :: ProcessId -> TaskDef -> [DataDef] -> ResultDef -> NodeId -> Process ()+spawnSlaveProcess masterProcess taskDef dataDefs resultDef slaveNode = do+  now <- liftIO getCurrentTime+  preparedTaskDef <- prepareSlaveForTask taskDef+  taskTransport <- return $ TaskTransport masterProcess (TaskMetaData (taskDescription taskDef dataDefs slaveNode) slaveNode (serializeTime now)) preparedTaskDef dataDefs resultDef+  _unusedSlaveProcessId <- spawn slaveNode (slaveTaskClosure taskTransport)+  return ()+  where+    prepareSlaveForTask :: TaskDef -> Process TaskDef+    prepareSlaveForTask (DeployFullBinary program) = do+      hash <- return $ RemoteStore.calculateHash program+      say $ "preparing for " ++ (show hash)+      _ <- spawn slaveNode (querySlavePreparationClosure (masterProcess, hash))+      prepared <- expect -- TODO match slaveNodes?+      case prepared of+       Prepared -> say $ "already prepared: " ++ (show hash)+       Unprepared -> do+         say $ "distributing " ++ (show hash)+         before <- liftIO getCurrentTime+         _ <- spawn slaveNode (prepareSlaveClosure (masterProcess, hash, program))+         ok <- expect -- TODO match slaveNodes?+         after <- liftIO getCurrentTime+         case ok of+          PreparationFinished -> say $ "distribution took: " ++ (show $ diffUTCTime after before)+      return $ PreparedDeployFullBinary hash+    prepareSlaveForTask d = return d -- nothing to prepare for other tasks (for now)++-- remote logic++handleSlaveTask :: TaskTransport -> Process ()+handleSlaveTask (TaskTransport masterProcess taskMetaData taskDef dataDefs resultDef) = do+  transportedResult <- (+    do+      acceptTime <- liftIO getCurrentTime+      say $ "processing: " ++ taskName+      results <- liftIO (processTasks taskDef dataDefs resultDef)+      say $ "processing done for: " ++ taskName+      processingDoneTime <- liftIO getCurrentTime+      return $ prepareResultsForResponding taskMetaData acceptTime processingDoneTime results+    ) `catch` buildError+  say $ "replying"+  send masterProcess (transportedResult :: TransportedResult)+  say $ "reply sent"+  where+    taskName = _taskName taskMetaData+    buildError :: SomeException -> Process TransportedResult+    buildError e = return $ TransportedResult taskMetaData emptyDuration emptyDuration $ Left $ "Task execution (for: "++taskName++") failed: " ++ (show e)+      where+        emptyDuration = fromIntegral (0 :: Integer)++prepareResultsForResponding :: TaskMetaData -> UTCTime -> UTCTime -> TasksExecutionResult -> TransportedResult+prepareResultsForResponding metaData acceptTime processingDoneTime (results, tasksRuntime) =+  TransportedResult metaData remoteRuntime (serializeTimeDiff tasksRuntime) $ Right results+  where+    remoteRuntime = serializeTimeDiff $ diffUTCTime processingDoneTime acceptTime++-- preparation negotiaion++handleQuerySlavePreparation :: Int -> Process QuerySlavePreparationResponse+handleQuerySlavePreparation hash = liftIO (RemoteStore.get hash) >>= return . (maybe Unprepared (\_ -> Prepared))++handlePrepareSlave :: Int -> ByteString -> Process PrepareSlaveResponse+handlePrepareSlave hash content = liftIO (RemoteStore.put hash content) >> return PreparationFinished++-- simple tasks+showSlaveNodes :: NodeConfig -> IO ()+showSlaveNodes config = withSlaveNodes config (+  \slaveNodes -> putStrLn ("Slave nodes: " ++ show slaveNodes))++showSlaveNodesWithData :: NodeConfig -> String -> IO ()+showSlaveNodesWithData slaveConfig hdfsFilePath = withSlaveNodes slaveConfig (+  \slaveNodes -> do+    nodesWithData <- findNodesWithData hdfsFilePath slaveNodes+    putStrLn $ "Found these nodes with data: " ++ show nodesWithData)++withSlaveNodes :: NodeConfig -> ([NodeId] -> IO ()) -> IO ()+withSlaveNodes (host, port) action = liftIO $ do+  backend <- initializeBackend host (show port) initRemoteTable+  startMaster backend (\slaveNodes -> liftIO (action slaveNodes))++shutdownSlaveNodes :: NodeConfig -> IO ()+shutdownSlaveNodes (host, port) = do+  backend <- initializeBackend host (show port) rtable+  startMaster backend $ \slaveNodes -> do+    say $ "found " ++ (show $ length slaveNodes) ++ " slave nodes, shutting them down"+    forM_ slaveNodes terminateSlave+    -- try terminateAllSlaves instead?++taskDescription :: TaskDef -> [DataDef] -> NodeId -> String+taskDescription t d n = "task: " ++ (describe t) ++ " " ++ (concat $ intersperse ", " $ map describe d) ++ " on " ++ (show n)

@@ -0,0 +1,55 @@+{-# LANGUAGE DeriveDataTypeable, DeriveGeneric, OverlappingInstances #-} --overlapping due to serialization+module Control.Distributed.Task.Distribution.TaskTransport where++import Control.Distributed.Process (ProcessId, NodeId)+import Control.Distributed.Process.Serializable (Serializable)+import Data.Binary (Binary)+import Data.ByteString.Lazy (ByteString)+import Data.Typeable (Typeable)+import GHC.Generics (Generic)++import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.Types.TaskTypes++--- task execution++data TaskTransport = TaskTransport {+  _masterProcess :: ProcessId,+  _taskMetaData :: TaskMetaData,+  _taskDef :: TaskDef,+  _dataDefs :: [DataDef],+  _resultDef :: ResultDef+  } deriving (Typeable, Generic)+instance Binary TaskTransport+instance Serializable TaskTransport++type TimeStamp = (Integer, Rational)++data TaskMetaData = TaskMetaData {+  _taskName :: String,+  _slaveNodeId :: NodeId,+  _taskDistributionStartTime :: TimeStamp+  } deriving (Typeable, Generic)+instance Binary TaskMetaData+instance Serializable TaskMetaData++data TransportedResult = TransportedResult {+  _transportedResultMetaData :: TaskMetaData,+  _transportedTotalRemoteRuntime :: Rational,+  _transportedTotalTasksRuntime :: Rational,+  _transportedResults :: Either String [TaskResult]+  } deriving (Typeable, Generic)+instance Binary TransportedResult+instance Serializable TransportedResult++--- task preparation++type QuerySlavePreparationRequest = (ProcessId, Int)+data QuerySlavePreparationResponse = Prepared | Unprepared deriving (Typeable, Generic)+instance Binary QuerySlavePreparationResponse+instance Serializable QuerySlavePreparationResponse++type PrepareSlaveRequest = (ProcessId, Int, ByteString)+data PrepareSlaveResponse = PreparationFinished deriving (Typeable, Generic)+instance Binary PrepareSlaveResponse+instance Serializable PrepareSlaveResponse

@@ -0,0 +1,48 @@+{-|+ Simple binary file storage for temporary files.+ Stores under <tempdir>/temp-binary-storage/+|-}+module Control.Distributed.Task.TaskSpawning.BinaryStorage (calculateHash, put, get, clearAll) where++import qualified Data.ByteString.Lazy as BL+import Data.Hashable (hash)+import System.Directory++calculateHash :: BL.ByteString -> Int+calculateHash = hash++put :: Int -> BL.ByteString -> IO ()+put fileHash content = do+  (filePath, fileExists) <- getFilePath fileHash+  if fileExists+    then return ()+    else BL.writeFile filePath content++get :: Int -> IO (Maybe FilePath)+get fileHash = do+  (filePath, fileExists) <- getFilePath fileHash+  if fileExists+    then return $ Just filePath+    else return $ Nothing++clearAll :: IO ()+clearAll = do+  homeDir <- getHomeDir+  removeFile homeDir++getFilePath :: Int -> IO (FilePath, Bool)+getFilePath fileHash = do+  homeDir <- getHomeDir+  filePath <- return $ homeDir ++ "/" ++ (show fileHash)+  fileExists <- doesFileExist filePath+  return (filePath, fileExists)++getHomeDir :: IO FilePath+getHomeDir = do+  sysTempDir <- getTemporaryDirectory+  homeDir <- return $ sysTempDir ++ "/temp-binary-storage/"+  hDExists <- doesDirectoryExist homeDir+  if hDExists+    then return ()+    else createDirectory homeDir+  return homeDir

@@ -0,0 +1,89 @@+module Control.Distributed.Task.TaskSpawning.DeployFullBinary (+  ExternalExecutionResult,+  deployAndRunFullBinary, deployAndRunExternalBinary, fullBinaryExecution, runExternalBinary,+  packIOHandling, unpackIOHandling, IOHandling(..)+  ) where++import qualified Codec.Compression.GZip as GZip+import Control.Concurrent.Async (async, wait)+import qualified Data.ByteString.Lazy.Char8 as BLC+import qualified Data.ByteString.Lazy as BL+import Data.List (intersperse)+import System.Process (readProcessWithExitCode)++import Control.Distributed.Task.DataAccess.DataSource+import Control.Distributed.Task.TaskSpawning.ExecutionUtil+import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.TaskSpawning.TaskDescription+import Control.Distributed.Task.TaskSpawning.TaskSpawningTypes+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.FileUtil+import Control.Distributed.Task.Util.Logging++type ExternalExecutionResult = ([TaskResult], NominalDiffTime)++deployAndRunFullBinary :: String -> IOHandling -> BL.ByteString -> IO ExternalExecutionResult+deployAndRunFullBinary mainArg = deployAndRunExternalBinary [mainArg]++deployAndRunExternalBinary :: [String] -> IOHandling -> BL.ByteString -> IO ExternalExecutionResult+deployAndRunExternalBinary programBaseArgs ioHandling program =+  withTempBLFile "distributed-program" program $ runExternalBinary programBaseArgs ioHandling++{-|+ Makes an external system call, parameters must match those read in RemoteExecutionSupport.+-}+runExternalBinary :: [String] -> IOHandling -> FilePath -> IO ExternalExecutionResult+runExternalBinary programBaseArgs ioHandling executablePath =+  measureDuration $ do+    readProcessWithExitCode "chmod" ["+x", executablePath] "" >>= expectSilentSuccess+    logInfo $ "worker: spawning external process: "++executablePath++" with baseArgs: "++show programBaseArgs+    processOutput <- executeExternal executablePath (programBaseArgs ++ [packIOHandling ioHandling])+    logInfo $ "worker: external process finished: "++executablePath+    return $ consumeResults processOutput++{-|+ Some methods parse the contents of stdout, thus these will fail in the case of logging to it (only ERROR at the moment).+-}+fullBinaryExecution :: IOHandling -> Task -> IO ()+fullBinaryExecution (IOHandling dataDefs resultDef) task = do+  logInfo $ "external binary: processing "++(concat $ intersperse ", " $ map describe dataDefs)+  tasks <- mapM (async . runSingle task resultDef) dataDefs+   -- reminder: trying to collect results here and submit them collectively makes it harder to force evaluation into the parallel part+  mapM_ wait tasks+  logInfo $ "external binary: tasks completed"++runSingle :: Task -> ResultDef -> DataDef -> IO ()+runSingle task resultDef dataDef = do+  taskInput <- loadData dataDef+  let result = task taskInput in do+    processedResult <- processResult result+    emitResult processedResult+  where+    processResult :: TaskResult -> IO TaskResult+    processResult taskResult =+      case resultDef of+       ReturnAsMessage -> return taskResult+       ReturnOnlyNumResults -> return [BLC.pack $ show $ length taskResult]+       (HdfsResult pre suf z) -> writeResultsToHdfs dataDef pre suf z taskResult++emitResult :: TaskResult -> IO ()+emitResult = BLC.putStrLn . BLC.concat . intersperse (BLC.pack "|")++consumeResults :: String -> [TaskResult]+consumeResults = map (BLC.split '|') . BLC.lines . BLC.pack++writeResultsToHdfs :: DataDef -> String -> String -> Bool -> TaskResult -> IO TaskResult+writeResultsToHdfs dataDef pre suf z taskResult = writeToHdfs >> return []+         where+           hdfsPath (HdfsData p) = pre++"/"++p++"/"++suf+           hdfsPath _ = error "implemented only for hdfs output"+           writeToHdfs =+             withTempBLCFile "move-to-hdfs" fileContent $ \tempFileName ->+               let (dirPart, filePart) = splitBasePath $ hdfsPath dataDef+               in copyToHdfs tempFileName dirPart filePart+               where+                 fileContent = (if z then GZip.compress else id) $ BLC.concat $ intersperse (BLC.pack "\n") taskResult+                 copyToHdfs localFile destPath destFilename = do+                   logInfo $ "external binary: copying "++localFile++" to "++destPath++"/"++destFilename+                   _ <- executeExternal "hdfs" ["dfs", "-mkdir", "-p", destPath] -- hadoop-rpc does not yet support writing, the external system call is an acceptable workaround+                   executeExternal "hdfs" ["dfs", "-copyFromLocal", localFile, destPath++"/"++destFilename]

@@ -0,0 +1,76 @@+module Control.Distributed.Task.TaskSpawning.DeployObjectCodeRelinked (+  loadObjectCode, deployAndRunObjectCodeRelinked+  ) where++import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Lazy.Char8 as BLC+import Data.List (intersperse)+import System.IO.Temp (withSystemTempDirectory)++import Control.Distributed.Task.TaskSpawning.DeployFullBinary+import Control.Distributed.Task.TaskSpawning.ExecutionUtil+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging++loadObjectCode :: IO BL.ByteString+loadObjectCode = do+  config <- getConfiguration+  BL.readFile (_objectCodePathOnMaster config ++ "/RemoteExecutable.o")++deployAndRunObjectCodeRelinked :: BLC.ByteString -> IOHandling -> IO ExternalExecutionResult+deployAndRunObjectCodeRelinked objectCode ioHandling =+  withSystemTempDirectory "object-code-build-dir" $ \builddir ->+    let objectCodeFilePath = builddir ++ "/RemoteExecutable.o"+        binaryPath = builddir++"/binary"+    in do+      logInfo "slave: preparing alternative main object code"+      config <- getConfiguration+      libs <- readFile "etc/remotelibs" >>= return . lines+      _ <- executeExternal "ghc"+           [+             "-no-link",+             "-outputdir", builddir,+             "-package-db", _packageDbPath config,+             "-i"++(_objectCodeResourcesPathSrc config),+             "Control.Distributed.Task.TaskSpawning.DeployFullBinary",+             "Control.Distributed.Task.Types.TaskTypes",+             "-i"++(_objectCodeResourcesPathExtra config),+             "RemoteExecutable",+             (_objectCodeResourcesPathExtra config)++"/RemoteExecutor.hs" --explicit for Main+           ]+      logInfo $ "slave: prepared frame in "++builddir+      logInfo "slave: storing transported object code"+      BL.writeFile objectCodeFilePath objectCode+      logInfo $ "slave: linking: " ++ binaryPath+      _ <- executeExternal "ghc"+           ([+               "-o", binaryPath,+               builddir++"/Main.o",+               objectCodeFilePath,+               "-package-db", _packageDbPath config+            ]+            ++ (map ((builddir++"/")++) taskDistributionDeps)+            ++ (asPackageOpts libs)+            )+      logInfo $ "slave: running " ++ binaryPath+      res <- runExternalBinary [] ioHandling binaryPath+      logInfo $ "executing task finished"+      logInfo $ "results: "++show res+      return res+  where+    taskDistributionDeps = [+      "Control/Distributed/Task/DataAccess/DataSource.o",+      "Control/Distributed/Task/DataAccess/HdfsDataSource.o",+      "Control/Distributed/Task/DataAccess/SimpleDataSource.o",+      "Control/Distributed/Task/TaskSpawning/DeployFullBinary.o",+      "Control/Distributed/Task/TaskSpawning/ExecutionUtil.o",+      "Control/Distributed/Task/TaskSpawning/TaskDescription.o",+      "Control/Distributed/Task/TaskSpawning/TaskDefinition.o",+      "Control/Distributed/Task/TaskSpawning/TaskSpawningTypes.o",+      "Control/Distributed/Task/Types/TaskTypes.o",+      "Control/Distributed/Task/Util/Logging.o",+      "Control/Distributed/Task/Util/Configuration.o",+      "Control/Distributed/Task/Util/ErrorHandling.o",+      "Control/Distributed/Task/Util/FileUtil.o"+      ]+    asPackageOpts = ("-package":) . (intersperse "-package")

@@ -0,0 +1,29 @@+module Control.Distributed.Task.TaskSpawning.DeploySerializedThunk (deployAndRunSerializedThunk, serializedThunkExecution, acceptAndExecuteSerializedThunk) where++import qualified Data.ByteString.Lazy as BL+import System.FilePath ()++import Control.Distributed.Task.TaskSpawning.DeployFullBinary (ExternalExecutionResult, deployAndRunExternalBinary, fullBinaryExecution)+import Control.Distributed.Task.TaskSpawning.FunctionSerialization (deserializeFunction)+import Control.Distributed.Task.TaskSpawning.TaskSpawningTypes+import Control.Distributed.Task.Types.TaskTypes++{-+ Deploys a given binary and executes it with the arguments defined by convention, including the serialized closure runnable in that program.++ The execution does not include error handling, this should be done on master/client.+-}+deployAndRunSerializedThunk :: String -> BL.ByteString -> IOHandling -> BL.ByteString -> IO ExternalExecutionResult+deployAndRunSerializedThunk mainArg taskFunction = deployAndRunExternalBinary [mainArg, show taskFunction]++{-+ Accepts the distributed, serialized closure as part of the spawned program and executes it.++ Parameter handling is done via simple serialization.+-}+--TODO refactor module borders to have fully unterstandable resonsibilities, shis should go up to have nothing to do with serialization of thunks?+acceptAndExecuteSerializedThunk :: BL.ByteString -> IO (TaskInput -> TaskResult)+acceptAndExecuteSerializedThunk taskFn = (deserializeFunction taskFn :: IO (TaskInput -> TaskResult)) >>= (\f -> return (take 10 . f))++serializedThunkExecution ::  IOHandling -> (TaskInput -> TaskResult) -> IO ()+serializedThunkExecution = fullBinaryExecution -- nothing conceptually different at this point

@@ -0,0 +1,111 @@+module Control.Distributed.Task.TaskSpawning.ExecutionUtil (+  withEnv,+  withTempBLFile,+  withTempBLCFile,+  withTempFile,+  ignoreIOExceptions,+  expectSilentSuccess,+  expectSuccess,+  createTempFilePath,+  serializeTaskInput,+  deserializeTaskInput,+  parseResultStrict, -- TODO move out of here+  executeExternal,+  measureDuration,+  readStdTillEOF+  ) where++import Control.Exception.Base (bracket, catch)+import qualified Data.ByteString.Char8 as BC+import qualified Data.ByteString.Lazy.Char8 as BLC+import qualified Data.ByteString.Lazy as BL+import Data.List (intersperse)+import Data.Time.Clock (diffUTCTime, NominalDiffTime, getCurrentTime)+import System.Directory (removeFile)+import System.Environment (lookupEnv, setEnv, unsetEnv)+import System.Exit (ExitCode(..))+import System.FilePath ()+import System.IO.Error (catchIOError, isEOFError)+import System.IO.Temp (withSystemTempFile)+import System.Process (readProcessWithExitCode)++import Control.Distributed.Task.Types.TaskTypes (TaskInput, TaskResult)+import Control.Distributed.Task.Util.ErrorHandling+import Control.Distributed.Task.Util.Logging++withEnv :: String -> String -> IO a -> IO a+withEnv key value action = do+  old <- lookupEnv key+  setEnv key value+  res <- action+  maybe (unsetEnv key) (\o -> setEnv key o) old+  return res++withTempBLFile :: FilePath -> BL.ByteString -> (FilePath -> IO result) -> IO result+withTempBLFile = withTempFile BL.writeFile++withTempBLCFile :: FilePath -> BLC.ByteString -> (FilePath -> IO result) -> IO result+withTempBLCFile = withTempFile BLC.writeFile++withTempFile :: (FilePath -> dataType -> IO ()) -> FilePath -> dataType -> (FilePath -> IO result) -> IO result+withTempFile writer filePathTemplate fileContent =+  bracket+    (do+      filePath <- createTempFilePath filePathTemplate+      writer filePath fileContent+      return filePath)+    (\filePath -> ignoreIOExceptions $ removeFile filePath)++ignoreIOExceptions :: IO () -> IO ()+ignoreIOExceptions io = io `catchIOError` (\_ -> return ())++expectSilentSuccess :: (ExitCode, String, String) -> IO ()+expectSilentSuccess executionOutput = expectSuccess executionOutput >>= \res -> case res of+  "" -> return ()+  _ -> error $ "no output expected, but got: " ++ res++expectSuccess :: (ExitCode, String, String) -> IO String+expectSuccess (ExitSuccess, result, []) = return result+expectSuccess (code, out, err) = error $ "command exited with unexpected status: "++show code++", output:\n"++out++"\nstderr:\n"++err++createTempFilePath :: String -> IO FilePath+createTempFilePath template = do+  -- a bit hackish, we only care about the generated file name, we have to do the file handler handling ourselves ...+  withSystemTempFile template (\ f _ -> return f)++serializeTaskInput :: TaskInput -> BLC.ByteString+serializeTaskInput = BLC.pack . show++deserializeTaskInput :: BLC.ByteString -> IO TaskInput+deserializeTaskInput s = withErrorAction logError "Could not read input data" $ return $ read $ BLC.unpack s++parseResultStrict :: BLC.ByteString -> IO TaskResult+parseResultStrict s = withErrorPrefix ("Cannot parse result: "++ (BLC.unpack s)) $ return $! (BLC.lines s :: TaskResult)++executeExternal :: FilePath -> [String] -> IO String+executeExternal executable args = do+  logInfo $ "executing: " ++ executable ++ " " ++ (filter (/='\n') $ concat $ intersperse " " args)+  result <- withErrorAction logError ("Could not run [" ++ (show executable) ++ "] successfully: ") (readProcessWithExitCode executable args "")+  expectSuccess result++measureDuration :: IO a -> IO (a, NominalDiffTime)+measureDuration action = do+  before <- getCurrentTime+  res <- action -- TODO eager enough?+  after <- getCurrentTime+  return (res, diffUTCTime after before)++readStdTillEOF :: IO TaskInput+readStdTillEOF = do+  l <- readLnUnlessEOF+  case l of+   Nothing -> return []+   (Just line) -> do+     rest <- readStdTillEOF+     return (line:rest)+  where+    readLnUnlessEOF :: IO (Maybe BL.ByteString)+    readLnUnlessEOF = (BC.getLine >>= return . Just . BLC.fromStrict) `catch` eofHandler+      where+        eofHandler :: IOError -> IO (Maybe BL.ByteString)+        eofHandler e = if isEOFError e then return Nothing else ioError e

@@ -0,0 +1,20 @@+module Control.Distributed.Task.TaskSpawning.FunctionSerialization (serializeFunction, deserializeFunction) where++import Data.Binary (encode, decode)+import qualified Data.ByteString.Lazy as L+import Data.Typeable (Typeable)+import qualified GHC.Packing as P++import Control.Distributed.Task.Util.ErrorHandling++serializeFunction :: (Typeable a, Typeable b) => (a -> b) -> IO L.ByteString+serializeFunction = serialize++serialize :: Typeable a => a -> IO L.ByteString+serialize a = P.trySerialize a >>= return . encode++deserializeFunction :: (Typeable a, Typeable b) => L.ByteString -> IO (a -> b)+deserializeFunction = deserialize++deserialize :: (Typeable a) => L.ByteString -> IO a+deserialize bs = withErrorPrefix ("Could not deserialize: "++(show bs)) $ P.deserialize $ decode bs

@@ -0,0 +1,52 @@+{-|+  Catches expected entry points for full binary deployment / thunk serialization.+  These modes deploy the itself as a program and are called remote with different arguments, which is handled here.+|-}+module Control.Distributed.Task.TaskSpawning.RemoteExecutionSupport (+  withRemoteExecutionSupport,+  withFullBinaryRemoteExecutionSupport,+  withSerializedThunkRemoteExecutionSupport+  ) where++import qualified System.Log.Logger as L+import System.Environment (getArgs, getExecutablePath)++import Control.Distributed.Task.TaskSpawning.DeployFullBinary (unpackIOHandling)+import Control.Distributed.Task.TaskSpawning.TaskSpawning (executeFullBinaryArg, executionWithinSlaveProcessForFullBinaryDeployment, executeSerializedThunkArg, executionWithinSlaveProcessForThunkSerialization)+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging++withRemoteExecutionSupport :: (TaskInput -> TaskResult) -> IO () -> IO ()+withRemoteExecutionSupport fn = withSerializedThunkRemoteExecutionSupport . withFullBinaryRemoteExecutionSupport fn++withFullBinaryRemoteExecutionSupport :: (TaskInput -> TaskResult) -> IO () -> IO ()+withFullBinaryRemoteExecutionSupport fn mainAction = do+  args <- getArgs+  case args of+   [mode, ioHandling] ->+     if mode == executeFullBinaryArg+     then+       initTaskLogging >>+       executionWithinSlaveProcessForFullBinaryDeployment (unpackIOHandling ioHandling) fn+     else mainAction+   _ -> mainAction++withSerializedThunkRemoteExecutionSupport :: IO () -> IO ()+withSerializedThunkRemoteExecutionSupport mainAction = do+  args <- getArgs+  case args of+   [mode, taskFn, ioHandling] ->+     if mode == executeSerializedThunkArg+     then+       initTaskLogging >>+       executionWithinSlaveProcessForThunkSerialization (unpackIOHandling ioHandling) taskFn+     else mainAction+   _ -> mainAction++initTaskLogging :: IO ()+initTaskLogging = do+  conf <- getConfiguration+  initLogging L.ERROR L.INFO (_taskLogFile conf)+  self <- getExecutablePath+  logInfo $ "started task execution for: "++self

@@ -0,0 +1,91 @@+module Control.Distributed.Task.TaskSpawning.SourceCodeExecution (+  processSourceCodeTasks+  ) where++import Control.Monad.IO.Class (MonadIO)+import Data.List (intersperse)+import Data.Time.Clock (NominalDiffTime)+import Data.Typeable (Typeable)+import qualified Language.Haskell.Interpreter as I+import System.Directory (getTemporaryDirectory, removeFile, removeDirectory)+import System.IO.Temp (createTempDirectory)+import System.FilePath ((</>))++import Control.Distributed.Task.DataAccess.DataSource (loadData)+import Control.Distributed.Task.TaskSpawning.ExecutionUtil+import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.TaskSpawning.TaskDescription+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.Configuration+import Control.Distributed.Task.Util.Logging++processSourceCodeTasks :: String -> String -> [DataDef] -> IO ([TaskResult], NominalDiffTime)+processSourceCodeTasks moduleName moduleContent dataDefs =+  measureDuration $ mapM runSourceCodeTask dataDefs+    where+      runSourceCodeTask :: DataDef -> IO TaskResult+      runSourceCodeTask dataDef = do+        logInfo $ "loading data for: "++describe dataDef+        taskInput <- loadData dataDef+        logInfo $ "applying data to task:"++moduleName+        result <- applySourceCodeTaskLogic taskInput+        return result+        where+          applySourceCodeTaskLogic taskInput = do+            putStrLn "compiling task from source code"+            taskFn <- loadTask (I.as :: Task) moduleName moduleContent+            putStrLn "applying data"+            return $ taskFn taskInput++loadTask :: (Typeable resultType) => resultType -> String -> String -> IO resultType+loadTask resultType moduleName moduleContent= do+  config <- getConfiguration+  iRes <- I.runInterpreter (loadTaskDef resultType moduleName moduleContent config)+  case iRes of+       Left err -> do+         printInterpreterError err+         error $ "could not load " ++ moduleName+       Right res -> return res++loadTaskDef :: Typeable a => a -> String -> String -> Configuration -> I.Interpreter a+loadTaskDef resultType moduleName moduleContent config = do+  sayI $ "Interpreter: Loading static modules and: " ++ moduleName ++ " ..."+  I.set [+    I.installedModulesInScope I.:= True,+    I.searchPath I.:= [_sourceCodeDistributionHome config]+    ]+  withTempModuleFile moduleName moduleContent loadModule+  func <- I.interpret "task" resultType+  sayI $ "done.\n"+  return func+    where+      loadModule moduleFilePath = do+        I.loadModules [moduleFilePath]+        I.setTopLevelModules [moduleName]+        I.setImportsQ (_sourceCodeModules config)++withTempModuleFile :: (MonadIO m) => String -> String -> (FilePath -> m a) -> m a+withTempModuleFile moduleName moduleContent moduleAction = do+  (moduleFile, moduleDir) <- I.liftIO $ writeModuleFile+  res <- moduleAction moduleFile -- FIXME Exception Handling -> cleanup+  I.liftIO $ cleanupModuleFile (moduleFile, moduleDir)+  return res+  where+    writeModuleFile :: IO (FilePath, FilePath)+    writeModuleFile = do+      tempDir <- getTemporaryDirectory+      moduleTempDir <- createTempDirectory tempDir moduleName -- FIXME a) hierarchical module names, b) could probably be easier implemented with withSystemTempDirectory, too+      moduleFile <- return $ moduleTempDir </> moduleName ++ ".hs"+      writeFile moduleFile moduleContent+      return (moduleFile, moduleTempDir)+    cleanupModuleFile :: (FilePath, FilePath) -> IO ()+    cleanupModuleFile (f, d) = do+      removeFile f+      removeDirectory d++printInterpreterError :: I.InterpreterError -> IO ()+printInterpreterError (I.WontCompile ghcErrors) = putStrLn $ "InterpreterError: " ++ (concat $ intersperse "\n" $ map I.errMsg ghcErrors)+printInterpreterError e = putStrLn $ "InterpreterError: " ++ (show e)++sayI :: String -> I.Interpreter ()+sayI = I.liftIO . putStr

@@ -0,0 +1,72 @@+{-# LANGUAGE DeriveDataTypeable, DeriveGeneric #-}+module Control.Distributed.Task.TaskSpawning.TaskDefinition where++import Control.Distributed.Process.Serializable (Serializable)+import Data.Binary (Binary)+import Data.ByteString.Lazy (ByteString)+import Data.Typeable (Typeable)+import GHC.Generics (Generic)++import Control.Distributed.Task.Types.HdfsConfigTypes++{-+  The way how code is to be distributed. The approaches have different disadvantages:++  - source code: very limited type checking, breaks integration of code in regular program+  - function serialization: special entry point in Main necessary+  - object code distribution: requires compilation intermediate result, cumbersome to add required libs/modules+-}+data TaskDef+ = SourceCodeModule {+   _moduleName :: String,+   _moduleContent :: String+   }+ | DeployFullBinary {+   _deployable :: ByteString+   }+ | PreparedDeployFullBinary {+   _preparedFullBinaryHash :: Int+   }+ | UnevaluatedThunk {+   _unevaluatedThunk :: ByteString,+   _deployable :: ByteString+   }+ | ObjectCodeModule {+   _objectCode :: ByteString+   } deriving (Typeable, Generic)+instance Binary TaskDef+instance Serializable TaskDef++{-+ Where data comes from:++ - hdfs data source+ - very simple file format for testing purposes, files with numbers expected relative to work directory+-}+data DataDef+  = HdfsData {+    _hdfsInputPath :: HdfsPath+    }+  | PseudoDB {+    _pseudoDBFilePath :: FilePath+    } deriving (Typeable, Generic)+instance Binary DataDef+instance Serializable DataDef++{-+ Where calculation results go:++ - simply respond the answer, aggregation happens on master application+ - only num results: for testing/benchmarking purposes+-}+data ResultDef+ = ReturnAsMessage+ | ReturnOnlyNumResults+ | HdfsResult {+   _outputPrefix :: String,+   _outputSuffix :: String,+   _outputZipped :: Bool+   }+ deriving (Typeable, Generic)+instance Binary ResultDef+instance Serializable ResultDef

@@ -0,0 +1,15 @@+module Control.Distributed.Task.TaskSpawning.TaskDescription where++import Control.Distributed.Task.TaskSpawning.TaskDefinition++class Describable a where+  describe :: a -> String+instance Describable TaskDef where+  describe (SourceCodeModule n _) = n+  describe (DeployFullBinary _) = "user function defined in main"+  describe (PreparedDeployFullBinary _) = "user function defined in main (prepared)"+  describe (UnevaluatedThunk _ _) = "unevaluated user function"+  describe (ObjectCodeModule _) = "object code module"+instance Describable DataDef where+  describe (HdfsData p) = p+  describe (PseudoDB p) = p

@@ -0,0 +1,87 @@+module Control.Distributed.Task.TaskSpawning.TaskSpawning (+  processTasks, TasksExecutionResult,+  fullBinarySerializationOnMaster, executeFullBinaryArg, executionWithinSlaveProcessForFullBinaryDeployment,+  serializedThunkSerializationOnMaster, executeSerializedThunkArg, executionWithinSlaveProcessForThunkSerialization,+  objectCodeSerializationOnMaster+  ) where++import qualified Data.ByteString.Lazy as BL+import Data.List (intersperse)++import qualified Control.Distributed.Task.TaskSpawning.BinaryStorage as RemoteStore+import qualified Control.Distributed.Task.TaskSpawning.DeployFullBinary as DFB+import qualified Control.Distributed.Task.TaskSpawning.DeploySerializedThunk as DST+import qualified Control.Distributed.Task.TaskSpawning.DeployObjectCodeRelinked as DOC+import Control.Distributed.Task.TaskSpawning.FunctionSerialization (serializeFunction, deserializeFunction)+import Control.Distributed.Task.TaskSpawning.SourceCodeExecution (processSourceCodeTasks)+import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Control.Distributed.Task.TaskSpawning.TaskDescription+import Control.Distributed.Task.TaskSpawning.TaskSpawningTypes+import Control.Distributed.Task.Types.TaskTypes+import Control.Distributed.Task.Util.ErrorHandling+import Control.Distributed.Task.Util.Logging++executeFullBinaryArg, executeSerializedThunkArg :: String+executeFullBinaryArg = "executefullbinary"+executeSerializedThunkArg = "executeserializedthunk"++type TasksExecutionResult = DFB.ExternalExecutionResult++{-|+  Apply the task on the data, producing either a location, where the results are stored, or the results directly.+  Which of those depends on the distribution type, when external programs are spawned the former can be more efficient,+  but if there is no such intermediate step, a direct result is better.+|-}+processTasks :: TaskDef -> [DataDef] -> ResultDef -> IO TasksExecutionResult+-- source code distribution behaves a bit different and only supports collectonmaster+processTasks (SourceCodeModule moduleName moduleContent) dataDefs _ = processSourceCodeTasks moduleName moduleContent dataDefs+processTasks taskDef dataDefs resultDef = do+  logInfo $ "spawning task for: "++(concat $ intersperse ", " $ map describe dataDefs)+  spawnExternalTask taskDef dataDefs resultDef++spawnExternalTask :: TaskDef -> [DataDef] -> ResultDef -> IO DFB.ExternalExecutionResult+spawnExternalTask (SourceCodeModule _ _) _ _ = error "source code distribution is handled differently"+-- Full binary deployment step 2/3: run within slave process to deploy the distributed task binary+spawnExternalTask (DeployFullBinary program) dataDefs resultDef =+  DFB.deployAndRunFullBinary executeFullBinaryArg (IOHandling dataDefs resultDef) program+spawnExternalTask (PreparedDeployFullBinary hash) dataDefs resultDef = do+  filePath_ <- RemoteStore.get hash+  maybe (error $ "no such program: "++show hash) (DFB.runExternalBinary [executeFullBinaryArg] (IOHandling dataDefs resultDef)) filePath_+-- Serialized thunk deployment step 2/3: run within slave process to deploy the distributed task binary+spawnExternalTask (UnevaluatedThunk function program) dataDefs resultDef =+  DST.deployAndRunSerializedThunk executeSerializedThunkArg function (IOHandling dataDefs resultDef) program+-- Partial binary deployment step 2/2: receive distribution on slave, link object file and spawn slave process, read its output,+-- the third step (accepting runtime arguments) is linked into the task executable (see RemoteExecutor)+spawnExternalTask (ObjectCodeModule objectCode) dataDefs resultDef =+  DOC.deployAndRunObjectCodeRelinked objectCode (IOHandling dataDefs resultDef)++-- Full binary deployment step 1/3+fullBinarySerializationOnMaster :: FilePath -> IO TaskDef+fullBinarySerializationOnMaster programPath = do+  currentExecutable <- BL.readFile programPath+  return $ DeployFullBinary currentExecutable++-- Serialized thunk deployment step 1/3: run within the client/master process to serialize itself.+serializedThunkSerializationOnMaster :: FilePath -> (TaskInput -> TaskResult) -> IO TaskDef+serializedThunkSerializationOnMaster programPath function = do+  program <- BL.readFile programPath -- TODO ByteString serialization should be contained within DST module+  taskFn <- serializeFunction function+  return $ UnevaluatedThunk taskFn program++-- Full binary deployment step 3/3: run within the spawned process for the distributed executable, applies data to distributed task.+executionWithinSlaveProcessForFullBinaryDeployment :: IOHandling -> (TaskInput -> TaskResult) -> IO ()+executionWithinSlaveProcessForFullBinaryDeployment = DFB.fullBinaryExecution++-- Serialized thunk deployment step 3/3: run within the spawned process for the distributed executable, applies data to distributed task.+executionWithinSlaveProcessForThunkSerialization :: IOHandling -> String -> IO ()+executionWithinSlaveProcessForThunkSerialization ioHandling taskFnArg = do+  taskFn <- withErrorAction logError ("Could not read task logic: " ++(show taskFnArg)) $ return $ (read taskFnArg :: BL.ByteString)+  logInfo "slave: deserializing task logic"+  logDebug $ "slave: got this task function: " ++ (show taskFn)+  function <- deserializeFunction taskFn :: IO (TaskInput -> TaskResult)+  serializeFunction function >>= \s -> logDebug $ "task deserialization done for: " ++ (show $ BL.unpack s)+  DST.serializedThunkExecution ioHandling function+  +-- Partial binary deployment step 1/2: start distribution of task on master+objectCodeSerializationOnMaster :: IO TaskDef+objectCodeSerializationOnMaster = DOC.loadObjectCode >>= \objectCode -> return $ ObjectCodeModule objectCode

@@ -0,0 +1,54 @@+module Control.Distributed.Task.TaskSpawning.TaskSpawningTypes (+  NominalDiffTime, -- reexport+  IOHandling(..), packIOHandling, unpackIOHandling+  ) where++import Data.Time.Clock (NominalDiffTime)++import Control.Distributed.Task.TaskSpawning.TaskDefinition+import Data.List (intersperse)+import Data.List.Split (splitOn)++-- task communication++data IOHandling = IOHandling [DataDef] ResultDef++packIOHandling :: IOHandling -> String+packIOHandling (IOHandling dataDefs resultDef) = (packResultDef resultDef)++"|"++(concat $ intersperse "|" $ map packDataDef dataDefs)+unpackIOHandling :: String -> IOHandling+unpackIOHandling s = let fields = splitOn "|" s+                     in if length fields < 2+                        then error $ "incompliete io handling: "++s+                        else IOHandling (map unpackDataDef $ tail fields) (unpackResultDef $ head fields)++splitOnExpect :: Int -> String -> String -> [String]+splitOnExpect len delim s = let es = splitOn delim s+                            in if (length es) /= len+                               then error $ "unexpected segment amount: "++s+                               else es++packDataDef :: DataDef -> String+packDataDef (HdfsData p) = "HdfsData:"++p+packDataDef (PseudoDB p) = "PseudoDB:"++p+unpackDataDef :: String -> DataDef+unpackDataDef s = let es = splitOnExpect 2 ":" s+                  in case es !! 0 of+                  "HdfsData" -> HdfsData $ es !! 1+                  "PseudoDB" -> PseudoDB $ es !! 1+                  str -> error $ "unknown data source value: " ++ str++packResultDef :: ResultDef -> String+packResultDef ReturnAsMessage = "ReturnAsMessage"+packResultDef (HdfsResult p s z) = "HdfsResult:"++p++":"++s++":"++show z+packResultDef ReturnOnlyNumResults = "ReturnOnlyNumResults"+unpackResultDef :: String -> ResultDef+unpackResultDef s = let es = splitOn ":" s+                    in if null es+                       then error $ "unknown segment amount"++s+                       else case es !! 0 of+                       "ReturnAsMessage" -> ReturnAsMessage+                       "ReturnOnlyNumResults" -> ReturnOnlyNumResults+                       "HdfsResult" -> if length es /= 4+                                       then error $ "unknown segment amount"++s+                                       else HdfsResult (es !! 1) (es !! 2) (read $ es !! 3)+                       str -> error $ "unknown return action value: " ++ str

@@ -0,0 +1,5 @@+module Control.Distributed.Task.Types.HdfsConfigTypes where++type HdfsConfig = (String, Int)+type HdfsPath = String+type HdfsLocation = (HdfsConfig, HdfsPath)

@@ -0,0 +1,7 @@+module Control.Distributed.Task.Types.TaskTypes where++import qualified Data.ByteString.Lazy as BLC++type TaskInput = [BLC.ByteString]+type TaskResult = [BLC.ByteString]+type Task = TaskInput -> TaskResult

@@ -0,0 +1,65 @@+module Control.Distributed.Task.Util.Configuration (+  Configuration(..), getConfiguration, DistributionStrategy(..)+  )where++import Data.List.Split (splitOn)++import Control.Distributed.Task.Types.HdfsConfigTypes++data Configuration = Configuration {+  _maxTasksPerNode :: Int,+  _hdfsConfig :: HdfsConfig,+  _pseudoDBPath :: FilePath,+  _distributionStrategy :: DistributionStrategy,+  _taskLogFile :: FilePath,+  _sourceCodeDistributionHome :: FilePath,+  _sourceCodeModules :: [(String, Maybe String)],+  _objectCodePathOnMaster :: FilePath,+  _packageDbPath :: FilePath,+  _objectCodeResourcesPathSrc :: FilePath,+  _objectCodeResourcesPathExtra :: FilePath+  }++data DistributionStrategy+  = FirstTaskWithData+  | AnywhereIsFine++getConfiguration :: IO Configuration+getConfiguration = do+  confFile <- readFile "etc/config"+  sourceCodeModulesFile <- readFile "etc/sourcecodemodules"+  return $ parseConfig confFile sourceCodeModulesFile+  where+    parseConfig conf sourceCodeModulesFile =+      Configuration+      (read $ f "max-tasks-per-node")+      (readEndpoint $ f "hdfs")+      (f "pseudo-db-path")+      (readStrat $ f "distribution-strategy")+      (f "task-log-file")+      (f "source-code-distribution-home")+      (map mkSourceCodeModule $ lines sourceCodeModulesFile)+      (f "object-code-path-on-master")+      (f "package-db-path")+      (f "object-code-resources-path-src")+      (f "object-code-resources-path-extra")+      where+        f = getConfig conf+        readStrat s = case s of+                       "local" -> FirstTaskWithData+                       "anywhere" -> AnywhereIsFine+                       _ -> error $ "unknown strategy: "++s+        readEndpoint str = let es = splitOn ":" str+                           in if length es == 2+                              then (es !! 0, read $ es !! 1)+                              else error $ "endpoint not properly configured in etc/config (example: hdfs=localhost:55555): "++str++        mkSourceCodeModule line = let es = splitOn ":" line in if (length es) == 1 then (es !! 0, Nothing) else (es !! 0, Just $ es !! 1)++getConfig :: String -> String -> String+getConfig file key =+  let conf = (filter (not . null . fst) . map parseConfig . map (splitOn "=") . lines) file+  in maybe (error $ "not configured: "++key) id $ lookup key conf+  where+    parseConfig :: [String] -> (String, String)+    parseConfig es = if length es < 2 then ("", "") else (head es, concat $ tail es)

@@ -0,0 +1,17 @@+module Control.Distributed.Task.Util.ErrorHandling (withErrorPrefix, withErrorAction) where++import Control.Exception (catch, SomeException)++withErrorPrefix :: String -> IO a -> IO a+withErrorPrefix = withErrorHandling Nothing++withErrorAction :: (String -> IO ()) -> String -> IO a -> IO a+withErrorAction = withErrorHandling . Just++withErrorHandling :: Maybe (String -> IO ()) -> String -> IO a -> IO a+withErrorHandling errorAction prefix action = action `catch` wrapError+  where+    wrapError :: SomeException -> IO a+    wrapError e = let errorMessage = prefix++": "++(show e) in do+        maybe (return ()) (\eA -> eA errorMessage) errorAction+        error $ errorMessage

@@ -0,0 +1,15 @@+module Control.Distributed.Task.Util.FileUtil where++import Data.List (intersperse)+import Data.List.Split (splitOn)++getFileNamePart :: FilePath -> String+getFileNamePart = snd . splitBasePath+  --let parts = splitOn "/" path in if null parts then "" else parts !! (length parts -1)++splitBasePath :: FilePath -> (FilePath, String)+splitBasePath path = let parts = splitOn "/" path in+  if null parts+  then ("", "")+  else (concat $ intersperse "/" $ take (length parts -1) parts,+        parts !! (length parts -1))

@@ -0,0 +1,35 @@+module Control.Distributed.Task.Util.Logging (logError, logWarn, logInfo, logDebug, logTrace, initLogging) where++import System.Directory (createDirectoryIfMissing)+import System.IO (stdout)+import qualified System.Log.Logger as L+import qualified System.Log.Handler as L (setFormatter)+import qualified System.Log.Handler.Simple as L+import qualified System.Log.Formatter as L++import Control.Distributed.Task.Util.FileUtil++logError, logWarn, logInfo, logDebug, logTrace :: String -> IO ()+logError = simpleLog L.errorM+logWarn = simpleLog L.warningM+logInfo = simpleLog L.infoM+-- as hslogger logging does not seem to be that performant when there is nothing is to log, debugLogging is "configured" here to a hard off+logDebug _ = return () --simpleLog L.debugM+logTrace _ = return () --simpleLog L.debugM++simpleLog :: (String -> String -> IO ()) -> String -> IO ()+simpleLog levelLogger = levelLogger L.rootLoggerName++initLogging :: L.Priority -> L.Priority -> FilePath -> IO ()+initLogging stdoutLogLevel fileLogLevel logfile = do+  L.updateGlobalLogger L.rootLoggerName (L.removeHandler)+  L.updateGlobalLogger L.rootLoggerName (L.setLevel $ max' stdoutLogLevel fileLogLevel)+  createDirectoryIfMissing True $ fst $ splitBasePath logfile+  addHandler' $ L.fileHandler logfile fileLogLevel+  addHandler' $ L.streamHandler stdout stdoutLogLevel+  where+    max' a b = if fromEnum a <= fromEnum b then a else b+    addHandler' logHandlerM = do+      logHandler <- logHandlerM+      h <- return $ L.setFormatter logHandler (L.simpleLogFormatter "[$time : $loggername : $prio] $msg")+      L.updateGlobalLogger L.rootLoggerName (L.addHandler h)

@@ -0,0 +1,19 @@+module Control.Distributed.Task.Util.SerializationUtil where++import Data.Time.Calendar (Day(..))+import Data.Time.Clock (UTCTime(..), NominalDiffTime)++type TimeStamp = (Integer, Rational)++serializeTime :: UTCTime -> TimeStamp+serializeTime (UTCTime (ModifiedJulianDay d) f) = (d, toRational f)++deserializeTime :: TimeStamp -> UTCTime+deserializeTime (d, f) = UTCTime (ModifiedJulianDay d) (fromRational f)++serializeTimeDiff :: NominalDiffTime -> Rational+serializeTimeDiff = toRational++deserializeTimeDiff :: Rational -> NominalDiffTime+deserializeTimeDiff = fromRational+

@@ -0,0 +1,188 @@+name:                task-distribution+version:             0.1.0.0+synopsis:            Initial project template from stack+description:         Please see README.md+homepage:            http://github.com/michaxm/task-distribution#readme+license:             BSD3+license-file:        LICENSE+author:              Axel Mannhardt+maintainer:          7a3ggptwts@snkmail.com+copyright:           2016 Axel Mannhardt+category:            Distributed Computing+build-type:          Simple+-- extra-source-files:+cabal-version:       >=1.18+++library+  hs-source-dirs:      src+  exposed-modules:     Control.Distributed.Task.Types.TaskTypes+                     , Control.Distributed.Task.Types.HdfsConfigTypes+                     , Control.Distributed.Task.Distribution.RunComputation+                     , Control.Distributed.Task.Distribution.TaskDistribution+                     , Control.Distributed.Task.Distribution.LogConfiguration+                     , Control.Distributed.Task.TaskSpawning.RemoteExecutionSupport+  other-modules:       Control.Distributed.Task.Distribution.TaskTransport+                     , Control.Distributed.Task.Distribution.DataLocality+                     , Control.Distributed.Task.TaskSpawning.TaskSpawning+                     , Control.Distributed.Task.TaskSpawning.TaskSpawningTypes+                     , Control.Distributed.Task.TaskSpawning.TaskDefinition+                     , Control.Distributed.Task.TaskSpawning.TaskDescription+                     , Control.Distributed.Task.TaskSpawning.ExecutionUtil+                     , Control.Distributed.Task.TaskSpawning.BinaryStorage+                     , Control.Distributed.Task.TaskSpawning.SourceCodeExecution+                     , Control.Distributed.Task.TaskSpawning.DeployFullBinary+                     , Control.Distributed.Task.TaskSpawning.DeploySerializedThunk+                     , Control.Distributed.Task.TaskSpawning.FunctionSerialization+                     , Control.Distributed.Task.TaskSpawning.DeployObjectCodeRelinked+                     , Control.Distributed.Task.DataAccess.DataSource+                     , Control.Distributed.Task.DataAccess.SimpleDataSource+                     , Control.Distributed.Task.DataAccess.HdfsDataSource+                     , Control.Distributed.Task.DataAccess.HdfsListing+                     , Control.Distributed.Task.Util.Configuration+                     , Control.Distributed.Task.Util.ErrorHandling+                     , Control.Distributed.Task.Util.FileUtil+                     , Control.Distributed.Task.Util.Logging+                     , Control.Distributed.Task.Util.SerializationUtil+  ghc-options:         -Wall -rtsopts -with-rtsopts=-N+  build-depends:       base >= 4.7 && < 5+                     , hint+                     , binary+                     , temporary, filepath, directory+                     , transformers+                     , bytestring+                     , packman+                     , process+                     , distributed-process+                     , distributed-process-simplelocalnet+                       -- for explicit bindings (instead of Template Haskell)+                     , rank1dynamic+                     , distributed-static+                     , text+                     , split+                     , hslogger+                     , time+                     , hashable+                     , zlib+                     , hadoop-rpc+                     , vector+                     , containers+                     , async+  default-language:    Haskell2010++executable slave+  hs-source-dirs:      app+  main-is:             Slave.hs+  other-modules:       +  ghc-options:         -threaded -rtsopts -with-rtsopts=-N -Wall+  build-depends:       base+                     , task-distribution+                     , hslogger+                     , containers+                     , json+                     , strings+                     , bytestring+  default-language:    Haskell2010++executable example+  hs-source-dirs:      app+  main-is:             Main.hs+  other-modules:       RemoteExecutable+                     , FullBinaryExamples+                     , DemoTask+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N -Wall+  build-depends:       base+                     , split+                     , task-distribution+                     , hslogger+                     , containers+                     , json+                     , strings+                     , bytestring+  default-language:    Haskell2010++executable task-distribution-object-code-remote+  hs-source-dirs:      object-code-app, src+  main-is:             RemoteExecutor.hs+  other-modules:       RemoteExecutable+                     , Control.Distributed.Task.Types.TaskTypes+                     , Control.Distributed.Task.TaskSpawning.DeployFullBinary+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , bytestring+                     , hslogger+                     , split+                     , directory+                     , time+                     , binary+                     , process+                     , temporary+                     , filepath+                     , distributed-process+                     , async+                     , hadoop-rpc+                     , vector+                     , text+                     , zlib+  default-language:    Haskell2010++executable run-demo-task+  hs-source-dirs:      app+  main-is:             RunDemoTask.hs+  other-modules:       DemoTask+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N -auto-all -caf-all+  build-depends:       base+                     , bytestring+                     , zlib+                     , split+                     , async+  default-language:    Haskell2010++executable test-visit-performance+  hs-source-dirs:      app+  main-is:             TestVisitCalculation.hs+  other-modules:       RemoteExecutable+                     , VisitCalculation+  ghc-options:         -threaded -rtsopts -with-rtsopts=-N -Wall+  build-depends:       base+                     , split+                     , containers+                     , json+                     , bytestring+  default-language:    Haskell2010++test-suite test-node-matching+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+                     , src+  main-is:             NodeMatchingTests.hs+  other-modules:       Control.Distributed.Task.Types.TaskTypes+                     , Control.Distributed.Task.Util.Configuration+                     , Control.Distributed.Task.Types.HdfsConfigTypes+                     , Control.Distributed.Task.Distribution.DataLocality+                     , Control.Distributed.Task.Util.Logging+  build-depends:       base+                     , hspec+                     , task-distribution+                     , hint+                     , binary+                     , temporary, filepath, directory+                     , transformers+                     , bytestring+                     , packman+                     , process+                     , distributed-process+                     , distributed-process-simplelocalnet+                     , rank1dynamic+                     , distributed-static+                     , text+                     , hadoop-rpc+                     , vector+                     , split+                     , hslogger+  ghc-options:         -O0+  default-language:    Haskell2010++source-repository head+  type:     git+  location: git@github.com:michaxm/task-distribution.git

@@ -0,0 +1,21 @@+import Data.List.Split (splitOn)+import Test.Hspec++import Control.Distributed.Task.Distribution.DataLocality (nodeMatcher)++main :: IO ()+main = hspec $ do+  describe "ClusterComputing.DataLocality" $ do+        context "Node/Host matching" $ do+            it "does not match everything" $ do+              nodeMatcher [] "nid://node:44441:0" "host" `shouldBe` False+            it "extracts host from nodeId string" $ do+              (dropWhile (=='/') . head . drop 1 . splitOn ":") "nid://localhost:44441:0" `shouldBe` "localhost"+            it "extracts host from host string" $ do+              (head . splitOn ":") "localhost:50010" `shouldBe` "localhost"+            it "matches same host" $ do+              nodeMatcher [] "nid://localhost:44441:0" "localhost:50010" `shouldBe` True+            it "matches special hdfs host" $ do+              nodeMatcher [("127.0.0.1", "localhost")] "nid://localhost:44441:0" "127.0.0.1:50010" `shouldBe` True+            it "matches special node host" $ do+              nodeMatcher [("127.0.0.1", "localhost")] "nid://127.0.0.1:44441:0" "localhost:50010" `shouldBe` True