Refactored OrderedSafeExecutor and OrderedScheduler

bookkeeper-benchmark/src/main/java/org/apache/bookkeeper/benchmark/BenchBookie.java

@@ -31,14 +31,14 @@
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.common.util.OrderedExecutor;
 import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.proto.BookieClient;
 import org.apache.bookkeeper.proto.BookieProtocol;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.util.ByteBufList;
-import org.apache.bookkeeper.util.OrderedSafeExecutor;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.HelpFormatter;
@@ -165,7 +165,7 @@ public static void main(String[] args)
             eventLoop = new NioEventLoopGroup();
         }
 
-        OrderedSafeExecutor executor = OrderedSafeExecutor.newBuilder()
+        OrderedExecutor executor = OrderedExecutor.newBuilder()
                 .name("BenchBookieClientScheduler")
                 .numThreads(1)
                 .build();

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/BoundedExecutorService.java

@@ -0,0 +1,109 @@
+/**
+ * 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.bookkeeper.common.util;
+
+import com.google.common.util.concurrent.ForwardingExecutorService;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Implements {@link ExecutorService} and allows limiting the number of tasks to
+ * be scheduled in the thread's queue.
+ */
+public class BoundedExecutorService extends ForwardingExecutorService {
+    private final BlockingQueue<Runnable> queue;
+    private final ThreadPoolExecutor thread;
+    private final int maxTasksInQueue;
+
+    public BoundedExecutorService(ThreadPoolExecutor thread, int maxTasksInQueue) {
+        this.queue = thread.getQueue();
+        this.thread = thread;
+        this.maxTasksInQueue = maxTasksInQueue;
+    }
+
+    @Override
+    protected ExecutorService delegate() {
+        return this.thread;
+    }
+
+    private void checkQueue(int numberOfTasks) {
+        if (maxTasksInQueue > 0 && (queue.size() + numberOfTasks) > maxTasksInQueue) {
+            throw new RejectedExecutionException("Queue at limit of " + maxTasksInQueue + " items");
+        }
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
+        checkQueue(tasks.size());
+        return super.invokeAll(tasks);
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+            throws InterruptedException {
+        checkQueue(tasks.size());
+        return super.invokeAll(tasks, timeout, unit);
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
+        checkQueue(tasks.size());
+        return super.invokeAny(tasks);
+    }
+
+    @Override
+    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
+            throws InterruptedException, ExecutionException, TimeoutException {
+        checkQueue(tasks.size());
+        return super.invokeAny(tasks, timeout, unit);
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        checkQueue(1);
+        super.execute(command);
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+        checkQueue(1);
+        return super.submit(task);
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+        checkQueue(1);
+        return super.submit(task);
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+        checkQueue(1);
+        return super.submit(task, result);
+    }
+}

bookkeeper-common/src/main/java/org/apache/bookkeeper/common/util/BoundedScheduledExecutorService.java

@@ -59,83 +59,83 @@ protected ListeningExecutorService delegate() {
 
     @Override
     public ListenableScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
-        this.checkQueue();
+        this.checkQueue(1);
         return this.thread.schedule(command, delay, unit);
     }
 
     @Override
     public <V> ListenableScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
-        this.checkQueue();
+        this.checkQueue(1);
         return this.thread.schedule(callable, delay, unit);
     }
 
     @Override
     public ListenableScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                             long initialDelay, long period, TimeUnit unit) {
-        this.checkQueue();
+        this.checkQueue(1);
         return this.thread.scheduleAtFixedRate(command, initialDelay, period, unit);
     }
 
     @Override
     public ListenableScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                                long initialDelay, long delay, TimeUnit unit) {
-        this.checkQueue();
+        this.checkQueue(1);
         return this.thread.scheduleAtFixedRate(command, initialDelay, delay, unit);
     }
 
     @Override
     public <T> ListenableFuture<T> submit(Callable<T> task) {
-        this.checkQueue();
+        this.checkQueue(1);
         return super.submit(task);
     }
 
     @Override
     public ListenableFuture<?> submit(Runnable task) {
-        this.checkQueue();
+        this.checkQueue(1);
         return super.submit(task);
     }
 
     @Override
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
-        this.checkQueue();
+        this.checkQueue(tasks.size());
         return super.invokeAll(tasks);
     }
 
     @Override
     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                          long timeout, TimeUnit unit) throws InterruptedException {
-        this.checkQueue();
+        this.checkQueue(tasks.size());
         return super.invokeAll(tasks, timeout, unit);
     }
 
     @Override
     public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
-        this.checkQueue();
+        this.checkQueue(tasks.size());
         return super.invokeAny(tasks);
     }
 
     @Override
     public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout,
                            TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-        this.checkQueue();
+        this.checkQueue(tasks.size());
         return super.invokeAny(tasks, timeout, unit);
     }
 
     @Override
     public <T> ListenableFuture<T> submit(Runnable task, T result) {
-        this.checkQueue();
+        this.checkQueue(1);
         return super.submit(task, result);
     }
 
     @Override
     public void execute(Runnable command) {
-        this.checkQueue();
+        this.checkQueue(1);
         super.execute(command);
     }
 
-    private void checkQueue() {
-        if (this.maxTasksInQueue > 0 && this.queue.size() >= this.maxTasksInQueue) {
-            throw new RejectedExecutionException("Queue at limit of " + this.maxTasksInQueue + " items");
+    private void checkQueue(int numberOfTasks) {
+        if (maxTasksInQueue > 0 && (queue.size() + numberOfTasks) > maxTasksInQueue) {
+            throw new RejectedExecutionException("Queue at limit of " + maxTasksInQueue + " items");
         }
     }