This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch thrift_test
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/thrift_test by this push:
new c0a4e6c use new service
c0a4e6c is described below
commit c0a4e6ceddf48be1a6fbcf8cddc49b84b974a7dd
Author: jt2594838 <[email protected]>
AuthorDate: Wed Jun 17 11:13:52 2020 +0800
use new service
---
.../org/apache/iotdb/db/service/RPCService.java | 76 +++++++++
.../org/apache/iotdb/db/service/TSServiceImpl.java | 173 +++++++++++++++++----
service-rpc/src/main/thrift/rpc.thrift | 4 +
3 files changed, 224 insertions(+), 29 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
b/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
index f3edd9f..f507118 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
@@ -20,6 +20,10 @@ package org.apache.iotdb.db.service;
import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.iotdb.db.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.db.concurrent.ThreadName;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -29,12 +33,21 @@ import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.exception.runtime.RPCServiceException;
import org.apache.iotdb.service.rpc.thrift.TSIService;
import org.apache.iotdb.service.rpc.thrift.TSIService.Processor;
+import org.apache.iotdb.service.rpc.thrift.TestService;
+import org.apache.iotdb.service.rpc.thrift.TestService.AsyncIface;
+import org.apache.iotdb.service.rpc.thrift.TestService.AsyncProcessor;
+import org.apache.iotdb.service.rpc.thrift.TestService.Iface;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
+import org.apache.thrift.server.THsHaServer;
+import org.apache.thrift.server.THsHaServer.Args;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadPoolServer;
import org.apache.thrift.transport.TFastFramedTransport;
+import org.apache.thrift.transport.TFastFramedTransport.Factory;
+import org.apache.thrift.transport.TNonblockingServerSocket;
+import org.apache.thrift.transport.TNonblockingServerTransport;
import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
@@ -54,6 +67,8 @@ public class RPCService implements RPCServiceMBean, IService {
private RPCServiceThread rpcServiceThread;
private TProtocolFactory protocolFactory;
private Processor<TSIService.Iface> processor;
+ private TestService.Processor<Iface> testSyncProcessor;
+ private TestService.AsyncProcessor<AsyncIface> testAsyncProcessor;
private TThreadPoolServer.Args poolArgs;
private TSServiceImpl impl;
@@ -180,7 +195,11 @@ public class RPCService implements RPCServiceMBean,
IService {
private class RPCServiceThread extends Thread {
private TServerSocket serverTransport;
+ private TServerSocket testServerSyncTransport;
+ private TNonblockingServerTransport testServerAsyncTransport;
private TServer poolServer;
+ private TServer testSyncServer;
+ private TServer testAsyncServer;
private CountDownLatch threadStopLatch;
public RPCServiceThread(CountDownLatch threadStopLatch)
@@ -194,6 +213,8 @@ public class RPCService implements RPCServiceMBean,
IService {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
impl = (TSServiceImpl)
Class.forName(config.getRpcImplClassName()).newInstance();
processor = new TSIService.Processor<>(impl);
+ testAsyncProcessor = new AsyncProcessor<>(impl);
+ testSyncProcessor = new TestService.Processor<>(impl);
this.threadStopLatch = threadStopLatch;
}
@@ -203,6 +224,60 @@ public class RPCService implements RPCServiceMBean,
IService {
logger.info("The RPC service thread begin to run...");
try {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ new Thread(() -> {
+ try {
+ testServerSyncTransport = new TServerSocket(new
InetSocketAddress(config.getRpcAddress(),
+ config.getRpcPort() + 1));
+ } catch (TTransportException e) {
+ e.printStackTrace();
+ }
+ //this is for testing.
+ if (!testServerSyncTransport.getServerSocket().isBound()) {
+ logger.error("The RPC service port is not bound.");
+ }
+ TThreadPoolServer.Args syncServerArgs =
+ new
TThreadPoolServer.Args(testServerSyncTransport).maxWorkerThreads(IoTDBDescriptor.
+
getInstance().getConfig().getRpcMaxConcurrentClientNum()).minWorkerThreads(1)
+ .stopTimeoutVal(
+
IoTDBDescriptor.getInstance().getConfig().getThriftServerAwaitTimeForStopService());
+ syncServerArgs.executorService =
IoTDBThreadPoolFactory.createThriftRpcClientThreadPool(syncServerArgs,
+ ThreadName.RPC_CLIENT.getName());
+ syncServerArgs.processor(testSyncProcessor);
+ syncServerArgs.protocolFactory(protocolFactory);
+ syncServerArgs.transportFactory(new Factory());
+ testSyncServer = new TThreadPoolServer(syncServerArgs);
+ testSyncServer.setServerEventHandler(new
RPCServiceThriftHandler(impl));
+ testSyncServer.serve();
+ }).start();
+
+ new Thread(() -> {
+ try {
+ testServerAsyncTransport = new TNonblockingServerSocket(new
InetSocketAddress(config.getRpcAddress(),
+ config.getRpcPort() + 2));
+ } catch (TTransportException e) {
+ e.printStackTrace();
+ }
+ Args asyncPoolArgs =
+ new
Args(testServerAsyncTransport).maxWorkerThreads(IoTDBDescriptor.
+ getInstance().getConfig().getRpcMaxConcurrentClientNum())
+ .minWorkerThreads(1);
+ asyncPoolArgs.executorService(new
ThreadPoolExecutor(asyncPoolArgs.minWorkerThreads,
+ asyncPoolArgs.maxWorkerThreads,
asyncPoolArgs.getStopTimeoutVal(), asyncPoolArgs.getStopTimeoutUnit(),
+ new SynchronousQueue<>(), new ThreadFactory() {
+ private AtomicLong threadIndex = new AtomicLong(0);
+ @Override
+ public Thread newThread(Runnable r) {
+ return new Thread(r, "TestAsyncClient" +
threadIndex.incrementAndGet());
+ }
+ }));
+ asyncPoolArgs.processor(testAsyncProcessor);
+ asyncPoolArgs.protocolFactory(protocolFactory);
+ asyncPoolArgs.transportFactory(new Factory());
+ testAsyncServer = new THsHaServer(asyncPoolArgs);
+ testAsyncServer.setServerEventHandler(new
RPCServiceThriftHandler(impl));
+ testAsyncServer.serve();
+ }).start();
+
serverTransport = new TServerSocket(new
InetSocketAddress(config.getRpcAddress(),
config.getRpcPort()));
//this is for testing.
@@ -221,6 +296,7 @@ public class RPCService implements RPCServiceMBean,
IService {
poolServer = new TThreadPoolServer(poolArgs);
poolServer.setServerEventHandler(new RPCServiceThriftHandler(impl));
poolServer.serve();
+
} catch (TTransportException e) {
throw new RPCServiceException(String.format("%s: failed to start %s,
because ", IoTDBConstant.GLOBAL_DB_NAME,
getID().getName()), e);
diff --git
a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
index 852f4c7..1ff710a 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/TSServiceImpl.java
@@ -88,7 +88,6 @@ import org.apache.iotdb.db.utils.QueryDataSetUtils;
import org.apache.iotdb.db.utils.SchemaUtils;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.RpcUtils;
-import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.ServerProperties;
import org.apache.iotdb.service.rpc.thrift.TSCancelOperationReq;
@@ -107,7 +106,6 @@ import
org.apache.iotdb.service.rpc.thrift.TSFetchResultsReq;
import org.apache.iotdb.service.rpc.thrift.TSFetchResultsResp;
import org.apache.iotdb.service.rpc.thrift.TSGetTimeZoneResp;
import org.apache.iotdb.service.rpc.thrift.TSIService;
-import org.apache.iotdb.service.rpc.thrift.TSIService.Client;
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertTabletReq;
@@ -119,6 +117,7 @@ import org.apache.iotdb.service.rpc.thrift.TSQueryDataSet;
import org.apache.iotdb.service.rpc.thrift.TSQueryNonAlignDataSet;
import org.apache.iotdb.service.rpc.thrift.TSSetTimeZoneReq;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
+import org.apache.iotdb.service.rpc.thrift.TestService;
import org.apache.iotdb.tsfile.common.conf.TSFileConfig;
import
org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
@@ -129,13 +128,15 @@ import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.apache.thrift.TException;
+import org.apache.thrift.async.AsyncMethodCallback;
+import org.apache.thrift.async.TAsyncClientManager;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.server.ServerContext;
import org.apache.thrift.transport.TFastFramedTransport;
+import org.apache.thrift.transport.TNonblockingSocket;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
-import org.eclipse.jetty.server.session.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -143,7 +144,8 @@ import org.slf4j.LoggerFactory;
/**
* Thrift RPC implementation at server side.
*/
-public class TSServiceImpl implements TSIService.Iface, ServerContext {
+public class TSServiceImpl implements TSIService.Iface, ServerContext,
TestService.Iface,
+ TestService.AsyncIface {
private static final Logger logger =
LoggerFactory.getLogger(TSServiceImpl.class);
private static final String INFO_NOT_LOGIN = "{}: Not login.";
@@ -1492,6 +1494,22 @@ public class TSServiceImpl implements TSIService.Iface,
ServerContext {
return SchemaUtils.getSeriesTypesByString(paths, aggregation);
}
+
+ private static int responseSize = 4;
+ @Override
+ public ByteBuffer testRequest(ByteBuffer load) {
+ ByteBuffer allocate = ByteBuffer.allocate(responseSize);
+ allocate.limit(responseSize);
+ return allocate;
+ }
+
+ @Override
+ public void testRequest(ByteBuffer load, AsyncMethodCallback<ByteBuffer>
resultHandler) {
+ ByteBuffer allocate = ByteBuffer.allocate(responseSize);
+ allocate.limit(responseSize);
+ resultHandler.onComplete(allocate);
+ }
+
@Override
public long requestCommitId(long headerId) {
return 0;
@@ -1638,41 +1656,138 @@ public class TSServiceImpl implements
TSIService.Iface, ServerContext {
return req;
}
- public static void main(String[] args) throws TException {
+ static class TestRequestHandler implements AsyncMethodCallback<ByteBuffer> {
+
+ private volatile boolean hasResult = false;
+ private volatile ByteBuffer result;
+ private volatile Exception e;
+
+ @Override
+ public void onComplete(ByteBuffer response) {
+ synchronized (this) {
+ result = response;
+ hasResult = true;
+ this.notifyAll();
+ }
+ }
+
+ @Override
+ public void onError(Exception exception) {
+ synchronized (this) {
+ e = exception;
+ hasResult = true;
+ this.notifyAll();
+ }
+ }
+
+ public ByteBuffer get() throws Exception {
+ synchronized (this) {
+ while (!hasResult) {
+ try {
+ this.wait();
+ } catch (InterruptedException ex) {
+ ex.printStackTrace();
+ break;
+ }
+ }
+ if (e != null) {
+ throw e;
+ } else {
+ return result;
+ }
+ }
+ }
+ }
+
+ // args : <ip> <port> <addAnEmptyRequestBeforeEachRequest> <loadSize>
<useAsyncClient>
+ // e.g. : 127.0.0.1 6668 0 1000 true
+ // 127.0.0.1 6669 0 1000 false
+ public static void main(String[] args) {
System.out.println("main ...");
AtomicInteger globalCnt = new AtomicInteger();
- AtomicLong globalTime = new AtomicLong();
long startTime = System.currentTimeMillis();
+
+ String ip = args[0];
+ // the synchronized test service is bound to rpcPort + 1 and the
asynchronized test service
+ // is bound to rpcPort + 2, so by default if you test sync client you
should use 6668 and
+ // 6669 for async client
+ int port = Integer.parseInt(args[1]);
int clientNum = Integer.parseInt(args[2]);
int addSync = Integer.parseInt(args[3]);
- int pointNum = Integer.parseInt(args[4]);
- List<TSInsertRecordsReq> reqs = new ArrayList<>(clientNum);
+ int loadSize = Integer.parseInt(args[4]);
+ boolean useSyncClient = Boolean.parseBoolean(args[5]);
+
+ if (useSyncClient) {
+ System.out.println("Test sync");
+ } else {
+ System.out.println("Test async");
+ }
+
ExecutorService pool = Executors.newFixedThreadPool(clientNum);
for (int i = 0; i < clientNum; i++) {
- reqs.add(prepareData(i * pointNum, pointNum));
- globalTime.addAndGet(pointNum);
- int finalI = i;
pool.submit(() -> {
- TTransport tTransport = new TFastFramedTransport(new TSocket(args[0],
- Integer.parseInt(args[1])));
- TProtocol protocol = new TBinaryProtocol(tTransport);
- Client client = new Client(protocol);
- tTransport.open();
- int cnt = 0;
- while (true) {
- if (addSync > 0) {
- client.requestCommitId(0);
- }
- client.insertRecords(finalReq(reqs.get(finalI),
globalTime.getAndAdd(pointNum), pointNum));
- cnt ++;
- //System.out.println(cnt);
- if (cnt % (100000 / clientNum) == 0) {
- int gc = globalCnt.addAndGet(100000 / clientNum);
- long consumedTime = System.currentTimeMillis() - startTime;
- System.out.println(String.format("%d request complete, time %d,
speed %f", gc,
- consumedTime, (double) gc / consumedTime));
+ try {
+ if (useSyncClient) {
+ TTransport tTransport = new TFastFramedTransport(new TSocket(ip,
port));
+ TProtocol protocol = new TBinaryProtocol(tTransport);
+ TestService.Client client = new TestService.Client(protocol);
+ tTransport.open();
+ int cnt = 0;
+ long lastBatchTime = System.currentTimeMillis();
+ while (true) {
+ if (addSync > 0) {
+ client.testRequest(ByteBuffer.allocate(0));
+ }
+ ByteBuffer allocate = ByteBuffer.allocate(loadSize);
+ allocate.limit(loadSize);
+ client.testRequest(allocate);
+ cnt ++;
+ //System.out.println(cnt);
+ if (cnt % (100000 / clientNum) == 0) {
+ int gc = globalCnt.addAndGet(100000 / clientNum);
+ long consumedTime = System.currentTimeMillis() - startTime;
+ long batchConsumedTime = System.currentTimeMillis() -
lastBatchTime;
+ lastBatchTime = System.currentTimeMillis();
+ System.out.println(String.format("%d request complete, time
%d, avg speed %f, "
+ + "batch speed %f",
+ gc, consumedTime, (double) gc / consumedTime,
+ (double) (100000 / clientNum) / batchConsumedTime));
+ }
+ }
+ } else {
+ TestService.AsyncClient client =
+ new TestService.AsyncClient(new TBinaryProtocol.Factory(),
+ new TAsyncClientManager(), new TNonblockingSocket(ip, port));
+ int cnt = 0;
+ long lastBatchTime = System.currentTimeMillis();
+ while (true) {
+ if (addSync > 0) {
+ TestRequestHandler handler = new TestRequestHandler();
+ client.testRequest(ByteBuffer.allocate(0), handler);
+ handler.get();
+ }
+ TestRequestHandler handler = new TestRequestHandler();
+ ByteBuffer allocate = ByteBuffer.allocate(loadSize);
+ allocate.limit(loadSize);
+ client.testRequest(allocate, handler);
+ handler.get();
+ cnt ++;
+ //System.out.println(cnt);
+ if (cnt % (100000 / clientNum) == 0) {
+ int gc = globalCnt.addAndGet(100000 / clientNum);
+ long consumedTime = System.currentTimeMillis() - startTime;
+ long batchConsumedTime = System.currentTimeMillis() -
lastBatchTime;
+ lastBatchTime = System.currentTimeMillis();
+ System.out.println(String.format("%d request complete, time
%d, avg speed %f, "
+ + "batch speed %f",
+ gc, consumedTime, (double) gc / consumedTime,
+ (double) (100000 / clientNum) / batchConsumedTime));
+ }
+ }
}
+ } catch (Exception e) {
+ e.printStackTrace();
}
});
}
diff --git a/service-rpc/src/main/thrift/rpc.thrift
b/service-rpc/src/main/thrift/rpc.thrift
index e68d2d5..2814ebe 100644
--- a/service-rpc/src/main/thrift/rpc.thrift
+++ b/service-rpc/src/main/thrift/rpc.thrift
@@ -315,3 +315,7 @@ service TSIService {
i64 requestCommitId(1:i64 headerId)
}
+
+service TestService {
+ binary testRequest(1:binary load)
+}
