github/arthas
4
0
Fork 0
mirror of https://github.com/alibaba/arthas.git synced 2024-04-21 10:21:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

init http api buffer pool on demand

Browse Source
This commit is contained in:
gongdewei
2020-09-24 14:30:33 +08:00
parent e40025a2df
commit c55cafcbdd
4 changed files with 119 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/com/taobao/arthas/core/server/ArthasBootstrap.java
+1
View File
@@ -417,6 +417,7 @@ public class ArthasBootstrap {
sessionManager.close();
sessionManager = null;
}
httpApiHandler = null;
if (timer != null) {
timer.cancel();
}
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/HttpRequestHandler.java
+1 -1
View File
@@ -73,7 +73,7 @@ public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequ
//handle http restful api
if ("/api".equals(path)) {
response = httpApiHandler.handle(request);
isHttpApiResponse = true;
isHttpApiResponse = (response!=null);
}
//handle webui requests
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/api/HttpApiBufferPool.java
+81
View File
@@ -0,0 +1,81 @@
package com.taobao.arthas.core.shell.term.impl.http.api;
import com.taobao.arthas.core.util.JsonUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* Lazy Buffer pool for http api handler
* @author gongdewei 2020/9/24
*/
public class HttpApiBufferPool {
private int jsonBufferSize = 1024 * 128;
private int poolSize = 8;
private ArrayBlockingQueue<ByteBuf> byteBufPool = new ArrayBlockingQueue<ByteBuf>(poolSize);
private ArrayBlockingQueue<char[]> charsBufPool = new ArrayBlockingQueue<char[]>(poolSize);
private ArrayBlockingQueue<byte[]> bytesPool = new ArrayBlockingQueue<byte[]>(poolSize);
public HttpApiBufferPool() {
JsonUtils.setSerializeWriterBufferThreshold(jsonBufferSize);
for (int i = 0; i < poolSize; i++) {
byteBufPool.offer(Unpooled.buffer(jsonBufferSize));
charsBufPool.offer(new char[jsonBufferSize]);
bytesPool.offer(new byte[jsonBufferSize]);
}
}
public void giveBackByteBuf(ByteBuf byteBuf) {
byteBuf.clear();
if (byteBuf.capacity() == jsonBufferSize) {
if (!byteBufPool.offer(byteBuf)) {
byteBuf.release();
}
} else {
//replace content ByteBuf
byteBuf.release();
if (byteBufPool.remainingCapacity() > 0) {
byteBufPool.offer(Unpooled.buffer(jsonBufferSize));
}
}
}
public ByteBuf pollByteBuf(long timeout, TimeUnit unit) throws InterruptedException {
return byteBufPool.poll(timeout, unit);
}
public boolean offerCharArray(char[] chars) {
return charsBufPool.offer(chars);
}
public char[] pollCharArray() {
return charsBufPool.poll();
}
public char[] pollCharArray(long timeout, TimeUnit unit) throws InterruptedException {
return charsBufPool.poll(timeout, unit);
}
public boolean offerByteArray(byte[] bytes) {
return bytesPool.offer(bytes);
}
public byte[] pollByteArray() {
return bytesPool.poll();
}
public byte[] pollByteArray(long timeout, TimeUnit unit) throws InterruptedException {
return bytesPool.poll(timeout, unit);
}
public int getJsonBufferSize() {
return jsonBufferSize;
}
public int getPoolSize() {
return poolSize;
}
}
core/src/main/java/com/taobao/arthas/core/shell/term/impl/http/api/HttpApiHandler.java
+36 -41
View File
@@ -4,7 +4,12 @@ import com.alibaba.arthas.deps.org.slf4j.Logger;
import com.alibaba.arthas.deps.org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.taobao.arthas.common.PidUtils;
import com.taobao.arthas.core.command.model.*;
import com.taobao.arthas.core.command.model.CommandRequestModel;
import com.taobao.arthas.core.command.model.InputStatus;
import com.taobao.arthas.core.command.model.InputStatusModel;
import com.taobao.arthas.core.command.model.MessageModel;
import com.taobao.arthas.core.command.model.ResultModel;
import com.taobao.arthas.core.command.model.WelcomeModel;
import com.taobao.arthas.core.distribution.PackingResultDistributor;
import com.taobao.arthas.core.distribution.ResultConsumer;
import com.taobao.arthas.core.distribution.ResultDistributor;
@@ -31,8 +36,12 @@ import com.taobao.arthas.core.util.JsonUtils;
import com.taobao.arthas.core.util.StringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.util.CharsetUtil;
import io.termd.core.function.Function;
@@ -40,9 +49,7 @@ import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
@@ -55,34 +62,21 @@ public class HttpApiHandler {
private static final Logger logger = LoggerFactory.getLogger(HttpApiHandler.class);
public static final int DEFAULT_EXEC_TIMEOUT = 30000;
private final SessionManager sessionManager;
private static HttpApiHandler instance;
private final InternalCommandManager commandManager;
private final JobController jobController;
private final HistoryManager historyManager;
private int jsonBufferSize = 1024 * 256;
private int poolSize = 8;
private ArrayBlockingQueue<ByteBuf> byteBufPool = new ArrayBlockingQueue<ByteBuf>(poolSize);
private ArrayBlockingQueue<char[]> charsBufPool = new ArrayBlockingQueue<char[]>(poolSize);
private ArrayBlockingQueue<byte[]> bytesPool = new ArrayBlockingQueue<byte[]>(poolSize);
private volatile HttpApiBufferPool bufferPool;
public HttpApiHandler(HistoryManager historyManager, SessionManager sessionManager) {
this.historyManager = historyManager;
this.sessionManager = sessionManager;
commandManager = this.sessionManager.getCommandManager();
jobController = this.sessionManager.getJobController();
//init buf pool
JsonUtils.setSerializeWriterBufferThreshold(jsonBufferSize);
for (int i = 0; i < poolSize; i++) {
byteBufPool.offer(Unpooled.buffer(jsonBufferSize));
charsBufPool.offer(new char[jsonBufferSize]);
bytesPool.offer(new byte[jsonBufferSize]);
}
}
public HttpResponse handle(FullHttpRequest request) throws Exception {
initBufferPool();
ApiResponse result;
String requestBody = null;
String requestId = null;
@@ -114,14 +108,15 @@ public class HttpApiHandler {
try {
//apply response content buf first
content = byteBufPool.poll(2000, TimeUnit.MILLISECONDS);
content = bufferPool.pollByteBuf(5000, TimeUnit.MILLISECONDS);
if (content == null) {
throw new ApiException("get response content buf failure");
}
content = content.retain();
//apply fastjson buf from pool
charsBuf = charsBufPool.poll();
bytesBuf = bytesPool.poll();
charsBuf = bufferPool.pollCharArray();
bytesBuf = bufferPool.pollByteArray();
if (charsBuf == null || bytesBuf == null) {
throw new ApiException("get json buf failure");
}
@@ -129,46 +124,46 @@ public class HttpApiHandler {
//create http response
DefaultFullHttpResponse response = new DefaultFullHttpResponse(request.protocolVersion(),
HttpResponseStatus.OK, content.retain());
HttpResponseStatus.OK, content);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json; charset=utf-8");
writeResult(response, result);
writeResultAsJson(response, result);
return response;
} catch (Exception e) {
//response is discarded
if (content != null) {
content.release();
byteBufPool.offer(content);
bufferPool.giveBackByteBuf(content);
}
throw e;
} finally {
//give back json buf to pool
JsonUtils.setSerializeWriterBufThreadLocal(null, null);
if (charsBuf != null) {
charsBufPool.offer(charsBuf);
bufferPool.offerCharArray(charsBuf);
}
if (bytesBuf != null) {
bytesPool.offer(bytesBuf);
bufferPool.offerByteArray(bytesBuf);
}
}
}
private void initBufferPool() {
if (bufferPool == null) {
synchronized (this) {
if (bufferPool == null) {
bufferPool = new HttpApiBufferPool();
}
}
}
}
public void onCompleted(DefaultFullHttpResponse httpResponse) {
ByteBuf content = httpResponse.content();
content.clear();
if (content.capacity() == jsonBufferSize) {
if (!byteBufPool.offer(content)) {
content.release();
}
} else {
//replace content ByteBuf
content.release();
if (byteBufPool.remainingCapacity() > 0) {
byteBufPool.offer(Unpooled.buffer(jsonBufferSize));
}
if (bufferPool != null) {
bufferPool.giveBackByteBuf(httpResponse.content());
}
}
private void writeResult(DefaultFullHttpResponse response, Object result) throws IOException {
private void writeResultAsJson(DefaultFullHttpResponse response, Object result) throws IOException {
ByteBufOutputStream out = new ByteBufOutputStream(response.content());
try {
JSON.writeJSONString(out, result);