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

# Class CachingStream

java.lang.Object
org.apache.sysds.runtime.instructions.ooc.CachingStream

All Implemented Interfaces:
  `OOCStreamable<IndexedMatrixValue>`

public class CachingStream extends Object implements OOCStreamable<IndexedMatrixValue>
A wrapper around LocalTaskQueue to consume the source stream and reset to consume again for other operators.

## Field Summary

Fields
Modifier and Type
Field
Description
`static final IDSequence`
`_streamSeq`

## Constructor Summary

Constructors
Constructor
Description
`CachingStream(OOCStream<IndexedMatrixValue> source)`
`CachingStream(OOCStream<IndexedMatrixValue> source, long streamId)`

## Method Summary

Modifier and Type
Method
Description
`void`
`activateIndexing()`
`void`
`addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`void`
`addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`void`
`clearDownstreamMessageRelays()`
`void`
`clearUpstreamMessageRelays()`
`OOCStream.QueueCallback<IndexedMatrixValue>`
`findCached(MatrixIndexes idx)`
`void`
`findCachedAsync(MatrixIndexes idx, Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> callback)`
`int`
`findCachedIndex(MatrixIndexes idx)`
`CompletableFuture<OOCStream.QueueCallback<IndexedMatrixValue>>`
`get(int idx)`
`CacheableData<?>`
`getData()`
`DataCharacteristics`
`getDataCharacteristics()`
`OOCStream<IndexedMatrixValue>`
`getReadStream()`
`CachingStream`
`getStreamCache()`
`OOCStream<IndexedMatrixValue>`
`getWriteStream()`
`boolean`
`hasStreamCache()`
`void`
`incrProcessingCount(int i, int count)`
Artificially increase the processing count of a block.
`void`
`incrProcessingCount(MatrixIndexes idx, int count)`
`void`
`incrSubscriberCount(int count)`
Artificially increase subscriber count.
`boolean`
`isProcessed()`
`void`
`messageDownstream(OOCStreamMessage msg)`
`void`
`messageUpstream(OOCStreamMessage msg)`
`OOCStream.QueueCallback<IndexedMatrixValue>`
`peekCached(MatrixIndexes idx)`
Finds a cached item without counting it as a consumption.
`void`
`peekCachedAsync(MatrixIndexes idx, Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> callback)`
Finds a cached item asynchronously without counting it as a consumption.
`BlockKey`
`peekCachedBlockKey(MatrixIndexes idx)`
`void`
`scheduleDeletion()`
`void`
`setData(CacheableData<?> data)`
`void`
`setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`void`
`setIXTransform(BiFunction<Boolean,IndexRange,IndexRange> transform)`
`void`
`setSubscriber(Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> subscriber, boolean incrConsumers)`
`void`
`setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)`
`String`
`toString()`

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

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

## Field Details

### _streamSeq

public static final IDSequence _streamSeq

## Constructor Details

### CachingStream

public CachingStream(OOCStream<IndexedMatrixValue> source)

### CachingStream

public CachingStream(OOCStream<IndexedMatrixValue> source, long streamId)

## Method Details

### scheduleDeletion

public void scheduleDeletion()

### toString

public String toString()

Overrides:
  `toString` in class `Object`

### get

public CompletableFuture<OOCStream.QueueCallback<IndexedMatrixValue>> get(int idx) throws InterruptedException, ExecutionException

Throws:
  `InterruptedException`
  `ExecutionException`

### findCachedIndex

public int findCachedIndex(MatrixIndexes idx)

### peekCachedBlockKey

public BlockKey peekCachedBlockKey(MatrixIndexes idx)

### findCached

public OOCStream.QueueCallback<IndexedMatrixValue> findCached(MatrixIndexes idx)

### findCachedAsync

public void findCachedAsync(MatrixIndexes idx, Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> callback)

### peekCachedAsync

public void peekCachedAsync(MatrixIndexes idx, Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> callback)
Finds a cached item asynchronously without counting it as a consumption.

### peekCached

public OOCStream.QueueCallback<IndexedMatrixValue> peekCached(MatrixIndexes idx)
Finds a cached item without counting it as a consumption.

### activateIndexing

public void activateIndexing()

### getReadStream

public OOCStream<IndexedMatrixValue> getReadStream()

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

### getWriteStream

public OOCStream<IndexedMatrixValue> getWriteStream()

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

### hasStreamCache

public boolean hasStreamCache()

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

### getStreamCache

public CachingStream getStreamCache()

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

### isProcessed

public boolean isProcessed()

Specified by:
  `isProcessed` in interface `OOCStreamable<IndexedMatrixValue>`

### getDataCharacteristics

public DataCharacteristics getDataCharacteristics()

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

### getData

public CacheableData<?> getData()

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

### setData

public void setData(CacheableData<?> data)

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

### messageUpstream

public void messageUpstream(OOCStreamMessage msg)

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

### messageDownstream

public void messageDownstream(OOCStreamMessage msg)

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

### setUpstreamMessageRelay

public void setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)

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

### setDownstreamMessageRelay

public void setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)

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

### addUpstreamMessageRelay

public void addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay)

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

### addDownstreamMessageRelay

public void addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay)

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

### clearUpstreamMessageRelays

public void clearUpstreamMessageRelays()

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

### clearDownstreamMessageRelays

public void clearDownstreamMessageRelays()

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

### setIXTransform

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

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

### setSubscriber

public void setSubscriber(Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> subscriber, boolean incrConsumers)

### incrSubscriberCount

public void incrSubscriberCount(int count)
Artificially increase subscriber count. Only use if certain blocks are accessed more than once.

### incrProcessingCount

public void incrProcessingCount(int i, int count)
Artificially increase the processing count of a block.

### incrProcessingCount

public void incrProcessingCount(MatrixIndexes idx, int count)