improve overall performance & stability

Author: yuriy-glotanov

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

@@ -1,7 +1,7 @@
 /**
  The MIT License (MIT)
 
- Copyright (c) 2010-2019 head systems, ltd
+ 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
@@ -31,7 +31,6 @@ this software and associated documentation files (the "Software"), to deal in
 import java.util.Properties;
 import java.io.FileInputStream;
 import java.io.File;
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.StringTokenizer;
 
@@ -104,8 +103,10 @@ public class Config {
     public final boolean SYNC_LOCK_ENABLE;
     public final int SYNC_PERIOD;
     public final int RETRIEVE_QUEUE_SIZE;
+    public final int RETRIEVE_THREADS_AMOUNT = 8;
     public final String CODEPAGE;
     public final String DATEFORMAT;
+    public final int TEST_DISTRIBUTE_MODE = 1;
 
     private final Properties p;
 

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

@@ -63,7 +63,7 @@ public class DataChunk implements Chunk {
     private volatile byte[] chunk;
     private Comparable id;
     private byte[] serializedId;
-    private Object entity;
+    private volatile Object entity;
     private Object undoentity;
     private Map<Integer, DataChunk> ics = new HashMap<>();
     private UndoChunk uc;
@@ -470,7 +470,7 @@ public Object getExistingEntity() {
         return this.entity;
     }
 
-    public Object getEntity () {
+    public synchronized Object getEntity () {
         if (entity==null) {
             try {
                 final ValueSet dcs = getDcsFromBytes();
@@ -580,10 +580,6 @@ public Object getEntity (Class c, Object[] params) {
         return entity;
     }
 
-    public void setFrameData(FrameData b) {
-        entity = b;
-    }
-
     @Deprecated
     public Object getUndoEntity () {
         if (undoentity==null) {
@@ -715,10 +711,6 @@ public void setEntity(Object entity) {
         }
     }
 
-    protected void setEntity(Object entity, boolean z) {
-        this.entity = entity;
-    }
-
     //method for transactional objects ONLY
     public synchronized DataChunk lock (Session s, LLT llt) throws Exception {
         final Object o = this.getEntity();

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

@@ -507,7 +507,7 @@ public synchronized void removeChunk (int ptr, LLT llt, boolean ignore) {
     }
 
     //returns all actual records
-    public synchronized ArrayList<Chunk> getFrameChunks (Session s) {
+    public ArrayList<Chunk> getFrameChunks (Session s) {
         final ArrayList<Chunk> res = new ArrayList<>();
 
         if (dataObject==null) {

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

@@ -1,7 +1,7 @@
 /**
  The MIT License (MIT)
 
- Copyright (c) 2010-2019 head systems, ltd
+ 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
@@ -32,7 +32,7 @@ this software and associated documentation files (the "Software"), to deal in
 import javax.persistence.*;
 import java.io.Serializable;
 import java.lang.reflect.Modifier;
-import java.net.MalformedURLException;
+import java.util.List;
 
 /**
  * @author Yuriy Glotanov
@@ -87,7 +87,7 @@ public class Cursor implements Serializable {
     @Transient
     private SQLSelect sqlStmt;
     @Transient
-    private String resultTargetName;
+    private List<String> resultTargetNames;
     @Transient
     private Session session;
     @Transient
@@ -101,22 +101,21 @@ public Cursor() {
         this.state = STATE_IDLE;
     }
 
-    public Cursor(String sql, String resultTargetName, int type) {
+    public Cursor(String sql, List<String> resultTargetNames, int type) {
         this.state = STATE_IDLE;
         this.sql = sql;
-        this.resultTargetName = resultTargetName;
+        this.resultTargetNames = resultTargetNames;
         this.type = type;
     }
 
     public Cursor(String sql, int type) {
         this.state = STATE_IDLE;
-        this.cursorId = cursorId;
         this.sql = sql;
         this.type = type;
     }
 
     //constructor for low-level storage function (initial first-time load table descriptions from datafile)
-    public Cursor (DataChunk chunk) throws IllegalAccessException, ClassNotFoundException, InternalException, MalformedURLException {
+    public Cursor (DataChunk chunk) throws IllegalAccessException, InternalException {
         final Object[] dcs = chunk.getDcs().getValueSet();
         final Class c = this.getClass();
         final java.lang.reflect.Field[] f = c.getDeclaredFields();
@@ -174,8 +173,8 @@ public void setSqlStmt(SQLSelect sqlStmt) {
         this.sqlStmt = sqlStmt;
     }
 
-    public String getResultTargetName() {
-        return resultTargetName;
+    public List<String> getResultTargetNames() {
+        return resultTargetNames;
     }
 
     public Session getSession() {

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

@@ -217,7 +217,6 @@ public synchronized FrameData createNewFrame(FrameData frame, int frameType, lon
             //replace chunk after usedSpace
             dc = new DataChunk(bd, s);
             db.insertChunk(dc, s, true, llt);
-            dc.setFrameData(bd);
             t.addIndexValue(dc);
         } else {
             //syncframe event should not persist new frame

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

@@ -190,17 +190,19 @@ public IndexFrame getIndexFrame() throws Exception {
         return (IndexFrame) frame;
     }
 
-    public synchronized ArrayList<Chunk> getFrameChunks(Session s) throws Exception {
+    public ArrayList<Chunk> getFrameChunks(Session s) throws Exception {
         if (getDataObject().isIndex()) {
             final IndexFrame f = getIndexFrame();
-            f.sort();
+            synchronized (this) {
+                f.sort();
+            }
             return f.getFrameChunks(s);
         } else {
             return getDataFrame().getFrameChunks(s);
         }
     }
 
-    public synchronized ArrayList<Object> getFrameEntities(Session s) throws Exception {
+    public ArrayList<Object> getFrameEntities(Session s) throws Exception {
         final ArrayList<Object> res = new ArrayList<>();
         for (Chunk c : getFrameChunks(s)) {
             res.add(((DataChunk)c).getEntity());

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

@@ -1,7 +1,7 @@
 /**
  The MIT License (MIT)
 
- Copyright (c) 2010-2019 head systems, ltd
+ 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
@@ -24,6 +24,8 @@ this software and associated documentation files (the "Software"), to deal in
 
 package su.interference.sql;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import su.interference.core.Config;
 import su.interference.metrics.Metrics;
 
@@ -38,6 +40,7 @@ this software and associated documentation files (the "Software"), to deal in
  */
 
 public class FrameApiJoin implements Serializable, Callable<FrameApiJoin> {
+    private final static Logger logger = LoggerFactory.getLogger(FrameApiJoin.class);
     private final int nodeId;
     private final transient FrameApi bd1;
     private final transient FrameApi bd2;
@@ -61,12 +64,20 @@ public FrameApiJoin(int nodeId, SQLCursor cur, FrameApi bd1, FrameApi bd2) {
         }
     }
 
