Merge pull request #1451 from OSGP/feature/throttling-service

Feature/throttling service

Author: jboon

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/PermitsByThrottlingConfig.java

@@ -12,6 +12,7 @@
 import java.util.concurrent.ConcurrentMap;
 import org.opensmartgridplatform.throttling.entities.ThrottlingConfig;
 import org.opensmartgridplatform.throttling.model.NetworkSegment;
+import org.opensmartgridplatform.throttling.model.PermitRequest;
 import org.opensmartgridplatform.throttling.model.ThrottlingSettings;
 import org.opensmartgridplatform.throttling.repositories.ThrottlingConfigRepository;
 import org.opensmartgridplatform.throttling.services.PermitService;
@@ -34,19 +35,21 @@ public class PermitsByThrottlingConfig {
   private final RateLimitService rateLimitService;
   private final boolean highPrioPoolEnabled;
   private final int maxWaitForHighPrioInMs;
+  private final int pauseWaitForHighPrioInMs;
 
   public PermitsByThrottlingConfig(
       final ThrottlingConfigRepository throttlingConfigRepository,
       final PermitService permitService,
       final RateLimitService rateLimitService,
       @Value("${wait.for.high.prio.enabled:true}") final boolean highPrioPoolEnabled,
-      @Value("${wait.for.high.prio.max.in.ms:10000}") final int maxWaitForHighPrioInMs) {
-
+      @Value("${wait.for.high.prio.max.in.ms:10000}") final int maxWaitForHighPrioInMs,
+      @Value("${wait.for.high.prio.pause.in.ms:200}") final int pauseWaitForHighPrioInMs) {
     this.throttlingConfigRepository = throttlingConfigRepository;
     this.permitService = permitService;
     this.rateLimitService = rateLimitService;
     this.highPrioPoolEnabled = highPrioPoolEnabled;
     this.maxWaitForHighPrioInMs = maxWaitForHighPrioInMs;
+    this.pauseWaitForHighPrioInMs = pauseWaitForHighPrioInMs;
   }
 
   /** Clears all cached permit counts and initializes the cached information from the database. */
@@ -67,7 +70,8 @@ public void initialize() {
                     this.permitService,
                     this.rateLimitService,
                     this.highPrioPoolEnabled,
-                    this.maxWaitForHighPrioInMs)));
+                    this.maxWaitForHighPrioInMs,
+                    this.pauseWaitForHighPrioInMs)));
 
     /* Remove config not in database */
     final List<Short> throttlingConfigIdsToBeRemoved =
@@ -96,15 +100,16 @@ public boolean requestPermit(
             networkSegment.throttlingConfigId(), this::createAndInitialize);
 
     return permitsPerNetworkSegment.requestPermit(
-        networkSegment, clientId, requestId, priority, throttlingSettings);
+        networkSegment, new PermitRequest(clientId, requestId), priority, throttlingSettings);
   }
 
   private PermitsPerNetworkSegment createAndInitialize(final short throttlingConfigId) {
     return new PermitsPerNetworkSegment(
         this.permitService,
         this.rateLimitService,
         this.highPrioPoolEnabled,
-        this.maxWaitForHighPrioInMs);
+        this.maxWaitForHighPrioInMs,
+        this.pauseWaitForHighPrioInMs);
   }
 
   public void newThrottlingConfigCreated(final short throttlingConfigId) {
@@ -119,7 +124,8 @@ public void newThrottlingConfigCreated(final short throttlingConfigId) {
             this.permitService,
             this.rateLimitService,
             this.highPrioPoolEnabled,
-            this.maxWaitForHighPrioInMs));
+            this.maxWaitForHighPrioInMs,
+            this.pauseWaitForHighPrioInMs));
   }
 
   public boolean releasePermit(
@@ -128,15 +134,19 @@ public boolean releasePermit(
     final PermitsPerNetworkSegment permitsPerNetworkSegment =
         this.permitsPerSegmentByConfig.get(networkSegment.throttlingConfigId());
     return permitsPerNetworkSegment != null
-        && permitsPerNetworkSegment.releasePermit(networkSegment, clientId, requestId);
+        && permitsPerNetworkSegment.releasePermit(
+            networkSegment, new PermitRequest(clientId, requestId));
   }
 
   public boolean discardPermit(final int clientId, final int requestId) {
     return this.permitService
         .findByClientIdAndRequestId(clientId, requestId)
         .map(
             permit ->
-                this.releasePermit(permit.networkSegment(), permit.clientId(), permit.requestId()))
+                this.releasePermit(
+                    permit.networkSegment(),
+                    permit.permitRequest().getClientId(),
+                    permit.permitRequest().getRequestId()))
         .orElse(false);
   }
 }

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/PermitsPerNetworkSegment.java

