com.bigdata.ha

Class QuorumServiceBase<S extends HAGlue,L extends AbstractJournal>

java.lang.Object

com.bigdata.quorum.AbstractQuorumClient<S>

com.bigdata.quorum.AbstractQuorumMember<S>

com.bigdata.ha.QuorumServiceBase<S,L>

All Implemented Interfaces:

QuorumCommit<S>, QuorumPipeline<S>, QuorumRead<S>, QuorumService<S>, QuorumClient<S>, QuorumListener, QuorumMember<S>, QuorumStateChangeListener, ServiceLookup<S>

public abstract class QuorumServiceBase<S extends HAGlue,L extends AbstractJournal>
extends AbstractQuorumMember<S>
implements QuorumService<S>

Abstract implementation provides the logic for distributing messages for the quorum 2-phase commit protocol, failover reads, etc.

Version:

$Id$

Author:

Bryan Thompson

Field Summary

Fields

Modifier and Type	Field and Description
protected static org.apache.log4j.Logger	log

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	QuorumServiceBase(String logicalServiceId, UUID serviceId, S service, L localService)

Method Summary

Methods

Modifier and Type	Method and Description
void	abort2Phase(long token) Used by the leader to send a message to each service joined with the quorum telling it to discard its write set, reloading all state from the last root block.
static <F extends Future<T>,T> void	cancelFutures(List<F> futures) Cancel the requests on the remote services (RMI).
CommitResponse	commit2Phase(CommitRequest req) Used by the leader to send a message to each joined service in the quorum telling it to commit using the root block from the corresponding prepare message.
ExecutorService	getExecutor() An ExecutorService which may be used by the QuorumMember to run various asynchronous tasks.
long	getLastCommitCounter() Return the lastCommitCounter for this service (based on its current root block).
long	getLastCommitTime() Return the lastCommitTime for this service (based on its current root block).
protected L	getLocalService() FIXME Return the IResourceManager, Journal, [@link DataService}, etc.
long	getPrepareTimeout() Return the configured timeout in milliseconds that the leader will await the other services to prepare for a 2-phase commit.
protected abstract long	getRetrySendTimeoutNanos()
S	getService() Return the local service implementation object (NOT the RMI proxy for that object).
abstract S	getService(UUID serviceId) Return the remote interface used to perform HA operations on a member of quorum.
UUID	getStoreUUID() The UUID of the backing store.
protected abstract void	handleReplicatedWrite(IHASyncRequest req, IHAWriteMessage msg, ByteBuffer data) Core implementation handles the message and payload when received on a service.
protected abstract void	incReceive(IHASyncRequest req, IHAWriteMessage msg, int nreads, int rdlen, int rem) Core implementation of callback for monitoring progress of replicated writes.
void	logRootBlock(IRootBlockView rootBlock) Log the root block for the commit point that closes the current write set onto the HALogWriter.
void	logWriteCacheBlock(IHAWriteMessage msg, ByteBuffer data) Log the IHAWriteMessage and the associated data (if necessary).
PrepareResponse	prepare2Phase(PrepareRequest req) Used by the leader to send a message to each joined service in the quorum instructing it to flush all writes to the backing channel, and acknowledge "yes" if ready to commit.
protected void	processEvents() Called from ErrorTask in HAJournalServer to ensure that events are processed before entering SeekConsensus.
void	purgeHALogs(long token) Purge the local HA log files.
byte[]	readFromQuorum(UUID storeId, long addr) Used by any service joined with the quorum to read a record from another service joined with the quorum in order to work around a "bad read" as identified by a checksum error on the local service.
Future<Void>	receiveAndReplicate(IHASyncRequest req, IHASendState snd, IHAWriteMessage msg) Return a Future for a task which will replicate an NIO buffer along the write pipeline.
Future<Void>	replicate(IHASyncRequest req, IHAWriteMessage msg, ByteBuffer b) Return a Future for a task which will replicate an NIO buffer along the write pipeline.
Future<IHAPipelineResetResponse>	resetPipeline(IHAPipelineResetRequest req) Reset the pipeline (blocking).

Methods inherited from class com.bigdata.quorum.AbstractQuorumMember

addListener, assertFollower, assertLeader, assertQuorum, consensus, getActor, getDownstreamServiceId, getLeader, getServiceId, isFollower, isJoinedMember, isLastInChain, isLeader, isMember, isPipelineMember, lostConsensus, memberAdd, memberRemove, pipelineAdd, pipelineChange, pipelineElectedLeader, pipelineRemove, pipelineUpstreamChange, quorumBreak, quorumMeet, removeListener, serviceJoin, serviceLeave

