diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,22 @@ +Copyright (c) 2013, Gregory Crosswhite +All rights reserved. + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright notice, + this list of conditions and the following disclaimer. + * Redistributions in binary form must reproduce the above copyright notice, + this list of conditions and the following disclaimer in the documentation + and/or other materials provided with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR +ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES +(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; +LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON +ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/LogicGrowsOnTrees-network.cabal b/LogicGrowsOnTrees-network.cabal new file mode 100644 --- /dev/null +++ b/LogicGrowsOnTrees-network.cabal @@ -0,0 +1,88 @@ +Name: LogicGrowsOnTrees-network +Version: 1.0.0 +License: BSD3 +License-file: LICENSE +Author: Gregory Crosswhite +Maintainer: Gregory Crosswhite +Synopsis: a adapter for the LogicGrowsOnTrees package that uses multiple processes running in a network +Cabal-version: >=1.8 +Build-type: Simple +Category: Control, Distributed Computing, Logic, Parallelism +Description: + This package provides a adapter for the LogicGrowsOnTrees package that uses + multiple processes running in a network for parallelism; see the module + documentation for more details. + +Source-Repository head + Type: git + Location: git://github.com/gcross/LogicGrowsOnTrees-network.git + +Source-Repository this + Type: git + Location: git://github.com/gcross/LogicGrowsOnTrees-network.git + Tag: 1.0.0 + +Library + Build-depends: + LogicGrowsOnTrees == 1.0.*, + base > 4 && < 5, + cereal >= 0.3.5 && < 0.3.6, + cmdtheline >= 0.2.2 && < 0.3, + composition >= 1.0.1 && < 1.1, + containers >= 0.4 && < 0.6, + hslogger == 1.2.*, + hslogger-template == 2.0.*, + lens >= 3.8.5 && < 3.10, + MonadCatchIO-transformers == 0.3.*, + mtl >= 2.1.2 && < 2.2, + network >= 2.3 && < 2.5, + pretty >= 1.1.1 && < 1.2, + transformers == 0.3.* + Hs-source-dirs: sources + Exposed-modules: + LogicGrowsOnTrees.Parallel.Adapter.Network + if flag(warnings) + GHC-Options: -Wall -fno-warn-name-shadowing + +Flag warnings + Description: Enables most warnings. + Default: False + +Flag examples + Description: Enable building the example executables. + Default: False + +Executable count-all-nqueens-solutions + Main-is: count-all-nqueens-solutions.hs + Hs-source-dirs: examples + Build-depends: + LogicGrowsOnTrees-network, + LogicGrowsOnTrees == 1.0.*, + base > 4 && < 5, + cmdtheline == 0.2.* + if flag(examples) + Buildable: True + else + Buildable: False + if flag(warnings) + GHC-Options: -Wall -fno-warn-name-shadowing + +Test-Suite tests + Type: exitcode-stdio-1.0 + Main-is: tests.hs + Hs-source-dirs: tests + Build-depends: + LogicGrowsOnTrees-network, + LogicGrowsOnTrees == 1.0.*, + base > 4 && < 5, + hslogger == 1.2.*, + hslogger-template == 2.0.*, + HUnit >= 1.2 && < 1.3, + network >= 2.3 && < 2.5, + random == 1.0.*, + stm >= 2.4 && < 2.5, + test-framework== 0.8.*, + test-framework-hunit == 0.3.*, + transformers == 0.3.* + if flag(warnings) + GHC-Options: -Wall -fno-warn-name-shadowing diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/examples/count-all-nqueens-solutions.hs b/examples/count-all-nqueens-solutions.hs new file mode 100644 --- /dev/null +++ b/examples/count-all-nqueens-solutions.hs @@ -0,0 +1,29 @@ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE UnicodeSyntax #-} + +import Data.Functor + +import System.Console.CmdTheLine + +import LogicGrowsOnTrees.Parallel.Adapter.Network +import LogicGrowsOnTrees.Parallel.Main +import LogicGrowsOnTrees.Utils.WordSum + +import LogicGrowsOnTrees.Examples.Queens + +main = + mainForExploreTree + driver + (getBoardSize <$> required (flip (pos 0) (posInfo + { posName = "BOARD_SIZE" + , posDoc = "board size" + } + ) Nothing)) + (defTI { termDoc = "count the number of n-queens solutions for a given board size" }) + (\_ (RunOutcome _ termination_reason) → do + case termination_reason of + Aborted _ → error "search aborted" + Completed (WordSum count) → print count + Failure _ message → error $ "error: " ++ message + ) + nqueensCount diff --git a/sources/LogicGrowsOnTrees/Parallel/Adapter/Network.hs b/sources/LogicGrowsOnTrees/Parallel/Adapter/Network.hs new file mode 100644 --- /dev/null +++ b/sources/LogicGrowsOnTrees/Parallel/Adapter/Network.hs @@ -0,0 +1,624 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE FlexibleContexts #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE NamedFieldPuns #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UnicodeSyntax #-} + +{-| This adapter implements parallelism by allowing multiple workers to connect + to a supervisor over the network. For this adapter, workers are started + separately from the supervisor, so the number of workers is not set by the + controller but by the number of workers that connect to supervisor. + -} +module LogicGrowsOnTrees.Parallel.Adapter.Network + ( + -- * Driver + driver + , driverNetwork + -- * Network + , Network(..) + , runNetwork + -- * Controller + , NetworkRequestQueueMonad(..) + , NetworkControllerMonad + , abort + , fork + , getCurrentProgressAsync + , getCurrentProgress + , getNumberOfWorkersAsync + , getNumberOfWorkers + , requestProgressUpdateAsync + , requestProgressUpdate + , setWorkloadBufferSize + -- * Outcome types + , RunOutcome(..) + , RunStatistics(..) + , TerminationReason(..) + -- * Miscellaneous types + , NetworkCallbacks(..) + , default_network_callbacks + , NetworkConfiguration(..) + , WorkerId(..) + , WrappedPortID(..) + -- * Generic runner functions + -- $runners + , runSupervisor + , runWorker + , runExplorer + -- * Utility functions + , showPortID + , getConfiguration + ) where + +import Prelude hiding (catch) + +import Control.Applicative (Applicative(..)) +import Control.Concurrent (ThreadId,forkIO,killThread,myThreadId,throwTo) +import Control.Exception (AsyncException(..),SomeException,bracket,catch,fromException) +import Control.Lens (use) +import Control.Lens.Operators ((%=)) +import Control.Lens.TH (makeLenses) +import Control.Monad (forever,when) +import Control.Monad.CatchIO (MonadCatchIO) +import Control.Monad.IO.Class (MonadIO(..)) +import Control.Monad.Trans.Reader (ask) +import Control.Monad.Trans.State.Strict (StateT,evalStateT) + +import Data.Composition ((.*)) +import Data.Functor ((<$>)) +import qualified Data.Map as Map +import Data.Map (Map) +import Data.Maybe (fromMaybe) +import Data.Monoid (Monoid(..)) +import Data.Serialize (Serialize) +import qualified Data.Set as Set +import Data.Set (Set) +import Data.Typeable (Typeable) + +import Network (HostName,PortID(..),PortNumber,accept,connectTo,listenOn,sClose,withSocketsDo) + +import System.Console.CmdTheLine +import System.IO (Handle) +import qualified System.Log.Logger as Logger +import System.Log.Logger (Priority(DEBUG,INFO)) +import System.Log.Logger.TH + +import Text.PrettyPrint (text) + +import LogicGrowsOnTrees +import LogicGrowsOnTrees.Parallel.Common.Message +import qualified LogicGrowsOnTrees.Parallel.Common.Process as Process +import LogicGrowsOnTrees.Parallel.Common.RequestQueue +import LogicGrowsOnTrees.Parallel.Common.Supervisor hiding (runSupervisor,getCurrentProgress,getNumberOfWorkers,setWorkloadBufferSize) +import LogicGrowsOnTrees.Parallel.ExplorationMode +import LogicGrowsOnTrees.Parallel.Main +import LogicGrowsOnTrees.Parallel.Purity +import LogicGrowsOnTrees.Utils.Handle + +-------------------------------------------------------------------------------- +----------------------------------- Loggers ------------------------------------ +-------------------------------------------------------------------------------- + +deriveLoggers "Logger" [DEBUG,INFO] + +-------------------------------------------------------------------------------- +----------------------------------- Network ------------------------------------ +-------------------------------------------------------------------------------- + +{-| This monad exists due to the quirk that on Windows one needs to initialize + the network system before using it via. 'withSocketsDo'; to ensure that this + happens, all computations that use the network are run in the 'Network' + monad which itself is then run using the 'runNetwork' function that is + equivalent to calling 'withSocketsDo'. + -} +newtype Network α = Network { unsafeRunNetwork :: IO α } + deriving (Applicative,Functor,Monad,MonadIO) + +{-| Initializes the network subsystem where required (e.g., on Windows). -} +runNetwork :: Network α → IO α +runNetwork = withSocketsDo . unsafeRunNetwork + +-------------------------------------------------------------------------------- +---------------------------- Mostly internal types ----------------------------- +-------------------------------------------------------------------------------- + +{- NOTE: These types have been placed here so that they can be seen by the + following declaration, as use of Template Haskell means that + declaration order now matters. + -} + +{-| The ID of a worker. -} +data WorkerId = WorkerId + { workerHostName :: HostName {-^ the address of the worker -} + , workerPortNumber :: PortNumber {-^ the port number of the worker -} + } deriving (Eq,Ord,Show,Typeable) + +data Worker = Worker + { workerHandle :: Handle + , workerThreadId :: ThreadId + } deriving (Eq,Show) + +data NetworkState = NetworkState + { _pending_add :: !(Set WorkerId) + , _pending_quit :: !(Set WorkerId) + , _workers :: !(Map WorkerId Worker) + } +makeLenses ''NetworkState + +type NetworkStateMonad = StateT NetworkState IO + +-------------------------------------------------------------------------------- +---------------------------------- Controller ---------------------------------- +-------------------------------------------------------------------------------- + +{-| This class extends 'RequestQueueMonad' with the ability to forcibly + disconnect a worker. + -} +class RequestQueueMonad m ⇒ NetworkRequestQueueMonad m where + {-| Forcibly disconnects the given worker; calling this function with the + `WorkerId` of a worker that is no longer connected to the system is + *not* an error; in that case, nothing will happen. + -} + disconnectWorker :: WorkerId → m () + +{-| This is the monad in which the network controller will run. -} +newtype NetworkControllerMonad exploration_mode α = + C (RequestQueueReader exploration_mode WorkerId NetworkStateMonad α) + deriving (Applicative,Functor,Monad,MonadCatchIO,MonadIO,RequestQueueMonad) + +instance HasExplorationMode (NetworkControllerMonad exploration_mode) where + type ExplorationModeFor (NetworkControllerMonad exploration_mode) = exploration_mode + +instance NetworkRequestQueueMonad (NetworkControllerMonad result) where + disconnectWorker worker_id = C $ ask >>= (enqueueRequest $ + debugM ("Disconnecting worker " ++ show worker_id) + >> + Map.lookup worker_id <$> use workers + >>= + maybe (pending_add %= Set.delete worker_id) + (liftIO . flip send QuitWorker . workerHandle) + ) + +-------------------------------------------------------------------------------- +--------------------------------- Other Types ---------------------------------- +-------------------------------------------------------------------------------- + +{-| Callbacks used to to notify when a worker has conneted or disconnected. -} +data NetworkCallbacks = NetworkCallbacks + { notifyConnected :: WorkerId → IO Bool + {-^ callback used to notify that a worker is about to connect; + return 'True' to allow the connection to proceed and 'False' + to veto it + -} + , notifyDisconnected :: WorkerId → IO () + {-^ callback used to notify that a worker has disconnected -} + } + +{-| A default set of callbacks for when you don't care about being notified of connections and disconnections. -} +default_network_callbacks :: NetworkCallbacks +default_network_callbacks = NetworkCallbacks + { notifyConnected = const (return True) + , notifyDisconnected = const (return ()) + } + +{-| Configuration information that indicates whether a process should be run in + supervisor or worker mode. + -} +data NetworkConfiguration shared_configuration supervisor_configuration = + {-| This constructor indicates that the process should run in supervisor mode. -} + SupervisorConfiguration + { shared_configuration :: shared_configuration {-^ configuration information shared between the supervisor and the worker -} + , supervisor_configuration :: supervisor_configuration {-^ configuration information specific to the supervisor -} + , supervisor_port :: WrappedPortID {-^ - for the supervisor, the port on which to listen -} + } + {-| This constructor indicates that the process should be run in worker mode. -} + | WorkerConfiguration + { supervisor_host_name :: HostName {-^ the address of the supervisor to which this worker should connect -} + , supervisor_port :: WrappedPortID {-^ - for the worker, the port to which to connect -} + } + +{-| A newtype wrapper around PortID in order to provide an instance of 'ArgVal'. -} +newtype WrappedPortID = WrappedPortID { unwrapPortID :: PortID } + +instance ArgVal WrappedPortID where + converter = (parsePortID,prettyPortID) + where + (parseInt,prettyInt) = converter + parsePortID = + either Left (\n → + if n >= 0 && n <= (65535 :: Int) + then Right . WrappedPortID . PortNumber . fromIntegral $ n + else Left . text $ "bad port number: must be between 0 and 65535, inclusive (was given " ++ show n ++ ")" + ) + . + parseInt + prettyPortID (WrappedPortID (PortNumber port_number)) = prettyInt . fromIntegral $ port_number + prettyPortID _ = error "a non-numeric port ID somehow made its way in to the configuration" + +instance ArgVal (Maybe WrappedPortID) where + converter = just + +-------------------------------------------------------------------------------- +------------------------------- Generic runners -------------------------------- +-------------------------------------------------------------------------------- + +{- $runners +In this section the full functionality of this module is exposed in case one +does not want the restrictions of the driver interface. If you decide to go in +this direction, then you need to decide whether you want there to be a single +executable for both the supervisor and worker with the process of determining in +which mode it should run taken care of for you, or whether you want to do this +yourself in order to give yourself more control (such as by having separate +supervisor and worker executables) at the price of more work. + +If you want to use a single executable with automated handling of the +supervisor and worker roles, then use 'runExplorer'. Otherwise, use +'runSupervisor' to run the supervisor loop and on each worker use 'runWorker'. + -} + +{-| This runs the supervisor, which will listen for connecting workers. -} +runSupervisor :: + ∀ exploration_mode. + ( Serialize (ProgressFor exploration_mode) + , Serialize (WorkerFinishedProgressFor exploration_mode) + ) ⇒ + ExplorationMode exploration_mode {-^ the exploration mode -} → + (Handle → IO ()) {-^ an action that writes any information needed by the worker to the given handle -} → + NetworkCallbacks {-^ callbacks used to signal when a worker has connected or disconnected; the connect callback has the ability to veto a worker from connecting -} → + PortID {-^ the port id on which to listen for connections -} → + ProgressFor exploration_mode {-^ the initial progress of the run -} → + NetworkControllerMonad exploration_mode () {-^ the controller of the supervisor -} → + Network (RunOutcomeFor exploration_mode) {-^ the outcome of the run -} +runSupervisor + exploration_mode + initializeWorker + NetworkCallbacks{..} + port_id + starting_progress + (C controller) + = liftIO $ do + request_queue ← newRequestQueue + + let receiveStolenWorkloadFromWorker = flip enqueueRequest request_queue .* receiveStolenWorkload + + receiveProgressUpdateFromWorker = flip enqueueRequest request_queue .* receiveProgressUpdate + + receiveFailureFromWorker = flip enqueueRequest request_queue .* receiveWorkerFailure + + receiveFinishedFromWorker worker_id final_progress = flip enqueueRequest request_queue $ do + removal_flag ← Set.member worker_id <$> use pending_quit + infoM $ if removal_flag + then "Worker " ++ show worker_id ++ " has finished, and will be removed." + else "Worker " ++ show worker_id ++ " has finished, and will look for another workload." + receiveWorkerFinishedWithRemovalFlag removal_flag worker_id final_progress + + receiveQuitFromWorker worker_id = flip enqueueRequest request_queue $ do + infoM $ "Worker " ++ show worker_id ++ " has quit." + pending_quit %= Set.delete worker_id + workers %= Map.delete worker_id + removeWorkerIfPresent worker_id + liftIO $ notifyDisconnected worker_id + + sendMessageToWorker message worker_id = do + use workers + >>= + liftIO + . + flip send message + . + workerHandle + . + fromJustOrBust ("Error looking up " ++ show worker_id ++ " to send a message") + . + Map.lookup worker_id + + broadcastMessageToWorkers message = mapM_ (sendMessageToWorker message) + + broadcastProgressUpdateToWorkers = broadcastMessageToWorkers RequestProgressUpdate + + broadcastWorkloadStealToWorkers = broadcastMessageToWorkers RequestWorkloadSteal + + receiveCurrentProgress = receiveProgress request_queue + + sendWorkloadToWorker workload worker_id = do + infoM $ "Activating worker " ++ show worker_id ++ " with workload " ++ show workload + sendMessageToWorker (StartWorkload workload) worker_id + + let port_id_description = showPortID port_id + supervisor_thread_id ← myThreadId + acceptor_thread_id ← forkIO $ + bracket + (debugM ("Acquiring lock on " ++ port_id_description) >> listenOn port_id) + (\socket → debugM ("Releasing lock on " ++ port_id_description) >> sClose socket) + (\socket → forever $ + accept socket >>= + \(workerHandle,workerHostName,workerPortNumber) → do + let identifier = workerHostName ++ ":" ++ show workerPortNumber + debugM $ "Received connection from " ++ identifier + let worker_id = WorkerId{..} + sendToWorker = send workerHandle + flip enqueueRequestAndWait request_queue $ pending_add %= Set.insert worker_id + allowed_to_connect ← liftIO $ notifyConnected worker_id + if allowed_to_connect + then + do debugM $ identifier ++ " is allowed to connect." + initializeWorker workerHandle + workerThreadId ← forkIO ( + receiveAndProcessMessagesFromWorkerUsingHandle + (MessageForSupervisorReceivers{..} :: MessageForSupervisorReceivers exploration_mode WorkerId) + workerHandle + worker_id + `catch` + (\e → case fromException e of + Just ThreadKilled → sendToWorker QuitWorker + Just UserInterrupt → sendToWorker QuitWorker + _ → do enqueueRequest (removeWorker worker_id) request_queue + sendToWorker QuitWorker `catch` (\(_::SomeException) → return ()) + liftIO $ notifyDisconnected worker_id + ) + ) + flip enqueueRequest request_queue $ do + is_pending_add ← Set.member worker_id <$> use pending_add + when is_pending_add $ do + pending_add %= Set.delete worker_id + workers %= Map.insert worker_id Worker{..} + addWorker worker_id + else + do debugM $ identifier ++ " is *not* allowed to connect." + sendToWorker QuitWorker + ) + `catch` + (\e → case fromException e of + Just ThreadKilled → return () + _ → throwTo supervisor_thread_id e + ) + forkControllerThread request_queue controller + flip evalStateT (NetworkState mempty mempty mempty) $ do + supervisor_outcome@SupervisorOutcome{supervisorRemainingWorkers} ← + runSupervisorStartingFrom + exploration_mode + starting_progress + SupervisorCallbacks{..} + (requestQueueProgram (return ()) request_queue) + broadcastMessageToWorkers QuitWorker supervisorRemainingWorkers + liftIO $ killThread acceptor_thread_id + killControllerThreads request_queue + return $ extractRunOutcomeFromSupervisorOutcome supervisor_outcome + +{-| Explores the given tree using multiple processes to achieve parallelism. + + This function grants access to all of the functionality of this adapter, + rather than having to go through the more restricted driver interface. The + signature of this function is very complicated because it is meant to be + used in both the supervisor and worker. The configuration information is + used to determine whether the program is being run in supervisor mode or in + worker mode; in the former case, the configuration is further split into + configuration information that is shared between the supervisor and the + worker and configuration information that is specific to the supervisor. + -} +runExplorer :: + ( Serialize shared_configuration + , Serialize (ProgressFor exploration_mode) + , Serialize (WorkerFinishedProgressFor exploration_mode) + ) ⇒ + (shared_configuration → ExplorationMode exploration_mode) + {-^ a function that constructs the exploration mode given the shared + configuration + -} → + Purity m n {-^ the purity of the tree -} → + IO (NetworkConfiguration shared_configuration supervisor_configuration) + {-^ an action that gets the configuration information (run on both + supervisor and worker processes); this also determines whether we + are in supervisor or worker mode based on whether the constructor + use is respectively 'SupervisorConfiguration' or + 'WorkerConfiguration' + -} → + (shared_configuration → IO ()) + {-^ an action that initializes the global state of the process given the + shared configuration (run on both supervisor and worker processes) + -} → + (shared_configuration → TreeT m (ResultFor exploration_mode)) + {-^ a function that constructs the tree from the shared configuration + (called only on the worker) + -} → + (shared_configuration → supervisor_configuration → IO (ProgressFor exploration_mode)) + {-^ an action that gets the starting progress given the full + configuration information (run only on the supervisor) + -} → + (shared_configuration → supervisor_configuration → NetworkControllerMonad exploration_mode ()) + {-^ a function that constructs the controller for the supervisor (called + only on the supervisor) + -} → + Network (Maybe ((shared_configuration,supervisor_configuration),RunOutcomeFor exploration_mode)) + {-^ if this process is the supervisor, then the outcome of the run as + well as the configuration information wrapped in 'Just'; otherwise + 'Nothing' + -} +runExplorer + constructExplorationMode + purity + getConfiguration + initializeGlobalState + constructTree + getStartingProgress + constructController + = do + configuration ← liftIO $ getConfiguration + case configuration of + SupervisorConfiguration{..} → do + liftIO $ initializeGlobalState shared_configuration + starting_progress ← liftIO $ getStartingProgress shared_configuration supervisor_configuration + termination_result ← + runSupervisor + (constructExplorationMode shared_configuration) + (flip send shared_configuration) + default_network_callbacks + (unwrapPortID supervisor_port) + starting_progress + (constructController shared_configuration supervisor_configuration) + return $ Just ((shared_configuration,supervisor_configuration),termination_result) + WorkerConfiguration{..} → liftIO $ do + handle ← connectTo supervisor_host_name (unwrapPortID supervisor_port) + shared_configuration ← receive handle + Process.runWorkerUsingHandles + (constructExplorationMode shared_configuration) + purity + (constructTree shared_configuration) + handle + handle + return Nothing + +{-| Runs a worker that connects to the supervisor via. the given address and port id. -} +runWorker :: + ( Serialize (ProgressFor exploration_mode) + , Serialize (WorkerFinishedProgressFor exploration_mode) + ) ⇒ + ExplorationMode exploration_mode {-^ the mode in to explore the tree -} → + Purity m n {-^ the purity of the tree -} → + TreeT m (ResultFor exploration_mode) {-^ the tree -} → + HostName {-^ the address of the supervisor -} → + PortID {-^ the port id on which the supervisor is listening -} → + Network () +runWorker exploration_mode purity tree host_name port_id = liftIO $ do + handle ← connectTo host_name port_id + Process.runWorkerUsingHandles exploration_mode purity tree handle handle + +-------------------------------------------------------------------------------- +------------------------------- Utility funtions ------------------------------- +-------------------------------------------------------------------------------- + +{-| Processes the command line and returns the network configuration; it uses + the first argument to determine whether the configuration should be for a + supervisor or for a worker. + -} +getConfiguration :: + Term shared_configuration {-^ configuration that is shared between the supervisor and the worker -} → + Term supervisor_configuration {-^ configuration that is specific to the supervisor -} → + TermInfo {-^ program information (you should at least set 'termDoc' with the program description) -} → + IO (NetworkConfiguration shared_configuration supervisor_configuration) {-^ the configuration obtained from the command line -} +getConfiguration shared_configuration_term supervisor_configuration_term term_info = + execChoice + (no_configuration_term,term_info) + [(supervisorConfigurationTermFor shared_configuration_term supervisor_configuration_term,defTI + { termName = "supervisor" + , termDoc = "Run the program in supervisor mode, waiting for network connections from workers on the specified port." + } + ) + ,(worker_configuration_term,defTI + { termName = "worker" + , termDoc = "Run the program in worker mode, connecting to the specified supervisor to receive workloads." + } + ) + ] + +{-| Constructs a string representation of a port id. (This function is needed + if using an older version of the @Network@ package that doesn't have a + 'Show' instance for 'PortID'.) + -} +showPortID :: PortID → String +showPortID (Service service_name) = "Service " ++ service_name +showPortID (PortNumber port_number) = "Port Number " ++ show port_number +#ifndef mingw32_HOST_OS +showPortID (UnixSocket unix_socket_name) = "Unix Socket " ++ unix_socket_name +#endif + +-------------------------------------------------------------------------------- +------------------------------------ Driver ------------------------------------ +-------------------------------------------------------------------------------- + +{- NOTE: This section is last so that it can see everything before it, as my + use of Template Haskell seems to have disturbed the ability of GHC to + see declarations out of order. + -} + +{-| This is the driver for the network adapter; it consists of a supervisor + that listens for connections and multiple workers that connect to the + supervisor, where the same executable is used for both the supervisor and + the worker. To start the supervisor, run the executable with "supervisor" as + the first argument and "-p PORTID" to specify the port id. To start a + worker, run the executable with "worker" as the first argument, the address + of the supervisor as the second, and the port id as the third. + -} +driver :: + ∀ shared_configuration supervisor_configuration m n exploration_mode. + ( Serialize shared_configuration + , Serialize (ProgressFor exploration_mode) + , Serialize (WorkerFinishedProgressFor exploration_mode) + ) ⇒ + Driver IO shared_configuration supervisor_configuration m n exploration_mode +driver = + case (driverNetwork :: Driver Network shared_configuration supervisor_configuration m n exploration_mode) of + Driver runDriver → Driver (runNetwork . runDriver) + +{-| This is the same as 'driver', but runs in the 'Network' monad. Use this + driver if you want to do other things with the network (such as starting + a subseqent parallel exploration) after the run completes. + -} +driverNetwork :: + ∀ shared_configuration supervisor_configuration m n exploration_mode. + ( Serialize shared_configuration + , Serialize (ProgressFor exploration_mode) + , Serialize (WorkerFinishedProgressFor exploration_mode) + ) ⇒ + Driver Network shared_configuration supervisor_configuration m n exploration_mode +driverNetwork = Driver $ \DriverParameters{..} → do + runExplorer + constructExplorationMode + purity + (getConfiguration shared_configuration_term supervisor_configuration_term program_info) + initializeGlobalState + constructTree + getStartingProgress + constructController + >>= + maybe (return ()) (liftIO . (notifyTerminated <$> fst . fst <*> snd . fst <*> snd)) + +-------------------------------------------------------------------------------- +----------------------------------- Internal ----------------------------------- +-------------------------------------------------------------------------------- + +fromJustOrBust message = fromMaybe (error message) + +no_configuration_term = ret (pure $ helpFail Plain Nothing) + +supervisorConfigurationTermFor shared_configuration_term supervisor_configuration_term = + SupervisorConfiguration + <$> shared_configuration_term + <*> supervisor_configuration_term + <*> (required + $ + opt Nothing + ((optInfo ["p","port"]) + { optName = "PORT" + , optDoc = "port on which to listen for workers" + } + ) + ) + +worker_configuration_term = + WorkerConfiguration + <$> (required + $ + pos 0 + Nothing + posInfo + { posName = "HOST_NAME" + , posDoc = "supervisor host name" + } + ) + <*> (required + $ + pos 1 + Nothing + posInfo + { posName = "HOST_PORT" + , posDoc = "supervisor host port" + } + ) diff --git a/tests/tests.hs b/tests/tests.hs new file mode 100644 --- /dev/null +++ b/tests/tests.hs @@ -0,0 +1,139 @@ +-- Language extensions {{{ +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE UnicodeSyntax #-} +{-# LANGUAGE ViewPatterns #-} +-- }}} + +-- Imports {{{ +import Control.Monad (forever,replicateM_) +import Control.Monad.IO.Class (MonadIO(..)) + +import Control.Concurrent (forkIO,threadDelay) +import Control.Concurrent.STM (atomically,modifyTVar,newTVarIO,readTVar,writeTVar) + +import Data.Functor ((<$>)) +import Data.IORef (modifyIORef,newIORef,readIORef) +import Data.Monoid ((<>),mempty) + +import GHC.Conc (unsafeIOToSTM) + +import Network (PortID(..)) + +import qualified System.Log.Logger as Logger +import System.Log.Logger (Priority(DEBUG,INFO),rootLoggerName,setLevel,updateGlobalLogger) +import System.Log.Logger.TH +import System.Random (randomRIO) + +import Test.Framework +import Test.Framework.Providers.HUnit +import Test.HUnit hiding (Test,Path) + +import LogicGrowsOnTrees +import LogicGrowsOnTrees.Checkpoint +import LogicGrowsOnTrees.Examples.Queens +import LogicGrowsOnTrees.Parallel.Adapter.Network +import LogicGrowsOnTrees.Parallel.ExplorationMode +import LogicGrowsOnTrees.Parallel.Purity (Purity(Pure)) +-- }}} + +-- Logging Functions {{{ +deriveLoggers "Logger" [DEBUG,INFO] +-- }}} + +-- Helper Functions {{{ +remdups :: (Eq a) => [a] -> [a] -- {{{ +remdups [] = [] +remdups (x : []) = [x] +remdups (x : xx : xs) + | x == xx = remdups (x : xs) + | otherwise = x : remdups (xx : xs) +-- }}} +-- }}} + +main = do + -- updateGlobalLogger rootLoggerName (setLevel DEBUG) + defaultMain tests + +tests = -- {{{ + [testCase "one process" . runTest $ \changeNumberOfWorkers → do + changeNumberOfWorkers (const 0) + changeNumberOfWorkers (const 1) + ,testCase "two processes" . runTest $ \changeNumberOfWorkers → do + changeNumberOfWorkers (3-) + ,testCase "many processes" . runTest $ \changeNumberOfWorkers → liftIO (randomRIO (0,1::Int)) >>= \i → case i of + 0 → changeNumberOfWorkers (\i → if i > 1 then i-1 else i) + 1 → changeNumberOfWorkers (+1) + ] + where + runTest generateNoise = do + let tree = nqueensCount 15 + port_id = PortNumber 54210 + progresses_ref ← newIORef [] + worker_ids_var ← newTVarIO [] + let changeNumberOfWorkers computeNewNumberOfWorkers = do + old_number_of_workers ← liftIO . atomically $ length <$> readTVar worker_ids_var + let new_number_of_workers = computeNewNumberOfWorkers old_number_of_workers + case new_number_of_workers `compare` old_number_of_workers of + EQ → return () + GT → liftIO + . + replicateM_ (new_number_of_workers-old_number_of_workers) + . + forkIO + . + unsafeRunNetwork + $ + runWorker + AllMode + Pure + tree + "localhost" + port_id + LT → replicateM_ (old_number_of_workers-new_number_of_workers) $ + (liftIO . atomically $ do + worker_ids ← readTVar worker_ids_var + let number_of_workers = length worker_ids + if number_of_workers > 0 + then do + index_to_remove ← unsafeIOToSTM $ randomRIO (0,number_of_workers-1) + writeTVar worker_ids_var $ take index_to_remove worker_ids ++ drop (number_of_workers+1) worker_ids + return . Just $ worker_ids !! index_to_remove + -- Because there is a delay between when a + -- disconnect request is made and when it is + -- processed, sometimes the number of workers + -- will decrease behind our back. + else return Nothing + ) >>= maybe (return ()) disconnectWorker + notifyConnected worker_id = do + atomically $ modifyTVar worker_ids_var (worker_id:) + return True + notifyDisconnected _ = return () + RunOutcome _ termination_reason ← + unsafeRunNetwork $ + runSupervisor + AllMode + (const $ return ()) + NetworkCallbacks{..} + port_id + mempty + (do changeNumberOfWorkers (const 1) + forever $ do + liftIO $ threadDelay 1000 + requestProgressUpdate >>= liftIO . modifyIORef progresses_ref . (:) + generateNoise changeNumberOfWorkers + ) + result ← case termination_reason of + Aborted _ → error "prematurely aborted" + Completed result → return result + Failure _ message → error message + let correct_result = exploreTree tree + result @?= correct_result + progresses ← remdups <$> readIORef progresses_ref + replicateM_ 4 $ randomRIO (0,length progresses-1) >>= \i → do + let Progress checkpoint result = progresses !! i + result @=? exploreTreeStartingFromCheckpoint (invertCheckpoint checkpoint) tree + correct_result @=? result <> (exploreTreeStartingFromCheckpoint checkpoint tree) +-- }}}