Package org.apache.sysds.runtime.instructions.ooc

# Class SubscribableTaskQueue<T>

java.lang.Object
org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue<OOCStream.QueueCallback<T>>
org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue<T>

All Implemented Interfaces:
  `OOCStream<T>`, `OOCStreamable<T>`

Direct Known Subclasses:
  `SourceOOCStream`

public class SubscribableTaskQueue<T> extends LocalTaskQueue<OOCStream.QueueCallback<T>> implements OOCStream<T>

## Nested Class Summary

## Nested classes/interfaces inherited from interface org.apache.sysds.runtime.instructions.ooc.OOCStream

`OOCStream.GroupQueueCallback<T>, OOCStream.QueueCallback<T>, OOCStream.SimpleQueueCallback<T>`

## Field Summary

### Fields inherited from class org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue

`MAX_SIZE, NO_MORE_TASKS`

## Constructor Summary

Constructors
Constructor
Description
`SubscribableTaskQueue()`

## Method Summary

Modifier and Type
Method
Description
`void`
`addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`void`
`addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`void`
`clearDownstreamMessageRelays()`
`void`
`clearUpstreamMessageRelays()`
`void`
`closeInput()`
Synchronized (logical) insert of a NO_MORE_TASKS symbol at the end of the FIFO queue in order to mark that no more tasks will be inserted into the queue.
`T`
`dequeue()`
`OOCStream.QueueCallback<T>`
`dequeueCB()`
`OOCStream.QueueCallback<T>`
`dequeueTask()`
Synchronized read and delete from the top of the FIFO queue.
`void`
`enqueue(OOCStream.QueueCallback<T> cb)`
`void`
`enqueue(T t)`
`void`
`enqueueTask(OOCStream.QueueCallback<T> t)`
Synchronized insert of a new task to the end of the FIFO queue.
`CacheableData<?>`
`getData()`
`DataCharacteristics`
`getDataCharacteristics()`
`OOCStream<T>`
`getReadStream()`
`CachingStream`
`getStreamCache()`
`OOCStream<T>`
`getWriteStream()`
`boolean`
`hasStreamCache()`
`void`
`messageDownstream(OOCStreamMessage msg)`
`void`
`messageUpstream(OOCStreamMessage msg)`
`void`
`propagateFailure(DMLRuntimeException re)`
`void`
`setData(CacheableData<?> data)`
`void`
`setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`void`
`setIXTransform(BiFunction<Boolean,IndexRange,IndexRange> transform)`
`void`
`setSubscriber(Consumer<OOCStream.QueueCallback<T>> subscriber)`
Registers a new subscriber that consumes the stream.
`void`
`setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`String`
`toString()`

### Methods inherited from class org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue

`hasNext, isProcessed`

### Methods inherited from class java.lang.Object

`equals, getClass, hashCode, notify, notifyAll, wait, wait, wait`

### Methods inherited from interface org.apache.sysds.runtime.instructions.ooc.OOCStreamable

`isProcessed`

## Constructor Details

### SubscribableTaskQueue

public SubscribableTaskQueue()

## Method Details

### enqueue

public void enqueue(T t)

Specified by:
  `enqueue` in interface `OOCStream<T>`

### enqueue

public void enqueue(OOCStream.QueueCallback<T> cb)

Specified by:
  `enqueue` in interface `OOCStream<T>`

### enqueueTask

public void enqueueTask(OOCStream.QueueCallback<T> t)
Description copied from class: `LocalTaskQueue`
Synchronized insert of a new task to the end of the FIFO queue.

Overrides:
  `enqueueTask` in class `LocalTaskQueue<OOCStream.QueueCallback<T>>`
Parameters:
  `t` - task

### dequeue

public T dequeue()

Specified by:
  `dequeue` in interface `OOCStream<T>`

### dequeueCB

public OOCStream.QueueCallback<T> dequeueCB()

Specified by:
  `dequeueCB` in interface `OOCStream<T>`

### dequeueTask

