org.corfudb.runtime.view.stream

Class AbstractQueuedStreamView

java.lang.Object

org.corfudb.runtime.view.stream.AbstractContextStreamView<org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext>

org.corfudb.runtime.view.stream.AbstractQueuedStreamView

All Implemented Interfaces:

AutoCloseable, Iterator<ILogData>, IStreamView

Direct Known Subclasses:

AddressMapStreamView, BackpointerStreamView

public abstract class AbstractQueuedStreamView
extends AbstractContextStreamView<org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext>

The abstract queued stream view implements a stream backed by a read queue.

A read queue is a priority queue where addresses can be inserted, and are dequeued in ascending order. Subclasses implement the fillReadQueue() function, which defines how the read queue should be filled, and the read() function, which reads an entry and updates the pointers for the stream view.

The addresses in the read queue must be global addresses.

This implementation does not handle bulk reads and depends on IStreamView's implementation of remainingUpTo(), which simply calls nextUpTo() under a lock until it returns null.

Created by mwei on 1/6/17.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	AbstractQueuedStreamView.BackpointerOp

Nested classes/interfaces inherited from interface org.corfudb.runtime.view.stream.IStreamView

IStreamView.SearchDirection

Constructor Summary

Constructors

Constructor and Description
AbstractQueuedStreamView(CorfuRuntime runtime, UUID streamId) Create a new queued stream view.

Method Summary

Modifier and Type	Method and Description
protected void	addToResolvedQueue(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context, long globalAddress, ILogData ld) Add the given address to the resolved queue of the given context.
long	append(Object object, Function<TokenResponse,Boolean> acquisitionCallback, Function<TokenResponse,Boolean> deacquisitionCallback) Append an object to the stream, returning the global address it was written at.
void	close()
ILogData	current() Retrieve the current entry in the stream, which was the entry previously returned by a call to next() or previous().
protected abstract boolean	discoverAddressSpace(UUID streamId, NavigableSet<Long> queue, long startAddress, long stopAddress, Function<ILogData,AbstractQueuedStreamView.BackpointerOp> filter, boolean checkpoint, long maxGlobal) Defines the strategy to discover addresses belonging to this stream.
protected boolean	fillFromResolved(long maxGlobal, org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context)
protected boolean	fillReadQueue(long maxGlobal, org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context) Fill the read queue for the current context.
long	find(long globalAddress, IStreamView.SearchDirection direction) Find the global address of the next entry in this stream, in the direction given.
void	gc(long trimMark) Garbage collect all the trimmed entries on this stream
long	getCurrentGlobalPosition() Get the current position of the pointer in this stream (global address).
boolean	getHasNext(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context) Return whether calling getNextEntry() may return more entries, given the context.
protected List<ILogData>	getNextEntries(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context, long maxGlobal, Function<ILogData,Boolean> contextCheckFn) Retrieve the next entries in the stream, given the context.
protected ILogData	getNextEntry(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context, long maxGlobal) Retrieve the next entry in the stream, given the context.
ILogData	previous() Retrieve the previous entry in the stream.
protected ILogData	read(long address)
protected ILogData	read(long address, long readStartTime) Reads data from an address in the address space.
protected List<ILogData>	readAll(List<Long> addresses)
protected ILogData	readRange(long address, List<Long> addresses)
protected abstract ILogData	removeFromQueue(NavigableSet<Long> queue) Remove next entry from the queue.
protected AbstractQueuedStreamView.BackpointerOp	resolveCheckpoint(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context, ILogData data, long maxGlobal)

Methods inherited from class org.corfudb.runtime.view.stream.AbstractContextStreamView

doesEntryUpdateContext, getCurrentContext, hasNext, nextUpTo, popContext, processEntryForContext, pushNewContext, remainingUpTo, reset, seek, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.corfudb.runtime.view.stream.IStreamView

append, getId, getTotalUpdates, next, remaining, spliterator, spliteratorUpTo, stream, streamUpTo

Methods inherited from interface java.util.Iterator

forEachRemaining, remove

Constructor Detail

AbstractQueuedStreamView

public AbstractQueuedStreamView(CorfuRuntime runtime,
                                UUID streamId)

Create a new queued stream view.

Parameters:

streamId - The ID of the stream

runtime - The runtime used to create this view.

Method Detail

addToResolvedQueue

protected void addToResolvedQueue(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context,
                                  long globalAddress,
                                  ILogData ld)

Add the given address to the resolved queue of the given context.

Parameters:

context - The context to add the address to

globalAddress - The resolved global address.

getNextEntry

protected ILogData getNextEntry(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context,
                                long maxGlobal)

Retrieve the next entry in the stream, given the context.

Specified by:

getNextEntry in class AbstractContextStreamView<org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext>

Parameters:

context - The context to retrieve the next entry from.

maxGlobal - The maximum global address to read to.

Returns:

Next ILogData for this context

removeFromQueue

protected abstract ILogData removeFromQueue(NavigableSet<Long> queue)

Remove next entry from the queue.

Parameters:

queue - queue of entries.

Returns:

next available entry. or null if there are no more entries or remaining entries are not part of this stream.

gc

public void gc(long trimMark)

Description copied from interface: IStreamView

Garbage collect all the trimmed entries on this stream

Parameters:

trimMark - start of the active log

append