@@ -4,66 +4,66 @@
 
 package org.opensmartgridplatform.throttling;
 
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.opensmartgridplatform.shared.wsheaderattribute.priority.MessagePriorityEnum;
 import org.opensmartgridplatform.throttling.model.NetworkSegment;
+import org.opensmartgridplatform.throttling.model.PermitRequest;
 import org.opensmartgridplatform.throttling.model.ThrottlingSettings;
 import org.opensmartgridplatform.throttling.services.PermitService;
 import org.opensmartgridplatform.throttling.services.RateLimitService;
 
 @Slf4j
 public class PermitsPerNetworkSegment {
+
   private final PermitService permitService;
   private final RateLimitService rateLimitService;
   private final boolean highPrioPoolEnabled;
   private final int maxWaitForHighPrioInMs;
+  private final int pauseWaitForHighPrioInMs;
 
   public PermitsPerNetworkSegment(
       final PermitService permitService,
       final RateLimitService rateLimitService,
       final boolean highPrioPoolEnabled,
-      final int maxWaitForHighPrioInMs) {
+      final int maxWaitForHighPrioInMs,
+      final int pauseWaitForHighPrioInMs) {
     this.permitService = permitService;
     this.rateLimitService = rateLimitService;
     this.highPrioPoolEnabled = highPrioPoolEnabled;
     this.maxWaitForHighPrioInMs = maxWaitForHighPrioInMs;
+    this.pauseWaitForHighPrioInMs = pauseWaitForHighPrioInMs;
   }
 
   public boolean requestPermit(
       final NetworkSegment networkSegment,
-      final int clientId,
-      final int requestId,
+      final PermitRequest permitRequest,
       final int priority,
       final ThrottlingSettings throttlingSettings) {
 
     final boolean newConnectionRequestAllowed =
         this.rateLimitService.isNewConnectionRequestAllowed(
             networkSegment.baseTransceiverStationId(), networkSegment.cellId(), throttlingSettings);
 
-    log.debug(
-        "Request [{}] for permit is {} by rate-limiter",
-        requestId,
-        newConnectionRequestAllowed ? "allowed" : "NOT allowed");
-
     if (newConnectionRequestAllowed) {
-      log.debug("Request [{}] for permit is allowed by rate-limiter", requestId);
+      log.debug("Request [{}] for permit is allowed by rate-limiter", permitRequest.getRequestId());
       return this.tryAcquiringPermit(
-          networkSegment, clientId, requestId, priority, throttlingSettings.getMaxConcurrency());
+          networkSegment, permitRequest, priority, throttlingSettings.getMaxConcurrency());
     }
 
-    log.debug("Request [{}] for permit is NOT allowed by rate-limiter", requestId);
+    log.debug(
+        "Request [{}] for permit is NOT allowed by rate-limiter", permitRequest.getRequestId());
     return false;
   }
 
   public boolean releasePermit(
-      final NetworkSegment networkSegment, final int clientId, final int requestId) {
-    return this.permitService.removePermit(networkSegment, clientId, requestId);
+      final NetworkSegment networkSegment, final PermitRequest permitRequest) {
+    return this.permitService.removePermit(networkSegment, permitRequest);
   }
 
   private boolean tryAcquiringPermit(
       final NetworkSegment networkSegment,
-      final int clientId,
-      final int requestId,
+      final PermitRequest permitRequest,
       final int priority,
       final int maxConcurrency) {
 
@@ -72,23 +72,26 @@ private boolean tryAcquiringPermit(
       return false;
     }
 
+    final boolean highPrio =
+        this.highPrioPoolEnabled && priority > MessagePriorityEnum.DEFAULT.getPriority();
+
     final boolean granted =
-        this.permitService.createPermit(networkSegment, clientId, requestId, maxConcurrency);
+        this.permitService.createPermit(networkSegment, permitRequest, maxConcurrency, highPrio);
 
     if (granted) {
-      log.debug("Request [{}] is granted a permit.", requestId);
+      log.debug("Request [{}] is granted a permit.", permitRequest.getRequestId());
       return true;
 
     } else {
-      log.debug("Request [{}], is NOT granted a permit.", requestId);
+      log.debug("Request [{}], is NOT granted a permit.", permitRequest.getRequestId());
 
-      if (this.highPrioPoolEnabled && priority > MessagePriorityEnum.DEFAULT.getPriority()) {
+      if (highPrio) {
         log.debug(
             "Request [{}] is a high priority request and high priority pool is enabled -> we will wait for a permit release...",
-            requestId);
+            permitRequest.getRequestId());
 
         return this.waitUntilPermitIsAvailable(
-            networkSegment, clientId, requestId, maxConcurrency, this.maxWaitForHighPrioInMs);
+            networkSegment, permitRequest, maxConcurrency, this.maxWaitForHighPrioInMs);
       }
     }
 
