</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <configuration>
+ <propertyExpansion>checkstyle.violationSeverity=error</propertyExpansion>
+ </configuration>
+ </plugin>
<plugin>
<groupId>org.jacoco</groupId>
<artifactId>jacoco-maven-plugin</artifactId>
}
/**
- * Callback invoked from {@link ClientTransaction} when a transaction has been sub
+ * Callback invoked from {@link ClientTransaction} when a transaction has been submitted.
*
* @param transaction Transaction handle
*/
static AbstractProxyHistory create(final DistributedDataStoreClientBehavior client,
final Optional<ShardBackendInfo> backendInfo, final LocalHistoryIdentifier identifier) {
final Optional<DataTree> dataTree = backendInfo.flatMap(ShardBackendInfo::getDataTree);
- return dataTree.isPresent() ? new LocalProxyHistory(client, identifier, dataTree.get()) : new RemoteProxyHistory(client, identifier);
+ return dataTree.isPresent() ? new LocalProxyHistory(client, identifier, dataTree.get())
+ : new RemoteProxyHistory(client, identifier);
}
@Override
return doCreateTransactionProxy(client, new TransactionIdentifier(identifier, txId.getTransactionId()));
}
- abstract AbstractProxyTransaction doCreateTransactionProxy(DistributedDataStoreClientBehavior client,
+ abstract AbstractProxyTransaction doCreateTransactionProxy(DistributedDataStoreClientBehavior clientBehavior,
TransactionIdentifier txId);
}
/**
* Class translating transaction operations towards a particular backend shard.
*
+ * <p>
* This class is not safe to access from multiple application threads, as is usual for transactions. Internal state
* transitions coming from interactions with backend are expected to be thread-safe.
*
+ * <p>
* This class interacts with the queueing mechanism in ClientActorBehavior, hence once we arrive at a decision
* to use either a local or remote implementation, we are stuck with it. We can re-evaluate on the next transaction.
*
}
/**
- * Seal this transaction before it is either
+ * Seals this transaction when ready.
*/
final void seal() {
checkSealed();
doAbort();
}
+ void abort(final VotingFuture<Void> ret) {
+ checkSealed();
+
+ sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), t -> {
+ if (t instanceof TransactionAbortSuccess) {
+ ret.voteYes();
+ } else if (t instanceof RequestFailure) {
+ ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+ } else {
+ ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
+ }
+ });
+ }
+
/**
* Commit this transaction, possibly in a coordinated fashion.
*
return ret;
}
- void abort(final VotingFuture<Void> ret) {
- checkSealed();
-
- sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), t -> {
- if (t instanceof TransactionAbortSuccess) {
- ret.voteYes();
- } else if (t instanceof RequestFailure) {
- ret.voteNo(((RequestFailure<?, ?>) t).getCause());
- } else {
- ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
- }
- });
- }
-
void canCommit(final VotingFuture<?> ret) {
checkSealed();
void preCommit(final VotingFuture<?> ret) {
checkSealed();
- sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), t-> {
+ sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
if (t instanceof TransactionPreCommitSuccess) {
ret.voteYes();
} else if (t instanceof RequestFailure) {
void doCommit(final VotingFuture<?> ret) {
checkSealed();
- sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t-> {
+ sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
if (t instanceof TransactionCommitSuccess) {
ret.voteYes();
} else if (t instanceof RequestFailure) {
});
}
+ abstract TransactionRequest<?> doCommit(boolean coordinated);
+
abstract void doDelete(final YangInstanceIdentifier path);
abstract void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
abstract CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path);
- abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(final YangInstanceIdentifier path);
+ abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(
+ final YangInstanceIdentifier path);
abstract void doSeal();
abstract void doAbort();
-
- abstract TransactionRequest<?> doCommit(boolean coordinated);
}
* Client-side view of a local history. This class tracks all state related to a particular history and routes
* frontend requests towards the backend.
*
+ * <p>
* This interface is used by the world outside of the actor system and in the actor system it is manifested via
* its client actor. That requires some state transfer with {@link DistributedDataStoreClientBehavior}. In order to
* reduce request latency, all messages are carbon-copied (and enqueued first) to the client actor.
/**
* Client-side view of a free-standing transaction.
*
+ * <p>
* This interface is used by the world outside of the actor system and in the actor system it is manifested via
* its client actor. That requires some state transfer with {@link DistributedDataStoreClientBehavior}. In order to
* reduce request latency, all messages are carbon-copied (and enqueued first) to the client actor.
*
+ * <p>
* It is internally composed of multiple {@link RemoteProxyTransaction}s, each responsible for a component shard.
*
+ * <p>
* Implementation is quite a bit complex, and involves cooperation with {@link AbstractClientHistory} for tracking
* gaps in transaction identifiers seen by backends.
*
+ * <p>
* These gaps need to be accounted for in the transaction setup message sent to a particular backend, so it can verify
* that the requested transaction is in-sequence. This is critical in ensuring that transactions (which are independent
* entities from message queueing perspective) do not get reodered -- thus allowing multiple in-flight transactions.
*
+ * <p>
* Alternative would be to force visibility by sending an abort request to all potential backends, but that would mean
* that even empty transactions increase load on all shards -- which would be a scalability issue.
*
+ * <p>
* Yet another alternative would be to introduce inter-transaction dependencies to the queueing layer in client actor,
* but that would require additional indirection and complexity.
*
final class ClientTransactionCommitCohort extends AbstractTransactionCommitCohort {
private final List<AbstractProxyTransaction> proxies;
- /**
- * @param clientTransaction
- */
ClientTransactionCommitCohort(final Collection<AbstractProxyTransaction> proxies) {
this.proxies = ImmutableList.copyOf(proxies);
}
return ret;
}
-}
\ No newline at end of file
+}
final class DirectTransactionCommitCohort extends AbstractTransactionCommitCohort {
private final AbstractProxyTransaction proxy;
- /**
- * @param clientTransaction
- */
DirectTransactionCommitCohort(final AbstractProxyTransaction proxy) {
this.proxy = Preconditions.checkNotNull(proxy);
}
* Client interface for interacting with the frontend actor. This interface is the primary access point through
* which the DistributedDataStore frontend interacts with backend Shards.
*
+ * <p>
* Keep this interface as clean as possible, as it needs to be implemented in thread-safe and highly-efficient manner.
*
* @author Robert Varga
return new DistributedDataStoreClientBehavior(context, actorContext);
}
- public static Props props(final @Nonnull MemberName memberName, @Nonnull final String storeName, final ActorContext ctx) {
+ public static Props props(@Nonnull final MemberName memberName, @Nonnull final String storeName,
+ final ActorContext ctx) {
final String name = "datastore-" + storeName;
final FrontendIdentifier frontendId = FrontendIdentifier.create(memberName, FrontendType.forName(name));
return Props.create(DistributedDataStoreClientActor.class,
() -> new DistributedDataStoreClientActor(frontendId, ctx));
}
- public static DistributedDataStoreClient getDistributedDataStoreClient(final @Nonnull ActorRef actor,
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ public static DistributedDataStoreClient getDistributedDataStoreClient(@Nonnull final ActorRef actor,
final long timeout, final TimeUnit unit) {
try {
return (DistributedDataStoreClient) Await.result(ExplicitAsk.ask(actor, GET_CLIENT_FACTORY,
* {@link ClientActorBehavior} acting as an intermediary between the backend actors and the DistributedDataStore
* frontend.
*
+ * <p>
* This class is not visible outside of this package because it breaks the actor containment. Services provided to
* Java world outside of actor containment are captured in {@link DistributedDataStoreClient}.
*
+ * <p>
* IMPORTANT: this class breaks actor containment via methods implementing {@link DistributedDataStoreClient} contract.
* When touching internal state, be mindful of the execution context from which execution context, Actor
* or POJO, is the state being accessed or modified.
*
+ * <p>
* THREAD SAFETY: this class must always be kept thread-safe, so that both the Actor System thread and the application
* threads can run concurrently. All state transitions must be made in a thread-safe manner. When in
* doubt, feel free to synchronize on this object.
*
+ * <p>
* PERFORMANCE: this class lies in a performance-critical fast path. All code needs to be concise and efficient, but
* performance must not come at the price of correctness. Any optimizations need to be carefully analyzed
* for correctness and performance impact.
*
+ * <p>
* TRADE-OFFS: part of the functionality runs in application threads without switching contexts, which makes it ideal
* for performing work and charging applications for it. That has two positive effects:
* - CPU usage is distributed across applications, minimizing work done in the actor thread
//
//
+ @SuppressWarnings("checkstyle:IllegalCatch")
private static <K, V extends LocalAbortable> V returnIfOperational(final Map<K , V> map, final K key, final V value,
final Throwable aborted) {
Verify.verify(map.put(key, value) == null);
* An {@link AbstractTransactionCommitCohort} for use with empty transactions. This relies on the fact that no backends
* have been touched, hence all state book-keeping needs to happen only locally and shares fate with the coordinator.
*
+ * <p>
* Therefore all methods can finish immediately without any effects.
*
* @author Robert Varga
final class GetClientRequest {
private final ActorRef replyTo;
- public GetClientRequest(final ActorRef replyTo) {
+ GetClientRequest(final ActorRef replyTo) {
this.replyTo = Preconditions.checkNotNull(replyTo);
}
ActorRef getReplyTo() {
return replyTo;
}
-}
\ No newline at end of file
+}
* Common interface for client histories and client transactions, which can be aborted immediately without replicating
* the effect to the backend. This is needed for abrupt shutdowns.
*
+ * <p>
* Since classes which need to expose this functionality do not need a base class, this is an abstract class and not
* an interface -- which allows us to not leak the {@link #localAbort(Throwable)} method.
*
* An {@link AbstractProxyTransaction} for dispatching a transaction towards a shard leader which is co-located with
* the client instance.
*
+ * <p>
* It requires a {@link DataTreeSnapshot}, which is used to instantiated a new {@link DataTreeModification}. Operations
* are then performed on this modification and once the transaction is submitted, the modification is sent to the shard
* leader.
*
+ * <p>
* This class is not thread-safe as usual with transactions. Since it does not interact with the backend until the
* transaction is submitted, at which point this class gets out of the picture, this is not a cause for concern.
*
if (cookie == null) {
cookie = nextShard++;
- Builder<String, Long> b = ImmutableBiMap.builder();
- b.putAll(shards);
- b.put(shardName, cookie);
- shards = b.build();
+ Builder<String, Long> builder = ImmutableBiMap.builder();
+ builder.putAll(shards);
+ builder.put(shardName, cookie);
+ shards = builder.build();
}
}
}
return new ShardBackendInfo(success.getBackend(),
nextSessionId.getAndIncrement(), success.getVersion(), shardName, UnsignedLong.fromLongBits(cookie),
success.getDataTree(), success.getMaxMessages());
- }).whenComplete((info, t) -> {
- if (t != null) {
- ret.completeExceptionally(t);
+ }).whenComplete((info, throwablw) -> {
+ if (throwablw != null) {
+ ret.completeExceptionally(throwablw);
} else {
ret.complete(info);
}
* An {@link AbstractProxyTransaction} for dispatching a transaction towards a shard leader whose location is currently
* not known or is known to be not co-located with the client.
*
+ * <p>
* It packages operations and sends them via the client actor queue to the shard leader. That queue is responsible for
* maintaining any submitted operations until the leader is discovered.
*
+ * <p>
* This class is not safe to access from multiple application threads, as is usual for transactions. Its internal state
* transitions based on backend responses are thread-safe.
*
}
}
- private void completeRead(final SettableFuture<Optional<NormalizedNode<?, ?>>> future, final Response<?, ?> response) {
+ private void completeRead(final SettableFuture<Optional<NormalizedNode<?, ?>>> future,
+ final Response<?, ?> response) {
LOG.debug("Read request completed with {}", response);
if (response instanceof ReadTransactionSuccess) {
* an exception. This exception corresponds to the cause reported by the first 'no' vote, with all subsequent votes
* added as suppressed exceptions.
*
+ * <p>
* Implementation is geared toward positive votes. Negative votes have to synchronize and therefore are more likely
* to see contention.
*
if (castVote()) {
synchronized (failures) {
resolveResult();
- }
+ }
}
}
* @param hasLeader true if the shard knows about leader ID
*/
abstract void onLeadershipChange(boolean isLeader, boolean hasLeader);
+
abstract void onMessage(M message, boolean isLeader, boolean hasLeader);
}
// in case a TransactionOperation results in another transaction operation being
// queued (eg a put operation from a client read Future callback that is notified
// synchronously).
- Collection<TransactionOperation> operationsBatch = null;
+ final Collection<TransactionOperation> operationsBatch;
synchronized (queuedTxOperations) {
if (queuedTxOperations.isEmpty()) {
// We're done invoking the TransactionOperations so we can now publish the
// TransactionContext.
localTransactionContext.operationHandOffComplete();
- if (!localTransactionContext.usesOperationLimiting()){
+ if (!localTransactionContext.usesOperationLimiting()) {
limiter.releaseAll();
}
transactionContext = localTransactionContext;
import akka.actor.ActorPath;
import akka.actor.ActorRef;
-public class RegisterChangeListenerReply{
+public class RegisterChangeListenerReply {
private final ActorRef listenerRegistration;
public RegisterChangeListenerReply(final ActorRef listenerRegistration) {
Code Review / controller.git / commitdiff