Fix for ZOOKEEPER-3072

Making the throttle check before passing over the request to the next thread will prevent the possibility of throttling code running after unthrottle
apache · Jul 6, 2018 · e27f95a · e27f95a
1 parent 25fd549
commit e27f95a
Show file tree

Hide file tree

Showing 2 changed files with 105 additions and 1 deletion.
diff --git a/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java b/src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java
@@ -1124,6 +1124,7 @@ public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOE
             }
             return;
         } else {
+            cnxn.incrOutstandingRequests(h);
             if (h.getType() == OpCode.sasl) {
                 Record rsp = processSasl(incomingBuffer,cnxn);
                 ReplyHeader rh = new ReplyHeader(h.getXid(), 0, KeeperException.Code.OK.intValue());
@@ -1140,7 +1141,6 @@ public void processPacket(ServerCnxn cnxn, ByteBuffer incomingBuffer) throws IOE
                 submitRequest(si);
             }
         }
-        cnxn.incrOutstandingRequests(h);
     }
 
     private Record processSasl(ByteBuffer incomingBuffer, ServerCnxn cnxn) throws IOException {

diff --git a/src/java/test/org/apache/zookeeper/test/AsyncHammerTest.java b/src/java/test/org/apache/zookeeper/test/AsyncHammerTest.java
@@ -31,6 +31,7 @@
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.ZKTestCase;
 import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.AsyncCallback.StatCallback;
 import org.apache.zookeeper.AsyncCallback.StringCallback;
 import org.apache.zookeeper.AsyncCallback.VoidCallback;
 import org.apache.zookeeper.ZooDefs.Ids;
@@ -165,6 +166,109 @@ public void processResult(int rc, String path, Object ctx) {
         }
     }
 
+    /**
+     * Send exists /zookeeper requests asynchronously, max 30 outstanding
+     */
+    class HammerThreadExists extends Thread implements StatCallback {
+        private static final int MAX_OUTSTANDING = 30;
+
+        private TestableZooKeeper zk;
+        private int outstanding;
+
+        private volatile boolean failed = false;
+
+        public HammerThreadExists(String name) {
+            super(name);
+        }
+
+        public void run() {
+            try {
+                CountdownWatcher watcher = new CountdownWatcher();
+                zk = new TestableZooKeeper(qb.hostPort, CONNECTION_TIMEOUT,
+                        watcher);
+                watcher.waitForConnected(CONNECTION_TIMEOUT);
+                while(bang) {
+                    incOutstanding(); // before create otw race
+                    zk.exists("/zookeeper", false, this, null);
+                }
+            } catch (InterruptedException e) {
+                if (bang) {
+                    LOG.error("sanity check Assert.failed!!!"); // sanity check
+                    return;
+                }
+            } catch (Exception e) {
+                LOG.error("Client create operation Assert.failed", e);
+                return;
+            } finally {
+                if (zk != null) {
+                    try {
+                        if (!zk.close(CONNECTION_TIMEOUT)) {
+                            failed = true;
+                            LOG.error("Client did not shutdown");
+                        }
+                    } catch (InterruptedException e) {
+                        LOG.info("Interrupted", e);
+                    }
+                }
+            }
+        }
+
+        private synchronized void incOutstanding() throws InterruptedException {
+            outstanding++;
+            while(outstanding > MAX_OUTSTANDING) {
+                wait();
+            }
+        }
+
+        private synchronized void decOutstanding() {
+            outstanding--;
+            Assert.assertTrue("outstanding >= 0", outstanding >= 0);
+            notifyAll();
+        }
+
+        public void process(WatchedEvent event) {
+            // ignore for purposes of this test
+        }
+
+        public void processResult(int rc, String path, Object ctx, Stat stat) {
+            if (rc != KeeperException.Code.OK.intValue()) {
+                if (bang) {
+                    failed = true;
+                    LOG.error("Exists Assert.failed for 0x"
+                            + Long.toHexString(zk.getSessionId())
+                            + "with rc:" + rc + " path:" + path);
+                }
+                decOutstanding();
+                return;
+            }
+            decOutstanding();
+        }
+    }
+
+    @Test
+    public void testExistsHammer() throws Exception {
+        System.setProperty("zookeeper.globalOutstandingLimit", "1");
+        setUp(false);
+        bang = true;
+        LOG.info("Starting hammers");
+        HammerThreadExists[] hammers = new HammerThreadExists[100];
+        for (int i = 0; i < hammers.length; i++) {
+            hammers[i] = new HammerThreadExists("HammerThread-" + i);
+            hammers[i].start();
+        }
+        LOG.info("Started hammers");
+        Thread.sleep(30000); // allow the clients to run for max 30sec
+        bang = false;
+        LOG.info("Stopping hammers");
+        for (int i = 0; i < hammers.length; i++) {
+            hammers[i].interrupt();
+            verifyThreadTerminated(hammers[i], 60000);
+            Assert.assertFalse(hammers[i].failed);
+        }
+        System.setProperty("zookeeper.globalOutstandingLimit", "1000");
+        tearDown();
+    }
+
     @Test
     public void testHammer() throws Exception {
         setUp(false);