Merge branch '#28_udp_socket_improvement' of emeroad/pinpoint

from pull-request 155

* refs/heads/#28_udp_socket_improvement:
  #28 implements flushThread stop  - add ThreadMXBeanUtils

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;
     }

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);
+    }
+
+}

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();
+        }
+
+    }
+}

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;
 	}

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();
+        }
+
+    }
 }

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;

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() }
+    }
+}