pre-release 2020.3

Author: yuriy-glotanov

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

@@ -107,6 +107,20 @@ public class Config {
     public final String CODEPAGE;
     public final String DATEFORMAT;
     public final int TEST_DISTRIBUTE_MODE = 1;
+    public final int CHECK_AVAIL_FRAME_TIMEOUT = 200;
+    // transport
+    public final int REMOTE_SYNC_TIMEOUT = 60000;
+    public final int READ_BUFFER_SIZE = 33554432;
+    public final int WRITE_BUFFER_SIZE = 33554432;
+    // cleanup
+    public final int CLEANUP_TIMEOUT = 3000;
+    public final int CLEANUP_PROTECTION_THR = 1000;
+    public final int IX_CLEANUP_PROTECTION_THR = 5000;
+    public final int HEAP_USE_THR_DATA = 60;
+    public final int HEAP_USE_THR_INDX = 80;
+    public final int HEAP_USE_THR_TEMP = 40;
+    public final int HEAP_USE_THR_UNDO = 50;
+
 
     private final Properties p;
 

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

@@ -84,7 +84,6 @@ public ValueSet getDcs() {
 
     //returns datacolumn set
     private ValueSet getDcsFromEntity() {
-        ValueSet dcs = null;
         try {
             final Field[] f = this.t == null ? this.entity.getClass().getDeclaredFields() : t.getFields();
             final List<Object> vs = new ArrayList<>();
@@ -98,11 +97,10 @@ private ValueSet getDcsFromEntity() {
                     vs.add(f[i].get(entity));
                 }
             }
-            dcs = new ValueSet(vs.toArray(new Object[]{}));
+            return new ValueSet(vs.toArray(new Object[]{}));
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
-        return dcs;
     }
 
     private ValueSet getDcsFromBytes() {
@@ -111,12 +109,12 @@ private ValueSet getDcsFromBytes() {
         }
 
         try {
-            Field[] cs = null;
+            Field[] cs;
             if (t != null) {
                 cs = t.getFields();
             } else {
-                Field[] f = class_.getDeclaredFields();
-                List<Field> ff = new ArrayList<Field>();
+                final Field[] f = class_.getDeclaredFields();
+                final List<Field> ff = new ArrayList<>();
                 for (int i = 0; i < f.length; i++) {
                     int m = f[i].getModifiers();
                     Transient ta = f[i].getAnnotation(Transient.class);
@@ -489,10 +487,10 @@ public synchronized Object getEntity () {
                     final ResultSetEntity rsa = (ResultSetEntity) ((Table) this.t).getTableClass().getAnnotation(ResultSetEntity.class);
                     final DataChunk dc = rsa == null ? (DataChunk) Instance.getInstance().getChunkByPointer(this.getHeader().getFramePtr(), this.getHeader().getFramePtrRowId().getRowPointer()) : this;
                     if (dc == null) {
-                        // todo during rframe.IndexFrame.init system directory not yet contains replicated FrameData objects
-                        logger.error("null datachunk found");
+                        logger.warn("null data record found during index entity construct");
+                    } else {
+                        dc.setIc(this);
                     }
-                    dc.setIc(this);
                     ((IndexChunk)o).setFramePtrRowId(this.getHeader().getFramePtrRowId());
                     ((IndexChunk)o).setDataChunk(dc);
                 }
@@ -546,7 +544,7 @@ public Object getEntity (Class c, Object[] params) {
         SystemEntity ca = (SystemEntity)c.getAnnotation(SystemEntity.class);
         try {
             if (ca!=null) { //System non-transactional
-                Object o = null;
+                Object o;
                 if (params!=null) {
                     final Class<?>[] cs = new Class<?>[params.length];
                     for (int i=0; i<params.length; i++) {

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

@@ -0,0 +1,74 @@
+/**
+ The MIT License (MIT)
+
+ Copyright (c) 2010-2020 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.List;
+
+/**
+ * @author Yuriy Glotanov
+ * @since 1.0
+ */
+
+public class IndexContainer {
+    private final List<IndexFrame> frameList;
+    private int fptr;
+    private int cptr;
+
+    public IndexContainer(List<IndexFrame> frameList) {
+        this.frameList = frameList;
+        fptr = 0;
+        cptr = 0;
+    }
+
+    public void reset() {
+        fptr = 0;
+        cptr = 0;
+    }
+
+    public Chunk next() {
+        if (frameList.size() == fptr) {
+            return null;
+        }
+        cptr++;
+        if (frameList.get(fptr).data.size() == cptr) {
+            fptr++;
+            cptr = 0;
+        }
+        if (frameList.size() == fptr) {
+            return null;
+        }
+        return frameList.get(fptr).data.get(cptr);
+    }
+
+    public Chunk get() {
+        if (frameList.size() == 0 || frameList.size() == fptr) {
+            return null;
+        }
+        if (frameList.get(fptr).data.size() == 0) {
+            return null;
+        }
+        return frameList.get(fptr).data.get(cptr);
+    }
+}

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

@@ -49,7 +49,7 @@ public class IndexElementList {
 
     public IndexElementList (int type) {
         this.type = type;
-        elementList = new CopyOnWriteArrayList<IndexElement>();
+        elementList = new CopyOnWriteArrayList<>();
     }
 
     private void _add (IndexElement e) {
@@ -79,7 +79,7 @@ public synchronized IndexElementList add (IndexElement e) {
                 this.lc = 0;
             } else {
                 if (e.getKey().compareTo(this.maxValue)>0) {
-
+                    throw new RuntimeException("internal error");
                 } else {
                     res.divided = true;
                     this._add(e);

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

@@ -28,6 +28,8 @@ this software and associated documentation files (the "Software"), to deal in
 import su.interference.persistent.*;
 import su.interference.persistent.Process;
 import su.interference.exception.*;
+
+import java.beans.Transient;
 import java.io.IOException;
 import java.io.File;
 import java.net.URL;
@@ -46,6 +48,9 @@ this software and associated documentation files (the "Software"), to deal in
 
 public class Instance implements Interference {
 
+    public static final String RELEASE = "2020.2";
+    public static final int SYSTEM_VERSION = 20201205;
+
     public static final String DATA_FILE = "datafile";
     public static final String INDX_FILE = "indxfile";
     public static final String JRNL_FILE = "jrnlfile";
@@ -58,8 +63,6 @@ public class Instance implements Interference {
     private static final int MAX_NODE_ID = 32;
     public static final int SESSION_EXPIRE = 7200000; //in ms
 
-    public static final int SYSTEM_VERSION = 20201122;
-
     public static final int SYSTEM_STATE_ONLINE = 1;
     public static final int SYSTEM_STATE_UP = 2;
     public static final int SYSTEM_STATE_FAIL = 3;
@@ -377,12 +380,13 @@ public void startupInstance(Session s) throws Exception, NoSuchMethodException,
             }
             TransportContext.getInstance().start();
             startProcesses(s);
+            checkOpenTransactions(s);
             systemState = Instance.SYSTEM_STATE_UP;
             //checkInMemoryIndexes();
             logger.info("\n----------------------------------------------------------------------\n" +
                           "------------------------ interference started ------------------------\n" +
                           "------------------ (c) head systems, ltd 2010-2020 -------------------\n" +
-                          "--------------------------- release 2020.2 ---------------------------\n" +
+                          "--------------------------- release "+RELEASE+" ---------------------------\n" +
                           "----------------------------------------------------------------------");
         } else {
 
@@ -427,6 +431,18 @@ private void checkInMemoryIndexes() {
         }
     }
 
+    private void checkOpenTransactions(Session s) {
+        for (Transaction t : getTransactions()) {
+            if (t.getCid() == 0 && t.getTransType() != Transaction.TRAN_THR) {
+                logger.info("Rollback incomplete transaction id="+t.getTransId());
+                t.retrieveTframes();
+                if (t.isLocal()) {
+                    t.rollback(s, false);
+                }
+            }
+        }
+    }
+
     @SuppressWarnings("unchecked")
     private void startProcesses(Session s) throws Exception {
         final Table t = getTableByName("su.interference.persistent.Process");
@@ -614,8 +630,7 @@ public synchronized Session getSessionBySid (long sid) {
     }
 
     public synchronized Transaction getTransactionById (long transId) {
-        if (systemState!=SYSTEM_STATE_UP) { return null; }
-        if (transId==0) { return null; }
+        if (transId == 0) { return null; }
         Table t = getTableByName("su.interference.persistent.Transaction");
         DataChunk dc = ((DataChunk)t.getIndexFieldByColumn("transId").getIndex().getObjectByKey(transId));
         if (dc==null) {
@@ -698,6 +713,18 @@ public TransFrame getTransFrameById(long transId, long cframeId, long uframe) {
         return null;
     }
 
+    public List<TransFrame> getTransFramesByTransId(long transId) {
+        List<TransFrame> res = new ArrayList<>();
+        final Table t = getTableByName("su.interference.persistent.TransFrame");
+        for (Map.Entry entry : ((Map<Object, Object>)t.getMapFieldByColumn("frameId").getMap()).entrySet()) {
+            final TransFrame tf = (TransFrame) ((DataChunk) entry.getValue()).getEntity();
+            if (tf.getTransId() == transId) {
+                res.add(tf);
+            }
+        }
+        return res;
+    }
+
     //used in unlock table mechanism
     @Deprecated
     public synchronized ArrayList<TransFrame> getTransFrameByObjectId (int objectId) {

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

@@ -332,7 +332,18 @@ public void createSystemDataFiles (Session s, LLT llt) throws Exception {
 
     public void openDataFiles () throws Exception {
         for (Map.Entry e : ifs.entrySet()) {
-            if (((DataFile)e.getValue()).checkFile()==1) {
+            final int state = ((DataFile)e.getValue()).checkFile();
+            if (state == DataFile.FILE_OK) {
+                ((DataFile)e.getValue()).openFile(Config.getConfig().DISKIO_MODE);
+            }
+            if (state == DataFile.FILE_VERSION_NOT_MATCH) {
+                final int filesv = ((DataFile)e.getValue()).checkSystemVersion();
+                logger.warn("\n----------------------------------------------------------------------\n" +
+                        "--- data files were created by a different version of the software ---\n" +
+                        "--- datafile version: "+filesv+" ---------------------------------------\n" +
+                        "--- software version: "+Instance.SYSTEM_VERSION+" ---------------------------------------\n" +
+                        "--- SERVICE MAY BE UNSTABLE ------------------------------------------\n" +
+                        "----------------------------------------------------------------------");
                 ((DataFile)e.getValue()).openFile(Config.getConfig().DISKIO_MODE);
             }
         }

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

@@ -41,11 +41,8 @@ public class SystemCleanUp implements Runnable, ManagedProcess {
     private volatile boolean f = true;
     CountDownLatch latch;
     private final static Logger logger = LoggerFactory.getLogger(SystemCleanUp.class);
-    private static final int CLEANUP_TIMEOUT = 3000;
     public static final int DATA_RETRIEVED_PRIORITY = 6;
     public static final int INDEX_RETRIEVED_PRIORITY = 9;
-    private static final int CLEANUP_PROTECTION_THR = 1000;
-    private static final int IX_CLEANUP_PROTECTION_THR = 5000;
 
     public void run () {
         Thread.currentThread().setName("interference-cleanup-thread");
@@ -58,7 +55,7 @@ public void run () {
             }
 
             try {
-                Thread.sleep(CLEANUP_TIMEOUT);
+                Thread.sleep(Config.getConfig().CLEANUP_TIMEOUT);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
@@ -74,7 +71,6 @@ public void stop() throws InterruptedException{
     }
 
     private void cleanUpFrames() {
-
         Metrics.get("systemCleanUp").start();
         int i = 0;
         int d = 0;
@@ -87,9 +83,9 @@ private void cleanUpFrames() {
         for (Object entry : Instance.getInstance().getFramesMap().entrySet()) {
             final FrameData f = (FrameData) ((DataChunk) ((Map.Entry) entry).getValue()).getEntity();
             final long frameAmount = f.getDataObject().getFrameAmount();
-            if (f.getDataFile().isData()) {
+            if (f.getDataFile().isData() && cleanupDataEnabled()) {
                 f.decreasePriority();
-                if (f.isSynced() && f.getObjectId() > 999 && f.getPriority() == 0 && frameAmount > CLEANUP_PROTECTION_THR) {
+                if (f.isSynced() && f.getObjectId() > 999 && f.getPriority() == 0 && frameAmount > Config.getConfig().CLEANUP_PROTECTION_THR) {
                     if (f.clearFrame()) {
                         d++;
                     }
@@ -98,9 +94,9 @@ private void cleanUpFrames() {
                     d_++;
                 }
             }
-            if (f.getDataFile().isIndex()) {
+            if (f.getDataFile().isIndex() && cleanupIndxEnabled()) {
                 f.decreasePriority();
-                if (f.isSynced() && f.getObjectId() > 999 && f.getPriority() == 0 && frameAmount > IX_CLEANUP_PROTECTION_THR) {
+                if (f.getPriority() == 0 && frameAmount > Config.getConfig().IX_CLEANUP_PROTECTION_THR) {
                     if (f.clearFrame()) {
                         x++;
                     }
@@ -109,8 +105,8 @@ private void cleanUpFrames() {
                     x_++;
                 }
             }
-            if (f.getDataFile().isTemp()) {
-                if (f.isSynced() && f.getObjectId() > 999 && frameAmount > CLEANUP_PROTECTION_THR) {
+            if (f.getDataFile().isTemp() && cleanupTempEnabled()) {
+                if (frameAmount > Config.getConfig().CLEANUP_PROTECTION_THR) {
                     if (f.clearFrame()) {
                         i++;
                     }
@@ -119,8 +115,8 @@ private void cleanUpFrames() {
                     i_++;
                 }
             }
-            if (f.getDataFile().isUndo()) {
-                if (f.isSynced() && frameAmount > CLEANUP_PROTECTION_THR) {
+            if (f.getDataFile().isUndo() && cleanupUndoEnabled()) {
+                if (f.isSynced() && frameAmount > Config.getConfig().CLEANUP_PROTECTION_THR) {
                     if (f.clearFrame()) {
                         u++;
                     }
@@ -148,4 +144,32 @@ public static void forceCleanUp() {
             }
         }
     }
+
+    private boolean cleanupDataEnabled() {
+        final long maxmem = Runtime.getRuntime().maxMemory();
+        final long alloc = Runtime.getRuntime().totalMemory();
+        final long allocpc = alloc * 100 / maxmem;
+        return allocpc > Config.getConfig().HEAP_USE_THR_DATA;
+    }
+
+    private boolean cleanupIndxEnabled() {
+        final long maxmem = Runtime.getRuntime().maxMemory();
+        final long alloc = Runtime.getRuntime().totalMemory();
+        final long allocpc = alloc * 100 / maxmem;
+        return allocpc > Config.getConfig().HEAP_USE_THR_INDX;
+    }
+
+    private boolean cleanupUndoEnabled() {
+        final long maxmem = Runtime.getRuntime().maxMemory();
+        final long alloc = Runtime.getRuntime().totalMemory();
+        final long allocpc = alloc * 100 / maxmem;
+        return allocpc > Config.getConfig().HEAP_USE_THR_UNDO;
+    }
+
+    private boolean cleanupTempEnabled() {
+        final long maxmem = Runtime.getRuntime().maxMemory();
+        final long alloc = Runtime.getRuntime().totalMemory();
+        final long allocpc = alloc * 100 / maxmem;
+        return allocpc > Config.getConfig().HEAP_USE_THR_TEMP;
+    }
 }