<feature>[thread]: support coalesce queue for batch dhcp (ZSTAC-83039)

compute/src/main/java/org/zstack/compute/zone/AbstractZone.java

@@ -8,7 +8,6 @@
 import org.zstack.header.zone.ZoneStateEvent;
 
 abstract class AbstractZone implements Zone {
-	private static DatabaseFacade dbf = Platform.getComponentLoader().getComponent(DatabaseFacade.class);
 	private final static StateMachine<ZoneState, ZoneStateEvent> stateMachine;
 
     static {

core/src/main/java/org/zstack/core/thread/AbstractCoalesceQueue.java

@@ -0,0 +1,189 @@
+package org.zstack.core.thread;
+
+import org.springframework.beans.factory.annotation.Autowire;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Configurable;
+import org.zstack.header.core.AbstractCompletion;
+import org.zstack.header.core.Completion;
+import org.zstack.header.core.ReturnValueCompletion;
+import org.zstack.header.errorcode.ErrorCode;
+import org.zstack.utils.Utils;
+import org.zstack.utils.logging.CLogger;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * Base implementation for coalesce queues.
+ *
+ * @param <T> Request Item Type
+ * @param <R> Batch Execution Result Type
+ * @param <V> Single Request Result Type
+ */
+@Configurable(preConstruction = true, autowire = Autowire.BY_TYPE)
+public abstract class AbstractCoalesceQueue<T, R, V> {
+    private static final CLogger logger = Utils.getLogger(AbstractCoalesceQueue.class);
+
+    @Autowired
+    private ThreadFacade thdf;
+
+    private final ConcurrentHashMap<String, SignatureQueue> signatureQueues = new ConcurrentHashMap<>();
+
+    protected class PendingRequest {
+        final T item;
+        final AbstractCompletion completion;
+
+        PendingRequest(T item, AbstractCompletion completion) {
+            this.item = item;
+            this.completion = completion;
+        }
+
+        @SuppressWarnings("unchecked")
+        void notifySuccess(V result) {
+            if (completion == null) {
+                return;
+            }
+
+            if (completion instanceof ReturnValueCompletion) {
+                ((ReturnValueCompletion<V>) completion).success(result);
+            } else if (completion instanceof Completion) {
+                ((Completion) completion).success();
+            }
+        }
+
+        void notifyFailure(ErrorCode errorCode) {
+            if (completion == null) {
+                return;
+            }
+
+            if (completion instanceof ReturnValueCompletion) {
+                ((ReturnValueCompletion<V>) completion).fail(errorCode);
+            } else if (completion instanceof Completion) {
+                ((Completion) completion).fail(errorCode);
+            }
+        }
+    }
+
+    private class SignatureQueue {
+        final String syncSignature;
+        List<PendingRequest> pendingList = Collections.synchronizedList(new ArrayList<>());
+
+        SignatureQueue(String syncSignature) {
+            this.syncSignature = syncSignature;
+        }
+
+        synchronized List<PendingRequest> takeAll() {
+            List<PendingRequest> toProcess = pendingList;
+            pendingList = Collections.synchronizedList(new ArrayList<>());
+            return toProcess;
+        }
+
+        synchronized void add(PendingRequest request) {
+            pendingList.add(request);
+        }
+
+        synchronized boolean isEmpty() {
+            return pendingList.isEmpty();
+        }
+    }
+
+    protected abstract String getName();
+
+    // Changed to take AbstractCompletion, subclasses cast it to specific type
+    protected abstract void executeBatch(List<T> items, AbstractCompletion completion);
+
+    protected abstract AbstractCompletion createBatchCompletion(String syncSignature, List<PendingRequest> requests, SyncTaskChain chain);
+
+    protected abstract V calculateResult(T item, R batchResult);
+
+    protected final void handleSuccess(String syncSignature, List<PendingRequest> requests, R batchResult, SyncTaskChain chain) {
+        for (PendingRequest req : requests) {
+            try {
+                V singleResult = calculateResult(req.item, batchResult);
+                req.notifySuccess(singleResult);
+            } catch (Throwable t) {
+                logger.warn(String.format("[%s] failed to calculate result for item %s", getName(), req.item), t);
+                req.notifyFailure(org.zstack.core.Platform.operr("failed to calculate result: %s", t.getMessage()));
+            }
+        }
+        cleanup(syncSignature);
+        chain.next();
+    }
+
+    protected final void handleFailure(String syncSignature, List<PendingRequest> requests, ErrorCode errorCode, SyncTaskChain chain) {
+        for (PendingRequest req : requests) {
+            req.notifyFailure(errorCode);
+        }
+        cleanup(syncSignature);
+        chain.next();
+    }
+
+    void setThreadFacade(ThreadFacade thdf) {
+        this.thdf = thdf;
+    }
+
+    protected final void submitRequest(String syncSignature, T item, AbstractCompletion completion) {
+        doSubmit(syncSignature, new PendingRequest(item, completion));
+    }
+
+    private void doSubmit(String syncSignature, PendingRequest request) {
+        SignatureQueue queue = signatureQueues.computeIfAbsent(syncSignature, SignatureQueue::new);
+        queue.add(request);
+
+        thdf.chainSubmit(new ChainTask(null) {
+            @Override
+            public String getSyncSignature() {
+                return String.format("coalesce-queue-%s-%s", AbstractCoalesceQueue.this.getName(), syncSignature);
+            }
+
+            @Override
+            public void run(SyncTaskChain chain) {
+                List<PendingRequest> requests = queue.takeAll();
+
+                if (requests.isEmpty()) {
+                    chain.next();
+                    return;
+                }
+
+                String name = getName();
+                logger.debug(String.format("[%s] coalescing %d requests for signature[%s]",
+                        name, requests.size(), syncSignature));
+
+
+                // Create the specific completion type (Completion or ReturnValueCompletion)
+                AbstractCompletion batchCompletion = createBatchCompletion(syncSignature, requests, chain);
+
+                // Execute batch with the direct completion object
+                List<T> items = requests.stream().map(req -> req.item).collect(Collectors.toList());
+                executeBatch(items, batchCompletion);
+            }
+
+            @Override
+            public String getName() {
+                return String.format("%s-coalesced-batch-%s", AbstractCoalesceQueue.this.getName(), syncSignature);
+            }
+
+            @Override
+            protected int getSyncLevel() {
+                return 1;
+            }
+        });
+    }
+
+    private void cleanup(String syncSignature) {
+        signatureQueues.computeIfPresent(syncSignature, (k, queue) -> {
+            if (queue.isEmpty()) {
+                return null;
+            }
+            return queue;
+        });
+    }
+
+    // For testing
+    int getActiveQueueCount() {
+        return signatureQueues.size();
+    }
+}

core/src/main/java/org/zstack/core/thread/CoalesceQueue.java

@@ -0,0 +1,61 @@
+package org.zstack.core.thread;
+
+import org.zstack.header.core.AbstractCompletion;
+import org.zstack.header.core.Completion;
+import org.zstack.header.errorcode.ErrorCode;
+
+import java.util.List;
+
+/**
+ * A coalesce queue for requests that do NOT expect a return value.
+ *
+ * @param <T> Request Item Type
+ */
+public abstract class CoalesceQueue<T> extends AbstractCoalesceQueue<T, Void, Void> {
+
+    /**
+     * Submit a request.
+     *
+     * @param syncSignature the sync signature; requests with the same signature will be coalesced
+     * @param item          the request item
+     * @param completion    the completion callback
+     */
+    public void submit(String syncSignature, T item, Completion completion) {
+        submitRequest(syncSignature, item, completion);
+    }
+
+    /**
+     * Executes the batched requests.
+     * <p>
+     * Subclasses must implement this method to process the coalesced items.
+     *
+     * @param items      the list of coalesced request items
+     * @param completion the completion callback for the batch execution
+     */
+    protected abstract void executeBatch(List<T> items, Completion completion);
+
+    @Override
+    protected final void executeBatch(List<T> items, AbstractCompletion batchCompletion) {
+        executeBatch(items, (Completion) batchCompletion);
+    }
+
+    @Override
+    protected final AbstractCompletion createBatchCompletion(String syncSignature, List<PendingRequest> requests, SyncTaskChain chain) {
+        return new Completion(chain) {
+            @Override
+            public void success() {
+                handleSuccess(syncSignature, requests, null, chain);
+            }
+
+            @Override
+            public void fail(ErrorCode errorCode) {
+                handleFailure(syncSignature, requests, errorCode, chain);
+            }
+        };
+    }
+
+    @Override
+    protected final Void calculateResult(T item, Void batchResult) {
+        return null;
+    }
+}

core/src/main/java/org/zstack/core/thread/ReturnValueCoalesceQueue.java

@@ -0,0 +1,43 @@
+package org.zstack.core.thread;
+
+import org.zstack.header.core.AbstractCompletion;
+import org.zstack.header.core.ReturnValueCompletion;
+import org.zstack.header.errorcode.ErrorCode;
+
+import java.util.List;
+
+/**
+ * A coalesce queue for requests that expect a return value.
+ *
+ * @param <T> Request Item Type
+ * @param <R> Batch Execution Result Type
+ * @param <V> Single Request Result Type
+ */
+public abstract class ReturnValueCoalesceQueue<T, R, V> extends AbstractCoalesceQueue<T, R, V> {
+
+    public void submit(String syncSignature, T item, ReturnValueCompletion<V> completion) {
+        submitRequest(syncSignature, item, completion);
+    }
+
+    protected abstract void executeBatch(List<T> items, ReturnValueCompletion<R> completion);
+
+    @Override
+    protected final void executeBatch(List<T> items, AbstractCompletion batchCompletion) {
+        executeBatch(items, (ReturnValueCompletion<R>) batchCompletion);
+    }
+
+    @Override
+    protected final AbstractCompletion createBatchCompletion(String syncSignature, List<PendingRequest> requests, SyncTaskChain chain) {
+        return new ReturnValueCompletion<R>(null) {
+            @Override
+            public void success(R batchResult) {
+                handleSuccess(syncSignature, requests, batchResult, chain);
+            }
+
+            @Override
+            public void fail(ErrorCode errorCode) {
+                handleFailure(syncSignature, requests, errorCode, chain);
+            }
+        };
+    }
+}

plugin/flatNetworkProvider/src/main/java/org/zstack/network/service/flat/FlatDhcpBackend.java

@@ -18,8 +18,7 @@
 import org.zstack.core.db.SQL;
 import org.zstack.core.defer.Defer;
 import org.zstack.core.defer.Deferred;
-import org.zstack.core.thread.SyncTask;
-import org.zstack.core.thread.ThreadFacade;
+import org.zstack.core.thread.*;
 import org.zstack.core.upgrade.GrayVersion;
 import org.zstack.core.workflow.SimpleFlowChain;
 import org.zstack.header.AbstractService;
@@ -121,6 +120,54 @@ public class FlatDhcpBackend extends AbstractService implements NetworkServiceDh
 
     private Map<String, L3NetworkGetIpStatisticExtensionPoint> getIpStatisticExts = new HashMap<>();
 
+    /**
+     * Request wrapper for DHCP apply coalescing.
+     */
+    private static class DhcpApplyRequest {
+        final String hostUuid;
+        final List<DhcpInfo> dhcpInfos;
+        final boolean rebuild;
+
+        DhcpApplyRequest(String hostUuid, List<DhcpInfo> dhcpInfos, boolean rebuild) {
+            this.hostUuid = hostUuid;
+            this.dhcpInfos = dhcpInfos;
+            this.rebuild = rebuild;
+        }
+    }
+
+    private class DhcpApplyQueue extends CoalesceQueue<DhcpApplyRequest> {
+        @Override
+        protected String getName() {
+            return "flat-dhcp-apply";
+        }
+
+        @Override
+        protected void executeBatch(List<DhcpApplyRequest> requests, Completion completion) {
+            if (requests.isEmpty()) {
+                completion.success();
+                return;
+            }
+
+            // All requests in the same batch have the same hostUuid
+            String hostUuid = requests.get(0).hostUuid;
+
+            // Merge all DhcpInfo from all requests, grouped by L3 network
+            // TODO: unify DHCP apply logic and switch to merged/batch flow everywhere
+            boolean anyRebuild = false;
+            List<DhcpInfo> mergedInfos = new ArrayList<>();
+            for (DhcpApplyRequest req : requests) {
+                anyRebuild = anyRebuild || req.rebuild;
+                mergedInfos.addAll(req.dhcpInfos);
+            }
+
+            logger.debug(String.format("Coalesced %d DHCP apply requests for host[uuid:%s]", requests.size(), hostUuid));
+
+            applyDhcpToHosts(mergedInfos, hostUuid, anyRebuild, completion);
+        }
+    }
+
+    private final DhcpApplyQueue dhcpApplyCoalesceQueue = new DhcpApplyQueue();
+
     public static final String APPLY_DHCP_PATH = "/flatnetworkprovider/dhcp/apply";
     public static final String BATCH_APPLY_DHCP_PATH = "/flatnetworkprovider/dhcp/batchApply";
     public static final String PREPARE_DHCP_PATH = "/flatnetworkprovider/dhcp/prepare";
@@ -2074,7 +2121,10 @@ public void applyDhcpService(List<DhcpStruct> dhcpStructList, VmInstanceSpec spe
             return;
         }
 
-        applyDhcpToHosts(toDhcpInfo(dhcpStructList), spec.getDestHost().getUuid(), false, completion);
+        String hostUuid = spec.getDestHost().getUuid();
+        DhcpApplyRequest request = new DhcpApplyRequest(hostUuid, toDhcpInfo(dhcpStructList), false);
+        // Use coalesce queue: requests to the same host will be merged into a single batch
+        dhcpApplyCoalesceQueue.submit(hostUuid, request, completion);
     }
 
     private void releaseDhcpService(List<DhcpInfo> info, final String vmUuid, final String hostUuid, final NoErrorCompletion completion) {