JobController.hs

-- ----------------------------------------------------------------------------
{- |
  Module     : Holumbus.MapReduce.JobController
  Copyright  : Copyright (C) 2008 Stefan Schmidt
  License    : MIT

  Maintainer : Stefan Schmidt (stefanschmidt@web.de)
  Stability  : experimental
  Portability: portable
  Version    : 0.1


-}
-- ----------------------------------------------------------------------------

{-# OPTIONS -fglasgow-exts #-}
module Holumbus.MapReduce.JobController
    ( TaskSendResult(..)
    , TaskSendFunction

    , JobController

    , printJobController

      -- * Creation and Destruction
    , newJobController
    , closeJobController
    , setFileSystemToJobController
    , setTaskSendHook

      -- * Job Controller 
    , startJobController
    , stopJobController
    , isJobControllerRunning
    , singleStepJobControlling

      -- * performing MapReduce-Jobs
      -- , startJob
      -- , stopJob
    , performJob

      -- * handling Task-Responses
    , setTaskCompleted
    , setTaskError
    )
where

import           Prelude hiding                 ( catch )

import           Control.Exception.Extensible   ( Exception
                                                , catch
                                                )

import           Control.Concurrent
import           Control.Monad

import qualified Data.ByteString.Lazy           as B
import           Data.Maybe
import           Data.Time
import           Data.Typeable
import qualified Data.Map                       as Map
import qualified Data.Set                       as Set
import           System.Log.Logger

import           Holumbus.MapReduce.Types
import qualified Holumbus.FileSystem.FileSystem as FS
import qualified Holumbus.Data.AccuMap          as AMap
import qualified Holumbus.Data.MultiMap         as MMap


localLogger :: String
localLogger = "Holumbus.MapReduce.JobController"

cycleLogger :: String
cycleLogger = "Holumbus.MapReduce.JobController.cycle"

-- ----------------------------------------------------------------------------
--
-- ----------------------------------------------------------------------------


newJobData  
  :: JobControllerData -> JobInfo
  -> IO (JobControllerData, JobData, MVar JobResult)
newJobData jcd info
  = do
    t <- getCurrentTime
    mVar <- newEmptyMVar
    let state = JSIdle
        parts = maybe 1 id $ getCurrentTaskPartValue state info
        jrc   = JobResultContainer mVar 
        jid   = jcd_NextJobId jcd
        jcd'  = jcd { jcd_NextJobId = (jid+1) }
        pairs = initialSplit parts (ji_Input info) 
        -- let pairs = zip (iterate (+1) 1) (map (\i -> [i]) (ji_Input info))
        accuMap = AMap.fromList pairs 
        outputMap = Map.insert JSIdle accuMap Map.empty
    return (jcd', JobData jid state outputMap info t t jrc, mVar)
    where
    initialSplit n ls = ps
      where
      ns = [(x `mod` n) + 1 | x <- [1..]]
      is = map (\a -> [a]) ls 
      ps = zip ns is


newTaskData 
  :: JobControllerData 
  -> JobId -> TaskType -> TaskState -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> ActionName -> TaskOutputType 
  -> IO (JobControllerData, TaskData)
newTaskData jcd jid tt ts opt n i a ot
  = do
    let tid = jcd_NextTaskId jcd
    let jcd' = jcd { jcd_NextTaskId = (tid+1) }
    return (jcd', TaskData jid tid tt ts opt n i [] ot a) 
    
data TaskSendResult = TSRSend | TSRNotSend | TSRError
  deriving (Show, Eq, Ord, Enum)    
    
-- ----------------------------------------------------------------------------
--
-- ----------------------------------------------------------------------------

intersections :: Ord a => [Set.Set a] -> Set.Set a  
intersections [] = Set.empty
intersections (s:ss) = foldl Set.intersection s ss 


mapAccumLM
  :: (Monad m) 
  => (acc -> x -> m (acc, y)) -- Function of of input list
                              -- and accumulator, returning new
                              -- accumulator and of result list
  -> acc                      -- Initial accumulator 
  -> [x]                      -- Input list
  -> m (acc, [y])             -- Final accumulator and result list
mapAccumLM _ s [] = return (s, [])
mapAccumLM f s (x:xs) 
  = do
    (s', y ) <- f s x
    (s'',ys) <- mapAccumLM f s' xs
    return (s'',y:ys)




type TaskSendFunction = TaskData -> IO (TaskSendResult)

dummyTaskSendFunction :: TaskData -> IO (TaskSendResult)
dummyTaskSendFunction _ = return TSRSend


type JobMap         = Map.Map JobId JobData
type TaskMap        = Map.Map TaskId TaskData
    
type StateJobIdMap  = MMap.MultiMap JobState JobId

type JobIdTaskIdMap = MMap.MultiMap JobId TaskId
type TypeTaskIdMap  = MMap.MultiMap TaskType TaskId
type StateTaskIdMap = MMap.MultiMap TaskState TaskId
   
    
data JobControlFunctions = JobControlFunctions {
    jcf_TaskSend :: TaskSendFunction
  }
instance Show JobControlFunctions where
  show _ = "JobControlFunctions"

    
data JobControllerData = JobControllerData {
  -- internal
    jcd_ServerThreadId :: Maybe ThreadId
  , jcd_ServerDelay    :: Int
  , jcd_FileSystem     :: Maybe FS.FileSystem
  -- control
  , jcd_NextJobId      :: JobId
  , jcd_NextTaskId     :: TaskId
  , jcd_Functions      :: JobControlFunctions
  -- job control
  , jcd_JobMap         :: ! JobMap
  , jcd_TaskMap        :: ! TaskMap
  , jcd_StateJobIdMap  :: ! StateJobIdMap
  , jcd_JobIdTaskIdMap :: ! JobIdTaskIdMap
  , jcd_TypeTaskIdMap  :: ! TypeTaskIdMap
  , jcd_StateTaskIdMap :: ! StateTaskIdMap
  } deriving (Show)


type JobController = MVar JobControllerData

printJobController :: JobController -> IO String
printJobController jc
  = withMVar jc $ \jcd -> return $ show jcd


data JobControllerException
    = KillServerException
      deriving (Show, Typeable)

instance Exception JobControllerException where

-- ----------------------------------------------------------------------------
-- Contruction / Destruction 
-- ----------------------------------------------------------------------------

defaultJobControllerData :: JobControllerData
defaultJobControllerData = jcd
  where
  jcd = JobControllerData
    Nothing
    10000 -- 10 milliseconds delay
    Nothing
    1
    1
    jcf
    Map.empty
    Map.empty
    MMap.empty
    MMap.empty
    MMap.empty
    MMap.empty
  jcf = JobControlFunctions
    dummyTaskSendFunction

newJobController :: IO JobController
newJobController
  = do
    let jcd = defaultJobControllerData
    jc <- newMVar jcd
    return jc 

closeJobController :: JobController -> IO ()
closeJobController jc
  = do
    stopJobController jc


setFileSystemToJobController :: FS.FileSystem -> JobController -> IO ()
setFileSystemToJobController fs jc
  = modifyMVar jc $
    \jcd -> return $ (jcd { jcd_FileSystem = Just fs }, ())


setTaskSendHook :: TaskSendFunction -> JobController -> IO ()
setTaskSendHook f jc
  = do
    modifyMVar jc $
      \jcd ->
      do
      let funs = jcd_Functions jcd
      let funs' = funs { jcf_TaskSend = f }
      return (jcd { jcd_Functions = funs' }, ())


-- ----------------------------------------------------------------------------
-- server functions
-- ----------------------------------------------------------------------------

startJobController :: JobController -> IO ()
startJobController jc
  = do
    modifyMVar jc $ 
      \jcd -> 
      do
      thd <- case (jcd_ServerThreadId jcd) of
        (Just i) -> return i
        (Nothing) ->
          do
          i <- forkIO $ doProcessing jc True
          return i
      return (jcd {jcd_ServerThreadId = (Just thd)}, ())



stopJobController :: JobController -> IO ()
stopJobController jc
  = do
    modifyMVar jc $ 
      \jcd -> 
      do
      case (jcd_ServerThreadId jcd) of
        (Nothing) -> return ()
        (Just i) -> 
          do
          throwTo i KillServerException
          yield
          return ()
      return (jcd {jcd_ServerThreadId = Nothing}, ())


isJobControllerRunning :: JobController -> IO Bool
isJobControllerRunning jc
  = withMVar jc $
      \jcd -> return $ isJust (jcd_ServerThreadId jcd)


singleStepJobControlling :: JobController -> IO ()
singleStepJobControlling jc
  = do
    singleStepAllowed <- withMVar jc $ 
      \jcd -> return $ isNothing (jcd_ServerThreadId jcd)
    if singleStepAllowed then doProcessing jc False else return ()

-- ----------------------------------------------------------------------------
-- private functions
-- ----------------------------------------------------------------------------

-- | get the JobIds, they can be filtered by JobState, if the filter is empty
--   all Jobs are returned.
getJobIds :: [JobState] -> JobControllerData -> [JobId]
getJobIds states jcd = Set.toList $ MMap.filterElements states (jcd_StateJobIdMap jcd)


-- | get the TaskIds, they can be filtered by JobId, TaskType and TaskState, if
--   a filter is empty, this attribute is not taken into account
getTaskIds :: [JobId] -> [TaskType] -> [TaskState] -> JobControllerData -> [TaskId]
getTaskIds jobs types states jcd = Set.toList $ intersections sets
  where
  jobSelected   = MMap.filterElements jobs (jcd_JobIdTaskIdMap jcd)
  typeSelected  = MMap.filterElements types (jcd_TypeTaskIdMap jcd)
  stateSelected = MMap.filterElements states (jcd_StateTaskIdMap jcd)
  sets = [jobSelected, typeSelected, stateSelected]


allTasksFinished :: JobControllerData -> JobId -> Bool
allTasksFinished jcd jid = null notFinishedTasks
  where
    notFinishedTasks = getTaskIds [jid] [] [TSIdle, TSSending, TSInProgress, TSCompleted] jcd



-- | adds a Job to the JobControllerData for execution
addJob :: JobData -> JobControllerData -> JobControllerData
addJob jd jcd = jcd { jcd_JobMap = jm', jcd_StateJobIdMap = sjm' }
  where
  jm' = Map.insert (jd_JobId jd) jd (jcd_JobMap jcd)
  sjm' = MMap.insert (jd_State jd) (jd_JobId jd) (jcd_StateJobIdMap jcd)


changeJobState :: JobId -> JobState -> JobControllerData -> JobControllerData
changeJobState jid js jcd = changeJobState' (Map.lookup jid (jcd_JobMap jcd))
  where
  changeJobState' (Nothing) = jcd
  changeJobState' (Just jd) = jcd { jcd_JobMap = jm', jcd_StateJobIdMap = sjm' }
    where
    jd' = jd { jd_State = js }  -- change JobData
    js' = jd_State jd           -- get old state
    jm' = Map.insert jid jd' (jcd_JobMap jcd) -- change JobData
    sjm' = MMap.insert js jid $ MMap.deleteElem js' jid (jcd_StateJobIdMap jcd) -- change StateJobIdMap


updateJob :: JobData -> JobControllerData -> JobControllerData
updateJob jd jcd = addJob jd $ maybe (jcd) (\_ -> changeJobState jid js jcd) maybeOldJob
  where
    maybeOldJob = (Map.lookup jid (jcd_JobMap jcd))
    jid = jd_JobId jd
    js = jd_State jd


addOutputToJob :: [(Int,[FunctionData])] -> JobData -> JobData
addOutputToJob outList jd = jd { jd_OutputMap = outputmap' }  
  where
  outAccu    = AMap.fromList outList
  state      = jd_State jd
  outputmap  = jd_OutputMap jd
  outputmap' = Map.insertWith (\m1 m2 -> AMap.union m1 m2) state outAccu outputmap 

addTask :: JobControllerData -> TaskData -> JobControllerData
addTask jcd td= jcd { jcd_TaskMap = tm', jcd_JobIdTaskIdMap = jtm', jcd_TypeTaskIdMap = ttm', jcd_StateTaskIdMap = stm' }
  where
    tid = td_TaskId td
    tm' = Map.insert tid td (jcd_TaskMap jcd)
    jtm' = MMap.insert (td_JobId td) tid (jcd_JobIdTaskIdMap jcd)
    ttm' = MMap.insert (td_Type td) tid (jcd_TypeTaskIdMap jcd)
    stm' = MMap.insert (td_State td) tid (jcd_StateTaskIdMap jcd)


changeTaskState :: TaskId -> TaskState -> JobControllerData -> JobControllerData
changeTaskState tid ts jcd = changeTaskState' (Map.lookup tid (jcd_TaskMap jcd))
  where
  changeTaskState' (Nothing) = jcd
  changeTaskState' (Just td) = 
    if isChangeTaskStateAllowed ts (td_State td)
      then jcd { jcd_TaskMap = tm', jcd_StateTaskIdMap = stm' }
      else jcd
    where
    td' = td { td_State = ts }  -- change TaskData
    ts' = td_State td           -- get old state
    tm' = Map.insert tid td' (jcd_TaskMap jcd) -- change TaskData
    stm' = MMap.insert ts tid $ MMap.deleteElem ts' tid (jcd_StateTaskIdMap jcd) -- change StateTaskIdMap


-- | newState oldState
isChangeTaskStateAllowed :: TaskState -> TaskState -> Bool
-- only Error-Tasks and Sending-Tasks can be reset to idle
isChangeTaskStateAllowed TSIdle       t = Set.member t $ Set.fromList [TSSending, TSError]
-- sending only from idle
isChangeTaskStateAllowed TSSending    t = Set.member t $ Set.fromList [TSIdle]
-- inProgress only from an idle Tasks or sending Task
isChangeTaskStateAllowed TSInProgress t = Set.member t $ Set.fromList [TSSending, TSIdle]
-- completed only sending, idle and inprogress
isChangeTaskStateAllowed TSCompleted  t = Set.member t $ Set.fromList [TSSending, TSIdle, TSInProgress]
-- finished all but not error
isChangeTaskStateAllowed TSFinished   t = Set.member t $ Set.fromList [TSSending, TSIdle, TSInProgress, TSCompleted]
-- we can always set a task to error
isChangeTaskStateAllowed TSError      _ = True
-- otherwise false (there are no other cases)
-- isChangeTaskStateAllowed _ _ = False

updateTaskOutput :: TaskId -> [(Int, [FunctionData])] -> JobControllerData -> JobControllerData
updateTaskOutput _ [] jcd = jcd
updateTaskOutput tid o jcd = updateTaskOuput' (Map.lookup tid (jcd_TaskMap jcd))
  where
  updateTaskOuput' (Nothing) = jcd
  updateTaskOuput' (Just td) = jcd { jcd_TaskMap = tm' }
    where
    td' = td { td_Output = o }  -- change TaskData
    tm' = Map.insert tid td' (jcd_TaskMap jcd) -- change TaskData


-- ----------------------------------------------------------------------------
-- Info an Debug
-- ----------------------------------------------------------------------------


-- ----------------------------------------------------------------------------
-- Task Controlling
-- ----------------------------------------------------------------------------


startJob 
  :: JobInfo -> JobController -> IO (JobId, MVar JobResult)
startJob ji jc
  = do
    modifyMVar jc $
      \jcd ->
      do
      -- test the job info... we don't want false jobs
      -- let (b, m) = testJobInfo ji (jcd_MapActionMap jcd) (jcd_ReduceActionMap jcd)
      -- if b 
      --   then do
      debugM localLogger $ "startJob: " ++ show ji
      
      (jcd',jd, jr) <- newJobData jcd ji
      return (addJob jd jcd', (jd_JobId jd, jr))
      --   else do
      --    return (jcd, Left m)

{-
stopJob :: JobId -> JobController -> IO ()
stopJob _ _
  = do
-}

performJob
  :: JobInfo -> JobController -> IO JobResult
performJob ji jc
  = do
    -- start the Job
    (_, mjr) <- startJob ji jc
    -- wait until the result is ready
    withMVar mjr $ \jr -> return jr

-- ----------------------------------------------------------------------------
-- Task Processing
-- ----------------------------------------------------------------------------


doProcessing :: JobController -> Bool -> IO ()
doProcessing jc loop
  = do
    catch (doProcessing' jc loop)
      handler
    where
      handler :: JobControllerException -> IO ()
      handler err = errorM cycleLogger $ "doProcessing: " ++ (show err)
      doProcessing' jc' loop'
        = do
          handleTasks jc'
          handleJobs jc'
          if loop' 
            then do
              delay <- withMVar jc' (\jcd -> return $ jcd_ServerDelay jcd)
              threadDelay delay
              doProcessing' jc' loop'
            else
              return ()



-- ----------------------------------------------------------------------------
-- Process Tasks
-- ----------------------------------------------------------------------------


-- | the send task doesn't block, because a timeout causes the whole job
--   controller to pause, we don't want this
sendTask :: JobController -> TaskData -> IO ()
sendTask jc td
  = do
    -- own thread for this... so it is non-blocking
    _ <- forkIO $ 
      do
      yield
      -- get the sendFunction (just read it)
      sendFunction <- withMVar jc $
        \jcd -> return $ jcf_TaskSend $ jcd_Functions jcd
      -- execute the send function (this might timeout or block)
      sendResult <- sendFunction td
      -- set the task in the correct state
      case sendResult of
        (TSRSend) ->
          do
          setTaskInProgress jc td
        (TSRNotSend) ->
          do
          -- this function can be called twice before it is finished
          -- and therefore, we introduced the "sending" state for tasks
          -- and if the sending is not ok, we have to set the task to idle again
          setTaskIdle jc td
        (TSRError) ->
          do
          setTaskError jc td
    return ()


-- toNextTaskState :: JobControllerData -> TaskData -> JobControllerData
-- toNextTaskState jcd td = changeTaskState (td_TaskId td) (getNextTaskState (td_State td)) jcd 

toSendingTaskState :: JobControllerData -> TaskData -> JobControllerData
toSendingTaskState jcd td = changeTaskState (td_TaskId td) TSSending jcd

finishTask :: JobControllerData -> TaskData -> JobControllerData
finishTask jcd td = changeTaskState (td_TaskId td) (getNextTaskState (td_State td)) jcd'  
-- toNextTaskState jcd' td
  where
    jd   = fromJust $ Map.lookup (td_JobId td) (jcd_JobMap jcd)
    jd'  = addOutputToJob (td_Output td) jd  
    jcd' = updateJob jd' jcd 


handleTasks :: JobController -> IO ()
handleTasks jc
  = modifyMVar jc $
      \jcd ->
      do
      infoM cycleLogger "processing Tasks:"
      -- get all runnning jobs (not idle...)
      let runningJobs = getJobIds [JSSplit, JSMap, JSCombine, JSReduce] jcd
      infoM cycleLogger $ "running Jobs:" ++ show runningJobs
      -- get all idle tasks
      let idleTasks = getTaskIds runningJobs [] [TSIdle] jcd
      infoM cycleLogger $ "idle Tasks:" ++ show idleTasks
      let idleTaskDatas = mapMaybe (\tid -> Map.lookup tid (jcd_TaskMap jcd)) idleTasks
      
      -- set these tasks to sending, so they won't be called again while the sending process
      let jcd1 = foldl toSendingTaskState jcd idleTaskDatas      
      -- 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
      
      -- get all completed Tasks
      let completedTasks = getTaskIds runningJobs [] [TSCompleted] jcd1
      let completedTaskDatas = mapMaybe (\tid -> Map.lookup tid (jcd_TaskMap jcd1)) completedTasks
      -- inProgressTasks = getTaskIds runningJobs [] [TSInProgress] jcd
      let jcd2 = foldl finishTask jcd1 completedTaskDatas
      return (jcd2,())


-- ----------------------------------------------------------------------------
-- Process Jobs
-- ----------------------------------------------------------------------------

toNextJobState :: JobControllerData -> JobData -> JobControllerData
toNextJobState jcd jd = changeJobState (jd_JobId jd) (getNextJobState (jd_State jd)) jcd

toIdleTaskState :: JobControllerData -> TaskData -> JobControllerData
toIdleTaskState jcd td = changeTaskState (td_TaskId td) TSIdle jcd

toInProgressTaskState :: JobControllerData -> TaskData -> JobControllerData
toInProgressTaskState jcd td = changeTaskState (td_TaskId td) TSInProgress jcd

toCompletedTaskState :: JobControllerData -> TaskData -> JobControllerData
toCompletedTaskState jcd td = updateTaskOutput tid o $ changeTaskState tid TSCompleted jcd
  where
  tid = td_TaskId td
  o = td_Output td

toErrorTaskState :: JobControllerData -> TaskData -> JobControllerData
toErrorTaskState jcd td = changeTaskState (td_TaskId td) TSError jcd


hasPhase :: JobData -> Bool
hasPhase jd = isJust $ getCurrentTaskAction jd


getCurrentTaskAction :: JobData -> Maybe ActionName
getCurrentTaskAction jd = getTaskAction' (jd_Info jd) (jd_State jd)
  where
  getTaskAction' ji JSSplit   = maybe Nothing (Just . ja_Name) (ji_SplitAction ji)
  getTaskAction' ji JSMap     = maybe Nothing (Just . ja_Name) (ji_MapAction ji)
  getTaskAction' ji JSCombine = maybe Nothing (Just . ja_Name) (ji_CombineAction ji)
  getTaskAction' ji JSReduce  = maybe Nothing (Just . ja_Name) (ji_ReduceAction ji) 
  getTaskAction' _  _         = Nothing


getCurrentTaskOutputType :: JobData -> TaskOutputType
getCurrentTaskOutputType jd = getTaskOutputType' (jd_Info jd) (jd_State jd)
  where
  getTaskOutputType' ji JSSplit   = maybe TOTFile (ja_Output) (ji_SplitAction ji)
  getTaskOutputType' ji JSMap     = maybe TOTFile (ja_Output) (ji_MapAction ji)
  getTaskOutputType' ji JSCombine = maybe TOTFile (ja_Output) (ji_CombineAction ji)
  getTaskOutputType' ji JSReduce  = maybe TOTRawTuple (ja_Output) (ji_ReduceAction ji)
  getTaskOutputType' _  _         = TOTFile


getCurrentTaskPartValue :: JobState -> JobInfo -> Maybe Int
getCurrentTaskPartValue state ji = gTPV state
  where
  gTPV JSIdle    = maybe (gTPV JSSplit)   (Just . ja_Count) (ji_SplitAction ji)
  gTPV JSSplit   = maybe (gTPV JSMap)     (Just . ja_Count) (ji_MapAction ji)
  gTPV JSMap     = maybe (gTPV JSCombine) (\_ -> Nothing)   (ji_CombineAction ji)
  gTPV JSCombine = maybe (gTPV JSReduce)  (Just . ja_Count) (ji_ReduceAction ji)
  gTPV JSReduce  = ji_NumOfResults ji
  gTPV _         = Nothing
  

createTasks :: JobControllerData -> JobData -> IO JobControllerData
createTasks jcd jd
  = do
    let state     = jd_State jd
        info      = jd_Info jd
        outputMap = (jd_OutputMap jd)
    -- our input is the output of the previous state
    let inputAccu = maybe (AMap.empty) (id) $ Map.lookup (getPrevJobState state) outputMap
    if (hasPhase jd)
      then do
        let inputList = AMap.toList inputAccu                
        let jid = jd_JobId jd
        let a = fromJust $ getCurrentTaskAction jd
        let ot = getCurrentTaskOutputType jd
        let opts = ji_Option info
        let n = getCurrentTaskPartValue state info
        let tt  = fromJust $ fromJobStatetoTaskType state
        -- create new tasks
        (jcd', taskDatas) <- mapAccumLM (\d i -> newTaskData d jid tt TSIdle opts n i a ot) jcd inputList
        -- add task to controller
        let jcd'' = foldl addTask jcd' taskDatas        
        return jcd''
      else do
        -- there is no action in this state, so we copy the input to the output
        let outputMap' = Map.insert state inputAccu outputMap 
        let jd' = jd { jd_OutputMap = outputMap' }
        return $ updateJob jd' jcd


createResults :: JobControllerData -> JobData -> IO JobControllerData
createResults jcd jd
  = do
    let outputAccu = maybe (AMap.empty) (id) $ Map.lookup JSCompleted (jd_OutputMap jd)
    let outputList = AMap.toList outputAccu
    let (JobResultContainer mVarResult) = jd_Result jd
    let res = JobResult $ concat $ map (snd) outputList
    putMVar mVarResult res
    return jcd  


handleJobs :: JobController -> IO ()
handleJobs jc
  = modifyMVar jc $
      \jcd ->
      do
      -- get all working jobs (idle or running)
      infoM cycleLogger "processing Jobs"
      let workingJobs = getJobIds [JSIdle, JSSplit, JSMap, JSCombine, JSReduce] jcd
      infoM cycleLogger $ "working jobs:" ++ show workingJobs
      -- process only jobs, whose current phase is done
      let jobsWithoutTasks = filter (allTasksFinished jcd) workingJobs
      infoM cycleLogger $ "jobsWithoutTasks:" ++ show jobsWithoutTasks
      -- get the old jobdatas
      let oldJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd)) jobsWithoutTasks
      debugM cycleLogger $ "oldJobDatas:\n" ++ show oldJobDatas
      -- change the job states to the next phase
      let jcd1 = foldl toNextJobState jcd oldJobDatas
      -- get the new jobdatas
      let newJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd1)) jobsWithoutTasks
      debugM cycleLogger $ "newJobDatas:\n" ++ show newJobDatas
      -- create new tasks for each Job
      jcd2 <- foldM createTasks jcd1 newJobDatas
      
      -- get all completed Jobs
      let completedJobs = getJobIds [JSCompleted] jcd2
      infoM cycleLogger $ "completed Jobs:" ++ show completedJobs
      -- get the completed JobDatas
      let completedJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd2)) completedJobs
      -- change the job states to the next phase
      let jcd3 = foldl toNextJobState jcd2 completedJobDatas
      -- get the new jobdatas
      let finishedJobDatas = mapMaybe (\jid -> Map.lookup jid (jcd_JobMap jcd3)) completedJobs
      -- create new tasks for each Job
      jcd4 <- foldM createResults jcd3 finishedJobDatas
      
      return (jcd4, ())




-- ----------------------------------------------------------------------------   
-- handling Task-Responses
-- ----------------------------------------------------------------------------


setTaskIdle :: JobController -> TaskData -> IO ()
setTaskIdle jc td
  = do
    debugM localLogger $ "setTaskIdle: waiting... TaskId: " ++ show (td_TaskId td)
    modifyMVar jc $
      \jcd ->
      do
      debugM localLogger $ "setTaskIdle: setting... TaskId: " ++ show (td_TaskId td)
      let jcd' = toIdleTaskState jcd td
      return (jcd', ())


setTaskInProgress :: JobController -> TaskData -> IO ()
setTaskInProgress jc td
  = do
    debugM localLogger $ "setTaskInProgress: waiting... TaskId: " ++ show (td_TaskId td)
    modifyMVar jc $
      \jcd ->
      do
      debugM localLogger $ "setTaskInProgress: setting... TaskId: " ++ show (td_TaskId td)
      let jcd' = toInProgressTaskState jcd td
      return (jcd', ())


setTaskCompleted :: JobController -> TaskData -> IO ()
setTaskCompleted jc td
  = do
    debugM localLogger $ "setTaskCompleted: waiting... TaskId: " ++ show (td_TaskId td)
    modifyMVar jc $
      \jcd ->
      do
      debugM localLogger $ "setTaskCompleted: setting... TaskId: " ++ show (td_TaskId td)
      let jcd' = toCompletedTaskState jcd td
      return (jcd', ())

    
setTaskError :: JobController -> TaskData -> IO ()
setTaskError jc td
  = do
    debugM localLogger $ "setTaskError: waiting... TaskId: " ++ show (td_TaskId td)
    modifyMVar jc $
      \jcd ->
      do
      debugM localLogger $ "setTaskError: setting... TaskId: " ++ show (td_TaskId td)
      let jcd' = toErrorTaskState jcd td
      return (jcd', ())