Holumbus-MapReduce 0.0.1 → 0.1.0

raw patch · 22 files changed

+1188/−82 lines, 22 filesdep +extensible-exceptionsdep +paralleldep +readlinedep ~Holumbus-Distributiondep ~Holumbus-Storagedep ~basenew-component:exe:MasterDaemonPVP ok

version bump matches the API change (PVP)

Dependencies added: extensible-exceptions, parallel, readline

Dependency ranges changed: Holumbus-Distribution, Holumbus-Storage, base

API changes (from Hackage documentation)

Files

• Examples/MapReduce/Crawler/Crawl.hs +1/−0
• Examples/MapReduce/Crawler/Index.hs +1/−0
• Examples/MapReduce/Grep/Grep.hs +1/−0
• Examples/MapReduce/Sort/Sort.hs +1/−0
• Examples/MapReduce/WordFrequency/WordFrequency.hs +2/−1
• Holumbus-MapReduce.cabal +18/−6
• Programs/MapReduceMasterDaemon/MasterDaemon.hs +85/−0
• README +1/−1
• source/Holumbus/Distribution/DMapReduce.hs +21/−0
• source/Holumbus/Distribution/Master/MasterData.hs +28/−7
• source/Holumbus/Distribution/Master/MasterPort.hs +3/−1
• source/Holumbus/Distribution/Master/MasterState.hs +96/−0
• source/Holumbus/Distribution/Messages.hs +1/−0
• source/Holumbus/Distribution/SimpleDMapReduceIO.hs +263/−0
• source/Holumbus/Distribution/Worker/WorkerData.hs +17/−7
• source/Holumbus/Distribution/Worker/WorkerPort.hs +3/−1
• source/Holumbus/MapReduce/DaemonInterface.hs +148/−0
• source/Holumbus/MapReduce/DaemonInterfaceWithFS.hs +301/−0
• source/Holumbus/MapReduce/JobController.hs +3/−4
• source/Holumbus/MapReduce/TaskProcessor.hs +7/−5
• source/Holumbus/MapReduce/Types.hs +169/−49
• source/Holumbus/Standalone/SMapReduce.hs +18/−0

@@ -41,6 +41,7 @@ import           Control.Monad hiding (when)  import           Data.Binary+--import           Holumbus.Common.MRBinary import           Data.Char import           Data.List import           Data.Maybe

@@ -38,6 +38,7 @@  import           Data.List import           Data.Binary hiding (Word)+--import           Holumbus.Common.MRBinary import qualified Data.Map    as M import qualified Data.IntMap as IM import qualified Data.IntSet as IS

@@ -24,6 +24,7 @@ where  import           Data.Binary+--import           Holumbus.Common.MRBinary import           Data.Maybe  import           System.Log.Logger

@@ -24,6 +24,7 @@ where  import           Data.Binary+--import           Holumbus.Common.MRBinary  import           System.Log.Logger 

@@ -24,6 +24,7 @@ where  import           Data.Binary+--import           Holumbus.Common.MRBinary  import           System.Log.Logger @@ -127,7 +128,7 @@      textList :: [(String,String)] textList-  = [("text1", "aaa bb c dd dd"),("text2", "aaa bb"),("text2", "aaa dd dd")]+  = [("text1", "aaa bb c dd dd"),("text2", "aaa bb"),("text3", "aaa dd dd")]   -- ----------------------------------------------------------------------------

@@ -1,9 +1,9 @@ name:          Holumbus-MapReduce-version:       0.0.1+version:       0.1.0 license:       OtherLicense license-file:  LICENSE author:        Uwe Schmidt, Stefan Schmidt-copyright:     Copyright (c) 2008 Uwe Schmidt, Stefan Schmidt+copyright:     Copyright (c) 2010 Stefan Schmidt, Uwe Schmidt, Sebastian Reese maintainer:    Stefan Schmidt <sts@holumbus.org> stability:     experimental category:      Distributed Computing@@ -48,7 +48,7 @@     Examples/Utils/PickleTest/PickleTest.hs  library-  build-depends:  base >= 3+  build-depends:  base >= 4 && < 5                 , binary     >= 0.4                 , bytestring >= 0.9                 , containers >= 0.1@@ -59,13 +59,17 @@                 , network    >= 2.1                 , time       >= 1.1                 , unix       >= 2.3-                , Holumbus-Distribution >= 0.0.1-                , Holumbus-Storage >= 0.0.1+                , parallel   >= 1.1.0.0+                , Holumbus-Distribution >= 0.0.1 && < 0.2+                , Holumbus-Storage >= 0.0.1 && < 0.2+                , extensible-exceptions+                , readline   >= 1.0    hs-source-dirs:                    source   exposed-modules:                     Holumbus.Distribution.DMapReduce+                , Holumbus.Distribution.SimpleDMapReduceIO                 , Holumbus.Distribution.Messages                 , Holumbus.Distribution.Worker                 , Holumbus.Distribution.Worker.WorkerData@@ -73,13 +77,16 @@                 , Holumbus.Distribution.Master                 , Holumbus.Distribution.Master.MasterData                 , Holumbus.Distribution.Master.MasterPort+                , Holumbus.Distribution.Master.MasterState                 , Holumbus.Standalone.SMapReduce                 , Holumbus.MapReduce.MapReduce                 , Holumbus.MapReduce.JobController                 , Holumbus.MapReduce.TaskProcessor                 , Holumbus.MapReduce.Types                 , Holumbus.MapReduce.UserInterface-  ghc-options: -Wall -threaded+                , Holumbus.MapReduce.DaemonInterface+                , Holumbus.MapReduce.DaemonInterfaceWithFS+  ghc-options: -Wall   extensions: Arrows DeriveDataTypeable ExistentialQuantification  executable Master@@ -87,3 +94,8 @@   hs-source-dirs: Programs/MapReduceMaster source   ghc-options: -Wall -threaded   extensions: Arrows DeriveDataTypeable ExistentialQuantification++executable MasterDaemon+  main-is: MasterDaemon.hs+  hs-source-dirs: Programs/MapReduceMasterDaemon source+  ghc-options: -Wall -threaded

@@ -0,0 +1,85 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : MasterDaemon+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1++  Daemonized Version of MASter with a tcp console+-}+-- ----------------------------------------------------------------------------++module Main(main) where++import           Holumbus.Common.Logging+import           Holumbus.Common.Utils                         ( handleAll )++import           Holumbus.Network.PortRegistry.PortRegistryPort+import qualified Holumbus.FileSystem.FileSystem                 as FS+import qualified Holumbus.Distribution.DMapReduce               as MR+import           Holumbus.Distribution.Master.MasterState+import qualified Holumbus.MapReduce.DaemonInterfaceWithFS       as UI+import           System.Environment+import           System.Log.Logger+import           System.Exit+++version :: String+version = "Daemon-Master 0.1"++prompt :: String+prompt = "# "++version++" > "++localLogger :: String+localLogger = "MasterDaemon"++pUsage :: IO ()+pUsage = do+  putStrLn "Usage: MasterDaemon ConsolePort Logfile Statefile"++params :: IO [String]+params = do+  args <- getArgs+  if length args /= 3 then do+    errorM localLogger "Wrong argument count"+    pUsage+    exitFailure+    else+      return args+      +main :: IO ()+main+  = do+    putStrLn version+    handleAll (\e -> errorM localLogger $ "EXCEPTION: " ++ show e) $+      do      +      (s_cport:logfile:statefile:[]) <- params+      initializeFileLogging logfile [(localLogger, INFO),("Holumbus.Network.DoWithServer",INFO),("Holumbus",ERROR),("measure",ERROR),("Holumbus.MapReduce.JobController.cycle",ERROR)]+      addResource statefile+      p <- newPortRegistryFromXmlFile "/tmp/registry.xml"      +      setPortRegistry p+      (mr,fs) <- initializeData+      UI.runDaemon (mr,fs) version (read s_cport) prompt +      deinitializeData (mr,fs)+++initializeData :: IO (MR.DMapReduce, FS.FileSystem)+initializeData +  = do    +    fs <- FS.mkFileSystemController FS.defaultFSControllerConfig+    let config  = MR.defaultMRMasterConfig+    mr <- MR.mkMapReduceMaster fs config+    return (mr,fs)+++deinitializeData :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitializeData (mr,fs)+  = do+    MR.closeMapReduce mr+    FS.closeFileSystem fs++-- ----------------------------------------------------------------------------

