com.bigdata.ha.pipeline

Class HASendService

java.lang.Object

com.bigdata.ha.pipeline.HASendService

public class HASendService
extends Object

A service for sending raw ByteBuffers across a socket. This service supports the HA write pipeline. This service is designed to be paired with an HAReceiveService, which typically is running on a different host. The HASendService provides only an efficient raw data transfer. The HA write pipeline coordinates the transfer of data using RMI messages which tell the receiver how much data to expect, the checksum of the data, etc.

Implementation

This class has a private single-threaded Executor to which it submits a SendTask. The SendTask will open a blocking-mode SocketChannel to the service at the configured InetSocketAddress and send the bytes remaining in a ByteBuffer to that service on that SocketChannel. The data will be sent on the socket using WritableByteChannel.write(ByteBuffer), which is optimized for the NIO transfer of direct ByteBuffers. Since this is a blocking-mode connection, the write operation will block until all bytes have been sent or the current thread is interrupted, e.g., by cancelling its Future.

The SendTask verifies that all bytes were sent as a post-condition (position() == limit). If there is any problem, then the SocketChannel is closed and the original exception is thrown out of SendTask#call(). If the socket is closed from the other end while we are still writing data, then that exception gets thrown out as well. The socket connection is closed as a post-condition (it does not survive across multiple sends). Closing the socket connection after each success or unsuccessful send gives us a strong indication of success or failure for the data transfer which is independent of the RMI message and makes it trivial to re-synchronize the HASendService since it is basically stateless.

Note: This class exposes its synchronization mechanism to HAReceiveService.

Author:

Martyn Cutcher, Bryan Thompson

See Also:

HAReceiveService

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	HASendService.IncSendTask This task implements the raw data transfer.

Constructor Summary

Constructors

Constructor and Description
HASendService() Designated constructor (flyweight initialization).

Method Summary

Methods

Modifier and Type	Method and Description
protected void	finalize() Extended to ensure that the private executor service is always terminated.
InetSocketAddress	getAddrNext() Return the current address to which this service will send data.
protected Callable<Void>	newIncSendTask(ByteBuffer buffer, byte[] marker) Factory for the SendTask.
Future<Void>	send(ByteBuffer buffer, byte[] marker) Send the bytes Buffer.remaining() in the buffer to the configured InetSocketAddress.
void	start(InetSocketAddress addrNext) Starts a thread which will transfer data to a service listening at the specified InetSocketAddress.
void	terminate() Immediate shutdown.
String	toString()

Methods inherited from class java.lang.Object

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

Constructor Detail

HASendService

public HASendService()

Designated constructor (flyweight initialization).

See Also:

start(InetSocketAddress)

Method Detail

toString

public String toString()

Overrides:

toString in class Object

getAddrNext

public InetSocketAddress getAddrNext()

Return the current address to which this service will send data.

Returns:

The current address -or- null if no address is set at this time.

See Also:

start(InetSocketAddress)

finalize

protected void finalize()
                 throws Throwable

Extended to ensure that the private executor service is always terminated.

Overrides:

finalize in class Object

Throws:

Throwable

start

public void start(InetSocketAddress addrNext)

Starts a thread which will transfer data to a service listening at the specified InetSocketAddress. A SocketChannel will be opened to the specified the connection to the socket specified in the constructor and start the thread pool on which the payloads will be send.

Note: This class exposes its synchronization mechanism to HAReceiveService.

Parameters:

addrNext - The Internet socket address of the downstream service.

Throws:

IllegalArgumentException - if the address is null.

IllegalStateException - if this service is already running.

See Also:

terminate()

terminate

public void terminate()

Immediate shutdown. Any transfer in process will be interrupted. It is safe to invoke this method whether or not the service is running.

Note: This class exposes its synchronization mechanism to HAReceiveService.

send

public Future<Void> send(ByteBuffer buffer,
                byte[] marker)
                  throws ImmediateDownstreamReplicationException,
                         InterruptedException

Send the bytes Buffer.remaining() in the buffer to the configured InetSocketAddress.

Note: This operation DOES NOT have a side effect on the position, limit or mark for the buffer.

Note: In order to use efficient NIO operations this MUST be a direct ByteBuffer.

Parameters:

buffer - The buffer.

marker - A marker that will be used to prefix the payload for the message in the write replication socket stream. The marker is used to ensure synchronization when reading on the stream.

Returns:

The Future which can be used to await the outcome of this operation.

Throws:

InterruptedException

ImmediateDownstreamReplicationException

IllegalArgumentException - if the buffer is null.

IllegalArgumentException - if the buffer is empty (no bytes remaining).

RejectedExecutionException - if this service has been shutdown.

newIncSendTask

protected Callable<Void> newIncSendTask(ByteBuffer buffer,
                            byte[] marker)

Factory for the SendTask.

Parameters:

buffer - The buffer whose data are to be sent.

marker - A marker that will be used to prefix the payload for the message in the write replication socket stream. The marker is used to ensure synchronization when reading on the stream.

Returns:

The task which will send the data to the configured InetSocketAddress.