modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccess.java | 44 -- modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccessHandler.java | 201 ++++++++++ modules/core/native-system/src/test/java/org/rhq/core/system/SigarAccessHandlerTest.java | 172 ++++++++ 3 files changed, 377 insertions(+), 40 deletions(-)
New commits: commit fe38a28bd0ba7df967ec8b6a7f5f2b4a6bb839d6 Author: Thomas Segismont tsegismo@redhat.com Date: Wed Jul 3 12:27:55 2013 +0200
Bug 923400 - Sigar creates high number of blocked threads (unbounded) if mount is gone
diff --git a/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccess.java b/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccess.java index 4319f9a..8c7cea9 100644 --- a/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccess.java +++ b/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccess.java @@ -1,6 +1,6 @@ /* * RHQ Management Platform - * Copyright (C) 2005-2008 Red Hat, Inc. + * Copyright (C) 2005-2013 Red Hat, Inc. * All rights reserved. * * This program is free software; you can redistribute it and/or modify @@ -13,17 +13,13 @@ * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + * along with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA */ package org.rhq.core.system;
-import java.lang.reflect.InvocationHandler; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; import java.lang.reflect.Proxy;
-import org.hyperic.sigar.Sigar; import org.hyperic.sigar.SigarProxy;
/** @@ -34,7 +30,6 @@ import org.hyperic.sigar.SigarProxy; * @author John Mazzitelli */ public class SigarAccess { - private static SigarAccessHandler invocationHandler; private static SigarProxy sigarProxy;
@@ -83,35 +78,4 @@ public class SigarAccess {
}
- private static class SigarAccessHandler implements InvocationHandler { - private Sigar sigar; - - public Object invoke(Object proxy, Method meth, Object[] args) throws Throwable { - // its possible in the time between this handler's creation and now, someone disabled the native layer. - // throw a runtime exception if the native system was disabled - if (SystemInfoFactory.isNativeSystemInfoDisabled()) { - throw new SystemInfoException("Native system has been disabled"); - } - - try { - synchronized (this) { - if (sigar == null) { - this.sigar = new Sigar(); - } - - return meth.invoke(sigar, args); - } - } catch (InvocationTargetException e) { - throw e.getTargetException(); - } - } - - public synchronized void close() { - if (this.sigar != null) { - this.sigar.close(); - this.sigar = null; - } - } - } - -} \ No newline at end of file +} diff --git a/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccessHandler.java b/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccessHandler.java new file mode 100644 index 0000000..a781641 --- /dev/null +++ b/modules/core/native-system/src/main/java/org/rhq/core/system/SigarAccessHandler.java @@ -0,0 +1,201 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 Red Hat, Inc. + * All rights reserved. + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation version 2 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA + */ + +package org.rhq.core.system; + +import static java.util.concurrent.TimeUnit.MINUTES; +import static java.util.concurrent.TimeUnit.SECONDS; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.hyperic.sigar.Sigar; + +/** + * An {@link InvocationHandler} for a {@link org.hyperic.sigar.SigarProxy}. + * <p>A single instance of this class will be created by the {@link SigarAccess} class.</p> + * This class holds a shared {@link Sigar} instance and serializes calls. If a thread waits more than + * 'sharedSigarLockMaxWait' seconds, it will be given a new Sigar instance, which will be destroyed at the end of the + * call. + * <p>Every 5 minutes, a background task checks that 'localSigarInstancesWarningThreshold' has not been exceeded. It it + * has, a warning message will be logged, optionally with a thread dump. + * <p>This class is configurable with System properties: + * <ul> + * <li><strong>sharedSigarLockMaxWait</strong>: maximum time in seconds a thread will wait for the shared Sigar lock + * acquistion; defaults to 2 seconds</li> + * <li><strong>localSigarInstancesWarningThreshold</strong>: threshold of currently living Sigar instances at which + * the background task will print warning messages; defaults to 50</li> + * <li><strong>maxLocalSigarInstances</strong>: maximum number of local Sigar instances which can be created, zero + * and negative values being interpreted as 'no limit'; defaults to 50</li> + * <li><strong>threadDumpOnlocalSigarInstancesWarningThreshold</strong>: if set to true (case insensitive), the + * background task will also log a thread dump when <strong>localSigarInstancesWarningThreshold</strong> is met</li> + * </ul> + * </p> + * + * @author Thomas Segismont + */ +class SigarAccessHandler implements InvocationHandler { + private static final Log LOG = LogFactory.getLog(SigarAccessHandler.class); + + private static final String LINE_SEPARATOR = System.getProperty("line.separator"); + private static final int SHARED_SIGAR_LOCK_MAX_WAIT = Integer.getInteger("sharedSigarLockMaxWait", 2); + private static final int LOCAL_SIGAR_INSTANCES_WARNING_THRESHOLD = Integer.getInteger( + "localSigarInstancesWarningThreshold", 50); + private static final int MAX_LOCAL_SIGAR_INSTANCES = Integer.getInteger("maxLocalSigarInstances", 50); + private static final boolean LIMIT_LOCAL_SIGAR_INSTANCES = MAX_LOCAL_SIGAR_INSTANCES > 0; + private static final boolean THREAD_DUMP_ON_SIGAR_INSTANCES_THRESHOLD = Boolean + .getBoolean("threadDumpOnlocalSigarInstancesWarningThreshold"); + + private final SigarFactory sigarFactory; + private final ReentrantLock sharedSigarLock; + private final ReentrantLock localSigarLock; + private final ScheduledExecutorService scheduledExecutorService; + private volatile int localSigarInstancesCount; + private Sigar sharedSigar; + + SigarAccessHandler() { + this(new DefaultSigarFactory()); + } + + SigarAccessHandler(SigarFactory sigarFactory) { + this.sigarFactory = sigarFactory; + sharedSigarLock = new ReentrantLock(); + localSigarLock = new ReentrantLock(); + scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); + scheduledExecutorService.scheduleWithFixedDelay(new ThresholdChecker(), 1, 5, MINUTES); + localSigarInstancesCount = 0; + } + + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + // its possible in the time between this handler's creation and now, someone disabled the native layer. + // throw a runtime exception if the native system was disabled + if (SystemInfoFactory.isNativeSystemInfoDisabled()) { + throw new SystemInfoException("Native system has been disabled"); + } + + // Acquire lock for shared Sigar instance. Wait 'sharedSigarLockMaxWait' seconds at most + boolean acquiredLock = sharedSigarLock.tryLock(SHARED_SIGAR_LOCK_MAX_WAIT, SECONDS); + if (acquiredLock) { + if (sharedSigar == null) { + this.sharedSigar = sigarFactory.createSigarInstance(); + } + try { + return method.invoke(sharedSigar, args); + } catch (InvocationTargetException e) { + throw e.getTargetException(); + } finally { + sharedSigarLock.unlock(); + } + } else { + Sigar localSigar = createLocalSigarInstance(); + try { + return method.invoke(localSigar, args); + } catch (InvocationTargetException e) { + throw e.getTargetException(); + } finally { + closeLocalSigarInstance(localSigar); + } + } + } + + private Sigar createLocalSigarInstance() { + localSigarLock.lock(); + try { + if (LIMIT_LOCAL_SIGAR_INSTANCES && localSigarInstancesCount >= MAX_LOCAL_SIGAR_INSTANCES) { + throw new RuntimeException("Too many Sigar instances created"); + } + Sigar sigarInstance = sigarFactory.createSigarInstance(); + localSigarInstancesCount++; + return sigarInstance; + } finally { + localSigarLock.unlock(); + } + } + + private void closeLocalSigarInstance(Sigar sigar) { + localSigarLock.lock(); + try { + sigar.close(); + localSigarInstancesCount--; + } finally { + localSigarLock.unlock(); + } + } + + void close() { + if (sharedSigar != null) { + sharedSigarLock.lock(); + try { + sharedSigar.close(); + sharedSigar = null; + } finally { + sharedSigarLock.unlock(); + } + } + scheduledExecutorService.shutdownNow(); + } + + int localSigarInstancesCount() { + return localSigarInstancesCount; + } + + interface SigarFactory { + Sigar createSigarInstance(); + } + + private static class DefaultSigarFactory implements SigarFactory { + @Override + public Sigar createSigarInstance() { + return new Sigar(); + } + } + + private class ThresholdChecker implements Runnable { + @Override + public void run() { + int currentCount = localSigarInstancesCount; + if (currentCount > LOCAL_SIGAR_INSTANCES_WARNING_THRESHOLD) { + StringBuilder sb = new StringBuilder(); + sb.append("There are ").append(currentCount).append(" local Sigar instances currently active. ") + .append("This may indicate that a call to the shared Sigar instance did not complete."); + if (THREAD_DUMP_ON_SIGAR_INSTANCES_THRESHOLD) { + sb.append(LINE_SEPARATOR); + threadDump(sb); + } + LOG.warn(sb.toString()); + } + } + + private void threadDump(StringBuilder sb) { + ThreadInfo[] threadInfos = ManagementFactory.getThreadMXBean().dumpAllThreads(true, true); + for (ThreadInfo threadInfo : threadInfos) { + sb.append(threadInfo); + sb.append(LINE_SEPARATOR); + } + } + } +} diff --git a/modules/core/native-system/src/test/java/org/rhq/core/system/SigarAccessHandlerTest.java b/modules/core/native-system/src/test/java/org/rhq/core/system/SigarAccessHandlerTest.java new file mode 100644 index 0000000..e993725 --- /dev/null +++ b/modules/core/native-system/src/test/java/org/rhq/core/system/SigarAccessHandlerTest.java @@ -0,0 +1,172 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 Red Hat, Inc. + * All rights reserved. + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation version 2 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA + */ + +package org.rhq.core.system; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +import java.lang.reflect.Method; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import org.hyperic.sigar.Sigar; +import org.hyperic.sigar.SigarProxy; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * @author Thomas Segismont + */ +@Test(singleThreaded = true) +public class SigarAccessHandlerTest { + private static final int THREAD_POOL_SIZE = 100; + + private Method getSwapMethod; + private Method getFileSystemMapMethod; + private Sigar sigarMock; + private ExecutorService executorService; + private SigarAccessHandler sigarAccessHandler; + + @BeforeClass + private void setFields() throws Exception { + getSwapMethod = SigarProxy.class.getMethod("getSwap", new Class[0]); + getFileSystemMapMethod = SigarProxy.class.getMethod("getFileSystemMap", new Class[0]); + sigarMock = mock(Sigar.class); + // Wait 15 seconds on call to getSwap + when(sigarMock.getSwap()).then(new Answer<Object>() { + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + Thread.sleep(1000 * 15); + return null; + } + }); + } + + @BeforeMethod + public void setUp() throws Exception { + sigarAccessHandler = new SigarAccessHandler(new SigarAccessHandler.SigarFactory() { + @Override + public Sigar createSigarInstance() { + return sigarMock; + } + }); + executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE); + } + + @AfterMethod + public void tearDown() throws Exception { + sigarAccessHandler.close(); + executorService.shutdownNow(); + + } + + @Test(timeOut = 1000 * 30) + public void testOnDemandSigarInstanceCreation() throws Throwable { + // Start concurrent invocations of the long running getSwap method + for (int i = 0; i < 10; i++) { + executorService.submit(new Runnable() { + @Override + public void run() { + try { + sigarAccessHandler.invoke(null, getSwapMethod, new Object[0]); + } catch (Throwable ignore) { + } + } + }); + } + Thread.sleep(1000 * 8); + + assertEquals(sigarAccessHandler.localSigarInstancesCount(), 9); + } + + @Test(timeOut = 1000 * 30) + public void testSigarInstanceDestruction() throws Throwable { + // Start concurrent invocations of the short running getFileSystemMapMethod method and an invocation of the long + // running getSwap method + for (int i = 0; i < 10; i++) { + if (i == 0) { + executorService.submit(new Runnable() { + @Override + public void run() { + try { + sigarAccessHandler.invoke(null, getSwapMethod, new Object[0]); + } catch (Throwable ignore) { + } + } + }); + } else { + executorService.submit(new Runnable() { + @Override + public void run() { + try { + sigarAccessHandler.invoke(null, getFileSystemMapMethod, new Object[0]); + } catch (Throwable ignore) { + } + } + }); + } + } + Thread.sleep(1000 * 8); + + assertEquals(sigarAccessHandler.localSigarInstancesCount(), 0); + } + + @Test(timeOut = 1000 * 30) + public void testMaxSigarInstanceCreation() throws Throwable { + // Start concurrent invocations of the long running getSwap method + List<Future<Throwable>> futures = new LinkedList<Future<Throwable>>(); + for (int i = 0; i < 60; i++) { + futures.add(executorService.submit(new Callable<Throwable>() { + @Override + public Throwable call() throws Exception { + try { + sigarAccessHandler.invoke(null, getSwapMethod, new Object[0]); + } catch (Throwable throwable) { + return throwable; + } + return null; + } + })); + } + Thread.sleep(1000 * 8); + + assertEquals(sigarAccessHandler.localSigarInstancesCount(), 50); + int failedCallsCount = 0; + for (Future<Throwable> future : futures) { + Throwable throwable = future.get(); + if (throwable != null) { + failedCallsCount++; + assertTrue(throwable instanceof RuntimeException); + assertEquals(throwable.getMessage(), "Too many Sigar instances created"); + } + } + assertEquals(failedCallsCount, 9); + } +}
rhq-commits@lists.fedorahosted.org