@@ -1,6 +1,6 @@ This is the Holumbus-MapReduce Framework -Version 0.0.1+Version 0.1.0  Stefan Schmidt sts@holumbus.org 

@@ -188,6 +188,27 @@         maybe (putStrLn "NOTHING") (\w' -> printDebug w') w         putStrLn "--------------------------------------------------------" +  getDebug (DMapReduce mr)+    = withMVar mr $+        \(DMapReduceData s t m w) ->+        do+        let line = "--------------------------------------------------------"+        tmp <- getDebug m+        mtmp <- maybe (return "NOTHING") (\w' -> getDebug w') w+        return (line+          ++"\n"++ "Distribtion - internal data\n"+          ++"\n"++line+          ++"\n"++ "SiteId:"+          ++"\n"++ show s+          ++"\n"++ "Type:"+          ++"\n"++ show t+          ++"\n"++line+          ++"\n"++ "Master:"+          ++"\n"++tmp+          ++"\n"++line+          ++"\n"++"Worker:"+          ++mtmp+          ++"\n"++line++"\n")   

@@ -28,7 +28,6 @@ import           Control.Concurrent  import           Data.List-import           Data.Maybe import qualified Data.Set                                       as Set  import           System.Log.Logger@@ -40,7 +39,7 @@ -- import           Holumbus.Network.Port -- import           Holumbus.Network.Messages import           Holumbus.Network.Communication-+import           Holumbus.Distribution.Master.MasterState import qualified Holumbus.MapReduce.MapReduce                   as MR import           Holumbus.MapReduce.JobController import           Holumbus.MapReduce.Types@@ -139,11 +138,11 @@     case msg of       (M.MReqTaskCompleted td) ->         do-        MC.receiveTaskCompleted td md+        _ <- MC.receiveTaskCompleted td md         return $ Just $ M.MRspSuccess       (M.MReqTaskError td) ->         do-        MC.receiveTaskError td md+        _ <- MC.receiveTaskError td md         return $ Just $ M.MRspSuccess       (M.MReqStartControlling) ->         do@@ -155,7 +154,7 @@         return $ Just $ M.MRspSuccess                (M.MReqIsControlling) ->         do-        MR.isControlling md+        _ <- MR.isControlling md         return $ Just $ M.MRspSuccess                (M.MReqSingleStep) ->         do@@ -175,6 +174,7 @@     let wp = WP.newWorkerPort cp     as <- WC.getActionNames wp     md <- readMVar m+    addWorker i (as,wp)     modifyMVar (md_WorkerController md) $       \wcd ->       do@@ -188,6 +188,7 @@ unregisterWorker m i _   = do     md <- readMVar m+    delWorker i     modifyMVar (md_WorkerController md) $       \wcd ->       do@@ -286,7 +287,7 @@             case info of               (Just ci) ->                 do-                WC.startTask td $ WP.newWorkerPort (ci_Port ci)+                _ <- WC.startTask td $ WP.newWorkerPort (ci_Port ci)                 let wcd' = addTaskToWorker (td_TaskId td) wid wcd                 return (wcd', TSRSend)                               (Nothing) ->@@ -388,4 +389,24 @@         putStrLn jc       where         gap = 20-      +  getDebug md+    = do+      mddebug <- getDebug (md_Server md)+      tmp <- (withMVar (md_WorkerController md) $+          \wcd -> +          do+          jc <- printJobController (md_JobController md)+          return (+            prettyRecordLine gap "TaskToWorkerMap: " (wcd_TaskToWorkerMap wcd)+            ++"\n"++ prettyRecordLine gap "WorkerToTaskMap" (wcd_WorkerToTaskMap wcd)+            ++"\n"++ prettyRecordLine gap "ActionToWorkerMap:" (wcd_ActionToWorkerMap wcd)        +            ++"\n"++ "JobController:"+            ++"\n"++  jc ++"\n"))+      return (+                 "Master-Object (full)"+        ++"\n"++ "--------------------------------------------------------"        +        ++"\n"++  "Server"+        ++ mddebug+        ++ tmp)+      where+        gap = 20      

@@ -69,7 +69,9 @@     = do       putStrLn "MasterPort:"       putStrLn $ show p-      + +  getDebug (MasterPort p)+    = return ("MasterPort:\n"++show p++"\n")              

