<commit_before>{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable #-}

-----------------------------------------------------------------------------
-- |
-- Module      :  Control.Distributed.Process.Platform.Call
-- Copyright   :  (c) Parallel Scientific (<NAME>) 2012
-- License     :  BSD3 (see the file LICENSE)
--
-- Maintainers :  <NAME>, <NAME>
-- Stability   :  experimental
-- Portability :  non-portable (requires concurrency)
--
-- This module provides a facility for Remote Procedure Call (rpc) style
-- interactions with Cloud Haskell processes.
--
-- Clients make synchronous calls to a running process (i.e., server) using the
-- 'callAt', 'callTimeout' and 'multicall' functions. Processes acting as the
-- server are constructed using Cloud Haskell's 'receive' family of primitives
-- and the 'callResponse' family of functions in this module.   
-----------------------------------------------------------------------------

module Control.Distributed.Process.Platform.Call 
  ( -- client API
    callAt
  , callTimeout
  , multicall
    -- server API
  , callResponse
  , callResponseIf
  , callResponseDefer
  , callResponseDeferIf
  , callForward
  , callResponseAsync
  ) where

import Control.Distributed.Process
import Control.Distributed.Process.Serializable (Serializable)
import Control.Monad (forM, forM_, join)
import Data.List (delete)
import qualified Data.Map as M
import Data.Maybe (listToMaybe)
import Data.Binary (Binary,get,put)
import Data.Typeable (Typeable)

import Control.Distributed.Process.Platform
import Control.Distributed.Process.Platform.Time

----------------------------------------------
-- * Multicall
----------------------------------------------

-- | Sends a message of type a to the given process, to be handled by a corresponding
-- callResponse... function, which will send back a message of type b.
-- The tag is per-process unique identifier of the transaction. If the timeout expires
-- or the target process dies, Nothing will be returned.
callTimeout :: (Serializable a, Serializable b) => ProcessId -> a -> Tag -> Timeout -> Process (Maybe b)
callTimeout pid msg tag time = 
  do res <- multicall [pid] msg tag time 
     return $ join (listToMaybe res)

-- | Like 'callTimeout', but with no timeout. Returns Nothing if the target process dies.
callAt :: (Serializable a, Serializable b) => ProcessId -> a -> Tag -> Process (Maybe b)
callAt pid msg tag = callTimeout pid msg tag infiniteWait

-- | Like 'callTimeout', but sends the message to multiple recipients and collects the results.
multicall :: forall a b.(Serializable a, Serializable b) => [ProcessId] -> a -> Tag -> Timeout -> Process [Maybe b]
multicall nodes msg tag time =
  do caller <- getSelfPid
     reciever <- spawnLocal $
         do reciever_pid <- getSelfPid
            mon_caller <- monitor caller
            () <- expect
            monitortags <- forM nodes monitor
            forM_ nodes $ \node -> send node (Multicall, node, reciever_pid, tag, msg)
            maybeTimeout time tag reciever_pid
            results <- recv nodes monitortags mon_caller
            send caller (MulticallResponse,tag,results)
     mon_reciever <- monitor reciever
     send reciever ()
     receiveWait
       [
         matchIf (\(MulticallResponse,mtag,_) -> mtag == tag)
                 (\(MulticallResponse,_,val) -> return val),
         matchIf (\(ProcessMonitorNotification ref _pid reason) -> ref == mon_reciever && reason /= DiedNormal)
                 (\_ -> error "multicall: unexpected termination of worker process")
       ]
   where
         recv nodes monitortags mon_caller = 
           do 
              let
                  ordered [] _ = []
                  ordered (x:xs) m = 
                      M.lookup x m : ordered xs m
                  recv1 ([],_,results) = return results
                  recv1 (_,[],results) = return results
                  recv1 (nodesleft,monitortagsleft,results) =
                     receiveWait
                         [
                            matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mon_caller)
                                    (\_ -> return Nothing),
                            matchIf (\(ProcessMonitorNotification ref pid reason) -> 
                                        ref `elem` monitortagsleft && pid `elem` nodesleft && reason /= DiedNormal)
                                    (\(ProcessMonitorNotification ref pid _reason) -> 
                                        return $ Just (delete pid nodesleft, delete ref monitortagsleft, results)),
                            matchIf (\(MulticallResponse,mtag,_,_) -> mtag == tag)
                                    (\(MulticallResponse,_,responder,msgx) -> 
                                        return $ Just (delete responder nodesleft, monitortagsleft, M.insert responder (msgx::b) results)),
                            matchIf (\(TimeoutNotification mtag) -> mtag == tag )
                                    (\_ -> return Nothing)
                         ]
                            >>= maybe (return results) recv1
              resultmap <- recv1 (nodes, monitortags, M.empty) :: Process (M.Map ProcessId b)
              return $ ordered nodes resultmap

