org.apache.flink.streaming.runtime.tasks.mailbox

Class TaskMailboxImpl

java.lang.Object

org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl

All Implemented Interfaces:

TaskMailbox

@ThreadSafe
public class TaskMailboxImpl
extends Object
implements TaskMailbox

Implementation of TaskMailbox in a BlockingQueue fashion and tailored towards our use case with multiple writers and single reader.

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailbox

TaskMailbox.MailboxClosedException, TaskMailbox.State

Field Summary

Fields inherited from interface org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailbox

MAX_PRIORITY, MIN_PRIORITY

Constructor Summary

Constructors

Constructor and Description
TaskMailboxImpl()
TaskMailboxImpl(Thread taskMailboxThread)

Method Summary

Modifier and Type	Method and Description
List<Mail>	close() Close the mailbox.
boolean	createBatch() Creates a batch of mails that can be taken with TaskMailbox.tryTakeFromBatch().
List<Mail>	drain() Drains the mailbox and returns all mails that were still enqueued.
TaskMailbox.State	getState() Returns the current state of the mailbox as defined by the lifecycle enum TaskMailbox.State.
boolean	hasMail() Returns true if the mailbox contains mail.
boolean	isMailboxThread() Check if the current thread is the mailbox thread.
void	put(Mail mail) Enqueues the given mail to the mailbox and blocks until there is capacity for a successful put.
void	putFirst(Mail mail) Adds the given action to the head of the mailbox.
void	quiesce() Quiesce the mailbox.
void	runExclusively(Runnable runnable) Runs the given code exclusively on this mailbox.
int	size() Returns the current number of mails in this mailbox.
Mail	take(int priority) This method returns the oldest mail from the mailbox (head of queue) or blocks until a mail is available.
Optional<Mail>	tryTake(int priority) Returns an optional with either the oldest mail from the mailbox (head of queue) if the mailbox is not empty or an empty optional otherwise.
Optional<Mail>	tryTakeFromBatch() Returns an optional with either the oldest mail from the batch (head of queue) if the batch is not empty or an empty optional otherwise.

Methods inherited from class java.lang.Object

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

Constructor Detail

TaskMailboxImpl

public TaskMailboxImpl(@Nonnull
                       Thread taskMailboxThread)

TaskMailboxImpl

@VisibleForTesting
public TaskMailboxImpl()

Method Detail

isMailboxThread

public boolean isMailboxThread()

Description copied from interface: TaskMailbox

Check if the current thread is the mailbox thread.

Read operations will fail if they are called from another thread.

Specified by:

isMailboxThread in interface TaskMailbox

Returns:

only true if called from the mailbox thread.

hasMail

public boolean hasMail()

Description copied from interface: TaskMailbox

Returns true if the mailbox contains mail.

Must be called from the mailbox thread (TaskMailbox.isMailboxThread().

Specified by:

hasMail in interface TaskMailbox

size

public int size()

Description copied from interface: TaskMailbox

Returns the current number of mails in this mailbox. (This includes mails in the batch not processed yet.)

Specified by:

size in interface TaskMailbox

Returns:

number of mails in the mailbox.

tryTake

public Optional<Mail> tryTake(int priority)

Description copied from interface: TaskMailbox

Returns an optional with either the oldest mail from the mailbox (head of queue) if the mailbox is not empty or an empty optional otherwise.

Must be called from the mailbox thread (TaskMailbox.isMailboxThread().

Specified by:

tryTake in interface TaskMailbox

Returns:

an optional with either the oldest mail from the mailbox (head of queue) if the mailbox is not empty or an empty optional otherwise.

take

@Nonnull
public Mail take(int priority)
                   throws InterruptedException,
                          IllegalStateException

Description copied from interface: TaskMailbox

This method returns the oldest mail from the mailbox (head of queue) or blocks until a mail is available.

Must be called from the mailbox thread (TaskMailbox.isMailboxThread().

Specified by:

take in interface TaskMailbox

Returns:

the oldest mail from the mailbox (head of queue).

Throws:

InterruptedException - on interruption.

IllegalStateException - if mailbox is already closed.

createBatch

public boolean createBatch()

Description copied from interface: TaskMailbox

Creates a batch of mails that can be taken with TaskMailbox.tryTakeFromBatch(). The batch does not affect TaskMailbox.tryTake(int) and TaskMailbox.take(int); that is, they return the same mails even if no batch had been created.

The default batch is empty. Thus, this method must be invoked once before TaskMailbox.tryTakeFromBatch().

If a batch is not completely consumed by TaskMailbox.tryTakeFromBatch(), its elements are carried over to the new batch.

Must be called from the mailbox thread (TaskMailbox.isMailboxThread().

Specified by:

createBatch in interface TaskMailbox

Returns:

true if there is at least one element in the batch; that is, if there is any mail at all at the time of the invocation.

tryTakeFromBatch

public Optional<Mail> tryTakeFromBatch()

Description copied from interface: TaskMailbox

Returns an optional with either the oldest mail from the batch (head of queue) if the batch is not empty or an empty optional otherwise.

Must be called from the mailbox thread (TaskMailbox.isMailboxThread().

Note that there is no blocking takeFromBatch as batches can only be created and consumed from the mailbox thread.

Specified by:

tryTakeFromBatch in interface TaskMailbox

Returns:

an optional with either the oldest mail from the batch (head of queue) if the batch is not empty or an empty optional otherwise.

put

public void put(@Nonnull
                Mail mail)

Description copied from interface: TaskMailbox

Enqueues the given mail to the mailbox and blocks until there is capacity for a successful put.

Mails can be added from any thread.

Specified by:

put in interface TaskMailbox

Parameters:

mail - the mail to enqueue.

putFirst

public void putFirst(@Nonnull
                     Mail mail)

Description copied from interface: TaskMailbox

Adds the given action to the head of the mailbox.

Mails can be added from any thread.

Specified by:

putFirst in interface TaskMailbox

Parameters:

mail - the mail to enqueue.

drain

public List<Mail> drain()

Description copied from interface: TaskMailbox

Drains the mailbox and returns all mails that were still enqueued.

Specified by:

drain in interface TaskMailbox

Returns:

list with all mails that where enqueued in the mailbox.

quiesce

public void quiesce()

Description copied from interface: TaskMailbox

Quiesce the mailbox. In this state, the mailbox supports only take operations and all pending and future put operations will throw TaskMailbox.MailboxClosedException.

Specified by:

quiesce in interface TaskMailbox

close

@Nonnull
public List<Mail> close()

Description copied from interface: TaskMailbox

Close the mailbox. In this state, all pending and future put operations and all pending and future take operations will throw TaskMailbox.MailboxClosedException. Returns all mails that were still enqueued.

Specified by:

close in interface TaskMailbox

Returns:

list with all mails that where enqueued in the mailbox at the time of closing.

getState

@Nonnull
public TaskMailbox.State getState()

Description copied from interface: TaskMailbox

Returns the current state of the mailbox as defined by the lifecycle enum TaskMailbox.State.

Specified by:

getState in interface TaskMailbox

Returns:

the current state of the mailbox.

runExclusively

public void runExclusively(Runnable runnable)

Description copied from interface: TaskMailbox

Runs the given code exclusively on this mailbox. No synchronized operations can be run concurrently to the given runnable (e.g., TaskMailbox.put(Mail) or modifying lifecycle methods).

Use this methods when you want to atomically execute code that uses different methods (e.g., check for state and then put message if open).

Specified by:

runExclusively in interface TaskMailbox

Parameters:

runnable - the runnable to execute