changed DataLayerService's server type from TThreadPoolServer to TThreadedSelectorServer for memory efficiency

Author: ruichuan

DataLayerService/src/main/java/org/microfunctions/data_layer/DataLayerServer.java

@@ -41,10 +41,10 @@
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TCompactProtocol;
 import org.apache.thrift.server.TServer;
-import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.server.TThreadedSelectorServer;
 import org.apache.thrift.transport.TFramedTransport;
-import org.apache.thrift.transport.TServerSocket;
-import org.apache.thrift.transport.TServerTransport;
+import org.apache.thrift.transport.TNonblockingServerSocket;
+import org.apache.thrift.transport.TNonblockingServerTransport;
 import org.apache.thrift.transport.TTransportException;
 import org.microfunctions.data_layer.DataLayerService.Iface;
 
@@ -66,7 +66,8 @@ public class DataLayerServer implements Iface, Callable<Object> {
 	private static final List<AbstractMap.SimpleEntry<String, Integer>> NO_KEYSPACES = new ArrayList<AbstractMap.SimpleEntry<String, Integer>>(0);
 	private static final List<String> NO_KEYS = new ArrayList<String>(0);
 
-	public static final int DEFAULT_MAX_WORKER_THREADS = Integer.MAX_VALUE;
+    public static final int DEFAULT_SELECTOR_THREADS = Math.max(2, 2 * Runtime.getRuntime().availableProcessors());
+    public static final int DEFAULT_WORKER_THREADS = Math.max(4, 4 * Runtime.getRuntime().availableProcessors());
 	public static final int DEFAULT_CLIENT_TIMEOUT = 0;
 	public static final int DEFAULT_MAX_FRAME_LENGTH = Integer.MAX_VALUE;
 
@@ -1625,13 +1626,15 @@ public boolean updateTableTypeCache(String action, String table, String tableTyp
 
 
 	public void start(InetSocketAddress bindAddr) throws TTransportException {
-		TServerTransport transport = new TServerSocket(bindAddr, DEFAULT_CLIENT_TIMEOUT);
-		TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport)
-				.transportFactory(new TFramedTransport.Factory(DEFAULT_MAX_FRAME_LENGTH))
-				.protocolFactory(new TCompactProtocol.Factory())
-				.processor(new DataLayerService.Processor<Iface>(this))
-				.maxWorkerThreads(DEFAULT_MAX_WORKER_THREADS);
-		server = new TThreadPoolServer(args);
+        TNonblockingServerTransport transport = new TNonblockingServerSocket(bindAddr, DEFAULT_CLIENT_TIMEOUT);
+        TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(transport)
+                .transportFactory(new TFramedTransport.Factory(DEFAULT_MAX_FRAME_LENGTH))
+                .protocolFactory(new TCompactProtocol.Factory())
+                .processor(new DataLayerService.Processor<Iface>(this))
+                .selectorThreads(DEFAULT_SELECTOR_THREADS)
+                .workerThreads(DEFAULT_WORKER_THREADS);
+        server = new TThreadedSelectorServer(args);
+
 		LOGGER.info("Listening on "+bindAddr);
 		server.serve();
 	}