Allow to configure BK for low latency busy-wait settings

bookkeeper-common/pom.xml

@@ -30,6 +30,11 @@
       <artifactId>bookkeeper-stats-api</artifactId>
       <version>${project.parent.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.bookkeeper</groupId>
+      <artifactId>cpu-affinity</artifactId>
+      <version>${project.parent.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>

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

@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -44,6 +45,8 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+import org.apache.bookkeeper.common.collections.BlockingMpscQueue;
+import org.apache.bookkeeper.common.util.affinity.CpuAffinity;
 import org.apache.bookkeeper.stats.Gauge;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.stats.OpStatsLogger;
@@ -68,6 +71,7 @@
 @Slf4j
 public class OrderedExecutor implements ExecutorService {
     public static final int NO_TASK_LIMIT = -1;
+    private static final int DEFAULT_MAX_ARRAY_QUEUE_SIZE = 10_000;
     protected static final long WARN_TIME_MICRO_SEC_DEFAULT = TimeUnit.SECONDS.toMicros(1);
 
     final String name;
@@ -80,7 +84,7 @@ public class OrderedExecutor implements ExecutorService {
     final boolean preserveMdcForTaskExecution;
     final long warnTimeMicroSec;
     final int maxTasksInQueue;
-
+    final boolean enableBusyWait;
 
     public static Builder newBuilder() {
         return new Builder();
@@ -98,7 +102,7 @@ public OrderedExecutor build() {
             }
             return new OrderedExecutor(name, numThreads, threadFactory, statsLogger,
                                            traceTaskExecution, preserveMdcForTaskExecution,
-                                           warnTimeMicroSec, maxTasksInQueue);
+                                           warnTimeMicroSec, maxTasksInQueue, enableBusyWait);
         }
     }
 
@@ -114,6 +118,7 @@ public abstract static class AbstractBuilder<T extends OrderedExecutor> {
         protected boolean preserveMdcForTaskExecution = false;
         protected long warnTimeMicroSec = WARN_TIME_MICRO_SEC_DEFAULT;
         protected int maxTasksInQueue = NO_TASK_LIMIT;
+        protected boolean enableBusyWait = false;
 
         public AbstractBuilder<T> name(String name) {
             this.name = name;
@@ -155,6 +160,11 @@ public AbstractBuilder<T> traceTaskWarnTimeMicroSec(long warnTimeMicroSec) {
             return this;
         }
 
+        public AbstractBuilder<T> enableBusyWait(boolean enableBusyWait) {
+            this.enableBusyWait = enableBusyWait;
+            return this;
+        }
+
         @SuppressWarnings("unchecked")
         public T build() {
             if (null == threadFactory) {
@@ -168,7 +178,8 @@ public T build() {
                 traceTaskExecution,
                 preserveMdcForTaskExecution,
                 warnTimeMicroSec,
-                maxTasksInQueue);
+                maxTasksInQueue,
+                enableBusyWait);
         }
     }
 
@@ -277,7 +288,15 @@ public T call() throws Exception {
     }
 
     protected ThreadPoolExecutor createSingleThreadExecutor(ThreadFactory factory) {
-        return new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), factory);
+        BlockingQueue<Runnable> queue;
+        if (enableBusyWait) {
+            // Use queue with busy-wait polling strategy
+            queue = new BlockingMpscQueue<>(maxTasksInQueue > 0 ? maxTasksInQueue : DEFAULT_MAX_ARRAY_QUEUE_SIZE);
+        } else {
+            // By default, use regular JDK LinkedBlockingQueue
+            queue = new LinkedBlockingQueue<>();
+        }
+        return new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, queue, factory);
     }
 
     protected ExecutorService getBoundedExecutor(ThreadPoolExecutor executor) {
@@ -361,25 +380,40 @@ public <T> Future<T> submit(Runnable task, T result) {
      */
     protected OrderedExecutor(String baseName, int numThreads, ThreadFactory threadFactory,
                                 StatsLogger statsLogger, boolean traceTaskExecution,
-                                boolean preserveMdcForTaskExecution, long warnTimeMicroSec, int maxTasksInQueue) {
+                                boolean preserveMdcForTaskExecution, long warnTimeMicroSec, int maxTasksInQueue,
+                                boolean enableBusyWait) {
         checkArgument(numThreads > 0);
         checkArgument(!StringUtils.isBlank(baseName));
 
         this.maxTasksInQueue = maxTasksInQueue;
         this.warnTimeMicroSec = warnTimeMicroSec;
+        this.enableBusyWait = enableBusyWait;
         name = baseName;
         threads = new ExecutorService[numThreads];
         threadIds = new long[numThreads];
         for (int i = 0; i < numThreads; i++) {
             ThreadPoolExecutor thread = createSingleThreadExecutor(
                     new ThreadFactoryBuilder().setNameFormat(name + "-" + getClass().getSimpleName() + "-" + i + "-%d")
                     .setThreadFactory(threadFactory).build());
+
             threads[i] = addExecutorDecorators(getBoundedExecutor(thread));
 
             final int idx = i;
             try {
                 threads[idx].submit(() -> {
                     threadIds[idx] = Thread.currentThread().getId();
+
+                    if (enableBusyWait) {
+                        // Try to acquire 1 CPU core to the executor thread. If it fails we
+                        // are just logging the error and continuing, falling back to
+                        // non-isolated CPUs.
+                        try {
+                            CpuAffinity.acquireCore();
+                        } catch (Throwable t) {
+                            log.warn("Failed to acquire CPU core for thread {}", Thread.currentThread().getName(),
+                                    t.getMessage(), t);
+                        }
+                    }
                 }).get();
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();

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

@@ -114,10 +114,9 @@ private OrderedScheduler(String baseName,
                                long warnTimeMicroSec,
                                int maxTasksInQueue) {
         super(baseName, numThreads, threadFactory, statsLogger, traceTaskExecution,
-                preserveMdcForTaskExecution, warnTimeMicroSec, maxTasksInQueue);
+                preserveMdcForTaskExecution, warnTimeMicroSec, maxTasksInQueue, false /* enableBusyWait */);
     }
 
-
     @Override
     protected ScheduledThreadPoolExecutor createSingleThreadExecutor(ThreadFactory factory) {
         return new ScheduledThreadPoolExecutor(1, factory);

bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java

@@ -40,14 +40,17 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.bookie.LedgerDirsManager.NoWritableLedgerDirException;
 import org.apache.bookkeeper.bookie.stats.JournalStats;
+import org.apache.bookkeeper.common.collections.BlockingMpscQueue;
 import org.apache.bookkeeper.common.collections.RecyclableArrayList;
+import org.apache.bookkeeper.common.util.affinity.CpuAffinity;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
 import org.apache.bookkeeper.stats.Counter;
@@ -56,7 +59,6 @@
 import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.bookkeeper.util.IOUtils;
 import org.apache.bookkeeper.util.MathUtils;
-import org.apache.bookkeeper.util.collections.GrowableArrayBlockingQueue;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -472,6 +474,15 @@ public ForceWriteThread(Thread threadToNotifyOnEx, boolean enableGroupForceWrite
         @Override
         public void run() {
             LOG.info("ForceWrite Thread started");
+
+            if (conf.isBusyWaitEnabled()) {
+                try {
+                    CpuAffinity.acquireCore();
+                } catch (Exception e) {
+                    LOG.warn("Unable to acquire CPU core for Journal ForceWrite thread: {}", e.getMessage(), e);
+                }
+            }
+
             boolean shouldForceWrite = true;
             int numReqInLastForceWrite = 0;
             while (running) {
@@ -611,8 +622,8 @@ static void writePaddingBytes(JournalChannel jc, ByteBuf paddingBuffer, int jour
     private final ExecutorService cbThreadPool;
 
     // journal entry queue to commit
-    final BlockingQueue<QueueEntry> queue = new GrowableArrayBlockingQueue<QueueEntry>();
-    final BlockingQueue<ForceWriteRequest> forceWriteRequests = new GrowableArrayBlockingQueue<ForceWriteRequest>();
+    final BlockingQueue<QueueEntry> queue;
+    final BlockingQueue<ForceWriteRequest> forceWriteRequests;
 
     volatile boolean running = true;
     private final LedgerDirsManager ledgerDirsManager;
@@ -628,6 +639,16 @@ public Journal(int journalIndex, File journalDirectory, ServerConfiguration conf
     public Journal(int journalIndex, File journalDirectory, ServerConfiguration conf,
             LedgerDirsManager ledgerDirsManager, StatsLogger statsLogger) {
         super("BookieJournal-" + conf.getBookiePort());
+
+        if (conf.isBusyWaitEnabled()) {
+            // To achieve lower latency, use busy-wait blocking queue implementation
+            queue = new BlockingMpscQueue<>(conf.getJournalQueueSize());
+            forceWriteRequests = new BlockingMpscQueue<>(conf.getJournalQueueSize());
+        } else {
+            queue = new ArrayBlockingQueue<>(conf.getJournalQueueSize());
+            forceWriteRequests = new ArrayBlockingQueue<>(conf.getJournalQueueSize());
+        }
+
         this.ledgerDirsManager = ledgerDirsManager;
         this.conf = conf;
         this.journalDirectory = journalDirectory;
@@ -906,6 +927,14 @@ public int getJournalQueueLength() {
     public void run() {
         LOG.info("Starting journal on {}", journalDirectory);
 
+        if (conf.isBusyWaitEnabled()) {
+            try {
+                CpuAffinity.acquireCore();
+            } catch (Exception e) {
+                LOG.warn("Unable to acquire CPU core for Journal thread: {}", e.getMessage(), e);
+            }
+        }
+
         RecyclableArrayList<QueueEntry> toFlush = entryListRecycler.newInstance();
         int numEntriesToFlush = 0;
         ByteBuf lenBuff = Unpooled.buffer(4);

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java

@@ -25,11 +25,11 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
 import io.netty.channel.EventLoopGroup;
-import io.netty.channel.epoll.EpollEventLoopGroup;
-import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.concurrent.DefaultThreadFactory;
+
 import java.io.IOException;
 import java.net.URI;
 import java.util.Collections;
@@ -39,9 +39,9 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+
 import org.apache.bookkeeper.client.AsyncCallback.CreateCallback;
 import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback;
 import org.apache.bookkeeper.client.AsyncCallback.IsClosedCallback;
@@ -78,9 +78,9 @@
 import org.apache.bookkeeper.proto.DataFormats;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.bookkeeper.util.EventLoopUtil;
 import org.apache.bookkeeper.util.SafeRunnable;
 import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.lang.SystemUtils;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZooKeeper;
 import org.slf4j.Logger;
@@ -404,6 +404,7 @@ public BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo
                 .traceTaskExecution(conf.getEnableTaskExecutionStats())
                 .preserveMdcForTaskExecution(conf.getPreserveMdcForTaskExecution())
                 .traceTaskWarnTimeMicroSec(conf.getTaskExecutionWarnTimeMicros())
+                .enableBusyWait(conf.isBusyWaitEnabled())
                 .build();
 
         // initialize stats logger
@@ -434,7 +435,8 @@ public BookKeeper(ClientConfiguration conf, ZooKeeper zk, EventLoopGroup eventLo
 
         // initialize event loop group
         if (null == eventLoopGroup) {
-            this.eventLoopGroup = getDefaultEventLoopGroup(conf);
+            this.eventLoopGroup = EventLoopUtil.getClientEventLoopGroup(conf,
+                    new DefaultThreadFactory("bookkeeper-io"));
             this.ownEventLoopGroup = true;
         } else {
             this.eventLoopGroup = eventLoopGroup;
@@ -601,8 +603,7 @@ BookieWatcher getBookieWatcher() {
         return bookieWatcher;
     }
 
-    @VisibleForTesting
-    OrderedExecutor getMainWorkerPool() {
+    public OrderedExecutor getMainWorkerPool() {
         return mainWorkerPool;
     }
 
@@ -1282,7 +1283,6 @@ public void asyncDeleteLedger(final long lId, final DeleteCallback cb, final Obj
      * @throws InterruptedException
      * @throws BKException
      */
-    @SuppressWarnings("unchecked")
     public void deleteLedger(long lId) throws InterruptedException, BKException {
         CompletableFuture<Void> future = new CompletableFuture<>();
         SyncDeleteCallback result = new SyncDeleteCallback(future);
@@ -1406,22 +1406,6 @@ public void close() throws BKException, InterruptedException {
         this.metadataDriver.close();
     }
 
-    static EventLoopGroup getDefaultEventLoopGroup(ClientConfiguration conf) {
-        ThreadFactory threadFactory = new DefaultThreadFactory("bookkeeper-io");
-        final int numThreads = conf.getNumIOThreads();
-
-        if (SystemUtils.IS_OS_LINUX) {
-            try {
-                return new EpollEventLoopGroup(numThreads, threadFactory);
-            } catch (Throwable t) {
-                LOG.warn("Could not use Netty Epoll event loop for bookie server: {}", t.getMessage());
-                return new NioEventLoopGroup(numThreads, threadFactory);
-            }
-        } else {
-            return new NioEventLoopGroup(numThreads, threadFactory);
-        }
-    }
-
     @Override
     public CreateBuilder newCreateLedgerOp() {
         return new LedgerCreateOp.CreateBuilderImpl(this);

bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/AbstractConfiguration.java

@@ -86,6 +86,8 @@ public abstract class AbstractConfiguration<T extends AbstractConfiguration>
             "storeSystemTimeAsLedgerUnderreplicatedMarkTime";
     protected static final String STORE_SYSTEMTIME_AS_LEDGER_CREATION_TIME = "storeSystemTimeAsLedgerCreationTime";
 
+    protected static final String ENABLE_BUSY_WAIT = "enableBusyWait";
+
     // Metastore settings, only being used when LEDGER_MANAGER_FACTORY_CLASS is MSLedgerManagerFactory
     protected static final String METASTORE_IMPL_CLASS = "metastoreImplClass";
     protected static final String METASTORE_MAX_ENTRIES_PER_SCAN = "metastoreMaxEntriesPerScan";
@@ -879,6 +881,42 @@ public T setPreserveMdcForTaskExecution(boolean enabled) {
         return getThis();
     }
 
+    /**
+     * Return whether the busy-wait is enabled for BookKeeper and Netty IO threads.
+     *
+     * <p>Default is false
+     *
+     * @return the value of the option
+     */
+    public boolean isBusyWaitEnabled() {
+        return getBoolean(ENABLE_BUSY_WAIT, false);
+    }
+
+    /**
+     * Option to enable busy-wait settings.
+     *
+     * <p>Default is false.
+     *
+     * <p>WARNING: This option will enable spin-waiting on executors and IO threads
+     * in order to reduce latency during context switches. The spinning will
+     * consume 100% CPU even when bookie is not doing any work. It is
+     * recommended to reduce the number of threads in the main workers pool
+     * ({@link ClientConfiguration#setNumWorkerThreads(int)}) and Netty event
+     * loop {@link ClientConfiguration#setNumIOThreads(int)} to only have few
+     * CPU cores busy.
+     * </p>
+     *
+     * @param busyWaitEanbled
+     *            if enabled, use spin-waiting strategy to reduce latency in
+     *            context switches
+     *
+     * @see #isBusyWaitEnabled()
+     */
+    public T setBusyWaitEnabled(boolean busyWaitEanbled) {
+        setProperty(ENABLE_BUSY_WAIT, busyWaitEanbled);
+        return getThis();
+    }
+
     /**
      * Return the flag indicating whether to limit stats logging.
      *