[Openrpg2 dev] SF.net SVN: openrpg:[101] trunk/src/openrpg2/common/core/route/ RouteThreadPool.java

[<sno...@us...>]

Revision: 101
          http://sourceforge.net/p/openrpg/svn/101
Author:   snowdog_
Date:     2013-09-17 00:06:30 +0000 (Tue, 17 Sep 2013)
Log Message:
-----------
Traced the startup issues to what appear to be stalling routing threads. Available pool of threads would empty and result in a NullPointer Exception in message sending code. Reworked RouteThreadPool to spawn more worker if required and reworked worker assignment method. Suspect actual bug to exist in either the RouteServiceThread or in the network layers handoff routines.

Modified Paths:
--------------
    trunk/src/openrpg2/common/core/route/RouteThreadPool.java

Modified: trunk/src/openrpg2/common/core/route/RouteThreadPool.java
===================================================================
--- trunk/src/openrpg2/common/core/route/RouteThreadPool.java	2013-09-16 21:10:47 UTC (rev 100)
+++ trunk/src/openrpg2/common/core/route/RouteThreadPool.java	2013-09-17 00:06:30 UTC (rev 101)
@@ -25,6 +25,7 @@
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.logging.Level;
 import java.util.logging.Logger;
 import openrpg2.common.core.engine.NetworkedModuleLocator;
 import openrpg2.common.core.network.NetworkMessageRelay;
@@ -41,6 +42,9 @@
     private NetworkedModuleLocator moduleLocator;
     private NetworkMessageRelay messageRelay;
     private int nextId = 1;
+    private final static int MAX_THREADS = 10;
+    private final static int MAX_ASSIGNMENT_ATTEMPTS = 10;
+    private final static int ASSIGNMENT_DELAY = 100;
     private Logger log = Logger.getLogger(this.getClass().getName());
     
     
@@ -50,18 +54,14 @@
     public RouteThreadPool(NetworkedModuleLocator locatorRef, NetworkMessageRelay msgRelay,  int initialSize) {
         moduleLocator = locatorRef;
         messageRelay = msgRelay;
+        if( initialSize > RouteThreadPool.MAX_THREADS){ initialSize = RouteThreadPool.MAX_THREADS; }
         for(int i = 0; i < initialSize; i++){
-            RouteServiceThread nst = spawnServiceThread();
-            synchronized(idle){
-                idle.add(nst);
-            }
-            synchronized(threadHandles){  
-                threadHandles.add(nst);
-            }
+            addServiceThreadToPool();
         }
-
     }
     
+    
+    
     protected void shutdownThreads(){
       Iterator i = threadHandles.iterator();
       while (i.hasNext()){
@@ -74,13 +74,28 @@
     
     protected RouteServiceThread getServiceThread(){
         RouteServiceThread nst = null;
-        synchronized(idle){
-            if( idle.size() > 0){
-                nst = (RouteServiceThread)idle.remove(0);
-            }else{
-                //System.out.println("RouteThreadPool::getServiceThread() No threads available");
+        int attempts = 0;
+        do{
+            synchronized(idle){
+                if( idle.size() > 0){
+                    nst = (RouteServiceThread)idle.remove(0);
+                }
             }
-        }
+            if( nst == null){
+                attempts += 1;
+                //System.out.println("RouteThreadPool::getServiceThread() No threads available; try #"+attempts);
+                if(( attempts > RouteThreadPool.MAX_ASSIGNMENT_ATTEMPTS)&&(threadHandles.size() < RouteThreadPool.MAX_THREADS)){
+                    addServiceThreadToPool();
+                }
+                if(( attempts > RouteThreadPool.MAX_ASSIGNMENT_ATTEMPTS)&&(threadHandles.size() >= RouteThreadPool.MAX_THREADS)){
+                    log.severe("ROUTING ERROR: No available routing threads!!");
+                    return null; 
+                }
+                try {
+                    Thread.sleep(RouteThreadPool.ASSIGNMENT_DELAY); //sleep while a routing threat frees up....
+                } catch (InterruptedException ex) {}
+            }
+        }while( nst == null);
         return nst;
     }
     
@@ -97,11 +112,20 @@
     private RouteServiceThread spawnServiceThread(){
         RouteServiceThread nst = new RouteServiceThread(this, this.messageRelay, this.moduleLocator, nextId);
         nst.setName("RouteThread-"+nextId);
-       log.finer("spawnServiceThread creating new thread ("+nst.getName()+")");
+       log.finer("spawning new service thread ("+nst.getName()+")");
         nextId++;
         nst.start();
         return nst;
     }
     
+    private void addServiceThreadToPool(){
+        RouteServiceThread nst = spawnServiceThread();
+        synchronized(idle){
+            idle.add(nst);
+        }
+        synchronized(threadHandles){  
+            threadHandles.add(nst);
+        }
+    }
     
 }

This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.