refactor: sliding expiration cache with cleanup task (#389)

Author: sophia-bq

common/lib/host_list_provider/monitoring/monitoring_host_list_provider.ts

@@ -16,7 +16,6 @@
 
 import { RdsHostListProvider } from "../rds_host_list_provider";
 import { HostInfo } from "../../host_info";
-import { SlidingExpirationCache } from "../../utils/sliding_expiration_cache";
 import { ClusterTopologyMonitor, ClusterTopologyMonitorImpl } from "./cluster_topology_monitor";
 import { PluginService } from "../../plugin_service";
 import { HostListProviderService } from "../../host_list_provider_service";
@@ -27,23 +26,25 @@ import { Messages } from "../../utils/messages";
 import { WrapperProperties } from "../../wrapper_property";
 import { BlockingHostListProvider } from "../host_list_provider";
 import { logger } from "../../../logutils";
+import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task";
 import { isDialectTopologyAware } from "../../utils/utils";
 
 export class MonitoringRdsHostListProvider extends RdsHostListProvider implements BlockingHostListProvider {
   static readonly CACHE_CLEANUP_NANOS: bigint = BigInt(60_000_000_000); // 1 minute.
   static readonly MONITOR_EXPIRATION_NANOS: bigint = BigInt(15 * 60_000_000_000); // 15 minutes.
   static readonly DEFAULT_TOPOLOGY_QUERY_TIMEOUT_MS = 5000; // 5 seconds.
 
-  private static monitors: SlidingExpirationCache<string, ClusterTopologyMonitor> = new SlidingExpirationCache(
+  private static monitors: SlidingExpirationCacheWithCleanupTask<string, ClusterTopologyMonitor> = new SlidingExpirationCacheWithCleanupTask(
     MonitoringRdsHostListProvider.CACHE_CLEANUP_NANOS,
     () => true,
-    async (monitor: ClusterTopologyMonitor) => {
+    async (item: ClusterTopologyMonitor) => {
       try {
-        await monitor.close();
+        await item.close();
       } catch {
         // Ignore.
       }
-    }
+    },
+    "MonitoringRdsHostListProvider.monitors"
   );
 
   private readonly pluginService: PluginService;
@@ -55,13 +56,7 @@ export class MonitoringRdsHostListProvider extends RdsHostListProvider implement
 
   async clearAll(): Promise<void> {
     RdsHostListProvider.clearAll();
-    // TODO: refactor when sliding-expiration-cache refactoring is merged.
-    for (const [key, monitor] of MonitoringRdsHostListProvider.monitors.entries) {
-      if (monitor !== undefined) {
-        await monitor.item.close();
-      }
-    }
-    MonitoringRdsHostListProvider.monitors.clear();
+    await MonitoringRdsHostListProvider.monitors.clear();
   }
 
   async queryForTopology(targetClient: ClientWrapper, dialect: DatabaseDialect): Promise<HostInfo[]> {

common/lib/internal_pooled_connection_provider.ts

@@ -17,7 +17,6 @@
 import { PluginService } from "./plugin_service";
 import { WrapperProperties } from "./wrapper_property";
 import { CanReleaseResources } from "./can_release_resources";
-import { SlidingExpirationCache } from "./utils/sliding_expiration_cache";
 import { PoolKey } from "./utils/pool_key";
 import { PooledConnectionProvider } from "./pooled_connection_provider";
 import { HostInfo } from "./host_info";
@@ -39,14 +38,16 @@ import { AwsPoolConfig } from "./aws_pool_config";
 import { LeastConnectionsHostSelector } from "./least_connections_host_selector";
 import { PoolClientWrapper } from "./pool_client_wrapper";
 import { logger } from "../logutils";
+import { SlidingExpirationCacheWithCleanupTask } from "./utils/sliding_expiration_cache_with_cleanup_task";
 
 export class InternalPooledConnectionProvider implements PooledConnectionProvider, CanReleaseResources {
   static readonly CACHE_CLEANUP_NANOS: bigint = BigInt(10 * 60_000_000_000); // 10 minutes
   static readonly POOL_EXPIRATION_NANOS: bigint = BigInt(30 * 60_000_000_000); // 30 minutes
-  protected static databasePools: SlidingExpirationCache<string, any> = new SlidingExpirationCache(
+  protected static databasePools: SlidingExpirationCacheWithCleanupTask<string, any> = new SlidingExpirationCacheWithCleanupTask(
     InternalPooledConnectionProvider.CACHE_CLEANUP_NANOS,
     (pool: any) => pool.getActiveCount() === 0,
-    (pool: any) => pool.end()
+    async (pool: any) => await pool.end(),
+    "InternalPooledConnectionProvider.databasePools"
   );
 
   private static readonly acceptedStrategies: Map<string, HostSelector> = new Map([
@@ -122,16 +123,7 @@ export class InternalPooledConnectionProvider implements PooledConnectionProvide
   }
 
   async releaseResources() {
-    for (const [_key, value] of InternalPooledConnectionProvider.databasePools.entries) {
-      if (value.item) {
-        await value.item.releaseResources();
-      }
-    }
-    InternalPooledConnectionProvider.clearDatabasePools();
-  }
-
-  static clearDatabasePools() {
-    InternalPooledConnectionProvider.databasePools.clear();
+    await InternalPooledConnectionProvider.databasePools.clear();
   }
 
   getHostInfoByStrategy(hosts: HostInfo[], role: HostRole, strategy: string, props?: Map<string, any>): HostInfo {
@@ -177,7 +169,7 @@ export class InternalPooledConnectionProvider implements PooledConnectionProvide
   }
 
   // for testing only
-  setDatabasePools(connectionPools: SlidingExpirationCache<string, any>): void {
+  setDatabasePools(connectionPools: SlidingExpirationCacheWithCleanupTask<string, any>): void {
     InternalPooledConnectionProvider.databasePools = connectionPools;
     LeastConnectionsHostSelector.setDatabasePools(connectionPools);
   }

common/lib/plugins/efm/host_monitoring_connection_plugin.ts

@@ -203,6 +203,6 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp
   }
 
   async releaseResources(): Promise<void> {
-    return this.monitorService.releaseResources();
+    await this.monitorService.releaseResources();
   }
 }

common/lib/plugins/efm/monitor_service.ts

@@ -19,9 +19,9 @@ import { HostInfo } from "../../host_info";
 import { AwsWrapperError, IllegalArgumentError } from "../../utils/errors";
 import { Monitor, MonitorImpl } from "./monitor";
 import { WrapperProperties } from "../../wrapper_property";
-import { SlidingExpirationCache } from "../../utils/sliding_expiration_cache";
 import { PluginService } from "../../plugin_service";
 import { Messages } from "../../utils/messages";
+import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task";
 import { ClientWrapper } from "../../client_wrapper";
 
 export interface MonitorService {
@@ -44,10 +44,13 @@ export interface MonitorService {
 
 export class MonitorServiceImpl implements MonitorService {
   private static readonly CACHE_CLEANUP_NANOS = BigInt(60_000_000_000);
-  protected static readonly monitors: SlidingExpirationCache<string, Monitor> = new SlidingExpirationCache(
+  protected static readonly monitors: SlidingExpirationCacheWithCleanupTask<string, Monitor> = new SlidingExpirationCacheWithCleanupTask(
     MonitorServiceImpl.CACHE_CLEANUP_NANOS,
     undefined,
-    () => {}
+    async (monitor: Monitor) => {
+      await monitor.releaseResources();
+    },
+    "efm/MonitorServiceImpl.monitors"
   );
   private readonly pluginService: PluginService;
   private cachedMonitorHostKeys: Set<string> | undefined;
@@ -108,7 +111,7 @@ export class MonitorServiceImpl implements MonitorService {
   }
 
   stopMonitoringForAllConnections(hostKeys: Set<string>) {
-    let monitor;
+    let monitor: Monitor;
     for (const hostKey of hostKeys) {
       monitor = MonitorServiceImpl.monitors.get(hostKey);
       if (monitor) {
@@ -119,8 +122,8 @@ export class MonitorServiceImpl implements MonitorService {
   }
 
   async getMonitor(hostKeys: Set<string>, hostInfo: HostInfo, properties: Map<string, any>): Promise<Monitor | null> {
-    let monitor;
-    let anyHostKey;
+    let monitor: Monitor;
+    let anyHostKey: string;
     for (const hostKey of hostKeys) {
       monitor = MonitorServiceImpl.monitors.get(hostKey);
       anyHostKey = hostKey;
@@ -159,16 +162,13 @@ export class MonitorServiceImpl implements MonitorService {
   }
 
   async releaseResources() {
-    for (const [_key, monitor] of MonitorServiceImpl.monitors.entries) {
-      if (monitor.item) {
-        await monitor.item.releaseResources();
-      }
-    }
+    await MonitorServiceImpl.monitors.clear();
     this.cachedMonitorHostKeys = undefined;
     this.cachedMonitorRef = undefined;
   }
 
-  static clearMonitors() {
-    MonitorServiceImpl.monitors.clear();
+  // Used for performance testing.
+  static async clearMonitors() {
+    await MonitorServiceImpl.monitors.clear();
   }
 }

common/lib/plugins/efm2/host_monitoring2_connection_plugin.ts

@@ -171,6 +171,6 @@ export class HostMonitoring2ConnectionPlugin extends AbstractConnectionPlugin im
   }
 
   async releaseResources(): Promise<void> {
-    return this.monitorService.releaseResources();
+    await this.monitorService.releaseResources();
   }
 }

common/lib/plugins/efm2/monitor_service.ts

@@ -19,13 +19,13 @@ import { HostInfo } from "../../host_info";
 import { AwsWrapperError } from "../../utils/errors";
 import { Monitor, MonitorImpl } from "./monitor";
 import { WrapperProperties } from "../../wrapper_property";
-import { SlidingExpirationCache } from "../../utils/sliding_expiration_cache";
 import { PluginService } from "../../plugin_service";
 import { Messages } from "../../utils/messages";
 import { TelemetryCounter } from "../../utils/telemetry/telemetry_counter";
 import { TelemetryFactory } from "../../utils/telemetry/telemetry_factory";
 import { ClientWrapper } from "../../client_wrapper";
 import { logger } from "../../../logutils";
+import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task";
 
 export interface MonitorService {
   startMonitoring(
@@ -52,18 +52,19 @@ export interface MonitorService {
 export class MonitorServiceImpl implements MonitorService {
   private static readonly CACHE_CLEANUP_NANOS = BigInt(60_000_000_000);
 
-  protected static readonly monitors: SlidingExpirationCache<string, Monitor> = new SlidingExpirationCache(
+  protected static readonly monitors: SlidingExpirationCacheWithCleanupTask<string, Monitor> = new SlidingExpirationCacheWithCleanupTask(
     MonitorServiceImpl.CACHE_CLEANUP_NANOS,
     (monitor: Monitor) => monitor.canDispose(),
     async (monitor: Monitor) => {
       {
         try {
-          await monitor.endMonitoringClient();
+          await monitor.releaseResources();
         } catch (error) {
           // ignore
         }
       }
-    }
+    },
+    "efm2/MonitorServiceImpl.monitors"
   );
   private readonly pluginService: PluginService;
   private telemetryFactory: TelemetryFactory;
@@ -133,7 +134,6 @@ export class MonitorServiceImpl implements MonitorService {
     failureDetectionCount: number
   ): Promise<Monitor | null> {
     const monitorKey: string = `${failureDetectionTimeMillis.toString()} ${failureDetectionIntervalMillis.toString()} ${failureDetectionCount.toString()} ${hostInfo.host}`;
-
     const cacheExpirationNanos = BigInt(WrapperProperties.MONITOR_DISPOSAL_TIME_MS.get(properties) * 1_000_000);
     return MonitorServiceImpl.monitors.computeIfAbsent(
       monitorKey,
@@ -152,15 +152,6 @@ export class MonitorServiceImpl implements MonitorService {
   }
 
   async releaseResources() {
-    for (const [_key, monitor] of MonitorServiceImpl.monitors.entries) {
-      if (monitor.item) {
-        await monitor.item.releaseResources();
-      }
-    }
-    MonitorServiceImpl.clearMonitors();
-  }
-
-  static clearMonitors() {
-    MonitorServiceImpl.monitors.clear();
+    await MonitorServiceImpl.monitors.clear();
   }
 }

common/lib/plugins/limitless/limitless_router_service.ts

@@ -31,6 +31,7 @@ import { HostRole } from "../../host_role";
 import { HostAvailability } from "../../host_availability/host_availability";
 import { HighestWeightHostSelector } from "../../highest_weight_host_selector";
 import { sleep } from "../../utils/utils";
+import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task";
 
 export interface LimitlessRouterService {
   startMonitor(hostInfo: HostInfo, properties: Map<string, any>): void;
@@ -40,11 +41,13 @@ export interface LimitlessRouterService {
 
 export class LimitlessRouterServiceImpl implements LimitlessRouterService {
   protected static readonly CACHE_CLEANUP_NANOS = BigInt(60_000_000_000); // 1 min
-  protected static readonly monitors: SlidingExpirationCache<string, LimitlessRouterMonitor> = new SlidingExpirationCache(
-    LimitlessRouterServiceImpl.CACHE_CLEANUP_NANOS,
-    undefined,
-    async (monitor) => await monitor.close()
-  );
+  protected static readonly monitors: SlidingExpirationCacheWithCleanupTask<string, LimitlessRouterMonitor> =
+    new SlidingExpirationCacheWithCleanupTask(
+      LimitlessRouterServiceImpl.CACHE_CLEANUP_NANOS,
+      undefined,
+      async (monitor: LimitlessRouterMonitor) => await monitor.close(),
+      "LimitlessRouterServiceImpl.monitors"
+    );
   protected static readonly limitlessRouterCache: SlidingExpirationCache<string, HostInfo[]> = new SlidingExpirationCache(
     LimitlessRouterServiceImpl.CACHE_CLEANUP_NANOS,
     undefined,
@@ -254,7 +257,7 @@ export class LimitlessRouterServiceImpl implements LimitlessRouterService {
     }
   }
 
-  static clearMonitors() {
-    LimitlessRouterServiceImpl.monitors.clear();
+  static async clearMonitors() {
+    await LimitlessRouterServiceImpl.monitors.clear();
   }
 }

common/lib/plugins/strategy/fastest_response/host_response_time_service.ts

@@ -45,7 +45,7 @@ export class HostResponseTimeServiceImpl implements HostResponseTimeService {
   readonly intervalMs: number;
   protected hosts: HostInfo[];
   private readonly telemetryFactory: TelemetryFactory;
-  protected static monitoringHosts: SlidingExpirationCache<string, any> = new SlidingExpirationCache(
+  protected static monitoringHosts: SlidingExpirationCache<string, HostResponseTimeMonitor> = new SlidingExpirationCache(
     HostResponseTimeServiceImpl.CACHE_CLEANUP_NANOS,
     undefined,
     async (monitor: HostResponseTimeMonitor) => {

common/lib/utils/locales/en.json

@@ -233,6 +233,11 @@
   "HostMonitor.detectedWriter": "Detected writer: '%s'.",
   "HostMonitor.endMonitoring": "Host monitor '%s' completed in '%s'.",
   "HostMonitor.writerHostChanged": "Writer host has changed from '%s' to '%s'.",
+  "SlidingExpirationCacheWithCleanupTask.cleaningUp": "Cleanup interval of '%s' minutes has passed, cleaning up sliding expiration cache '%s'.",
+  "SlidingExpirationCacheWithCleanupTask.cleanUpTaskInterrupted": "Sliding expiration cache '%s' cleanup task has been interrupted and is exiting.",
+  "SlidingExpirationCacheWithCleanupTask.cleanUpTaskStopped": "Sliding expiration cache '%s' cleanup task has been stopped and is exiting.",
+  "SlidingExpirationCacheWithCleanupTask.clear": "Sliding expiration cache '%s' has been cleared, all resources are released.",
+  "SlidingExpirationCacheWithCleanupTask.cleanUpTaskInitialized": "Sliding expiration cache '%s' cleanup task has been initialized.",
   "HostMonitoringConnectionPlugin.monitoringDeactivated": "Monitoring deactivated for method '%s'.",
   "CustomEndpointPlugin.connectionRequestToCustomEndpoint": "Detected a connection request to a custom endpoint URL: '%s'.",
   "CustomEndpointPlugin.errorParsingEndpointIdentifier": "Unable to parse custom endpoint identifier from URL: '%s'.",

common/lib/utils/sliding_expiration_cache.ts

@@ -37,7 +37,7 @@ class CacheItem<V> {
 }
 
 export class SlidingExpirationCache<K, V> {
-  private _cleanupIntervalNanos: bigint = BigInt(10 * 60_000_000_000); // 10 minutes
+  protected _cleanupIntervalNanos: bigint = BigInt(10 * 60_000_000_000); // 10 minutes
   private readonly _shouldDisposeFunc?: (item: V) => boolean;
   private readonly _itemDisposalFunc?: (item: V) => void;
   map: Map<K, CacheItem<V>> = new Map<K, CacheItem<V>>();
@@ -116,7 +116,7 @@ export class SlidingExpirationCache<K, V> {
       return cacheItem;
     });
 
-    if (item != undefined && item != null && this._itemDisposalFunc != null) {
+    if (item != undefined && this._itemDisposalFunc != null) {
       this._itemDisposalFunc(item);
     }
   }