Merge remote-tracking branch 'xiao/storm-425'

Nathan Marz · Nathan Marz · commit 78f18123aa1f · 2013-01-04T23:10:22.000-08:00
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -36,6 +36,8 @@ ui.port: 8080
 ui.childopts: "-Xmx768m"
 
 drpc.port: 3772
+drpc.worker.threads: 64
+drpc.queue.size: 128
 drpc.invocations.port: 3773
 drpc.request.timeout.secs: 600
 
diff --git a/src/clj/backtype/storm/daemon/drpc.clj b/src/clj/backtype/storm/daemon/drpc.clj
@@ -6,7 +6,7 @@
   (:import [backtype.storm.generated DistributedRPC DistributedRPC$Iface DistributedRPC$Processor
             DRPCRequest DRPCExecutionException DistributedRPCInvocations DistributedRPCInvocations$Iface
             DistributedRPCInvocations$Processor])
-  (:import [java.util.concurrent Semaphore ConcurrentLinkedQueue])
+  (:import [java.util.concurrent Semaphore ConcurrentLinkedQueue ThreadPoolExecutor ArrayBlockingQueue TimeUnit])
   (:import [backtype.storm.daemon Shutdownable])
   (:import [java.net InetAddress])
   (:use [backtype.storm bootstrap config log])
@@ -100,6 +100,8 @@
 (defn launch-server!
   ([]
     (let [conf (read-storm-config)
+          worker-threads (int (conf DRPC-WORKER-THREADS))
+          queue-size (int (conf DRPC-QUEUE-SIZE))
           service-handler (service-handler)
           ;; requests and returns need to be on separate thread pools, since calls to
           ;; "execute" don't unblock until other thrift methods are called. So if 
@@ -108,6 +110,8 @@
           handler-server (THsHaServer. (-> (TNonblockingServerSocket. (int (conf DRPC-PORT)))
                                              (THsHaServer$Args.)
                                              (.workerThreads 64)
+                                             (.executorService (ThreadPoolExecutor. worker-threads worker-threads 
+                                                                 60 TimeUnit/SECONDS (ArrayBlockingQueue. queue-size)))
                                              (.protocolFactory (TBinaryProtocol$Factory.))
                                              (.processor (DistributedRPC$Processor. service-handler))
                                              ))
diff --git a/src/jvm/backtype/storm/Config.java b/src/jvm/backtype/storm/Config.java
@@ -222,6 +222,16 @@ public class Config extends HashMap<String, Object> {
      */
     public static String DRPC_PORT = "drpc.port";
     
+    /**
+     * DRPC thrift server worker threads 
+     */
+    public static String DRPC_WORKER_THREADS = "drpc.worker.threads";
+
+    /**
+     * DRPC thrift server queue size 
+     */
+    public static String DRPC_QUEUE_SIZE = "drpc.queue.size";
+    
     /**
      * This port on Storm DRPC is used by DRPC topologies to receive function invocations and send results back. 
      */
