Changeset 101 for branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java

Timestamp:

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

Author:

stoda

Message:

PoolThreadSupervisor? done

TODO: revise all threads

File:

• branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java (modified) (7 diffs)

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();
         }