Merge branch 'cassandra-2.1' into trunk

Conflicts:
	src/java/org/apache/cassandra/utils/memory/NativeAllocator.java

CHANGES.txt

@@ -40,6 +40,7 @@
 
 
 2.1.3
+ * Scale memtable slab allocation logarithmically (CASSANDRA-7882)
  * cassandra-stress simultaneous inserts over same seed (CASSANDRA-7964)
  * Reduce cassandra-stress sampling memory requirements (CASSANDRA-7926)
  * Ensure memtable flush cannot expire commit log entries from its future (CASSANDRA-8383)

src/java/org/apache/cassandra/utils/memory/NativeAllocator.java

@@ -18,8 +18,8 @@
 package org.apache.cassandra.utils.memory;
 
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.cassandra.config.CFMetaData;
@@ -36,25 +36,26 @@ import org.apache.cassandra.db.NativeDeletedCell;
 import org.apache.cassandra.db.NativeExpiringCell;
 import org.apache.cassandra.io.util.IAllocator;
 import org.apache.cassandra.utils.concurrent.OpOrder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class NativeAllocator extends MemtableAllocator
 {
-    private static final Logger logger = LoggerFactory.getLogger(NativeAllocator.class);
+    private final static int MAX_REGION_SIZE = 1 * 1024 * 1024;
-
-    private final static int REGION_SIZE = 1024 * 1024;
     private final static int MAX_CLONED_SIZE = 128 * 1024; // bigger than this don't go in the region
+    private final static int MIN_REGION_SIZE = 8 * 1024;
 
     private static final IAllocator allocator = DatabaseDescriptor.getoffHeapMemoryAllocator();
     
     // globally stash any Regions we allocate but are beaten to using, and use these up before allocating any more
-    private static final ConcurrentLinkedQueue<Region> RACE_ALLOCATED = new ConcurrentLinkedQueue<>();
+    private static final Map<Integer, RaceAllocated> RACE_ALLOCATED = new HashMap<>();
+
+    static
+    {
+        for(int i = MIN_REGION_SIZE ; i <= MAX_REGION_SIZE; i *= 2)
+            RACE_ALLOCATED.put(i, new RaceAllocated());
+    }
 
     private final AtomicReference<Region> currentRegion = new AtomicReference<>();
-    private final AtomicInteger regionCount = new AtomicInteger(0);
     private final ConcurrentLinkedQueue<Region> regions = new ConcurrentLinkedQueue<>();
-    private AtomicLong unslabbed = new AtomicLong(0);
 
     protected NativeAllocator(NativePool pool)
     {
@@ -99,35 +100,63 @@ public class NativeAllocator extends MemtableAllocator
     public long allocate(int size, OpOrder.Group opGroup)
     {
         assert size >= 0;
-        offHeap().allocate(size, opGroup);
-        // satisfy large allocations directly from JVM since they don't cause fragmentation
-        // as badly, and fill up our regions quickly
         if (size > MAX_CLONED_SIZE)
-        {
+            return allocateOversize(size, opGroup);
-            unslabbed.addAndGet(size);
-            Region region = new Region(allocator.allocate(size), size);
-            regions.add(region);
-
-            long peer;
-            if ((peer = region.allocate(size)) == -1)
-                throw new AssertionError();
-
-            return peer;
-        }
 
         while (true)
         {
-            Region region = getRegion();
+            Region region = currentRegion.get();
-
             long peer;
-            if ((peer = region.allocate(size)) > 0)
+            if (region != null && (peer = region.allocate(size)) > 0)
                 return peer;
 
-            // not enough space!
+            trySwapRegion(region, size);
-            currentRegion.compareAndSet(region, null);
         }
     }
 
+    private void trySwapRegion(Region current, int minSize)
+    {
+        // decide how big we want the new region to be:
+        //  * if there is no prior region, we set it to min size
+        //  * otherwise we double its size; if it's too small to fit the allocation, we round it up to 4-8x its size
+        int size;
+        if (current == null) size = MIN_REGION_SIZE;
+        else size = current.capacity * 2;
+        if (minSize > size)
+            size = Integer.highestOneBit(minSize) << 3;
+        size = Math.min(MAX_REGION_SIZE, size);
+
+        // first we try and repurpose a previously allocated region
+        RaceAllocated raceAllocated = RACE_ALLOCATED.get(size);
+        Region next = raceAllocated.poll();
+
+        // if there are none, we allocate one
+        if (next == null)
+            next = new Region(allocator.allocate(size), size);
+
+        // we try to swap in the region we've obtained;
+        // if we fail to swap the region, we try to stash it for repurposing later; if we're out of stash room, we free it
+        if (currentRegion.compareAndSet(current, next))
+            regions.add(next);
+        else if (!raceAllocated.stash(next))
+            allocator.free(next.peer);
+    }
+
+    private long allocateOversize(int size, OpOrder.Group opGroup)
+    {
+        // satisfy large allocations directly from JVM since they don't cause fragmentation
+        // as badly, and fill up our regions quickly
+        offHeap().allocate(size, opGroup);
+        Region region = new Region(allocator.allocate(size), size);
+        regions.add(region);
+
+        long peer;
+        if ((peer = region.allocate(size)) == -1)
+            throw new AssertionError();
+
+        return peer;
+    }
+
     public void setDiscarded()
     {
         for (Region region : regions)
@@ -135,34 +164,24 @@ public class NativeAllocator extends MemtableAllocator
         super.setDiscarded();
     }
 
-    /**
+    // used to ensure we don't keep loads of race allocated regions around indefinitely. keeps the total bound on wasted memory low.
-     * Get the current region, or, if there is no current region, allocate a new one
+    private static class RaceAllocated
-     */
-    private Region getRegion()
     {
-        while (true)
+        final ConcurrentLinkedQueue<Region> stash = new ConcurrentLinkedQueue<>();
+        final Semaphore permits = new Semaphore(8);
+        boolean stash(Region region)
         {
-            // Try to get the region
+            if (!permits.tryAcquire())
-            Region region = currentRegion.get();
+                return false;
-            if (region != null)
+            stash.add(region);
-                return region;
+            return true;
-
+        }
-            // No current region, so we want to allocate one. We race
+        Region poll()
-            // against other allocators to CAS in a Region, and if we fail we stash the region for re-use
+        {
-            region = RACE_ALLOCATED.poll();
+            Region next = stash.poll();
-            if (region == null)
+            if (next != null)
-                region = new Region(allocator.allocate(REGION_SIZE), REGION_SIZE);
+                permits.release();
-            if (currentRegion.compareAndSet(null, region))
+            return next;
-            {
-                regions.add(region);
-                regionCount.incrementAndGet();
-                logger.trace("{} regions now allocated in {}", regionCount, this);
-                return region;
-            }
-
-            // someone else won race - that's fine, we'll try to grab theirs
-            // in the next iteration of the loop.
-            RACE_ALLOCATED.add(region);
         }
     }
 
@@ -181,7 +200,7 @@ public class NativeAllocator extends MemtableAllocator
          */
         private final long peer;
 
-        private final long capacity;
+        private final int capacity;
 
         /**
          * Offset for the next allocation, or the sentinel value -1
@@ -200,7 +219,7 @@ public class NativeAllocator extends MemtableAllocator
          *
          * @param peer peer
          */
-        private Region(long peer, long capacity)
+        private Region(long peer, int capacity)
         {
             this.peer = peer;
             this.capacity = capacity;