Ts batch improvements

@@ -252,14 +252,17 @@ sql:
       batch_size: "${SQL_ATTRIBUTES_BATCH_SIZE:10000}"
       batch_max_delay: "${SQL_ATTRIBUTES_BATCH_MAX_DELAY_MS:100}"
       stats_print_interval_ms: "${SQL_ATTRIBUTES_BATCH_STATS_PRINT_MS:10000}"
+      batch_threads: "${SQL_ATTRIBUTES_BATCH_THREADS:4}"
     ts:
       batch_size: "${SQL_TS_BATCH_SIZE:10000}"
       batch_max_delay: "${SQL_TS_BATCH_MAX_DELAY_MS:100}"
       stats_print_interval_ms: "${SQL_TS_BATCH_STATS_PRINT_MS:10000}"
+      batch_threads: "${SQL_TS_BATCH_THREADS:4}"
     ts_latest:
       batch_size: "${SQL_TS_LATEST_BATCH_SIZE:10000}"
       batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
       stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
+      batch_threads: "${SQL_TS_LATEST_BATCH_THREADS:4}"
     # Specify whether to remove null characters from strValue of attributes and timeseries before insert
     remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}"
     postgres:
@@ -268,6 +271,7 @@ sql:
     timescale:
       # Specify Interval size for new data chunks storage.
       chunk_time_interval: "${SQL_TIMESCALE_CHUNK_TIME_INTERVAL:604800000}"
+      batch_threads: "${SQL_TIMESCALE_BATCH_THREADS:4}"
     ttl:
       ts:
         enabled: "${SQL_TTL_TS_ENABLED:true}"

@@ -41,16 +41,14 @@ public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
     private final TbSqlBlockingQueueParams params;
 
     private ExecutorService executor;
-    private ScheduledLogExecutorComponent logExecutor;
 
     public TbSqlBlockingQueue(TbSqlBlockingQueueParams params) {
         this.params = params;
     }
 
     @Override
