Changeset 101 for branches/supervisor

Timestamp:

10/11/13 13:44:19 (11 years ago)

Author:

stoda

Message:

PoolThreadSupervisor? done

TODO: revise all threads

Location:

branches/supervisor/src

Files:

• main/java/omq/common/util/ParameterQueue.java (modified) (1 diff)
• main/java/omq/server/InvocationThread.java (modified) (3 diffs)
• main/java/omq/server/RemoteObject.java (modified) (4 diffs)
• main/java/omq/server/RemoteThreadPool.java (modified) (7 diffs)
• test/java/omq/test/remotePool (added)
• test/java/omq/test/remotePool/RemotePoolTest.java (added)
• test/java/omq/test/remotePool/Sleep.java (added)
• test/java/omq/test/remotePool/SleepImpl.java (added)

branches/supervisor/src/main/java/omq/common/util/ParameterQueue.java

@@ -138 +138 @@
 
         /**
+         * Set the minimum number of threads in a pool
+         */
+        public static String MIN_POOL_THREADS = "omq.min_num_threads";
+
+        /**
+         * Set the maximum number of threads in a pool
+         */
+        public static String MAX_POOL_THREADS = "omq.max_num_threads";
+
+        /**
+         * Set the refresh time to see how many messages are on the queue
+         */
+        public static String POOL_REFRESH_TIME = "omq.refresh_time";
+
+        /**
+         * Set the maximum number of threads in a pool
+         */
+        public static String KEEP_ALIVE_TIME = "omq.keep_alive_time";
+
+        /**
+         * Set the maximum number of messages per thread to create a new one
+         */
+        public static String MAX_MESSAGES_PER_THREAD = "omq.max_messages_per_thread";
+
+        /**
          * Time in milis by default is set in a minute
          */

branches/supervisor/src/main/java/omq/server/InvocationThread.java

@@ -40 +40 @@
         private long lastExec;
 
-        private RemoteThreadPool pool; // TODO posar això bé
+        private RemoteThreadPool pool;
 
         // Broker
@@ -52 +52 @@
         private boolean killed = false;
 
-        public InvocationThread(RemoteObject obj, Broker broker) throws Exception {
+        public InvocationThread(RemoteObject obj) throws Exception {
                 this.obj = obj;
                 this.UID = obj.getRef();
                 this.env = obj.getEnv();
-                this.broker = broker;
+                this.broker = obj.getBroker();
+                this.pool = obj.getPool();
                 this.serializer = broker.getSerializer();
                 this.lastExec = 0;
@@ -151 +152 @@
                                 logger.error(e);
                         } catch (Exception e) {
+                                e.printStackTrace();
                                 logger.error(e);
                         }

branches/supervisor/src/main/java/omq/server/RemoteObject.java

@@ -35 +35 @@
         private transient RemoteThreadPool pool;
         private transient Map<String, List<Class<?>>> params;
-        private transient List<InvocationThread> invocationList;
 
         private static final Map<String, Class<?>> primitiveClasses = new HashMap<String, Class<?>>();
@@ -74 +73 @@
                 }
 
-                // Get num threads to use
-                int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1"));
-                invocationList = new ArrayList<InvocationThread>(numThreads);
-
-                // Start invocation threads
-                for (int i = 0; i < numThreads; i++) {
-                        InvocationThread iThread = new InvocationThread(this, broker);
-                        invocationList.add(iThread);
-                        iThread.start();
-                }
-
+                // Get pool information
+                int minPoolThreads = Integer.parseInt(env.getProperty(ParameterQueue.MIN_POOL_THREADS, "1"));
+                int maxPoolThreads = Integer.parseInt(env.getProperty(ParameterQueue.MAX_POOL_THREADS, "1"));
+                long refresh = Long.parseLong(env.getProperty(ParameterQueue.POOL_REFRESH_TIME, "60000"));
+                long keepAliveTime = Long.parseLong(env.getProperty(ParameterQueue.KEEP_ALIVE_TIME, "30000"));
+                int maxMessagesPerThread = Integer.parseInt(env.getProperty(ParameterQueue.MAX_MESSAGES_PER_THREAD, "5"));
+
+                // Create the pool & start it
+                pool = new RemoteThreadPool(minPoolThreads, maxPoolThreads, refresh, keepAliveTime, maxMessagesPerThread, this, broker);
+                pool.start();
         }
 