public long append(Object object,
                   Function<TokenResponse,Boolean> acquisitionCallback,
                   Function<TokenResponse,Boolean> deacquisitionCallback)

Append an object to the stream, returning the global address it was written at.

Optionally, provide a method to be called when an address is acquired, and also a method to be called when an address is released (due to an unsuccessful append).

We loop forever trying to write, and automatically retrying if we get overwritten (hole filled).

Parameters:

object - The object to append.

acquisitionCallback - A function to call when an address is acquired. It should return true to continue with the append.

deacquisitionCallback - A function to call when an address is released. It should return true to retry writing.

Returns:

The (global) address the object was written at.

read

protected ILogData read(long address,
                        long readStartTime)

Reads data from an address in the address space. It will give the writer a chance to complete based on the time when the reads of which this individual read is a step started. If the reads have been going on for longer than the grace period given for a writer to complete a write, the subsequent individual read calls will immediately fill the hole on absence of data at the given address.

Parameters:

address - address to read.

readStartTime - start time of the range of reads.

Returns:

log data at the address.

readAll

@Nonnull
protected List<ILogData> readAll(@Nonnull
                                          List<Long> addresses)

getNextEntries

protected List<ILogData> getNextEntries(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context,
                                        long maxGlobal,
                                        Function<ILogData,Boolean> contextCheckFn)

Retrieve the next entries in the stream, given the context.

This function is designed to implement a bulk read. In a bulk read, one of the entries may cause the context to change - the implementation should check if the entry changes the context and stop reading if this occurs, returning the entry that caused contextCheckFn to return true.

The default implementation simply calls getNextEntry.

In the queued implementation, we just read all entries in the read queue in parallel. If there is any entry which changes the context, we cut the list off there.

Overrides:

getNextEntries in class AbstractContextStreamView<org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext>

Parameters:

context - The context to retrieve the next entry from.

maxGlobal - The maximum global address to read to.

contextCheckFn - A function which returns true if the entry changes the stream context.

Returns:

A list of the next entries for this context

fillReadQueue

protected boolean fillReadQueue(long maxGlobal,
                                org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context)

Fill the read queue for the current context. This method is called whenever a client requests a read, but there are no addresses left in the read queue.

This method returns true if entries were added to the read queue, false otherwise.

Parameters:

maxGlobal - The maximum global address to read to.

context - The current stream context.

Returns:

True, if entries were added to the read queue, False, otherwise.

discoverAddressSpace

protected abstract boolean discoverAddressSpace(UUID streamId,
                                                NavigableSet<Long> queue,
                                                long startAddress,
                                                long stopAddress,
                                                Function<ILogData,AbstractQueuedStreamView.BackpointerOp> filter,
                                                boolean checkpoint,
                                                long maxGlobal)

Defines the strategy to discover addresses belonging to this stream. We currently support two mechanisms: - Following backpointers (@see org.corfudb.runtime.view.stream.BackpointerStreamView) - Requesting the sequencer for the complete address map of a stream. (@see org.corfudb.runtime.view.stream.AddressMapStreamView)

Parameters:

streamId - stream unique identifier.

queue - queue to fill up.

startAddress - read start address (inclusive)

stopAddress - read stop address (exclusive)

filter - filter to apply to data

checkpoint - true if checkpoint discovery, false otherwise.

maxGlobal - max address to resolve discovery.

Returns:

true if addresses were discovered, false, otherwise.

fillFromResolved

protected boolean fillFromResolved(long maxGlobal,
                                   org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context)

read

protected ILogData read(long address)

readRange

@Nonnull
protected ILogData readRange(long address,
                                      @Nonnull
                                      List<Long> addresses)

getHasNext

public boolean getHasNext(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context)

Return whether calling getNextEntry() may return more entries, given the context.

We indicate we may have entries available if the read queue contains entries to read -or- if the next token is greater than our log pointer.

Specified by:

getHasNext in class AbstractContextStreamView<org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext>

Parameters:

context - The context to retrieve the next entry from.

Returns:

True, if getNextEntry() may return an entry. False otherwise.

close

public void close()

Specified by:

close in interface AutoCloseable

Overrides:

close in class AbstractContextStreamView<org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext>

find

public long find(long globalAddress,
                 IStreamView.SearchDirection direction)

Find the global address of the next entry in this stream, in the direction given.

Parameters:

globalAddress - The global address to start searching from.

direction - The direction to search.

Returns:

The global address of the next entry in the stream, or Address.NOT_FOUND if no entry was found.

resolveCheckpoint

protected AbstractQueuedStreamView.BackpointerOp resolveCheckpoint(org.corfudb.runtime.view.stream.AbstractQueuedStreamView.QueuedStreamContext context,
                                                                   ILogData data,
                                                                   long maxGlobal)

previous

public ILogData previous()

Retrieve the previous entry in the stream. If there are no previous entries, NULL will be returned.

Returns:

The previous entry in the stream, or NULL, if no entries are available.

current

public ILogData current()

Retrieve the current entry in the stream, which was the entry previously returned by a call to next() or previous(). If the stream was never read from, NULL will be returned.

Returns:

The current entry in the stream.

getCurrentGlobalPosition

public long getCurrentGlobalPosition()

Get the current position of the pointer in this stream (global address).

Returns:

The position of the pointer in this stream (global address), or Address.NON_ADDRESS.