Re-work the flow control for host allocation

Signed-off-by: Qun Li <qun.li@zstack.io>

Author: live4thee

compute/src/main/java/org/zstack/compute/allocator/HostAllocatorGlobalConfig.java

@@ -21,5 +21,5 @@ public class HostAllocatorGlobalConfig {
     @GlobalConfigValidation
     public static GlobalConfig PAGINATION_LIMIT = new GlobalConfig(CATEGORY, "paginationLimit");
     @GlobalConfigValidation
-    public static GlobalConfig HOST_ALLOCATOR_CONCURRENT_LEVEL = new GlobalConfig(CATEGORY, "hostAllocator.concurrentLevel");
+    public static GlobalConfig HOST_ALLOCATOR_ALLOW_CONCURRENT = new GlobalConfig(CATEGORY, "hostAllocator.concurrent");
 }

compute/src/main/java/org/zstack/compute/allocator/HostAllocatorManagerImpl.java

@@ -9,11 +9,14 @@
 import org.zstack.core.db.SimpleQuery;
 import org.zstack.core.db.SimpleQuery.Op;
 import org.zstack.core.errorcode.ErrorFacade;
-import org.zstack.core.thread.SyncTask;
+import org.zstack.core.thread.ChainTask;
+import org.zstack.core.thread.SyncTaskChain;
 import org.zstack.core.thread.ThreadFacade;
 import org.zstack.header.AbstractService;
 import org.zstack.header.allocator.*;
 import org.zstack.header.cluster.ReportHostCapacityMessage;
+import org.zstack.header.core.Completion;
+import org.zstack.header.core.NopeCompletion;
 import org.zstack.header.core.ReturnValueCompletion;
 import org.zstack.header.core.workflow.Flow;
 import org.zstack.header.core.workflow.FlowRollback;
@@ -302,6 +305,40 @@ private void handle(ReportHostCapacityMessage msg) {
     }
 
     private void handle(final AllocateHostMsg msg) {
+        if (HostAllocatorGlobalConfig.HOST_ALLOCATOR_ALLOW_CONCURRENT.value(Boolean.class)) {
+            doHandleAllocateHost(msg, new NopeCompletion(null));
+            return;
+        }
+
+        thdf.chainSubmit(new ChainTask(msg) {
+            @Override
+            public String getSyncSignature() {
+                return "host-allocator";
+            }
+
+            @Override
+            public void run(SyncTaskChain chain) {
+                doHandleAllocateHost(msg, new Completion(chain) {
+                    @Override
+                    public void success() {
+                        chain.next();
+                    }
+
+                    @Override
+                    public void fail(ErrorCode errorCode) {
+                        chain.next();
+                    }
+                });
+            }
+
+            @Override
+            public String getName() {
+                return "allocate-host-for-vm-" + msg.getVmInstance().getUuid();
+            }
+        });
+    }
+
+    private void doHandleAllocateHost(final AllocateHostMsg msg, Completion completion) {
         HostAllocatorSpec spec = HostAllocatorSpec.fromAllocationMsg(msg);
         spec.setBackupStoragePrimaryStorageMetrics(backupStoragePrimaryStorageMetrics);
 
@@ -333,19 +370,23 @@ public void success(List<HostInventory> hosts) {
                     if (hosts.isEmpty()){
                         reply.setHosts(new ArrayList<>());
                         bus.reply(msg, reply);
+                        completion.success();
                         return;
                     }
+
                     sortors.dryRunSort(spec, hosts, new ReturnValueCompletion<List<HostInventory>>(msg) {
                         @Override
                         public void success(List<HostInventory> returnValue) {
                             reply.setHosts(returnValue);
                             bus.reply(msg, reply);
+                            completion.success();
                         }
 
                         @Override
                         public void fail(ErrorCode errorCode) {
                             reply.setError(errorCode);
                             bus.reply(msg, reply);
+                            completion.fail(errorCode);
                         }
                     });
                 }
@@ -354,50 +395,27 @@ public void fail(ErrorCode errorCode) {
                 public void fail(ErrorCode errorCode) {
                     reply.setError(errorCode);
                     bus.reply(msg, reply);
+                    completion.fail(errorCode);
                 }
             });
         } else {
             final AllocateHostReply reply = new AllocateHostReply();
             strategy.allocate(spec, new ReturnValueCompletion<List<HostInventory>>(msg) {
                 @Override
                 public void success(List<HostInventory> hosts) {
-                    // flow control for host reservation
-                    Integer n = HostAllocatorGlobalConfig.HOST_ALLOCATOR_CONCURRENT_LEVEL.value(Integer.class);
-                    final int syncLevel = (n == null || n < 0) ? 0 : n;
-
-                    thdf.syncSubmit(new SyncTask<Void>() {
-                        @Override
-                        public String getSyncSignature() {
-                            return "host-reserve-flow-control";
-                        }
-
+                    sortors.sort(spec, hosts, new ReturnValueCompletion<HostInventory>(msg) {
                         @Override
-                        public int getSyncLevel() {
-                            return syncLevel;
-                        }
-
-                        @Override
-                        public String getName() {
-                            return "reserve-host";
+                        public void success(HostInventory returnValue) {
+                            reply.setHost(returnValue);
+                            bus.reply(msg, reply);
+                            completion.success();
                         }
 
                         @Override
-                        public Void call() throws Exception {
-                            sortors.sort(spec, hosts, new ReturnValueCompletion<HostInventory>(msg) {
-                                @Override
-                                public void success(HostInventory returnValue) {
-                                    reply.setHost(returnValue);
-                                    bus.reply(msg, reply);
-                                }
-
-                                @Override
-                                public void fail(ErrorCode errorCode) {
-                                    reply.setError(errorCode);
-                                    bus.reply(msg, reply);
-                                }
-                            });
-
-                            return null;
+                        public void fail(ErrorCode errorCode) {
+                            reply.setError(errorCode);
+                            bus.reply(msg, reply);
+                            completion.fail(errorCode);
                         }
                     });
                 }
@@ -406,6 +424,7 @@ public void fail(ErrorCode errorCode) {
                 public void fail(ErrorCode errorCode) {
                     reply.setError(errorCode);
                     bus.reply(msg, reply);
+                    completion.fail(errorCode);
                 }
             });
         }

conf/globalConfig/hostAllocator.xml

@@ -41,10 +41,10 @@
     </config>
 
     <config>
-        <name>hostAllocator.concurrentLevel</name>
-        <description>maximum concurrency number of host allocator</description>
+        <name>hostAllocator.concurrent</name>
+        <description>allow host allocator run concurrently</description>
         <category>hostAllocator</category>
-        <defaultValue>2</defaultValue>
-        <type>java.lang.Integer</type>
+        <defaultValue>false</defaultValue>
+        <type>java.lang.Boolean</type>
     </config>
 </globalConfig>