@@ -100 +98 @@
         public void kill() throws IOException {
                 logger.info("Killing objectmq: " + this.getRef());
-                for (InvocationThread iThread : invocationList) {
-                        iThread.kill();
-                }
+                pool.kill();
         }
 
@@ -205 +201 @@
         }
 
+        public RemoteThreadPool getPool() {
+                return pool;
+        }
+
         public Properties getEnv() {
                 return env;

branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java

@@ -6 +6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.log4j.Logger;
+
 import omq.common.broker.Broker;
 
@@ -11 +13 @@
 import com.rabbitmq.client.Channel;
 
+/**
+ * 
+ * @author Sergi Toda <sergi.toda@estudiants.urv.cat>
+ * 
+ */
+
 public class RemoteThreadPool extends Thread {
+        private static final Logger logger = Logger.getLogger(RemoteThreadPool.class.getName());
         private List<InvocationThread> workers;
         private AtomicInteger busy;
@@ -40 +49 @@
         public void run() {
 
-                // Crear aquí tots els fils?
+                /*
+                 * Create and start minPoolThreads
+                 */
+                logger.info("ObjectMQ reference: " + obj.getRef() + ", creating: " + minPoolThreads + ", maxPoolThreads: " + maxPoolThreads + ", refresh time: "
+                                + refresh + ", keepAlive: " + keepAliveTime + ", maxMessagesPerThread: " + maxMessagesPerThread);
+
+                for (int i = 0; i < minPoolThreads; i++) {
+                        try {
+                                InvocationThread iThread = new InvocationThread(obj);
+                                workers.add(iThread);
+                                iThread.start();
+                        } catch (Exception e) {
+                                logger.error("Error while creating pool threads", e);
+                                e.printStackTrace();
+                        }
+                }
 
                 while (!killed) {
@@ -55 +79 @@
                                 if (numWorkers < maxPoolThreads && division >= maxMessagesPerThread) {
                                         // Create a new thread
-                                        InvocationThread worker = new InvocationThread(obj, broker);
+                                        System.out.println("Add worker");
+                                        InvocationThread worker = new InvocationThread(obj);
                                         workers.add(worker);
                                         worker.start();
                                 } else if (numWorkers > minPoolThreads && busy.get() < numWorkers) {
                                         // Kill idle threads
+                                        System.out.println("Kill lazy workers, numWorkers = " + numWorkers + ", minPool = " + minPoolThreads + ", busy = " + busy.get());
                                         stopIdleThreads();
                                 }
@@ -66 +92 @@
 
                         } catch (Exception e) {
-
+                                e.printStackTrace();
+                                logger.error(e);
                         }
                 }
@@ -75 +102 @@
                 long now = System.currentTimeMillis();
 
+                int i = 0;
                 for (InvocationThread worker : workers) {
+                        // Ensure there are at least minThreads available
+                        if (workers.size() == minPoolThreads) {
+                                break;
+                        }
                         long lastExec = worker.getLastExecution();
-                        if (worker.isIdle() && (lastExec - now) > keepAliveTime) {
+                        System.out.println("last - now = " + (now - lastExec) + " keep alive = " + keepAliveTime);
+                        if (worker.isIdle() && (now - lastExec) > keepAliveTime) {
                                 // Kill thread
                                 try {
                                         worker.kill();
+                                        workers.remove(i);
                                 } catch (IOException e) {
                                         // TODO Auto-generated catch block
@@ -87 +121 @@
 
                         }
+                        i++;
                 }
+        }
+
+        public void kill() throws IOException {
+                killed = true;
+                for (InvocationThread iThread : workers) {
+                        iThread.kill();
+                }
+                interrupt();
         }