Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id D2167200C4D for ; Tue, 21 Mar 2017 23:24:26 +0100 (CET) Received: by cust-asf.ponee.io (Postfix) id D0F5E160B6E; Tue, 21 Mar 2017 22:24:26 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id B20F0160B81 for ; Tue, 21 Mar 2017 23:24:24 +0100 (CET) Received: (qmail 95669 invoked by uid 500); 21 Mar 2017 22:24:23 -0000 Mailing-List: contact commits-help@arrow.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@arrow.apache.org Delivered-To: mailing list commits@arrow.apache.org Received: (qmail 95633 invoked by uid 99); 21 Mar 2017 22:24:23 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 21 Mar 2017 22:24:23 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id B4273DFF36; Tue, 21 Mar 2017 22:24:23 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: julien@apache.org To: commits@arrow.apache.org Date: Tue, 21 Mar 2017 22:24:24 -0000 Message-Id: In-Reply-To: <00fdfa547a53442089284957236ae872@git.apache.org> References: <00fdfa547a53442089284957236ae872@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [2/3] arrow git commit: ARROW-208: Add checkstyle policy to java project archived-at: Tue, 21 Mar 2017 22:24:27 -0000 http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/BaseAllocator.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/BaseAllocator.java b/java/memory/src/main/java/org/apache/arrow/memory/BaseAllocator.java index 9edafbc..aaa7ce8 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/BaseAllocator.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/BaseAllocator.java @@ -6,57 +6,54 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *

* Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ + package org.apache.arrow.memory; -import java.util.Arrays; -import java.util.IdentityHashMap; -import java.util.Set; -import java.util.concurrent.atomic.AtomicInteger; +import com.google.common.base.Preconditions; + +import io.netty.buffer.ArrowBuf; +import io.netty.buffer.UnsafeDirectLittleEndian; import org.apache.arrow.memory.AllocationManager.BufferLedger; import org.apache.arrow.memory.util.AssertionUtil; import org.apache.arrow.memory.util.HistoricalLog; -import com.google.common.base.Preconditions; - -import io.netty.buffer.ArrowBuf; -import io.netty.buffer.UnsafeDirectLittleEndian; +import java.util.Arrays; +import java.util.IdentityHashMap; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; public abstract class BaseAllocator extends Accountant implements BufferAllocator { - private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BaseAllocator.class); public static final String DEBUG_ALLOCATOR = "arrow.memory.debug.allocator"; - public static final int DEBUG_LOG_LENGTH = 6; public static final boolean DEBUG = AssertionUtil.isAssertionsEnabled() || Boolean.parseBoolean(System.getProperty(DEBUG_ALLOCATOR, "false")); + private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BaseAllocator + .class); + // Package exposed for sharing between AllocatorManger and BaseAllocator objects + final String name; + final RootAllocator root; private final Object DEBUG_LOCK = DEBUG ? new Object() : null; - private final AllocationListener listener; private final BaseAllocator parentAllocator; private final ArrowByteBufAllocator thisAsByteBufAllocator; private final IdentityHashMap childAllocators; private final ArrowBuf empty; - - private volatile boolean isClosed = false; // the allocator has been closed - - // Package exposed for sharing between AllocatorManger and BaseAllocator objects - final String name; - final RootAllocator root; - // members used purely for debugging private final IdentityHashMap childLedgers; private final IdentityHashMap reservations; private final HistoricalLog historicalLog; + private volatile boolean isClosed = false; // the allocator has been closed protected BaseAllocator( final AllocationListener listener, @@ -91,7 +88,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato this.root = (RootAllocator) this; empty = createEmpty(); } else { - throw new IllegalStateException("An parent allocator must either carry a root or be the root."); + throw new IllegalStateException("An parent allocator must either carry a root or be the " + + "root."); } this.parentAllocator = parentAllocator; @@ -114,11 +112,52 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } + private static String createErrorMsg(final BufferAllocator allocator, final int rounded, final + int requested) { + if (rounded != requested) { + return String.format( + "Unable to allocate buffer of size %d (rounded from %d) due to memory limit. Current " + + "allocation: %d", + rounded, requested, allocator.getAllocatedMemory()); + } else { + return String.format("Unable to allocate buffer of size %d due to memory limit. Current " + + "allocation: %d", + rounded, allocator.getAllocatedMemory()); + } + } + + /** + * Rounds up the provided value to the nearest power of two. + * + * @param val An integer value. + * @return The closest power of two of that value. + */ + static int nextPowerOfTwo(int val) { + int highestBit = Integer.highestOneBit(val); + if (highestBit == val) { + return val; + } else { + return highestBit << 1; + } + } + + public static StringBuilder indent(StringBuilder sb, int indent) { + final char[] indentation = new char[indent * 2]; + Arrays.fill(indentation, ' '); + sb.append(indentation); + return sb; + } + + public static boolean isDebug() { + return DEBUG; + } + @Override public void assertOpen() { if (AssertionUtil.ASSERT_ENABLED) { if (isClosed) { - throw new IllegalStateException("Attempting operation on allocator when allocator is closed.\n" + throw new IllegalStateException("Attempting operation on allocator when allocator is " + + "closed.\n" + toVerboseString()); } } @@ -136,7 +175,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } /** - * For debug/verification purposes only. Allows an AllocationManager to tell the allocator that we have a new ledger + * For debug/verification purposes only. Allows an AllocationManager to tell the allocator that + * we have a new ledger * associated with this allocator. */ void associateLedger(BufferLedger ledger) { @@ -149,7 +189,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } /** - * For debug/verification purposes only. Allows an AllocationManager to tell the allocator that we are removing a + * For debug/verification purposes only. Allows an AllocationManager to tell the allocator that + * we are removing a * ledger associated with this allocator */ void dissociateLedger(BufferLedger ledger) { @@ -167,8 +208,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato /** * Track when a ChildAllocator of this BaseAllocator is closed. Used for debugging purposes. * - * @param childAllocator - * The child allocator that has been closed. + * @param childAllocator The child allocator that has been closed. */ private void childClosed(final BaseAllocator childAllocator) { assertOpen(); @@ -187,17 +227,6 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } } - private static String createErrorMsg(final BufferAllocator allocator, final int rounded, final int requested) { - if (rounded != requested) { - return String.format( - "Unable to allocate buffer of size %d (rounded from %d) due to memory limit. Current allocation: %d", - rounded, requested, allocator.getAllocatedMemory()); - } else { - return String.format("Unable to allocate buffer of size %d due to memory limit. Current allocation: %d", - rounded, allocator.getAllocatedMemory()); - } - } - @Override public ArrowBuf buffer(final int initialRequestSize) { assertOpen(); @@ -205,7 +234,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato return buffer(initialRequestSize, null); } - private ArrowBuf createEmpty(){ + private ArrowBuf createEmpty() { assertOpen(); return new ArrowBuf(new AtomicInteger(), null, AllocationManager.EMPTY, null, null, 0, 0, true); @@ -221,7 +250,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato return empty; } - // round to next largest power of two if we're within a chunk since that is how our allocator operates + // round to next largest power of two if we're within a chunk since that is how our allocator + // operates final int actualRequestSize = initialRequestSize < AllocationManager.CHUNK_SIZE ? nextPowerOfTwo(initialRequestSize) : initialRequestSize; @@ -245,10 +275,12 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } /** - * Used by usual allocation as well as for allocating a pre-reserved buffer. Skips the typical accounting associated + * Used by usual allocation as well as for allocating a pre-reserved buffer. Skips the typical + * accounting associated * with creating a new buffer. */ - private ArrowBuf bufferWithoutReservation(final int size, BufferManager bufferManager) throws OutOfMemoryException { + private ArrowBuf bufferWithoutReservation(final int size, BufferManager bufferManager) throws + OutOfMemoryException { assertOpen(); final AllocationManager manager = new AllocationManager(this, size); @@ -274,185 +306,20 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato final long maxAllocation) { assertOpen(); - final ChildAllocator childAllocator = new ChildAllocator(this, name, initReservation, maxAllocation); + final ChildAllocator childAllocator = new ChildAllocator(this, name, initReservation, + maxAllocation); if (DEBUG) { synchronized (DEBUG_LOCK) { childAllocators.put(childAllocator, childAllocator); - historicalLog.recordEvent("allocator[%s] created new child allocator[%s]", name, childAllocator.name); + historicalLog.recordEvent("allocator[%s] created new child allocator[%s]", name, + childAllocator.name); } } return childAllocator; } - public class Reservation implements AllocationReservation { - private int nBytes = 0; - private boolean used = false; - private boolean closed = false; - private final HistoricalLog historicalLog; - - public Reservation() { - if (DEBUG) { - historicalLog = new HistoricalLog("Reservation[allocator[%s], %d]", name, System.identityHashCode(this)); - historicalLog.recordEvent("created"); - synchronized (DEBUG_LOCK) { - reservations.put(this, this); - } - } else { - historicalLog = null; - } - } - - @Override - public boolean add(final int nBytes) { - assertOpen(); - - Preconditions.checkArgument(nBytes >= 0, "nBytes(%d) < 0", nBytes); - Preconditions.checkState(!closed, "Attempt to increase reservation after reservation has been closed"); - Preconditions.checkState(!used, "Attempt to increase reservation after reservation has been used"); - - // we round up to next power of two since all reservations are done in powers of two. This may overestimate the - // preallocation since someone may perceive additions to be power of two. If this becomes a problem, we can look - // at - // modifying this behavior so that we maintain what we reserve and what the user asked for and make sure to only - // round to power of two as necessary. - final int nBytesTwo = BaseAllocator.nextPowerOfTwo(nBytes); - if (!reserve(nBytesTwo)) { - return false; - } - - this.nBytes += nBytesTwo; - return true; - } - - @Override - public ArrowBuf allocateBuffer() { - assertOpen(); - - Preconditions.checkState(!closed, "Attempt to allocate after closed"); - Preconditions.checkState(!used, "Attempt to allocate more than once"); - - final ArrowBuf arrowBuf = allocate(nBytes); - used = true; - return arrowBuf; - } - - @Override - public int getSize() { - return nBytes; - } - - @Override - public boolean isUsed() { - return used; - } - - @Override - public boolean isClosed() { - return closed; - } - - @Override - public void close() { - assertOpen(); - - if (closed) { - return; - } - - if (DEBUG) { - if (!isClosed()) { - final Object object; - synchronized (DEBUG_LOCK) { - object = reservations.remove(this); - } - if (object == null) { - final StringBuilder sb = new StringBuilder(); - print(sb, 0, Verbosity.LOG_WITH_STACKTRACE); - logger.debug(sb.toString()); - throw new IllegalStateException( - String.format("Didn't find closing reservation[%d]", System.identityHashCode(this))); - } - - historicalLog.recordEvent("closed"); - } - } - - if (!used) { - releaseReservation(nBytes); - } - - closed = true; - } - - @Override - public boolean reserve(int nBytes) { - assertOpen(); - - final AllocationOutcome outcome = BaseAllocator.this.allocateBytes(nBytes); - - if (DEBUG) { - historicalLog.recordEvent("reserve(%d) => %s", nBytes, Boolean.toString(outcome.isOk())); - } - - return outcome.isOk(); - } - - /** - * Allocate the a buffer of the requested size. - * - *