@@ -97,26 +100,35 @@ private boolean tryAcquiringPermit(
 
   private boolean waitUntilPermitIsAvailable(
       final NetworkSegment networkSegment,
-      final int clientId,
-      final int requestId,
+      final PermitRequest permitRequest,
       final int maxConcurrency,
       final int maxWaitForHighPrioInMs) {
 
     final long startTime = System.currentTimeMillis();
 
-    log.debug("High priority request [{}] is waiting until permit is available.", requestId);
+    log.debug(
+        "High priority request [{}] is waiting until permit is available.",
+        permitRequest.getRequestId());
 
     while (System.currentTimeMillis() - startTime < maxWaitForHighPrioInMs) {
 
       final boolean granted =
-          this.permitService.createPermitWithHighPriority(
-              networkSegment, clientId, requestId, maxConcurrency);
+          this.permitService.createPermit(networkSegment, permitRequest, maxConcurrency, true);
 
       if (!granted) {
+
+        if (this.pauseWaitForHighPrioInMs > -1) {
+          try {
+            TimeUnit.MILLISECONDS.sleep(this.pauseWaitForHighPrioInMs);
+          } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+          }
+        }
+
         continue;
       }
 
-      log.debug("High priority request [{}] is granted a permit.", requestId);
+      log.debug("High priority request [{}] is granted a permit.", permitRequest.getRequestId());
       return true;
     }
 

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/config/RedisConfig.java

@@ -61,7 +61,7 @@ public class RedisConfig {
   @Value("${redis.connection.pool-size:64}")
   private int redisConnectionPoolSize;
 
-  @Bean
+  @Bean(destroyMethod = "shutdown")
   public RedissonClient redissonClient(final Config redissonConfig) {
     this.installJCAProvider();
     return Redisson.create(redissonConfig);

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/model/Permit.java

@@ -4,4 +4,4 @@
 
 package org.opensmartgridplatform.throttling.model;
 
-public record Permit(NetworkSegment networkSegment, int clientId, int requestId) {}
+public record Permit(NetworkSegment networkSegment, PermitRequest permitRequest) {}

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/model/PermitKey.java

@@ -13,6 +13,7 @@ public class PermitKey {
 
   private static final String KEY_FORMAT = "osgp-network-segment-permit-%s_%s_%s";
   private static final String LOCK_FORMAT = "osgp-network-segment-lock-%s_%s_%s";
+  private static final String LOBBY_FORMAT = "osgp-network-segment-lobby-%s_%s_%s";
 
   private short throttlingConfigID;
   private int baseTransceiverStationId;
@@ -42,6 +43,14 @@ public String lockId() {
         this.asString(this.cellId));
   }
 
+  public String lobby() {
+    return String.format(
+        LOBBY_FORMAT,
+        this.asString(this.throttlingConfigID),
+        this.asString(this.baseTransceiverStationId),
+        this.asString(this.cellId));
+  }
+
   private String asString(final int value) {
     return this.asString(value, false);
   }

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/model/PermitRequest.java

@@ -0,0 +1,23 @@
+// SPDX-FileCopyrightText: Copyright Contributors to the GXF project
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package org.opensmartgridplatform.throttling.model;
+
+import java.util.UUID;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@EqualsAndHashCode
+@Getter
+public class PermitRequest {
+  final int clientId;
+  final int requestId;
+  final UUID uuid;
+
+  public PermitRequest(final int clientId, final int requestId) {
+    this.clientId = clientId;
+    this.requestId = requestId;
+    this.uuid = UUID.randomUUID();
+  }
+}

osgp/platform/osgp-throttling-service/src/main/java/org/opensmartgridplatform/throttling/services/PermitService.java

@@ -7,23 +7,17 @@
 import java.util.Optional;
 import org.opensmartgridplatform.throttling.model.NetworkSegment;
 import org.opensmartgridplatform.throttling.model.Permit;
+import org.opensmartgridplatform.throttling.model.PermitRequest;
 
 public interface PermitService {
 
   boolean createPermit(
-      final NetworkSegment networkSegment,
-      final int clientId,
-      final int requestId,
-      final int maxConcurrentRequests);
+      NetworkSegment networkSegment,
+      PermitRequest permitRequest,
+      int maxConcurrentRequests,
+      boolean highPrioRequest);
 
-  boolean createPermitWithHighPriority(
-      final NetworkSegment networkSegment,
-      final int clientId,
-      final int requestId,
-      final int maxConcurrentRequests);
-
-  boolean removePermit(
-      final NetworkSegment networkSegment, final int clientId, final int requestId);
+  boolean removePermit(NetworkSegment networkSegment, PermitRequest permitRequest);
 
   Optional<Permit> findByClientIdAndRequestId(int clientId, int requestId);