X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-binding-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fbinding%2Fcodegen%2Fimpl%2FSingletonHolder.java;h=1ec4aa2d30bc9da7307891a41050bfc33e55642a;hp=4141bba2d44ec1ec13502c984efc36e282f4f912;hb=4fd83cbe09def7cc6c725cdbd8a09397eafff277;hpb=abca01a0016ee55b6cdcb71c81419c4b143bb63a diff --git a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/codegen/impl/SingletonHolder.java b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/codegen/impl/SingletonHolder.java index 4141bba2d4..1ec4aa2d30 100644 --- a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/codegen/impl/SingletonHolder.java +++ b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/codegen/impl/SingletonHolder.java @@ -7,23 +7,31 @@ */ package org.opendaylight.controller.sal.binding.codegen.impl; -import com.google.common.util.concurrent.ListeningExecutorService; -import com.google.common.util.concurrent.MoreExecutors; -import com.google.common.util.concurrent.ThreadFactoryBuilder; -import javassist.ClassPool; -import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator; -import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory; - import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import javassist.ClassPool; + +import org.apache.commons.lang3.StringUtils; +import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator; +import org.opendaylight.controller.sal.binding.spi.NotificationInvokerFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.util.concurrent.ForwardingBlockingQueue; +import com.google.common.util.concurrent.ListeningExecutorService; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.ThreadFactoryBuilder; + public class SingletonHolder { + private static final Logger logger = LoggerFactory.getLogger(SingletonHolder.class); public static final ClassPool CLASS_POOL = ClassPool.getDefault(); public static final org.opendaylight.controller.sal.binding.codegen.impl.RuntimeCodeGenerator RPC_GENERATOR_IMPL = new org.opendaylight.controller.sal.binding.codegen.impl.RuntimeCodeGenerator( @@ -34,8 +42,9 @@ public class SingletonHolder { 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 MAX_NOTIFICATION_QUEUE_SIZE = 1000; public static final int NOTIFICATION_THREAD_LIFE = 15; + private static final String NOTIFICATION_QUEUE_SIZE_PROPERTY = "mdsal.notificationqueue.size"; private static ListeningExecutorService NOTIFICATION_EXECUTOR = null; private static ListeningExecutorService COMMIT_EXECUTOR = null; @@ -46,37 +55,58 @@ public class SingletonHolder { * should use service injection to make the executor configurable. */ @Deprecated - public static synchronized final ListeningExecutorService getDefaultNotificationExecutor() { + public static synchronized ListeningExecutorService getDefaultNotificationExecutor() { if (NOTIFICATION_EXECUTOR == null) { + int queueSize = MAX_NOTIFICATION_QUEUE_SIZE; + String queueValue = System.getProperty(NOTIFICATION_QUEUE_SIZE_PROPERTY); + if (StringUtils.isNotBlank(queueValue)) { + try { + queueSize = Integer.parseInt(queueValue); + logger.trace("Queue size was set to {}", queueSize); + } catch (NumberFormatException e) { + logger.warn("Cannot parse {} as set by {}, using default {}", queueValue, + NOTIFICATION_QUEUE_SIZE_PROPERTY, queueSize); + } + } + // 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 queue = new LinkedBlockingQueue(MAX_NOTIFICATION_QUEUE_SIZE) { + final BlockingQueue delegate = new LinkedBlockingQueue<>(queueSize); + final BlockingQueue queue = new ForwardingBlockingQueue() { @Override - public boolean offer(Runnable r) { - // ThreadPoolExecutor will spawn a new thread after core size is reached only if the queue.offer returns false. + protected BlockingQueue delegate() { + return delegate; + } + + @Override + public boolean offer(final Runnable r) { + // ThreadPoolExecutor will spawn a new thread after core size is reached only + // if the queue.offer returns false. return false; } }; - ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-notification-%d").build(); + final ThreadFactory factory = new ThreadFactoryBuilder() + .setDaemon(true) + .setNameFormat("md-sal-binding-notification-%d") + .build(); - ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_NOTIFICATION_THREADS, MAX_NOTIFICATION_THREADS, - NOTIFICATION_THREAD_LIFE, TimeUnit.SECONDS, queue , factory, + final ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_NOTIFICATION_THREADS, MAX_NOTIFICATION_THREADS, + NOTIFICATION_THREAD_LIFE, TimeUnit.SECONDS, queue, factory, new RejectedExecutionHandler() { - // if the max threads are met, then it will raise a rejectedExecution. We then push to the queue. - @Override - public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { - try { - executor.getQueue().put(r); - } catch (InterruptedException e) { - Thread.currentThread().interrupt();// set interrupt flag after clearing - throw new IllegalStateException(e); - } - } - }); + // if the max threads are met, then it will raise a rejectedExecution. We then push to the queue. + @Override + public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) { + try { + executor.getQueue().put(r); + } catch (InterruptedException e) { + throw new RejectedExecutionException("Interrupted while waiting on the queue", e); + } + } + }); NOTIFICATION_EXECUTOR = MoreExecutors.listeningDecorator(executor); } @@ -89,7 +119,7 @@ public class SingletonHolder { * should use service injection to make the executor configurable. */ @Deprecated - public static synchronized final ListeningExecutorService getDefaultCommitExecutor() { + public static synchronized ListeningExecutorService getDefaultCommitExecutor() { if (COMMIT_EXECUTOR == null) { ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-commit-%d").build(); /*