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

Author: ruichuan

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

@@ -40,11 +40,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 +65,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 +1625,14 @@ 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)
+	    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))
-                .maxWorkerThreads(DEFAULT_MAX_WORKER_THREADS);
-        server = new TThreadPoolServer(args);
+                .selectorThreads(DEFAULT_SELECTOR_THREADS)
+                .workerThreads(DEFAULT_WORKER_THREADS);
+        server = new TThreadedSelectorServer(args);
 
 		LOGGER.info("Listening on "+bindAddr);
 		server.serve();