content: {-# LANGUAGE DeriveDataTypeable #-}
  
  -- | This module provides data dependency resolution and
  -- fault tolerance via /promises/ (known elsewhere as /futures/).
  -- It's implemented in terms of the "Remote.Process" module.
  module Remote.Task (
                     -- * Tasks and promises
                     TaskM, Promise, PromiseList(..),
                     runTask,
                     newPromise, newPromiseAt, newPromiseNear, newPromiseHere, newPromiseAtRole,
                     toPromise, toPromiseAt, toPromiseNear, toPromiseImm,
                     readPromise, 
  
                     -- * MapReduce
                     MapReduce(..),
                     mapReduce,
  
                     -- * Useful auxilliaries
                     chunkify,
                     shuffle,
                     tsay,
                     tlogS,
                     Locality(..),
                     TaskException(..),
  
                     -- * Internals, not for general use
                     __remoteCallMetaData, serialEncodeA, serialDecodeA
                     ) where
  
  import Remote.Reg (putReg,getEntryByIdent,RemoteCallMetaData)
  import Remote.Encoding (serialEncodePure,hGetPayload,hPutPayload,Payload,getPayloadContent,Serializable,serialDecode,serialEncode)
  import Remote.Process (roundtripQuery, ServiceException(..), TransmitStatus(..),diffTime,getConfig,Config(..),matchProcessDown,terminate,nullPid,monitorProcess,TransmitException(..),MonitorAction(..),ptry,LogConfig(..),getLogConfig,setNodeLogConfig,nodeFromPid,LogLevel(..),LogTarget(..),logS,getLookup,say,LogSphere,NodeId,ProcessM,ProcessId,PayloadDisposition(..),getSelfPid,getSelfNode,matchUnknownThrow,receiveWait,receiveTimeout,roundtripResponse,roundtripResponseAsync,roundtripQueryImpl,match,makePayloadClosure,spawn,spawnLocal,spawnLocalAnd,setDaemonic,send,makeClosure)
  import Remote.Closure (Closure(..))
  import Remote.Peer (getPeers)
  
  import Data.Dynamic (Dynamic, toDyn, fromDynamic)
  import System.IO (withFile,IOMode(..))
  import System.Directory (renameFile)
  import Data.Binary (Binary,get,put,putWord8,getWord8)
  import Control.Exception (SomeException,Exception,throw)
  import Data.Typeable (Typeable)
  import Control.Applicative (Applicative(..))
  import Control.Monad (liftM,when)
  import Control.Monad.Trans (liftIO)
  import Control.Concurrent.MVar (MVar,modifyMVar,modifyMVar_,newMVar,newEmptyMVar,takeMVar,putMVar,readMVar,withMVar)
  import qualified Data.Map as Map (Map,insert,lookup,empty,insertWith',toList)
  import Data.List ((\\),union,nub,groupBy,sortBy,delete)
  import Data.Time (UTCTime,getCurrentTime)
  
  -- imports required for hashClosure; is there a lighter-weight of doing this?
  import Data.Digest.Pure.MD5 (md5)
  import Data.ByteString.Lazy.UTF8 (fromString)
  import qualified Data.ByteString.Lazy as B (concat)
  
  ----------------------------------------------
  -- * Promises and tasks
  ----------------------------------------------
  
  
  type PromiseId = Integer
  
  type Hash = String
  
  data PromiseList a = PlChunk a (Promise (PromiseList a))
                     | PlNil deriving Typeable
  
  instance (Serializable a) => Binary (PromiseList a) where
     put (PlChunk a p) = putWord8 0 >> put a >> put p
     put PlNil = putWord8 1
     get = do w <- getWord8
              case w of
                 0 -> do a <- get
                         p <- get
                         return $ PlChunk a p
                 1 -> return PlNil
  
  -- | The basic data type for expressing data dependence
  -- in the 'TaskM' monad. A Promise represents a value that
  -- may or may not have been computed yet; thus, it's like
  -- a distributed thunk (in the sense of a non-strict unit
  -- of evaluation). These are created by 'newPromise' and friends,
  -- and the underlying value can be gotten with 'readPromise'.
  data Promise a = PromiseBasic { _psRedeemer :: ProcessId, _psId :: PromiseId } 
                 | PromiseImmediate a deriving Typeable
  -- psRedeemer should maybe be wrapped in an IORef so that it can be updated in case of node failure
  
  instance (Serializable a) => Binary (Promise a) where
     put (PromiseBasic a b) = putWord8 0 >> put a >> put b
     put (PromiseImmediate a) = putWord8 1 >> put a
     get = do a <- getWord8
              case a of
                0 -> do b <- get
                        c <- get
                        return $ PromiseBasic b c
                1 -> do b <- get
                        return $ PromiseImmediate b
  
  -- | Stores the data produced by a promise, in one of its
  -- various forms. If it's currently in memory, we keep it
  -- as a payload, to be decoded by its ultimate user (who
  -- of course has the static type information), the time
  -- it was last touched (so we know when to flush it),
  -- and perhaps also a decoded version, so that it doesn't
  -- need to be decoded repeatedly: this makes this go a lot
  -- faster. If it's been flushed to disk, we keep track of
  -- where, and if the promise didn't complete, but threw
  -- an exception during its execution, we mark there here
  -- as well: the exception will be propagated to
  -- dependents.
  data PromiseStorage = PromiseInMemory PromiseData UTCTime (Maybe Dynamic)
                      | PromiseOnDisk FilePath
                      | PromiseException String
  
  type PromiseData = Payload
  {- UNUSED
  type TimeStamp = UTCTime
  -}
  
  -- | Keeps track of what we know about currently running promises.
  -- The closure and locality and provided by the initial call to
  -- newPromise, the nodeboss is where it is currently running.
  -- We need this info to deal with complaints.
  data PromiseRecord = PromiseRecord ProcessId (Closure PromiseData) Locality
  
  data MasterState = MasterState
       {
           -- | Promise IDs are allocated serially from here
           msNextId :: PromiseId,
   
           -- | All currently known nodes, with the role, node ID, and node boss. Updated asychronously by prober thread
           msNodes :: MVar [(String,NodeId,ProcessId)],
  
           -- | Given a nodeboss, which promises belong to it. Not sure what this is good for
           msAllocation :: Map.Map ProcessId [PromiseId],
  
           -- | Given a promise, what do we know about it. Include its nodeboss, its closure, and its locality preference
           msPromises :: Map.Map PromiseId PromiseRecord,
  
           -- | The locality preference of new worker tasks, if not specified otherwise
           msDefaultLocality :: Locality
       }
  
  data MmNewPromise = MmNewPromise (Closure Payload) Locality Queueing deriving (Typeable)
  instance Binary MmNewPromise where 
    get = do a <- get
             l <- get
             q <- get
             return $ MmNewPromise a l q
    put (MmNewPromise a l q) = put a >> put l >> put q
  
  data MmNewPromiseResponse = MmNewPromiseResponse ProcessId PromiseId 
                            | MmNewPromiseResponseFail deriving (Typeable)
  instance Binary MmNewPromiseResponse where 
     put (MmNewPromiseResponse a b) =
             do putWord8 0
                put a
                put b
     put MmNewPromiseResponseFail = putWord8 1
     get = do a <- getWord8
              case a of
                0 -> do b <- get 
                        c <- get
                        return $ MmNewPromiseResponse b c
                1 -> return MmNewPromiseResponseFail
  
  data MmStatus = MmStatus deriving Typeable
  instance Binary MmStatus where
    get = return MmStatus
    put MmStatus = put ()
  
  data MmStatusResponse = MmStatusResponse [NodeId] (Map.Map ProcessId [PromiseId]) deriving Typeable
  instance Binary MmStatusResponse where
    get = do a <- get
             b <- get
             return $ MmStatusResponse a b
    put (MmStatusResponse a b) = put a >> put b
  
  data MmComplain = MmComplain ProcessId PromiseId deriving (Typeable)
  instance Binary MmComplain where 
     put (MmComplain a b) = put a >> put b
     get = do a <- get
              b <- get
              return $ MmComplain a b
  
  data MmComplainResponse = MmComplainResponse ProcessId deriving (Typeable)
  instance Binary MmComplainResponse where
     put (MmComplainResponse a) = put a
     get = do a <- get
              return $ MmComplainResponse a
  
  data TmNewPeer = TmNewPeer NodeId deriving (Typeable)
  instance Binary TmNewPeer where 
     get = do a <- get
              return $ TmNewPeer a
     put (TmNewPeer nid) = put nid
  
  data NmStart = NmStart PromiseId (Closure Payload) Queueing deriving (Typeable)
  instance Binary NmStart where 
    get = do a <- get
             b <- get
             c <- get
             return $ NmStart a b c
    put (NmStart a b c) = put a >> put b >> put c
  
  data NmStartResponse = NmStartResponse Bool deriving (Typeable)
  instance Binary NmStartResponse where 
    get = do a <- get
             return $ NmStartResponse a
    put (NmStartResponse a) = put a
  
  data NmRedeem = NmRedeem PromiseId deriving (Typeable)
  instance Binary NmRedeem where 
     get = do a <- get
              return $ NmRedeem a
     put (NmRedeem prid) = put prid
  
  data NmRedeemResponse = NmRedeemResponse Payload
                        | NmRedeemResponseUnknown
                        | NmRedeemResponseException deriving (Typeable)
  instance Binary NmRedeemResponse where 
     get = do a <- getWord8
              case a of
                0 -> do b <- get
                        return $ NmRedeemResponse b
                1 -> return NmRedeemResponseUnknown
                2 -> return NmRedeemResponseException
     put (NmRedeemResponse a) = putWord8 0 >> put a
     put (NmRedeemResponseUnknown) = putWord8 1
     put (NmRedeemResponseException) = putWord8 2
  
  data TaskException = TaskException String deriving (Show,Typeable)
  instance Exception TaskException
  
  -- | (Currently ignored.)
  data Queueing = QuNone
                | QuExclusive
                | QuSmall deriving (Typeable,Ord,Eq)
  
  defaultQueueing :: Queueing
  defaultQueueing = QuNone
  
  instance Binary Queueing where
     put QuNone = putWord8 0
     put QuExclusive = putWord8 1
     put QuSmall = putWord8 2
     get = do a <- getWord8
              case a of
                 0 -> return QuNone
                 1 -> return QuExclusive
                 2 -> return QuSmall
  
  -- | A specification of preference
  -- of where a promise should be allocated,
  -- among the nodes visible to the master.
  data Locality = LcUnrestricted -- ^ The promise can be placed anywhere.
                | LcDefault -- ^ The default preference is applied, which is for nodes having a role of NODE of WORKER
                | LcByRole [String] -- ^ Nodes having the given roles will be preferred
                | LcByNode [NodeId] -- ^ The given nodes will be preferred
  
  instance Binary Locality where
     put LcUnrestricted = putWord8 0
     put LcDefault = putWord8 1
     put (LcByRole a) = putWord8 2 >> put a
     put (LcByNode a) = putWord8 3 >> put a
     get = do a <- getWord8
              case a of
                 0 -> return LcUnrestricted
                 1 -> return LcDefault
                 2 -> do r <- get
                         return $ LcByRole r
                 3 -> do r <- get
                         return $ LcByNode r
  
  defaultLocality :: Locality
  defaultLocality = LcByRole ["WORKER","NODE"]
  
  taskError :: String -> a
  taskError s = throw $ TaskException s
  
  serialEncodeA :: (Serializable a) => a -> TaskM Payload
  serialEncodeA = liftTask . liftIO . serialEncode
  
  serialDecodeA :: (Serializable a) => Payload -> TaskM (Maybe a)
  serialDecodeA = liftTask . liftIO . serialDecode
  
  monitorTask :: ProcessId -> ProcessId -> ProcessM TransmitStatus
  monitorTask monitor monitee
     = do res <- ptry $ monitorProcess monitor monitee MaMonitor 
          case res of
             Right _ -> return QteOK
             Left (ServiceException e) -> return $ QteOther e
  
  roundtripImpl :: (Serializable a, Serializable b) => ProcessId -> a -> ProcessM (Either TransmitStatus b)
  roundtripImpl pid dat = roundtripQueryImpl 0 PldUser pid dat id []
  
  roundtrip :: (Serializable a, Serializable b) => ProcessId -> a -> TaskM (Either TransmitStatus b)
  roundtrip apid dat = 
     TaskM $ \ts ->
       case Map.lookup apid (tsMonitoring ts) of
         Nothing -> do mypid <- getSelfPid
                       res0 <- monitorTask mypid apid
                       case res0 of
                          QteOK -> 
                            do res <- roundtripImpl apid dat
                               return (ts {tsMonitoring=Map.insert apid () (tsMonitoring ts)},res)
                          _ -> return (ts,Left res0)
         Just _ -> do res <- roundtripImpl apid dat
                      return (ts,res)
  
  -- roundtrip a b = liftTask $ roundtripQueryUnsafe PldUser a b
  
  spawnDaemonic :: ProcessM () -> ProcessM ProcessId
  spawnDaemonic p = spawnLocalAnd p setDaemonic
  
  runWorkerNode :: ProcessId -> NodeId -> ProcessM ProcessId
  runWorkerNode masterpid nid = 
       do clo <- makeClosure "Remote.Task.runWorkerNode__impl" (masterpid) :: ProcessM (Closure (ProcessM ()))
          spawn nid clo 
  
  runWorkerNode__impl :: Payload -> ProcessM ()
  runWorkerNode__impl pl = 
     do setDaemonic -- maybe it's good to have the node manager be daemonic, but prolly not. If so, the MASTERPID must be terminated when user-provided MASTERPROC ends
        mpid <- liftIO $ serialDecode pl
        case mpid of
           Just masterpid -> handler masterpid
           Nothing -> error "Failure to extract in rwn__impl"
   where handler masterpid = startNodeManager masterpid
  
  passthrough__implPl :: Payload -> TaskM Payload
  passthrough__implPl pl = return pl
  
  passthrough__closure :: (Serializable a) => a -> Closure (TaskM a)
  passthrough__closure a = Closure "Remote.Task.passthrough__impl" (serialEncodePure a)
  
  __remoteCallMetaData :: RemoteCallMetaData
  __remoteCallMetaData x = putReg runWorkerNode__impl "Remote.Task.runWorkerNode__impl" 
                          (putReg passthrough__implPl "Remote.Task.passthrough__implPl" x)
  
  updatePromiseInMemory :: PromiseStorage -> IO PromiseStorage
  updatePromiseInMemory (PromiseInMemory p _ d) = do utc <- getCurrentTime
                                                     return $ PromiseInMemory p utc d
  updatePromiseInMemory other = return other
  
  makePromiseInMemory :: PromiseData -> Maybe Dynamic -> IO PromiseStorage
  makePromiseInMemory p dyn = do utc <- getCurrentTime
                                 return $ PromiseInMemory p utc dyn
  
  forwardLogs :: Maybe ProcessId -> ProcessM ()
  forwardLogs masterpid = 
          do lc <- getLogConfig
             selfnid <- getSelfNode
             let newlc = lc {logTarget = case masterpid of
                                               Just mp
                                                  | nodeFromPid mp /= selfnid -> LtForward $ nodeFromPid mp
                                               _ -> LtStdout}     
               in setNodeLogConfig newlc
  
  hashClosure :: Closure a -> Hash
  hashClosure (Closure s pl) = show $ md5 $ B.concat [fromString s, getPayloadContent pl]
  
  undiskify :: FilePath -> MVar PromiseStorage -> ProcessM (Maybe PromiseData)
  undiskify fpIn mps =
        do wrap $ liftIO $ modifyMVar mps (\val ->
              case val of
                PromiseOnDisk fp -> 
                  do pl <- withFile fp ReadMode hGetPayload
                     inmem <- makePromiseInMemory pl Nothing
                     return (inmem,Just pl)
                PromiseInMemory payload _ _ -> return (val,Just payload)
                _ -> return (val,Nothing))
    where wrap a = do res <- ptry a
                      case res of
                        Left e -> do logS "TSK" LoCritical $ "Error reading promise from file "++fpIn++": "++show (e::IOError)
                                     return Nothing
                        Right r -> return r
  
  diskify :: FilePath -> MVar PromiseStorage -> Bool -> ProcessM ()
  diskify fp mps reallywrite =
        do cfg <- getConfig
           when (cfgPromiseFlushDelay cfg > 0)
              (handler (cfgPromiseFlushDelay cfg))
      where
           handler delay =
             do _ <- receiveTimeout delay []
                again <- wrap $ liftIO $ modifyMVar mps (\val ->
                     case val of
                         PromiseInMemory payload utc _ ->
                             do now <- getCurrentTime
                                if diffTime now utc > delay
                                   then do when reallywrite $
                                               do liftIO $ withFile tmp WriteMode (\h -> hPutPayload h payload)
                                                  renameFile tmp fp
                                           return (PromiseOnDisk fp,False)
                                   else return (val,True)
                         _      -> return (val,False))
                when again
                   (diskify fp mps reallywrite)
           tmp = fp ++ ".tmp"
           wrap a = do res <- ptry a
                       case res of
                           Left z -> do logS "TSK" LoImportant $ "Error writing promise to disk on file "++fp++": "++show (z::IOError)
                                        return False
                           Right v -> return v
  
  startNodeWorker :: ProcessId -> NodeBossState -> 
                     MVar PromiseStorage -> Closure Payload -> ProcessM ()
  startNodeWorker masterpid nbs mps clo@(Closure cloname cloarg) = 
    do self <- getSelfPid
       _ <- spawnLocalAnd (starter self) (prefix self)
       return ()
    where 
          prefix nodeboss =
            do self <- getSelfPid
               monitorProcess self nodeboss MaLink
               setDaemonic 
          starter nodeboss = -- TODO try to do an undiskify here, if the promise is left over from a previous, failed run
              let initialState = TaskState {tsMaster=masterpid,tsNodeBoss=Just nodeboss, 
                                            tsPromiseCache=nsPromiseCache nbs, tsRedeemerForwarding=nsRedeemerForwarding nbs,
                                            tsMonitoring=Map.empty}
                  tasker = do tbl <- liftTask $ getLookup
                              case getEntryByIdent tbl cloname of
                                Just funval -> 
                                            do val <- funval cloarg
                                               p <- liftTaskIO $ makePromiseInMemory val Nothing
                                               liftTaskIO $ putMVar mps p
                                               cfg <- liftTask $ getConfig
                                               let cachefile = cfgPromisePrefix cfg++hashClosure clo
                                               liftTask $ diskify cachefile mps True
                                Nothing -> taskError $ "Failed looking up "++cloname++" in closure table"
               in do res <- ptry $ runTaskM tasker initialState :: ProcessM (Either SomeException (TaskState,()))
                     case res of
                       Left ex -> liftIO (putMVar mps (PromiseException (show ex))) >> throw ex
                       Right _ -> return ()
  
  data NodeBossState = 
          NodeBossState
          {
             nsPromiseCache :: MVar (Map.Map PromiseId (MVar PromiseStorage)),
             nsRedeemerForwarding :: MVar (Map.Map PromiseId ProcessId)
          }
  
  startNodeManager :: ProcessId -> ProcessM ()
  startNodeManager masterpid = 
    let
        handler :: NodeBossState -> ProcessM a
        handler state = 
          let promisecache = nsPromiseCache state
              nmStart = roundtripResponse (\(NmStart promise clo _queueing) -> 
                 do promisestore <- liftIO $ newEmptyMVar
                    ret <- liftIO $ modifyMVar promisecache 
                      (\pc ->               let newpc = Map.insert promise promisestore pc 
                                             in return (newpc,True))
                    when (ret)
                         (startNodeWorker masterpid state promisestore clo)
                    return (NmStartResponse ret,state)) 
              nmTermination = matchProcessDown masterpid $
                       do forwardLogs Nothing
                          logS "TSK" LoInformation $ "Terminating nodeboss after my master "++show masterpid++" is gone"
                          terminate
              nmRedeem = roundtripResponseAsync (\(NmRedeem promise) ans -> 
                       let answerer = do pc <- liftIO $ readMVar promisecache
                                         case Map.lookup promise pc of
                                           Nothing -> ans NmRedeemResponseUnknown
                                           Just v -> do rv <- liftIO $ readMVar v -- possibly long wait
                                                        case rv of
                                                           PromiseInMemory rrv _ _ -> 
                                                                  do liftIO $ modifyMVar_ v (\_ -> updatePromiseInMemory rv)
                                                                     ans (NmRedeemResponse rrv)
                                                           PromiseOnDisk fp -> do mpd <- undiskify fp v
                                                                                  case mpd of
                                                                                     Nothing -> 
                                                                                       ans (NmRedeemResponseUnknown)
                                                                                     Just a ->
                                                                                       ans (NmRedeemResponse a)
                                                                                  diskify fp v False
                                                           PromiseException _ -> ans NmRedeemResponseException
                        in do _ <- spawnLocal answerer
                              return state) False
           in receiveWait [nmStart, nmRedeem, nmTermination, matchUnknownThrow] >>= handler
     in do forwardLogs $ Just masterpid
           mypid <- getSelfPid
           monitorProcess mypid masterpid MaMonitor
           logS "TSK" LoInformation $ "Starting a nodeboss owned by " ++ show masterpid
           pc <- liftIO $ newMVar Map.empty
           pf <- liftIO $ newMVar Map.empty
           let initState = NodeBossState {nsPromiseCache=pc,nsRedeemerForwarding=pf}
           handler initState
  
  -- | Starts a new context for executing a 'TaskM' environment.
  -- The node on which this function is run becomes a new master
  -- in a Task application; as a result, the application should
  -- only call this function once. The master will attempt to
  -- control all nodes that it can find; if you are going to be
  -- running more than one CH application on a single network,
  -- be sure to give each application a different network
  -- magic (via cfgNetworkMagic). The master TaskM environment
  -- created by this function can then spawn other threads,
  -- locally or remotely, using 'newPromise' and friends.
  
  runTask :: TaskM a -> ProcessM a
  runTask = startMaster
  
  startMaster :: TaskM a -> ProcessM a
  startMaster proc = 
      do mvmaster <- liftIO $ newEmptyMVar
         mvdone <- liftIO $ newEmptyMVar
         master <- runMaster (masterproc mvdone mvmaster)
         liftIO $ putMVar mvmaster master
         liftIO $ takeMVar mvdone
     where masterproc mvdone mvmaster nodeboss = 
             do master <- liftIO $ takeMVar mvmaster
                pc <- liftIO $ newMVar Map.empty
                pf <- liftIO $ newMVar Map.empty
                let initialState = TaskState {tsMaster=master,tsNodeBoss=Just nodeboss,
                                              tsPromiseCache=pc, tsRedeemerForwarding=pf,
                                              tsMonitoring=Map.empty}
                res <- liftM snd $ runTaskM proc initialState
                liftIO $ putMVar mvdone res
  
  {- UNUSED
  type LocationSelector = MasterState -> ProcessM (NodeId,ProcessId)
  -}
  
  selectLocation :: MasterState -> Locality -> ProcessM (Maybe (String,NodeId,ProcessId))
  selectLocation ms locality =
     let nodes = msNodes ms
      in liftIO $ modifyMVar nodes
            (\n -> case n of
                      [] -> return (n,Nothing)
                      _ -> let dflt = (rotate n,Just $ head n)
                               filterify f = case filter f n of
                                               [] -> return dflt
                                               (a:_) -> return ((delete a n) ++ [a],Just a)
                            in  case cond locality of
                                  LcUnrestricted -> return dflt
                                  LcDefault -> return dflt
                                  LcByRole l -> filterify (\(r,_,_) -> r `elem` l)
                                  LcByNode l -> filterify (\(_,r,_) -> r `elem` l))
        where rotate [] = []
              rotate (h:t) = t ++ [h]
              cond l = case l of
                         LcDefault -> msDefaultLocality ms
                         _ -> l
  
  countLocations :: MasterState -> ProcessM Int
  countLocations ms = liftIO $ withMVar (msNodes ms) (\a -> return $ length a)
  
  findPeers :: ProcessM [(String,NodeId)]
  findPeers = liftM (concat . (map (\(role,v) -> [ (role,x) | x <- v] )) . Map.toList) getPeers
  
  sendSilent :: (Serializable a) => ProcessId -> a -> ProcessM ()
  sendSilent pid a = do res <- ptry $ send pid a
                        case res of
                          Left (TransmitException _) -> return ()
                          Right _ -> return ()
  
  {- UNUSED
  getStatus :: TaskM ()
  getStatus = 
     do master <- getMaster
        res <- roundtrip master MmStatus
        case res of
          Left _ -> return ()
          Right (MmStatusResponse nodes promises) ->
                let verboseNodes = intercalate ", " (map show nodes)
                    verbosePromises = intercalate "\n" $ map (\(nb,l) -> (show nb)++" -- "++intercalate "," (map show l)) (Map.toList promises) 
                 in tsay $ "\nKnown nodes: " ++ verboseNodes ++ "\n\nNodebosses: " ++ verbosePromises
  -}
  
  runMaster :: (ProcessId -> ProcessM ()) -> ProcessM ProcessId
  runMaster masterproc = 
    let
       probeOnce nodes seen masterpid =
          do recentlist <- findPeers -- TODO if a node fails to response to a probe even once, it's gone forever; be more flexible
             let newseen = seen `union` recentlist
             let topidlist = recentlist \\ seen
             let cleanOut n = filter (\(_,nid,_) -> nid `elem` (map snd recentlist)) n
             newlypidded <- mapM (\(role,nid) -> 
                               do pid <- runWorkerNode masterpid nid
                                  return (role,nid,pid)) topidlist           
             (_newlist,totalseen) <- liftIO $ modifyMVar nodes (\oldlist ->
                          return ((cleanOut oldlist) ++ newlypidded,(recentlist,newseen)))
             let newlyadded = totalseen \\ seen
             mapM_ (\nid -> sendSilent masterpid (TmNewPeer nid)) (map snd newlyadded)
             return totalseen
       proberDelay = 10000000 -- how often do we check the network to see what nodes are available?
       prober nodes seen masterpid =
          do totalseen <- probeOnce nodes seen masterpid
             _ <- receiveTimeout proberDelay [matchUnknownThrow]
             prober nodes totalseen masterpid
       master state = 
          let 
              tryAlloc clo promiseid locality queueing =
                do ns <- selectLocation state locality
                   case ns of
                      Nothing -> do logS "TSK" LoCritical "Attempt to allocate a task, but no nodes found"
                                    return Nothing
                      Just (_,nid,nodeboss) -> 
                           do res <- roundtripQuery PldUser nodeboss (NmStart promiseid clo queueing) -- roundtripQuery monitors and then unmonitors, which generates a lot of traffic; we probably don't need to do this
                              case res of
                                Left e -> 
                                   do logS "TSK" LoImportant $ "Failed attempt to start "++show clo++" on " ++show nid ++": "++show e
                                      return Nothing
                                Right (NmStartResponse True) -> return $ Just nodeboss
                                _ -> do logS "TSK" LoImportant $ "Failed attempt to start "++show clo++" on " ++show nid
                                        return Nothing
              basicAllocate clo promiseid locality queueing =
                 do count <- countLocations state
                    res1 <- tryAlloc clo promiseid locality queueing
                    case res1 of 
                       Just _ -> return res1
                       Nothing ->  -- TODO we should try all matching locations before moving on to Unrestricted
                         do res <- stubborn count $ tryAlloc clo promiseid LcUnrestricted queueing
                            case res of
                               Nothing -> do logS "TSK" LoCritical $ "Terminally failed to start "++show clo
                                             return res
                               _ -> return res
              statusMsg = roundtripResponse
                (\x -> case x of
                          MmStatus -> 
                           do thenodes <- liftIO $ readMVar $ msNodes state
                              let knownNodes = map (\(_,n,_) -> n) thenodes
                                  proctree = msAllocation state
                              return (MmStatusResponse knownNodes proctree,state))
              complainMsg = roundtripResponse
                (\x -> case x of
                         MmComplain procid promid -> 
                           case Map.lookup promid (msPromises state) of
                              Nothing -> return (MmComplainResponse nullPid,state) -- failure
                              Just (PromiseRecord curprocid curclo curlocality) 
                                 | curprocid /= procid -> return (MmComplainResponse curprocid,state)
                                 | otherwise -> 
                                     do res <- basicAllocate curclo promid curlocality defaultQueueing
                                        case res of
                                          Nothing -> return (MmComplainResponse nullPid,state) -- failure
                                          Just newprocid ->
                                             let newpromises = Map.insert promid (PromiseRecord newprocid curclo curlocality) (msPromises state)
                                              in return (MmComplainResponse newprocid,state {msPromises=newpromises})) 
              promiseMsg = roundtripResponse 
                (\x -> case x of
                         MmNewPromise clo locality queueing -> 
                            do 
                               let promiseid = msNextId state
                               res <- basicAllocate clo promiseid locality queueing
                               case res of
                                  Just nodeboss -> 
                                          let newstate = state {msAllocation=newAllocation,msPromises=newPromises,msNextId=promiseid+1}
                                              newAllocation = Map.insertWith' (\a b -> nub $ a++b) nodeboss [promiseid] (msAllocation state)
                                              newPromises = Map.insert promiseid (PromiseRecord nodeboss clo locality) (msPromises state)
                                           in return (MmNewPromiseResponse nodeboss promiseid,newstate)
                                  Nothing -> 
                                           return (MmNewPromiseResponseFail,state))
              simpleMsg = match 
                (\x -> case x of
                         TmNewPeer nid -> do logS "TSK" LoInformation $ "Found new peer " ++show nid
                                             return state)
           in receiveWait [simpleMsg, promiseMsg, complainMsg,statusMsg] >>= master -- TODO matchUnknownThrow
     in do nodes <- liftIO $ newMVar []
           selfnode <- getSelfNode
           selfpid <- getSelfPid
           let initState = MasterState {msNextId=0, msAllocation=Map.empty, msPromises=Map.empty, msNodes=nodes, msDefaultLocality = defaultLocality}
           masterpid <- spawnDaemonic (master initState)
           seennodes <- probeOnce nodes [] masterpid
           let getByNid _ [] = Nothing
               getByNid nid ((_,n,nodeboss):xs) = if nid==n then Just nodeboss else getByNid nid xs
           res <- liftIO $ withMVar nodes (\n -> return $ getByNid selfnode n)
           _ <- case res of
               Nothing -> taskError "Can't find self: make sure cfgKnownHosts includes the master"
               Just x -> spawnLocalAnd (masterproc x) (do myself <- getSelfPid
                                                          monitorProcess selfpid myself MaLinkError)
           _ <- spawnDaemonic (prober nodes seennodes masterpid)
           return masterpid
  
  stubborn :: (Monad m) => Int -> m (Maybe a) -> m (Maybe a)
  stubborn 0 a = a
  stubborn n a | n>0
               = do r <- a
                    case r of
                       Just _ -> return r
                       Nothing -> stubborn (n-1) a
  
  -- TODO: setDefaultLocality :: Locality -> TaskM ()
  
  -- | Like 'newPromise', but creates a promise whose
  -- values is already known. In other words, it puts
  -- a given, already-calculated value in a promise.
  -- Conceptually (but not syntactically, due to closures),
  -- you can consider it like this:
  --
  -- > toPromise a = newPromise (return a)
  toPromise :: (Serializable a) => a -> TaskM (Promise a)
  toPromise = toPromiseAt LcDefault
  
  -- | A variant of 'toPromise' that lets the user
  -- express a locality preference, i.e. some information
  -- about which node will become the owner of the
  -- new promise. These preferences will not necessarily
  -- be respected.
  toPromiseAt :: (Serializable a) => Locality -> a -> TaskM (Promise a)
  toPromiseAt locality a = newPromiseAt locality (passthrough__closure a)
  
  -- | Similar to 'toPromiseAt' and 'newPromiseNear'
  toPromiseNear :: (Serializable a,Serializable b) => Promise b -> a -> TaskM (Promise a)
  toPromiseNear (PromiseImmediate _) = toPromise
  -- TODO should I consult tsRedeemerForwarding here?
  toPromiseNear (PromiseBasic prhost _prid) = toPromiseAt (LcByNode [nodeFromPid prhost])
  
  -- | Creates an /immediate promise/, which is to say, a promise
  -- in name only. Unlike a regular promise (created by 'toPromise'), 
  -- this kind of promise contains the value directly. The 
  -- advantage is that promise redemption is very fast, requiring
  -- no network communication. The downside is that it the
  -- underlying data will be copied along with the promise.
  -- Useful only for small data.
  toPromiseImm :: (Serializable a) => a -> TaskM (Promise a)
  toPromiseImm = return . PromiseImmediate
  
  -- | Given a function (expressed here as a closure, see "Remote.Call")
  -- that computes a value, returns a token identifying that value.
  -- This token, a 'Promise' can be moved about even if the
  -- value hasn't been computed yet. The computing function 
  -- will be started somewhere among the nodes visible to the
  -- current master, preferring those nodes that correspond
  -- to the 'defaultLocality'. Afterwards, attempts to
  -- redeem the promise with 'readPromise' will contact the node
  -- where the function is executing.
  newPromise :: (Serializable a) => Closure (TaskM a) -> TaskM (Promise a)
  newPromise = newPromiseAt LcDefault
  
  -- | A variant of 'newPromise' that prefers to start
  -- the computing function on the same node as the caller.
  -- Useful if you plan to use the resulting value
  -- locally.
  newPromiseHere :: (Serializable a) => Closure (TaskM a) -> TaskM (Promise a)
  newPromiseHere clo =
       do mynode <- liftTask $ getSelfNode
          newPromiseAt (LcByNode [mynode]) clo
  
  -- | A variant of 'newPromise' that prefers to start
  -- the computing function on the same node where some
  -- other promise lives. The other promise is not
  -- evaluated.
  newPromiseNear :: (Serializable a, Serializable b) => Promise b -> Closure (TaskM a) -> TaskM (Promise a)
  newPromiseNear (PromiseImmediate _) = newPromise
  newPromiseNear (PromiseBasic prhost _prid) = newPromiseAt (LcByNode [nodeFromPid prhost]) 
  
  -- | A variant of 'newPromise' that prefers to start
  -- the computing functions on some set of nodes that
  -- have a given role (assigned by the cfgRole configuration
  -- option).
  newPromiseAtRole :: (Serializable a) => String -> Closure (TaskM a) -> TaskM (Promise a)
  newPromiseAtRole role clo = newPromiseAt (LcByRole [role]) clo
  
  -- | A variant of 'newPromise' that lets the user
  -- specify a 'Locality'. The other flavors of newPromise,
  -- such as 'newPromiseAtRole', 'newPromiseNear', and
  -- 'newPromiseHere' at just shorthand for a call to this function.
  newPromiseAt :: (Serializable a) => Locality -> Closure (TaskM a) -> TaskM (Promise a)
  newPromiseAt locality clo = 
     let realclo = makePayloadClosure clo
      in case realclo of
            Just plclo -> do master <- getMaster
                             res <- roundtrip master (MmNewPromise plclo locality defaultQueueing)
                             case res of
                                Right (MmNewPromiseResponse pid prid) -> return $ PromiseBasic pid prid
                                Right (MmNewPromiseResponseFail) ->
                                            taskError $ "Spawning of closure "++show clo++" by newPromise failed"
                                Left tms -> taskError $ "Spawning of closure "++show clo++" by newPromise resulted in "++show tms
            Nothing -> taskError $ "The specified closure, "++show clo++", can't produce payloads"
  
  -- | Given a promise, gets the value that is being
  -- calculated. If the calculation has finished,
  -- the owning node will be contacted and the data
  -- moved to the current node. If the calculation
  -- has not finished, this function will block
  -- until it has. If the calculation failed
  -- by throwing an exception (e.g. divide by zero),
  -- then this function will throw an excption as well
  -- (a 'TaskException'). If the node owning the
  -- promise is not accessible, the calculation
  -- will be restarted.
  readPromise :: (Serializable a) => Promise a -> TaskM a
  readPromise (PromiseImmediate a) = return a
  readPromise thepromise@(PromiseBasic prhost prid) = 
         do mp <- lookupCachedPromise prid
            case mp of
              Nothing -> do fprhost <- liftM (maybe prhost id) $ lookupForwardedRedeemer prid
                            res <- roundtrip fprhost (NmRedeem prid)
                            case res of
                               Left e -> do tlogS "TSK" LoInformation $ "Complaining about promise " ++ show prid ++" on " ++show fprhost++" because of "++show e
                                            complain fprhost prid
                               Right NmRedeemResponseUnknown -> 
                                         do tlogS "TSK" LoInformation $ "Complaining about promise " ++ show prid ++" on " ++show fprhost++" because allegedly unknown"
                                            complain fprhost prid
                               Right (NmRedeemResponse thedata) -> 
                                                    do extracted <- extractFromPayload thedata
                                                       promiseinmem <- liftTaskIO $ makePromiseInMemory thedata (Just $ toDyn extracted)
                                                       putPromiseInCache prid promiseinmem
                                                       return extracted
                               Right NmRedeemResponseException ->
                                    taskError "Failed promise redemption" -- don't redeem, this is a terminal failure
              Just mv -> do val <- liftTaskIO $ readMVar mv -- possible long wait here
                            case val of -- TODO this read/write MVars should be combined!
                               PromiseInMemory v _utc thedyn -> 
                                       case thedyn of
                                         Just thedynvalue -> 
                                           case fromDynamic thedynvalue of
                                             Nothing -> do liftTask $ logS "TSK" LoStandard "Insufficiently dynamic promise cache"
                                                           extractFromPayload v
                                             Just realval -> do updated <- liftTaskIO $ makePromiseInMemory v thedyn
                                                                putPromiseInCache prid updated
                                                                return realval
                                         Nothing -> do extracted <- extractFromPayload v
                                                       updated <- liftTaskIO $ makePromiseInMemory v (Just $ toDyn extracted)
                                                       putPromiseInCache prid updated
                                                       return extracted
                               PromiseException _ -> taskError $ "Redemption of promise failed"
                               PromiseOnDisk fp -> do mpd <- liftTask $ undiskify fp mv
                                                      _ <- liftTask $ spawnLocal $ diskify fp mv False
                                                      case mpd of
                                                         Just dat -> extractFromPayload dat
                                                         _ -> taskError "Promise extraction from disk failed"
       where extractFromPayload v = do out <- liftTaskIO $ serialDecode v
                                       case out of
                                         Just r -> return r
                                         Nothing -> taskError "Unexpected payload type"
             complain fprhost prid =
                       do master <- getMaster
                          response <- roundtrip master (MmComplain fprhost prid)
                          case response of 
                            Left a -> taskError $ "Couldn't file complaint with master about " ++ show fprhost ++ " because " ++ show a
                            Right (MmComplainResponse newhost) 
                              | newhost == nullPid -> taskError $ "Couldn't file complaint with master about " ++ show fprhost
                              | otherwise -> do setForwardedRedeemer prid newhost
                                                readPromise thepromise
  
  data TaskState = TaskState
        {
           tsMaster :: ProcessId,
           tsNodeBoss :: Maybe ProcessId,
           tsPromiseCache :: MVar (Map.Map PromiseId (MVar PromiseStorage)),
           tsRedeemerForwarding :: MVar (Map.Map PromiseId ProcessId),
           tsMonitoring :: Map.Map ProcessId ()
        }
  
  data TaskM a = TaskM { runTaskM :: TaskState -> ProcessM (TaskState, a) } deriving (Typeable)
  
  instance Monad TaskM where
     m >>= k = TaskM $ \ts -> do
                  (ts',a) <- runTaskM m ts
                  (ts'',a') <- runTaskM (k a) (ts')
                  return (ts'',a')              
     return x = TaskM $ \ts -> return $ (ts,x)
  
  instance Functor TaskM where
      f `fmap` m =
          TaskM $ \ts ->
          runTaskM m ts >>= \(ts', x) ->
          return (ts', f x)
  
  instance Applicative TaskM where
      mf <*> mx =
          TaskM $ \ts ->
          runTaskM mf ts >>= \(ts', f) ->
          runTaskM mx ts' >>= \(ts'', x) ->
          return (ts'', f x)
      pure = return
  
  lookupForwardedRedeemer :: PromiseId -> TaskM (Maybe ProcessId)
  lookupForwardedRedeemer q = 
     TaskM $ \ts ->
       liftIO $ withMVar (tsRedeemerForwarding ts) $ (\fwd ->
         let lo = Map.lookup q fwd 
          in return (ts,lo))
  
  setForwardedRedeemer :: PromiseId -> ProcessId -> TaskM ()
  setForwardedRedeemer from to =
     TaskM $ \ts -> liftIO $ modifyMVar (tsRedeemerForwarding ts)  (\fwd ->
       let newmap = Map.insert from to fwd 
        in return ( newmap,(ts,())  ) )
  
  lookupCachedPromise :: PromiseId -> TaskM (Maybe (MVar PromiseStorage))
  lookupCachedPromise prid = TaskM $ \ts -> 
                 do mv <- liftIO $ withMVar (tsPromiseCache ts)
                        (\pc -> return $ Map.lookup prid pc)
                    return (ts,mv)
  
  putPromiseInCache :: PromiseId -> PromiseStorage -> TaskM ()
  putPromiseInCache prid ps = TaskM $ \ts ->
          do liftIO $ modifyMVar_ (tsPromiseCache ts)
                  (\pc -> do mv <- newMVar ps
                             return $ Map.insert prid mv pc)
             return (ts,())
  
  getMaster :: TaskM ProcessId
  getMaster = TaskM $ \ts -> return (ts,tsMaster ts)
  
  liftTask :: ProcessM a -> TaskM a
  liftTask a = TaskM $ \ts -> a >>= (\x -> return (ts,x))
  
  liftTaskIO :: IO a -> TaskM a
  liftTaskIO = liftTask . liftIO
  
  -- | A Task-monadic version of 'Remote.Process.say'.
  -- Puts text messages in the log.
  tsay :: String -> TaskM ()
  tsay a = liftTask $ say a
  
  -- | Writes various kinds of messages to the
  -- "Remote.Process" log.
  tlogS :: LogSphere -> LogLevel -> String -> TaskM ()
  tlogS a b c = liftTask $ logS a b c
  
  ----------------------------------------------
  -- * MapReduce
  ----------------------------------------------
  
  -- | A data structure that stores the important
  -- user-provided functions that are the namesakes
  -- of the MapReduce algorithm.
  -- The number of mapper processes can be controlled
  -- by the user by controlling the length of the string
  -- returned by mtChunkify. The number of reducer
  -- promises is controlled by the number of values
  -- values returned by shuffler.
  -- The user must provide their own mapper and reducer.
  -- For many cases, the default chunkifier ('chunkify')
  -- and shuffler ('shuffle') are adequate.
  data MapReduce rawinput input middle1 middle2 result 
      = MapReduce
        {
          mtMapper :: input -> Closure (TaskM [middle1]),
          mtReducer :: middle2 -> Closure (TaskM result),
          mtChunkify :: rawinput -> [input],
          mtShuffle :: [middle1] -> [middle2]
        }
  
  -- | A convenient way to provide the 'mtShuffle' function
  -- as part of 'mapReduce'. 
  shuffle :: Ord a => [(a,b)] -> [(a,[b])]
  shuffle q = 
      let semi = groupBy (\(a,_) (b,_) -> a==b) (sortBy (\(a,_) (b,_) -> compare a b) q)
       in map (\x -> (fst $ head x,map snd x)) semi 
  
  -- | A convenient way to provide the 'mtChunkify' function
  -- as part of 'mapReduce'. 
  chunkify :: Int -> [a] -> [[a]] 
  chunkify numChunks l 
    | numChunks <= 0 = taskError "Can't chunkify into less than one chunk"
    | otherwise = splitSize (ceiling ((fromIntegral (length l) / fromIntegral numChunks) :: Double)) l
     where
        splitSize _ [] = []
        splitSize i v = let (first,second) = splitAt i v 
                         in first : splitSize i second
  
  -- | The MapReduce algorithm, implemented in a very
  -- simple form on top of the Task layer. Its
  -- use depends on four user-determined data types:
  --
  -- * input -- The data type provided as the input to the algorithm as a whole and given to the mapper.
  --
  -- * middle1 -- The output of the mapper. This may include some /key/ which is used by the shuffler to allocate data to reducers.
  -- If you use the default shuffler, 'shuffle', this type must have the form @Ord a => (a,b)@.
  --
  -- * middle2 -- The output of the shuffler. The default shuffler emits a type in the form @Ord => (a,[b])@. Each middle2 output 
  -- by shuffler is given to a separate reducer.
  --
  -- * result -- The output of the reducer, upon being given a bunch of middles.
  mapReduce :: (Serializable i,Serializable k,Serializable m,Serializable r) =>
               MapReduce ri i k m r -> ri -> TaskM [r]
  mapReduce mr inputs =
      let chunks = (mtChunkify mr) inputs 
       in do 
             pmapResult <- mapM (\chunk -> 
                   newPromise ((mtMapper mr) chunk) ) chunks
             mapResult <- mapM readPromise pmapResult
             let shuffled = (mtShuffle mr) (concat mapResult)
             pres <- mapM (\mid2 -> 
                    newPromise ((mtReducer mr) mid2)) shuffled
             mapM readPromise pres
  
  