Methods inherited from class com.bigdata.quorum.AbstractQuorumClient

disconnected, getLogicalServiceZPath, getQuorum, notify, start, terminate

Methods inherited from class java.lang.Object

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

Methods inherited from interface com.bigdata.ha.QuorumService

discardWriteSet, enterErrorState, getPID, getServiceDir, installRootBlocks

Methods inherited from interface com.bigdata.quorum.QuorumMember

assertLeader, getActor, getDownstreamServiceId, getServiceId, isFollower, isJoinedMember, isLastInChain, isLeader, isMember, isPipelineMember

Methods inherited from interface com.bigdata.quorum.QuorumClient

disconnected, getLeader, getLogicalServiceZPath, getQuorum, start, terminate

Methods inherited from interface com.bigdata.quorum.QuorumListener

notify

Methods inherited from interface com.bigdata.quorum.QuorumStateChangeListener

consensus, lostConsensus, memberAdd, memberRemove, pipelineAdd, pipelineChange, pipelineElectedLeader, pipelineRemove, pipelineUpstreamChange, quorumBreak, quorumMeet, serviceJoin, serviceLeave

Field Detail

log

protected static final transient org.apache.log4j.Logger log

Constructor Detail

QuorumServiceBase

protected QuorumServiceBase(String logicalServiceId,
                 UUID serviceId,
                 S service,
                 L localService)

Parameters:

logicalServiceId - The identifier of the logical service.

serviceId - The UUID for this service (a physical instance of the logical service).

service - The interface for the local service that is exposed to remote clients (typically as a smart proxy).

localService - The local service implementation.

Method Detail

getRetrySendTimeoutNanos

protected abstract long getRetrySendTimeoutNanos()

getService

public S getService()

Description copied from interface: QuorumMember

Return the local service implementation object (NOT the RMI proxy for that object).

A service operating as a QuorumMember has both Remote interface and a non-remote interface. These interfaces typically define similar methods (for example, compare HACommitGlue, which is Remote, with QuorumCommit, which is non-remote). The Remote interface in turn will have a local implementation object inside of the JVM and an exported proxy for that Remote interface. This method returns the local implementation object for the Remote interface and is intended to facilitate operations which the QuorumMember executes against its own Remote interface. While the Remote interface will be exposed to other services using a proxy, how that happens is outside of the scope of this method.

Specified by:

getService in interface QuorumMember<S extends HAGlue>

getService

public abstract S getService(UUID serviceId)

Return the remote interface used to perform HA operations on a member of quorum.

Specified by:

getService in interface QuorumClient<S extends HAGlue>

Specified by:

getService in interface ServiceLookup<S extends HAGlue>

Specified by:

getService in class AbstractQuorumMember<S extends HAGlue>

Parameters:

serviceId - The UUID associated with the service.

Returns:

The remote interface for that quorum member.

getLocalService

protected L getLocalService()

FIXME Return the IResourceManager, Journal, [@link DataService}, etc. Probably rename to getResourceManager().

getExecutor

public ExecutorService getExecutor()

Description copied from interface: QuorumMember

An ExecutorService which may be used by the QuorumMember to run various asynchronous tasks.

Specified by:

getExecutor in interface QuorumMember<S extends HAGlue>

receiveAndReplicate

public Future<Void> receiveAndReplicate(IHASyncRequest req,
                               IHASendState snd,
                               IHAWriteMessage msg)
                                 throws IOException

Description copied from interface: QuorumPipeline

Return a Future for a task which will replicate an NIO buffer along the write pipeline. This method is invoked for any node except the master, including the last node in the failover chain.

Specified by:

receiveAndReplicate in interface QuorumPipeline<S extends HAGlue>

Parameters:

req - A synchronization request (optional). This is only non-null when historical WriteCache blocks are being replayed down the write pipeline in order to synchronize a service.

snd - Metadata about the state of the sender and potentially the routing of the payload along the write replication pipeline.

msg - The RMI metadata about the payload.

Throws:

IOException

replicate

public Future<Void> replicate(IHASyncRequest req,
                     IHAWriteMessage msg,
                     ByteBuffer b)
                       throws IOException

Description copied from interface: QuorumPipeline

Return a Future for a task which will replicate an NIO buffer along the write pipeline. This method is only invoked by the quorum leader. The payload is replicated to the first follower in the write pipeline. That follower will accept the payload (and replicate it if necessary) using #receiveAndReplicate(IHAWriteMessage).

