--- /dev/null
+/*
+ * Copyright (c) 2019 PANTHEON.tech, s.r.o. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.cluster.datastore.persisted;
+
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Verify.verify;
+import static com.google.common.math.IntMath.ceilingPowerOfTwo;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableList.Builder;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.Iterator;
+import org.opendaylight.yangtools.concepts.Variant;
+
+/**
+ * An {@link OutputStream} implementation which collects data is a series of {@code byte[]} chunks, each of which has
+ * a fixed maximum size. This is generally preferable to {@link ByteArrayOutputStream}, as that can result in huge
+ * byte arrays -- which can create unnecessary pressure on the GC (as well as lot of copying).
+ *
+ * <p>
+ * This class takes a different approach: it recognizes that result of buffering will be collected at some point, when
+ * the stream is already closed (and thus unmodifiable). Thus it splits the process into two steps:
+ * <ul>
+ * <li>Data acquisition, during which we start with an initial (power-of-two) size and proceed to fill it up. Once the
+ * buffer is full, we stash it, allocate a new buffer twice its size and repeat the process. Once we hit
+ * {@link #MAX_ARRAY_SIZE}, we do not grow subsequent buffer. We also can skip some intermediate sizes if data
+ * is introduced in large chunks via {@link #write(byte[], int, int)}.</li>
+ * <li>Buffer consolidation, which occurs when the stream is {@link #close() closed}. At this point we construct the
+ * final collection of buffers.</li>
+ * </ul>
+ *
+ * <p>
+ * The data acquisition strategy results in predictably-sized buffers, which are growing exponentially in size until
+ * they hit maximum size. Intrinsic property here is that the total capacity of chunks created during the ramp up is
+ * guaranteed to fit into {@code MAX_ARRAY_SIZE}, hence they can readily be compacted into a single buffer, which
+ * replaces them. Combined with the requirement to trim the last buffer to have accurate length, this algorithm
+ * guarantees total number of internal copy operations is capped at {@code 2 * MAX_ARRAY_SIZE}. The number of produced
+ * chunks is also well-controlled:
+ * <ul>
+ * <li>for slowly-built data, we will maintain perfect packing</li>
+ * <li>for fast-startup data, we will be at most one one chunk away from packing perfectly</li>
+ * </ul>
+ *
+ * @author Robert Varga
+ * @author Tomas Olvecky
+ */
+final class ChunkedOutputStream extends OutputStream {
+ static final int MAX_ARRAY_SIZE = ceilingPowerOfTwo(Integer.getInteger(
+ "org.opendaylight.controller.cluster.datastore.persisted.max-array-size", 256 * 1024));
+ private static final int MIN_ARRAY_SIZE = 32;
+
+ // byte[] or a List
+ private Object result;
+ // Lazily-allocated to reduce pressure for single-chunk streams
+ private Deque<byte[]> prevChunks;
+
+ private byte[] currentChunk;
+ private int currentOffset;
+ private int size;
+
+ ChunkedOutputStream(final int requestedInitialCapacity) {
+ currentChunk = new byte[initialCapacity(requestedInitialCapacity)];
+ }
+
+ @Override
+ @SuppressWarnings("checkstyle:ParameterName")
+ public void write(final int b) throws IOException {
+ checkNotClosed();
+ ensureOneByte();
+ currentChunk[currentOffset] = (byte) b;
+ currentOffset++;
+ size++;
+ }
+
+ @Override
+ @SuppressWarnings("checkstyle:ParameterName")
+ public void write(final byte[] b, final int off, final int len) throws IOException {
+ if (len < 0) {
+ throw new IndexOutOfBoundsException();
+ }
+ checkNotClosed();
+
+ int fromOffset = off;
+ int toCopy = len;
+
+ while (toCopy != 0) {
+ final int count = ensureMoreBytes(toCopy);
+ System.arraycopy(b, fromOffset, currentChunk, currentOffset, count);
+ currentOffset += count;
+ size += count;
+ fromOffset += count;
+ toCopy -= count;
+ }
+ }
+
+ @Override
+ public void close() {
+ if (result == null) {
+ result = computeResult();
+ prevChunks = null;
+ currentChunk = null;
+ }
+ }
+
+ int size() {
+ return size;
+ }
+
+ ChunkedByteArray toChunkedByteArray() {
+ checkClosed();
+ return new ChunkedByteArray(size, result instanceof byte[] ? ImmutableList.of((byte[]) result)
+ : (ImmutableList<byte[]>) result);
+ }
+
+ Variant<byte[], ChunkedByteArray> toVariant() {
+ checkClosed();
+ return result instanceof byte[] ? Variant.ofFirst((byte[]) result)
+ : Variant.ofSecond(new ChunkedByteArray(size, (ImmutableList<byte[]>) result));
+ }
+
+ private Object computeResult() {
+ if (prevChunks == null) {
+ // Simple case: it's only the current buffer, return that
+ return trimChunk(currentChunk, currentOffset);
+ }
+ if (size <= MAX_ARRAY_SIZE) {
+ // We have collected less than full chunk of data, let's have just one chunk ...
+ final byte[] singleChunk;
+ if (currentOffset == 0 && prevChunks.size() == 1) {
+ // ... which we have readily available
+ return prevChunks.getFirst();
+ }
+
+ // ... which we need to collect
+ singleChunk = new byte[size];
+ int offset = 0;
+ for (byte[] chunk : prevChunks) {
+ System.arraycopy(chunk, 0, singleChunk, offset, chunk.length);
+ offset += chunk.length;
+ }
+ System.arraycopy(currentChunk, 0, singleChunk, offset, currentOffset);
+ return singleChunk;
+ }
+
+ // Determine number of chunks to aggregate and their required storage. Normally storage would be MAX_ARRAY_SIZE,
+ // but we can have faster-than-exponential startup, which ends up needing less storage -- and we do not want to
+ // end up trimming this array.
+ int headSize = 0;
+ int headCount = 0;
+ final Iterator<byte[]> it = prevChunks.iterator();
+ do {
+ final byte[] chunk = it.next();
+ if (chunk.length == MAX_ARRAY_SIZE) {
+ break;
+ }
+
+ headSize += chunk.length;
+ headCount++;
+ } while (it.hasNext());
+
+ // Compact initial chunks into a single one
+ final byte[] head = new byte[headSize];
+ int offset = 0;
+ for (int i = 0; i < headCount; ++i) {
+ final byte[] chunk = prevChunks.removeFirst();
+ System.arraycopy(chunk, 0, head, offset, chunk.length);
+ offset += chunk.length;
+ }
+ verify(offset == head.length);
+ prevChunks.addFirst(head);
+
+ // Now append the current chunk if need be, potentially trimming it
+ if (currentOffset == 0) {
+ return ImmutableList.copyOf(prevChunks);
+ }
+
+ final Builder<byte[]> builder = ImmutableList.builderWithExpectedSize(prevChunks.size() + 1);
+ builder.addAll(prevChunks);
+ builder.add(trimChunk(currentChunk, currentOffset));
+ return builder.build();
+ }
+
+ // Ensure a single byte
+ private void ensureOneByte() {
+ if (currentChunk.length == currentOffset) {
+ nextChunk(nextChunkSize(currentChunk.length));
+ }
+ }
+
+ // Ensure more than one byte, returns the number of bytes available
+ private int ensureMoreBytes(final int requested) {
+ int available = currentChunk.length - currentOffset;
+ if (available == 0) {
+ nextChunk(nextChunkSize(currentChunk.length, requested));
+ available = currentChunk.length;
+ }
+ final int count = Math.min(requested, available);
+ verify(count > 0);
+ return count;
+ }
+
+ private void nextChunk(final int chunkSize) {
+ if (prevChunks == null) {
+ prevChunks = new ArrayDeque<>();
+ }
+
+ prevChunks.addLast(currentChunk);
+ currentChunk = new byte[chunkSize];
+ currentOffset = 0;
+ }
+
+ private void checkClosed() {
+ checkState(result != null, "Stream has not been closed yet");
+ }
+
+ private void checkNotClosed() throws IOException {
+ if (result != null) {
+ throw new IOException("Stream is already closed");
+ }
+ }
+
+ private static int nextChunkSize(final int currentSize, final int requested) {
+ return currentSize == MAX_ARRAY_SIZE || requested >= MAX_ARRAY_SIZE
+ ? MAX_ARRAY_SIZE : Math.max(currentSize * 2, ceilingPowerOfTwo(requested));
+ }
+
+ private static int nextChunkSize(final int currentSize) {
+ return currentSize < MAX_ARRAY_SIZE ? currentSize * 2 : MAX_ARRAY_SIZE;
+ }
+
+ private static int initialCapacity(final int requestedSize) {
+ if (requestedSize < MIN_ARRAY_SIZE) {
+ return MIN_ARRAY_SIZE;
+ }
+ if (requestedSize > MAX_ARRAY_SIZE) {
+ return MAX_ARRAY_SIZE;
+ }
+ return ceilingPowerOfTwo(requestedSize);
+ }
+
+ private static byte[] trimChunk(final byte[] chunk, final int length) {
+ return chunk.length == length ? chunk : Arrays.copyOf(chunk, length);
+ }
+}
Code Review / controller.git / blobdiff