@@ -0,0 +1,96 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.Distribution.Master.MasterState+  Copyright  : Copyright (C) 2010 Sebastian Reese+  License    : MIT++  Maintainer : Stefan Schmidt (stefanschmidt@web.de)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.Distribution.Master.MasterState +(+    State+  , States+  , StateResource+  , addResource+  , delResource+  , addWorker+  , delWorker+)where++import qualified Data.List as List+import qualified Data.Map                                as Map+import           System.IO.Unsafe+import           Control.Concurrent.MVar+import           Holumbus.MapReduce.Types+import           Holumbus.Common.FileHandling+import qualified Holumbus.Distribution.Worker.WorkerPort as WP+import           Holumbus.Network.Communication+import           Holumbus.Network.Port++type States = Map.Map IdType State+type State = ([ActionName], WP.WorkerPort)+type StateResource = FilePath++{-# NOINLINE resources #-}+resources :: MVar [StateResource]+resources = unsafePerformIO $ newMVar ([])++addResource :: FilePath -> IO ()+addResource fp = do+  modifyMVar_ resources (return . (fp:))+  updateStats++delResource :: FilePath -> IO ()+delResource fp = do+  modifyMVar_ resources (return . List.delete fp)+  updateStats                      ++{-# NOINLINE states #-}+states :: MVar States+states = unsafePerformIO $ newMVar (Map.empty)++addWorker :: IdType -> State -> IO ()+addWorker i state = do+  modifyMVar_ states (return . Map.insert i state)+  updateStats++delWorker :: IdType -> IO ()+delWorker i = do+  modifyMVar_ states (return . Map.delete i)+  updateStats++updateStats :: IO ()+updateStats = do+  resources' <- readMVar resources+  states'    <- readMVar states+  clearResources resources'+  putState resources' states'+    +putState :: [StateResource] -> States -> IO ()+putState ress = mapM_ ( flip printState ress ) . Map.toList+    +printState :: (IdType,State) -> [StateResource] -> IO ()+printState (i,(actions, wp)) = mapM_ (flip appendToTextFile (mkLine i wp actions))++clearResources :: [StateResource] -> IO ()+clearResources = mapM_ (flip writeToTextFile $"")++mkLine :: IdType -> WP.WorkerPort -> [ActionName] -> String+mkLine i wp actions = (show i)++"\t"++(showActions actions)++"\t"++(showWorkerPort wp)++ "\n"++showActions :: [ActionName] -> String+showActions [] = ""+showActions (x:[]) = x+showActions (x:xs) = x ++ " / " ++ showActions xs++showWorkerPort :: WP.WorkerPort -> String+showWorkerPort (WP.WorkerPort (ClientPort (Port _ m))) = case m of+  Nothing                     -> "no socket"+  (Just (SocketId host port)) -> host ++ ":" ++ show port

@@ -33,6 +33,7 @@   import           Data.Binary+--import           Holumbus.Common.MRBinary  import           Holumbus.Network.Messages import           Holumbus.MapReduce.Types

@@ -0,0 +1,263 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Examples2.SimpleMR.SimpleDMapReduce+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+-}+-- ----------------------------------------------------------------------------+module Holumbus.Distribution.SimpleDMapReduceIO+(+   MapFunction+ , ReduceFunction+ , client+ , worker+ , partition'+ , Priority(..)+ , putTimeStamp+)+where++import           Holumbus.Network.PortRegistry.PortRegistryPort+import           Holumbus.MapReduce.Types+import           Holumbus.Common.FileHandling+import           Data.Binary+--import           Holumbus.Common.MRBinary+import           Data.Maybe+import           Control.Parallel.Strategies+import qualified Holumbus.Distribution.DMapReduce               as MR+import qualified Holumbus.FileSystem.FileSystem                 as FS+import qualified Holumbus.MapReduce.DaemonInterface             as DI+import qualified Holumbus.Data.KeyMap                           as KMap+import           Holumbus.Common.Logging+import           Holumbus.Common.Utils ( handleAll )+import           System.Log.Logger+import           System.Environment+import           System.Exit+import           Data.Time.Clock.POSIX+import qualified Data.ByteString.Lazy as B++{-+splitConfiguration+  :: (Hash k1, NFData v1, NFData k1, Binary a, Binary k1, Binary v1)+  => SplitConfiguration a k1 v1+splitConfiguration+  = SplitConfiguration+      defaultSplit+      defaultInputReader+      defaultOutputWriter+-}++mapConfiguration+  :: (Hash k2, NFData v1, NFData k1, NFData v2, NFData k2, Ord k2, Binary a, Binary k1, Binary v1, Binary k2, Binary v2)+  => MapFunction a k1 v1 k2 v2+  -> MapConfiguration a k1 v1 k2 v2+mapConfiguration fct+  = MapConfiguration+      fct+      hashedPartition+      defaultInputReader+      defaultOutputWriter++reduceConfiguration+  :: (Hash k2, NFData v2, NFData k2, NFData v3, Ord k2, Binary a, Binary k2, Binary v2, Binary v3)+  => ReduceFunction a k2 v2 v3+  -> ReduceConfiguration a k2 v2 v3+reduceConfiguration fct+  = ReduceConfiguration+      defaultMerge+      fct+      hashedPartition+      defaultInputReader+      defaultOutputWriter+++{-+actionConfig+-}+actionConfig :: (Hash k1, Hash k2, Binary a, NFData k1, NFData k2, Ord k2, Binary k1, Binary k2, NFData v1, NFData v4, NFData v2, NFData v3, Binary v1, Binary v3, Binary v2, Binary v4) => MapFunction a k1 v1 k2 v2 -> ReduceFunction a k2 v3 v4 -> ActionConfiguration a k1 v1 k2 v2 v3 v4+actionConfig m r = (defaultActionConfiguration "ID") {+           ac_Split   = Nothing -- Just splitConfiguration  +         , ac_Map     = Just . mapConfiguration    $ m+         , ac_Reduce  = Just . reduceConfiguration $ r+         }++{- ---------------------------------------------------------------------------------------------------------+   The simple client fucntions+  --------------------------------------------------------------------------------------------------------- -}   +  +  +{-+ The simple client+ + doMapReduce :: ActionConfiguration a k1 v1 k2 v2 v3 v4+  -> a               -- ^ options+  -> [(k1,v1)]       -- ^ input (Tuples)+  -> [FileId]        -- ^ input (Files)+  -> Int             -- ^ number of splitters+  -> Int             -- ^ number of mappers+  -> Int             -- ^ number of reducers+  -> Int             -- ^ number of results+  -> TaskOutputType  -- ^ type of the result (file of raw)+  -> mr -> IO ([(k2,v4)],[FileId])+-}+client :: ( Show k1, Show k2, Show v1, Show v2, Show v3, Show v4+          , Binary v1, Binary v3, Binary v2, Binary v4+          , Binary a, Binary k1, Binary k2+          , NFData k1, NFData k2, NFData v1, NFData v4, NFData v2, NFData v3+          , Ord k2, Hash k1, Hash k2) =>+          MapFunction a k1 v1 k2 v2 -> ReduceFunction a k2 v3 v4+          -> a -> (Int,Int,Int) -> [[(k1,v1)]] -> IO [(k2,v4)]+client m r a (splitters,mappers,reducers) lss = do+      -- create port registry+      p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+      setPortRegistry p      +      +      -- make filesystem+      fs <- FS.mkFileSystemClient FS.defaultFSClientConfig+      siteid <- FS.getMySiteId fs+      putStrLn $ "My FS Siteid is: " ++ show siteid+      +      -- create mapreduce data+      mr <- initializeData+      +      --_ <- readline "Press enter to continue .."+      -- create the filenames and store the data to the map reduce filesystem+--      let filenames = map (\i -> "initial_input_"++show i) [1..(length lss)]+--      FS.createFiles (zipWith (\fn c -> (fn,listToByteString c)) filenames lss) fs+      let (files,filenames) = prepareFiles lss 0+      FS.createFiles files fs+      -- mapM_ (\(filename,ls) -> FS.createFile filename (listToByteString ls) fs) $ zip filenames lss+      +      -- do the map reduce job+      putTimeStamp "SimpleDMR Begin MR"+      (_,fids) <- MR.doMapReduce (actionConfig m r) a [] filenames splitters mappers reducers 1 TOTFile mr+      putTimeStamp "SimpleDMR End MR"+      +      -- get the results from filesystem+      result <- merge fids fs+      +      -- close file- and mapreducesystem+      deinitializeData mr+      FS.closeFileSystem fs+      +      -- finally, return the result+      return result++prepareFiles :: Binary a => [[a]] -> Int -> ([(String,B.ByteString)],[String])+prepareFiles []     _ = ([],[])+prepareFiles (x:xs) i = ((fn,bin):files,fn:filenames)+  where+  fn = ("Initial_input_"++show i)+  bin = listToByteString x+  (files,filenames) = prepareFiles xs (i+1)+      +merge :: (Show k2, Show v4, Hash k2, Binary k2, Binary v4, NFData k2, NFData v4) => [FS.FileId] -> FS.FileSystem -> IO [(k2,v4)]+merge fids fs = do+   mayberesult <- mapM ( flip FS.getFileContent fs) fids+   let result = concat . map parseByteStringToList $ catMaybes mayberesult+   return result+   ++{-+The simple client's init function+-}+initializeData :: IO (MR.DMapReduce)+initializeData +  = do    +    let config = MR.defaultMRClientConfig+    MR.mkMapReduceClient config++{-+The simple client's deinit function+-}+deinitializeData :: MR.DMapReduce -> IO ()+deinitializeData mr+  = do+    MR.closeMapReduce mr++{- ---------------------------------------------------------------------------------------------------------+   The simple worker fucntions+  --------------------------------------------------------------------------------------------------------- -}   +version :: String+version = "SimpleWorker v 0.1"++prompt :: String+prompt = "# "++version++" > "++localLogger :: String+localLogger = "Holumbus.Distribution.SimpleDMapReduceIO.worker"++pUsage :: IO ()+pUsage = do+  putStrLn "Usage: SimpleWorker ConsolePort Logfile"++params :: IO [String]+params = do+  args <- getArgs+  if length args /= 2 then do+    errorM localLogger "Wrong argument count"+    pUsage+    exitFailure+    else+      return args++{-+ The simpleWorker+-}+worker :: (Show k1, Show k2, Show v1, Show v2, Hash k1, Hash k2, Binary a, NFData k1, NFData k2, Ord k2, Binary k1, Binary k2, NFData v1,NFData v4, NFData v2, NFData v3, Binary v1, Binary v3, Binary v2, Binary v4, Show v4, Show v3) => MapFunction a k1 v1 k2 v2  -> ReduceFunction a k2 v3 v4 -> [(String,Priority)] -> IO ()+worker m r loggers = do+  handleAll (\e -> errorM localLogger $ "EXCEPTION: " ++ show e) $+    do +    (s_cport:logfile:[]) <- params+    --initializeFileLogging logfile [(localLogger, INFO),("Holumbus.Network.DoWithServer",INFO),("Holumbus.MapReduce.Types", INFO)]+    initializeFileLogging logfile ([(localLogger, INFO),("Holumbus.Network.DoWithServer",INFO)]++loggers)+    p <- newPortRegistryFromXmlFile "/tmp/registry.xml"+    setPortRegistry p+    (mr,fs) <- initWorker m r+    DI.runDaemon mr version (read s_cport) prompt +    deinitWorker (mr,fs)++{-+ The simpleWorker's init functin+-}+initWorker :: (Hash k1, Show k1, Show k2, Show v1, Show v2, Show v3, Show v4, Hash k2, Binary a, NFData k1, NFData k2, Ord k2, Binary k1, Binary k2, NFData v1, NFData v4, NFData v2, NFData v3, Binary v1, Binary v3, Binary v2, Binary v4) => MapFunction a k1 v1 k2 v2  -> ReduceFunction a k2 v3 v4 -> IO (MR.DMapReduce, FS.FileSystem)+initWorker m r+  = do+    fs <- FS.mkFileSystemNode FS.defaultFSNodeConfig+    mr <- MR.mkMapReduceWorker fs actionMap MR.defaultMRWorkerConfig+    return (mr,fs)+    where+      actionMap :: ActionMap+      actionMap = KMap.insert (readActionConfiguration (actionConfig m r)) KMap.empty++{-+ The simpleWorker's deinit functin+-}+deinitWorker :: (MR.DMapReduce, FS.FileSystem) -> IO ()+deinitWorker (mr,fs)+  = do+    MR.closeMapReduce mr+    FS.closeFileSystem fs+    +{-++partition'++first list, list of + -}+partition' :: [a] -> [[a]] -> [[a]]+partition'     _        [] = []+partition'    []       xss = xss+partition'    us   (_xs:[]) = [us]+partition' (u:us) (xs:xss) = partition' us (xss ++ [xs'])+  where xs' = (u:xs)++putTimeStamp :: String -> IO ()+putTimeStamp s = do+  t1 <- getPOSIXTime+  putStrLn (s++" : "++ show t1)

@@ -26,8 +26,6 @@  import           Control.Concurrent -import           Data.Maybe- import           System.Log.Logger  import           Holumbus.Common.Debug@@ -90,15 +88,19 @@     case msg of       (M.WReqStartTask td) ->         do-        startTask td wd+        infoM localLogger "recieved start task"+        _ <- startTask td wd+        infoM localLogger "task started"         return $ Just $ M.WRspSuccess       (M.WReqStopTask tid) ->         do-        stopTask tid wd+        infoM localLogger "stop task"+        _ <- stopTask tid wd+        infoM localLogger "task stopped"         return $ Just $ M.WRspSuccess       (M.WReqStopAllTasks) ->         do-        stopAllTasks wd+        _ <- stopAllTasks wd         return $ Just $ M.WRspSuccess       (M.WReqGetActionNames) ->         do@@ -120,7 +122,7 @@         \mp ->         do         debugM localLogger $ "completed Task" ++ show (td_TaskId td)-        MC.receiveTaskCompleted td mp+        _ <- MC.receiveTaskCompleted td mp         return (mp, True)  @@ -131,7 +133,7 @@         \mp ->         do         debugM localLogger $ "error Task" ++ show (td_TaskId td)-        MC.receiveTaskError td mp+        _ <- MC.receiveTaskError td mp         return (mp, True)  @@ -190,3 +192,11 @@       tp <- TP.printTaskProcessor (wd_TaskProcessor wd)       putStrLn "TaskProcessor:"       putStrLn tp+  getDebug wd+    = do+      tmp <- getDebug (wd_Client wd)+      tp <- TP.printTaskProcessor (wd_TaskProcessor wd)+      return ("Worker-Object (full)"+        ++"\n"++tmp+        ++"\n"++"TaskProcessor:"+        ++"\n"++tp++"\n")

@@ -16,7 +16,7 @@ module Holumbus.Distribution.Worker.WorkerPort ( -- * Datatypes-  WorkerPort+  WorkerPort(..)    -- * Creation and Destruction , newWorkerPort@@ -100,3 +100,5 @@       = do         putStrLn "WorkerPort:"         putStrLn $ show p+  getDebug (WorkerPort p)+      = return ("WorkerPort:\n"++show p++"\n")

@@ -0,0 +1,148 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.MapReduce.DaemonInterface+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.MapReduce.DaemonInterface+    (+      -- * operations+      runDaemon+    )+where++import           Data.Maybe++import           Holumbus.Common.Utils                         ( handleAll )+import           Holumbus.Common.FileHandling+import qualified Holumbus.Common.Debug                          as DEBUG+import qualified Holumbus.Console.ServerConsole                       as Console++import qualified Holumbus.MapReduce.MapReduce                   as MR+import qualified Holumbus.MapReduce.Types                       as T++++-- ----------------------------------------------------------------------------+-- Operations+-- ----------------------------------------------------------------------------++-- | runs the user interface... just add an fileSystem an a fancy version-number+runDaemon :: (MR.MapReduce mr) => mr -> String -> Int -> String -> IO ()+runDaemon mr version port prompt+  = do+    -- starts the console with the specified commands+    Console.startServerConsole (createConsole version) mr port prompt++++-- ----------------------------------------------------------------------------+-- private functions+-- ----------------------------------------------------------------------------+++createConsole :: (MR.MapReduce mr) => String -> Console.ConsoleData mr+createConsole version =+  Console.addConsoleCommand "site"      getMySiteId "" $+  Console.addConsoleCommand "mrtype"    getMapReduceType "" $+  Console.addConsoleCommand "startC"    startControlling "" $+  Console.addConsoleCommand "stopC"     stopControlling "" $+  Console.addConsoleCommand "isC"       isControlling "" $+  Console.addConsoleCommand "step"      doSingleStep "" $+  Console.addConsoleCommand "mrJob"     doMapReduceJob "" $+  Console.addConsoleCommand "parse"     parseJob "" $+  Console.addConsoleCommand "debug"     printDebug "" $ +  Console.addConsoleCommand "version"   (printVersion version) "prints the version" $ +  Console.initializeConsole++ +getMySiteId :: (MR.MapReduce mr) => mr -> [String] -> IO String+getMySiteId mr _+  = handleAll (\e -> return . show $ e) $+      do+      s <- MR.getMySiteId mr+      return . show $ s+++getMapReduceType :: (MR.MapReduce mr) => mr -> [String] -> IO String+getMapReduceType mr _+  = handleAll (\e -> return . show $ e) $+      do+      t <- MR.getMapReduceType mr+      return . show $ t++  +startControlling :: (MR.MapReduce mr) => mr -> [String] -> IO String+startControlling mr _+  = handleAll (\e -> return . show $ e) $+      do+      MR.startControlling mr+      return "OK"+++stopControlling :: (MR.MapReduce mr) => mr -> [String] -> IO String+stopControlling mr _+  = handleAll (\e -> return . show $ e) $+      do+      MR.stopControlling mr+      return "OK"+++isControlling :: (MR.MapReduce mr) => mr -> [String] -> IO String+isControlling mr _+  = handleAll (\e -> return . show $ e) $+      do+      b <- MR.isControlling mr+      return . show $ b+++  +doSingleStep :: (MR.MapReduce mr) => mr -> [String] -> IO String+doSingleStep mr _+  = handleAll (\e -> return . show $ e) $+      do+      MR.doSingleStep mr+      return "OK"+++  +doMapReduceJob :: (MR.MapReduce mr) => mr -> [String] -> IO String+doMapReduceJob mr opts+  = do+    handleAll (\e -> return . show $ e) $+      do+      (mbName,_) <- Console.nextOption opts+      jobInfo <- (loadFromXmlFile (fromJust mbName))::IO T.JobInfo+      r <- MR.doMapReduceJob jobInfo mr+      return . (++"RESULT:\n") . show $  r+++parseJob :: (MR.MapReduce mr) => mr -> [String] -> IO String+parseJob _ opts+  = do+    handleAll (\e -> return . show $ e) $+      do+      (mbName,_) <- Console.nextOption opts+      jobInfo <- (loadFromXmlFile (fromJust mbName))::IO T.JobInfo+      return . show $ jobInfo+++printDebug :: (MR.MapReduce mr) => mr -> [String] -> IO String+printDebug mr _+  = do+    handleAll (\e -> return . show $ e) $ DEBUG.getDebug  mr++  +printVersion :: (MR.MapReduce mr) => String -> mr -> [String] -> IO String+printVersion version _ _+  = return version+

@@ -0,0 +1,301 @@+-- ----------------------------------------------------------------------------+{- |+  Module     : Holumbus.MapReduce.DaemonInterface+  Copyright  : Copyright (C) 2009 Sebastian Reese+  License    : MIT++  Maintainer : Sebastian Reese (str@holumbus.org)+  Stability  : experimental+  Portability: portable+  Version    : 0.1+++-}+-- ----------------------------------------------------------------------------++module Holumbus.MapReduce.DaemonInterfaceWithFS+    (+      -- * operations+      runDaemon+    )+where++import           Data.Maybe+import Data.Binary+import Data.List+import           Holumbus.Common.Utils                         ( handleAll )+import           Holumbus.Common.FileHandling+import qualified Holumbus.Common.Debug                          as DEBUG+import qualified Holumbus.Console.ServerConsole                       as Console++import qualified Holumbus.MapReduce.MapReduce                   as MR+import qualified Holumbus.MapReduce.Types                       as T+import qualified Holumbus.FileSystem.FileSystem                 as FS+import qualified Holumbus.FileSystem.Storage                    as S++++-- ----------------------------------------------------------------------------+-- Operations+-- ----------------------------------------------------------------------------++-- | runs the user interface... just add an fileSystem an a fancy version-number+runDaemon :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> String -> Int -> String -> IO ()+runDaemon mrfs version port prompt+  = do+    -- starts the console with the specified commands+    Console.startServerConsole (createConsole version) mrfs port prompt++++-- ----------------------------------------------------------------------------+-- private functions+-- ----------------------------------------------------------------------------+++createConsole :: (MR.MapReduce mr) => String -> Console.ConsoleData (mr,FS.FileSystem)+createConsole version =+  Console.addConsoleCommand "site"      getMySiteId "" $+  Console.addConsoleCommand "mrtype"    getMapReduceType "" $+  Console.addConsoleCommand "startC"    startControlling "" $+  Console.addConsoleCommand "stopC"     stopControlling "" $+  Console.addConsoleCommand "isC"       isControlling "" $+  Console.addConsoleCommand "step"      doSingleStep "" $+  Console.addConsoleCommand "mrJob"     doMapReduceJob "" $+  Console.addConsoleCommand "parse"     parseJob "" $+  Console.addConsoleCommand "debug"     printDebug "" $ +  Console.addConsoleCommand "version"   (printVersion version) "prints the version" $ +  Console.addConsoleCommand "fsid"        getMyFSSiteId     "get my siteId" $+  Console.addConsoleCommand "sites"     getFileSites    "get all sites with the given file name" $+  Console.addConsoleCommand "with"      getNearestNodePortWithFile      "gets the nearest node-port with a file (DEBUG)" $+  Console.addConsoleCommand "for"       getNearestNodePortForFile       "gets the nearest node-port for a (new) file (DEBUG)" $  +  Console.addConsoleCommand "contains"  containsFile    "is file in filesystem or not" $+  Console.addConsoleCommand "create"    createFile      "adds a file" $+  Console.addConsoleCommand "createS"   createFiles     "adds a list of files [(Filename,FileContent)]" $+  Console.addConsoleCommand "append"    append          "appends to a file" $+  Console.addConsoleCommand "delete"    deleteFile      "deletes a file" $+  Console.addConsoleCommand "content"   getFileContent  "gets the content of a file" $+  Console.addConsoleCommand "data"      getFileData     "gets the metadata of a file" $+  Console.addConsoleCommand "local"     isFileLocal     "test, if the file is on the local node" $+  Console.addConsoleCommand "fsdebug"   printFSDebug    "prints internal state of the filesystem (DEBUG)" $ +  Console.initializeConsole++ +getMySiteId :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getMySiteId (mr,_) _+  = handleAll (\e -> return . show $ e) $+      do+      s <- MR.getMySiteId mr+      return . show $ s+++getMapReduceType :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getMapReduceType (mr,_) _+  = handleAll (\e -> return . show $ e) $+      do+      t <- MR.getMapReduceType mr+      return . show $ t++  +startControlling :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+startControlling (mr,_) _+  = handleAll (\e -> return . show $ e) $+      do+      MR.startControlling mr+      return "OK"+++stopControlling :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+stopControlling (mr,_) _+  = handleAll (\e -> return . show $ e) $+      do+      MR.stopControlling mr+      return "OK"+++isControlling :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+isControlling (mr,_) _+  = handleAll (\e -> return . show $ e) $+      do+      b <- MR.isControlling mr+      return . show $ b+++  +doSingleStep :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+doSingleStep (mr,_) _+  = handleAll (\e -> return . show $ e) $+      do+      MR.doSingleStep mr+      return "OK"+++  +doMapReduceJob :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+doMapReduceJob (mr,_) opts+  = do+    handleAll (\e -> return . show $ e) $+      do+      (mbName,_) <- Console.nextOption opts+      jobInfo <- (loadFromXmlFile (fromJust mbName))::IO T.JobInfo+      r <- MR.doMapReduceJob jobInfo mr+      return . (++"RESULT:\n") . show $  r+++parseJob :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+parseJob _ opts+  = do+    handleAll (\e -> return . show $ e) $+      do+      (mbName,_) <- Console.nextOption opts+      jobInfo <- (loadFromXmlFile (fromJust mbName))::IO T.JobInfo+      return . show $ jobInfo+++printDebug :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+printDebug (mr,_) _+  = do+    handleAll (\e -> return . show $ e) $ DEBUG.getDebug  mr++  +printVersion :: (MR.MapReduce mr) => String -> (mr,FS.FileSystem) -> [String] -> IO String+printVersion version _ _+  = return version++getFileNameAndContent :: [String] -> (S.FileId, S.FileContent)+getFileNameAndContent []   = error "no filename given"+getFileNameAndContent (x:xs) = (x, encode $ intercalate " " xs)+-- getFileNameAndContent (x:xs) = (x, S.TextFile $ intercalate " " xs)++getFileNamesAndContent :: [String] -> [(S.FileId, S.FileContent)]+getFileNamesAndContent [] = []+getFileNamesAndContent (_:[]) = error "no content given"+getFileNamesAndContent (fid:c:xs) = ((fid, encode c):getFileNamesAndContent xs)+++getFileNameAndContentSize :: [String] -> (S.FileId, Integer)+getFileNameAndContentSize []   = error "no filename given"+getFileNameAndContentSize (x1:[]) = (x1, 0)+getFileNameAndContentSize (x1:x2:_) = (x1, read x2)+++getMyFSSiteId :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getMyFSSiteId (_,fs) _+  = do+    handleAll (\e -> return $ show e) $+      do+      i <- FS.getMySiteId fs+      return $ show i+     ++getFileSites :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getFileSites (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      s <- FS.getFileSites n fs+      return $ show s+++getNearestNodePortWithFile :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getNearestNodePortWithFile (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      p <- FS.getNearestNodePortWithFile n fs+      return $ show p+++getNearestNodePortForFile :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getNearestNodePortForFile (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, s) = getFileNameAndContentSize opts+      p <- FS.getNearestNodePortForFile n s fs+      return $ show p+++containsFile :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+containsFile (_,fs) opts +  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      b <- FS.containsFile n fs+      return $ show b+++createFile :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+createFile (_,fs) opts +  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, c) = getFileNameAndContent opts+      FS.createFile n c fs +      return ""++createFiles :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+createFiles (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let l = getFileNamesAndContent opts+      FS.createFiles l fs+      return ""++append :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+append (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, c) = getFileNameAndContent opts+      FS.appendFile n c fs+      return ""+  ++deleteFile :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+deleteFile (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      FS.deleteFile n fs+      return ""+++getFileContent :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getFileContent (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      c <- FS.getFileContent n fs+      return $ show c+++getFileData :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+getFileData (_,fs) opts+  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      d <- FS.getFileData n fs+      return $ show d+      ++isFileLocal :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+isFileLocal (_,fs) opts +  = do+    handleAll (\e -> return $ show e) $+      do+      let (n, _) = getFileNameAndContent opts+      b <- FS.isFileLocal n fs+      return $ show b++printFSDebug :: (MR.MapReduce mr) => (mr,FS.FileSystem) -> [String] -> IO String+printFSDebug (_,fs) _+  = do+    handleAll (\e -> return . show $ e) $ DEBUG.getDebug  fs

@@ -47,8 +47,7 @@  import           Prelude hiding                 ( catch ) -import           Control.Exception              ( Exception-                                                , throwTo+import           Control.Exception.Extensible   ( Exception                                                 , catch                                                 ) @@ -500,7 +499,7 @@ sendTask jc td   = do     -- own thread for this... so it is non-blocking-    forkIO $ +    _ <- forkIO $        do       yield       -- get the sendFunction (just read it)@@ -559,7 +558,7 @@       -- get all sending taskdatas                         let sendingTaskDatas = mapMaybe (\tid -> Map.lookup tid (jcd_TaskMap jcd1)) idleTasks       -- send all idle Tasks (non-blocking)-      mapM (sendTask jc) sendingTaskDatas+      _ <- mapM (sendTask jc) sendingTaskDatas              -- get all completed Tasks       let completedTasks = getTaskIds runningJobs [] [TSCompleted] jcd1

@@ -48,9 +48,8 @@  import           Prelude hiding                 ( catch ) -import		 Control.Exception		( Exception+import		 Control.Exception.Extensible   ( Exception 						, throw-						, throwTo 						, catch 						) import           Control.Concurrent@@ -358,7 +357,7 @@   = do     debugM localLogger $ "waiting to stop all Tasks"     tids <- withMVar tp $ \tpd -> return $ getTasksIds tpd-    mapM (\tid -> stopTask tid tp) tids+    _ <- mapM (\tid -> stopTask tid tp) tids     return ()  @@ -483,6 +482,7 @@         ) $          do yield            td' <- performTask td tp+           infoM taskLogger "report task completed"            reportCompletedTask td' tp  -- not used, because we are doi    @@ -510,7 +510,7 @@ -- Performing a Task -- ---------------------------------------------------------------------------- --- | doing a map task+-- | doing a task performTask :: TaskData -> TaskProcessor-> IO TaskData performTask td tp   = do@@ -539,7 +539,9 @@           (Nothing) -> throw (UnkownTaskException $ td_Action td)                  let env = mkActionEnvironment td fs+        infoM taskLogger "starting action"         bout <- action env opt parts bin         let td' = td { td_Output = bout }-        debugM taskLogger $ "output td: " ++ show td'+        infoM taskLogger "finished action"+        --debugM taskLogger $ "output td: " ++ show td'         return td'

@@ -17,7 +17,9 @@ {-# LANGUAGE Arrows, NoMonomorphismRestriction #-} module Holumbus.MapReduce.Types (-  FunctionData(..)+  Hash(..)+, hashedPartition  +,  FunctionData(..)    {- , encodeTuple@@ -60,6 +62,19 @@  , OptionsDecoder ++-- ----------------------------------------------------------------------------+-- remove this, when work done!+-- defaultFunctions (only for profiling)+, defaultInputReader+, defaultOutputWriter+, defaultSplit+, defaultPartition+, defaultMerge+, readConnector+, writeConnector+-- ----------------------------------------------------------------------------+ , ActionConfiguration(..) , SplitConfiguration(..) , MapConfiguration(..)@@ -88,11 +103,14 @@ , ReduceMerge  , ReduceFunction , ReducePartition+, SplitFunction+, SplitAction ) where  import           Control.Concurrent import           Data.Binary+--import           Holumbus.Common.MRBinary import qualified Data.ByteString.Lazy as B import qualified Data.Map as Map import qualified Data.HashTable as Hash@@ -100,16 +118,16 @@ import           Data.Time  import           System.Log.Logger+import           Data.Time.Clock.POSIX  import           Text.XML.HXT.Arrow  import qualified Holumbus.Data.AccuMap as AMap import           Holumbus.Common.FileHandling import           Holumbus.Common.Utils-import qualified Holumbus.Data.AccuMap as AMap import qualified Holumbus.Data.KeyMap as KMap import qualified Holumbus.FileSystem.FileSystem as FS-+import           Control.Parallel.Strategies  localLogger :: String localLogger = "Holumbus.MapReduce.Types"@@ -117,7 +135,11 @@ -- ---------------------------------------------------------------------------- -- general datatypes -- -----------------------------------------------------------------------------+class Hash a where+  hash :: Int -> a -> Int+  +instance Hash Int where+  hash n k = mod k n  data FunctionData   = TupleFunctionData B.ByteString@@ -489,32 +511,70 @@ type OutputWriter k2 v2 = [(k2,v2)] -> IO B.ByteString  readConnector-  :: (Binary k1, Binary v1)+  :: (NFData k1, NFData v1, Binary k1, Binary v1)   => InputReader k1 v1   -> ActionEnvironment   -> [FunctionData]    -> IO [(k1,v1)] readConnector ic ae ls   = do-    debugM localLogger $ "readConnector: " ++ show ls-    os <- mapM (readInput (ae_FileSystem ae)) ls-    return $ concat $ catMaybes os+    let (tuples,filenames) = splitList ls++    mayBeTupleContent <- mapM readInput tuples+    fileContent <- read2 filenames+    +    return . concat $ (catMaybes mayBeTupleContent) ++ fileContent     where    ++    -- split list into tuople data and filenames+    splitList :: [FunctionData] -> ([B.ByteString],[FS.FileId])+    splitList [] = ([],[])+    splitList ((TupleFunctionData t):xs) = ((t:ts),fs)+      where+      (ts,fs) = splitList xs+    splitList ((FileFunctionData f):xs)  = (ts,f:fs)+      where+      (ts,fs) = splitList xs+    + +    -- read all files at once+    --read2 :: [FS.FileId] -> IO [[(k1,v1)]]+    read2 files = do+      contents <- FS.getMultiFileContent files fs+      encodedcontents <- mapM (ic . snd) contents+      return encodedcontents++      where+        fs = ae_FileSystem ae++     -- readInput :: FS.FileSystem -> FunctionData -> IO (Maybe [(k1,v1)])-    readInput _  (TupleFunctionData t) = return $ Just $ [decode t] -    readInput fs (FileFunctionData f)+    readInput t = return $ Just $ [decode t] +{-    readInput fs (FileFunctionData f)       = do                 debugM localLogger $ "loadInputList: getting content for: " ++ f         mbc <- FS.getFileContent f fs-        debugM localLogger $ "loadInputList: content is: " ++ show mbc         if isNothing mbc           then do             debugM localLogger $ "loadInputList: no content found"             return Nothing           else do             d <- ic $ fromJust mbc-            return $ Just d+            return $ Just d +-} +--+--myMapM :: (a -> IO b) -> [a] -> IO [b]+--myMapM f [] = return []+--myMapM f (x:xs) = do+--   infoM localLogger $ "myMapM: do f"+--   b <- f x+--   infoM localLogger $ "myMapM: do recursiv map"+--   bs <- myMapM f xs+--   infoM localLogger $ "myMapM: do return"+--   return (b:bs)+   +       writeConnector    :: (Binary k2, Binary v2)@@ -524,10 +584,23 @@   -> IO [(Int,[FunctionData])] writeConnector oc ae ls   = do-    debugM localLogger $ "writeConnector: "-    os <- mapM (writeOutput (ae_FileSystem ae) tot) ls+    infoM localLogger $ "writeConnector: " ++ (show . length $ ls)+    os <- case tot of+      TOTRawTuple -> mapM (writeOutput (ae_FileSystem ae) tot) ls+      _           -> write2     return $ catMaybes os     where ++    write2 = do+      bincontents <- mapM (\(_,c) -> oc c) ls+      let filelist = zip filenames bincontents;+          +      FS.createFiles filelist fs+      return $ (zipWith (\(i,_) fn -> Just (i,[FileFunctionData fn]) ) ls filenames)+      where+      filenames = map (\(i,_) -> "j" ++ show (td_JobId td) ++ "_t" ++ show (td_TaskId td) ++ "_i" ++ show i) ls+      fs = ae_FileSystem ae+      td   = ae_TaskData ae     tot  = td_OutputType $ ae_TaskData ae     -- writeOutput :: FS.FileSystem -> TaskOutputType -> (Int,[(k2,v2)]) -> IO (Maybe (Int,[FunctionData]))@@ -538,21 +611,23 @@     -- TODO exception werfen      writeOutput fs _ (i,ts)       = do-        c <- oc ts+        infoM localLogger "oc ts"+        c <-  oc ts        +        infoM localLogger "appendfile"         FS.appendFile fn c fs+        infoM localLogger "return just"         return $ Just (i,[FileFunctionData fn])         where-        fn = "j" ++ show (td_JobId td) ++ "_t" ++ show (td_TaskId td) ++ "_i" ++ show i+        fn =  "j" ++ show (td_JobId td) ++ "_t" ++ show (td_TaskId td) ++ "_i" ++ show i  -defaultInputReader :: (Binary k1, Binary v1) => InputReader k1 v1-defaultInputReader b-  = return $ parseByteStringToList b+defaultInputReader :: (NFData v1, NFData k1, Binary k1, Binary v1) => InputReader k1 v1+defaultInputReader = return . parseByteStringToList  -defaultOutputWriter :: (Binary k2, Binary v2) => OutputWriter k2 v2-defaultOutputWriter ls-  = return $ B.concat $ map encode ls+defaultOutputWriter :: (NFData v2, NFData k2, Binary k2, Binary v2) => OutputWriter k2 v2+defaultOutputWriter = return . listToByteString +--  = return $ B.concat $ map encode ls   type ActionName = String@@ -591,18 +666,18 @@   }  -data ReduceConfiguration a k2 v2 v3+data ReduceConfiguration a k2 v3 v4   = ReduceConfiguration {-    rc_Merge     :: ReduceMerge a k2 v2-  , rc_Function  :: ReduceFunction a k2 v2 v3-  , rc_Partition :: ReducePartition a k2 v3-  , rc_Reader    :: InputReader k2 v2-  , rc_Writer    :: OutputWriter k2 v3+    rc_Merge     :: ReduceMerge a k2 v3+  , rc_Function  :: ReduceFunction a k2 v3 v4+  , rc_Partition :: ReducePartition a k2 v4+  , rc_Reader    :: InputReader k2 v3+  , rc_Writer    :: OutputWriter k2 v4   }   defaultActionConfiguration-  :: (Binary a, Binary k1, Binary v1, Binary k2, Binary v4)+  :: (NFData v1, NFData k1, Binary a, Binary k1, Binary v1, Binary k2, Binary v4)   => ActionName   -> ActionConfiguration a k1 v1 k2 v2 v3 v4 defaultActionConfiguration name@@ -620,7 +695,7 @@   defaultSplitConfiguration-  :: (Binary a, Binary k1, Binary v1)+  :: (NFData v1, NFData k1, Binary a, Binary k1, Binary v1)   => SplitConfiguration a k1 v1 defaultSplitConfiguration    = SplitConfiguration@@ -630,7 +705,7 @@   defaultMapConfiguration-  :: (Ord k2, Binary a, Binary k1, Binary v1, Binary k2, Binary v2)+  :: (NFData v1, NFData k1, NFData v2, NFData k2, Ord k2, Binary a, Binary k1, Binary v1, Binary k2, Binary v2)   => MapFunction a k1 v1 k2 v2   -> MapConfiguration a k1 v1 k2 v2 defaultMapConfiguration fct@@ -642,7 +717,7 @@   defaultReduceConfiguration-  :: (Ord k2, Binary a, Binary k2, Binary v2, Binary v3)+  :: (NFData v2, NFData k2, NFData v3, Ord k2, Binary a, Binary k2, Binary v2, Binary v3)   => ReduceFunction a k2 v2 v3   -> ReduceConfiguration a k2 v2 v3 defaultReduceConfiguration fct@@ -655,9 +730,10 @@   defaultSplit-  :: SplitFunction a k1 v1-defaultSplit _ _ n ls -  = return partedList+  :: (NFData k1, NFData v1) => SplitFunction a k1 v1+defaultSplit _ _ n ls = do+  infoM localLogger "defaultSplit"+  return partedList     where     partedList = AMap.toList $ AMap.fromList ps     ns = [(x `mod` n) + 1 | x <- [0..]]@@ -665,6 +741,18 @@     ps = zip ns is  +hashedPartition :: (Hash k2, Binary k2, Binary v2, NFData k2, NFData v2) => MapPartition a k2 v2+hashedPartition _ _ 1 l = return . (:[]) . (,) 1 $ l+hashedPartition _ _ n l = do +  infoM localLogger "hashedPartition: map"+  let a =  map (\t -> (hash n (fst t),[t])) l+  infoM localLogger "hashedPartition: fromList"+  let b = AMap.fromList a+  infoM localLogger "hashedPartition: toList"+  let c = AMap.toList b+  infoM localLogger "hashedPartition: return"+  return $ c+ defaultPartition   :: (Binary k2, Binary v2)   => MapPartition a k2 v2@@ -674,7 +762,7 @@ defaultPartition _ _ n ls   = do     -- calculate partition-Values-    let markedList = map (\t@(k,_) -> (hash k,t)) ls+    let markedList = map (\t@(k,_) -> (hash' k,t)) ls     -- merge them          -- TODO this might change (revert) the order of the Elements...@@ -683,7 +771,7 @@     where     -- calculate a hash-value, because we only have the Binary-Instance, we     -- can only use the Bytestring of the Value-    hash k = ((fromIntegral $ Hash.hashString (show $ encode k)) `mod` n) + 1+    hash' k = ((fromIntegral $ Hash.hashString (show $ encode k)) `mod` n) + 1   defaultMerge@@ -753,6 +841,10 @@  readActionConfiguration   :: ( Ord k2, Binary a+     , Show k1, Show v1+     , Show k2, Show v2, Show v3, Show v4+     , NFData k1, NFData v1+     , NFData k2, NFData v2, NFData v3      , Binary k1, Binary v1      , Binary k2, Binary v2      , Binary v3, Binary v4)@@ -810,7 +902,7 @@ type SplitFunction a k1 v1 = SplitAction a k1 v1  performSplitAction-  :: (Binary a, Binary k1, Binary v1)+  :: (NFData k1, NFData v1, Binary a, Binary k1, Binary v1, Show k1, Show v1)   => OptionsDecoder a   -> SplitFunction a k1 v1   -> InputReader k1 v1@@ -825,17 +917,20 @@     let a = optDec opts          infoM localLogger "performSplitAction"+    putTimeStamp "Begin performSplitAction"          infoM localLogger "reading inputList"     inputList <- readConnector reader env ls-        +    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  input is: " ++ show inputList ++ "\n\n"     infoM localLogger "doing split"     partedList <- case n of       (Just n') -> fct env a n' inputList       (Nothing) -> return [(i,inputList)]     +    debugM localLogger $ ">>>>>>>>>>>>>>>>>> splittet list is: " ++ show partedList ++ "\n\n"     infoM localLogger "writing outputlist"     outputList <- writeConnector writer env partedList+    putTimeStamp "End performSplitAction"     return outputList  @@ -856,8 +951,8 @@       performMapAction-  :: (Ord k2,-      Binary a, Binary k1, Binary v1, Binary k2, Binary v2)+  :: (Ord k2, Show k1, Show k2, Show v1, Show v2,+      Binary a, Binary k1, Binary v1, Binary k2, Binary v2, NFData k2, NFData v2, NFData v1, NFData k1)   => OptionsDecoder a   -> MapFunction a k1 v1 k2 v2   -> MapPartition a k2 v2@@ -874,21 +969,31 @@     let a = optDec opts          infoM localLogger "performMapAction"+    putTimeStamp "Begin performMapAction"     -    infoM localLogger "reading inputList"+    infoM localLogger $ "reading inputList: " ++ show (i,ls)     inputList <- readConnector reader env ls-    -    infoM localLogger "doing map"+    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  input is: " ++ show inputList ++ "\n\n"+        +    infoM localLogger ("doing map: " ++ (show . length $ inputList))     mappedList <- mapM (\(k1, v1) -> fct env a k1 v1) inputList     let tupleList = concat mappedList-     + +    infoM localLogger $ "map result: " ++ (show . length $ tupleList)    +    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  mapped list is: " ++ show tupleList ++ "\n\n"+     infoM localLogger "doing partition"     partedList <- case n of       (Just n') -> part env a n' tupleList       (Nothing) -> return [(i,tupleList)]++    infoM localLogger $ "map parted result: " ++ (show . length $ partedList)    +    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  partitioned list is: " ++ show partedList ++ "\n\n"     -    infoM localLogger "writing outputlist"+    infoM localLogger "writing outputlist: begin"     outputList <- writeConnector writer env partedList+    infoM localLogger "writing outputlist: done"    +    putTimeStamp "End performMapAction"     return outputList  @@ -913,7 +1018,8 @@       performReduceAction-  :: (Ord k2,+  :: (Ord k2, Show k2, Show v2, Show v3,+      NFData k2, NFData v2,       Binary a, Binary k2, Binary v2, Binary v3)   => OptionsDecoder a   -> ReduceMerge a k2 v2@@ -932,25 +1038,34 @@     let a = optDec opts        infoM localLogger "performReduceAction"+    putTimeStamp "Begin performReduceAction"          infoM localLogger "reading inputList"     inputList <- readConnector reader env ls-    +    infoM localLogger $ ">>>>>>>>>>>>>>>>>>  input is: " ++ (show . length $ inputList) ++ "\n\n"+    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  input is: " ++ show inputList ++ "\n\n"+             infoM localLogger "doing merge"     mergedList <- merge env a inputList+    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  mergedList is: " ++ show mergedList     ++ "\n\n"          infoM localLogger "doing reduce"     maybesList <- mapM (\(k2,v2s) -> performReduceFunction a k2 v2s) mergedList     let tupleList = catMaybes maybesList+    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  tupleList is: " ++ show tupleList ++ "\n\n"          infoM localLogger "doing partition"      partedList <- case n of       (Just n') -> part env a n' tupleList       (Nothing) -> return [(i,tupleList)] ++    debugM localLogger $ ">>>>>>>>>>>>>>>>>>  partedList is: " ++ show partedList ++ "\n\n"     -    infoM localLogger "writing outputlist"+    infoM localLogger "writing outputlist: begin"     outputList <- writeConnector writer env partedList+    infoM localLogger "writing outputlist: done"     +    putTimeStamp "End performReduceAction"     return outputList     where       performReduceFunction a k2 v2s@@ -959,3 +1074,8 @@           case mbV3 of             (Nothing) -> return Nothing             (Just v3) -> return $ Just (k2,v3)++putTimeStamp :: String -> IO ()+putTimeStamp s = do+  t1 <- getPOSIXTime+  infoM localLogger (s++" : "++ show t1)

@@ -192,6 +192,24 @@         actions <- getActions (sad_TaskProcessor sad)         putStrLn $ show $ actions +  getDebug (SMapReduce sa)+    = do+      withMVar sa $+        \sad ->+        do+        showJC <- printJobController (sad_JobController sad)+        showTP <- printTaskProcessor (sad_TaskProcessor sad)+        actions <- getActions (sad_TaskProcessor sad)+        let line= "--------------------------------------------------------"+        return (line+          ++"\n"++ "Job-Controller"+          ++"\n"++ showJC+          ++"\n"++ line+          ++"\n"++ "Task-Processor"+          ++"\n"++ showTP+          ++"\n"++ line+          ++"\n"++ "Actions"+          ++"\n"++ show actions++"\n")