-    public FrameApiJoin call() throws Exception {
+    public FrameApiJoin call() throws InterruptedException {
+        final Thread thread = Thread.currentThread();
         if (nodeId == Config.getConfig().LOCAL_NODE_ID) {
+            thread.setName("interference-sql-join-task-" + thread.getId());
             Metrics.get("localTask").start();
-            result = frameJoinTask.call();
+            try {
+                result = frameJoinTask.call();
+            } catch (Exception e) {
+                failed = true;
+                logger.error("join task", e);
+            }
             Metrics.get("localTask").stop();
         } else {
+            thread.setName("interference-sql-remote-task-" + thread.getId());
             Metrics.get("remoteTask").start();
             latch.await();
             Metrics.get("remoteTask").stop();

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

@@ -1,7 +1,7 @@
 /**
  The MIT License (MIT)
 
- Copyright (c) 2010-2019 head systems, ltd
+ 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
@@ -26,8 +26,6 @@ this software and associated documentation files (the "Software"), to deal in
 
 import su.interference.exception.InternalException;
 
-import java.io.IOException;
-import java.lang.reflect.InvocationTargetException;
 import java.net.MalformedURLException;
 import java.util.List;
 
@@ -50,5 +48,6 @@ public interface FrameIterator {
     List<Integer> getObjectIds();
     boolean isLeftfs();
     void setLeftfs(boolean leftfs);
+    boolean noDistribute();
 
 }