X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-binding-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fbinding%2Fcodegen%2Fimpl%2FSingletonHolder.java;h=6bcd74d213526b83c94b43087443e50ba9385da6;hb=412db94945c5db5d2da918f5e23bd3abcecc4d10;hp=cf754cc74e7c5b386c9b27209726d111f3ebb147;hpb=d7992a5c497b98387469ec24f99f3570f517b44f;p=controller.git 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 cf754cc74e..6bcd74d213 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,6 +7,7 @@ */ package org.opendaylight.controller.sal.binding.codegen.impl; +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; @@ -14,14 +15,14 @@ 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.opendaylight.yangtools.sal.binding.generator.util.JavassistUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -29,10 +30,7 @@ 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( - CLASS_POOL); - public static final RuntimeCodeGenerator RPC_GENERATOR = RPC_GENERATOR_IMPL; - public static final NotificationInvokerFactory INVOKER_FACTORY = RPC_GENERATOR_IMPL.getInvokerFactory(); + public static final JavassistUtils JAVASSIST = JavassistUtils.forClassPool(CLASS_POOL); public static final int CORE_NOTIFICATION_THREADS = 4; public static final int MAX_NOTIFICATION_THREADS = 32; @@ -54,44 +52,54 @@ public class SingletonHolder { if (NOTIFICATION_EXECUTOR == null) { int queueSize = MAX_NOTIFICATION_QUEUE_SIZE; - String queueValue = System.getProperty(NOTIFICATION_QUEUE_SIZE_PROPERTY); + final 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) { + } catch (final 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(queueSize) { + final BlockingQueue delegate = new LinkedBlockingQueue<>(queueSize); + final BlockingQueue queue = new ForwardingBlockingQueue() { + @Override + protected BlockingQueue delegate() { + return delegate; + } + @Override - public boolean offer(Runnable r) { - // ThreadPoolExecutor will spawn a new thread after core size is reached only if the queue.offer returns false. + 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 (final InterruptedException e) { + throw new RejectedExecutionException("Interrupted while waiting on the queue", e); + } + } + }); NOTIFICATION_EXECUTOR = MoreExecutors.listeningDecorator(executor); } @@ -106,7 +114,7 @@ public class SingletonHolder { @Deprecated public static synchronized ListeningExecutorService getDefaultCommitExecutor() { if (COMMIT_EXECUTOR == null) { - ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-commit-%d").build(); + final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-commit-%d").build(); /* * FIXME: this used to be newCacheThreadPool(), but MD-SAL does not have transaction * ordering guarantees, which means that using a concurrent threadpool results @@ -114,7 +122,7 @@ public class SingletonHolder { * in inconsistent data being present. Once proper primitives are introduced, * concurrency can be reintroduced. */ - ExecutorService executor = Executors.newSingleThreadExecutor(factory); + final ExecutorService executor = Executors.newSingleThreadExecutor(factory); COMMIT_EXECUTOR = MoreExecutors.listeningDecorator(executor); } @@ -123,7 +131,7 @@ public class SingletonHolder { public static ExecutorService getDefaultChangeEventExecutor() { if (CHANGE_EVENT_EXECUTOR == null) { - ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-change-%d").build(); + final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat("md-sal-binding-change-%d").build(); /* * FIXME: this used to be newCacheThreadPool(), but MD-SAL does not have transaction * ordering guarantees, which means that using a concurrent threadpool results @@ -131,7 +139,7 @@ public class SingletonHolder { * in inconsistent data being present. Once proper primitives are introduced, * concurrency can be reintroduced. */ - ExecutorService executor = Executors.newSingleThreadExecutor(factory); + final ExecutorService executor = Executors.newSingleThreadExecutor(factory); CHANGE_EVENT_EXECUTOR = MoreExecutors.listeningDecorator(executor); }