minor bugfix / improvements. Multithreading fixes

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>de.caluga</groupId>
     <artifactId>morphium</artifactId>
-    <version>2.2.14</version>
+    <version>2.2.15</version>
     <packaging>jar</packaging>
     <parent>
         <groupId>org.sonatype.oss</groupId>

src/de/caluga/morphium/Morphium.java

@@ -115,6 +115,9 @@ public Morphium(MorphiumConfig cfg) {
 
     }
 
+    public ThreadPoolExecutor getAsyncOperationsThreadPool() {
+        return asyncOperationsThreadPool;
+    }
     public void setConfig(MorphiumConfig cfg) {
         if (config != null) {
             throw new RuntimeException("Cannot change config!");

src/de/caluga/morphium/query/QueryFactory.java

@@ -18,7 +18,7 @@ public interface QueryFactory {
 
     public void setQueryImpl(Class<? extends Query> queryImpl);
 
-    void setExecutor(ThreadPoolExecutor ex);
+    public void setExecutor(ThreadPoolExecutor ex);
 
-    ThreadPoolExecutor getExecutor();
+    public ThreadPoolExecutor getExecutor(Morphium m);
 }

src/de/caluga/morphium/query/QueryFactoryImpl.java

@@ -15,7 +15,7 @@
  */
 public class QueryFactoryImpl implements QueryFactory {
     private Class<? extends Query> queryImpl;
-    ThreadPoolExecutor executor = null;
+    private ThreadPoolExecutor executor = null;
 
     public QueryFactoryImpl() {
     }
@@ -30,9 +30,9 @@ public QueryFactoryImpl(Class<? extends Query> qi) {
     }
 
     @Override
-    public ThreadPoolExecutor getExecutor() {
+    public ThreadPoolExecutor getExecutor(Morphium m) {
         if (executor == null) {
-            executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
+            executor = new ThreadPoolExecutor(m.getConfig().getMaxConnections() / 2, (int) (m.getConfig().getMaxConnections() * m.getConfig().getBlockingThreadsMultiplier() * 0.9),
                     60L, TimeUnit.SECONDS,
                     new SynchronousQueue<Runnable>());
         }
@@ -55,8 +55,7 @@ public <T> Query<T> createQuery(Morphium m, Class<? extends T> type) {
             Query<T> q = queryImpl.newInstance();
             q.setMorphium(m);
             q.setType(type);
-            q.setExecutor(getExecutor());
-
+            q.setExecutor(getExecutor(m));
             return q;
         } catch (InstantiationException e) {
             throw new RuntimeException(e);

src/de/caluga/morphium/query/QueryImpl.java

@@ -12,9 +12,7 @@
 import java.lang.reflect.Field;
 import java.text.SimpleDateFormat;
 import java.util.*;
-import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
 
 /**
  * User: Stpehan Bösebeck
@@ -83,11 +81,6 @@ public ServerAddress getServer() {
     }
 
     public ThreadPoolExecutor getExecutor() {
-        if (executor == null) {
-            executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
-                    60L, TimeUnit.SECONDS,
-                    new SynchronousQueue<Runnable>());
-        }
         return executor;
     }
 

src/de/caluga/morphium/writer/MorphiumWriterImpl.java

@@ -13,7 +13,7 @@
 import java.lang.reflect.Method;
 import java.text.SimpleDateFormat;
 import java.util.*;
-import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
@@ -33,9 +33,7 @@ public class MorphiumWriterImpl implements MorphiumWriter {
     private Morphium morphium;
     private int maximumRetries = 10;
     private int pause = 250;
-    private ThreadPoolExecutor executor = new ThreadPoolExecutor(0, Integer.MAX_VALUE,
-            60L, TimeUnit.SECONDS,
-            new ArrayBlockingQueue<Runnable>(1000, true));
+    private ThreadPoolExecutor executor = null;
 
     @Override
     public void setMaximumQueingTries(int n) {
@@ -51,8 +49,9 @@ public void setPauseBetweenTries(int p) {
     public void setMorphium(Morphium m) {
         morphium = m;
         if (m != null) {
-            executor.setCorePoolSize(m.getConfig().getMaxConnections() / 2);
-            executor.setMaximumPoolSize((int) (m.getConfig().getMaxConnections() * m.getConfig().getBlockingThreadsMultiplier() * 0.9));
+            executor = new ThreadPoolExecutor(m.getConfig().getMaxConnections() / 2, (int) (m.getConfig().getMaxConnections() * m.getConfig().getBlockingThreadsMultiplier() * 0.9),
+                    60L, TimeUnit.SECONDS,
+                    new SynchronousQueue<Runnable>());
         }
     }
 

test/de/caluga/test/mongo/suite/CacheSyncTest.java

@@ -165,7 +165,7 @@ public void idCacheTest() throws Exception {
         dur = System.currentTimeMillis() - start;
         log.info("Storing with synchronizer: " + dur + " ms");
 
-        Thread.sleep(10000);
+        Thread.sleep(15000);
         start = System.currentTimeMillis();
         int notFoundCounter = 0;
         for (int i = 0; i < 100; i++) {

test/de/caluga/test/mongo/suite/WriteBufferCountTest.java

@@ -6,6 +6,8 @@
 import de.caluga.morphium.query.Query;
 import org.junit.Test;
 
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadMXBean;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -57,4 +59,13 @@ private int waitForWriteProcessToBeScheduled() {
         }
         return c;
     }
+
+
+    @Test
+    public void threadNumberTest() throws Exception {
+        ThreadMXBean thbean = ManagementFactory.getThreadMXBean();
+        log.info("Running threads: " + thbean.getThreadCount());
+        assert (thbean.getThreadCount() < 1000);
+
+    }
 }