d53cb48e501c75c25e3532ce2c71e481717612ad
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / Shard.java
1 /*
2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.actor.ActorRef;
12 import akka.actor.ActorSelection;
13 import akka.actor.Cancellable;
14 import akka.actor.PoisonPill;
15 import akka.actor.Props;
16 import akka.event.Logging;
17 import akka.event.LoggingAdapter;
18 import akka.japi.Creator;
19 import akka.persistence.RecoveryFailure;
20 import akka.serialization.Serialization;
21 import com.google.common.annotations.VisibleForTesting;
22 import com.google.common.base.Optional;
23 import com.google.common.base.Preconditions;
24 import com.google.common.collect.Lists;
25 import com.google.common.util.concurrent.FutureCallback;
26 import com.google.common.util.concurrent.Futures;
27 import com.google.common.util.concurrent.ListenableFuture;
28 import com.google.protobuf.ByteString;
29 import com.google.protobuf.InvalidProtocolBufferException;
30 import java.util.Collection;
31 import java.util.HashMap;
32 import java.util.List;
33 import java.util.Map;
34 import java.util.concurrent.ExecutionException;
35 import java.util.concurrent.TimeUnit;
36 import javax.annotation.Nonnull;
37 import org.opendaylight.controller.cluster.DataPersistenceProvider;
38 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
39 import org.opendaylight.controller.cluster.common.actor.MeteringBehavior;
40 import org.opendaylight.controller.cluster.datastore.ShardCommitCoordinator.CohortEntry;
41 import org.opendaylight.controller.cluster.datastore.compat.BackwardsCompatibleThreePhaseCommitCohort;
42 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
43 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
44 import org.opendaylight.controller.cluster.datastore.identifiers.ShardTransactionIdentifier;
45 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardMBeanFactory;
46 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
47 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
48 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
49 import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
50 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
51 import org.opendaylight.controller.cluster.datastore.messages.CloseTransactionChain;
52 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
53 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
54 import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
55 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
56 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
57 import org.opendaylight.controller.cluster.datastore.messages.ForwardedReadyTransaction;
58 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
59 import org.opendaylight.controller.cluster.datastore.messages.ReadData;
60 import org.opendaylight.controller.cluster.datastore.messages.ReadDataReply;
61 import org.opendaylight.controller.cluster.datastore.messages.ReadyTransactionReply;
62 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
63 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
64 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
65 import org.opendaylight.controller.cluster.datastore.modification.Modification;
66 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
67 import org.opendaylight.controller.cluster.datastore.node.NormalizedNodeToNodeCodec;
68 import org.opendaylight.controller.cluster.raft.RaftActor;
69 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
70 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
71 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.CompositeModificationPayload;
72 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
73 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
74 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
75 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreFactory;
76 import org.opendaylight.controller.protobuff.messages.common.NormalizedNodeMessages;
77 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
78 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
79 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionFactory;
80 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
81 import org.opendaylight.yangtools.concepts.ListenerRegistration;
82 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
83 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
84 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
85 import scala.concurrent.duration.Duration;
86 import scala.concurrent.duration.FiniteDuration;
87
88 /**
89  * A Shard represents a portion of the logical data tree <br/>
90  * <p>
91  * Our Shard uses InMemoryDataStore as it's internal representation and delegates all requests it
92  * </p>
93  */
94 public class Shard extends RaftActor {
95
96     private static final Object COMMIT_TRANSACTION_REPLY = new CommitTransactionReply().toSerializable();
97
98     private static final Object TX_COMMIT_TIMEOUT_CHECK_MESSAGE = "txCommitTimeoutCheck";
99
100     public static final String DEFAULT_NAME = "default";
101
102     // The state of this Shard
103     private final InMemoryDOMDataStore store;
104
105     private final LoggingAdapter LOG =
106         Logging.getLogger(getContext().system(), this);
107
108     /// The name of this shard
109     private final ShardIdentifier name;
110
111     private final ShardStats shardMBean;
112
113     private final List<ActorSelection> dataChangeListeners =  Lists.newArrayList();
114
115     private final List<DelayedListenerRegistration> delayedListenerRegistrations =
116                                                                        Lists.newArrayList();
117
118     private final DatastoreContext datastoreContext;
119
120     private final DataPersistenceProvider dataPersistenceProvider;
121
122     private SchemaContext schemaContext;
123
124     private ActorRef createSnapshotTransaction;
125
126     private int createSnapshotTransactionCounter;
127
128     private final ShardCommitCoordinator commitCoordinator;
129
130     private final long transactionCommitTimeout;
131
132     private Cancellable txCommitTimeoutCheckSchedule;
133
134     /**
135      * Coordinates persistence recovery on startup.
136      */
137     private ShardRecoveryCoordinator recoveryCoordinator;
138     private List<Object> currentLogRecoveryBatch;
139
140     private final Map<String, DOMStoreTransactionChain> transactionChains = new HashMap<>();
141
142     protected Shard(final ShardIdentifier name, final Map<ShardIdentifier, String> peerAddresses,
143             final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
144         super(name.toString(), mapPeerAddresses(peerAddresses),
145                 Optional.of(datastoreContext.getShardRaftConfig()));
146
147         this.name = name;
148         this.datastoreContext = datastoreContext;
149         this.schemaContext = schemaContext;
150         this.dataPersistenceProvider = (datastoreContext.isPersistent()) ? new PersistentDataProvider() : new NonPersistentRaftDataProvider();
151
152         LOG.info("Shard created : {} persistent : {}", name, datastoreContext.isPersistent());
153
154         store = InMemoryDOMDataStoreFactory.create(name.toString(), null,
155                 datastoreContext.getDataStoreProperties());
156
157         if(schemaContext != null) {
158             store.onGlobalContextUpdated(schemaContext);
159         }
160
161         shardMBean = ShardMBeanFactory.getShardStatsMBean(name.toString(),
162                 datastoreContext.getDataStoreMXBeanType());
163         shardMBean.setNotificationManager(store.getDataChangeListenerNotificationManager());
164
165         if (isMetricsCaptureEnabled()) {
166             getContext().become(new MeteringBehavior(this));
167         }
168
169         commitCoordinator = new ShardCommitCoordinator(TimeUnit.SECONDS.convert(1, TimeUnit.MINUTES),
170                 datastoreContext.getShardTransactionCommitQueueCapacity());
171
172         transactionCommitTimeout = TimeUnit.MILLISECONDS.convert(
173                 datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS);
174     }
175
176     private static Map<String, String> mapPeerAddresses(
177         final Map<ShardIdentifier, String> peerAddresses) {
178         Map<String, String> map = new HashMap<>();
179
180         for (Map.Entry<ShardIdentifier, String> entry : peerAddresses
181             .entrySet()) {
182             map.put(entry.getKey().toString(), entry.getValue());
183         }
184
185         return map;
186     }
187
188     public static Props props(final ShardIdentifier name,
189         final Map<ShardIdentifier, String> peerAddresses,
190         final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
191         Preconditions.checkNotNull(name, "name should not be null");
192         Preconditions.checkNotNull(peerAddresses, "peerAddresses should not be null");
193         Preconditions.checkNotNull(datastoreContext, "dataStoreContext should not be null");
194         Preconditions.checkNotNull(schemaContext, "schemaContext should not be null");
195
196         return Props.create(new ShardCreator(name, peerAddresses, datastoreContext, schemaContext));
197     }
198
199     @Override
200     public void postStop() {
201         super.postStop();
202
203         if(txCommitTimeoutCheckSchedule != null) {
204             txCommitTimeoutCheckSchedule.cancel();
205         }
206     }
207
208     @Override
209     public void onReceiveRecover(final Object message) throws Exception {
210         if(LOG.isDebugEnabled()) {
211             LOG.debug("onReceiveRecover: Received message {} from {}",
212                 message.getClass().toString(),
213                 getSender());
214         }
215
216         if (message instanceof RecoveryFailure){
217             LOG.error(((RecoveryFailure) message).cause(), "Recovery failed because of this cause");
218
219             // Even though recovery failed, we still need to finish our recovery, eg send the
220             // ActorInitialized message and start the txCommitTimeoutCheckSchedule.
221             onRecoveryComplete();
222         } else {
223             super.onReceiveRecover(message);
224         }
225     }
226
227     @Override
228     public void onReceiveCommand(final Object message) throws Exception {
229         if(LOG.isDebugEnabled()) {
230             LOG.debug("onReceiveCommand: Received message {} from {}", message, getSender());
231         }
232
233         if(message.getClass().equals(ReadDataReply.SERIALIZABLE_CLASS)) {
234             handleReadDataReply(message);
235         } else if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
236             handleCreateTransaction(message);
237         } else if(message instanceof ForwardedReadyTransaction) {
238             handleForwardedReadyTransaction((ForwardedReadyTransaction)message);
239         } else if(message.getClass().equals(CanCommitTransaction.SERIALIZABLE_CLASS)) {
240             handleCanCommitTransaction(CanCommitTransaction.fromSerializable(message));
241         } else if(message.getClass().equals(CommitTransaction.SERIALIZABLE_CLASS)) {
242             handleCommitTransaction(CommitTransaction.fromSerializable(message));
243         } else if(message.getClass().equals(AbortTransaction.SERIALIZABLE_CLASS)) {
244             handleAbortTransaction(AbortTransaction.fromSerializable(message));
245         } else if (message.getClass().equals(CloseTransactionChain.SERIALIZABLE_CLASS)){
246             closeTransactionChain(CloseTransactionChain.fromSerializable(message));
247         } else if (message instanceof RegisterChangeListener) {
248             registerChangeListener((RegisterChangeListener) message);
249         } else if (message instanceof UpdateSchemaContext) {
250             updateSchemaContext((UpdateSchemaContext) message);
251         } else if (message instanceof PeerAddressResolved) {
252             PeerAddressResolved resolved = (PeerAddressResolved) message;
253             setPeerAddress(resolved.getPeerId().toString(),
254                 resolved.getPeerAddress());
255         } else if(message.equals(TX_COMMIT_TIMEOUT_CHECK_MESSAGE)) {
256             handleTransactionCommitTimeoutCheck();
257         } else {
258             super.onReceiveCommand(message);
259         }
260     }
261
262     private void handleTransactionCommitTimeoutCheck() {
263         CohortEntry cohortEntry = commitCoordinator.getCurrentCohortEntry();
264         if(cohortEntry != null) {
265             long elapsed = System.currentTimeMillis() - cohortEntry.getLastAccessTime();
266             if(elapsed > transactionCommitTimeout) {
267                 LOG.warning("Current transaction {} has timed out after {} ms - aborting",
268                         cohortEntry.getTransactionID(), transactionCommitTimeout);
269
270                 doAbortTransaction(cohortEntry.getTransactionID(), null);
271             }
272         }
273     }
274
275     private void handleCommitTransaction(final CommitTransaction commit) {
276         final String transactionID = commit.getTransactionID();
277
278         LOG.debug("Committing transaction {}", transactionID);
279
280         // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to
281         // this transaction.
282         final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
283         if(cohortEntry == null) {
284             // We're not the current Tx - the Tx was likely expired b/c it took too long in
285             // between the canCommit and commit messages.
286             IllegalStateException ex = new IllegalStateException(
287                     String.format("Cannot commit transaction %s - it is not the current transaction",
288                             transactionID));
289             LOG.error(ex.getMessage());
290             shardMBean.incrementFailedTransactionsCount();
291             getSender().tell(new akka.actor.Status.Failure(ex), getSelf());
292             return;
293         }
294
295         // We perform the preCommit phase here atomically with the commit phase. This is an
296         // optimization to eliminate the overhead of an extra preCommit message. We lose front-end
297         // coordination of preCommit across shards in case of failure but preCommit should not
298         // normally fail since we ensure only one concurrent 3-phase commit.
299
300         try {
301             // We block on the future here so we don't have to worry about possibly accessing our
302             // state on a different thread outside of our dispatcher. Also, the data store
303             // currently uses a same thread executor anyway.
304             cohortEntry.getCohort().preCommit().get();
305
306             Shard.this.persistData(getSender(), transactionID,
307                     new CompositeModificationPayload(cohortEntry.getModification().toSerializable()));
308         } catch (InterruptedException | ExecutionException e) {
309             LOG.error(e, "An exception occurred while preCommitting transaction {}",
310                     cohortEntry.getTransactionID());
311             shardMBean.incrementFailedTransactionsCount();
312             getSender().tell(new akka.actor.Status.Failure(e), getSelf());
313         }
314
315         cohortEntry.updateLastAccessTime();
316     }
317
318     private void finishCommit(@Nonnull final ActorRef sender, final @Nonnull String transactionID) {
319         // With persistence enabled, this method is called via applyState by the leader strategy
320         // after the commit has been replicated to a majority of the followers.
321
322         CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
323         if(cohortEntry == null) {
324             // The transaction is no longer the current commit. This can happen if the transaction
325             // was aborted prior, most likely due to timeout in the front-end. We need to finish
326             // committing the transaction though since it was successfully persisted and replicated
327             // however we can't use the original cohort b/c it was already preCommitted and may
328             // conflict with the current commit or may have been aborted so we commit with a new
329             // transaction.
330             cohortEntry = commitCoordinator.getAndRemoveCohortEntry(transactionID);
331             if(cohortEntry != null) {
332                 commitWithNewTransaction(cohortEntry.getModification());
333                 sender.tell(COMMIT_TRANSACTION_REPLY, getSelf());
334             } else {
335                 // This really shouldn't happen - it likely means that persistence or replication
336                 // took so long to complete such that the cohort entry was expired from the cache.
337                 IllegalStateException ex = new IllegalStateException(
338                         String.format("Could not finish committing transaction %s - no CohortEntry found",
339                                 transactionID));
340                 LOG.error(ex.getMessage());
341                 sender.tell(new akka.actor.Status.Failure(ex), getSelf());
342             }
343
344             return;
345         }
346
347         LOG.debug("Finishing commit for transaction {}", cohortEntry.getTransactionID());
348
349         try {
350             // We block on the future here so we don't have to worry about possibly accessing our
351             // state on a different thread outside of our dispatcher. Also, the data store
352             // currently uses a same thread executor anyway.
353             cohortEntry.getCohort().commit().get();
354
355             sender.tell(COMMIT_TRANSACTION_REPLY, getSelf());
356
357             shardMBean.incrementCommittedTransactionCount();
358             shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis());
359
360         } catch (InterruptedException | ExecutionException e) {
361             sender.tell(new akka.actor.Status.Failure(e), getSelf());
362
363             LOG.error(e, "An exception occurred while committing transaction {}", transactionID);
364             shardMBean.incrementFailedTransactionsCount();
365         }
366
367         commitCoordinator.currentTransactionComplete(transactionID, true);
368     }
369
370     private void handleCanCommitTransaction(final CanCommitTransaction canCommit) {
371         LOG.debug("Can committing transaction {}", canCommit.getTransactionID());
372         commitCoordinator.handleCanCommit(canCommit, getSender(), self());
373     }
374
375     private void handleForwardedReadyTransaction(ForwardedReadyTransaction ready) {
376         LOG.debug("Readying transaction {}, client version {}", ready.getTransactionID(),
377                 ready.getTxnClientVersion());
378
379         // This message is forwarded by the ShardTransaction on ready. We cache the cohort in the
380         // commitCoordinator in preparation for the subsequent three phase commit initiated by
381         // the front-end.
382         commitCoordinator.transactionReady(ready.getTransactionID(), ready.getCohort(),
383                 ready.getModification());
384
385         // Return our actor path as we'll handle the three phase commit, except if the Tx client
386         // version < 1 (Helium-1 version). This means the Tx was initiated by a base Helium version
387         // node. In that case, the subsequent 3-phase commit messages won't contain the
388         // transactionId so to maintain backwards compatibility, we create a separate cohort actor
389         // to provide the compatible behavior.
390         ActorRef replyActorPath = self();
391         if(ready.getTxnClientVersion() < CreateTransaction.HELIUM_1_VERSION) {
392             LOG.debug("Creating BackwardsCompatibleThreePhaseCommitCohort");
393             replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props(
394                     ready.getTransactionID()));
395         }
396
397         ReadyTransactionReply readyTransactionReply = new ReadyTransactionReply(
398                 Serialization.serializedActorPath(replyActorPath));
399         getSender().tell(ready.isReturnSerialized() ? readyTransactionReply.toSerializable() :
400                 readyTransactionReply, getSelf());
401     }
402
403     private void handleAbortTransaction(final AbortTransaction abort) {
404         doAbortTransaction(abort.getTransactionID(), getSender());
405     }
406
407     private void doAbortTransaction(final String transactionID, final ActorRef sender) {
408         final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID);
409         if(cohortEntry != null) {
410             LOG.debug("Aborting transaction {}", transactionID);
411
412             // We don't remove the cached cohort entry here (ie pass false) in case the Tx was
413             // aborted during replication in which case we may still commit locally if replication
414             // succeeds.
415             commitCoordinator.currentTransactionComplete(transactionID, false);
416
417             final ListenableFuture<Void> future = cohortEntry.getCohort().abort();
418             final ActorRef self = getSelf();
419
420             Futures.addCallback(future, new FutureCallback<Void>() {
421                 @Override
422                 public void onSuccess(final Void v) {
423                     shardMBean.incrementAbortTransactionsCount();
424
425                     if(sender != null) {
426                         sender.tell(new AbortTransactionReply().toSerializable(), self);
427                     }
428                 }
429
430                 @Override
431                 public void onFailure(final Throwable t) {
432                     LOG.error(t, "An exception happened during abort");
433
434                     if(sender != null) {
435                         sender.tell(new akka.actor.Status.Failure(t), self);
436                     }
437                 }
438             });
439         }
440     }
441
442     private void handleCreateTransaction(final Object message) {
443         if (isLeader()) {
444             createTransaction(CreateTransaction.fromSerializable(message));
445         } else if (getLeader() != null) {
446             getLeader().forward(message, getContext());
447         } else {
448             getSender().tell(new akka.actor.Status.Failure(new NoShardLeaderException(
449                 "Could not find shard leader so transaction cannot be created. This typically happens" +
450                 " when the system is coming up or recovering and a leader is being elected. Try again" +
451                 " later.")), getSelf());
452         }
453     }
454
455     private void handleReadDataReply(final Object message) {
456         // This must be for install snapshot. Don't want to open this up and trigger
457         // deSerialization
458
459         self().tell(new CaptureSnapshotReply(ReadDataReply.getNormalizedNodeByteString(message)),
460                 self());
461
462         createSnapshotTransaction = null;
463
464         // Send a PoisonPill instead of sending close transaction because we do not really need
465         // a response
466         getSender().tell(PoisonPill.getInstance(), self());
467     }
468
469     private void closeTransactionChain(final CloseTransactionChain closeTransactionChain) {
470         DOMStoreTransactionChain chain =
471             transactionChains.remove(closeTransactionChain.getTransactionChainId());
472
473         if(chain != null) {
474             chain.close();
475         }
476     }
477
478     private ActorRef createTypedTransactionActor(int transactionType,
479             ShardTransactionIdentifier transactionId, String transactionChainId, int clientVersion ) {
480
481         DOMStoreTransactionFactory factory = store;
482
483         if(!transactionChainId.isEmpty()) {
484             factory = transactionChains.get(transactionChainId);
485             if(factory == null){
486                 DOMStoreTransactionChain transactionChain = store.createTransactionChain();
487                 transactionChains.put(transactionChainId, transactionChain);
488                 factory = transactionChain;
489             }
490         }
491
492         if(this.schemaContext == null) {
493             throw new IllegalStateException("SchemaContext is not set");
494         }
495
496         if (transactionType == TransactionProxy.TransactionType.READ_ONLY.ordinal()) {
497
498             shardMBean.incrementReadOnlyTransactionCount();
499
500             return getContext().actorOf(
501                 ShardTransaction.props(factory.newReadOnlyTransaction(), getSelf(),
502                         schemaContext,datastoreContext, shardMBean,
503                         transactionId.getRemoteTransactionId(), clientVersion),
504                         transactionId.toString());
505
506         } else if (transactionType == TransactionProxy.TransactionType.READ_WRITE.ordinal()) {
507
508             shardMBean.incrementReadWriteTransactionCount();
509
510             return getContext().actorOf(
511                 ShardTransaction.props(factory.newReadWriteTransaction(), getSelf(),
512                         schemaContext, datastoreContext, shardMBean,
513                         transactionId.getRemoteTransactionId(), clientVersion),
514                         transactionId.toString());
515
516
517         } else if (transactionType == TransactionProxy.TransactionType.WRITE_ONLY.ordinal()) {
518
519             shardMBean.incrementWriteOnlyTransactionCount();
520
521             return getContext().actorOf(
522                 ShardTransaction.props(factory.newWriteOnlyTransaction(), getSelf(),
523                         schemaContext, datastoreContext, shardMBean,
524                         transactionId.getRemoteTransactionId(), clientVersion),
525                         transactionId.toString());
526         } else {
527             throw new IllegalArgumentException(
528                 "Shard="+name + ":CreateTransaction message has unidentified transaction type="
529                     + transactionType);
530         }
531     }
532
533     private void createTransaction(CreateTransaction createTransaction) {
534         try {
535             ActorRef transactionActor = createTransaction(createTransaction.getTransactionType(),
536                 createTransaction.getTransactionId(), createTransaction.getTransactionChainId(),
537                 createTransaction.getVersion());
538
539             getSender().tell(new CreateTransactionReply(Serialization.serializedActorPath(transactionActor),
540                     createTransaction.getTransactionId()).toSerializable(), getSelf());
541         } catch (Exception e) {
542             getSender().tell(new akka.actor.Status.Failure(e), getSelf());
543         }
544     }
545
546     private ActorRef createTransaction(int transactionType, String remoteTransactionId,
547             String transactionChainId, int clientVersion) {
548
549         ShardTransactionIdentifier transactionId =
550             ShardTransactionIdentifier.builder()
551                 .remoteTransactionId(remoteTransactionId)
552                 .build();
553
554         if(LOG.isDebugEnabled()) {
555             LOG.debug("Creating transaction : {} ", transactionId);
556         }
557
558         ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId,
559                 transactionChainId, clientVersion);
560
561         return transactionActor;
562     }
563
564     private void syncCommitTransaction(final DOMStoreWriteTransaction transaction)
565         throws ExecutionException, InterruptedException {
566         DOMStoreThreePhaseCommitCohort commitCohort = transaction.ready();
567         commitCohort.preCommit().get();
568         commitCohort.commit().get();
569     }
570
571     private void commitWithNewTransaction(final Modification modification) {
572         DOMStoreWriteTransaction tx = store.newWriteOnlyTransaction();
573         modification.apply(tx);
574         try {
575             syncCommitTransaction(tx);
576             shardMBean.incrementCommittedTransactionCount();
577             shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis());
578         } catch (InterruptedException | ExecutionException e) {
579             shardMBean.incrementFailedTransactionsCount();
580             LOG.error(e, "Failed to commit");
581         }
582     }
583
584     private void updateSchemaContext(final UpdateSchemaContext message) {
585         this.schemaContext = message.getSchemaContext();
586         updateSchemaContext(message.getSchemaContext());
587         store.onGlobalContextUpdated(message.getSchemaContext());
588     }
589
590     @VisibleForTesting
591     void updateSchemaContext(final SchemaContext schemaContext) {
592         store.onGlobalContextUpdated(schemaContext);
593     }
594
595     private void registerChangeListener(final RegisterChangeListener registerChangeListener) {
596
597         LOG.debug("registerDataChangeListener for {}", registerChangeListener.getPath());
598
599         ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
600                                                      NormalizedNode<?, ?>>> registration;
601         if(isLeader()) {
602             registration = doChangeListenerRegistration(registerChangeListener);
603         } else {
604             LOG.debug("Shard is not the leader - delaying registration");
605
606             DelayedListenerRegistration delayedReg =
607                     new DelayedListenerRegistration(registerChangeListener);
608             delayedListenerRegistrations.add(delayedReg);
609             registration = delayedReg;
610         }
611
612         ActorRef listenerRegistration = getContext().actorOf(
613                 DataChangeListenerRegistration.props(registration));
614
615         LOG.debug("registerDataChangeListener sending reply, listenerRegistrationPath = {} ",
616                     listenerRegistration.path());
617
618         getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf());
619     }
620
621     private ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
622                                                NormalizedNode<?, ?>>> doChangeListenerRegistration(
623             final RegisterChangeListener registerChangeListener) {
624
625         ActorSelection dataChangeListenerPath = getContext().system().actorSelection(
626                 registerChangeListener.getDataChangeListenerPath());
627
628         // Notify the listener if notifications should be enabled or not
629         // If this shard is the leader then it will enable notifications else
630         // it will not
631         dataChangeListenerPath.tell(new EnableNotification(true), getSelf());
632
633         // Now store a reference to the data change listener so it can be notified
634         // at a later point if notifications should be enabled or disabled
635         dataChangeListeners.add(dataChangeListenerPath);
636
637         AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> listener =
638                 new DataChangeListenerProxy(schemaContext, dataChangeListenerPath);
639
640         LOG.debug("Registering for path {}", registerChangeListener.getPath());
641
642         return store.registerChangeListener(registerChangeListener.getPath(), listener,
643                 registerChangeListener.getScope());
644     }
645
646     private boolean isMetricsCaptureEnabled(){
647         CommonConfig config = new CommonConfig(getContext().system().settings().config());
648         return config.isMetricCaptureEnabled();
649     }
650
651     @Override
652     protected
653     void startLogRecoveryBatch(final int maxBatchSize) {
654         currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize);
655
656         if(LOG.isDebugEnabled()) {
657             LOG.debug("{} : starting log recovery batch with max size {}", persistenceId(), maxBatchSize);
658         }
659     }
660
661     @Override
662     protected void appendRecoveredLogEntry(final Payload data) {
663         if (data instanceof CompositeModificationPayload) {
664             currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification());
665         } else {
666             LOG.error("Unknown state received {} during recovery", data);
667         }
668     }
669
670     @Override
671     protected void applyRecoverySnapshot(final ByteString snapshot) {
672         if(recoveryCoordinator == null) {
673             recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext);
674         }
675
676         recoveryCoordinator.submit(snapshot, store.newWriteOnlyTransaction());
677
678         if(LOG.isDebugEnabled()) {
679             LOG.debug("{} : submitted recovery sbapshot", persistenceId());
680         }
681     }
682
683     @Override
684     protected void applyCurrentLogRecoveryBatch() {
685         if(recoveryCoordinator == null) {
686             recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext);
687         }
688
689         recoveryCoordinator.submit(currentLogRecoveryBatch, store.newWriteOnlyTransaction());
690
691         if(LOG.isDebugEnabled()) {
692             LOG.debug("{} : submitted log recovery batch with size {}", persistenceId(),
693                     currentLogRecoveryBatch.size());
694         }
695     }
696
697     @Override
698     protected void onRecoveryComplete() {
699         if(recoveryCoordinator != null) {
700             Collection<DOMStoreWriteTransaction> txList = recoveryCoordinator.getTransactions();
701
702             if(LOG.isDebugEnabled()) {
703                 LOG.debug("{} : recovery complete - committing {} Tx's", persistenceId(), txList.size());
704             }
705
706             for(DOMStoreWriteTransaction tx: txList) {
707                 try {
708                     syncCommitTransaction(tx);
709                     shardMBean.incrementCommittedTransactionCount();
710                 } catch (InterruptedException | ExecutionException e) {
711                     shardMBean.incrementFailedTransactionsCount();
712                     LOG.error(e, "Failed to commit");
713                 }
714             }
715         }
716
717         recoveryCoordinator = null;
718         currentLogRecoveryBatch = null;
719         updateJournalStats();
720
721         //notify shard manager
722         getContext().parent().tell(new ActorInitialized(), getSelf());
723
724         // Being paranoid here - this method should only be called once but just in case...
725         if(txCommitTimeoutCheckSchedule == null) {
726             // Schedule a message to be periodically sent to check if the current in-progress
727             // transaction should be expired and aborted.
728             FiniteDuration period = Duration.create(transactionCommitTimeout / 3, TimeUnit.MILLISECONDS);
729             txCommitTimeoutCheckSchedule = getContext().system().scheduler().schedule(
730                     period, period, getSelf(),
731                     TX_COMMIT_TIMEOUT_CHECK_MESSAGE, getContext().dispatcher(), ActorRef.noSender());
732         }
733     }
734
735     @Override
736     protected void applyState(final ActorRef clientActor, final String identifier, final Object data) {
737
738         if (data instanceof CompositeModificationPayload) {
739             Object modification = ((CompositeModificationPayload) data).getModification();
740
741             if(modification == null) {
742                 LOG.error(
743                      "modification is null - this is very unexpected, clientActor = {}, identifier = {}",
744                      identifier, clientActor != null ? clientActor.path().toString() : null);
745             } else if(clientActor == null) {
746                 // There's no clientActor to which to send a commit reply so we must be applying
747                 // replicated state from the leader.
748                 commitWithNewTransaction(MutableCompositeModification.fromSerializable(
749                         modification, schemaContext));
750             } else {
751                 // This must be the OK to commit after replication consensus.
752                 finishCommit(clientActor, identifier);
753             }
754         } else {
755             LOG.error("Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}",
756                     data, data.getClass().getClassLoader(),
757                     CompositeModificationPayload.class.getClassLoader());
758         }
759
760         updateJournalStats();
761
762     }
763
764     private void updateJournalStats() {
765         ReplicatedLogEntry lastLogEntry = getLastLogEntry();
766
767         if (lastLogEntry != null) {
768             shardMBean.setLastLogIndex(lastLogEntry.getIndex());
769             shardMBean.setLastLogTerm(lastLogEntry.getTerm());
770         }
771
772         shardMBean.setCommitIndex(getCommitIndex());
773         shardMBean.setLastApplied(getLastApplied());
774     }
775
776     @Override
777     protected void createSnapshot() {
778         if (createSnapshotTransaction == null) {
779
780             // Create a transaction. We are really going to treat the transaction as a worker
781             // so that this actor does not get block building the snapshot
782             createSnapshotTransaction = createTransaction(
783                 TransactionProxy.TransactionType.READ_ONLY.ordinal(),
784                 "createSnapshot" + ++createSnapshotTransactionCounter, "",
785                 CreateTransaction.CURRENT_VERSION);
786
787             createSnapshotTransaction.tell(
788                 new ReadData(YangInstanceIdentifier.builder().build()).toSerializable(), self());
789
790         }
791     }
792
793     @VisibleForTesting
794     @Override
795     protected void applySnapshot(final ByteString snapshot) {
796         // Since this will be done only on Recovery or when this actor is a Follower
797         // we can safely commit everything in here. We not need to worry about event notifications
798         // as they would have already been disabled on the follower
799
800         LOG.info("Applying snapshot");
801         try {
802             DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction();
803             NormalizedNodeMessages.Node serializedNode = NormalizedNodeMessages.Node.parseFrom(snapshot);
804             NormalizedNode<?, ?> node = new NormalizedNodeToNodeCodec(schemaContext)
805                 .decode(serializedNode);
806
807             // delete everything first
808             transaction.delete(YangInstanceIdentifier.builder().build());
809
810             // Add everything from the remote node back
811             transaction.write(YangInstanceIdentifier.builder().build(), node);
812             syncCommitTransaction(transaction);
813         } catch (InvalidProtocolBufferException | InterruptedException | ExecutionException e) {
814             LOG.error(e, "An exception occurred when applying snapshot");
815         } finally {
816             LOG.info("Done applying snapshot");
817         }
818     }
819
820     @Override
821     protected void onStateChanged() {
822         boolean isLeader = isLeader();
823         for (ActorSelection dataChangeListener : dataChangeListeners) {
824             dataChangeListener.tell(new EnableNotification(isLeader), getSelf());
825         }
826
827         if(isLeader) {
828             for(DelayedListenerRegistration reg: delayedListenerRegistrations) {
829                 if(!reg.isClosed()) {
830                     reg.setDelegate(doChangeListenerRegistration(reg.getRegisterChangeListener()));
831                 }
832             }
833
834             delayedListenerRegistrations.clear();
835         }
836
837         shardMBean.setRaftState(getRaftState().name());
838         shardMBean.setCurrentTerm(getCurrentTerm());
839
840         // If this actor is no longer the leader close all the transaction chains
841         if(!isLeader){
842             for(Map.Entry<String, DOMStoreTransactionChain> entry : transactionChains.entrySet()){
843                 if(LOG.isDebugEnabled()) {
844                     LOG.debug(
845                         "onStateChanged: Closing transaction chain {} because shard {} is no longer the leader",
846                         entry.getKey(), getId());
847                 }
848                 entry.getValue().close();
849             }
850
851             transactionChains.clear();
852         }
853     }
854
855     @Override
856     protected DataPersistenceProvider persistence() {
857         return dataPersistenceProvider;
858     }
859
860     @Override protected void onLeaderChanged(final String oldLeader, final String newLeader) {
861         shardMBean.setLeader(newLeader);
862     }
863
864     @Override public String persistenceId() {
865         return this.name.toString();
866     }
867
868     @VisibleForTesting
869     DataPersistenceProvider getDataPersistenceProvider() {
870         return dataPersistenceProvider;
871     }
872
873     private static class ShardCreator implements Creator<Shard> {
874
875         private static final long serialVersionUID = 1L;
876
877         final ShardIdentifier name;
878         final Map<ShardIdentifier, String> peerAddresses;
879         final DatastoreContext datastoreContext;
880         final SchemaContext schemaContext;
881
882         ShardCreator(final ShardIdentifier name, final Map<ShardIdentifier, String> peerAddresses,
883                 final DatastoreContext datastoreContext, final SchemaContext schemaContext) {
884             this.name = name;
885             this.peerAddresses = peerAddresses;
886             this.datastoreContext = datastoreContext;
887             this.schemaContext = schemaContext;
888         }
889
890         @Override
891         public Shard create() throws Exception {
892             return new Shard(name, peerAddresses, datastoreContext, schemaContext);
893         }
894     }
895
896     @VisibleForTesting
897     InMemoryDOMDataStore getDataStore() {
898         return store;
899     }
900
901     @VisibleForTesting
902     ShardStats getShardMBean() {
903         return shardMBean;
904     }
905
906     private static class DelayedListenerRegistration implements
907         ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>>> {
908
909         private volatile boolean closed;
910
911         private final RegisterChangeListener registerChangeListener;
912
913         private volatile ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
914                                                              NormalizedNode<?, ?>>> delegate;
915
916         DelayedListenerRegistration(final RegisterChangeListener registerChangeListener) {
917             this.registerChangeListener = registerChangeListener;
918         }
919
920         void setDelegate( final ListenerRegistration<AsyncDataChangeListener<YangInstanceIdentifier,
921                                             NormalizedNode<?, ?>>> registration) {
922             this.delegate = registration;
923         }
924
925         boolean isClosed() {
926             return closed;
927         }
928
929         RegisterChangeListener getRegisterChangeListener() {
930             return registerChangeListener;
931         }
932
933         @Override
934         public AsyncDataChangeListener<YangInstanceIdentifier, NormalizedNode<?, ?>> getInstance() {
935             return delegate != null ? delegate.getInstance() : null;
936         }
937
938         @Override
939         public void close() {
940             closed = true;
941             if(delegate != null) {
942                 delegate.close();
943             }
944         }
945     }
946 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.