cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From alek...@apache.org
Subject [1/3] cassandra git commit: Introduce backpressure for hints
Date Tue, 23 Feb 2016 15:33:26 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-3.0 fe37e0644 -> 037d24efd
  refs/heads/trunk 4b27287cd -> fc9c6faa2


Introduce backpressure for hints

patch by Ariel Weisberg; reviewed by Benedict Elliott Smith for
CASSANDRA-10972


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/037d24ef
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/037d24ef
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/037d24ef

Branch: refs/heads/cassandra-3.0
Commit: 037d24efdf83bd2736556f9880c5e1f6be48fa77
Parents: fe37e06
Author: Ariel Weisberg <ariel.weisberg@datastax.com>
Authored: Mon Dec 28 16:32:05 2015 -0500
Committer: Aleksey Yeschenko <aleksey@apache.org>
Committed: Tue Feb 23 15:28:41 2016 +0000

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 build.xml                                       | 14 +++-
 .../apache/cassandra/hints/HintsBufferPool.java | 34 ++++++---
 .../cassandra/hints/HintsWriteExecutor.java     |  3 +-
 .../cassandra/hints/HintsBufferPoolTest.java    | 75 ++++++++++++++++++++
 .../apache/cassandra/hints/HintsBufferTest.java |  2 +-
 6 files changed, 114 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/037d24ef/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index da91594..a675016 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.4
+ * Introduce backpressure for hints (CASSANDRA-10972)
  * Fix ClusteringPrefix not being able to read tombstone range boundaries (CASSANDRA-11158)
  * Prevent logging in sandboxed state (CASSANDRA-11033)
  * Disallow drop/alter operations of UDTs used by UDAs (CASSANDRA-10721)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/037d24ef/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index d27b77a..6ef99fd 100644
--- a/build.xml
+++ b/build.xml
@@ -111,6 +111,8 @@
     <property name="jacoco.finalexecfile" value="${jacoco.export.dir}/jacoco.exec" />
     <property name="jacoco.version" value="0.7.5.201505241946"/>
 
+    <property name="byteman.version" value="3.0.3"/>
+
     <property name="ecj.version" value="4.4.2"/>
 
     <condition property="maven-ant-tasks.jar.exists">
@@ -382,6 +384,11 @@
           <dependency groupId="org.jacoco" artifactId="org.jacoco.agent" version="${jacoco.version}"/>
           <dependency groupId="org.jacoco" artifactId="org.jacoco.ant" version="${jacoco.version}"/>
 
+          <dependency groupId="org.jboss.byteman" artifactId="byteman" version="${byteman.version}"/>
+          <dependency groupId="org.jboss.byteman" artifactId="byteman-submit" version="${byteman.version}"/>
+          <dependency groupId="org.jboss.byteman" artifactId="byteman-bmunit" version="${byteman.version}"/>
+
+
           <dependency groupId="org.openjdk.jmh" artifactId="jmh-core" version="1.1.1"/>
           <dependency groupId="org.openjdk.jmh" artifactId="jmh-generator-annprocess"
version="1.1.1"/>
 
@@ -479,7 +486,10 @@
                 artifactId="cassandra-parent"
                 version="${version}"/>
         <dependency groupId="org.jacoco" artifactId="org.jacoco.agent"/>
-        <dependency groupId="org.jacoco" artifactId="org.jacoco.ant"/>
+        <dependency groupId="org.jacoco" artifactId="org.jacoco.ant" />
+        <dependency groupId="org.jboss.byteman" artifactId="byteman"/>
+        <dependency groupId="org.jboss.byteman" artifactId="byteman-submit"/>
+        <dependency groupId="org.jboss.byteman" artifactId="byteman-bmunit"/>
       </artifact:pom>
 
       <artifact:pom id="test-deps-pom"
@@ -1165,6 +1175,7 @@
         <jvmarg value="-Dcassandra.keepBriefBrief=${cassandra.keepBriefBrief}" />
 	<optjvmargs/>
         <classpath>
+          <pathelement path="${java.class.path}"/>
           <path refid="cassandra.classpath" />
           <pathelement location="${test.classes}"/>
           <pathelement location="${test.conf}"/>
@@ -1701,6 +1712,7 @@
   <classpathentry kind="output" path="build/classes/main"/>
   <classpathentry kind="lib" path="build/classes/thrift" sourcepath="interface/thrift/gen-java/"/>
   <classpathentry kind="lib" path="test/conf"/>
+  <classpathentry kind="lib" path="${java.home}/../lib/tools.jar"/>
 ]]>
 	</echo>	  
   	<path id="eclipse-project-libs-path">

http://git-wip-us.apache.org/repos/asf/cassandra/blob/037d24ef/src/java/org/apache/cassandra/hints/HintsBufferPool.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hints/HintsBufferPool.java b/src/java/org/apache/cassandra/hints/HintsBufferPool.java
index 83b155a..25f9bc1 100644
--- a/src/java/org/apache/cassandra/hints/HintsBufferPool.java
+++ b/src/java/org/apache/cassandra/hints/HintsBufferPool.java
@@ -17,10 +17,11 @@
  */
 package org.apache.cassandra.hints;
 
-import java.util.Queue;
 import java.util.UUID;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
 
