From bff310b6913ab750fb506884bb29515098177bc0 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Thu, 8 Aug 2024 17:31:05 +0200 Subject: [PATCH] DefaultTbLocalSubscriptionService WS single lock refactored by tenantId; TbEntityLocalSubsInfo performance optimizations for remove subscription --- .../DefaultTbLocalSubscriptionService.java | 36 ++++++++++++++----- .../subscription/TbAlarmsSubscription.java | 8 ++--- .../subscription/TbAttributeSubscription.java | 8 ++--- .../subscription/TbEntityLocalSubsInfo.java | 32 ++++++++++++++--- .../service/subscription/TbSubscription.java | 24 +++---------- .../subscription/TbSubscriptionUtils.java | 5 +-- .../subscription/TbSubscriptionsInfo.java | 9 +++-- .../TbTimeSeriesSubscription.java | 8 ++--- .../sub/AbstractNotificationSubscription.java | 5 +++ .../sub/NotificationsCountSubscription.java | 6 +++- .../sub/NotificationsSubscription.java | 6 +++- 11 files changed, 87 insertions(+), 60 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index 6e99731305..0d29d374a3 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -22,6 +22,7 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Lazy; import org.springframework.context.event.EventListener; import org.springframework.stereotype.Service; +import org.springframework.util.ConcurrentReferenceHashMap; import org.thingsboard.common.util.DeduplicationUtil; import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.ThingsBoardExecutors; @@ -120,12 +121,12 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer private String serviceId; private ExecutorService subscriptionUpdateExecutor; - private final Lock subsLock = new ReentrantLock(); + private final ConcurrentReferenceHashMap locks = new ConcurrentReferenceHashMap<>(16, ConcurrentReferenceHashMap.ReferenceType.SOFT); @PostConstruct public void initExecutor() { subscriptionUpdateExecutor = ThingsBoardExecutors.newWorkStealingPool(20, getClass()); - tsCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("ts-sub-callback")); + tsCallBackExecutor = Executors.newFixedThreadPool(8, ThingsBoardThreadFactory.forName("ts-sub-callback")); //since we are using locks by TenantId serviceId = serviceInfoProvider.getServiceId(); staleSessionCleanupExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("stale-session-cleanup")); staleSessionCleanupExecutor.scheduleWithFixedDelay(this::cleanupStaleSessions, 60, 60, TimeUnit.SECONDS); @@ -176,6 +177,10 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer }); } + Lock getSubsLock(TenantId tenantId) { + return locks.computeIfAbsent(tenantId, x -> new ReentrantLock()); + } + @Override public void addSubscription(TbSubscription subscription, WebSocketSessionRef sessionRef) { TenantId tenantId = subscription.getTenantId(); @@ -191,6 +196,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer log.debug("[{}][{}] Register subscription: {}", tenantId, entityId, subscription); SubscriptionModificationResult result; + final Lock subsLock = getSubsLock(tenantId); subsLock.lock(); try { Map> sessionSubscriptions = subscriptionsBySessionId.computeIfAbsent(subscription.getSessionId(), k -> new ConcurrentHashMap<>()); @@ -219,14 +225,18 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer log.debug("[{}][{}] Processing sub event callback: {}.", entityId, seqNumber, entityUpdatesInfo); entityUpdates.put(entityId, entityUpdatesInfo); Set> pendingSubs = null; - subsLock.lock(); + Lock subsLock = null; try { TbEntityLocalSubsInfo entitySubs = subscriptionsByEntityId.get(entityId); if (entitySubs != null) { + subsLock = getSubsLock(entitySubs.getTenantId()); + subsLock.lock(); pendingSubs = entitySubs.clearPendingSubscriptions(seqNumber); } } finally { - subsLock.unlock(); + if (subsLock != null) { + subsLock.unlock(); + } } if (pendingSubs != null) { pendingSubs.forEach(this::checkMissedUpdates); @@ -238,12 +248,14 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer public void cancelSubscription(String sessionId, int subscriptionId) { log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId); SubscriptionModificationResult result = null; - subsLock.lock(); + Lock subsLock = null; try { Map> sessionSubscriptions = subscriptionsBySessionId.get(sessionId); if (sessionSubscriptions != null) { TbSubscription subscription = sessionSubscriptions.remove(subscriptionId); if (subscription != null) { + subsLock = getSubsLock(subscription.getTenantId()); + subsLock.lock(); if (sessionSubscriptions.isEmpty()) { subscriptionsBySessionId.remove(sessionId); } @@ -255,7 +267,9 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer log.debug("[{}] No session subscriptions found!", sessionId); } } finally { - subsLock.unlock(); + if (subsLock != null) { + subsLock.unlock(); + } } if (result != null && result.hasEvent()) { pushSubscriptionEvent(result); @@ -266,18 +280,24 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer public void cancelAllSessionSubscriptions(String sessionId) { log.debug("[{}] Going to remove session subscriptions.", sessionId); List results = new ArrayList<>(); - subsLock.lock(); + Lock subsLock = null; try { Map> sessionSubscriptions = subscriptionsBySessionId.remove(sessionId); if (sessionSubscriptions != null) { for (TbSubscription subscription : sessionSubscriptions.values()) { + if (subsLock == null) { + subsLock = getSubsLock(subscription.getTenantId()); + subsLock.lock(); + } results.add(modifySubscription(subscription.getTenantId(), subscription.getEntityId(), subscription, false)); } } else { log.debug("[{}] No session subscriptions found!", sessionId); } } finally { - subsLock.unlock(); + if (subsLock != null) { + subsLock.unlock(); + } } results.stream().filter(SubscriptionModificationResult::hasEvent).forEach(this::pushSubscriptionEvent); } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java index a1d0aec29f..d59f8d74c3 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbAlarmsSubscription.java @@ -36,12 +36,8 @@ public class TbAlarmsSubscription extends TbSubscription> subs = ConcurrentHashMap.newKeySet(); private volatile TbSubscriptionsInfo state = new TbSubscriptionsInfo(); @@ -88,6 +84,7 @@ public class TbEntityLocalSubsInfo { if (!newState.attrAllKeys) { if (attrSub.isAllKeys()) { newState.attrAllKeys = true; + newState.attrKeys = null; stateChanged = true; } else { if (newState.attrKeys == null) { @@ -104,6 +101,7 @@ public class TbEntityLocalSubsInfo { if (!newState.tsAllKeys) { if (tsSub.isAllKeys()) { newState.tsAllKeys = true; + newState.tsKeys = null; stateChanged = true; } else { if (newState.tsKeys == null) { @@ -137,8 +135,30 @@ public class TbEntityLocalSubsInfo { return toEvent(ComponentLifecycleEvent.DELETED); } TbSubscriptionsInfo oldState = state.copy(); - TbSubscriptionsInfo newState = new TbSubscriptionsInfo(); + + //copy unchanged state only + TbSubscriptionsInfo newState = state.copy(); + switch (sub.getType()) { + case NOTIFICATIONS: + case NOTIFICATIONS_COUNT: + newState.notifications = false; + break; + case ALARMS: + newState.alarms = false; + break; + case ATTRIBUTES: + newState.attrAllKeys = false; + newState.attrKeys = null; + break; + case TIMESERIES: + newState.tsAllKeys = false; + newState.tsKeys = null; + } + for (TbSubscription subscription : subs) { + if (subscription.getType() != sub.getType()) { + continue; // skip unchanged types + } switch (subscription.getType()) { case NOTIFICATIONS: case NOTIFICATIONS_COUNT: @@ -155,6 +175,7 @@ public class TbEntityLocalSubsInfo { var attrSub = (TbAttributeSubscription) subscription; if (!newState.attrAllKeys && attrSub.isAllKeys()) { newState.attrAllKeys = true; + newState.attrKeys = null; continue; } if (newState.attrKeys == null) { @@ -167,6 +188,7 @@ public class TbEntityLocalSubsInfo { var tsSub = (TbTimeSeriesSubscription) subscription; if (!newState.tsAllKeys && tsSub.isAllKeys()) { newState.tsAllKeys = true; + newState.tsKeys = null; continue; } if (newState.tsKeys == null) { diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java index d1822a6ffe..68c042ea51 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscription.java @@ -15,41 +15,25 @@ */ package org.thingsboard.server.service.subscription; -import lombok.AllArgsConstructor; import lombok.Data; +import lombok.EqualsAndHashCode; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; -import java.util.Objects; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; @Data -@AllArgsConstructor +@EqualsAndHashCode(cacheStrategy = EqualsAndHashCode.CacheStrategy.LAZY, doNotUseGetters = true) public abstract class TbSubscription { + @EqualsAndHashCode.Exclude private final String serviceId; private final String sessionId; private final int subscriptionId; private final TenantId tenantId; private final EntityId entityId; private final TbSubscriptionType type; + @EqualsAndHashCode.Exclude private final BiConsumer, T> updateProcessor; - @Override - public boolean equals(Object o) { - if (this == o) return true; - if (o == null || getClass() != o.getClass()) return false; - TbSubscription that = (TbSubscription) o; - return subscriptionId == that.subscriptionId && - sessionId.equals(that.sessionId) && - tenantId.equals(that.tenantId) && - entityId.equals(that.entityId) && - type == that.type; - } - - @Override - public int hashCode() { - return Objects.hash(sessionId, subscriptionId, tenantId, entityId, type); - } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java index 2d8440fa20..d691b70c55 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java @@ -107,8 +107,9 @@ public class TbSubscriptionUtils { .type(event); if (!ComponentLifecycleEvent.DELETED.equals(event)) { builder.info(new TbSubscriptionsInfo(proto.getNotifications(), proto.getAlarms(), - proto.getTsAllKeys(), proto.getTsKeysCount() > 0 ? new HashSet<>(proto.getTsKeysList()) : null, - proto.getAttrAllKeys(), proto.getAttrKeysCount() > 0 ? new HashSet<>(proto.getAttrKeysList()) : null, + proto.getTsAllKeys(), proto.getAttrAllKeys(), + proto.getTsKeysCount() > 0 ? new HashSet<>(proto.getTsKeysList()) : null, + proto.getAttrKeysCount() > 0 ? new HashSet<>(proto.getAttrKeysList()) : null, proto.getSeqNumber())); } return builder.build(); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionsInfo.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionsInfo.java index aa8ea70eab..7430d2e386 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionsInfo.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionsInfo.java @@ -20,25 +20,24 @@ import lombok.EqualsAndHashCode; import lombok.RequiredArgsConstructor; import lombok.ToString; -import java.util.HashSet; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; /** * Information about the local websocket subscriptions. */ @RequiredArgsConstructor @AllArgsConstructor -@EqualsAndHashCode(exclude = {"seqNumber"}) +@EqualsAndHashCode @ToString public class TbSubscriptionsInfo { protected boolean notifications; protected boolean alarms; protected boolean tsAllKeys; + protected boolean attrAllKeys; // primitives first for equals performance protected Set tsKeys; - protected boolean attrAllKeys; protected Set attrKeys; + @EqualsAndHashCode.Exclude protected int seqNumber; public boolean isEmpty() { @@ -50,7 +49,7 @@ public class TbSubscriptionsInfo { } protected TbSubscriptionsInfo copy(int seqNumber) { - return new TbSubscriptionsInfo(notifications, alarms, tsAllKeys, tsKeys, attrAllKeys, attrKeys, seqNumber); + return new TbSubscriptionsInfo(notifications, alarms, tsAllKeys, attrAllKeys, tsKeys, attrKeys, seqNumber); } } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeSeriesSubscription.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeSeriesSubscription.java index 10a24106ad..0fc9a80122 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeSeriesSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbTimeSeriesSubscription.java @@ -53,12 +53,8 @@ public class TbTimeSeriesSubscription extends TbSubscription extends TbSubscription super(serviceId, sessionId, subscriptionId, tenantId, entityId, type, updateProcessor); } + @Override + protected boolean canEqual(final Object other) { + return other instanceof AbstractNotificationSubscription; + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java index 1e812c3b31..491acedb1e 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsCountSubscription.java @@ -23,7 +23,6 @@ import org.thingsboard.server.service.subscription.TbSubscription; import org.thingsboard.server.service.subscription.TbSubscriptionType; import org.thingsboard.server.service.ws.notification.cmd.UnreadNotificationsCountUpdate; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; @Getter @@ -35,6 +34,11 @@ public class NotificationsCountSubscription extends AbstractNotificationSubscrip super(serviceId, sessionId, subscriptionId, tenantId, entityId, TbSubscriptionType.NOTIFICATIONS_COUNT, updateProcessor); } + @Override + protected boolean canEqual(final Object other) { + return other instanceof NotificationsCountSubscription; + } + public UnreadNotificationsCountUpdate createUpdate() { return UnreadNotificationsCountUpdate.builder() .cmdId(getSubscriptionId()) diff --git a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java index 97dc15f6b0..867f48b72e 100644 --- a/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java +++ b/application/src/main/java/org/thingsboard/server/service/ws/notification/sub/NotificationsSubscription.java @@ -30,7 +30,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; import java.util.stream.Collectors; @@ -48,6 +47,11 @@ public class NotificationsSubscription extends AbstractNotificationSubscription< this.limit = limit; } + @Override + protected boolean canEqual(final Object other) { + return other instanceof NotificationsSubscription; + } + public UnreadNotificationsUpdate createFullUpdate() { return UnreadNotificationsUpdate.builder() .cmdId(getSubscriptionId())