diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -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.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/app/DemoTask.hs b/app/DemoTask.hs
new file mode 100644
--- /dev/null
+++ b/app/DemoTask.hs
@@ -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)
diff --git a/app/FullBinaryExamples.hs b/app/FullBinaryExamples.hs
new file mode 100644
--- /dev/null
+++ b/app/FullBinaryExamples.hs
@@ -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 )
diff --git a/app/Main.hs b/app/Main.hs
new file mode 100644
--- /dev/null
+++ b/app/Main.hs
@@ -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
diff --git a/app/RemoteExecutable.hs b/app/RemoteExecutable.hs
new file mode 100644
--- /dev/null
+++ b/app/RemoteExecutable.hs
@@ -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)
diff --git a/app/RunDemoTask.hs b/app/RunDemoTask.hs
new file mode 100644
--- /dev/null
+++ b/app/RunDemoTask.hs
@@ -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
diff --git a/app/Slave.hs b/app/Slave.hs
new file mode 100644
--- /dev/null
+++ b/app/Slave.hs
@@ -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>"
diff --git a/app/TestVisitCalculation.hs b/app/TestVisitCalculation.hs
new file mode 100644
--- /dev/null
+++ b/app/TestVisitCalculation.hs
@@ -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
diff --git a/app/VisitCalculation.hs b/app/VisitCalculation.hs
new file mode 100644
--- /dev/null
+++ b/app/VisitCalculation.hs
@@ -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)
diff --git a/object-code-app/RemoteExecutable.hs b/object-code-app/RemoteExecutable.hs
new file mode 100644
--- /dev/null
+++ b/object-code-app/RemoteExecutable.hs
@@ -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"
diff --git a/object-code-app/RemoteExecutor.hs b/object-code-app/RemoteExecutor.hs
new file mode 100644
--- /dev/null
+++ b/object-code-app/RemoteExecutor.hs
@@ -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
diff --git a/src/Control/Distributed/Task/DataAccess/DataSource.hs b/src/Control/Distributed/Task/DataAccess/DataSource.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/DataAccess/DataSource.hs
@@ -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
diff --git a/src/Control/Distributed/Task/DataAccess/HdfsDataSource.hs b/src/Control/Distributed/Task/DataAccess/HdfsDataSource.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/DataAccess/HdfsDataSource.hs
@@ -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"
diff --git a/src/Control/Distributed/Task/DataAccess/HdfsListing.hs b/src/Control/Distributed/Task/DataAccess/HdfsListing.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/DataAccess/HdfsListing.hs
@@ -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
diff --git a/src/Control/Distributed/Task/DataAccess/SimpleDataSource.hs b/src/Control/Distributed/Task/DataAccess/SimpleDataSource.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/DataAccess/SimpleDataSource.hs
@@ -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
diff --git a/src/Control/Distributed/Task/Distribution/DataLocality.hs b/src/Control/Distributed/Task/Distribution/DataLocality.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Distribution/DataLocality.hs
@@ -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))
diff --git a/src/Control/Distributed/Task/Distribution/LogConfiguration.hs b/src/Control/Distributed/Task/Distribution/LogConfiguration.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Distribution/LogConfiguration.hs
@@ -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")
diff --git a/src/Control/Distributed/Task/Distribution/RunComputation.hs b/src/Control/Distributed/Task/Distribution/RunComputation.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Distribution/RunComputation.hs
@@ -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)
diff --git a/src/Control/Distributed/Task/Distribution/TaskDistribution.hs b/src/Control/Distributed/Task/Distribution/TaskDistribution.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Distribution/TaskDistribution.hs
@@ -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)
diff --git a/src/Control/Distributed/Task/Distribution/TaskTransport.hs b/src/Control/Distributed/Task/Distribution/TaskTransport.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Distribution/TaskTransport.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/BinaryStorage.hs b/src/Control/Distributed/Task/TaskSpawning/BinaryStorage.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/BinaryStorage.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/DeployFullBinary.hs b/src/Control/Distributed/Task/TaskSpawning/DeployFullBinary.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/DeployFullBinary.hs
@@ -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]
diff --git a/src/Control/Distributed/Task/TaskSpawning/DeployObjectCodeRelinked.hs b/src/Control/Distributed/Task/TaskSpawning/DeployObjectCodeRelinked.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/DeployObjectCodeRelinked.hs
@@ -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")
diff --git a/src/Control/Distributed/Task/TaskSpawning/DeploySerializedThunk.hs b/src/Control/Distributed/Task/TaskSpawning/DeploySerializedThunk.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/DeploySerializedThunk.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/ExecutionUtil.hs b/src/Control/Distributed/Task/TaskSpawning/ExecutionUtil.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/ExecutionUtil.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/FunctionSerialization.hs b/src/Control/Distributed/Task/TaskSpawning/FunctionSerialization.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/FunctionSerialization.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs b/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/RemoteExecutionSupport.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/SourceCodeExecution.hs b/src/Control/Distributed/Task/TaskSpawning/SourceCodeExecution.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/SourceCodeExecution.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/TaskDefinition.hs b/src/Control/Distributed/Task/TaskSpawning/TaskDefinition.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/TaskDefinition.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/TaskDescription.hs b/src/Control/Distributed/Task/TaskSpawning/TaskDescription.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/TaskDescription.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/TaskSpawning.hs b/src/Control/Distributed/Task/TaskSpawning/TaskSpawning.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/TaskSpawning.hs
@@ -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
diff --git a/src/Control/Distributed/Task/TaskSpawning/TaskSpawningTypes.hs b/src/Control/Distributed/Task/TaskSpawning/TaskSpawningTypes.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/TaskSpawning/TaskSpawningTypes.hs
@@ -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
diff --git a/src/Control/Distributed/Task/Types/HdfsConfigTypes.hs b/src/Control/Distributed/Task/Types/HdfsConfigTypes.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Types/HdfsConfigTypes.hs
@@ -0,0 +1,5 @@
+module Control.Distributed.Task.Types.HdfsConfigTypes where
+
+type HdfsConfig = (String, Int)
+type HdfsPath = String
+type HdfsLocation = (HdfsConfig, HdfsPath)
diff --git a/src/Control/Distributed/Task/Types/TaskTypes.hs b/src/Control/Distributed/Task/Types/TaskTypes.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Types/TaskTypes.hs
@@ -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
diff --git a/src/Control/Distributed/Task/Util/Configuration.hs b/src/Control/Distributed/Task/Util/Configuration.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Util/Configuration.hs
@@ -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)
diff --git a/src/Control/Distributed/Task/Util/ErrorHandling.hs b/src/Control/Distributed/Task/Util/ErrorHandling.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Util/ErrorHandling.hs
@@ -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
diff --git a/src/Control/Distributed/Task/Util/FileUtil.hs b/src/Control/Distributed/Task/Util/FileUtil.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Util/FileUtil.hs
@@ -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))
diff --git a/src/Control/Distributed/Task/Util/Logging.hs b/src/Control/Distributed/Task/Util/Logging.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Util/Logging.hs
@@ -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)
diff --git a/src/Control/Distributed/Task/Util/SerializationUtil.hs b/src/Control/Distributed/Task/Util/SerializationUtil.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Task/Util/SerializationUtil.hs
@@ -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
+
diff --git a/task-distribution.cabal b/task-distribution.cabal
new file mode 100644
--- /dev/null
+++ b/task-distribution.cabal
@@ -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
diff --git a/test/NodeMatchingTests.hs b/test/NodeMatchingTests.hs
new file mode 100644
--- /dev/null
+++ b/test/NodeMatchingTests.hs
@@ -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