groovyx.gpars.dataflow.operator
Class ForkingDataflowOperatorActor

java.lang.Object
  groovyx.gpars.serial.WithSerialId
      groovyx.gpars.actor.impl.MessageStream
          groovyx.gpars.actor.Actor
              groovyx.gpars.actor.AbstractLoopingActor
                  groovyx.gpars.actor.StaticDispatchActor<java.lang.Object>
                      groovyx.gpars.dataflow.operator.DataflowProcessorActor
                          groovyx.gpars.dataflow.operator.DataflowOperatorActor
                              groovyx.gpars.dataflow.operator.ForkingDataflowOperatorActor

All Implemented Interfaces:

java.io.Serializable

final class ForkingDataflowOperatorActor
extends DataflowOperatorActor

An operator's internal actor. Repeatedly polls inputs and once they're all available it performs the operator's body. The operator's body is executed in as a separate task, allowing multiple copies of the body to be run concurrently. The maxForks property guards the maximum number or concurrently run copies.

Author:

Vaclav Pech

Nested Class Summary

Nested classes/interfaces inherited from class groovyx.gpars.actor.Actor

Actor.MyRemoteHandle, Actor.RemoteActor

Nested classes/interfaces inherited from class groovyx.gpars.actor.impl.MessageStream

MessageStream.RemoteMessageStream, MessageStream.SendTo

Field Summary

private int	maxForks
private java.util.concurrent.Semaphore	semaphore
private Pool	threadPool

Fields inherited from class groovyx.gpars.dataflow.operator.DataflowProcessorActor

CANNOT_OBTAIN_THE_SEMAPHORE_TO_FORK_OPERATOR_S_BODY, code, inputs, outputs, owningProcessor, stoppingGently

Fields inherited from class groovyx.gpars.actor.Actor

ACTOR_HAS_ALREADY_BEEN_STARTED, CANNOT_SEND_REPLIES_NO_SENDER_HAS_BEEN_REGISTERED, currentThread, parallelGroup, START_MESSAGE, STOP_MESSAGE, TERMINATE_MESSAGE, TIMEOUT, TIMEOUT_MESSAGE, timer

Fields inherited from class groovyx.gpars.serial.WithSerialId

serialHandle

Constructor Summary

ForkingDataflowOperatorActor(DataflowOperator owningOperator, PGroup group, java.util.List outputs, java.util.List inputs, groovy.lang.Closure code, int maxForks)

Method Summary

protected void	forwardPoisonPill(java.lang.Object data)
(package private) void	startTask(java.util.List<java.lang.Object> results)

Methods inherited from class groovyx.gpars.dataflow.operator.DataflowOperatorActor

afterStart, onMessage

Methods inherited from class groovyx.gpars.dataflow.operator.DataflowProcessorActor

afterStop, fireMessageArrived, checkPoison, isControlMessage, onException, registerChannelListenersToAllInputs, reportException, send

Methods inherited from class groovyx.gpars.actor.AbstractLoopingActor

getSender, hasBeenStopped, initialize, isActive, isFair, makeFair, reply, replyIfExists, setParallelGroup, setTimeout, silentStart, start, stop, sweepNextMessage, terminate

Methods inherited from class groovyx.gpars.actor.Actor

createActorMessage, createRemoteHandle, deregisterCurrentActorWithThread, getJoinLatch, getParallelGroup, handleException, handleInterrupt, handleStart, handleTermination, handleTimeout, isActorThread, join, join, join, join, onStop, registerCurrentActorWithThread, sendAndContinue, sendAndPromise, threadBoundActor

Methods inherited from class groovyx.gpars.actor.impl.MessageStream

call, getRemoteClass, leftShift, send, send, sendAndWait, sendAndWait, sendAndWait

Methods inherited from class groovyx.gpars.serial.WithSerialId

getOrCreateSerialHandle, writeReplace

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

semaphore

private final java.util.concurrent.Semaphore semaphore

threadPool

private final Pool threadPool

maxForks

private final int maxForks

Constructor Detail

ForkingDataflowOperatorActor

ForkingDataflowOperatorActor(DataflowOperator owningOperator,
                             PGroup group,
                             java.util.List outputs,
                             java.util.List inputs,
                             groovy.lang.Closure code,
                             int maxForks)

Method Detail

startTask

void startTask(java.util.List<java.lang.Object> results)

Overrides:

startTask in class DataflowOperatorActor

forwardPoisonPill

protected void forwardPoisonPill(java.lang.Object data)

Overrides:

forwardPoisonPill in class DataflowProcessorActor