-    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction) {
-        this.logExecutor = logExecutor;
-        executor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("sql-queue-" + params.getLogName().toLowerCase()));
+    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, int index) {
+        executor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("sql-queue-" + index + "-" + params.getLogName().toLowerCase()));
         executor.submit(() -> {
             String logName = params.getLogName();
             int batchSize = params.getBatchSize();
@@ -94,7 +92,7 @@ public class TbSqlBlockingQueue<E> implements TbSqlQueue<E> {
 
         logExecutor.scheduleAtFixedRate(() -> {
             if (queue.size() > 0 || addedCount.get() > 0 || savedCount.get() > 0 || failedCount.get() > 0) {
-                log.info("[{}] queueSize [{}] totalAdded [{}] totalSaved [{}] totalFailed [{}]",
+                log.info("Queue-{} [{}] queueSize [{}] totalAdded [{}] totalSaved [{}] totalFailed [{}]", index,
                         params.getLogName(), queue.size(), addedCount.getAndSet(0), savedCount.getAndSet(0), failedCount.getAndSet(0));
             }
         }, params.getStatsPrintIntervalMs(), params.getStatsPrintIntervalMs(), TimeUnit.MILLISECONDS);

+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.sql;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+@Slf4j
+@Data
+public class TbSqlBlockingQueueWrapper<E> {
+    private final CopyOnWriteArrayList<TbSqlBlockingQueue<E>> queues = new CopyOnWriteArrayList<>();
+    private final TbSqlBlockingQueueParams params;
+    private ScheduledLogExecutorComponent logExecutor;
+    private final Function<E, Integer> hashCodeFunction;
+    private final int maxThreads;
+
+    public void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction) {
+        for (int i = 0; i < maxThreads; i++) {
+            TbSqlBlockingQueue<E> queue = new TbSqlBlockingQueue<>(params);
+            queues.add(queue);
+            queue.init(logExecutor, saveFunction, i);
+        }
+    }
+
+    public ListenableFuture<Void> add(E element) {
+        int hash = hashCodeFunction.apply(element);
+        int queueIndex = (hash & 0x7FFFFFFF) % maxThreads;
+        return queues.get(queueIndex).add(element);
+    }
+
+    public void destroy() {
+        queues.forEach(TbSqlBlockingQueue::destroy);
+    }
+}

@@ -22,7 +22,7 @@ import java.util.function.Consumer;
 
 public interface TbSqlQueue<E> {
 
-    void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction);
+    void init(ScheduledLogExecutorComponent logExecutor, Consumer<List<E>> saveFunction, int queueIndex);
 
     void destroy();
 

@@ -32,8 +32,8 @@ import org.thingsboard.server.dao.model.sql.AttributeKvCompositeKey;
 import org.thingsboard.server.dao.model.sql.AttributeKvEntity;
 import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService;
 import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent;
-import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
+import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
 import org.thingsboard.server.dao.util.SqlDao;
 
 import javax.annotation.PostConstruct;
@@ -41,6 +41,7 @@ import javax.annotation.PreDestroy;
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static org.thingsboard.server.common.data.UUIDConverter.fromTimeUUID;
@@ -68,7 +69,10 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
     @Value("${sql.attributes.stats_print_interval_ms:1000}")
     private long statsPrintIntervalMs;
 
-    private TbSqlBlockingQueue<AttributeKvEntity> queue;
+    @Value("${sql.attributes.batch_threads:4}")
+    private int batchThreads;
+
+    private TbSqlBlockingQueueWrapper<AttributeKvEntity> queue;
 
     @PostConstruct
     private void init() {
@@ -78,7 +82,9 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
                 .maxDelay(maxDelay)
                 .statsPrintIntervalMs(statsPrintIntervalMs)
                 .build();
-        queue = new TbSqlBlockingQueue<>(params);
+
+        Function<AttributeKvEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getId().getEntityId().hashCode() : 0;
+        queue = new TbSqlBlockingQueueWrapper<>(params, hashcodeFunction, batchThreads);
         queue.init(logExecutor, v -> attributeKvInsertRepository.saveOrUpdate(v));
     }
 

@@ -17,6 +17,7 @@ package org.thingsboard.server.dao.sqlts;
 
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import lombok.extern.slf4j.Slf4j;
@@ -31,8 +32,8 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.DaoUtil;
 import org.thingsboard.server.dao.model.sqlts.ts.TsKvEntity;
-import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
+import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
 import org.thingsboard.server.dao.sqlts.insert.InsertTsRepository;
 import org.thingsboard.server.dao.sqlts.ts.TsKvRepository;
 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
@@ -43,6 +44,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Slf4j
@@ -54,7 +56,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
     @Autowired
     protected InsertTsRepository<TsKvEntity> insertRepository;
 
-    protected TbSqlBlockingQueue<TsKvEntity> tsQueue;
+    protected TbSqlBlockingQueueWrapper<TsKvEntity> tsQueue;
 
     @PostConstruct
     protected void init() {
@@ -65,7 +67,9 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
                 .maxDelay(tsMaxDelay)
                 .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .build();
-        tsQueue = new TbSqlBlockingQueue<>(tsParams);
+
+        Function<TsKvEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getEntityId().hashCode() : 0;
+        tsQueue = new TbSqlBlockingQueueWrapper<>(tsParams, hashcodeFunction, tsBatchThreads);
         tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
     }
 

@@ -41,8 +41,8 @@ import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService;
 import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent;
-import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
+import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
 import org.thingsboard.server.dao.sqlts.dictionary.TsKvDictionaryRepository;
 import org.thingsboard.server.dao.sqlts.insert.latest.InsertLatestTsRepository;
 import org.thingsboard.server.dao.sqlts.latest.SearchTsKvLatestRepository;
@@ -52,7 +52,9 @@ import org.thingsboard.server.dao.timeseries.SimpleListenableFuture;
 import javax.annotation.Nullable;
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
@@ -82,7 +84,7 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
     @Autowired
     private TsKvDictionaryRepository dictionaryRepository;
 
-    private TbSqlBlockingQueue<TsKvLatestEntity> tsLatestQueue;
+    private TbSqlBlockingQueueWrapper<TsKvLatestEntity> tsLatestQueue;
 
     @Value("${sql.ts_latest.batch_size:1000}")
     private int tsLatestBatchSize;
@@ -93,6 +95,9 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
     @Value("${sql.ts_latest.stats_print_interval_ms:1000}")
     private long tsLatestStatsPrintIntervalMs;
 
+    @Value("${sql.ts_latest.batch_threads:4}")
+    private int tsLatestBatchThreads;
+
     @Autowired
     protected ScheduledLogExecutorComponent logExecutor;
 
@@ -105,6 +110,12 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
     @Value("${sql.ts.stats_print_interval_ms:1000}")
     protected long tsStatsPrintIntervalMs;
 
+    @Value("${sql.ts.batch_threads:4}")
+    protected int tsBatchThreads;
+
+    @Value("${sql.timescale.batch_threads:4}")
+    protected int timescaleBatchThreads;
+
     @PostConstruct
     protected void init() {
         TbSqlBlockingQueueParams tsLatestParams = TbSqlBlockingQueueParams.builder()
@@ -113,8 +124,23 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
                 .maxDelay(tsLatestMaxDelay)
                 .statsPrintIntervalMs(tsLatestStatsPrintIntervalMs)
                 .build();
-        tsLatestQueue = new TbSqlBlockingQueue<>(tsLatestParams);
-        tsLatestQueue.init(logExecutor, v -> insertLatestTsRepository.saveOrUpdate(v));
+
+        java.util.function.Function<TsKvLatestEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getEntityId().hashCode() : 0;
+        tsLatestQueue = new TbSqlBlockingQueueWrapper<>(tsLatestParams, hashcodeFunction, tsLatestBatchThreads);
+
+        tsLatestQueue.init(logExecutor, v -> {
+            Map<TsKey, List<TsKvLatestEntity>> tsMap =
+                    v.stream().collect(Collectors.groupingBy(ts -> new TsKey(ts.getEntityId(), ts.getStrKey())));
+
+            List<TsKvLatestEntity> latestEntities =
+                    tsMap.keySet()
+                            .stream()
+                            .map(tsMap::get)
+                            .map(list -> list.stream().max(Comparator.comparing(TsKvLatestEntity::getTs)).get())
+                            .collect(Collectors.toList());
+
+            insertLatestTsRepository.saveOrUpdate(latestEntities);
+        });
     }
 
     @PreDestroy

+/**
+ * Copyright © 2016-2020 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.sqlts;
+
+import lombok.Data;
+
+import java.util.UUID;
+
+@Data
+public class TsKey {
+    private final UUID entityId;
+    private final String key;
+}

@@ -33,8 +33,8 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.DaoUtil;
 import org.thingsboard.server.dao.model.sqlts.timescale.ts.TimescaleTsKvEntity;
-import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
+import org.thingsboard.server.dao.sql.TbSqlBlockingQueueWrapper;
 import org.thingsboard.server.dao.sqlts.AbstractSqlTimeseriesDao;
 import org.thingsboard.server.dao.sqlts.insert.InsertTsRepository;
 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
@@ -48,6 +48,7 @@ import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
 
 @Component
 @Slf4j
@@ -63,7 +64,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
     @Autowired
     protected InsertTsRepository<TimescaleTsKvEntity> insertRepository;
 
-    protected TbSqlBlockingQueue<TimescaleTsKvEntity> tsQueue;
+    protected TbSqlBlockingQueueWrapper<TimescaleTsKvEntity> tsQueue;
 
     @PostConstruct
     protected void init() {
@@ -74,7 +75,10 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 .maxDelay(tsMaxDelay)
                 .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .build();
-        tsQueue = new TbSqlBlockingQueue<>(tsParams);
+
+        Function<TimescaleTsKvEntity, Integer> hashcodeFunction = entity -> entity != null ? entity.getEntityId().hashCode() : 0;
+        tsQueue = new TbSqlBlockingQueueWrapper<>(tsParams, hashcodeFunction, timescaleBatchThreads);
+
         tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
     }