Top Description Fields Constructors Methods
rx.internal.util

public Class RxRingBuffer

extends Object
implements Subscription
Class Inheritance
All Implemented Interfaces
rx.Subscription
Imports
java.util.Queue, rx.Observer, .Subscription, rx.exceptions.MissingBackpressureException, rx.internal.operators.NotificationLite, rx.internal.util.atomic.SpscAtomicArrayQueue, rx.internal.util.unsafe.SpmcArrayQueue, .SpscArrayQueue, .UnsafeAccess
This assumes Spsc or Spmc usage. This means only a single producer calling the on* methods. This is the Rx contract of an Observer (see http://reactivex.io/documentation/contract.html). Concurrent invocations of on* methods will not be thread-safe.

Field Summary

Modifier and TypeField and Description
private Queue<Object>
queue

Queue implementation testing that led to current choices of data structures: With synchronized LinkedList 

 Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 19118392.046  1002814.238    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    17891.641      252.747    ops/s

With MpscPaddedQueue (single consumer, so failing 1 unit test)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 22164483.238  3035027.348    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    23154.303      602.548    ops/s


With ConcurrentLinkedQueue (tracking count separately)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 17353906.092   378756.411    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    19224.411     1010.610    ops/s

With ConcurrentLinkedQueue (using queue.size() method for count)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 23951121.098  1982380.330    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5     1142.351       33.592    ops/s

With SynchronizedQueue (synchronized LinkedList ...
private final int
size

public static final int
SIZE

public volatile Object
terminalState

We store the terminal state separately so it doesn't count against the size.

Constructor Summary

AccessConstructor and Description
private
RxRingBuffer(Queue<Object> queue, int size)

private
RxRingBuffer(boolean spmc, int size)

pack-priv
RxRingBuffer()

Method Summary

Modifier and TypeMethod and Description
public boolean
accept(Object o, Observer<T> child)

public Throwable
asError(Object o)

public int
available()

public int
capacity()

public int
count()

public static RxRingBuffer
getSpmcInstance()

public static RxRingBuffer
getSpscInstance()

public Object
getValue(Object o)

public boolean
isCompleted(Object o)

public boolean
isEmpty()

public boolean
isError(Object o)

public boolean
isUnsubscribed()

Implements rx.Subscription.isUnsubscribed.

Indicates whether this Subscription is currently unsubscribed.
public void
onCompleted()

public void
onError(Throwable t)

public void
onNext(Object
the value to buffer
o)

public Object
peek()

public Object
poll()

public synchronized void
release()

public void
unsubscribe()

Implements rx.Subscription.unsubscribe.

Stops the receipt of notifications on the Subscriber that was registered when this Subscription was received.
Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAlltoStringwaitwaitwait

Field Detail

queueback to summary
private Queue<Object> queue

Queue implementation testing that led to current choices of data structures: With synchronized LinkedList 

 Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 19118392.046  1002814.238    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    17891.641      252.747    ops/s

With MpscPaddedQueue (single consumer, so failing 1 unit test)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 22164483.238  3035027.348    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    23154.303      602.548    ops/s


With ConcurrentLinkedQueue (tracking count separately)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 17353906.092   378756.411    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    19224.411     1010.610    ops/s

With ConcurrentLinkedQueue (using queue.size() method for count)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 23951121.098  1982380.330    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5     1142.351       33.592    ops/s

With SynchronizedQueue (synchronized LinkedList ... no object pooling)

r.i.RxRingBufferPerf.createUseAndDestroy1       thrpt         5 33231667.136   685757.510    ops/s
r.i.RxRingBufferPerf.createUseAndDestroy1000    thrpt         5    74623.614     5493.766    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 22907359.257   707026.632    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    22222.410      320.829    ops/s

With ArrayBlockingQueue

Benchmark                                            Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.createUseAndDestroy1       thrpt         5  2389804.664    68990.804    ops/s
r.i.RxRingBufferPerf.createUseAndDestroy1000    thrpt         5    27384.274     1411.789    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 26497037.559    91176.247    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    17985.144      237.771    ops/s

With ArrayBlockingQueue and Object Pool

Benchmark                                            Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.createUseAndDestroy1       thrpt         5 12465685.522   399070.770    ops/s
r.i.RxRingBufferPerf.createUseAndDestroy1000    thrpt         5    27701.294      395.217    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 26399625.086   695639.436    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    17985.427      253.190    ops/s

With SpscArrayQueue (single consumer, so failing 1 unit test)
 - requires access to Unsafe

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.createUseAndDestroy1       thrpt         5  1922996.035    49183.766    ops/s
r.i.RxRingBufferPerf.createUseAndDestroy1000    thrpt         5    70890.186     1382.550    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 80637811.605  3509706.954    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    71822.453     4127.660    ops/s


With SpscArrayQueue and Object Pool (object pool improves createUseAndDestroy1 by 10x)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.createUseAndDestroy1       thrpt         5 25220069.264  1329078.785    ops/s
r.i.RxRingBufferPerf.createUseAndDestroy1000    thrpt         5    72313.457     3535.447    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1       thrpt         5 81863840.884  2191416.069    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    73140.822     1528.764    ops/s

With SpmcArrayQueue
 - requires access to Unsafe

Benchmark                                            Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1       thrpt         5 27630345.474   769219.142    ops/s
r.i.RxRingBufferPerf.spmcCreateUseAndDestroy1000    thrpt         5    80052.046     4059.541    ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1       thrpt         5 44449524.222   563068.793    ops/s
r.i.RxRingBufferPerf.spmcRingBufferAddRemove1000    thrpt         5    65231.253     1805.732    ops/s

With SpmcArrayQueue and ObjectPool (object pool improves createUseAndDestroy1 by 10x)

Benchmark                                        Mode   Samples        Score  Score error    Units
r.i.RxRingBufferPerf.createUseAndDestroy1       thrpt         5 18489343.061  1011872.825    ops/s
r.i.RxRingBufferPerf.createUseAndDestroy1000    thrpt         5    46416.434     1439.144    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove        thrpt         5 38280945.847  1071801.279    ops/s
r.i.RxRingBufferPerf.ringBufferAddRemove1000    thrpt         5    42337.663     1052.231    ops/s

--------------

When UnsafeAccess.isUnsafeAvailable() == true we can use the Spmc/SpscArrayQueue implementations.
sizeback to summary
private final int size
SIZEback to summary
public static final int SIZE
terminalStateback to summary
public volatile Object terminalState

We store the terminal state separately so it doesn't count against the size. We don't just +1 the size since some of the queues require sizes that are a power of 2. This is a subjective thing ... wanting to keep the size (ie 1024) the actual number of onNext that can be sent rather than something like 1023 onNext + 1 terminal event. It also simplifies checking that we have received only 1 terminal event, as we don't need to peek at the last item or retain a boolean flag.

Constructor Detail

RxRingBufferback to summary
private RxRingBuffer(Queue<Object> queue, int size)
RxRingBufferback to summary
private RxRingBuffer(boolean spmc, int size)
RxRingBufferback to summary
pack-priv RxRingBuffer()

Method Detail

acceptback to summary
public boolean accept(Object o, Observer<T> child)
Annotations
@SuppressWarnings:unchecked, rawtypes
asErrorback to summary
public Throwable asError(Object o)
availableback to summary
public int available()
capacityback to summary
public int capacity()
countback to summary
public int count()
getSpmcInstanceback to summary
public static RxRingBuffer getSpmcInstance()
getSpscInstanceback to summary
public static RxRingBuffer getSpscInstance()
getValueback to summary
public Object getValue(Object o)
isCompletedback to summary
public boolean isCompleted(Object o)
isEmptyback to summary
public boolean isEmpty()
isErrorback to summary
public boolean isError(Object o)
isUnsubscribedback to summary
public boolean isUnsubscribed()

Implements rx.Subscription.isUnsubscribed.

Doc from rx.Subscription.isUnsubscribed.

Indicates whether this Subscription is currently unsubscribed.

Returns:boolean

true if this Subscription is currently unsubscribed, false otherwise

Annotations
@Override
onCompletedback to summary
public void onCompleted()
onErrorback to summary
public void onError(Throwable t)
onNextback to summary
public void onNext(Object o) throws MissingBackpressureException
Parameters
o:Object

the value to buffer

Exceptions
MissingBackpressureException:
if more onNext are sent than have been requested
peekback to summary
public Object peek()
pollback to summary
public Object poll()
releaseback to summary
public synchronized void release()
unsubscribeback to summary
public void unsubscribe()

Implements rx.Subscription.unsubscribe.

Doc from rx.Subscription.unsubscribe.

Stops the receipt of notifications on the Subscriber that was registered when this Subscription was received.

This allows deregistering an Subscriber before it has finished receiving all events (i.e. before onCompleted is called).

Annotations
@Override