Class SubmissionPublisher.BufferedSubscription<T>
- All Implemented Interfaces:
ForkJoinPool.ManagedBlocker
,Flow.Subscription
- Enclosing class:
- SubmissionPublisher<T>
The publisher guarantees a single producer via its lock. We ensure in this class that there is at most one consumer. The request and cancel methods must be fully thread-safe but are coded to exploit the most common case in which they are only called by consumers (usually within onNext).
Execution control is managed using the ACTIVE ctl bit. We ensure that a task is active when consumable items (and usually, SUBSCRIBE, ERROR or COMPLETE signals) are present and there is demand (unfilled requests). This is complicated on the creation side by the possibility of exceptions when trying to execute tasks. These eventually force DISABLED state, but sometimes not directly. On the task side, termination (clearing ACTIVE) that would otherwise race with producers or request() calls uses the CONSUME keep-alive bit to force a recheck.
The ctl field also manages run state. When DISABLED, no further updates are possible. Disabling may be preceded by setting ERROR or COMPLETE (or both -- ERROR has precedence), in which case the associated Subscriber methods are invoked, possibly synchronously if there is no active consumer task (including cases where execute() failed). The cancel() method is supported by treating as ERROR but suppressing onError signal.
Support for blocking also exploits the fact that there is only one possible waiter. ManagedBlocker-compatible control fields are placed in this class itself rather than in wait-nodes. Blocking control relies on the "waiter" field. Producers set the field before trying to block, but must then recheck (via offer) before parking. Signalling then just unparks and clears waiter field. If the producer and/or consumer are using a ForkJoinPool, the producer attempts to help run consumer tasks via ForkJoinPool.helpAsyncBlocker before blocking.
-
Field Summary
FieldsModifier and TypeFieldDescriptionprivate static final int
(package private) static final int
(package private) Object[]
private static final int
(package private) static final int
(package private) static final int
(package private) int
private static final long
(package private) static final int
Initial buffer capacity used when maxBufferCapacity is greater.(package private) long
private static final long
(package private) static final int
(package private) static final int
(package private) Executor
(package private) int
private static final long
(package private) static final long
(package private) int
(package private) SubmissionPublisher.BufferedSubscription<T>
(package private) SubmissionPublisher.BufferedSubscription<T>
(package private) BiConsumer<? super Flow.Subscriber<? super T>,
? super Throwable> (package private) Throwable
(package private) T
(package private) int
(package private) static final int
(package private) Flow.Subscriber<? super T>
(package private) int
private static final long
(package private) long
private static final sun.misc.Unsafe
(package private) Thread
-
Constructor Summary
ConstructorsConstructorDescriptionBufferedSubscription
(Flow.Subscriber<? super T> subscriber, Executor executor, BiConsumer<? super Flow.Subscriber<? super T>, ? super Throwable> onNextHandler, int maxBufferCapacity) -
Method Summary
Modifier and TypeMethodDescriptionfinal boolean
block()
void
cancel()
Causes consumer task to exit if active (without reporting onError unless there is already a pending error), and disables.private boolean
checkControl
(Flow.Subscriber<? super T> s, int c) Responds to control events in consume().private boolean
checkDemand
(int c) Responds to apparent zero demand in consume().private boolean
checkEmpty
(Flow.Subscriber<? super T> s, int c) Responds to apparent emptiness in consume().(package private) final void
consume()
Consumer loop, called from ConsumerTask, or indirectly when helping during submit.private void
detach()
Nulls out most fields, mainly to avoid garbage retention until publisher unsubscribes, but also to help cleanly stop upon error by nulling required components.(package private) final int
Returns estimated number of buffered items, or -1 if disabled.private int
growAndAdd
(Object[] a, T item) Tries to create or expand buffer, then adds item if possible.private void
handleOnNext
(Flow.Subscriber<? super T> s, Throwable ex) Processes exception in Subscriber.onNext.(package private) final boolean
final boolean
(package private) final int
Tries to add item and start consumer task if necessary.(package private) final void
(package private) final void
Issues error signal, asynchronously if a task is running, else synchronously.(package private) final void
void
request
(long n) Adds to demand and possibly starts task.private void
private int
startOnOffer
(int stat) Tries to start consumer task after offer.private void
Tries to start consumer task upon a signal or request; disables on failure.(package private) final int
Spins/helps/blocks while offer returns 0.(package private) final int
timedOffer
(T item, long nanos) Timeout version; similar to submit.toString()
-
Field Details
-
timeout
long timeout -
demand
volatile long demand -
maxCapacity
int maxCapacity -
putStat
int putStat -
ctl
volatile int ctl -
head
volatile int head -
tail
int tail -
array
Object[] array -
subscriber
Flow.Subscriber<? super T> subscriber -
executor
Executor executor -
onNextHandler
BiConsumer<? super Flow.Subscriber<? super T>,? super Throwable> onNextHandler -
pendingError
-
waiter
-
putItem
T putItem -
next
-
nextRetry
SubmissionPublisher.BufferedSubscription<T> nextRetry -
ACTIVE
static final int ACTIVE- See Also:
-
CONSUME
static final int CONSUME- See Also:
-
DISABLED
static final int DISABLED- See Also:
-
ERROR
static final int ERROR- See Also:
-
SUBSCRIBE
static final int SUBSCRIBE- See Also:
-
COMPLETE
static final int COMPLETE- See Also:
-
INTERRUPTED
static final long INTERRUPTED- See Also:
-
DEFAULT_INITIAL_CAP
static final int DEFAULT_INITIAL_CAPInitial buffer capacity used when maxBufferCapacity is greater. Must be a power of two.- See Also:
-
U
private static final sun.misc.Unsafe U -
CTL
private static final long CTL -
TAIL
private static final long TAIL -
HEAD
private static final long HEAD -
DEMAND
private static final long DEMAND -
ABASE
private static final int ABASE -
ASHIFT
private static final int ASHIFT
-
-
Constructor Details
-
BufferedSubscription
BufferedSubscription(Flow.Subscriber<? super T> subscriber, Executor executor, BiConsumer<? super Flow.Subscriber<? super T>, ? super Throwable> onNextHandler, int maxBufferCapacity)
-
-
Method Details
-
toString
-
isDisabled
final boolean isDisabled() -
estimateLag
final int estimateLag()Returns estimated number of buffered items, or -1 if disabled. -
offer
Tries to add item and start consumer task if necessary.- Returns:
- -1 if disabled, 0 if dropped, else estimated lag
-
growAndAdd
Tries to create or expand buffer, then adds item if possible. -
submit
Spins/helps/blocks while offer returns 0. Called only if initial offer return 0. -
timedOffer
Timeout version; similar to submit. -
startOnOffer
private int startOnOffer(int stat) Tries to start consumer task after offer.- Returns:
- -1 if now disabled, else argument
-
signalWaiter
-
detach
private void detach()Nulls out most fields, mainly to avoid garbage retention until publisher unsubscribes, but also to help cleanly stop upon error by nulling required components. -
onError
Issues error signal, asynchronously if a task is running, else synchronously. -
startOrDisable
private void startOrDisable()Tries to start consumer task upon a signal or request; disables on failure. -
onComplete
final void onComplete() -
onSubscribe
final void onSubscribe() -
cancel
public void cancel()Causes consumer task to exit if active (without reporting onError unless there is already a pending error), and disables.- Specified by:
cancel
in interfaceFlow.Subscription
-
request
public void request(long n) Adds to demand and possibly starts task.- Specified by:
request
in interfaceFlow.Subscription
- Parameters:
n
- the increment of demand; a value ofLong.MAX_VALUE
may be considered as effectively unbounded
-
isReleasable
public final boolean isReleasable()- Specified by:
isReleasable
in interfaceForkJoinPool.ManagedBlocker
-
block
public final boolean block()- Specified by:
block
in interfaceForkJoinPool.ManagedBlocker
-
consume
final void consume()Consumer loop, called from ConsumerTask, or indirectly when helping during submit. -
checkControl
Responds to control events in consume(). -
checkEmpty
Responds to apparent emptiness in consume(). -
checkDemand
private boolean checkDemand(int c) Responds to apparent zero demand in consume(). -
handleOnNext
Processes exception in Subscriber.onNext.
-