session 2 step hazelcast batch insert

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/RequirementsAndObligationCreation.java

@@ -2,7 +2,9 @@ package ru.spcex.clearing.session.stage.impl;
 
 import java.time.LocalDate;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
@@ -47,6 +49,7 @@ import ru.spcex.platform.enumeration.RegistryTradingParams;
 import ru.spcex.platform.enumeration.RegistryUnit;
 import ru.spcex.platform.enumeration.Side;
 import ru.spcex.platform.imdg.api.Imdg;
+import ru.spcex.platform.imdg.api.ImdgId;
 import ru.spcex.platform.imdg.api.ImdgProvider;
 import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
 import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
@@ -65,7 +68,8 @@ import ru.spcex.platform.utils.collection.Pair;
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
 public class RequirementsAndObligationCreation implements ISessionStage {
     private final Logger log = LoggerFactory.getLogger(getClass());
-    private final Imdg<Registry> registryImdg;
+    protected final Imdg<Registry> registryImdg;
+    private final ImdgId idGen;
     private final ImdgProvider imdgProvider;
     protected final RequirementsAndObligationCreationCash cash;
 
@@ -73,6 +77,7 @@ public class RequirementsAndObligationCreation implements ISessionStage {
     @Autowired
     public RequirementsAndObligationCreation(ImdgProvider imdgProvider) {
         this.registryImdg = imdgProvider.getImdg(IMDGDistributedNames.Map_Registry, Registry.class);
+        this.idGen = imdgProvider.getImdgIdGenerator();
         this.imdgProvider = imdgProvider;
         this.cash = new RequirementsAndObligationCreationCash();
     }
@@ -85,10 +90,10 @@ public class RequirementsAndObligationCreation implements ISessionStage {
                 try {
                     RequirementsAndObligationCreationPayload data
                         = (RequirementsAndObligationCreationPayload) task.getData();
-                    return createRegisters(data.getExecutions(), data.getSessionId());
+                    return createRegisters(data.getExecutions(), data.getSessionId(), null);
                 } catch (ClassCastException e) {
                     List<ExecutionCommon> data = (List<ExecutionCommon>) task.getData();
-                    return createRegisters(data, null);
+                    return createRegisters(data, null, null);
                 }
             } finally {
                 doRgsSearch = null;
@@ -120,9 +125,29 @@ public class RequirementsAndObligationCreation implements ISessionStage {
         return this.doRgsSearch;
     }
 
+    protected static int calcMapCapacity(int num) {
+        double divided = num / 0.75;
+        return (int) Math.ceil(divided);
+    }
 
-    protected StageResult<?> createRegisters(List<ExecutionCommon> data, Long sessionId) {
+    protected Map<Long, Registry> storageMap(List<ExecutionCommon> data) {
+        ExecutionType eType = data.stream()
+            .findFirst()
+            .map(ExecutionCommon::type)
+            .orElse(ExecutionType.ExecutionDeposit);
+        int initialCapacity;
+        if (eType.equals(ExecutionType.ExecutionDeposit)) {
+            initialCapacity = calcMapCapacity(data.size() * 4);
+        } else {
+            initialCapacity = calcMapCapacity(data.size() * 2);
+        }
+        log.trace("executions size {}; storage map initial capacity {}", data.size(), initialCapacity);
+        return new HashMap<>(initialCapacity, 1f);
+    }
+
+    protected StageResult<?> createRegisters(List<ExecutionCommon> data, Long sessionId, Map<Long, Registry> rgsStorage) {
         data.sort(tradeTimeComparator.thenComparing(execIdComparator));
+        Map<Long, Registry> newRgss = rgsStorage != null ? rgsStorage : storageMap(data);
         //см. описание к #matchExecutions
         for (int i = 0; i < data.size(); ) {
             Pair<ExecutionCommon, ExecutionCommon> matched = matchExecutions(data, i);
@@ -171,7 +196,8 @@ public class RequirementsAndObligationCreation implements ISessionStage {
                                         .registryDesignation(regDsgn)
                                         .returnDeposit(returnedDep)
                                         .build();
-                                registryImdg.insert(newRegister);
+                                newRegister.setId(idGen.nextId());
+                                newRgss.put(newRegister.getId(), newRegister);
                                 log.debug("executions id {} and {}: created rgs.id={}", partyExec.getId(), counterExec.getId(), newRegister.getId());
                             });
             findAndUpdateOrCreate.accept(partyExec, RegistryDesignation.O, false);
@@ -189,6 +215,10 @@ public class RequirementsAndObligationCreation implements ISessionStage {
                 findAndUpdateOrCreate.accept(counterExec, RegistryDesignation.O, true);
             }
         }
+        if (rgsStorage == null) {
+            log.info("batch insert {} registries into map", newRgss.size());
+            registryImdg.putAll(newRgss);
+        }
         return new StageResult<>(null, true);
     }
 

clearing-parent/clearing-service/src/main/java/ru/spcex/clearing/session/stage/impl/RequirementsAndObligationCreationCompound.java

@@ -2,7 +2,9 @@ package ru.spcex.clearing.session.stage.impl;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Stream;
 import org.slf4j.Logger;
@@ -12,6 +14,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Service;
 import ru.clearing.classes.statics.data.execution.ExecutionCommon;
+import ru.clearing.classes.statics.data.registry.Registry;
 import ru.spcex.clearing.session.stage.StageResult;
 import ru.spcex.clearing.session.stage.Task;
 import ru.spcex.clearing.session.stage.TaskType;
@@ -48,12 +51,20 @@ public class RequirementsAndObligationCreationCompound extends RequirementsAndOb
         throw new IllegalStateException("Unknown task type: " + task.getTaskType());
     }
 
+    protected Map<Long, Registry> storageMap(List<ExecutionCommon> executionFond, List<ExecutionCommon> executionCurrency, List<ExecutionCommon> executionDeposit) {
+        int depSize = executionDeposit.size() * 4;
+        int othersSize = Math.max(executionFond.size() * 2, executionCurrency.size() * 2);
+        int capacity = calcMapCapacity(Math.max(depSize, othersSize));
+        return new HashMap<>(capacity, 1f);
+    }
+
 
     private StageResult<?> mergeReqsAndClaims(RequirementsAndObligationCreationCompoundPayload data) {
         List<ExecutionCommon> tmpList = new ArrayList<>(Arrays.asList(new ExecutionCommon[2]));
         List<ExecutionCommon> executionFond = data.getExecutionsTrdt();
         List<ExecutionCommon> executionCurrency = data.getExecutionsCurrency();
         List<ExecutionCommon> executionDeposit = data.getExecutionsFinal();
+        Map<Long, Registry> newRgss = storageMap(executionFond, executionCurrency, executionDeposit);
         Stream.of(executionFond, executionCurrency, executionDeposit).forEach(c -> c.sort(tradeTimeComparator.thenComparing(execIdComparator)));
 
         int efs = executionFond.size();
@@ -72,9 +83,10 @@ public class RequirementsAndObligationCreationCompound extends RequirementsAndOb
             if (pair.getFirst().type().equals(ExecutionType.ExecutionDeposit)) ied[0] += 2;
             if (pair.getFirst().type().equals(ExecutionType.ExecutionCurrency)) iec[0] += 2;
             pair.map(f -> tmpList.set(0, f), s -> tmpList.set(1, s));
-            createRegisters(tmpList, data.getSessionId());
+            createRegisters(tmpList, data.getSessionId(), newRgss);
         }
-
+        log.info("batch insert {} registries into map", newRgss.size());
+        registryImdg.putAll(newRgss);
         return new StageResult<>(null, true);
     }
 

platform-parent/platform-imdg-api-hazelcast-impl/src/main/java/ru/spcex/platform/imdg/iml/hazelcast/adapter/ImdgHazelcast.java

@@ -1,5 +1,11 @@
 package ru.spcex.platform.imdg.iml.hazelcast.adapter;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import com.hazelcast.aggregation.Aggregators;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.ILock;
@@ -19,8 +25,6 @@ import ru.spcex.platform.imdg.iml.hazelcast.adapter.predicate.ImdgPredicateBuild
 import ru.spcex.platform.imdg.iml.hazelcast.adapter.predicate.ImdgPredicateHazelcast;
 import ru.spcex.platform.utils.log.ExceptionUtils;
 
-import java.util.*;
-
 public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private IdGenerator idGenerator;
@@ -98,6 +102,11 @@ public class ImdgHazelcast<T extends SpcexObjectBase> implements Imdg<T> {
         }
     }
 
+    @Override
+    public void putAll(Map<Long, T> m) {
+        map.putAll(m);
+    }
+
     @Override
     public void update(T paramT) {
         map.put(paramT.getId(), paramT);

platform-parent/platform-imdg-api/src/main/java/ru/spcex/platform/imdg/api/Imdg.java

@@ -1,11 +1,10 @@
 package ru.spcex.platform.imdg.api;
 
-import ru.spcex.platform.classes.base.SpcexObjectBase;
-import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
-import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
-
 import java.util.Collection;
 import java.util.Map;
+import ru.spcex.platform.classes.base.SpcexObjectBase;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicate;
+import ru.spcex.platform.imdg.api.predicate.ImdgPredicateBuilder;
 
 public interface Imdg<T extends SpcexObjectBase> {
     default String getMapName() {
@@ -20,6 +19,8 @@ public interface Imdg<T extends SpcexObjectBase> {
         throw new UnsupportedOperationException("not implemented lockAndPerform");
     }
 
+    default void putAll(Map<Long, T> batch) {throw new UnsupportedOperationException("not implemented update");}
+
     default void update(T paramT) {
         throw new UnsupportedOperationException("not implemented update");
     }