pre-release 2020.1 final beta

Author: yuriy-glotanov

src/main/java/su/interference/core/Config.java

@@ -59,6 +59,7 @@ public class Config {
     private static final String P_DISKIO_MODE = "diskio.mode";
     private static final String P_SYNC_LOCK_ENABLE = "sync.lock.enable";
     private static final String P_SYNC_PERIOD = "sync.period";
+    private static final String P_RETRIEVE_QUEUE_SIZE = "retrieve.queue.size";
     private static final String P_CODEPAGE = "codepage";
     private static final String P_DATEFORMAT = "dateformat";
     private static final int MAX_NODE_ID = 64;
@@ -80,7 +81,8 @@ public class Config {
     private static final int FILES_AMOUNT_DEFAULT=4;
     private static final String DISKIO_MODE_DEFAULT="rws";
     private static final boolean SYNC_LOCK_ENABLE_DEFAULT=false;
-    private static final int SYNC_PERIOD_DEFAULT=5;
+    private static final int SYNC_PERIOD_DEFAULT=2000;
+    private static final int RETRIEVE_QUEUE_SIZE_DEFAULT=10000;
     private static final String CODEPAGE_DEFAULT="UTF-8";
     private static final String DATEFORMAT_DEFAULT="dd.MM.yyyy";
 
@@ -101,6 +103,7 @@ public class Config {
     public final String DISKIO_MODE;
     public final boolean SYNC_LOCK_ENABLE;
     public final int SYNC_PERIOD;
+    public final int RETRIEVE_QUEUE_SIZE;
     public final String CODEPAGE;
     public final String DATEFORMAT;
 
@@ -158,6 +161,7 @@ private Config() {
             DISKIO_MODE = validateDiskioMode(p.getProperty(P_DISKIO_MODE));
             SYNC_LOCK_ENABLE = validateSyncLock(p.getProperty(P_SYNC_LOCK_ENABLE));
             SYNC_PERIOD = validateSyncPeriod(p.getProperty(P_SYNC_PERIOD));
+            RETRIEVE_QUEUE_SIZE = validateQueueSize(p.getProperty(P_RETRIEVE_QUEUE_SIZE));
         } else {
             DB_PATH = DB_PATH_DEFAULT;
             JOURNAL_PATH = JOURNAL_PATH_DEFAULT;
@@ -178,6 +182,7 @@ private Config() {
             DISKIO_MODE = DISKIO_MODE_DEFAULT;
             SYNC_LOCK_ENABLE = SYNC_LOCK_ENABLE_DEFAULT;
             SYNC_PERIOD = SYNC_PERIOD_DEFAULT;
+            RETRIEVE_QUEUE_SIZE = RETRIEVE_QUEUE_SIZE_DEFAULT;
         }
         System.setProperty("com.sun.management.jmxremote.port","8111");
         System.setProperty("com.sun.management.jmxremote.authenticate","false");
@@ -324,7 +329,19 @@ private static String validateDiskioMode(String mode) {
     private static int validateSyncPeriod(String value) {
         try {
             int p = Integer.valueOf(value);
-            if (p > 0 && p < 10000) {
+            if (p > 9 && p < 60001) {
+                return p;
+            }
+        } catch(NumberFormatException e) {
+            logger.error("sync period value is not valid - use default value");
+        }
+        return SYNC_PERIOD_DEFAULT;
+    }
+
+    private static int validateQueueSize(String value) {
+        try {
+            int p = Integer.valueOf(value);
+            if (p > 0 && p < 100000000) {
                 return p;
             }
         } catch(NumberFormatException e) {

src/main/java/su/interference/core/ManagedCallable.java

@@ -1,7 +1,36 @@
+/**
+ The MIT License (MIT)
+
+ Copyright (c) 2010-2019 head systems, ltd
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+ */
+
 package su.interference.core;
 
 import java.util.concurrent.Callable;
 
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
 public interface ManagedCallable<T> extends Callable<T> {
     void stop();
 }

src/main/java/su/interference/core/RetrieveQueue.java

@@ -1,7 +1,36 @@
+/**
+ The MIT License (MIT)
+
+ Copyright (c) 2010-2019 head systems, ltd
+
+ Permission is hereby granted, free of charge, to any person obtaining a copy of
+ this software and associated documentation files (the "Software"), to deal in
+ the Software without restriction, including without limitation the rights to
+ use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+ the Software, and to permit persons to whom the Software is furnished to do so,
+ subject to the following conditions:
+
+ The above copyright notice and this permission notice shall be included in all
+ copies or substantial portions of the Software.
+
+ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+ FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+ COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+ IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+ */
+
 package su.interference.core;
 
 import java.util.concurrent.LinkedBlockingQueue;
 
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
 public class RetrieveQueue {
     private final LinkedBlockingQueue<Chunk> q;
     private final ManagedCallable r;
@@ -17,7 +46,7 @@ public Object poll() {
             if (retrieve) {
                 Chunk c = q.take();
                 if (c.isTerminate()) {
-                    retrieve = false;
+                    stop();
                     return null;
                 }
                 return c.getEntity();
@@ -33,7 +62,7 @@ public Chunk cpoll() {
             if (retrieve) {
                 Chunk c = q.take();
                 if (c.isTerminate()) {
-                    retrieve = false;
+                    stop();
                     return null;
                 }
                 return c;

src/main/java/su/interference/core/SyncQueue.java

@@ -141,7 +141,7 @@ public void run () {
             }
 
             try {
-                final int period = Config.getConfig().SYNC_PERIOD*1000;
+                final int period = Config.getConfig().SYNC_PERIOD;
                 Thread.sleep(period);
             } catch (InterruptedException e) {
                 e.printStackTrace();

src/main/java/su/interference/persistent/Session.java

@@ -35,17 +35,15 @@ this software and associated documentation files (the "Software"), to deal in
 import su.interference.proxy.RSProxyFactory;
 import su.interference.sql.ResultSet;
 import su.interference.sql.SQLColumn;
+import su.interference.sql.SQLCursor;
 import su.interference.sql.SQLSelect;
 
 import java.util.*;
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Modifier;
 import java.net.MalformedURLException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.*;
 import javax.persistence.*;
 
 /**
@@ -285,9 +283,18 @@ public Object find (Class c, long id) throws Exception {
 
     protected synchronized RetrieveQueue getContentQueue(Table t) {
         if (t != null) {
-            retrieveQueue = t.getContentQueue(this);
-            Future f = rqpool.submit(retrieveQueue.getR());
-            return retrieveQueue;
+            try {
+                retrieveQueue = t.getContentQueue(this);
+                Future f = rqpool.submit(retrieveQueue.getR());
+                Object o = f.get();
+                return retrieveQueue;
+            } catch (Exception e) {
+                if (e instanceof ExecutionException) {
+                    e.getCause().printStackTrace();
+                } else {
+                    e.printStackTrace();
+                }
+            }
         }
         return null;
     }
@@ -299,6 +306,10 @@ public void closeQueue() {
         retrieveQueue = null;
     }
 
+    public void closeStreamQueue() {
+        SQLCursor.removeStreamQueue(this);
+    }
+
     public void close() {
         rqpool.shutdownNow();
         streampool.isShutdown();

src/main/java/su/interference/persistent/Table.java

@@ -1291,7 +1291,7 @@ protected RetrieveQueue getContentQueue(Session s) {
     protected synchronized RetrieveQueue getTableContentQueue(Session s) {
         final Table t = this;
         final int ptr = 0;
-        final LinkedBlockingQueue<Chunk> q = new LinkedBlockingQueue<>();
+        final LinkedBlockingQueue<Chunk> q = new LinkedBlockingQueue<>(Config.getConfig().RETRIEVE_QUEUE_SIZE);
         final AtomicBoolean stopped = new AtomicBoolean();
         final Chunk tc = new DataChunk();
         tc.setTerminate(true);
@@ -1330,7 +1330,7 @@ public void stop() {
     protected synchronized RetrieveQueue getIndexContentQueue(Session s) {
         final Table t = this;
         final int ptr = 0;
-        final LinkedBlockingQueue<Chunk> q = new LinkedBlockingQueue<>();
+        final LinkedBlockingQueue<Chunk> q = new LinkedBlockingQueue<>(Config.getConfig().RETRIEVE_QUEUE_SIZE);
         final AtomicBoolean stopped = new AtomicBoolean();
         final Chunk tc = new DataChunk();
         tc.setTerminate(true);
@@ -1506,11 +1506,6 @@ public DataChunk getChunkByEntity (Object o, Session s) throws IOException, Invo
                 }
             }
         } else {
-            if (this.idfield!=null&&this.generatedfield!=null) {
-                if (this.idfield.equals(this.generatedfield)) {
-                    return null;
-                }
-            }
             final EntityContainer to = (EntityContainer)o;
             final Table idt = getFirstIndexByIdColumn();
             if (to.getDataChunk()==null) {

src/main/java/su/interference/proxy/GenericResultImpl.java

@@ -62,6 +62,7 @@ public String toString() {
                     sb.append(m.getName().substring(3).substring(1));
                     sb.append(":");
                     sb.append(m.invoke(this, null));
+                    sb.append(" ");
                 }
             }
         } catch (Exception e) {

src/main/java/su/interference/sql/CList.java

@@ -29,13 +29,11 @@ this software and associated documentation files (the "Software"), to deal in
 import su.interference.sqlexception.MissingTablesDescription;
 import su.interference.sqlexception.AmbiguousColumnName;
 import su.interference.sqlexception.InvalidColumnDescription;
-import su.interference.core.Instance;
 import su.interference.exception.InternalException;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.lang.reflect.Field;
-import java.net.MalformedURLException;
 import java.util.List;
 
 /**

src/main/java/su/interference/sql/FrameGroupTask.java

@@ -81,15 +81,17 @@ public void run() {
                 }
             }
             //return last group after stop
+/*
             final DataChunk gdc = sqlg.add(null);
             if (gdc != null) {
                 Object oo = gdc.getEntity(gtable);
                 target.persist(oo, s);
             }
+*/
 
 
         } catch (Exception e) {
-            ((StreamQueue) target).stop(s);
+            ((StreamQueue) target).stop();
             logger.error("", e);
         }
     }

src/main/java/su/interference/sql/FrameJoinTask.java

@@ -30,9 +30,6 @@ this software and associated documentation files (the "Software"), to deal in
 import su.interference.metrics.Metrics;
 import su.interference.persistent.*;
 import su.interference.proxy.GenericResult;
-import su.interference.transport.SQLEvent;
-import su.interference.transport.TransportContext;
-import su.interference.transport.TransportEvent;
 
 import java.util.List;
 import java.util.concurrent.Callable;