- * The implementation of the allocator's inner class provides this. - * - * @param nBytes - * the size of the buffer requested - * @return the buffer, or null, if the request cannot be satisfied - */ - private ArrowBuf allocate(int nBytes) { - assertOpen(); - - boolean success = false; - - /* - * The reservation already added the requested bytes to the allocators owned and allocated bytes via reserve(). - * This ensures that they can't go away. But when we ask for the buffer here, that will add to the allocated bytes - * as well, so we need to return the same number back to avoid double-counting them. - */ - try { - final ArrowBuf arrowBuf = BaseAllocator.this.bufferWithoutReservation(nBytes, null); - - listener.onAllocation(nBytes); - if (DEBUG) { - historicalLog.recordEvent("allocate() => %s", String.format("ArrowBuf[%d]", arrowBuf.getId())); - } - success = true; - return arrowBuf; - } finally { - if (!success) { - releaseBytes(nBytes); - } - } - } - - /** - * Return the reservation back to the allocator without having used it. - * - * @param nBytes - * the size of the reservation - */ - private void releaseReservation(int nBytes) { - assertOpen(); - - releaseBytes(nBytes); - - if (DEBUG) { - historicalLog.recordEvent("releaseReservation(%d)", nBytes); - } - } - - } - @Override public AllocationReservation newReservation() { assertOpen(); @@ -460,7 +327,6 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato return new Reservation(); } - @Override public synchronized void close() { /* @@ -474,7 +340,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato isClosed = true; if (DEBUG) { - synchronized(DEBUG_LOCK) { + synchronized (DEBUG_LOCK) { verifyAllocator(); // are there outstanding child allocators? @@ -488,7 +354,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } throw new IllegalStateException( - String.format("Allocator[%s] closed with outstanding child allocators.\n%s", name, toString())); + String.format("Allocator[%s] closed with outstanding child allocators.\n%s", name, + toString())); } // are there outstanding buffers? @@ -501,7 +368,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato if (reservations.size() != 0) { throw new IllegalStateException( - String.format("Allocator[%s] closed with outstanding reservations (%d).\n%s", name, reservations.size(), + String.format("Allocator[%s] closed with outstanding reservations (%d).\n%s", name, + reservations.size(), toString())); } @@ -512,7 +380,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato final long allocated = getAllocatedMemory(); if (allocated > 0) { throw new IllegalStateException( - String.format("Memory was leaked by query. Memory leaked: (%d)\n%s", allocated, toString())); + String.format("Memory was leaked by query. Memory leaked: (%d)\n%s", allocated, + toString())); } // we need to release our memory to our parent before we tell it we've closed. @@ -543,7 +412,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } /** - * Provide a verbose string of the current allocator state. Includes the state of all child allocators, along with + * Provide a verbose string of the current allocator state. Includes the state of all child + * allocators, along with * historical logs of each object and including stacktraces. * * @return A Verbose string of current allocator state. @@ -560,47 +430,31 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } /** - * Rounds up the provided value to the nearest power of two. - * - * @param val - * An integer value. - * @return The closest power of two of that value. - */ - static int nextPowerOfTwo(int val) { - int highestBit = Integer.highestOneBit(val); - if (highestBit == val) { - return val; - } else { - return highestBit << 1; - } - } - - - /** * Verifies the accounting state of the allocator. Only works for DEBUG. * - * @throws IllegalStateException - * when any problems are found + * @throws IllegalStateException when any problems are found */ void verifyAllocator() { - final IdentityHashMap buffersSeen = new IdentityHashMap<>(); + final IdentityHashMap buffersSeen = new + IdentityHashMap<>(); verifyAllocator(buffersSeen); } /** * Verifies the accounting state of the allocator. Only works for DEBUG. - * *

- * This overload is used for recursive calls, allowing for checking that ArrowBufs are unique across all allocators + *

+ * This overload is used for recursive calls, allowing for checking that ArrowBufs are unique + * across all allocators * that are checked. *

* - * @param buffersSeen - * a map of buffers that have already been seen when walking a tree of allocators - * @throws IllegalStateException - * when any problems are found + * @param buffersSeen a map of buffers that have already been seen when walking a tree of + * allocators + * @throws IllegalStateException when any problems are found */ - private void verifyAllocator(final IdentityHashMap buffersSeen) { + private void verifyAllocator(final IdentityHashMap + buffersSeen) { // The remaining tests can only be performed if we're in debug mode. if (!DEBUG) { return; @@ -618,7 +472,8 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato /* * Verify my relationships with my descendants. * - * The sum of direct child allocators' owned memory must be <= my allocated memory; my allocated memory also + * The sum of direct child allocators' owned memory must be <= my allocated memory; my + * allocated memory also * includes ArrowBuf's directly allocated by me. */ long childTotal = 0; @@ -648,11 +503,13 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato final UnsafeDirectLittleEndian udle = ledger.getUnderlying(); /* - * Even when shared, ArrowBufs are rewrapped, so we should never see the same instance twice. + * Even when shared, ArrowBufs are rewrapped, so we should never see the same instance + * twice. */ final BaseAllocator otherOwner = buffersSeen.get(udle); if (otherOwner != null) { - throw new IllegalStateException("This allocator's ArrowBuf already owned by another allocator"); + throw new IllegalStateException("This allocator's ArrowBuf already owned by another " + + "allocator"); } buffersSeen.put(udle, this); @@ -713,12 +570,14 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato if (allocated2 != allocated) { throw new IllegalStateException(String.format( - "allocator[%s]: allocated t1 (%d) + allocated t2 (%d). Someone released memory while in verification.", + "allocator[%s]: allocated t1 (%d) + allocated t2 (%d). Someone released memory " + + "while in verification.", name, allocated, allocated2)); } throw new IllegalStateException(String.format( - "allocator[%s]: buffer space (%d) + prealloc space (%d) + child space (%d) != allocated (%d)", + "allocator[%s]: buffer space (%d) + prealloc space (%d) + child space (%d) != " + + "allocated (%d)", name, bufferTotal, reservedTotal, childTotal, allocated)); } } @@ -777,14 +636,6 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } } - - public static StringBuilder indent(StringBuilder sb, int indent) { - final char[] indentation = new char[indent * 2]; - Arrays.fill(indentation, ' '); - sb.append(indentation); - return sb; - } - public static enum Verbosity { BASIC(false, false), // only include basic information LOG(true, false), // include basic @@ -800,7 +651,179 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato } } - public static boolean isDebug() { - return DEBUG; + public class Reservation implements AllocationReservation { + + private final HistoricalLog historicalLog; + private int nBytes = 0; + private boolean used = false; + private boolean closed = false; + + public Reservation() { + if (DEBUG) { + historicalLog = new HistoricalLog("Reservation[allocator[%s], %d]", name, System + .identityHashCode(this)); + historicalLog.recordEvent("created"); + synchronized (DEBUG_LOCK) { + reservations.put(this, this); + } + } else { + historicalLog = null; + } + } + + @Override + public boolean add(final int nBytes) { + assertOpen(); + + Preconditions.checkArgument(nBytes >= 0, "nBytes(%d) < 0", nBytes); + Preconditions.checkState(!closed, "Attempt to increase reservation after reservation has " + + "been closed"); + Preconditions.checkState(!used, "Attempt to increase reservation after reservation has been" + + " used"); + + // we round up to next power of two since all reservations are done in powers of two. This + // may overestimate the + // preallocation since someone may perceive additions to be power of two. If this becomes a + // problem, we can look + // at + // modifying this behavior so that we maintain what we reserve and what the user asked for + // and make sure to only + // round to power of two as necessary. + final int nBytesTwo = BaseAllocator.nextPowerOfTwo(nBytes); + if (!reserve(nBytesTwo)) { + return false; + } + + this.nBytes += nBytesTwo; + return true; + } + + @Override + public ArrowBuf allocateBuffer() { + assertOpen(); + + Preconditions.checkState(!closed, "Attempt to allocate after closed"); + Preconditions.checkState(!used, "Attempt to allocate more than once"); + + final ArrowBuf arrowBuf = allocate(nBytes); + used = true; + return arrowBuf; + } + + @Override + public int getSize() { + return nBytes; + } + + @Override + public boolean isUsed() { + return used; + } + + @Override + public boolean isClosed() { + return closed; + } + + @Override + public void close() { + assertOpen(); + + if (closed) { + return; + } + + if (DEBUG) { + if (!isClosed()) { + final Object object; + synchronized (DEBUG_LOCK) { + object = reservations.remove(this); + } + if (object == null) { + final StringBuilder sb = new StringBuilder(); + print(sb, 0, Verbosity.LOG_WITH_STACKTRACE); + logger.debug(sb.toString()); + throw new IllegalStateException( + String.format("Didn't find closing reservation[%d]", System.identityHashCode + (this))); + } + + historicalLog.recordEvent("closed"); + } + } + + if (!used) { + releaseReservation(nBytes); + } + + closed = true; + } + + @Override + public boolean reserve(int nBytes) { + assertOpen(); + + final AllocationOutcome outcome = BaseAllocator.this.allocateBytes(nBytes); + + if (DEBUG) { + historicalLog.recordEvent("reserve(%d) => %s", nBytes, Boolean.toString(outcome.isOk())); + } + + return outcome.isOk(); + } + + /** + * Allocate the a buffer of the requested size. + *

+ *

+ * The implementation of the allocator's inner class provides this. + * + * @param nBytes the size of the buffer requested + * @return the buffer, or null, if the request cannot be satisfied + */ + private ArrowBuf allocate(int nBytes) { + assertOpen(); + + boolean success = false; + + /* + * The reservation already added the requested bytes to the allocators owned and allocated + * bytes via reserve(). + * This ensures that they can't go away. But when we ask for the buffer here, that will add + * to the allocated bytes + * as well, so we need to return the same number back to avoid double-counting them. + */ + try { + final ArrowBuf arrowBuf = BaseAllocator.this.bufferWithoutReservation(nBytes, null); + + listener.onAllocation(nBytes); + if (DEBUG) { + historicalLog.recordEvent("allocate() => %s", String.format("ArrowBuf[%d]", arrowBuf + .getId())); + } + success = true; + return arrowBuf; + } finally { + if (!success) { + releaseBytes(nBytes); + } + } + } + + /** + * Return the reservation back to the allocator without having used it. + * + * @param nBytes the size of the reservation + */ + private void releaseReservation(int nBytes) { + assertOpen(); + + releaseBytes(nBytes); + + if (DEBUG) { + historicalLog.recordEvent("releaseReservation(%d)", nBytes); + } + } + } } http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/BoundsChecking.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/BoundsChecking.java b/java/memory/src/main/java/org/apache/arrow/memory/BoundsChecking.java index 4e88c73..b0e9cd8 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/BoundsChecking.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/BoundsChecking.java @@ -6,21 +6,22 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *

