More improvments

- Decr. sleep to 5ms within submit_task() calls
- Removed lock base SDKClient booking from pool

Author: ashwin2002

src/main/java/RestServer/TaskRequest.java

@@ -702,7 +702,7 @@ public ResponseEntity<Map<String, Object>> doc_load() {
         }
 
         // Calculate effective number of workers needed
-        int effectiveWorkers = Math.min(ws.workers, 
+        int effectiveWorkers = Math.min(ws.workers,
             (int)((totalDocsToProcess + docsPerWorker - 1) / docsPerWorker));  // ceil division
 
         System.out.println("Smart worker counting: Total docs=" + totalDocsToProcess +

src/main/java/couchbase/sdk/SDKClientPool.java

@@ -85,10 +85,10 @@ public void create_clients(String bucket_name, Server server, int req_clients) t
     }
 
     public SDKClient get_client_for_bucket(String bucket_name, String scope, String collection) {
-        String cache_key = bucket_name + ":" + scope + ":" + collection;
-
+        String col_name = scope + collection;
+        
         // Check if client is already cached for this collection
-        ClientInfo existing = clientCache.get(cache_key);
+        ClientInfo existing = clientCache.get(col_name);
         if (existing != null) {
             existing.counter.incrementAndGet();
             return existing.client;
@@ -113,7 +113,7 @@ public SDKClient get_client_for_bucket(String bucket_name, String scope, String
         busyClients.computeIfAbsent(bucket_name, k -> new ConcurrentLinkedQueue<>()).add(client);
 
         // Cache client reference with thread-safe counter
-        clientCache.put(cache_key, new ClientInfo(client, new AtomicInteger(1)));
+        clientCache.put(col_name, new ClientInfo(client, new AtomicInteger(1)));
 
         return client;
     }
@@ -122,22 +122,22 @@ public void release_client(SDKClient client) {
         if (client == null || client.bucket == null) {
             return;
         }
-
+        
         String bucket_key = client.bucket;
-        String cache_key = bucket_key + ":" + client.scope + ":" + client.collection;
-
+        String col_name = client.scope + client.collection;
+        
         // Get cached client info
-        ClientInfo info = clientCache.get(cache_key);
+        ClientInfo info = clientCache.get(col_name);
         if (info == null) {
             return;
         }
-
+        
         // Decrement counter atomically
         int newCount = info.counter.decrementAndGet();
-
+        
         if (newCount == 0) {
             // Remove from cache atomically
-            clientCache.remove(cache_key);
+            clientCache.remove(col_name);
 
             // Remove from busy pool and add to idle pool atomically
             ConcurrentLinkedQueue<SDKClient> busyPool = busyClients.get(bucket_key);

src/main/java/couchbase/sdk/SharedClusterManager.java

@@ -23,17 +23,17 @@
  */
 public class SharedClusterManager {
     private static Logger logger = LogManager.getLogger(SharedClusterManager.class);
-    
+
     // Common KV connections setting for massively parallel collection loads
     // Increased from default 5 to 500 to support 5,000 collections loading in parallel
-    private static final int DEFAULT_KV_CONNECTIONS = 500;
-    
+    private static final int DEFAULT_KV_CONNECTIONS = 5;
+
     // Shared ClusterEnvironment with optimized connection settings
     private static ClusterEnvironment sharedEnvironment;
-    
+
     // Store cluster instances per server connection string
     private static ConcurrentHashMap<String, ClusterWrapper> clusterMap = new ConcurrentHashMap<>();
-    
+
     // Initialize the shared environment once (lazy initialization)
     private static void initializeSharedEnvironment() {
         if (sharedEnvironment == null) {
@@ -51,7 +51,7 @@ private static void initializeSharedEnvironment() {
             }
         }
     }
-    
+
     /**
      * Get or create a shared Cluster instance for the given server connection
      */
@@ -72,27 +72,27 @@ public static synchronized Cluster getCluster(Server server) throws Authenticati
 
         return wrapper.cluster;
     }
-    
+
     /**
      * Release reference to the shared Cluster instance
      */
     public static synchronized void releaseCluster(Server server) {
         String clusterKey = getClusterKey(server);
         ClusterWrapper wrapper = clusterMap.get(clusterKey);
-        
+
         if (wrapper != null) {
             int refCount = wrapper.decrementRefCount();
-            logger.debug("Released Cluster instance for server: " + server.ip + 
+            logger.debug("Released Cluster instance for server: " + server.ip +
                         " (ref count: " + refCount + ")");
-            
+
             if (refCount == 0) {
                 logger.info("No more references, disconnecting Cluster for server: " + server.ip);
                 wrapper.cluster.disconnect();
                 clusterMap.remove(clusterKey);
             }
         }
     }
-    
+
     /**
      * Shutdown all cluster instances and the shared environment
      */
@@ -110,7 +110,7 @@ public static synchronized void shutdownAll() {
             logger.info("Shared Cluster Environment shutdown complete");
         }
     }
-    
+
     private static Cluster createCluster(Server server) throws AuthenticationFailureException {
         ClusterOptions clusterOptions;
         try {
@@ -121,52 +121,52 @@ private static Cluster createCluster(Server server) throws AuthenticationFailure
                 clusterOptions = ClusterOptions.clusterOptions(server.rest_username, server.rest_password)
                         .environment(createNonTLSEnvironment());
             }
-            
+
             Cluster cluster = Cluster.connect(server.ip, clusterOptions);
             logger.info("Cluster connection successful: " + server.ip);
             return cluster;
         } catch (AuthenticationFailureException e) {
-            logger.error("Authentication failed for server: " + server.ip + 
+            logger.error("Authentication failed for server: " + server.ip +
                         " with user: " + server.rest_username);
             throw e;
         } catch (Exception e) {
             logger.error("Failed to connect Cluster to server: " + server.ip, e);
             throw new RuntimeException("Cluster connection failed", e);
         }
     }
-    
+
     private static ClusterEnvironment createNonTLSEnvironment() {
         return ClusterEnvironment.builder()
                 .timeoutConfig(TimeoutConfig.builder().kvTimeout(Duration.ofSeconds(10)))
                 .ioConfig(IoConfig.enableDnsSrv(true))
                 .ioConfig(IoConfig.numKvConnections(DEFAULT_KV_CONNECTIONS))
                 .build();
     }
-    
+
     private static String getClusterKey(Server server) {
         return server.ip + ":" + server.memcached_port;
     }
-    
+
     /**
      * Wrapper class to track reference count for shared Cluster instances
      */
     private static class ClusterWrapper {
         Cluster cluster;
         AtomicInteger refCount;
-        
+
         ClusterWrapper(Cluster cluster) {
             this.cluster = cluster;
             this.refCount = new AtomicInteger(1);
         }
-        
+
         void incrementRefCount() {
             refCount.incrementAndGet();
         }
-        
+
         int decrementRefCount() {
             return refCount.decrementAndGet();
         }
-        
+
         int getRefCount() {
             return refCount.get();
         }