Index: branches/supervisor/src/main/java/omq/common/util/ParameterQueue.java =================================================================== --- branches/supervisor/src/main/java/omq/common/util/ParameterQueue.java (revision 100) +++ branches/supervisor/src/main/java/omq/common/util/ParameterQueue.java (revision 101) @@ -138,4 +138,29 @@ /** + * 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 */ Index: branches/supervisor/src/main/java/omq/server/InvocationThread.java =================================================================== --- branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 100) +++ branches/supervisor/src/main/java/omq/server/InvocationThread.java (revision 101) @@ -40,5 +40,5 @@ private long lastExec; - private RemoteThreadPool pool; // TODO posar això bé + private RemoteThreadPool pool; // Broker @@ -52,9 +52,10 @@ 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,4 +152,5 @@ logger.error(e); } catch (Exception e) { + e.printStackTrace(); logger.error(e); } Index: branches/supervisor/src/main/java/omq/server/RemoteObject.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 100) +++ branches/supervisor/src/main/java/omq/server/RemoteObject.java (revision 101) @@ -35,5 +35,4 @@ private transient RemoteThreadPool pool; private transient Map>> params; - private transient List invocationList; private static final Map> primitiveClasses = new HashMap>(); @@ -74,15 +73,14 @@ } - // Get num threads to use - int numThreads = Integer.parseInt(env.getProperty(ParameterQueue.NUM_THREADS, "1")); - invocationList = new ArrayList(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,7 +98,5 @@ public void kill() throws IOException { logger.info("Killing objectmq: " + this.getRef()); - for (InvocationThread iThread : invocationList) { - iThread.kill(); - } + pool.kill(); } @@ -205,4 +201,8 @@ } + public RemoteThreadPool getPool() { + return pool; + } + public Properties getEnv() { return env; Index: branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java =================================================================== --- branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java (revision 100) +++ branches/supervisor/src/main/java/omq/server/RemoteThreadPool.java (revision 101) @@ -6,4 +6,6 @@ import java.util.concurrent.atomic.AtomicInteger; +import org.apache.log4j.Logger; + import omq.common.broker.Broker; @@ -11,5 +13,12 @@ import com.rabbitmq.client.Channel; +/** + * + * @author Sergi Toda + * + */ + public class RemoteThreadPool extends Thread { + private static final Logger logger = Logger.getLogger(RemoteThreadPool.class.getName()); private List workers; private AtomicInteger busy; @@ -40,5 +49,20 @@ 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,9 +79,11 @@ 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,5 +92,6 @@ } catch (Exception e) { - + e.printStackTrace(); + logger.error(e); } } @@ -75,10 +102,17 @@ 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,5 +121,14 @@ } + i++; } + } + + public void kill() throws IOException { + killed = true; + for (InvocationThread iThread : workers) { + iThread.kill(); + } + interrupt(); } Index: branches/supervisor/src/test/java/omq/test/remotePool/RemotePoolTest.java =================================================================== --- branches/supervisor/src/test/java/omq/test/remotePool/RemotePoolTest.java (revision 101) +++ branches/supervisor/src/test/java/omq/test/remotePool/RemotePoolTest.java (revision 101) @@ -0,0 +1,60 @@ +package omq.test.remotePool; + +import java.util.Properties; + +import omq.common.broker.Broker; +import omq.common.util.ParameterQueue; + +import org.junit.BeforeClass; +import org.junit.Test; + +public class RemotePoolTest { + private static String NAME = "remotePool.sleep"; + + @BeforeClass + public static void Server() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + env.setProperty(ParameterQueue.MIN_POOL_THREADS, "1"); + env.setProperty(ParameterQueue.MAX_POOL_THREADS, "4"); + env.setProperty(ParameterQueue.POOL_REFRESH_TIME, "1000"); + env.setProperty(ParameterQueue.KEEP_ALIVE_TIME, "2000"); + env.setProperty(ParameterQueue.MAX_MESSAGES_PER_THREAD, "2"); + + Broker broker = new Broker(env); + + SleepImpl sleep = new SleepImpl(); + broker.bind(NAME, sleep); + } + + @Test + public void test() throws Exception { + Properties env = new Properties(); + env.setProperty(ParameterQueue.USER_NAME, "guest"); + env.setProperty(ParameterQueue.USER_PASS, "guest"); + + // Get host info of rabbimq (where it is) + env.setProperty(ParameterQueue.RABBIT_HOST, "127.0.0.1"); + env.setProperty(ParameterQueue.RABBIT_PORT, "5672"); + + Broker broker = new Broker(env); + Sleep sleep = broker.lookup(NAME, Sleep.class); + for (int j = 0; j < 2; j++) { + + for (int i = 0; i < 4; i++) { + sleep.sleep(); + } + Thread.sleep(3000); + for (int i = 0; i < 4; i++) { + sleep.sleep(); + } + Thread.sleep(10000); + } + Thread.sleep(20000 * 1000); + } +} Index: branches/supervisor/src/test/java/omq/test/remotePool/Sleep.java =================================================================== --- branches/supervisor/src/test/java/omq/test/remotePool/Sleep.java (revision 101) +++ branches/supervisor/src/test/java/omq/test/remotePool/Sleep.java (revision 101) @@ -0,0 +1,11 @@ +package omq.test.remotePool; + +import omq.Remote; +import omq.client.annotation.AsyncMethod; +import omq.client.annotation.RemoteInterface; + +@RemoteInterface +public interface Sleep extends Remote { + @AsyncMethod + public void sleep(); +} Index: branches/supervisor/src/test/java/omq/test/remotePool/SleepImpl.java =================================================================== --- branches/supervisor/src/test/java/omq/test/remotePool/SleepImpl.java (revision 101) +++ branches/supervisor/src/test/java/omq/test/remotePool/SleepImpl.java (revision 101) @@ -0,0 +1,24 @@ +package omq.test.remotePool; + +import omq.client.annotation.AsyncMethod; +import omq.server.RemoteObject; + +public class SleepImpl extends RemoteObject implements Sleep { + + /** + * + */ + private static final long serialVersionUID = 1L; + + @Override + @AsyncMethod + public void sleep() { + try { + System.out.println("I'm going to sleep!!!!!!!! ThreadID: " + Thread.currentThread().getId()); + Thread.sleep(1500); + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + +}