<commit_before>module ClusterComputing.TaskDistribution (
  startWorkerNode,
  executeDistributed,
  showWorkerNodes,
  showWorkerNodesWithData,
  shutdownWorkerNodes) where

import Control.Distributed.Process (Process, ProcessId, NodeId,
                                    say, getSelfPid, spawn, send, expect, catch,
                                    RemoteTable)
import Control.Distributed.Process.Backend.SimpleLocalnet (initializeBackend, startMaster, startSlave, terminateSlave)
import qualified Control.Distributed.Process.Serializable as PS
import qualified Control.Distributed.Static as S
import Control.Distributed.Process.Closure (staticDecode)
import Control.Distributed.Process.Node (initRemoteTable)
import Control.Distributed.Process.Serializable (Serializable)
import Control.Exception.Base (SomeException)
import Control.Monad (forM_)
import Control.Monad.IO.Class
import qualified Data.Binary as B (encode)
import Data.List (delete)
import Data.List.Split (splitOn)
import qualified Data.Rank1Dynamic as R1 (toDynamic)
import Data.Time.Clock (UTCTime, diffUTCTime, NominalDiffTime, getCurrentTime)

import ClusterComputing.DataLocality (findNodesWithData)
import ClusterComputing.HdfsWriter (writeEntriesToFile)
import ClusterComputing.LogConfiguration
import ClusterComputing.TaskTransport
import TaskSpawning.TaskSpawning (processTask, RunStat)
import TaskSpawning.TaskTypes

{-
 The bits in this file are arranged so that the less verbose Template Haskell version would work. That version is not used due to incompability
 issues between packman and Template Haskell though, but is retained in the commens for documentational purposes.
-}

-- BEGIN bindings for node communication
{-
 This is the final building block of the worker task execution, calling TaskSpawning.processTask.
-}
workerTask :: TaskTransport -> Process () -- TODO: have a node local config?
workerTask (TaskTransport masterProcess taskMetaData taskDef dataDef resultDef) = do
  handledResult <- (
    do
      acceptTime <- liftIO getCurrentTime
      say $ "processing: " ++ taskName
      result <- liftIO (processTask taskDef dataDef >>= return)
      say $ "processing done for: " ++ taskName
      processingDoneTime <- liftIO getCurrentTime
      liftIO (handleWorkerResult dataDef resultDef result acceptTime processingDoneTime) >>= return . Right
    ) `catch` buildError
  say $ "replying"
  send masterProcess ((taskMetaData, handledResult) :: TransportedResult)
  where
    taskName = _taskName taskMetaData
    buildError :: SomeException -> Process (Either String a)
    buildError e = return $ Left $ "Task execution (for: "++taskName++") failed: " ++ (format $ show e)
      where
        format [] = []
        format ('\\':'n':'\\':'t':rest) = "\n\t" ++ (format rest)
        format (x:rest) = x:[] ++ (format rest)

type TransportedResult = (TaskMetaData, Either String (TaskResult, RemoteRunStat)) -- signature here defines transported type, handle with care

handleWorkerResult :: DataDef -> ResultDef -> (TaskResult, RunStat) -> UTCTime -> UTCTime -> IO (TaskResult, RemoteRunStat)
handleWorkerResult dataDef resultDef (taskResult, runStat) acceptTime processingDoneTime = do
  res <- handleResult dataDef resultDef
  return (res, serializedRunStat runStat)
  where
    handleResult :: DataDef -> ResultDef -> IO TaskResult
    handleResult _ ReturnAsMessage = return taskResult
    handleResult (HdfsData (config, path)) (HdfsResult outputPrefix) = do
      writeEntriesToFile config (outputPrefix ++ "/" ++ fileNamePart) taskResult >> return []
      where fileNamePart = let parts = splitOn "/" path in if null parts then "" else parts !! (length parts -1)
    handleResult _ (HdfsResult _) = error "storage to hdfs with other data source than hdfs currently not supported"
    handleResult _ ReturnOnlyNumResults = return (taskResult >>= \rs -> [show $ length rs])
    serializedRunStat (d, t, e) =
      RemoteRunStat (serializeTimeDiff $ diffUTCTime processingDoneTime acceptTime) (serializeTimeDiff d) (serializeTimeDiff t) (serializeTimeDiff e)

-- template haskell vs. its result<commit_msg>added measuring hdfs result storage<commit_after>module ClusterComputing.TaskDistribution (
  startWorkerNode,
  executeDistributed,
  showWorkerNodes,
  showWorkerNodesWithData,
  shutdownWorkerNodes) where

