<commit_before>
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE NoMonomorphismRestriction #-}
{-# LANGUAGE DeriveFunctor, DeriveGeneric, DeriveDataTypeable #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE RankNTypes #-}
-- | Controller processes for node and actor
module DNA.Controller (
      -- * Interaction with remote ACP
      ACP(..)
    , terminateACP
      -- * Starting ACP
    , startAcpLoop
      -- * Control messages for ACP
    , Res(..)
    , ReqSpawnShell(..)
    , ReqSpawnGroup(..)
    , ReqConnect(..)
    , ReqResources(..)
    , ReqResourcesGrp(..)
      -- * Node controller
    , nodeController
    , spawnHierachically
    , __remoteTable
    ) where

import Control.Applicative
import Control.Monad
import Control.Monad.State
import Control.Monad.Except
import Control.Distributed.Process
import Control.Distributed.Process.Closure
import Control.Distributed.Process.Serializable (Serializable)
import Data.Binary   (Binary)
import Data.Typeable (Typeable)
import qualified Data.Foldable   as T
import qualified Data.Set        as Set
import           Data.Set          (Set)
import qualified Data.Map.Strict as Map
import           Data.Map.Strict   (Map)
import Text.Printf
import GHC.Generics (Generic)

import DNA.Lens
import DNA.Types
import DNA.Logging



----------------------------------------------------------------
-- Interaction with 
----------------------------------------------------------------

-- | Send command to remote ACP to terminate abnormally
terminateACP :: MonadProcess m => ACP -> m ()
terminateACP (ACP pid) = liftP $ do
  send pid Terminate


-- | Command for ACP to terminate
data Terminate = Terminate
                deriving (Show,Eq,Ord,Typeable,Generic)
instance Binary Terminate


----------------------------------------------------------------
-- ACP implementation
----------------------------------------------------------------

-- | Start execution of actor controller process
startAcpLoop :: Closure (Process ()) -- ^ Closure to self
             -> ProcessId            -- ^ PID of process being monitored
             -> VirtualCAD           -- ^ Allocated nodes
             -> Process ()
startAcpLoop self pid (VirtualCAD _ n nodes) = do
    let loop s = do
            ms <- acpStep s
            case ms of
              Right s' -> loop s'
              -- FIXME: check status of child processes
              Left  Done     -> logMessage "ACP" "Done"
              Left (Fatal m) -> do<commit_msg>Raise fatal error when actor completes execution with running children<commit_after>
{-# LANGUAGE FlexibleContexts #-}
{-# LANGUAGE NoMonomorphismRestriction #-}
{-# LANGUAGE DeriveFunctor, DeriveGeneric, DeriveDataTypeable #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE GADTs #-}
{-# LANGUAGE TemplateHaskell #-}
{-# LANGUAGE RankNTypes #-}
-- | Controller processes for node and actor
module DNA.Controller (
      -- * Interaction with remote ACP
      ACP(..)
    , terminateACP
      -- * Starting ACP
    , startAcpLoop
      -- * Control messages for ACP
    , Res(..)
    , ReqSpawnShell(..)
    , ReqSpawnGroup(..)
    , ReqConnect(..)
    , ReqResources(..)
    , ReqResourcesGrp(..)
      -- * Node controller
    , nodeController
    , spawnHierachically
    , __remoteTable
    ) where

import Control.Applicative
import Control.Monad
import Control.Monad.State
import Control.Monad.Except
import Control.Distributed.Process
import Control.Distributed.Process.Closure
import Control.Distributed.Process.Serializable (Serializable)
import Data.Binary   (Binary)
import Data.Typeable (Typeable)
import qualified Data.Foldable   as T
import qualified Data.Set        as Set
import           Data.Set          (Set)
import qualified Data.Map.Strict as Map
import           Data.Map.Strict   (Map)
import Text.Printf
import GHC.Generics (Generic)

import DNA.Lens
import DNA.Types
import DNA.Logging



----------------------------------------------------------------
-- Interaction with 
----------------------------------------------------------------

-- | Send command to remote ACP to terminate abnormally
terminateACP :: MonadProcess m => ACP -> m ()
terminateACP (ACP pid) = liftP $ do
  send pid Terminate


-- | Command for ACP to terminate
data Terminate = Terminate
                deriving (Show,Eq,Ord,Typeable,Generic)
instance Binary Terminate


----------------------------------------------------------------
-- ACP implementation
----------------------------------------------------------------

-- | Start execution of actor controller process
startAcpLoop :: Closure (Process ()) -- ^ Closure to self
             -> ProcessId            -- ^ PID of process being monitored
             -> VirtualCAD           -- ^ Allocated nodes
             -> Process ()
startAcpLoop self pid (VirtualCAD _ n nodes) = do
    let loop s = do
            ms <- acpStep s
            case ms of
              Right s' -> loop s'
              Left  Done
                  | Map.null (s^.stChildren) -> logMessage "ACP" "Done"
                  | otherwise -> do logMessage "ACP" "Unterminated children"
                                    error "Unterminated children"
              Left (Fatal m) -> do
{'fraction_non_alphanumeric': 0.18350710900473935, 'fraction_numerical': 0.0, 'mean_word_length': 1.7536534446764092, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3362397', 'n_tokens_mistral': 1463, 'n_tokens_neox': 1381, 'n_words': 411}