diff --git a/commons/src/main/java/com/navercorp/pinpoint/common/util/PinpointThreadFactory.java b/commons/src/main/java/com/navercorp/pinpoint/common/util/PinpointThreadFactory.java index dbb6e95ca..c1ce38dd2 100644 --- a/commons/src/main/java/com/navercorp/pinpoint/common/util/PinpointThreadFactory.java +++ b/commons/src/main/java/com/navercorp/pinpoint/common/util/PinpointThreadFactory.java @@ -45,7 +45,7 @@ public class PinpointThreadFactory implements ThreadFactory { String newThreadName = createThreadName(); Thread thread = new Thread(job, newThreadName); if (daemon) { - thread.setDaemon(daemon); + thread.setDaemon(true); } return thread; } diff --git a/commons/src/main/java/com/navercorp/pinpoint/common/util/ThreadMXBeanUtils.java b/commons/src/main/java/com/navercorp/pinpoint/common/util/ThreadMXBeanUtils.java new file mode 100644 index 000000000..7a45fdf1b --- /dev/null +++ b/commons/src/main/java/com/navercorp/pinpoint/common/util/ThreadMXBeanUtils.java @@ -0,0 +1,76 @@ +package com.nhn.pinpoint.common.util; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.logging.Level; +import java.util.logging.Logger; + +/** + * @author emeroad + */ +public final class ThreadMXBeanUtils { + + private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean(); + + private static final boolean OBJECT_MONITOR_USAGE_SUPPORT; + private static final boolean SYNCHRONIZER_USAGE_SUPPORT; + // check support -> getWaitedTime(), getBlockedTime() + private static final boolean CONTENTION_MONITORING_SUPPORT; + + private ThreadMXBeanUtils() { + } + + static { + OBJECT_MONITOR_USAGE_SUPPORT = THREAD_MX_BEAN.isObjectMonitorUsageSupported(); + SYNCHRONIZER_USAGE_SUPPORT = THREAD_MX_BEAN.isSynchronizerUsageSupported(); + CONTENTION_MONITORING_SUPPORT = THREAD_MX_BEAN.isThreadContentionMonitoringSupported(); + logOption(); + } + + private static void logOption() { + final Logger logger = Logger.getLogger(ThreadMXBeanUtils.class.getName()); + if (logger.isLoggable(Level.INFO)) { + final StringBuilder builder = new StringBuilder(); + builder.append("ThreadMXBean SupportOption:{OBJECT_MONITOR_USAGE_SUPPORT="); + builder.append(OBJECT_MONITOR_USAGE_SUPPORT); + builder.append("}, {SYNCHRONIZER_USAGE_SUPPORT="); + builder.append(SYNCHRONIZER_USAGE_SUPPORT); + builder.append("}, {CONTENTION_MONITORING_SUPPORT="); + builder.append(CONTENTION_MONITORING_SUPPORT); + builder.append('}'); + logger.info(builder.toString()); + } + } + + public static ThreadInfo[] dumpAllThread() { +// try { + return THREAD_MX_BEAN.dumpAllThreads(OBJECT_MONITOR_USAGE_SUPPORT, SYNCHRONIZER_USAGE_SUPPORT); +// ?? handle exception +// } catch (java.lang.SecurityException se) { +// log?? +// return new ThreadInfo[]{}; +// } catch (java.lang.UnsupportedOperationException ue) { +// log?? +// return new ThreadInfo[]{}; +// } + } + + public static boolean findThreadName(ThreadInfo[] threadInfos, String threadName) { + if (threadInfos == null) { + return false; + } + for (ThreadInfo threadInfo : threadInfos) { + if (threadInfo.getThreadName().equals(threadName)) { + return true; + } + } + return false; + } + + public static boolean findThreadName(String threadName) { + final ThreadInfo[] threadInfos = dumpAllThread(); + return findThreadName(threadInfos, threadName); + } + +} diff --git a/commons/src/test/java/com/navercorp/pinpoint/common/util/ThreadMXBeanUtilsTest.java b/commons/src/test/java/com/navercorp/pinpoint/common/util/ThreadMXBeanUtilsTest.java new file mode 100644 index 000000000..9b126b760 --- /dev/null +++ b/commons/src/test/java/com/navercorp/pinpoint/common/util/ThreadMXBeanUtilsTest.java @@ -0,0 +1,64 @@ +package com.nhn.pinpoint.common.util; + +import junit.framework.Assert; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.management.ThreadInfo; +import java.util.Arrays; +import java.util.concurrent.CountDownLatch; + +public class ThreadMXBeanUtilsTest { + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Test + public void testName() throws Exception { + ThreadInfo[] threadInfos = ThreadMXBeanUtils.dumpAllThread(); + + Assert.assertNotNull(threadInfos); + logger.trace("thread:{}", Arrays.toString(threadInfos)); + } + + @Test + public void testHasThreadName() { + + String threadName = "ThreadMXBeanUtils-test-thread"; + + Assert.assertFalse(ThreadMXBeanUtils.findThreadName(threadName)); + + WaitingRunnable waiting = new WaitingRunnable(); + Thread thread = new Thread(waiting, threadName); + thread.start(); + + Assert.assertTrue(ThreadMXBeanUtils.findThreadName(threadName)); + + waiting.stop(); + try { + thread.join(2000); + } catch (InterruptedException e) { + Assert.fail(); + } + + Assert.assertFalse(ThreadMXBeanUtils.findThreadName(threadName)); + } + + private class WaitingRunnable implements Runnable { + + private final CountDownLatch latch = new CountDownLatch(1); + + @Override + public void run() { + try { + latch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + public void stop() { + latch.countDown(); + } + + } +} \ No newline at end of file diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/service/ThreadDumpService.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/service/ThreadDumpService.java index d21c9e5f2..fb3e05ef2 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/service/ThreadDumpService.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/service/ThreadDumpService.java @@ -9,6 +9,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import com.nhn.pinpoint.common.util.ThreadMXBeanUtils; import org.apache.thrift.TBase; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -145,8 +146,7 @@ public class ThreadDumpService implements ProfilerRequestCommandService { } private ThreadInfo[] getAllThreadInfo() { - ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean(); - ThreadInfo[] threadInfos = threadMxBean.getThreadInfo(threadMxBean.getAllThreadIds(), 100); + ThreadInfo[] threadInfos = ThreadMXBeanUtils.dumpAllThread(); return threadInfos; } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSender.java index 315388263..5136ec5c3 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSender.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSender.java @@ -6,14 +6,11 @@ import java.util.concurrent.TimeUnit; import org.apache.thrift.TBase; import org.apache.thrift.TException; -import org.apache.thrift.protocol.TCompactProtocol; -import org.apache.thrift.protocol.TProtocolFactory; import com.nhn.pinpoint.common.util.PinpointThreadFactory; import com.nhn.pinpoint.thrift.io.ChunkHeaderBufferedTBaseSerializer; import com.nhn.pinpoint.thrift.io.ChunkHeaderBufferedTBaseSerializerFactory; import com.nhn.pinpoint.thrift.io.ChunkHeaderBufferedTBaseSerializerFlushHandler; -import com.nhn.pinpoint.thrift.io.TBaseLocator; /** * split & buffering @@ -25,10 +22,14 @@ import com.nhn.pinpoint.thrift.io.TBaseLocator; */ public class BufferedUdpDataSender extends UdpDataSender { private static final int CHUNK_SIZE = 1024 * 16; - private static final String SCHEDULED_FLUSH = "ScheduledFlush"; + + private static final String SCHEDULED_FLUSH = "BufferedUdpDataSender-ScheduledFlush"; private final ChunkHeaderBufferedTBaseSerializer chunkHeaderBufferedSerializer = new ChunkHeaderBufferedTBaseSerializerFactory().createSerializer(); + private final Thread flushThread; + + public BufferedUdpDataSender(String host, int port, String threadName, int queueSize) { this(host, port, threadName, queueSize, SOCKET_TIMEOUT, SEND_BUFFER_SIZE, CHUNK_SIZE); } @@ -55,22 +56,30 @@ public class BufferedUdpDataSender extends UdpDataSender { try { udpSocket.send(reusePacket); - logger.debug("Data sent. {size={}}", internalBufferSize); + if (isDebug) { + logger.debug("Data sent. {size={}}", internalBufferSize); + } } catch (IOException e) { logger.warn("packet send error. size:{}", internalBufferSize, e); } } }); - startScheduledFlush(); + flushThread = startScheduledFlush(); } - private void startScheduledFlush() { + // for test + String getFlushThreadName() { + return flushThread.getName(); + } + + private Thread startScheduledFlush() { final ThreadFactory threadFactory = new PinpointThreadFactory(SCHEDULED_FLUSH, true); final Thread thread = threadFactory.newThread(new Runnable() { @Override public void run() { - while (true) { + final Thread currentThread = Thread.currentThread(); + while (!currentThread.isInterrupted()) { try { chunkHeaderBufferedSerializer.flush(); } catch (TException e) { @@ -79,13 +88,18 @@ public class BufferedUdpDataSender extends UdpDataSender { try { TimeUnit.MILLISECONDS.sleep(1000); } catch (InterruptedException ignored) { + currentThread.interrupt(); } } + logger.info("stop ScheduledFlush {} - {}", currentThread.getName(), currentThread.getId()); } }); + logger.info("stop ScheduledFlush {} - {}", thread.getName(), thread.getId()); thread.start(); + return thread; } + @Override protected void sendPacket(Object message) { if (message instanceof TBase) { @@ -101,4 +115,22 @@ public class BufferedUdpDataSender extends UdpDataSender { return; } } + + @Override + public void stop() { + super.stop(); + stopFlushThread(); + } + + private void stopFlushThread() { + final Thread flushThread = this.flushThread; + // terminate thread + flushThread.interrupt(); + try { + flushThread.join(5000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + } } \ No newline at end of file diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSender.java index 7793821dc..ca5bb6b38 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSender.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSender.java @@ -23,8 +23,8 @@ import com.nhn.pinpoint.thrift.io.NetworkAvailabilityCheckPacket; public class UdpDataSender extends AbstractDataSender implements DataSender { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); - private final boolean isTrace = logger.isTraceEnabled(); - private final boolean isDebug = logger.isDebugEnabled(); + protected final boolean isTrace = logger.isTraceEnabled(); + protected final boolean isDebug = logger.isDebugEnabled(); public static final int SOCKET_TIMEOUT = 1000 * 5; public static final int SEND_BUFFER_SIZE = 1024 * 64 * 16; diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSenderTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSenderTest.java new file mode 100644 index 000000000..445c9abdf --- /dev/null +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/BufferedUdpDataSenderTest.java @@ -0,0 +1,31 @@ +package com.nhn.pinpoint.profiler.sender; + +import com.nhn.pinpoint.common.util.ThreadMXBeanUtils; +import junit.framework.Assert; +import org.junit.Test; + + +public class BufferedUdpDataSenderTest { + + @Test + public void testSendPacket() throws Exception { + + + + } + + @Test + public void testStop_StopFlushThread() throws Exception { + + final BufferedUdpDataSender sender = new BufferedUdpDataSender("localhost", 9999, "testUdpSender", 100); + + final String flushThreadName = sender.getFlushThreadName(); + + Assert.assertTrue(ThreadMXBeanUtils.findThreadName(flushThreadName)); + + sender.stop(); + + Assert.assertFalse(ThreadMXBeanUtils.findThreadName(flushThreadName)); + // ?? finally { send.stop() } + } +} \ No newline at end of file