data MulticallResponseType a =
         MulticallAccept
       | MulticallForward ProcessId a
       | MulticallReject deriving Eq

callResponseImpl :: (Serializable a,Serializable b) => (a -> MulticallResponseType c) -> 
                         (a -> (b -> Process())-> Process c) -> Match c
callResponseImpl cond proc = 
    matchIf (\(Multicall,_responder,_,_,msg) -> 
                 case cond msg of
                    MulticallReject -> False
                    _ -> True) 
            (\wholemsg@(Multicall,responder,sender,tag,msg) -> 
                 case cond msg of
                   MulticallForward target ret -> -- TODO sender should get a ProcessMonitorNotification if target dies, or we should link target
                     do send target wholemsg
                        return ret
                   MulticallReject -> error "multicallResponseImpl: Indecisive condition"
                   MulticallAccept -><commit_msg>kill some more compiler warnings<commit_after>{-# LANGUAGE ScopedTypeVariables, DeriveDataTypeable #-}

-----------------------------------------------------------------------------
-- |
-- Module      :  Control.Distributed.Process.Platform.Call
-- Copyright   :  (c) Parallel Scientific (<NAME>) 2012
-- License     :  BSD3 (see the file LICENSE)
--
-- Maintainers :  <NAME>, <NAME>
-- Stability   :  experimental
-- Portability :  non-portable (requires concurrency)
--
-- This module provides a facility for Remote Procedure Call (rpc) style
-- interactions with Cloud Haskell processes.
--
-- Clients make synchronous calls to a running process (i.e., server) using the
-- 'callAt', 'callTimeout' and 'multicall' functions. Processes acting as the
-- server are constructed using Cloud Haskell's 'receive' family of primitives
-- and the 'callResponse' family of functions in this module.   
-----------------------------------------------------------------------------

module Control.Distributed.Process.Platform.Call 
  ( -- client API
    callAt
  , callTimeout
  , multicall
    -- server API
  , callResponse
  , callResponseIf
  , callResponseDefer
  , callResponseDeferIf
  , callForward
  , callResponseAsync
  ) where

import Control.Distributed.Process
import Control.Distributed.Process.Serializable (Serializable)
import Control.Monad (forM, forM_, join)
import Data.List (delete)
import qualified Data.Map as M
import Data.Maybe (listToMaybe)
import Data.Binary (Binary,get,put)
import Data.Typeable (Typeable)

import Control.Distributed.Process.Platform
import Control.Distributed.Process.Platform.Time

----------------------------------------------
-- * Multicall
----------------------------------------------

-- | Sends a message of type a to the given process, to be handled by a corresponding
-- callResponse... function, which will send back a message of type b.
-- The tag is per-process unique identifier of the transaction. If the timeout expires
-- or the target process dies, Nothing will be returned.
callTimeout :: (Serializable a, Serializable b) => ProcessId -> a -> Tag -> Timeout -> Process (Maybe b)
callTimeout pid msg tag time = 
  do res <- multicall [pid] msg tag time 
     return $ join (listToMaybe res)

-- | Like 'callTimeout', but with no timeout. Returns Nothing if the target process dies.
callAt :: (Serializable a, Serializable b) => ProcessId -> a -> Tag -> Process (Maybe b)
callAt pid msg tag = callTimeout pid msg tag infiniteWait

-- | Like 'callTimeout', but sends the message to multiple recipients and collects the results.
multicall :: forall a b.(Serializable a, Serializable b) => [ProcessId] -> a -> Tag -> Timeout -> Process [Maybe b]
multicall nodes msg tag time =
  do caller <- getSelfPid
     reciever <- spawnLocal $
         do reciever_pid <- getSelfPid
            mon_caller <- monitor caller
            () <- expect
            monitortags <- forM nodes monitor
            forM_ nodes $ \node -> send node (Multicall, node, reciever_pid, tag, msg)
            maybeTimeout time tag reciever_pid
            results <- recv nodes monitortags mon_caller
            send caller (MulticallResponse,tag,results)
     mon_reciever <- monitor reciever
     send reciever ()
     receiveWait
       [
         matchIf (\(MulticallResponse,mtag,_) -> mtag == tag)
                 (\(MulticallResponse,_,val) -> return val),
         matchIf (\(ProcessMonitorNotification ref _pid reason) -> ref == mon_reciever && reason /= DiedNormal)
                 (\_ -> error "multicall: unexpected termination of worker process")
       ]
   where
         recv nodes' monitortags mon_caller = 
           do 
              let
                  ordered [] _ = []
                  ordered (x:xs) m = 
                      M.lookup x m : ordered xs m
                  recv1 ([],_,results) = return results
                  recv1 (_,[],results) = return results
                  recv1 (nodesleft,monitortagsleft,results) =
                     receiveWait
                         [
                            matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mon_caller)
                                    (\_ -> return Nothing),
                            matchIf (\(ProcessMonitorNotification ref pid reason) -> 
                                        ref `elem` monitortagsleft && pid `elem` nodesleft && reason /= DiedNormal)
                                    (\(ProcessMonitorNotification ref pid _reason) -> 
                                        return $ Just (delete pid nodesleft, delete ref monitortagsleft, results)),
                            matchIf (\(MulticallResponse,mtag,_,_) -> mtag == tag)
                                    (\(MulticallResponse,_,responder,msgx) -> 
                                        return $ Just (delete responder nodesleft, monitortagsleft, M.insert responder (msgx::b) results)),
                            matchIf (\(TimeoutNotification mtag) -> mtag == tag )
                                    (\_ -> return Nothing)
                         ]
                            >>= maybe (return results) recv1
              resultmap <- recv1 (nodes', monitortags, M.empty) :: Process (M.Map ProcessId b)
              return $ ordered nodes' resultmap

data MulticallResponseType a =
         MulticallAccept
       | MulticallForward ProcessId a
       | MulticallReject deriving Eq

callResponseImpl :: (Serializable a,Serializable b) => (a -> MulticallResponseType c) -> 
                         (a -> (b -> Process())-> Process c) -> Match c
callResponseImpl cond proc = 
    matchIf (\(Multicall,_responder,_,_,msg) -> 
                 case cond msg of
                    MulticallReject -> False
                    _ -> True) 
            (\wholemsg@(Multicall,responder,sender,tag,msg) -> 
                 case cond msg of
                   MulticallForward target ret -> -- TODO sender should get a ProcessMonitorNotification if target dies, or we should link target
                     do send target wholemsg
                        return ret
                   MulticallReject -> error "multicallResponseImpl: Indecisive condition"
                   MulticallAccept ->
{'fraction_non_alphanumeric': 0.12566414426018355, 'fraction_numerical': 0.0016100466913540493, 'mean_word_length': 1.784801613987895, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 95, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '818777', 'n_tokens_mistral': 3139, 'n_tokens_neox': 2994, 'n_words': 1024}