import Control.Distributed.Process (Process, ProcessId, NodeId,
                                    say, getSelfPid, spawn, send, expect, catch,
                                    RemoteTable)
import Control.Distributed.Process.Backend.SimpleLocalnet (initializeBackend, startMaster, startSlave, terminateSlave)
import qualified Control.Distributed.Process.Serializable as PS
import qualified Control.Distributed.Static as S
import Control.Distributed.Process.Closure (staticDecode)
import Control.Distributed.Process.Node (initRemoteTable)
import Control.Distributed.Process.Serializable (Serializable)
import Control.Exception.Base (SomeException)
import Control.Monad (forM_)
import Control.Monad.IO.Class
import qualified Data.Binary as B (encode)
import Data.List (delete)
import Data.List.Split (splitOn)
import qualified Data.Rank1Dynamic as R1 (toDynamic)
import Data.Time.Clock (UTCTime, diffUTCTime, NominalDiffTime, getCurrentTime)

import ClusterComputing.DataLocality (findNodesWithData)
import ClusterComputing.HdfsWriter (writeEntriesToFile)
import ClusterComputing.LogConfiguration
import ClusterComputing.TaskTransport
import TaskSpawning.ExecutionUtil (measureDuration)
import TaskSpawning.TaskSpawning (processTask, RunStat)
import TaskSpawning.TaskTypes

{-
 The bits in this file are arranged so that the less verbose Template Haskell version would work. That version is not used due to incompability
 issues between packman and Template Haskell though, but is retained in the commens for documentational purposes.
-}

-- BEGIN bindings for node communication
{-
 This is the final building block of the worker task execution, calling TaskSpawning.processTask.
-}
workerTask :: TaskTransport -> Process () -- TODO: have a node local config?
workerTask (TaskTransport masterProcess taskMetaData taskDef dataDef resultDef) = do
  handledResult <- (
    do
      acceptTime <- liftIO getCurrentTime
      say $ "processing: " ++ taskName
      result <- liftIO (processTask taskDef dataDef >>= return)
      say $ "processing done for: " ++ taskName
      processingDoneTime <- liftIO getCurrentTime
      liftIO (handleWorkerResult dataDef resultDef result acceptTime processingDoneTime) >>= return . Right
    ) `catch` buildError
  say $ "replying"
  send masterProcess ((taskMetaData, handledResult) :: TransportedResult)
  where
    taskName = _taskName taskMetaData
    buildError :: SomeException -> Process (Either String a)
    buildError e = return $ Left $ "Task execution (for: "++taskName++") failed: " ++ (format $ show e)
      where
        format [] = []
        format ('\\':'n':'\\':'t':rest) = "\n\t" ++ (format rest)
        format (x:rest) = x:[] ++ (format rest)

type TransportedResult = (TaskMetaData, Either String (TaskResult, RemoteRunStat)) -- signature here defines transported type, handle with care

handleWorkerResult :: DataDef -> ResultDef -> (TaskResult, RunStat) -> UTCTime -> UTCTime -> IO (TaskResult, RemoteRunStat)
handleWorkerResult dataDef resultDef (taskResult, runStat) acceptTime processingDoneTime = do
  res <- handleResult dataDef resultDef
  return (res, serializedRunStat runStat)
  where
    handleResult :: DataDef -> ResultDef -> IO TaskResult
    handleResult _ ReturnAsMessage = return taskResult
    handleResult (HdfsData (config, path)) (HdfsResult outputPrefix) = do
      storeDur <- measureDuration $ writeEntriesToFile config (outputPrefix ++ "/" ++ fileNamePart) taskResult
      putStrLn $ "stored result data in: " ++ (show storeDur)
      return []
      where fileNamePart = let parts = splitOn "/" path in if null parts then "" else parts !! (length parts -1)
    handleResult _ (HdfsResult _) = error "storage to hdfs with other data source than hdfs currently not supported"
    handleResult _ ReturnOnlyNumResults = return (taskResult >>= \rs -> [show $ length rs])
    serializedRunStat (d, t, e) =
      RemoteRunStat (serializeTimeDiff $ diffUTCTime processingDoneTime acceptTime) (serializeTimeDiff d) (serializeTimeDiff t) (serializeTimeDiff e)

-- template haskell vs. its result
{'fraction_non_alphanumeric': 0.08475181526008808, 'fraction_numerical': 0.0007142006903940007, 'mean_word_length': 4.382447149263292, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4139722', 'n_tokens_mistral': 2249, 'n_tokens_neox': 2196, 'n_words': 747}