Solve #1162, Dispatch actions to specified type of executor (#1332)

Author: littlezhou

smart-engine/src/main/java/org/smartdata/server/cluster/NodeInfo.java

@@ -0,0 +1,50 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.smartdata.server.cluster;
+
+
+/**
+ * Represent each nodes that SSM services (SmartServers and SmartAgents) running on.
+ *
+ */
+public class NodeInfo {
+  private String id;
+  private String location;
+
+  public NodeInfo(String id, String location) {
+    this.id = id;
+    this.location = location;
+  }
+
+  public String getId() {
+    return id;
+  }
+
+  public void setId(String id) {
+    this.id = id;
+  }
+
+  public String getLocation() {
+    return location;
+  }
+
+  public void setLocation(String location) {
+    this.location = location;
+  }
+}
+

smart-engine/src/main/java/org/smartdata/server/engine/ActiveServerInfo.java

@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.smartdata.server.engine;
+
+import org.smartdata.server.cluster.NodeInfo;
+
+public class ActiveServerInfo extends NodeInfo {
+  private static ActiveServerInfo inst;
+
+  private ActiveServerInfo() {
+    super("ActiveSSMServer", "127.0.0.1");
+  }
+
+  public static ActiveServerInfo getInstance() {
+    if (inst == null) {
+      inst = new ActiveServerInfo();
+    }
+    return inst;
+  }
+}

smart-engine/src/main/java/org/smartdata/server/engine/CmdletManager.java

@@ -92,13 +92,13 @@ public class CmdletManager extends AbstractService {
   private Map<String, Long> fileLocks;
   private ListMultimap<String, ActionScheduler> schedulers = ArrayListMultimap.create();
   private List<ActionSchedulerService> schedulerServices = new ArrayList<>();
+  private long totalScheduled = 0;
 
   public CmdletManager(ServerContext context) {
     super(context);
 
     this.metaStore = context.getMetaStore();
     this.executorService = Executors.newScheduledThreadPool(2);
-    this.dispatcher = new CmdletDispatcher(context, this);
     this.runningCmdlets = new ArrayList<>();
     this.pendingCmdlet = new LinkedList<>();
     this.schedulingCmdlet = new LinkedList<>();
@@ -107,6 +107,8 @@ public CmdletManager(ServerContext context) {
     this.idToCmdlets = new ConcurrentHashMap<>();
     this.idToActions = new ConcurrentHashMap<>();
     this.fileLocks = new ConcurrentHashMap<>();
+    this.dispatcher = new CmdletDispatcher(context, this, scheduledCmdlet,
+        idToLaunchCmdlet, runningCmdlets, schedulers);
   }
 
   @VisibleForTesting
@@ -270,18 +272,19 @@ private void syncCmdAction(CmdletInfo cmdletInfo,
   @Override
   public void start() throws IOException {
     LOG.info("Starting ...");
-    executorService.scheduleAtFixedRate(new ScheduleTask(), 100, 50,  TimeUnit.MILLISECONDS);
-    executorService.scheduleAtFixedRate(
-        new DispatchTask(this.dispatcher), 200, 100, TimeUnit.MILLISECONDS);
+    executorService.scheduleAtFixedRate(new ScheduleTask(), 100, 50, TimeUnit.MILLISECONDS);
+
     for (ActionSchedulerService s : schedulerServices) {
       s.start();
     }
+    dispatcher.start();
     LOG.info("Started.");
   }
 
   @Override
   public void stop() throws IOException {
     LOG.info("Stopping ...");
+    dispatcher.stop();
     for (int i = schedulerServices.size() - 1; i >= 0; i--) {
       schedulerServices.get(i).stop();
     }
@@ -967,44 +970,12 @@ public void run() {
         int nScheduled;
         do {
           nScheduled = scheduleCmdlet();
+          totalScheduled += nScheduled;
         } while (nScheduled != 0);
       } catch (IOException e) {
         LOG.error("Exception when Scheduling Cmdlet. "
             + scheduledCmdlet.size() + " cmdlets are pending for dispatch.", e);
       }
     }
   }
-
-  private class DispatchTask implements Runnable {
-    private final CmdletDispatcher dispatcher;
-
-    public DispatchTask(CmdletDispatcher dispatcher) {
-      this.dispatcher = dispatcher;
-    }
-
-    @Override
-    public void run() {
-      while (dispatcher.canDispatchMore()) {
-        try {
-          LaunchCmdlet launchCmdlet = getNextCmdletToRun();
-          if (launchCmdlet == null) {
-            break;
-          } else {
-            cmdletPreExecutionProcess(launchCmdlet);
-            dispatcher.dispatch(launchCmdlet);
-          }
-        } catch (IOException e) {
-          LOG.error("Cmdlet dispatcher error", e);
-        }
-      }
-    }
-  }
-
-  public void cmdletPreExecutionProcess(LaunchCmdlet cmdlet) {
-    for (LaunchAction action : cmdlet.getLaunchActions()) {
-      for (ActionScheduler p : schedulers.get(action.getActionType())) {
-        p.onPreDispatch(action);
-      }
-    }
-  }
 }

smart-engine/src/main/java/org/smartdata/server/engine/StandbyServerInfo.java

@@ -17,28 +17,10 @@
  */
 package org.smartdata.server.engine;
 
-public class StandbyServerInfo {
-  private String id;
-  private String location;
+import org.smartdata.server.cluster.NodeInfo;
 
+public class StandbyServerInfo extends NodeInfo {
   public StandbyServerInfo(String id, String location) {
-    this.id = id;
-    this.location = location;
-  }
-
-  public String getId() {
-    return id;
-  }
-
-  public void setId(String id) {
-    this.id = id;
-  }
-
-  public String getLocation() {
-    return location;
-  }
-
-  public void setLocation(String location) {
-    this.location = location;
+    super(id, location);
   }
 }

smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/metric/fetcher/DispatcherConf.java renamed to smart-engine/src/main/java/org/smartdata/server/engine/cmdlet/CmdletDispatchPolicy.java

@@ -15,8 +15,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.smartdata.hdfs.metric.fetcher;
+package org.smartdata.server.engine.cmdlet;
 
-public class DispatcherConf {
-  int maxConcurrentMovesPerNode;
+
+public enum CmdletDispatchPolicy {
+    ANY,
+    PREFER_AGENT,
+    PREFER_REMOTE_SSM,
+    PREFER_LOCAL;
+//    MUST_AGENT,
+//    MUST_REMOTE_SSM,
+//    MUST_LOCAL;
 }