More improvements added.

Author: ritalrw

src/main/java/Loader.java

@@ -145,7 +145,7 @@ public static void main(String[] args) throws IOException {
         Option validate = new Option("validate", "validate", true, "Validate Data during Reads");
         options.addOption(validate);
 
-        Option gtm = new Option("gtm", "gtm", true, "Go for max doc ops");
+        Option gtm = new Option("gtm", "gtm", true, "Go for max doc ops (disables rate limiting)");
         options.addOption(gtm);
 
         Option deleted = new Option("deleted", "deleted", true, "To verify deleted docs");

src/main/java/couchbase/loadgen/WorkLoadGenerate.java

@@ -388,10 +388,14 @@ else if(ops < dg.ws.ops/dg.ws.workers && flag) {
             ops = 0;
             Instant end = Instant.now();
             timeElapsed = Duration.between(start, end);
-            if(!this.dg.ws.gtm && timeElapsed.toMillis() < 1000)
+            // Smarter throttling: only sleep if significantly under time limit, and reduce sleep duration
+            if(!this.dg.ws.gtm && timeElapsed.toMillis() < 900)
                 try {
-                    long i =  (long) ((1000-timeElapsed.toMillis()));
-                    TimeUnit.MILLISECONDS.sleep(i);
+                    // Reduce sleep to 70% of needed time to allow burst periods
+                    long sleepTime = (long) ((900-timeElapsed.toMillis()) * 0.7);
+                    if(sleepTime > 10) { // Only sleep if > 10ms
+                        TimeUnit.MILLISECONDS.sleep(sleepTime);
+                    }
                 } catch (InterruptedException e) {
                     e.printStackTrace();
                 }

src/main/java/couchbase/sdk/DocOps.java

@@ -38,7 +38,7 @@ public List<Result> bulkInsert(Collection collection, List<Tuple2<String, Object
 
         // Emit error Results as part of the stream and collect at the end
         // This is thread-safe and avoids synchronization overhead
-        int concurrency = Math.min(documents.size(), 1000);
+        int concurrency = Math.min(documents.size(), 5000);
         return Flux.fromIterable(documents)
                 .flatMap(documentToInsert -> {
                   String k = documentToInsert.getT1();
@@ -58,7 +58,7 @@ public List<Result> bulkUpsert(Collection collection, List<Tuple2<String, Object
 
         // Emit error Results as part of the stream and collect at the end
         // This is thread-safe and avoids synchronization overhead
-        int concurrency = Math.min(documents.size(), 1000);
+        int concurrency = Math.min(documents.size(), 5000);
         return Flux.fromIterable(documents)
                 .flatMap(documentToInsert -> {
                   String k = documentToInsert.getT1();
@@ -74,7 +74,7 @@ public List<Result> bulkUpsert(Collection collection, List<Tuple2<String, Object
 
     public List<Tuple2<String, Object>> bulkGets(Collection collection, List<Tuple2<String, Object>> documents, GetOptions getOptions) {
         final ReactiveCollection reactiveCollection = collection.reactive();
-        int concurrency = Math.min(documents.size(), 1000);
+        int concurrency = Math.min(documents.size(), 5000);
         List<Tuple2<String, Object>> returnValue = Flux.fromIterable(documents)
                 .flatMap(new Function<Tuple2<String, Object>, Publisher<Tuple2<String, Object>>>() {
                     public Publisher<Tuple2<String, Object>> apply(Tuple2<String, Object> documentToInsert) {
@@ -105,7 +105,7 @@ public List<Result> bulkDelete(Collection collection, List<String> keys, RemoveO
 
         // Emit error Results as part of the stream and collect at the end
         // This is thread-safe and avoids synchronization overhead
-        int concurrency = Math.min(keys.size(), 1000);
+        int concurrency = Math.min(keys.size(), 5000);
         return Flux.fromIterable(keys)
                 .flatMap(key -> {
                   return reactiveCollection.remove(key, removeOptions)
@@ -119,7 +119,7 @@ public List<Result> bulkDelete(Collection collection, List<String> keys, RemoveO
     public List<ConcurrentHashMap<String, Object>> bulkReplace(Collection collection, List<Tuple2<String, Object>> documents,
             ReplaceOptions replaceOptions) {
         final ReactiveCollection reactiveCollection = collection.reactive();
-        int concurrency = Math.min(documents.size(), 1000);
+        int concurrency = Math.min(documents.size(), 5000);
         List<ConcurrentHashMap<String, Object>> returnValue = Flux.fromIterable(documents)
                 .flatMap(new Function<Tuple2<String, Object>, Publisher<ConcurrentHashMap<String, Object>>>() {
                     public Publisher<ConcurrentHashMap<String, Object>> apply(Tuple2<String, Object> documentToInsert) {
@@ -153,7 +153,7 @@ public Mono<ConcurrentHashMap<String, Object>> apply(Throwable error) {
     public List<ConcurrentHashMap<String, Object>> bulkTouch(Collection collection, List<String> keys, final int exp,
             TouchOptions touchOptions, Duration exp_duration) {
         final ReactiveCollection reactiveCollection = collection.reactive();
-        int concurrency = Math.min(keys.size(), 1000);
+        int concurrency = Math.min(keys.size(), 5000);
         List<ConcurrentHashMap<String, Object>> returnValue = Flux.fromIterable(keys)
                 .flatMap(new Function<String, Publisher<ConcurrentHashMap<String, Object>>>() {
                     public Publisher<ConcurrentHashMap<String, Object>> apply(String key){

src/main/java/couchbase/sdk/SDKClient.java

@@ -35,12 +35,13 @@ public class SDKClient {
             .securityConfig(SecurityConfig.enableTls(true)
             .trustManagerFactory(InsecureTrustManagerFactory.INSTANCE))
             .ioConfig(IoConfig.enableDnsSrv(true))
-            .ioConfig(IoConfig.numKvConnections(200))
+            .ioConfig(IoConfig.numKvConnections(500))
             .build();
 
     public static ClusterEnvironment env2 = ClusterEnvironment.builder()
             .timeoutConfig(TimeoutConfig.builder().kvTimeout(Duration.ofSeconds(10)))
-            .ioConfig(IoConfig.enableDnsSrv(true)).ioConfig(IoConfig.numKvConnections(200))
+            .ioConfig(IoConfig.enableDnsSrv(true))
+            .ioConfig(IoConfig.numKvConnections(500))
             .build();
 
     public SDKClient(Server master, String bucket, String scope, String collection) {