LogicGrowsOnTrees.Parallel.Common.Worker

The Worker module contains the workhorse code of the parallelization infrastructure in the form of the forkWorkerThread function, which explores a tree step by step while continuously polling for requests; for more details see forkWorkerThread.

Types

Worker interaction

data ProgressUpdate progress

A progress update sent to the supervisor; it has a component which contains information about how much of the tree has been explored and what results have been found so far, as well as the remaining Workload to be completed by this worker.

Constructors

ProgressUpdate

Fields

progressUpdateProgress :: progress

progressUpdateRemainingWorkload :: Workload

Instances

Eq progress => Eq (ProgressUpdate progress)

Show progress => Show (ProgressUpdate progress)

Serialize progress_1627696524 => Serialize (ProgressUpdate progress_1627696524)

type ProgressUpdateFor exploration_mode = ProgressUpdate (ProgressFor exploration_mode)

A convenient type alias for the type of ProgressUpdate associated with the given exploration mode.

data StolenWorkload progress

A stolen workload sent to the supervisor; in addition to a component with the stolen Workload itself, it also has a ProgressUpdate component, which is required in order to maintain the invariant that all of the Workloads that the supervisor has on file (both assigned to workers and unassigned) plus the current progress equals the full tree.

Constructors

StolenWorkload

Fields

stolenWorkloadProgressUpdate :: ProgressUpdate progress

stolenWorkload :: Workload

Instances

Eq progress => Eq (StolenWorkload progress)

Show progress => Show (StolenWorkload progress)

Serialize progress_1627696606 => Serialize (StolenWorkload progress_1627696606)

type StolenWorkloadFor exploration_mode = StolenWorkload (ProgressFor exploration_mode)

A convenient type alias for the type of StolenWorkload associated with the the given exploration mode.

type WorkerRequestQueue progress = IORef [WorkerRequest progress]

A queue of worker requests.

NOTE: Although the type is a list, and requests are added by prepending them to the list, it still acts as a queue because the worker will reverse the list before processing the requests.

type WorkerRequestQueueFor exploration_mode = WorkerRequestQueue (ProgressFor exploration_mode)

A convenient type alias for the type of WorkerRequestQueue associated with the given exploration mode.

data WorkerEnvironment progress

The environment of a running worker.

Constructors

WorkerEnvironment

Fields

workerInitialPath :: Path

the initial path of the worker's workload

workerThreadId :: ThreadId

the thread id of the worker thread

workerPendingRequests :: WorkerRequestQueue progress

the request queue for the worker

workerTerminationFlag :: IVar ()

an IVar that is filled when the worker terminates

type WorkerEnvironmentFor exploration_mode = WorkerEnvironment (ProgressFor exploration_mode)

A convenient type alias for the type of WorkerEnvironment associated with the given exploration mode.

data WorkerTerminationReason worker_final_progress

The reason why a worker terminated.

Constructors

WorkerFinished worker_final_progress

worker completed normally without error; included is the final result

WorkerFailed String

worker failed; included is the message of the failure (this would have been a value of type SomeException if it were not for the fact that this value will often have to be sent over communication channels and exceptions cannot be serialized (as they have unknown type), meaning that it usually has to be turned into a String via show anyway)

WorkerAborted

worker was aborted by either an external request or the ThreadKilled or UserInterrupt exceptions

Instances

Functor WorkerTerminationReason

The Functor instance lets you manipulate the final progress value when the termination reason is WorkerFinished.

Eq worker_final_progress => Eq (WorkerTerminationReason worker_final_progress)

Show worker_final_progress => Show (WorkerTerminationReason worker_final_progress)

type WorkerTerminationReasonFor exploration_mode = WorkerTerminationReason (WorkerFinishedProgressFor exploration_mode)

A convenient type alias for the type of WorkerTerminationReason associated with the given exploration mode.

type family WorkerPushActionFor exploration_mode :: *

The action that a worker can take to push a result to the supervisor; this type is effectively null (with the exact value absurd) for all modes except FoundModeUsingPush.

Functions

forkWorkerThread

Arguments

:: ExplorationMode exploration_mode

the mode in to explore the tree

-> Purity m n

the purity of the tree

-> (WorkerTerminationReasonFor exploration_mode -> IO ())

the action to run when the worker has terminated

-> TreeT m (ResultFor exploration_mode)

the tree

-> Workload

the workload for the worker

-> WorkerPushActionFor exploration_mode

the action to push a result to the supervisor; this should be equal to absurd except when the exploration mode is FoundModeUsingPush.

-> IO (WorkerEnvironmentFor exploration_mode)

the environment for the worker

The forkWorkerThread function is the workhorse of the parallization infrastructure; it explores a tree in a separate thread while polling for requests. Specifically, the worker alternates between stepping through the tree and checking to see if there are any new requests in the queue.

The worker is optimized around the observation that the vast majority of its time is spent exploring the tree rather than responding to requests, and so the amount of overhead needed to check if any requests are present needs to be minimized at the cost of possibly delaying a response to an incoming request. For this reason, it uses an IORef for the queue to minimize the cost of peeking at it rather than an MVar or some other thread synchronization variable; the trade-off is that if a request is added to the queue by a different processor then it might not be noticed immediately the caches get synchronized. Likewise, the request queue uses the List type rather than something like Data.Sequence for simplicity; the vast majority of the time the worker will encounter an empty list, and on the rare occasion when the list is non-empty it will be short enough that reverseing it will not pose a significant cost.

At any given point in the exploration, there is an initial path which locates the subtree that was given as the original workload, a cursor which indicates the subtree within this subtree that makes up the current workload, and the context which indicates the current location in the subtree that is being explored. All workers start with an empty cursor; when a workload is stolen, decisions made early on in the the context are frozen and moved into the cursor because if they were not then when the worker backtracked it would explore a workload that it just gave away, resulting in some results being observed twice.

The worker terminates either if it finishes exploring all the nodes in its (current) workload, if an error occurs, or if it is aborted either via. the ThreadKilled and UserInterrupt exceptions or by an abort request placed in the request queue.

sendAbortRequest :: WorkerRequestQueue progress -> IO ()

Sends a request to abort.

sendProgressUpdateRequest

Arguments

:: WorkerRequestQueue progress

the request queue

-> (ProgressUpdate progress -> IO ())

the action to perform when the update is available

-> IO ()

Sends a request for a progress update along with a response action to perform when the progress update is available.

sendWorkloadStealRequest

Arguments

:: WorkerRequestQueue progress

the request queue

-> (Maybe (StolenWorkload progress) -> IO ())

the action to perform when the update is available

-> IO ()

Sends a request to steal a workload along with a response action to perform when the progress update is available.

exploreTreeGeneric :: (WorkerPushActionFor exploration_mode ~ (Void -> ()), ResultFor exploration_mode ~ α) => ExplorationMode exploration_mode -> Purity m n -> TreeT m α -> IO (WorkerTerminationReason (FinalResultFor exploration_mode))

Explores a tree with the specified purity using the given mode by forking a worker thread and waiting for it to finish; it exists to facilitate testing and benchmarking and is not a function that you are likely to ever have a need for yourself.