X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fpersisted%2FCommitTransactionPayload.java;h=3fc636b851022fcc39e65acee3365113c05b4666;hb=99f80f27bee37bb23e345420bf14bb7bb4793c28;hp=b2164133b5b1209c53484852414bf88bc84a43a1;hpb=9b319f491af1c65705b69e8a182aab5006a2f959;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java
index b2164133b5..3fc636b851 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/persisted/CommitTransactionPayload.java
@@ -8,8 +8,8 @@
 package org.opendaylight.controller.cluster.datastore.persisted;
 
 import static com.google.common.base.Verify.verifyNotNull;
+import static com.google.common.math.IntMath.ceilingPowerOfTwo;
 import static java.util.Objects.requireNonNull;
-import static org.opendaylight.controller.cluster.datastore.persisted.ChunkedOutputStream.MAX_ARRAY_SIZE;
 
 import com.google.common.annotations.Beta;
 import com.google.common.annotations.VisibleForTesting;
@@ -26,12 +26,16 @@ import java.io.Serializable;
 import java.io.StreamCorruptedException;
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.Map.Entry;
+import org.eclipse.jdt.annotation.NonNull;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
+import org.opendaylight.controller.cluster.datastore.persisted.DataTreeCandidateInputOutput.DataTreeCandidateWithVersion;
+import org.opendaylight.controller.cluster.io.ChunkedByteArray;
+import org.opendaylight.controller.cluster.io.ChunkedOutputStream;
 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.IdentifiablePayload;
-import org.opendaylight.yangtools.concepts.Variant;
+import org.opendaylight.yangtools.concepts.Either;
 import org.opendaylight.yangtools.yang.data.api.schema.stream.ReusableStreamReceiver;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
 import org.opendaylight.yangtools.yang.data.impl.schema.ReusableImmutableNormalizedNodeStreamWriter;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -47,34 +51,43 @@ public abstract class CommitTransactionPayload extends IdentifiablePayload<Trans
     private static final Logger LOG = LoggerFactory.getLogger(CommitTransactionPayload.class);
     private static final long serialVersionUID = 1L;
 
-    private volatile Entry<TransactionIdentifier, DataTreeCandidate> candidate = null;
+    private static final int MAX_ARRAY_SIZE = ceilingPowerOfTwo(Integer.getInteger(
+        "org.opendaylight.controller.cluster.datastore.persisted.max-array-size", 256 * 1024));
+
+    private volatile Entry<TransactionIdentifier, DataTreeCandidateWithVersion> candidate = null;
 
     CommitTransactionPayload() {
 
     }
 
-    public static CommitTransactionPayload create(final TransactionIdentifier transactionId,
-            final DataTreeCandidate candidate, final int initialSerializedBufferCapacity) throws IOException {
-
-        final ChunkedOutputStream cos = new ChunkedOutputStream(initialSerializedBufferCapacity);
+    public static @NonNull CommitTransactionPayload create(final TransactionIdentifier transactionId,
+            final DataTreeCandidate candidate, final PayloadVersion version, final int initialSerializedBufferCapacity)
+                    throws IOException {
+        final ChunkedOutputStream cos = new ChunkedOutputStream(initialSerializedBufferCapacity, MAX_ARRAY_SIZE);
         try (DataOutputStream dos = new DataOutputStream(cos)) {
             transactionId.writeTo(dos);
-            DataTreeCandidateInputOutput.writeDataTreeCandidate(dos, candidate);
+            DataTreeCandidateInputOutput.writeDataTreeCandidate(dos, version, candidate);
         }
 
-        final Variant<byte[], ChunkedByteArray> source = cos.toVariant();
+        final Either<byte[], ChunkedByteArray> source = cos.toVariant();
         LOG.debug("Initial buffer capacity {}, actual serialized size {}", initialSerializedBufferCapacity, cos.size());
         return source.isFirst() ? new Simple(source.getFirst()) : new Chunked(source.getSecond());
     }
 
     @VisibleForTesting
-    public static CommitTransactionPayload create(final TransactionIdentifier transactionId,
+    public static @NonNull CommitTransactionPayload create(final TransactionIdentifier transactionId,
+            final DataTreeCandidate candidate, final PayloadVersion version) throws IOException {
+        return create(transactionId, candidate, version, 512);
+    }
+
+    @VisibleForTesting
+    public static @NonNull CommitTransactionPayload create(final TransactionIdentifier transactionId,
             final DataTreeCandidate candidate) throws IOException {
-        return create(transactionId, candidate, 512);
+        return create(transactionId, candidate, PayloadVersion.current());
     }
 
-    public Entry<TransactionIdentifier, DataTreeCandidate> getCandidate() throws IOException {
-        Entry<TransactionIdentifier, DataTreeCandidate> localCandidate = candidate;
+    public @NonNull Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate() throws IOException {
+        Entry<TransactionIdentifier, DataTreeCandidateWithVersion> localCandidate = candidate;
         if (localCandidate == null) {
             synchronized (this) {
                 localCandidate = candidate;
@@ -86,13 +99,14 @@ public abstract class CommitTransactionPayload extends IdentifiablePayload<Trans
         return localCandidate;
     }
 
-    public final Entry<TransactionIdentifier, DataTreeCandidate> getCandidate(
+    public final @NonNull Entry<TransactionIdentifier, DataTreeCandidateWithVersion> getCandidate(
             final ReusableStreamReceiver receiver) throws IOException {
         final DataInput in = newDataInput();
         return new SimpleImmutableEntry<>(TransactionIdentifier.readFrom(in),
                 DataTreeCandidateInputOutput.readDataTreeCandidate(in, receiver));
     }
 
+    @Override
     public TransactionIdentifier getIdentifier() {
         try  {
             return getCandidate().getKey();
@@ -101,6 +115,17 @@ public abstract class CommitTransactionPayload extends IdentifiablePayload<Trans
         }
     }
 
+    /**
+     * The cached candidate needs to be cleared after it is done applying to the DataTree, otherwise it would be keeping
+     * deserialized in memory which are not needed anymore leading to wasted memory. This lets the payload know that
+     * this was the last time the candidate was needed ant it is safe to be cleared.
+     */
+    public Entry<TransactionIdentifier, DataTreeCandidateWithVersion> acquireCandidate() throws IOException {
+        final Entry<TransactionIdentifier, DataTreeCandidateWithVersion> localCandidate = getCandidate();
+        candidate = null;
+        return localCandidate;
+    }
+
     abstract void writeBytes(ObjectOutput out) throws IOException;
 
     abstract DataInput newDataInput();