When assertions are on, complains if there are any outstanding buffers; to avoid * that, release all buffers before the allocator is closed. */ @@ -87,19 +85,18 @@ public interface BufferAllocator extends AutoCloseable { public long getAllocatedMemory(); /** - * Set the maximum amount of memory this allocator is allowed to allocate. + * Return the current maximum limit this allocator imposes. * - * @param newLimit - * The new Limit to apply to allocations + * @return Limit in number of bytes. */ - public void setLimit(long newLimit); + public long getLimit(); /** - * Return the current maximum limit this allocator imposes. + * Set the maximum amount of memory this allocator is allowed to allocate. * - * @return Limit in number of bytes. + * @param newLimit The new Limit to apply to allocations */ - public long getLimit(); + public void setLimit(long newLimit); /** * Returns the peak amount of memory allocated from this allocator. @@ -118,25 +115,31 @@ public interface BufferAllocator extends AutoCloseable { public AllocationReservation newReservation(); /** - * Get a reference to the empty buffer associated with this allocator. Empty buffers are special because we don't - * worry about them leaking or managing reference counts on them since they don't actually point to any memory. + * Get a reference to the empty buffer associated with this allocator. Empty buffers are + * special because we don't + * worry about them leaking or managing reference counts on them since they don't actually + * point to any memory. */ public ArrowBuf getEmpty(); /** - * Return the name of this allocator. This is a human readable name that can help debugging. Typically provides + * Return the name of this allocator. This is a human readable name that can help debugging. + * Typically provides * coordinates about where this allocator was created */ public String getName(); /** - * Return whether or not this allocator (or one if its parents) is over its limits. In the case that an allocator is - * over its limit, all consumers of that allocator should aggressively try to addrss the overlimit situation. + * Return whether or not this allocator (or one if its parents) is over its limits. In the case + * that an allocator is + * over its limit, all consumers of that allocator should aggressively try to addrss the + * overlimit situation. */ public boolean isOverLimit(); /** - * Return a verbose string describing this allocator. If in DEBUG mode, this will also include relevant stacktraces + * Return a verbose string describing this allocator. If in DEBUG mode, this will also include + * relevant stacktraces * and historical logs for underlying objects * * @return A very verbose description of the allocator hierarchy. @@ -144,7 +147,8 @@ public interface BufferAllocator extends AutoCloseable { public String toVerboseString(); /** - * Asserts (using java assertions) that the provided allocator is currently open. If assertions are disabled, this is + * Asserts (using java assertions) that the provided allocator is currently open. If assertions + * are disabled, this is * a no-op. */ public void assertOpen(); http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/BufferManager.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/BufferManager.java b/java/memory/src/main/java/org/apache/arrow/memory/BufferManager.java index 8969434..2fe763e 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/BufferManager.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/BufferManager.java @@ -15,6 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. ******************************************************************************/ + package org.apache.arrow.memory; import io.netty.buffer.ArrowBuf; @@ -24,7 +25,7 @@ import io.netty.buffer.ArrowBuf; * re-allocation the old buffer will be freed. Managing a list of these buffers * prevents some parts of the system from needing to define a correct location * to place the final call to free them. - * + *

* The current uses of these types of buffers are within the pluggable components of Drill. * In UDFs, memory management should not be a concern. We provide access to re-allocatable * ArrowBufs to give UDF writers general purpose buffers we can account for. To prevent the need @@ -38,12 +39,9 @@ public interface BufferManager extends AutoCloseable { /** * Replace an old buffer with a new version at least of the provided size. Does not copy data. * - * @param old - * Old Buffer that the user is no longer going to use. - * @param newSize - * Size of new replacement buffer. - * @return - * A new version of the buffer. + * @param old Old Buffer that the user is no longer going to use. + * @param newSize Size of new replacement buffer. + * @return A new version of the buffer. */ public ArrowBuf replace(ArrowBuf old, int newSize); @@ -57,8 +55,7 @@ public interface BufferManager extends AutoCloseable { /** * Get a managed buffer of at least a certain size. * - * @param size - * The desired size + * @param size The desired size * @return A buffer */ public ArrowBuf getManagedBuffer(int size); http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/ChildAllocator.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/ChildAllocator.java b/java/memory/src/main/java/org/apache/arrow/memory/ChildAllocator.java index 11c9063..f9a6dc7 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/ChildAllocator.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/ChildAllocator.java @@ -6,15 +6,16 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *

Child allocators can only be created by the root, or other children, so * this class is package private.

*/ class ChildAllocator extends BaseAllocator { + /** * Constructor. * * @param parentAllocator parent allocator -- the one creating this child - * @param name the name of this child allocator + * @param name the name of this child allocator * @param initReservation initial amount of space to reserve (obtained from the parent) - * @param maxAllocation maximum amount of space that can be obtained from this allocator; - * note this includes direct allocations (via {@see BufferAllocator#buffer(int, int)} - * et al) and requests from descendant allocators. Depending on the allocation policy in - * force, even less memory may be available + * @param maxAllocation maximum amount of space that can be obtained from this allocator; note + * this includes direct allocations (via {@see BufferAllocator#buffer(int, + *int)} et al) and requests from descendant allocators. Depending on the + * allocation policy in force, even less memory may be available */ ChildAllocator( BaseAllocator parentAllocator, http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/OutOfMemoryException.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/OutOfMemoryException.java b/java/memory/src/main/java/org/apache/arrow/memory/OutOfMemoryException.java index 6ba0284..c36584c 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/OutOfMemoryException.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/OutOfMemoryException.java @@ -6,28 +6,31 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + *

+ * Unless required by applicable law or agreed to in writing, software distributed under the License + * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express + * or implied. See the License for the specific language governing permissions and limitations under + * the License. + *

+ * Memory Allocation, Account and Management + *

+ * See the README.md file in this directory for detailed information about Arrow's memory allocation + * subsystem. + *

+ * Memory Allocation, Account and Management + *

+ * See the README.md file in this directory for detailed information about Arrow's memory + * allocation subsystem. + *

+ * Memory Allocation, Account and Management + *

+ * See the README.md file in this directory for detailed information about Arrow's memory + * allocation subsystem. + *

+ * Memory Allocation, Account and Management + *

+ * See the README.md file in this directory for detailed information about Arrow's memory + * allocation subsystem. */ /** * Memory Allocation, Account and Management * - * See the README.md file in this directory for detailed information about Arrow's memory allocation subsystem. + * See the README.md file in this directory for detailed information about Arrow's memory + * allocation subsystem. * */ + package org.apache.arrow.memory; http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/util/AssertionUtil.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/util/AssertionUtil.java b/java/memory/src/main/java/org/apache/arrow/memory/util/AssertionUtil.java index 28d0785..710f572 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/util/AssertionUtil.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/util/AssertionUtil.java @@ -6,32 +6,33 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *

This form supports the specification of a limit that will limit the * number of historical entries kept (which keeps down the amount of memory * used). With the limit, the first entry made is always kept (under the @@ -70,12 +60,12 @@ public class HistoricalLog { * Each time a new entry is made, the oldest that is not the first is dropped. *

* - * @param limit the maximum number of historical entries that will be kept, - * not including the first entry made - * @param idStringFormat {@link String#format} format string that can be used - * to identify this object in a log. Including some kind of unique identifier - * that can be associated with the object instance is best. - * @param args for the format string, or nothing if none are required + * @param limit the maximum number of historical entries that will be kept, not including + * the first entry made + * @param idStringFormat {@link String#format} format string that can be used to identify this + * object in a log. Including some kind of unique identifier that can be + * associated with the object instance is best. + * @param args for the format string, or nothing if none are required */ public HistoricalLog(final int limit, final String idStringFormat, Object... args) { this.limit = limit; @@ -88,7 +78,7 @@ public class HistoricalLog { * at the time this is called. * * @param noteFormat {@link String#format} format string that describes the event - * @param args for the format string, or nothing if none are required + * @param args for the format string, or nothing if none are required */ public synchronized void recordEvent(final String noteFormat, Object... args) { final String note = String.format(noteFormat, args); @@ -114,22 +104,13 @@ public class HistoricalLog { } /** - * Write the history of this object to the given {@link StringBuilder}. The history - * includes the identifying string provided at construction time, and all the recorded - * events with their stack traces. - * - * @param sb {@link StringBuilder} to write to - * @param additional an extra string that will be written between the identifying - * information and the history; often used for a current piece of state - */ - - /** * * @param sb * @param indent * @param includeStackTrace */ - public synchronized void buildHistory(final StringBuilder sb, int indent, boolean includeStackTrace) { + public synchronized void buildHistory(final StringBuilder sb, int indent, boolean + includeStackTrace) { final char[] indentation = new char[indent]; final char[] innerIndentation = new char[indent + 2]; Arrays.fill(indentation, ' '); @@ -140,7 +121,6 @@ public class HistoricalLog { .append(idString) .append('\n'); - if (firstEvent != null) { sb.append(innerIndentation) .append(firstEvent.time) @@ -151,7 +131,7 @@ public class HistoricalLog { firstEvent.stackTrace.writeToBuilder(sb, indent + 2); } - for(final Event event : history) { + for (final Event event : history) { if (event == firstEvent) { continue; } @@ -171,6 +151,16 @@ public class HistoricalLog { } /** + * Write the history of this object to the given {@link StringBuilder}. The history + * includes the identifying string provided at construction time, and all the recorded + * events with their stack traces. + * + * @param sb {@link StringBuilder} to write to + * @param additional an extra string that will be written between the identifying + * information and the history; often used for a current piece of state + */ + + /** * Write the history of this object to the given {@link Logger}. The history * includes the identifying string provided at construction time, and all the recorded * events with their stack traces. @@ -182,4 +172,17 @@ public class HistoricalLog { buildHistory(sb, 0, true); logger.debug(sb.toString()); } + + private static class Event { + + private final String note; // the event text + private final StackTrace stackTrace; // where the event occurred + private final long time; + + public Event(final String note) { + this.note = note; + this.time = System.nanoTime(); + stackTrace = new StackTrace(); + } + } } http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/memory/src/main/java/org/apache/arrow/memory/util/StackTrace.java ---------------------------------------------------------------------- diff --git a/java/memory/src/main/java/org/apache/arrow/memory/util/StackTrace.java b/java/memory/src/main/java/org/apache/arrow/memory/util/StackTrace.java index 638c2fb..bb4ea6c 100644 --- a/java/memory/src/main/java/org/apache/arrow/memory/util/StackTrace.java +++ b/java/memory/src/main/java/org/apache/arrow/memory/util/StackTrace.java @@ -6,15 +6,16 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *

* Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.arrow.tools; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.IOException; -import java.io.OutputStream; +package org.apache.arrow.tools; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.memory.RootAllocator; @@ -29,6 +24,12 @@ import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.file.ArrowFileReader; import org.apache.arrow.vector.stream.ArrowStreamWriter; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStream; + /** * Converts an Arrow file to an Arrow stream. The file should be specified as the * first argument and the output is written to standard out. http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/tools/src/main/java/org/apache/arrow/tools/Integration.java ---------------------------------------------------------------------- diff --git a/java/tools/src/main/java/org/apache/arrow/tools/Integration.java b/java/tools/src/main/java/org/apache/arrow/tools/Integration.java index 5d4849c..453693d 100644 --- a/java/tools/src/main/java/org/apache/arrow/tools/Integration.java +++ b/java/tools/src/main/java/org/apache/arrow/tools/Integration.java @@ -16,15 +16,8 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.arrow.tools; -import java.io.File; -import java.io.FileInputStream; -import java.io.FileOutputStream; -import java.io.IOException; -import java.util.Arrays; -import java.util.Iterator; -import java.util.List; +package org.apache.arrow.tools; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.memory.RootAllocator; @@ -44,8 +37,25 @@ import org.apache.commons.cli.PosixParser; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.Arrays; +import java.util.Iterator; +import java.util.List; + public class Integration { private static final Logger LOGGER = LoggerFactory.getLogger(Integration.class); + private final Options options; + + Integration() { + this.options = new Options(); + this.options.addOption("a", "arrow", true, "arrow file"); + this.options.addOption("j", "json", true, "json file"); + this.options.addOption("c", "command", true, "command to execute: " + Arrays.toString(Command + .values())); + } public static void main(String[] args) { try { @@ -59,20 +69,61 @@ public class Integration { } } - private final Options options; + private static void fatalError(String message, Throwable e) { + System.err.println(message); + System.err.println(e.getMessage()); + LOGGER.error(message, e); + System.exit(1); + } + + private File validateFile(String type, String fileName, boolean shouldExist) { + if (fileName == null) { + throw new IllegalArgumentException("missing " + type + " file parameter"); + } + File f = new File(fileName); + if (shouldExist && (!f.exists() || f.isDirectory())) { + throw new IllegalArgumentException(type + " file not found: " + f.getAbsolutePath()); + } + if (!shouldExist && f.exists()) { + throw new IllegalArgumentException(type + " file already exists: " + f.getAbsolutePath()); + } + return f; + } + + void run(String[] args) throws ParseException, IOException { + CommandLineParser parser = new PosixParser(); + CommandLine cmd = parser.parse(options, args, false); + + + Command command = toCommand(cmd.getOptionValue("command")); + File arrowFile = validateFile("arrow", cmd.getOptionValue("arrow"), command.arrowExists); + File jsonFile = validateFile("json", cmd.getOptionValue("json"), command.jsonExists); + command.execute(arrowFile, jsonFile); + } + + private Command toCommand(String commandName) { + try { + return Command.valueOf(commandName); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Unknown command: " + commandName + " expected one of " + + Arrays.toString(Command.values())); + } + } enum Command { ARROW_TO_JSON(true, false) { @Override public void execute(File arrowFile, File jsonFile) throws IOException { - try(BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE); - FileInputStream fileInputStream = new FileInputStream(arrowFile); - ArrowFileReader arrowReader = new ArrowFileReader(fileInputStream.getChannel(), allocator)) { + try (BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE); + FileInputStream fileInputStream = new FileInputStream(arrowFile); + ArrowFileReader arrowReader = new ArrowFileReader(fileInputStream.getChannel(), + allocator)) { VectorSchemaRoot root = arrowReader.getVectorSchemaRoot(); Schema schema = root.getSchema(); LOGGER.debug("Input file size: " + arrowFile.length()); LOGGER.debug("Found schema: " + schema); - try (JsonFileWriter writer = new JsonFileWriter(jsonFile, JsonFileWriter.config().pretty(true))) { + try (JsonFileWriter writer = new JsonFileWriter(jsonFile, JsonFileWriter.config() + .pretty(true))) { writer.start(schema); for (ArrowBlock rbBlock : arrowReader.getRecordBlocks()) { arrowReader.loadRecordBatch(rbBlock); @@ -94,7 +145,8 @@ public class Integration { try (FileOutputStream fileOutputStream = new FileOutputStream(arrowFile); VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator); // TODO json dictionaries - ArrowFileWriter arrowWriter = new ArrowFileWriter(root, null, fileOutputStream.getChannel())) { + ArrowFileWriter arrowWriter = new ArrowFileWriter(root, null, fileOutputStream + .getChannel())) { arrowWriter.start(); reader.read(root); while (root.getRowCount() != 0) { @@ -113,7 +165,8 @@ public class Integration { try (BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE); JsonFileReader jsonReader = new JsonFileReader(jsonFile, allocator); FileInputStream fileInputStream = new FileInputStream(arrowFile); - ArrowFileReader arrowReader = new ArrowFileReader(fileInputStream.getChannel(), allocator)) { + ArrowFileReader arrowReader = new ArrowFileReader(fileInputStream.getChannel(), + allocator)) { Schema jsonSchema = jsonReader.start(); VectorSchemaRoot arrowRoot = arrowReader.getVectorSchemaRoot(); Schema arrowSchema = arrowRoot.getSchema(); @@ -135,7 +188,8 @@ public class Integration { boolean hasMoreJSON = jsonRoot != null; boolean hasMoreArrow = iterator.hasNext(); if (hasMoreJSON || hasMoreArrow) { - throw new IllegalArgumentException("Unexpected RecordBatches. J:" + hasMoreJSON + " A:" + hasMoreArrow); + throw new IllegalArgumentException("Unexpected RecordBatches. J:" + hasMoreJSON + " " + + "A:" + hasMoreArrow); } } } @@ -153,51 +207,4 @@ public class Integration { } - Integration() { - this.options = new Options(); - this.options.addOption("a", "arrow", true, "arrow file"); - this.options.addOption("j", "json", true, "json file"); - this.options.addOption("c", "command", true, "command to execute: " + Arrays.toString(Command.values())); - } - - private File validateFile(String type, String fileName, boolean shouldExist) { - if (fileName == null) { - throw new IllegalArgumentException("missing " + type + " file parameter"); - } - File f = new File(fileName); - if (shouldExist && (!f.exists() || f.isDirectory())) { - throw new IllegalArgumentException(type + " file not found: " + f.getAbsolutePath()); - } - if (!shouldExist && f.exists()) { - throw new IllegalArgumentException(type + " file already exists: " + f.getAbsolutePath()); - } - return f; - } - - void run(String[] args) throws ParseException, IOException { - CommandLineParser parser = new PosixParser(); - CommandLine cmd = parser.parse(options, args, false); - - - Command command = toCommand(cmd.getOptionValue("command")); - File arrowFile = validateFile("arrow", cmd.getOptionValue("arrow"), command.arrowExists); - File jsonFile = validateFile("json", cmd.getOptionValue("json"), command.jsonExists); - command.execute(arrowFile, jsonFile); - } - - private Command toCommand(String commandName) { - try { - return Command.valueOf(commandName); - } catch (IllegalArgumentException e) { - throw new IllegalArgumentException("Unknown command: " + commandName + " expected one of " + Arrays.toString(Command.values())); - } - } - - private static void fatalError(String message, Throwable e) { - System.err.println(message); - System.err.println(e.getMessage()); - LOGGER.error(message, e); - System.exit(1); - } - } http://git-wip-us.apache.org/repos/asf/arrow/blob/a9a57013/java/tools/src/main/java/org/apache/arrow/tools/StreamToFile.java ---------------------------------------------------------------------- diff --git a/java/tools/src/main/java/org/apache/arrow/tools/StreamToFile.java b/java/tools/src/main/java/org/apache/arrow/tools/StreamToFile.java index 3b79d5b..41dfd34 100644 --- a/java/tools/src/main/java/org/apache/arrow/tools/StreamToFile.java +++ b/java/tools/src/main/java/org/apache/arrow/tools/StreamToFile.java @@ -6,17 +6,24 @@ * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at - * + *

* http://www.apache.org/licenses/LICENSE-2.0 - * + *