2 * Copyright (c) 2014 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore;
10 import com.google.common.collect.Lists;
11 import com.google.common.util.concurrent.ThreadFactoryBuilder;
12 import java.util.Collection;
13 import java.util.Collections;
14 import java.util.List;
15 import java.util.concurrent.ExecutorService;
16 import java.util.concurrent.Executors;
17 import java.util.concurrent.TimeUnit;
18 import org.opendaylight.controller.cluster.datastore.modification.MutableCompositeModification;
19 import org.opendaylight.controller.cluster.datastore.utils.SerializationUtils;
20 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
21 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
22 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
23 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
28 * Coordinates persistence recovery of journal log entries and snapshots for a shard. Each snapshot
29 * and journal log entry batch are de-serialized and applied to their own write transaction
30 * instance in parallel on a thread pool for faster recovery time. However the transactions are
31 * committed to the data store in the order the corresponding snapshot or log batch are received
32 * to preserve data store integrity.
34 * @author Thomas Panetelis
36 class ShardRecoveryCoordinator {
38 private static final int TIME_OUT = 10;
40 private static final Logger LOG = LoggerFactory.getLogger(ShardRecoveryCoordinator.class);
42 private final List<DOMStoreWriteTransaction> resultingTxList = Lists.newArrayList();
43 private final SchemaContext schemaContext;
44 private final String shardName;
45 private final ExecutorService executor;
47 ShardRecoveryCoordinator(String shardName, SchemaContext schemaContext) {
48 this.schemaContext = schemaContext;
49 this.shardName = shardName;
51 executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(),
52 new ThreadFactoryBuilder().setDaemon(true)
53 .setNameFormat("ShardRecovery-" + shardName + "-%d").build());
57 * Submits a batch of journal log entries.
59 * @param logEntries the serialized journal log entries
60 * @param resultingTx the write Tx to which to apply the entries
62 void submit(List<Object> logEntries, DOMStoreWriteTransaction resultingTx) {
63 LogRecoveryTask task = new LogRecoveryTask(logEntries, resultingTx);
64 resultingTxList.add(resultingTx);
65 executor.execute(task);
71 * @param snapshotBytes the serialized snapshot
72 * @param resultingTx the write Tx to which to apply the entries
74 void submit(byte[] snapshotBytes, DOMStoreWriteTransaction resultingTx) {
75 SnapshotRecoveryTask task = new SnapshotRecoveryTask(snapshotBytes, resultingTx);
76 resultingTxList.add(resultingTx);
77 executor.execute(task);
80 Collection<DOMStoreWriteTransaction> getTransactions() {
81 // Shutdown the executor and wait for task completion.
85 if(executor.awaitTermination(TIME_OUT, TimeUnit.MINUTES)) {
86 return resultingTxList;
88 LOG.error("Recovery for shard {} timed out after {} minutes", shardName, TIME_OUT);
90 } catch (InterruptedException e) {
91 Thread.currentThread().interrupt();
94 return Collections.emptyList();
97 private static abstract class ShardRecoveryTask implements Runnable {
99 final DOMStoreWriteTransaction resultingTx;
101 ShardRecoveryTask(DOMStoreWriteTransaction resultingTx) {
102 this.resultingTx = resultingTx;
106 private class LogRecoveryTask extends ShardRecoveryTask {
108 private final List<Object> logEntries;
110 LogRecoveryTask(List<Object> logEntries, DOMStoreWriteTransaction resultingTx) {
112 this.logEntries = logEntries;
117 for(int i = 0; i < logEntries.size(); i++) {
118 MutableCompositeModification.fromSerializable(
119 logEntries.get(i)).apply(resultingTx);
120 // Null out to GC quicker.
121 logEntries.set(i, null);
126 private class SnapshotRecoveryTask extends ShardRecoveryTask {
128 private final byte[] snapshotBytes;
130 SnapshotRecoveryTask(byte[] snapshotBytes, DOMStoreWriteTransaction resultingTx) {
132 this.snapshotBytes = snapshotBytes;
137 NormalizedNode<?, ?> node = SerializationUtils.deserializeNormalizedNode(snapshotBytes);
139 // delete everything first
140 resultingTx.delete(YangInstanceIdentifier.builder().build());
142 // Add everything from the remote node back
143 resultingTx.write(YangInstanceIdentifier.builder().build(), node);