From 590694b0e42b04907317bbcde3400199fd9a399c Mon Sep 17 00:00:00 2001 From: "koo.taejin" Date: Tue, 26 Apr 2016 17:42:35 +0900 Subject: [PATCH] Added UdpDataSender using NIO #1711 added NioUdpDataSender --- agent/src/main/resources/pinpoint.config | 2 + .../config/DefaultProfilerConfig.java | 30 ++- .../bootstrap/config/ProfilerConfig.java | 4 + .../pinpoint/profiler/DefaultAgent.java | 25 +-- .../profiler/sender/NioUDPDataSender.java | 191 ++++++++++++++++++ .../profiler/sender/UdpDataSenderFactory.java | 40 ++++ .../profiler/sender/UdpDataSenderType.java | 11 + .../profiler/sender/NioUdpDataSenderTest.java | 113 +++++++++++ .../agent/src/main/resources/pinpoint.config | 2 + quickstart/conf/pinpoint.config | 2 + .../thrift/io/ByteBufferOutputStream.java | 89 ++++++++ .../thrift/io/HeaderTBaseSerializer2.java | 50 +++++ .../io/HeaderTBaseSerializerFactory2.java | 39 ++++ .../thrift/io/ResettableOutputStream.java | 12 ++ .../thrift/io/TOutputStreamTransport.java | 55 +++++ 15 files changed, 647 insertions(+), 18 deletions(-) create mode 100644 profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/NioUDPDataSender.java create mode 100644 profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderFactory.java create mode 100644 profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderType.java create mode 100644 profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/NioUdpDataSenderTest.java create mode 100644 thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ByteBufferOutputStream.java create mode 100644 thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializer2.java create mode 100644 thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializerFactory2.java create mode 100644 thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ResettableOutputStream.java create mode 100644 thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TOutputStreamTransport.java diff --git a/agent/src/main/resources/pinpoint.config b/agent/src/main/resources/pinpoint.config index 847f9f936..7dd3fda84 100644 --- a/agent/src/main/resources/pinpoint.config +++ b/agent/src/main/resources/pinpoint.config @@ -47,12 +47,14 @@ profiler.spandatasender.write.queue.size=5120 #profiler.spandatasender.socket.sendbuffersize=1048576 #profiler.spandatasender.socket.timeout=3000 profiler.spandatasender.chunk.size=16384 +profiler.spandatasender.socket.type=OIO # Capacity of the StatDataSender write queue. profiler.statdatasender.write.queue.size=5120 #profiler.statdatasender.socket.sendbuffersize=1048576 #profiler.statdatasender.socket.timeout=3000 profiler.statdatasender.chunk.size=16384 +profiler.statdatasender.socket.type=OIO # Interval to retry sending agent info. Unit is milliseconds. profiler.agentInfo.send.retry.interval=300000 diff --git a/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/DefaultProfilerConfig.java b/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/DefaultProfilerConfig.java index d20915da8..1e2e1d258 100644 --- a/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/DefaultProfilerConfig.java +++ b/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/DefaultProfilerConfig.java @@ -16,6 +16,11 @@ package com.navercorp.pinpoint.bootstrap.config; +import com.navercorp.pinpoint.bootstrap.logging.JavaLoggerFactory; +import com.navercorp.pinpoint.bootstrap.util.NumberUtils; +import com.navercorp.pinpoint.bootstrap.util.spring.PropertyPlaceholderHelper; +import com.navercorp.pinpoint.common.util.PropertyUtils; + import java.io.FileNotFoundException; import java.io.IOException; import java.util.Arrays; @@ -28,11 +33,6 @@ import java.util.logging.Level; import java.util.logging.Logger; import java.util.regex.Pattern; -import com.navercorp.pinpoint.bootstrap.logging.JavaLoggerFactory; -import com.navercorp.pinpoint.bootstrap.util.NumberUtils; -import com.navercorp.pinpoint.bootstrap.util.spring.PropertyPlaceholderHelper; -import com.navercorp.pinpoint.common.util.PropertyUtils; - /** * @author emeroad * @author netspider @@ -101,12 +101,14 @@ public class DefaultProfilerConfig implements ProfilerConfig { private int spanDataSenderSocketSendBufferSize = 1024 * 64 * 16; private int spanDataSenderSocketTimeout = 1000 * 3; private int spanDataSenderChunkSize = 1024 * 16; + private String spanDataSenderSocketType = "OIO"; private int statDataSenderWriteQueueSize = 1024 * 5; private int statDataSenderSocketSendBufferSize = 1024 * 64 * 16; private int statDataSenderSocketTimeout = 1000 * 3; private int statDataSenderChunkSize = 1024 * 16; - + private String statDataSenderSocketType = "OIO"; + private boolean tcpDataSenderCommandAcceptEnable = false; private boolean traceAgentActiveThread = true; @@ -247,6 +249,11 @@ public class DefaultProfilerConfig implements ProfilerConfig { return statDataSenderSocketTimeout; } + @Override + public String getStatDataSenderSocketType() { + return statDataSenderSocketType; + } + @Override public int getSpanDataSenderWriteQueueSize() { return spanDataSenderWriteQueueSize; @@ -272,6 +279,11 @@ public class DefaultProfilerConfig implements ProfilerConfig { return spanDataSenderSocketTimeout; } + @Override + public String getSpanDataSenderSocketType() { + return spanDataSenderSocketType; + } + @Override public int getSpanDataSenderChunkSize() { return spanDataSenderChunkSize; @@ -568,11 +580,13 @@ public class DefaultProfilerConfig implements ProfilerConfig { this.spanDataSenderSocketSendBufferSize = readInt("profiler.spandatasender.socket.sendbuffersize", 1024 * 64 * 16); this.spanDataSenderSocketTimeout = readInt("profiler.spandatasender.socket.timeout", 1000 * 3); this.spanDataSenderChunkSize = readInt("profiler.spandatasender.chunk.size", 1024 * 16); + this.spanDataSenderSocketType = readString("profiler.spandatasender.socket.type", "OIO"); this.statDataSenderWriteQueueSize = readInt("profiler.statdatasender.write.queue.size", 1024 * 5); this.statDataSenderSocketSendBufferSize = readInt("profiler.statdatasender.socket.sendbuffersize", 1024 * 64 * 16); this.statDataSenderSocketTimeout = readInt("profiler.statdatasender.socket.timeout", 1000 * 3); this.statDataSenderChunkSize = readInt("profiler.statdatasender.chunk.size", 1024 * 16); + this.statDataSenderSocketType = readString("profiler.statdatasender.socket.type", "OIO"); this.tcpDataSenderCommandAcceptEnable = readBoolean("profiler.tcpdatasender.command.accept.enable", false); @@ -817,6 +831,8 @@ public class DefaultProfilerConfig implements ProfilerConfig { builder.append(spanDataSenderSocketTimeout); builder.append(", spanDataSenderChunkSize="); builder.append(spanDataSenderChunkSize); + builder.append(", spanDataSenderSocketType="); + builder.append(spanDataSenderSocketType); builder.append(", statDataSenderWriteQueueSize="); builder.append(statDataSenderWriteQueueSize); builder.append(", statDataSenderSocketSendBufferSize="); @@ -825,6 +841,8 @@ public class DefaultProfilerConfig implements ProfilerConfig { builder.append(statDataSenderSocketTimeout); builder.append(", statDataSenderChunkSize="); builder.append(statDataSenderChunkSize); + builder.append(", statDataSenderSocketType="); + builder.append(statDataSenderSocketType); builder.append(", tcpDataSenderCommandAcceptEnable="); builder.append(tcpDataSenderCommandAcceptEnable); builder.append(", traceAgentActiveThread="); diff --git a/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/ProfilerConfig.java b/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/ProfilerConfig.java index 4fb093e70..1c9efe35d 100644 --- a/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/ProfilerConfig.java +++ b/bootstrap-core/src/main/java/com/navercorp/pinpoint/bootstrap/config/ProfilerConfig.java @@ -43,6 +43,8 @@ public interface ProfilerConfig { int getStatDataSenderSocketTimeout(); + String getStatDataSenderSocketType(); + int getSpanDataSenderWriteQueueSize(); int getSpanDataSenderSocketSendBufferSize(); @@ -53,6 +55,8 @@ public interface ProfilerConfig { int getSpanDataSenderSocketTimeout(); + String getSpanDataSenderSocketType(); + int getSpanDataSenderChunkSize(); int getStatDataSenderChunkSize(); diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java index 214052e6f..3ae750993 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java @@ -16,15 +16,6 @@ package com.navercorp.pinpoint.profiler; -import java.lang.instrument.Instrumentation; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.Set; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import com.navercorp.pinpoint.ProductInfo; import com.navercorp.pinpoint.bootstrap.Agent; import com.navercorp.pinpoint.bootstrap.AgentOption; @@ -61,13 +52,21 @@ import com.navercorp.pinpoint.profiler.sampler.SamplerFactory; import com.navercorp.pinpoint.profiler.sender.DataSender; import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender; import com.navercorp.pinpoint.profiler.sender.TcpDataSender; -import com.navercorp.pinpoint.profiler.sender.UdpDataSender; +import com.navercorp.pinpoint.profiler.sender.UdpDataSenderFactory; import com.navercorp.pinpoint.profiler.util.ApplicationServerTypeResolver; import com.navercorp.pinpoint.profiler.util.RuntimeMXBeanUtils; import com.navercorp.pinpoint.rpc.ClassPreLoader; import com.navercorp.pinpoint.rpc.client.PinpointClient; import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import com.navercorp.pinpoint.rpc.util.ClientFactoryUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.instrument.Instrumentation; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; /** * @author emeroad @@ -364,11 +363,13 @@ public class DefaultAgent implements Agent { } protected DataSender createUdpStatDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize) { - return new UdpDataSender(this.profilerConfig.getCollectorStatServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize); + UdpDataSenderFactory factory = new UdpDataSenderFactory(this.profilerConfig.getCollectorStatServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize); + return factory.create(profilerConfig.getStatDataSenderSocketType()); } protected DataSender createUdpSpanDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize) { - return new UdpDataSender(this.profilerConfig.getCollectorSpanServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize); + UdpDataSenderFactory factory = new UdpDataSenderFactory(this.profilerConfig.getCollectorSpanServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize); + return factory.create(profilerConfig.getSpanDataSenderSocketType()); } protected EnhancedDataSender getTcpDataSender() { diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/NioUDPDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/NioUDPDataSender.java new file mode 100644 index 000000000..86601e058 --- /dev/null +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/NioUDPDataSender.java @@ -0,0 +1,191 @@ +package com.navercorp.pinpoint.profiler.sender; + +import com.navercorp.pinpoint.rpc.PinpointSocketException; +import com.navercorp.pinpoint.rpc.buffer.ByteBufferFactory; +import com.navercorp.pinpoint.rpc.buffer.ByteBufferFactoryLocator; +import com.navercorp.pinpoint.rpc.buffer.ByteBufferType; +import com.navercorp.pinpoint.thrift.io.ByteBufferOutputStream; +import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializer2; +import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializerFactory2; +import com.navercorp.pinpoint.thrift.io.NetworkAvailabilityCheckPacket; +import org.apache.thrift.TBase; +import org.apache.thrift.TException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.DatagramSocket; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.DatagramChannel; +import java.util.Arrays; + +/** + * @Author Taejin Koo + */ +public class NioUDPDataSender extends AbstractDataSender implements DataSender { + + protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + protected final boolean isDebug = logger.isDebugEnabled(); + + public static final int SOCKET_TIMEOUT = 1000 * 5; + public static final int SEND_BUFFER_SIZE = 1024 * 64 * 16; + public static final int UDP_MAX_PACKET_LENGTH = 65507; + + private final DatagramChannel datagramChannel; + private final HeaderTBaseSerializer2 serializer; + private final ByteBufferOutputStream byteBufferOutputStream; + + private final AsyncQueueingExecutor executor; + + private volatile boolean closed = false; + + public NioUDPDataSender(String host, int port, String threadName, int queueSize) { + this(host, port, threadName, queueSize, SOCKET_TIMEOUT, SEND_BUFFER_SIZE); + } + + public NioUDPDataSender(String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize) { + if (host == null ) { + throw new NullPointerException("host must not be null"); + } + if (threadName == null) { + throw new NullPointerException("threadName must not be null"); + } + if (queueSize <= 0) { + throw new IllegalArgumentException("queueSize"); + } + if (timeout <= 0) { + throw new IllegalArgumentException("timeout"); + } + if (sendBufferSize <= 0) { + throw new IllegalArgumentException("sendBufferSize"); + } + + // TODO If fail to create socket, stop agent start + logger.info("NioUDPDataSender initialized. host={}, port={}", host, port); + this.datagramChannel = createChannel(host, port, timeout, sendBufferSize); + + HeaderTBaseSerializerFactory2 serializerFactory = new HeaderTBaseSerializerFactory2(); + this.serializer = serializerFactory.createSerializer(); + + ByteBufferFactory bufferFactory = ByteBufferFactoryLocator.getFactory(ByteBufferType.DIRECT); + ByteBuffer byteBuffer = bufferFactory.getBuffer(UDP_MAX_PACKET_LENGTH); + this.byteBufferOutputStream = new ByteBufferOutputStream(byteBuffer); + + this.executor = createAsyncQueueingExecutor(queueSize, threadName); + } + + private DatagramChannel createChannel(String host, int port, int timeout, int sendBufferSize) { + DatagramChannel datagramChannel = null; + DatagramSocket socket = null; + try { + datagramChannel = DatagramChannel.open(); + socket = datagramChannel.socket(); + socket.setSoTimeout(timeout); + socket.setSendBufferSize(sendBufferSize); + + if (logger.isWarnEnabled()) { + final int checkSendBufferSize = socket.getSendBufferSize(); + if (sendBufferSize != checkSendBufferSize) { + logger.warn("DatagramChannel.setSendBufferSize() error. {}!={}", sendBufferSize, checkSendBufferSize); + } + } + + InetSocketAddress serverAddress = new InetSocketAddress(host, port); + datagramChannel.connect(serverAddress); + + return datagramChannel; + } catch (IOException e) { + if (socket != null) { + socket.close(); + } + + if (datagramChannel != null) { + try { + datagramChannel.close(); + } catch (IOException ignored) { + } + } + + throw new IllegalStateException("DatagramChannel create fail. Cause" + e.getMessage(), e); + } + } + + @Override + public boolean send(TBase data) { + return executor.execute(data); + } + + @Override + public void stop() { + try { + closed = true; + executor.stop(); + } finally { + try { + byteBufferOutputStream.close(); + } catch (IOException e) { + // ignore + } + } + } + + public boolean isNetworkAvailable() { + final NetworkAvailabilityCheckPacket dto = new NetworkAvailabilityCheckPacket(); + try { + sendPacket(dto); + + if (logger.isInfoEnabled()) { + logger.info("Data sent. {}", dto); + } + + final byte[] receiveData = new byte[NetworkAvailabilityCheckPacket.DATA_OK.length]; + datagramChannel.read(ByteBuffer.wrap(receiveData)); + + if (logger.isInfoEnabled()) { + logger.info("Data received. {}", Arrays.toString(receiveData)); + } + + return Arrays.equals(NetworkAvailabilityCheckPacket.DATA_OK , receiveData); + } catch (IOException e) { + logger.warn("packet send error {}", dto, e); + return false; + } + } + + protected void sendPacket(Object message) { + if (closed) { + throw new PinpointSocketException("NioUDPDataSender already closed."); + } + + if (message instanceof TBase) { + byteBufferOutputStream.clear(); + + final TBase dto = (TBase) message; + // do not copy bytes because it's single threaded + + try { + serializer.serialize(dto, byteBufferOutputStream); + } catch (TException e) { + throw new PinpointSocketException("Serialize " + dto + " failed. Error:" + e.getMessage(), e); + } + ByteBuffer byteBuffer = byteBufferOutputStream.getByteBuffer(); + int bufferSize = byteBuffer.remaining(); + try { + datagramChannel.write(byteBuffer); + } catch (IOException e) { + final Thread currentThread = Thread.currentThread(); + if (currentThread.isInterrupted()) { + logger.warn("{} thread interrupted.", currentThread.getName()); + throw new PinpointSocketException(currentThread.getName() + " thread interrupted.", e); + } else { + throw new PinpointSocketException("packet send error. size:" + bufferSize + ", " + dto, e); + } + } + } else { + logger.warn("sendPacket fail. invalid type:{}", message != null ? message.getClass() : null); + return; + } + } + +} diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderFactory.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderFactory.java new file mode 100644 index 000000000..8ec5e150b --- /dev/null +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderFactory.java @@ -0,0 +1,40 @@ +package com.navercorp.pinpoint.profiler.sender; + +/** + * @Author Taejin Koo + */ +public final class UdpDataSenderFactory { + +// String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize + + private final String host; + private final int port; + private final String threadName; + private final int queueSize; + private final int timeout; + private final int sendBufferSize; + + public UdpDataSenderFactory(String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize) { + this.host = host; + this.port = port; + this.threadName = threadName; + this.queueSize = queueSize; + this.timeout = timeout; + this.sendBufferSize = sendBufferSize; + } + + public DataSender create(String typeName) { + return create(UdpDataSenderType.valueOf(typeName)); + } + + public DataSender create(UdpDataSenderType type) { + if (type == UdpDataSenderType.NIO) { + return new NioUDPDataSender(host, port, threadName, queueSize, timeout, sendBufferSize); + } else if (type == UdpDataSenderType.OIO) { + return new UdpDataSender(host, port, threadName, queueSize, timeout, sendBufferSize); + } else { + throw new IllegalArgumentException("Unknown type."); + } + } + +} diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderType.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderType.java new file mode 100644 index 000000000..2bc362884 --- /dev/null +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/UdpDataSenderType.java @@ -0,0 +1,11 @@ +package com.navercorp.pinpoint.profiler.sender; + +/** + * @Author Taejin Koo + */ +public enum UdpDataSenderType { + + OIO, + NIO; + +} diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/NioUdpDataSenderTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/NioUdpDataSenderTest.java new file mode 100644 index 000000000..504f1b031 --- /dev/null +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/NioUdpDataSenderTest.java @@ -0,0 +1,113 @@ +package com.navercorp.pinpoint.profiler.sender; + +import com.navercorp.pinpoint.thrift.dto.TAgentInfo; +import org.apache.commons.lang3.RandomStringUtils; +import org.apache.thrift.TBase; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.util.SocketUtils; + +import java.io.IOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.SocketException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * @Author Taejin Koo + */ +public class NioUdpDataSenderTest { + + // The correct maximum UDP message size is 65507, as determined by the following formula: + // 0xffff - (sizeof(IP Header) + sizeof(UDP Header)) = 65535-(20+8) = 65507 + private static int AcceptedSize = 65507; + private final Logger logger = LoggerFactory.getLogger(this.getClass().getName()); + // port conflict against base port. so increased 5 + private int PORT = SocketUtils.findAvailableUdpPort(61112); + private DatagramSocket receiver; + + @Before + public void setUp() throws SocketException { + receiver = new DatagramSocket(PORT); + receiver.setSoTimeout(1000); + } + + @After + public void setDown() throws InterruptedException { + close(receiver); + // port conflict happens when testcases run continuously so port number is increased. + PORT = SocketUtils.findAvailableUdpPort(61112); + } + + private void close(DatagramSocket socket) { + if (socket == null) { + return; + } + socket.close(); + + } + + @Test + public void sendTest1() throws Exception { + NioUDPDataSender sender = new NioUDPDataSender("localhost", PORT, "test", 128, 1000, 1024 * 64 * 100); + + int sendMessageCount = 10; + TAgentInfo agentInfo = new TAgentInfo(); + for (int i = 0; i < 10; i++) { + sender.send(agentInfo); + } + + try { + waitMessageReceived(sendMessageCount); + } finally { + sender.stop(); + } + } + + @Test(expected = IOException.class) + public void exceedMessageSendTest() throws InterruptedException, IOException { + String random = RandomStringUtils.randomAlphabetic(UdpDataSender.UDP_MAX_PACKET_LENGTH + 100); + + TAgentInfo agentInfo = new TAgentInfo(); + agentInfo.setAgentId(random); + + NioUDPDataSender sender = new NioUDPDataSender("localhost", PORT, "test", 128, 1000, 1024 * 64 * 100); + sender.send(agentInfo); + + waitMessageReceived(1); + } + + + private boolean sendMessage_getLimit(TBase tbase, long waitTimeMillis) throws InterruptedException { + final AtomicBoolean limitCounter = new AtomicBoolean(false); + final CountDownLatch latch = new CountDownLatch(1); + + NioUDPDataSender sender = new NioUDPDataSender("localhost", PORT, "test", 128, 1000, 1024 * 64 * 100); + try { + sender.send(tbase); + latch.await(waitTimeMillis, TimeUnit.MILLISECONDS); + } finally { + sender.stop(); + } + return limitCounter.get(); + } + + + private void waitMessageReceived(int expectReceiveMessageCount) throws IOException { + byte[] receiveData = new byte[65535]; + DatagramPacket datagramPacket = new DatagramPacket(receiveData, 0, receiveData.length); + + int remainCount = expectReceiveMessageCount; + + while (remainCount > 0) { + remainCount--; + receiver.receive(datagramPacket); + } + } + +} diff --git a/quickstart/agent/src/main/resources/pinpoint.config b/quickstart/agent/src/main/resources/pinpoint.config index 93919fe50..54ab0fe5c 100644 --- a/quickstart/agent/src/main/resources/pinpoint.config +++ b/quickstart/agent/src/main/resources/pinpoint.config @@ -39,11 +39,13 @@ profiler.spandatasender.write.queue.size=5120 #profiler.spandatasender.socket.sendbuffersize=1048576 #profiler.spandatasender.socket.timeout=3000 profiler.spandatasender.chunk.size=16384 +profiler.spandatasender.socket.type=OIO profiler.statdatasender.write.queue.size=5120 #profiler.statdatasender.socket.sendbuffersize=1048576 #profiler.statdatasender.socket.timeout=3000 profiler.statdatasender.chunk.size=16384 +profiler.statdatasender.socket.type=OIO profiler.agentInfo.send.retry.interval=300000 diff --git a/quickstart/conf/pinpoint.config b/quickstart/conf/pinpoint.config index 8bcaa3571..fbd92e488 100644 --- a/quickstart/conf/pinpoint.config +++ b/quickstart/conf/pinpoint.config @@ -39,11 +39,13 @@ profiler.spandatasender.write.queue.size=5120 #profiler.spandatasender.socket.sendbuffersize=1048576 #profiler.spandatasender.socket.timeout=3000 profiler.spandatasender.chunk.size=16384 +profiler.spandatasender.socket.type=OIO profiler.statdatasender.write.queue.size=5120 #profiler.statdatasender.socket.sendbuffersize=1048576 #profiler.statdatasender.socket.timeout=3000 profiler.statdatasender.chunk.size=16384 +profiler.statdatasender.socket.type=OIO profiler.agentInfo.send.retry.interval=300000 diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ByteBufferOutputStream.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ByteBufferOutputStream.java new file mode 100644 index 000000000..92fda3e22 --- /dev/null +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ByteBufferOutputStream.java @@ -0,0 +1,89 @@ +package com.navercorp.pinpoint.thrift.io; + +import sun.misc.Cleaner; +import sun.nio.ch.DirectBuffer; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.ByteBuffer; + +/** + * Not Thread Safe. + * + * @Author Taejin Koo + */ +public class ByteBufferOutputStream extends OutputStream implements ResettableOutputStream { + + private final ByteBuffer byteBuffer; + + public ByteBufferOutputStream(ByteBuffer byteBuffer) { + this.byteBuffer = byteBuffer; + } + + @Override + public void write(int b) throws IOException { + checkWriteAvailable(1); + byteBuffer.put((byte) b); + } + + @Override + public void write(byte[] byteArray, int offset, int length) throws IOException { + checkWriteAvailable(length); + byteBuffer.put(byteArray, offset, length); + } + + private void checkWriteAvailable(int size) { + if (byteBuffer.remaining() >= size) { + return; + } + + byteBuffer.limit(byteBuffer.capacity()); + if (byteBuffer.remaining() < size) { + throw new BufferOverflowException("write failed remaining-size:" + byteBuffer.remaining() + ", input-size:" + size + "."); + } + } + + @Override + public void flush() throws IOException { + // ignore + } + + @Override + public void close() throws IOException { + deallocate(byteBuffer); + } + + private void deallocate(ByteBuffer byteBuffer) { + if (byteBuffer == null) { + return; + } + + if (byteBuffer instanceof DirectBuffer) { + Cleaner cleaner = ((DirectBuffer) byteBuffer).cleaner(); + if (cleaner != null) { + cleaner.clean(); + } + } + } + + + public ByteBuffer getByteBuffer() { + byteBuffer.flip(); + return byteBuffer; + } + + public void clear() { + byteBuffer.clear(); + } + + @Override + public void mark() { + byteBuffer.mark(); + } + + @Override + public void resetToMarkIndex() { + byteBuffer.reset(); + } + +} diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializer2.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializer2.java new file mode 100644 index 000000000..29a0a9a36 --- /dev/null +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializer2.java @@ -0,0 +1,50 @@ +package com.navercorp.pinpoint.thrift.io; + +import org.apache.thrift.TBase; +import org.apache.thrift.TException; +import org.apache.thrift.protocol.TProtocol; +import org.apache.thrift.protocol.TProtocolFactory; + +import java.io.OutputStream; + +/** + * + * Caution. not thread safe + * + * @Author Taejin Koo + */ +public class HeaderTBaseSerializer2 { + + private static final String UTF8 = "UTF8"; + + private final TOutputStreamTransport tOutputStreamTransport; + private final TProtocol protocol; + private final TBaseLocator tBaseLocator; + + public HeaderTBaseSerializer2(TProtocolFactory protocolFactory, TBaseLocator tBaseLocator) { + this.tOutputStreamTransport = new TOutputStreamTransport(); + this.protocol = protocolFactory.getProtocol(tOutputStreamTransport); + this.tBaseLocator = tBaseLocator; + } + + public void serialize(TBase base, OutputStream outputStream) throws TException { + tOutputStreamTransport.open(outputStream); + try { + final Header header = tBaseLocator.headerLookup(base); + writeHeader(header); + base.write(protocol); + } finally { + tOutputStreamTransport.close(); + } + } + + private void writeHeader(Header header) throws TException { + protocol.writeByte(header.getSignature()); + protocol.writeByte(header.getVersion()); + // fixed size regardless protocol + short type = header.getType(); + protocol.writeByte(BytesUtils.writeShort1(type)); + protocol.writeByte(BytesUtils.writeShort2(type)); + } + +} diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializerFactory2.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializerFactory2.java new file mode 100644 index 000000000..e17b85124 --- /dev/null +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/HeaderTBaseSerializerFactory2.java @@ -0,0 +1,39 @@ +package com.navercorp.pinpoint.thrift.io; + +import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.protocol.TProtocolFactory; + +/** + * @Author Taejin Koo + */ +public class HeaderTBaseSerializerFactory2 implements SerializerFactory { + + private static final TProtocolFactory DEFAULT_PROTOCOL_FACTORY = new TCompactProtocol.Factory(); + private static final TBaseLocator DEFAULT_TBASE_LOCATOR = new DefaultTBaseLocator(); + + private final TProtocolFactory protocolFactory; + private final TBaseLocator tBaseLocator; + + public HeaderTBaseSerializerFactory2() { + this(DEFAULT_PROTOCOL_FACTORY, DEFAULT_TBASE_LOCATOR); + } + + public HeaderTBaseSerializerFactory2(TProtocolFactory protocolFactory) { + this(protocolFactory, DEFAULT_TBASE_LOCATOR); + } + + public HeaderTBaseSerializerFactory2(TBaseLocator tBaseLocator) { + this(DEFAULT_PROTOCOL_FACTORY, tBaseLocator); + } + + public HeaderTBaseSerializerFactory2(TProtocolFactory protocolFactory, TBaseLocator tBaseLocator) { + this.protocolFactory = protocolFactory; + this.tBaseLocator = tBaseLocator; + } + + @Override + public HeaderTBaseSerializer2 createSerializer() { + return new HeaderTBaseSerializer2(protocolFactory, tBaseLocator); + } + +} diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ResettableOutputStream.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ResettableOutputStream.java new file mode 100644 index 000000000..d0b22202f --- /dev/null +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ResettableOutputStream.java @@ -0,0 +1,12 @@ +package com.navercorp.pinpoint.thrift.io; + +/** + * @Author Taejin Koo + */ +public interface ResettableOutputStream { + + void mark(); + + void resetToMarkIndex(); + +} diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TOutputStreamTransport.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TOutputStreamTransport.java new file mode 100644 index 000000000..69bae14e2 --- /dev/null +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TOutputStreamTransport.java @@ -0,0 +1,55 @@ +package com.navercorp.pinpoint.thrift.io; + +import org.apache.thrift.transport.TTransport; +import org.apache.thrift.transport.TTransportException; + +import java.io.IOException; +import java.io.OutputStream; + +/** + * + * Caution. not thread safe + * + * @Author Taejin Koo + */ +public class TOutputStreamTransport extends TTransport { + + private OutputStream outputStream; + + @Override + public boolean isOpen() { + return outputStream != null; + } + + @Override + public void open() throws TTransportException { + } + + public void open(OutputStream outputStream) throws TTransportException { + this.outputStream = outputStream; + } + + @Override + public void close() { + this.outputStream = null; + } + + @Override + public int read(byte[] bytes, int index, int length) throws TTransportException { + throw new UnsupportedOperationException(); + } + + @Override + public void write(byte[] bytes, int index, int length) throws TTransportException { + if (this.outputStream == null) { + throw new TTransportException(1, "TOutputStreamTransport is not opend."); + } else { + try { + this.outputStream.write(bytes, index, length); + } catch (IOException var5) { + throw new TTransportException(0, var5); + } + } + } + +}