home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-17 00:36:02 +10:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #409 from koo-taejin/issue_286

Browse Source 
Optimize udp efficiency. #286
This commit is contained in:
koo-taejin
2015-05-12 16:23:33 +09:00
parent 68fa974a4e fc86922734
commit cb8ce1aa25
2 changed files with 23 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/SpanStreamUdpSender.java
+19 -8
View File
@@ -43,8 +43,10 @@ public class SpanStreamUdpSender extends AbstractDataSender {
public static final int SOCKET_TIMEOUT = 1000 * 5;
public static final int SEND_BUFFER_SIZE = 1024 * 64 * 16;
public static final int DEFAULT_BUFFER_SIZE = 1024 * 16;
public static final int UDP_MAX_PACKET_LENGTH = 65507;
private final SpanStreamSendDataFactory spanStreamSendDataFactory;
@@ -60,8 +62,12 @@ public class SpanStreamUdpSender extends AbstractDataSender {
public SpanStreamUdpSender(String host, int port, String threadName, int queueSize) {
this(host, port, threadName, queueSize, SOCKET_TIMEOUT, SEND_BUFFER_SIZE);
}
public SpanStreamUdpSender(String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize) {
this(host, port, threadName, queueSize, SOCKET_TIMEOUT, SEND_BUFFER_SIZE, DEFAULT_BUFFER_SIZE);
}
public SpanStreamUdpSender(String host, int port, String threadName, int queueSize, int timeout, int sendBufferSize, int dataBufferSize) {
if (host == null) {
throw new NullPointerException("host must not be null");
}
@@ -82,12 +88,12 @@ public class SpanStreamUdpSender extends AbstractDataSender {
logger.info("UdpDataSender initialized. host={}, port={}", host, port);
this.udpChannel = createChannel(host, port, timeout, sendBufferSize);
HeaderTBaseSerializerPoolFactory headerTBaseSerializerPoolFactory = new HeaderTBaseSerializerPoolFactory(false, sendBufferSize, true);
HeaderTBaseSerializerPoolFactory headerTBaseSerializerPoolFactory = new HeaderTBaseSerializerPoolFactory(false, dataBufferSize, true);
this.serializerPool = new ObjectPool<HeaderTBaseSerializer>(headerTBaseSerializerPoolFactory, 16);
this.spanStreamSendDataSerializer = new SpanStreamSendDataSerializer();
this.spanStreamSendDataFactory = new SpanStreamSendDataFactory(sendBufferSize, 16, serializerPool);
this.spanStreamSendDataFactory = new SpanStreamSendDataFactory(dataBufferSize, 16, serializerPool);
this.standbySpanStreamDataSendWorker = new StandbySpanStreamDataSendWorker(new FlushHandler(), new StandbySpanStreamDataStorage());
this.standbySpanStreamDataSendWorker.start();
@@ -176,7 +182,7 @@ public class SpanStreamUdpSender extends AbstractDataSender {
serializerPool.returnObject(serializer);
return;
}
doAddAndFlush(compositeSpanStreamData, serializer);
}
@@ -205,9 +211,9 @@ public class SpanStreamUdpSender extends AbstractDataSender {
}
try {
if (!currentSpanStreamSendData.addBuffer(compositeSpanStreamData.getByteBuffer())) {
if (!currentSpanStreamSendData.addBuffer(compositeSpanStreamData.getByteBuffer())) {
SendDataPlaner sendDataPlaner = new SpanChunkStreamSendDataPlaner(compositeSpanStreamData, spanStreamSendDataFactory);
Iterator<SpanStreamSendData> sendDataIterator = sendDataPlaner.getSendDataIterator(currentSpanStreamSendData, serializer);
while (sendDataIterator.hasNext()) {
SpanStreamSendData sendData = sendDataIterator.next();
@@ -220,8 +226,9 @@ public class SpanStreamUdpSender extends AbstractDataSender {
}
}
}
} else {
boolean isAdded = standbySpanStreamDataSendWorker.addStandbySpanStreamData(currentSpanStreamSendData);
}
} catch (IOException e) {
logger.warn("UDPChannel write fail.", e);
}
@@ -236,6 +243,8 @@ public class SpanStreamUdpSender extends AbstractDataSender {
long sentBufferSize = udpChannel.write(byteBuffers);
if (remainingLength != sentBufferSize) {
logger.warn("sent buffer {}/{}.", sentBufferSize, remainingLength);
} else {
logger.debug("Data sent. size:{}, {}", sentBufferSize);
}
}
} finally {
@@ -257,7 +266,9 @@ public class SpanStreamUdpSender extends AbstractDataSender {
long sentBufferSize = udpChannel.write(byteBuffers);
if (remainingLength != sentBufferSize) {
logger.warn("sent buffer {}/{}.", sentBufferSize, remainingLength);
} else {
}
}
} catch (IOException e) {
logger.warn("Failed to flush span stream data.", e);
profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/planer/AbstractSpanStreamSendDataPlaner.java
+4
View File
@@ -73,6 +73,10 @@ public abstract class AbstractSpanStreamSendDataPlaner implements SendDataPlaner
int buffersLength = compositeSpanStreamData.getComponentsBufferCapacity();
if (currentSpanStreamSendData.isAvailableBufferCapacity(buffersLength)) {
if (currentSpanStreamSendData.getAvailableGatheringComponentsCount() < 2) {
currentSpanStreamSendData.setFlushMode();
}
currentSpanStreamSendData.addBuffer(compositeSpanStreamData.getByteBuffer(), serializer);
spanStreamSendDataList.add(currentSpanStreamSendData);