Fix a hot loop in DirectExchangeClient

Fix a hot loop in DirectExchangeClient

The fix changes queuedClients to LinkedHashSet, this helps with the filter condition
in scheduleRequestIfNecessary. In addition two for loops are modified to become one
loop on Iterator.

core/trino-main/src/main/java/io/trino/operator/DirectExchangeClient.java

@@ -35,8 +35,8 @@
 
 import java.io.Closeable;
 import java.net.URI;
-import java.util.Deque;
-import java.util.LinkedList;
+import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -73,7 +73,7 @@ public class DirectExchangeClient
     private final Map<URI, HttpPageBufferClient> allClients = new ConcurrentHashMap<>();
 
     @GuardedBy("this")
-    private final Deque<HttpPageBufferClient> queuedClients = new LinkedList<>();
+    private final Set<HttpPageBufferClient> queuedClients = new LinkedHashSet<>();
 
     private final Set<HttpPageBufferClient> completedClients = newConcurrentHashSet();
     private final DirectExchangeBuffer buffer;
@@ -284,17 +284,22 @@ synchronized int scheduleRequestIfNecessary()
         long projectedBytesToBeRequested = 0;
         int clientCount = 0;
 
-        for (HttpPageBufferClient client : queuedClients) {
+        Iterator<HttpPageBufferClient> clientIterator = queuedClients.iterator();
+        while (clientIterator.hasNext()) {
+            HttpPageBufferClient client = clientIterator.next();
             if (projectedBytesToBeRequested >= neededBytes * concurrentRequestMultiplier - reservedBytesForScheduledClients) {
                 break;
             }
             projectedBytesToBeRequested += client.getAverageRequestSizeInBytes();
-            clientCount++;
-        }
-        for (int i = 0; i < clientCount; i++) {
-            HttpPageBufferClient client = queuedClients.poll();
+
             client.scheduleRequest();
+
+            // Remove the client from the queuedClient's set.
+            clientIterator.remove();
+
+            clientCount++;
         }
+
         return clientCount;
     }
 
@@ -304,7 +309,7 @@ public ListenableFuture<Void> isBlocked()
     }
 
     @VisibleForTesting
-    Deque<HttpPageBufferClient> getQueuedClients()
+    Set<HttpPageBufferClient> getQueuedClients()
     {
         return queuedClients;
     }