go back to ThreadPoolServer in QueueService and DataLayerService

Author: Istemi Ekin Akkus

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

@@ -37,14 +37,14 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-
 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.TNonblockingServerSocket;
-import org.apache.thrift.transport.TNonblockingServerTransport;
+import org.apache.thrift.transport.TServerSocket;
+import org.apache.thrift.transport.TServerTransport;
 import org.apache.thrift.transport.TTransportException;
 import org.microfunctions.data_layer.DataLayerService.Iface;
 
@@ -66,8 +66,7 @@ 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_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_MAX_WORKER_THREADS = Integer.MAX_VALUE;
 	public static final int DEFAULT_CLIENT_TIMEOUT = 0;
 	public static final int DEFAULT_MAX_FRAME_LENGTH = Integer.MAX_VALUE;
 
@@ -1626,14 +1625,13 @@ public boolean updateTableTypeCache(String action, String table, String tableTyp
 
 
 	public void start(InetSocketAddress bindAddr) throws TTransportException {
-        TNonblockingServerTransport transport = new TNonblockingServerSocket(bindAddr, DEFAULT_CLIENT_TIMEOUT);
-        TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(transport)
+    	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))
-                .selectorThreads(DEFAULT_SELECTOR_THREADS)
-                .workerThreads(DEFAULT_WORKER_THREADS);
-        server = new TThreadedSelectorServer(args);
+                .maxWorkerThreads(DEFAULT_MAX_WORKER_THREADS);
+        server = new TThreadPoolServer(args);
 
 		LOGGER.info("Listening on "+bindAddr);
 		server.serve();

QueueService/src/main/java/org/microfunctions/queue/local/LocalQueueServer.java

@@ -24,17 +24,17 @@
 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.TNonblockingServerSocket;
-import org.apache.thrift.transport.TNonblockingServerTransport;
+import org.apache.thrift.transport.TServerSocket;
+import org.apache.thrift.transport.TServerTransport;
 import org.apache.thrift.transport.TTransportException;
 import org.microfunctions.queue.local.LocalQueueService.Iface;
 
 public class LocalQueueServer implements Iface, Runnable {
 
-    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_MAX_WORKER_THREADS = Integer.MAX_VALUE;
     public static final int DEFAULT_CLIENT_TIMEOUT = 0;
     public static final int DEFAULT_MAX_FRAME_LENGTH = Integer.MAX_VALUE;
 
@@ -116,26 +116,25 @@ public long freeMemory() throws TException {
         return Runtime.getRuntime().maxMemory() - Runtime.getRuntime().totalMemory() + Runtime.getRuntime().freeMemory();
     }
 
-    public void start (InetSocketAddress bindAddr, int nSelectorThreads, int nWorkerThreads, int clientTimeout) throws TTransportException {
-        TNonblockingServerTransport transport = new TNonblockingServerSocket(bindAddr, clientTimeout);
-        TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(transport)
+    public void start (InetSocketAddress bindAddr, int maxWorkerThreads, int clientTimeout) throws TTransportException {
+        TServerTransport transport = new TServerSocket(bindAddr, clientTimeout);
+        TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport)
                 .transportFactory(new TFramedTransport.Factory(DEFAULT_MAX_FRAME_LENGTH))
                 .protocolFactory(new TCompactProtocol.Factory())
                 .processor(new LocalQueueService.Processor<Iface>(this))
-                .selectorThreads(nSelectorThreads)
-                .workerThreads(nWorkerThreads);
-        server = new TThreadedSelectorServer(args);
+                .maxWorkerThreads(maxWorkerThreads);
+        server = new TThreadPoolServer(args);
 
         LOGGER.info("Starting local queue...");
         server.serve();
     }
 
-    public void start (InetSocketAddress bindAddr, int nSelectorThreads, int nWorkerThreads) throws TTransportException {
-        this.start(bindAddr, nSelectorThreads, nWorkerThreads, DEFAULT_CLIENT_TIMEOUT);
+    public void start (InetSocketAddress bindAddr, int maxWorkerThreads) throws TTransportException {
+        this.start(bindAddr, maxWorkerThreads, DEFAULT_CLIENT_TIMEOUT);
     }
 
     public void start (InetSocketAddress bindAddr) throws TTransportException {
-        this.start(bindAddr, DEFAULT_SELECTOR_THREADS, DEFAULT_WORKER_THREADS, DEFAULT_CLIENT_TIMEOUT);
+        this.start(bindAddr, DEFAULT_MAX_WORKER_THREADS, DEFAULT_CLIENT_TIMEOUT);
     }
 
     public void stop () {