Note: The implementation of this method should be robust to changes in the write pipeline. Specifically, if a follower leaves the write pipeline, it should attempt to retransmit the message and the payload while allowing time for the write pipeline to be reconfigured in response to the related QuorumMember events.

Specified by:

replicate in interface QuorumPipeline<S extends HAGlue>

Parameters:

req - A synchronization request (optional). This is only non-null when historical WriteCache blocks are being replayed down the write pipeline in order to synchronize a service.

msg - The RMI metadata about the payload.

b - The payload. The bytes from the position to the limit will be transmitted (note that the #of bytes remaining in the buffer MUST agree with IHAWriteMessageBase.getSize()).

Throws:

IOException

resetPipeline

public Future<IHAPipelineResetResponse> resetPipeline(IHAPipelineResetRequest req)
                                               throws IOException

Description copied from interface: QuorumPipeline

Reset the pipeline (blocking). This message is used to handle an error in pipeline replication. If replication fails, the socket connections both upstream and downstream of the point of failure can be left in an indeterminate state with partially buffered data. In order to bring the pipeline back into a known state (without forcing a quorum break) we message each service in the pipeline to reset its HAReceiveService (including the inner HASendService). The next message and payload relayed from the leader will cause new socket connections to be established.

Specified by:

resetPipeline in interface QuorumPipeline<S extends HAGlue>

Returns:

The Future for the operation on the local service.

Throws:

IOException

handleReplicatedWrite

protected abstract void handleReplicatedWrite(IHASyncRequest req,
                         IHAWriteMessage msg,
                         ByteBuffer data)
                                       throws Exception

Core implementation handles the message and payload when received on a service.

Note: Replication of the message and payload is handled by the caller. The implementation of this method is NOT responsible for replication.

Parameters:

req - The synchronization request (optional). When non- null the message and payload are historical data. When null they are live data.

msg - Metadata about a buffer containing data replicated to this node.

data - The buffer containing the data.

Throws:

Exception

See Also:

QuorumPipelineImpl#handleReplicatedWrite(IHAWriteMessage, ByteBuffer)

incReceive

protected abstract void incReceive(IHASyncRequest req,
              IHAWriteMessage msg,
              int nreads,
              int rdlen,
              int rem)
                            throws Exception

Core implementation of callback for monitoring progress of replicated writes.

Parameters:

req - The synchronization request (optional). When non- null the message and payload are historical data. When null they are live data.

msg - Metadata about a buffer containing data replicated to this node.

rdlen - The number of bytes read from the socket in this read.

rem - The number of bytes remaining before the payload has been fully read.

Throws:

Exception

logWriteCacheBlock

public void logWriteCacheBlock(IHAWriteMessage msg,
                      ByteBuffer data)
                        throws IOException

Log the IHAWriteMessage and the associated data (if necessary). The log file for the current write set will be deleted if the quorum is fully met at the next 2-phase commit.

Note: The default implementation is a NOP.

Specified by:

logWriteCacheBlock in interface QuorumPipeline<S extends HAGlue>

Parameters:

msg - The IHAWriteMessage.

data - The WriteCache block.

Throws:

IOException

purgeHALogs

public void purgeHALogs(long token)

Purge the local HA log files. This should be invoked when the service goes through a commit point in which the quorum is fully met. At that moment, we no longer require these log files to resynchronize any service.

Note: The open HALog file is NOT removed by this method.

Note: The default implementation is a NOP.

Specified by:

purgeHALogs in interface QuorumPipeline<S extends HAGlue>

Parameters:

token - The quorum token. The quorum must remain fully met for this token across this operation. If that invariant is violated then the operation will terminate once this violation is noticed.

logRootBlock

public void logRootBlock(IRootBlockView rootBlock)
                  throws IOException

Log the root block for the commit point that closes the current write set onto the HALogWriter.

Note: This method is ONLY invoked as part of the 2-phase commit protocol. Therefore, this method ONLY applies to the live HALog file. A service is atomically either joined with the met quorum at a 2-phase commit point or not joined. The PREPARE and COMMIT messages are ONLY generated for services that were joined with the met quorum as of that atomic decision point in the commit protocol. Therefore, this method is never called for a service that was not joined as of that atomic decision point.

Note: The default implementation is a NOP.

Specified by:

logRootBlock in interface QuorumPipeline<S extends HAGlue>

Parameters:

rootBlock - The root block for the commit point that was just achieved.

Throws:

IOException

abort2Phase

public void abort2Phase(long token)
                 throws IOException,
                        InterruptedException

Description copied from interface: QuorumCommit

Used by the leader to send a message to each service joined with the quorum telling it to discard its write set, reloading all state from the last root block. If the node has not observed the corresponding "prepare" message then it should ignore this message.

Specified by:

abort2Phase in interface QuorumCommit<S extends HAGlue>

Parameters:

token - The quorum token.

Throws:

IOException

InterruptedException

commit2Phase

public CommitResponse commit2Phase(CommitRequest req)
                            throws IOException,
                                   InterruptedException

Description copied from interface: QuorumCommit

Used by the leader to send a message to each joined service in the quorum telling it to commit using the root block from the corresponding prepare message. The commit MAY NOT go forward unless both the current quorum token and the lastCommitTime on this message agree with the quorum token and lastCommitTime in the root block from the last "prepare" message.

Specified by:

commit2Phase in interface QuorumCommit<S extends HAGlue>

Throws:

IOException

InterruptedException

prepare2Phase

public PrepareResponse prepare2Phase(PrepareRequest req)
                              throws InterruptedException,
                                     TimeoutException,
                                     IOException

Description copied from interface: QuorumCommit

Used by the leader to send a message to each joined service in the quorum instructing it to flush all writes to the backing channel, and acknowledge "yes" if ready to commit. If the service can not prepare for any reason, then it must return "no". The service must save a copy of the root block for use with the next commit message.

Specified by:

prepare2Phase in interface QuorumCommit<S extends HAGlue>

Returns:

The outcome of the distributed PREPARE request, indicating whether each service is prepared to commit.

Throws:

InterruptedException

TimeoutException

IOException

getStoreUUID

public final UUID getStoreUUID()

Description copied from interface: QuorumPipeline

The UUID of the backing store.

Specified by:

getStoreUUID in interface QuorumPipeline<S extends HAGlue>

getLastCommitTime

public final long getLastCommitTime()

Description copied from interface: QuorumService

Return the lastCommitTime for this service (based on its current root block).

Specified by:

getLastCommitTime in interface QuorumPipeline<S extends HAGlue>

Specified by:

getLastCommitTime in interface QuorumService<S extends HAGlue>

getLastCommitCounter

public final long getLastCommitCounter()

Description copied from interface: QuorumService

Return the lastCommitCounter for this service (based on its current root block).

Specified by:

getLastCommitCounter in interface QuorumPipeline<S extends HAGlue>

Specified by:

getLastCommitCounter in interface QuorumService<S extends HAGlue>

getPrepareTimeout

public long getPrepareTimeout()

Description copied from interface: QuorumService

Return the configured timeout in milliseconds that the leader will await the other services to prepare for a 2-phase commit.

Specified by:

getPrepareTimeout in interface QuorumService<S extends HAGlue>

readFromQuorum

public byte[] readFromQuorum(UUID storeId,
                    long addr)
                      throws InterruptedException,
                             IOException

Description copied from interface: QuorumRead

Used by any service joined with the quorum to read a record from another service joined with the quorum in order to work around a "bad read" as identified by a checksum error on the local service.

Note: This is NOT the normal path for reading on a record from a service. This is used to handle bad reads (when a checksum or IO error is reported by the local disk) by reading the record from another member of the quorum.

Specified by:

readFromQuorum in interface QuorumRead<S extends HAGlue>

Parameters:

storeId - The UUID of the IRawStore from which the record should be read.

addr - The address of a record on that store.

Returns:

The record.

Throws:

InterruptedException

IOException

See Also:

HAGlue#readFromDisk(UUID, long)

processEvents

protected void processEvents()

Called from ErrorTask in HAJournalServer to ensure that events are processed before entering SeekConsensus.

See Also:

HAJournalServer reports "follower" but is in SeekConsensus and is not participating in commits

cancelFutures

public static <F extends Future<T>,T> void cancelFutures(List<F> futures)

Cancel the requests on the remote services (RMI). This is a best effort implementation. Any RMI related errors are trapped and ignored in order to be robust to failures in RMI when we try to cancel the futures.

NOte: This is not being done in parallel. However, due to a DGC thread leak issue, we now use ThickFutures. Thus, the tasks that are being cancelled are all local tasks running on the #executorService. If that local task is doing an RMI, then cancelling it will cause an interrupt in the NIO request.