public static final int CORE_NOTIFICATION_THREADS = 4;
public static final int MAX_NOTIFICATION_THREADS = 32;
+ // block caller thread after MAX_NOTIFICATION_THREADS + MAX_NOTIFICATION_QUEUE_SIZE pending notifications
+ public static final int MAX_NOTIFICATION_QUEUE_SIZE = 10;
public static final int NOTIFICATION_THREAD_LIFE = 15;
private static ListeningExecutorService NOTIFICATION_EXECUTOR = null;
public static synchronized final ListeningExecutorService getDefaultNotificationExecutor() {
if (NOTIFICATION_EXECUTOR == null) {
- // Overriding the queue since we need an unbounded queue
- // and threadpoolexecutor would not create new threads if the queue is not full
- BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>() {
+ // Overriding the queue:
+ // ThreadPoolExecutor would not create new threads if the queue is not full, thus adding
+ // occurs in RejectedExecutionHandler.
+ // This impl saturates threadpool first, then queue. When both are full caller will get blocked.
+ BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(MAX_NOTIFICATION_QUEUE_SIZE) {
@Override
public boolean offer(Runnable r) {
- if (size() <= 1) {
- // if the queue is empty (or has just 1), no need to rampup the threads
- return super.offer(r);
- } else {
- // if the queue is not empty, force the queue to return false.
- // threadpoolexecutor will spawn a new thread if the queue.offer returns false.
- return false;
- }
+ // ThreadPoolExecutor will spawn a new thread after core size is reached only if the queue.offer returns false.
+ return false;
}
};
try {
executor.getQueue().put(r);
} catch (InterruptedException e) {
- e.printStackTrace();
+ Thread.currentThread().interrupt();// set interrupt flag after clearing
+ throw new IllegalStateException(e);
}
}
});
--- /dev/null
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.sal.binding.codegen.impl;
+
+import com.google.common.util.concurrent.ListeningExecutorService;
+import java.lang.reflect.Field;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Ignore
+public class SingletonHolderTest {
+ private static final Logger logger = LoggerFactory.getLogger(SingletonHolderTest.class);
+
+ @Test
+ public void testNotificationExecutor() throws Exception {
+ ListeningExecutorService executor = SingletonHolder.getDefaultNotificationExecutor();
+ ThreadPoolExecutor tpExecutor = (ThreadPoolExecutor) setAccessible(executor.getClass().getDeclaredField("delegate")).get(executor);
+ BlockingQueue<Runnable> queue = tpExecutor.getQueue();
+
+ for (int idx = 0; idx < 100; idx++) {
+ final int idx2 = idx;
+ logger.info("Adding {}\t{}\t{}", idx, queue.size(), tpExecutor.getActiveCount());
+ executor.execute(new Runnable() {
+
+ @Override
+ public void run() {
+ logger.info("in {}", idx2);
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ logger.info("out {}", idx2);
+ }
+ });
+ }
+ executor.shutdown();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ }
+
+ private static Field setAccessible(Field field) {
+ field.setAccessible(true);
+ return field;
+ }
+}