content: -- ----------------------------------------------------------------------------
  {- |
    Module     : Holumbus.MapReduce.Types
    Copyright  : Copyright (C) 2008 Stefan Schmidt
    License    : MIT
  
    Maintainer : Stefan Schmidt (stefanschmidt@web.de)
    Stability  : experimental
    Portability: portable
    Version    : 0.1
  
  
  -}
  -- ----------------------------------------------------------------------------
  
  {-# OPTIONS -fglasgow-exts #-}
  {-# LANGUAGE Arrows, NoMonomorphismRestriction #-}
  module Holumbus.MapReduce.Types
  (
    Hash(..)
  , hashedPartition  
  ,  FunctionData(..)
    
  {-
  , encodeTuple
  , decodeTuple
  , decodeTupleList
  , encodeTupleList
  -}
  
  -- * TaskData
  , TaskId
  , TaskType(..)
  , TaskState(..)
  , getNextTaskState
  , TaskOutputType(..)
  , TaskData(..)
  
  -- * JobData
  , JobId
  , JobState(..)
  , getNextJobState
  , getPrevJobState
  , fromJobStatetoTaskType
  , OutputMap
  , JobAction(..)
  , JobInfo(..)
  , JobData(..)
  , JobResultContainer(..)
  , JobResult(..)
  
  
  -- * TaskAction
  , ActionName
  , ActionInfo
  
  , ActionEnvironment(..)
  , mkActionEnvironment
  
  , InputReader
  , OutputWriter
  
  , OptionsDecoder
  
  
  -- ----------------------------------------------------------------------------
  -- remove this, when work done!
  -- defaultFunctions (only for profiling)
  , defaultInputReader
  , defaultOutputWriter
  , defaultSplit
  , defaultPartition
  , defaultMerge
  , readConnector
  , writeConnector
  -- ----------------------------------------------------------------------------
  
  , ActionConfiguration(..)
  , SplitConfiguration(..)
  , MapConfiguration(..)
  , ReduceConfiguration(..)
  , defaultActionConfiguration
  , defaultSplitConfiguration
  , defaultMapConfiguration
  , defaultReduceConfiguration
  , readActionConfiguration
  , createJobInfoFromConfiguration
  , createListsFromJobResult
  
  , ActionData(..)
  , getActionForTaskType
  , ActionMap
  
  -- * MapAction
  , MapAction
  -- , BinaryMapAction 
  , MapFunction
  , MapPartition
  
  -- * Combine-Reduce-Action
  , ReduceAction
  -- , BinaryReduceAction
  , 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
  import           Data.Maybe
  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.KeyMap as KMap
  import qualified Holumbus.FileSystem.FileSystem as FS
  import           Control.DeepSeq
  
  localLogger :: String
  localLogger = "Holumbus.MapReduce.Types"
  
  -- ----------------------------------------------------------------------------
  -- 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
    | FileFunctionData FS.FileId
    deriving (Show, Eq, Ord)
  
  instance Binary FunctionData where
    put (TupleFunctionData t)  = putWord8 0 >> put t
    put (FileFunctionData f) = putWord8 1 >> put f
    get
      = do
        t <- getWord8
        case t of
          1 -> get >>= \f -> return (FileFunctionData f)
          _ -> get >>= \b -> return (TupleFunctionData b)
          
          
  instance XmlPickler FunctionData where
    xpickle = xpFunctionData
          
  xpFunctionData :: PU FunctionData
  xpFunctionData
    = xpElem "data" $
        xpAlt tag ps
        where
        tag (TupleFunctionData _) = 0
        tag (FileFunctionData _)  = 1
        ps = [xpWrap (\p -> TupleFunctionData (encodeTuple p), \(TupleFunctionData p) -> decodeTuple p) xpRawFunctionData
             ,xpWrap (\f -> FileFunctionData f, \(FileFunctionData f) -> f) xpFileFunctionData ]
        xpRawFunctionData = xpPair (xpElem "key" xpText0) (xpElem "value" xpText0)
        xpFileFunctionData = xpElem "filename" xpText
  
  
          
  -- ----------------------------------------------------------------------------
  -- general encoding / decoding
  -- ----------------------------------------------------------------------------
  
  encodeTuple :: (Binary k, Binary v) => (k, v) -> B.ByteString
  encodeTuple t = encode t
  
  
  decodeTuple :: (Binary k, Binary v) => B.ByteString -> (k, v)
  decodeTuple t = decode t
  
  
  -- encodeTupleList :: (Binary k, Binary v) => [(k, v)] -> [B.ByteString]
  -- encodeTupleList ls = map encodeTuple ls
  
  
  -- decodeTupleList :: (Binary k, Binary v) => [B.ByteString] -> [(k, v)]
  -- decodeTupleList ls = map decodeTuple ls
  
  
  
  
  -- ----------------------------------------------------------------------------
  -- Task DataTypes
  -- ----------------------------------------------------------------------------
  
  
  -- | the task id (should be unique in the system)
  type TaskId = Integer
  
  
  -- | which type (map, combine, reduce)
  data TaskType = TTSplit | TTMap | TTCombine | TTReduce | TTError
    deriving (Show, Eq, Ord)
  
  instance Binary TaskType where
    put (TTSplit)   = putWord8 1
    put (TTMap)     = putWord8 2
    put (TTCombine) = putWord8 3
    put (TTReduce)  = putWord8 4
    put (TTError)   = putWord8 0
    get
      = do
        t <- getWord8
        case t of
         1 -> return (TTSplit)
         2 -> return (TTMap)
         3 -> return (TTCombine)
         4 -> return (TTReduce)
         _ -> return (TTError)
  
  
  -- | the task state
  data TaskState = TSIdle | TSSending | TSInProgress | TSCompleted | TSFinished | TSError
    deriving (Show, Eq, Ord, Enum)
  
  instance Binary TaskState where
    put (TSIdle)       = putWord8 1
    put (TSSending)    = putWord8 2
    put (TSInProgress) = putWord8 3
    put (TSCompleted)  = putWord8 4
    put (TSFinished)   = putWord8 5
    put (TSError)      = putWord8 0
    get
      = do
        t <- getWord8
        case t of
          1 -> return (TSIdle)
          2 -> return (TSSending)
          3 -> return (TSInProgress)
          4 -> return (TSCompleted)
          5 -> return (TSFinished)
          _ -> return (TSError)
  
  getNextTaskState :: TaskState -> TaskState
  getNextTaskState TSError    = TSError
  getNextTaskState TSFinished = TSFinished
  getNextTaskState s          = succ s
  
  
  data TaskOutputType = TOTRawTuple | TOTFile
    deriving (Show, Read, Eq, Ord, Enum)
    
  instance Binary TaskOutputType where
    put (TOTRawTuple) = putWord8 0
    put (TOTFile)     = putWord8 1
    get
      = do
       t <- getWord8
       case t of
         0 -> return (TOTRawTuple)
         _ -> return (TOTFile)
  
  instance XmlPickler TaskOutputType where
    xpickle = xpAttr "output" $ xpPrim
  
  
  -- | the TaskData, contains all information to do the task
  data TaskData = TaskData {
      td_JobId        :: ! JobId
    , td_TaskId       :: ! TaskId
    , td_Type         :: ! TaskType
    , td_State        :: ! TaskState
    , td_Option       :: ! B.ByteString
    , td_PartValue    :: ! (Maybe Int) 
    , td_Input        :: ! (Int, [FunctionData])
    , td_Output       :: ! [(Int,[FunctionData])]
    , td_OutputType   :: ! TaskOutputType
    , td_Action       :: ! ActionName
    } deriving (Show, Eq, Ord)
  
  instance Binary TaskData where
    put (TaskData jid tid tt ts opt pv i o ot a)
      = put jid >> put tid >> put tt >> put ts >> put opt >> put pv >> put i >> put o >> put ot >> put a
    get
      = do
        jid <- get
        tid <- get
        tt  <- get
        ts  <- get
        opt <- get
        pv  <- get
        i   <- get
        o   <- get
        ot  <- get
        a   <- get
        return (TaskData jid tid tt ts opt pv i o ot a)
  
  
  
  
  -- ----------------------------------------------------------------------------
  -- Job Datatypes
  -- ----------------------------------------------------------------------------
  
  
  -- | the job id (should be unique in the system)
  type JobId = Integer
  
  
  -- | the job state
  data JobState = JSPlanned | JSIdle | JSSplit | JSMap | JSCombine | JSReduce | JSCompleted | JSFinished | JSError
    deriving(Show, Eq, Ord, Enum)
  
  instance Binary JobState where
    put (JSPlanned)    = putWord8 1
    put (JSIdle)       = putWord8 2
    put (JSSplit)      = putWord8 3
    put (JSMap)        = putWord8 4
    put (JSCombine)    = putWord8 5
    put (JSReduce)     = putWord8 6
    put (JSCompleted)  = putWord8 7
    put (JSFinished)   = putWord8 8
    put (JSError)      = putWord8 0
    get
      = do
        t <- getWord8
        case t of
          1 -> return (JSPlanned)
          2 -> return (JSIdle)
          3 -> return (JSSplit)
          4 -> return (JSMap)
          5 -> return (JSCombine)
          6 -> return (JSReduce)
          7 -> return (JSCompleted)
          8 -> return (JSFinished)
          _ -> return (JSError)
  
  getNextJobState :: JobState -> JobState
  getNextJobState JSError    = JSError
  getNextJobState JSFinished = JSFinished
  getNextJobState s          = succ s
  
  getPrevJobState :: JobState -> JobState
  getPrevJobState JSIdle  = JSIdle
  getPrevJobState JSError = JSError
  getPrevJobState s       = pred s
  
  fromJobStatetoTaskType :: JobState -> Maybe TaskType
  fromJobStatetoTaskType JSSplit   = Just TTSplit
  fromJobStatetoTaskType JSMap     = Just TTMap
  fromJobStatetoTaskType JSCombine = Just TTCombine
  fromJobStatetoTaskType JSReduce  = Just TTReduce
  fromJobStatetoTaskType _         = Nothing
  
  
  
  type OutputMap = Map.Map JobState (AMap.AccuMap Int FunctionData)
  
  data JobAction = JobAction {
      ja_Name   :: ! ActionName
    , ja_Output :: ! TaskOutputType
    , ja_Count  :: ! Int
    } deriving (Show, Eq)
  
  instance Binary JobAction where
    put (JobAction n o c) = put n >> put o >> put c
    get 
      = do
        n <- get
        o <- get
        c <- get
        return (JobAction n o c)
  
  instance XmlPickler JobAction where
    xpickle = xpJobAction
    
  xpJobAction :: PU JobAction
  xpJobAction =
    xpWrap
      (\(n, o, c) -> JobAction n o c, 
       \(JobAction n o c) -> (n, o, c)) xpAction
    where
    xpAction = xpTriple (xpFunction) (xpickle) (xpCount)
    xpFunction = xpAttr "name" xpText
    xpCount = xpWrap ((maybe 1 id), Just) $ xpOption $ xpAttr "count" xpickle
        
  
  
  -- | defines a job, this is all data the user has to give to run a job
  data JobInfo = JobInfo {
      ji_Description       :: ! String
    , ji_Option            :: ! B.ByteString
    , ji_SplitAction       :: ! (Maybe JobAction)
    , ji_MapAction         :: ! (Maybe JobAction)
    , ji_CombineAction     :: ! (Maybe JobAction)
    , ji_ReduceAction      :: ! (Maybe JobAction)
    , ji_NumOfResults      :: ! (Maybe Int)
    , ji_Input             :: ! [FunctionData]
    } deriving (Show, Eq)
  
  instance Binary JobInfo where
    put (JobInfo d opt sa ma ca ra nor i)
      = put d >> put opt >> 
        put sa >> put ma >> put ca >> put ra >> put nor >> put i
    get
      = do
        d   <- get
        opt <- get
        sa  <- get
        ma  <- get
        ca  <- get
        ra  <- get
        nor  <- get
        i   <- get
        return (JobInfo d opt sa ma ca ra nor i)
     
  instance XmlPickler JobInfo where
    xpickle = xpJobInfo
    
  xpJobInfo :: PU JobInfo
  xpJobInfo = 
      xpElem "jobInfo" $
      xpWrap
        (\(n, (sa,ma,ca,ra), nor, i) -> JobInfo n (encode ()) sa ma ca ra nor i, 
         \(JobInfo n _ sa ma ca ra nor i) -> (n, (sa,ma,ca,ra), nor, i)) xpJob
      where
      xpJob = 
        xp4Tuple
          (xpAttr "name" xpText0)
          (xpActions)
          (xpCount)
          (xpFunctionDataList)
      xpActions = xp4Tuple (xpSplitAction) (xpMapAction) (xpCombineAction) (xpReduceAction)
      xpSplitAction   = xpOption $ xpElem "split" $ xpJobAction
      xpMapAction     = xpOption $ xpElem "map" $ xpJobAction
      xpCombineAction = xpOption $ xpElem "combine" $ xpJobAction
      xpReduceAction  = xpOption $ xpElem "reduce" $ xpJobAction
      xpFunctionDataList
        = xpWrap (filterEmptyList,setEmptyList) $ xpOption $
          xpElem "inputList" $ xpList xpFunctionData
      xpCount = xpOption $ xpAttr "numOfResults" xpickle
      
  
  -- | the job data, include the user-input and some additional control-data
  data JobData = JobData {
      jd_JobId       :: JobId
    , jd_State       :: JobState
    , jd_OutputMap   :: OutputMap
    , jd_Info        :: JobInfo
    , jd_startTime   :: UTCTime
    , jd_endTime     :: UTCTime
    , jd_Result      :: JobResultContainer
    }
  
  instance Show JobData where
    show (JobData jid state om _ t1 t2 _) =
      "JobId:\t" ++ show jid ++ "\n" ++
      "State:\t" ++ show state ++ "\n" ++
      "Info:\tJobInfo\n" ++
      "StartTime:\t" ++  show t1 ++ "\n" ++
      "EndTime:\t" ++ show t2 ++ "\n" ++
      "OutputMap:\n" ++ show om ++ "\n"
  
  
  data JobResultContainer = JobResultContainer (MVar JobResult)
  
  instance Show JobResultContainer where
    show _ = "JobResultContainer"
  
  -- | the result of the job, given by the master
  data JobResult = JobResult {
      jr_Output        :: [FunctionData]
    } deriving (Show)
  
  instance Binary JobResult where
    put (JobResult o) = put o
    get
      = do
        o <- get
        return (JobResult o)
  
  
  
  -- ----------------------------------------------------------------------------
  -- Reader and Writer
  -- ----------------------------------------------------------------------------
  
  -- | the ActionEnvironment contains all data that might be needed
  --   during an action. So far, it only keeps the current task data and
  --   a reference to the global filesystem and the options. 
  --   This is a good place to implement counters for the map-reduce-system 
  --   or other stuff...
  data ActionEnvironment = ActionEnvironment {
      ae_TaskData   :: TaskData
    , ae_FileSystem :: FS.FileSystem
    }
  
  mkActionEnvironment :: TaskData -> FS.FileSystem -> ActionEnvironment
  mkActionEnvironment td fs = ActionEnvironment td fs
  
  
  type InputReader k1 v1 = B.ByteString -> IO [(k1,v1)]
  
  type OutputWriter k2 v2 = [(k2,v2)] -> IO B.ByteString
  
  readConnector
    :: (NFData k1, NFData v1, Binary k1, Binary v1)
    => InputReader k1 v1
    -> ActionEnvironment
    -> [FunctionData] 
    -> IO [(k1,v1)]
  readConnector ic ae ls
    = do
      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 t = return $ Just $ [decode t] 
  {-    readInput fs (FileFunctionData f)
        = do        
          debugM localLogger $ "loadInputList: getting content for: " ++ f
          mbc <- FS.getFileContent f fs
          if isNothing mbc
            then do
              debugM localLogger $ "loadInputList: no content found"
              return Nothing
            else do
              d <- ic $ fromJust mbc
              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)
    => OutputWriter k2 v2
    -> ActionEnvironment
    -> [(Int,[(k2,v2)])] 
    -> IO [(Int,[FunctionData])]
  writeConnector oc ae ls
    = do
      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]))
      writeOutput _  TOTRawTuple (i,ts) 
        = return $ Just $ (i,bs)
        where
        bs = map (\t -> TupleFunctionData $ encode t) ts
      -- TODO exception werfen 
      writeOutput fs _ (i,ts)
        = do
          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
  
  
  defaultInputReader :: (NFData v1, NFData k1, Binary k1, Binary v1) => InputReader k1 v1
  defaultInputReader = return . parseByteStringToList
  
  
  defaultOutputWriter :: (NFData v2, NFData k2, Binary k2, Binary v2) => OutputWriter k2 v2
  defaultOutputWriter = return . listToByteString 
  --  = return $ B.concat $ map encode ls
  
  
  type ActionName = String
  
  type ActionInfo = String
  
  data ActionConfiguration a k1 v1 k2 v2 v3 v4
    = ActionConfiguration {
      ac_Name          :: ActionName
    , ac_Info          :: ActionInfo
    , ac_OptEncoder    :: OptionsEncoder a
    , ac_OptDecoder    :: OptionsDecoder a
    , ac_InputEncoder  :: InputEncoder k1 v1
    , ac_OutputDecoder :: OutputDecoder k2 v4
    , ac_Split         :: Maybe (SplitConfiguration a k1 v1)
    , ac_Map           :: Maybe (MapConfiguration a k1 v1 k2 v2)
    , ac_Combine       :: Maybe (ReduceConfiguration a k2 v2 v3)
    , ac_Reduce        :: Maybe (ReduceConfiguration a k2 v3 v4)
    }  
  
  
  data SplitConfiguration a k1 v1
    = SplitConfiguration {
      sc_Function :: SplitFunction a k1 v1
    , sc_Reader   :: InputReader k1 v1
    , sc_Writer   :: OutputWriter k1 v1
    }
  
  
  data MapConfiguration a k1 v1 k2 v2
    = MapConfiguration {
      mc_Function  :: MapFunction a k1 v1 k2 v2
    , mc_Partition :: MapPartition a k2 v2
    , mc_Reader    :: InputReader k1 v1
    , mc_Writer    :: OutputWriter k2 v2
    }
  
  
  data ReduceConfiguration a k2 v3 v4
    = ReduceConfiguration {
      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
    :: (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
    = ActionConfiguration
        name
        ""
        defaultOptionsEncoder
        defaultOptionsDecoder
        defaultInputEncoder
        defaultOutputDecoder
        (Just defaultSplitConfiguration)
        Nothing
        Nothing
        Nothing
  
  
  defaultSplitConfiguration
    :: (NFData v1, NFData k1, Binary a, Binary k1, Binary v1)
    => SplitConfiguration a k1 v1
  defaultSplitConfiguration 
    = SplitConfiguration
        defaultSplit
        defaultInputReader
        defaultOutputWriter 
  
  
  defaultMapConfiguration
    :: (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
    = MapConfiguration
        fct
        defaultPartition
        defaultInputReader
        defaultOutputWriter
  
  
  defaultReduceConfiguration
    :: (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
    = ReduceConfiguration
        defaultMerge
        fct
        defaultPartition
        defaultInputReader
        defaultOutputWriter
  
  
  defaultSplit
    :: (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..]]
      is = map (\a -> [a]) ls 
      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
  defaultPartition _ _ 1 ls
    -- To make it faster, wenn no partition is used
    = return [(1,ls)]
  defaultPartition _ _ n ls
    = do
      -- calculate partition-Values
      let markedList = map (\t@(k,_) -> (hash' k,t)) ls
      -- merge them
      
      -- TODO this might change (revert) the order of the Elements...
      let resultList = AMap.toList $ AMap.fromTupleList markedList
      return resultList
      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
  
  
  defaultMerge
    :: (Ord k2, Binary k2, Binary v2)
    => ReduceMerge a k2 v2
  defaultMerge _ _ ls
    = return $ AMap.toList $ AMap.fromTupleList ls
  
  
  
  data ActionData
    = ActionData {
      ad_Name    :: ActionName
    , ad_Info    :: ActionInfo
    , ad_Split   :: Maybe BinarySplitAction
    , ad_Map     :: Maybe BinaryMapAction
    , ad_Combine :: Maybe BinaryReduceAction
    , ad_Reduce  :: Maybe BinaryReduceAction
    }
  
  instance KMap.Key ActionData where
    getKey = ad_Name
  
  instance Show ActionData where
    show (ActionData n i _ _ _ _) = "{ActionData name:\"" ++ n ++ "\" info:\"" ++ i ++ "\"}"
  
  
  type ActionMap = KMap.KeyMap ActionData
  
  type OptionsEncoder a = a -> B.ByteString
  type OptionsDecoder a = B.ByteString -> a
  
  defaultOptionsEncoder :: (Binary a) => OptionsEncoder a
  defaultOptionsEncoder = encode
  
  defaultOptionsDecoder :: (Binary a) => OptionsDecoder a
  defaultOptionsDecoder = decode
  
  
  type InputEncoder k1 v1 = [(k1,v1)] -> [FS.FileId] -> [FunctionData]
  type OutputDecoder k2 v4 = [FunctionData] -> ([(k2,v4)],[FS.FileId])
  
  defaultInputEncoder :: (Binary k1, Binary v1) => InputEncoder k1 v1
  defaultInputEncoder ls1 ls2 = ls1' ++ ls2'
    where
    ls1' = map (\t -> TupleFunctionData (encode t)) ls1
    ls2' = map (\f -> FileFunctionData f) ls2
  
  defaultOutputDecoder :: (Binary k2, Binary v4) => OutputDecoder k2 v4
  defaultOutputDecoder ls = (ls1, ls2)
    where
    ls1 = map (\t -> decode t) $ catMaybes ls1'
    ls2 = catMaybes ls2'
    (ls1',ls2') = unzip $ map (splitter) ls
    splitter (TupleFunctionData t) = (Just t, Nothing)
    splitter (FileFunctionData f)  = (Nothing, Just f)
    
  
  getActionForTaskType :: TaskType -> ActionData -> Maybe BinaryReduceAction
  getActionForTaskType TTSplit   ad = ad_Split ad
  getActionForTaskType TTMap     ad = ad_Map ad
  getActionForTaskType TTCombine ad = ad_Combine ad
  getActionForTaskType TTReduce  ad = ad_Reduce ad
  getActionForTaskType _         _  = Nothing
  
    
  
  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)
    => ActionConfiguration a k1 v1 k2 v2 v3 v4 
    -> ActionData
  readActionConfiguration (ActionConfiguration n i _ optDec _ _ sc mc cc rc)
    = ActionData n i sf mf cf rf
    where
    sf = maybe Nothing (\(SplitConfiguration f ir ow) -> Just $ performSplitAction optDec f ir ow) sc 
    mf = maybe Nothing (\(MapConfiguration f p ir ow) -> Just $ performMapAction optDec f p ir ow) mc
    cf = maybe Nothing (\(ReduceConfiguration m f p ir ow) -> Just $ performReduceAction optDec m f p ir ow) cc
    rf = maybe Nothing (\(ReduceConfiguration m f p ir ow) -> Just $ performReduceAction optDec m f p ir ow) rc
  
  
  
  createJobInfoFromConfiguration
    :: ActionConfiguration a k1 v1 k2 v2 v3 v4
    -> a              -- ^ options
    -> [(k1,v1)]      -- ^ input (Tuples)
    -> [FS.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)
    -> JobInfo
  createJobInfoFromConfiguration (ActionConfiguration n _ optEnc _ inEnc _ sConf mConf cConf rConf) opts' ls1 ls2 sCnt mCnt rCnt nor' rt
    = JobInfo n opts sa ma ca ra nor i
    where
    opts = optEnc opts'
    sa   = maybe Nothing (\_ -> Just $ JobAction n TOTFile sCnt) sConf
    ma   = maybe Nothing (\_ -> Just $ JobAction n TOTFile mCnt) mConf
    ca   = maybe Nothing (\_ -> Just $ JobAction n TOTFile mCnt) cConf
    ra   = maybe Nothing (\_ -> Just $ JobAction n rt      rCnt) rConf
    nor  = Just nor'
    i    = inEnc ls1 ls2
    
      
  createListsFromJobResult
    :: ActionConfiguration a k1 v1 k2 v2 v3 v4
    -> JobResult
    -> ([(k2,v4)],[FS.FileId])
  createListsFromJobResult ac jr = (ac_OutputDecoder ac) (jr_Output jr)
      
  -- ----------------------------------------------------------------------------
  -- SplitAction
  -- ----------------------------------------------------------------------------
  
  -- | general SplitAction
  type SplitAction a k1 v1 = ActionEnvironment -> a -> Int -> [(k1,v1)] -> IO [(Int, [(k1,v1)])]
  
  -- | SplitAction on ByteStrings
  type BinarySplitAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])] 
  
  type SplitFunction a k1 v1 = SplitAction a k1 v1
  
  performSplitAction
    :: (NFData k1, NFData v1, Binary a, Binary k1, Binary v1, Show k1, Show v1)
    => OptionsDecoder a
    -> SplitFunction a k1 v1
    -> InputReader k1 v1
    -> OutputWriter k1 v1
    -> ActionEnvironment
    -> B.ByteString
    -> Maybe Int
    -> (Int,[FunctionData])
    -> IO [(Int, [FunctionData])]
  performSplitAction optDec fct reader writer env opts n (i,ls)
    = do
      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
  
  
  -- ----------------------------------------------------------------------------
  -- MapAction
  -- ----------------------------------------------------------------------------
  
  
  -- | general MapAction
  type MapAction a k1 v1 k2 v2 = ActionEnvironment -> a -> Int -> [(k1,v1)] -> IO [(Int, [(k2,v2)])]
  
  -- | MapAction on ByteStrings
  type BinaryMapAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])] 
  
  type MapFunction a k1 v1 k2 v2 = ActionEnvironment -> a -> k1 -> v1 -> IO [(k2, v2)]
  
  type MapPartition a k2 v2 = ActionEnvironment -> a -> Int -> [(k2,v2)] -> IO [(Int, [(k2,v2)])]
      
  
  performMapAction
    :: (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
    -> InputReader k1 v1
    -> OutputWriter k2 v2
    -> ActionEnvironment
    -> B.ByteString
    -> Maybe Int
    -> (Int,[FunctionData])
    -> IO [(Int, [FunctionData])]
  performMapAction optDec fct part reader writer env opts n (i,ls)
    = do
      -- decode the options
      let a = optDec opts
      
      infoM localLogger "performMapAction"
      putTimeStamp "Begin performMapAction"
      
      infoM localLogger $ "reading inputList: " ++ show (i,ls)
      inputList <- readConnector reader env ls
      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: begin"
      outputList <- writeConnector writer env partedList
      infoM localLogger "writing outputlist: done"    
      putTimeStamp "End performMapAction"
      return outputList
  
  
  
  
  
  -- ----------------------------------------------------------------------------
  -- Combine- / ReduceTask
  -- ----------------------------------------------------------------------------
  
  -- | general MapAction
  type ReduceAction a k2 v2 v3 = ActionEnvironment -> a -> Int -> [(k2,v2)] -> IO [(Int, [(k2,v3)])]
  
  -- | MapAction on ByteStrings
  type BinaryReduceAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])] 
  
  type ReduceMerge a k2 v2 = ActionEnvironment -> a -> [(k2,v2)] -> IO [(k2,[v2])]
  
  type ReduceFunction a k2 v2 v3 = ActionEnvironment -> a -> k2 -> [v2] -> IO (Maybe v3)
  
  type ReducePartition a k2 v3 = ActionEnvironment -> a -> Int -> [(k2,v3)] -> IO [(Int, [(k2,v3)])]
      
  
  performReduceAction
    :: (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
    -> ReduceFunction a k2 v2 v3
    -> ReducePartition a k2 v3
    -> InputReader k2 v2
    -> OutputWriter k2 v3
    -> ActionEnvironment
    -> B.ByteString
    -> Maybe Int
    -> (Int,[FunctionData])
    -> IO [(Int, [FunctionData])]
  performReduceAction optDec merge fct part reader writer env opts n (i,ls)
    = do
      -- decode the options
      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: begin"
      outputList <- writeConnector writer env partedList
      infoM localLogger "writing outputlist: done"
      
      putTimeStamp "End performReduceAction"
      return outputList
      where
        performReduceFunction a k2 v2s
          = do
            mbV3 <- fct env a k2 v2s
            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)
  