LogicGrowsOnTrees-network (empty) → 1.0.0
raw patch · 6 files changed
+904/−0 lines, 6 filesdep +HUnitdep +LogicGrowsOnTreesdep +LogicGrowsOnTrees-networksetup-changed
Dependencies added: HUnit, LogicGrowsOnTrees, LogicGrowsOnTrees-network, MonadCatchIO-transformers, base, cereal, cmdtheline, composition, containers, hslogger, hslogger-template, lens, mtl, network, pretty, random, stm, test-framework, test-framework-hunit, transformers
Files
- LICENSE +22/−0
- LogicGrowsOnTrees-network.cabal +88/−0
- Setup.hs +2/−0
- examples/count-all-nqueens-solutions.hs +29/−0
- sources/LogicGrowsOnTrees/Parallel/Adapter/Network.hs +624/−0
- tests/tests.hs +139/−0
+ LICENSE view
@@ -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.
+ LogicGrowsOnTrees-network.cabal view
@@ -0,0 +1,88 @@+Name: LogicGrowsOnTrees-network+Version: 1.0.0+License: BSD3+License-file: LICENSE+Author: Gregory Crosswhite+Maintainer: Gregory Crosswhite <gcrosswhite@gmail.com>+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
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ examples/count-all-nqueens-solutions.hs view
@@ -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
+ sources/LogicGrowsOnTrees/Parallel/Adapter/Network.hs view
@@ -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"+ }+ )
+ tests/tests.hs view
@@ -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)+-- }}}