+import org.apache.cassandra.config.Config;
 import org.apache.cassandra.net.MessagingService;
 
 /**
@@ -34,15 +35,16 @@ final class HintsBufferPool
         void flush(HintsBuffer buffer, HintsBufferPool pool);
     }
 
+    static final int MAX_ALLOCATED_BUFFERS = Integer.getInteger(Config.PROPERTY_PREFIX +
"MAX_HINT_BUFFERS", 3);
     private volatile HintsBuffer currentBuffer;
-    private final Queue<HintsBuffer> reserveBuffers;
+    private final BlockingQueue<HintsBuffer> reserveBuffers;
     private final int bufferSize;
     private final FlushCallback flushCallback;
+    private int allocatedBuffers = 0;
 
     HintsBufferPool(int bufferSize, FlushCallback flushCallback)
     {
-        reserveBuffers = new ConcurrentLinkedQueue<>();
-
+        reserveBuffers = new LinkedBlockingQueue<>();
         this.bufferSize = bufferSize;
         this.flushCallback = flushCallback;
     }
@@ -78,13 +80,10 @@ final class HintsBufferPool
         }
     }
 
-    boolean offer(HintsBuffer buffer)
+    void offer(HintsBuffer buffer)
     {
-        if (!reserveBuffers.isEmpty())
-            return false;
-
-        reserveBuffers.offer(buffer);
-        return true;
+        if (!reserveBuffers.offer(buffer))
+            throw new RuntimeException("Failed to store buffer");
     }
 
     // A wrapper to ensure a non-null currentBuffer value on the first call.
@@ -108,6 +107,18 @@ final class HintsBufferPool
             return false;
 
         HintsBuffer buffer = reserveBuffers.poll();
+        if (buffer == null && allocatedBuffers >= MAX_ALLOCATED_BUFFERS)
+        {
+            try
+            {
+                //This BlockingQueue.take is a target for byteman in HintsBufferPoolTest
+                buffer = reserveBuffers.take();
+            }
+            catch (InterruptedException e)
+            {
+                throw new RuntimeException(e);
+            }
+        }
         currentBuffer = buffer == null ? createBuffer() : buffer;
 
         return true;
@@ -115,6 +126,7 @@ final class HintsBufferPool
 
     private HintsBuffer createBuffer()
     {
+        allocatedBuffers++;
         return HintsBuffer.create(bufferSize);
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/037d24ef/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java b/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java
index 1a69a4f..eb1bffe 100644
--- a/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java
+++ b/src/java/org/apache/cassandra/hints/HintsWriteExecutor.java
@@ -133,8 +133,7 @@ final class HintsWriteExecutor
             finally
             {
                 HintsBuffer recycledBuffer = buffer.recycle();
-                if (!bufferPool.offer(recycledBuffer))
-                    recycledBuffer.free();
+                bufferPool.offer(recycledBuffer);
             }
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/037d24ef/test/unit/org/apache/cassandra/hints/HintsBufferPoolTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/hints/HintsBufferPoolTest.java b/test/unit/org/apache/cassandra/hints/HintsBufferPoolTest.java
new file mode 100644
index 0000000..3fb912e
--- /dev/null
+++ b/test/unit/org/apache/cassandra/hints/HintsBufferPoolTest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * 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.cassandra.hints;
+
+import org.apache.cassandra.Util;
+import org.jboss.byteman.contrib.bmunit.BMRule;
+import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import com.google.common.collect.ImmutableList;
+
+import static junit.framework.Assert.*;
+
+import java.lang.Thread.State;
+import java.util.Queue;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+@RunWith(BMUnitRunner.class)
+public class HintsBufferPoolTest
+{
+
+    @BeforeClass
+    public static void defineSchema()
+    {
+        HintsBufferTest.defineSchema();
+    }
+
+    /*
+     * Check that the hints buffer pool will only drain a limited number of buffers
+     */
+    static volatile boolean blockedOnBackpressure = false;
+    @Test
+    @BMRule(name = "Greatest name in the world",
+            targetClass="HintsBufferPool",
+            targetMethod="switchCurrentBuffer",
+            targetLocation="AT INVOKE java.util.concurrent.BlockingQueue.take",
+            action="org.apache.cassandra.hints.HintsBufferPoolTest.blockedOnBackpressure
= true;")
+    public void testBackpressure() throws Exception
+    {
+        Queue<HintsBuffer> returnedBuffers = new ConcurrentLinkedQueue<>();
+        HintsBufferPool pool = new HintsBufferPool(256, (buffer, p) -> returnedBuffers.offer(buffer));
+
+        Thread blocked = new Thread(() -> {
+            for (int ii = 0; ii < 512; ii++)
+                pool.write(ImmutableList.of(UUID.randomUUID()), HintsBufferTest.createHint(ii,
ii));
+        });
+        blocked.start();
+
+        Util.spinAssertEquals(State.WAITING, () -> blocked.getState(), 1);
+
+        while (blocked.isAlive())
+            if (!returnedBuffers.isEmpty())
+                pool.offer(returnedBuffers.poll().recycle());
+
+        assertTrue(blockedOnBackpressure);
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/037d24ef/test/unit/org/apache/cassandra/hints/HintsBufferTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/hints/HintsBufferTest.java b/test/unit/org/apache/cassandra/hints/HintsBufferTest.java
index ebc333a..78ea4f4 100644
--- a/test/unit/org/apache/cassandra/hints/HintsBufferTest.java
+++ b/test/unit/org/apache/cassandra/hints/HintsBufferTest.java
@@ -188,7 +188,7 @@ public class HintsBufferTest
         return idx;
     }
 
-    private static Hint createHint(int idx, long baseTimestamp)
+    static Hint createHint(int idx, long baseTimestamp)
     {
         long timestamp = baseTimestamp + idx;
         return Hint.create(createMutation(idx, TimeUnit.MILLISECONDS.toMicros(timestamp)),
timestamp);


Mime
View raw message