public OOCStream.QueueCallback<T> dequeueTask()
Description copied from class: `LocalTaskQueue`
Synchronized read and delete from the top of the FIFO queue.

Overrides:
  `dequeueTask` in class `LocalTaskQueue<OOCStream.QueueCallback<T>>`
Returns:
  task

### closeInput

public void closeInput()
Description copied from class: `LocalTaskQueue`
Synchronized (logical) insert of a NO_MORE_TASKS symbol at the end of the FIFO queue in order to mark that no more tasks will be inserted into the queue.

Specified by:
  `closeInput` in interface `OOCStream<T>`
Overrides:
  `closeInput` in class `LocalTaskQueue<OOCStream.QueueCallback<T>>`

### setSubscriber

public void setSubscriber(Consumer<OOCStream.QueueCallback<T>> subscriber)
Description copied from interface: `OOCStream`
Registers a new subscriber that consumes the stream. While there is no guarantee for any specific order, the closing item LocalTaskQueue.NO_MORE_TASKS is guaranteed to be invoked after every other item has finished processing. Thus, the NO_MORE_TASKS callback can be used to free dependent resources and close output streams.

Specified by:
  `setSubscriber` in interface `OOCStream<T>`

### propagateFailure

public void propagateFailure(DMLRuntimeException re)

Specified by:
  `propagateFailure` in interface `OOCStream<T>`
Overrides:
  `propagateFailure` in class `LocalTaskQueue<OOCStream.QueueCallback<T>>`

### getReadStream

public OOCStream<T> getReadStream()

Specified by:
  `getReadStream` in interface `OOCStreamable<T>`

### getWriteStream

public OOCStream<T> getWriteStream()

Specified by:
  `getWriteStream` in interface `OOCStreamable<T>`

### messageUpstream

public void messageUpstream(OOCStreamMessage msg)

Specified by:
  `messageUpstream` in interface `OOCStreamable<T>`

### messageDownstream

public void messageDownstream(OOCStreamMessage msg)

Specified by:
  `messageDownstream` in interface `OOCStreamable<T>`

### hasStreamCache

public boolean hasStreamCache()

Specified by:
  `hasStreamCache` in interface `OOCStreamable<T>`

### getStreamCache

public CachingStream getStreamCache()

Specified by:
  `getStreamCache` in interface `OOCStreamable<T>`

### getDataCharacteristics

public DataCharacteristics getDataCharacteristics()

Specified by:
  `getDataCharacteristics` in interface `OOCStreamable<T>`

### getData

public CacheableData<?> getData()

Specified by:
  `getData` in interface `OOCStreamable<T>`

### setData

public void setData(CacheableData<?> data)

Specified by:
  `setData` in interface `OOCStreamable<T>`

### setUpstreamMessageRelay

public void setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)

Specified by:
  `setUpstreamMessageRelay` in interface `OOCStreamable<T>`

### setDownstreamMessageRelay

public void setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)

Specified by:
  `setDownstreamMessageRelay` in interface `OOCStreamable<T>`

### addUpstreamMessageRelay

public void addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)

Specified by:
  `addUpstreamMessageRelay` in interface `OOCStreamable<T>`

### addDownstreamMessageRelay

public void addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)

Specified by:
  `addDownstreamMessageRelay` in interface `OOCStreamable<T>`

### clearUpstreamMessageRelays

public void clearUpstreamMessageRelays()

Specified by:
  `clearUpstreamMessageRelays` in interface `OOCStreamable<T>`

### clearDownstreamMessageRelays

public void clearDownstreamMessageRelays()

Specified by:
  `clearDownstreamMessageRelays` in interface `OOCStreamable<T>`

### setIXTransform

public void setIXTransform(BiFunction<Boolean,IndexRange,IndexRange> transform)

Specified by:
  `setIXTransform` in interface `OOCStreamable<T>`

### toString

public String toString()

Overrides:
  `toString` in class `LocalTaskQueue<OOCStream.QueueCallback<T>>`