Enable worker core pinning in CPU nightly benchmark

.github/workflows/benchmark_nightly.yml

@@ -43,8 +43,6 @@ jobs:
           pip install -r benchmarks/requirements-ab.txt
       - name: Benchmark cpu nightly
         if: ${{ matrix.hardware == 'cpu' }}
-        env:
-          OMP_NUM_THREADS: 1
         run: python benchmarks/auto_benchmark.py --input benchmarks/benchmark_config_cpu.yaml --skip false
       - name: Benchmark gpu nightly
         if: ${{ matrix.hardware == 'gpu' }}

benchmarks/auto_benchmark.py

@@ -80,6 +80,12 @@ def report_cmd(self, cmd):
 
         self.bm_config["report_cmd"] = " ".join(cmd_options)
 
+    def enable_launcher_with_logical_core(self):
+        if self.bm_config["hardware"] == "cpu":
+            with open("./benchmarks/config.properties", "a") as f:
+                f.write("cpu_launcher_enable=true\n")
+                f.write("cpu_launcher_args=--use_logical_core\n")
+
     def load_config(self):
         report_cmd = None
         for k, v in self.yaml_dict.items():
@@ -100,6 +106,8 @@ def load_config(self):
             else "{}/cpu".format(MODEL_JSON_CONFIG_PATH)
         )
 
+        self.enable_launcher_with_logical_core()
+
         if self.skip_ts_install:
             self.bm_config["version"] = get_torchserve_version()
 

frontend/server/src/main/java/org/pytorch/serve/wlm/WorkerLifeCycle.java

@@ -35,7 +35,6 @@ public class WorkerLifeCycle {
     private Connector connector;
     private ReaderThread errReader;
     private ReaderThread outReader;
-    private String launcherArgs;
     private int numWorker;
     private int currNumRunningWorkers;
 
@@ -50,10 +49,11 @@ public Process getProcess() {
         return process;
     }
 
-    public ArrayList<String> launcherArgsToList() {
+    public ArrayList<String> launcherArgsToList(String launcherArgs) {
         ArrayList<String> arrlist = new ArrayList<String>();
         arrlist.add("-m");
-        arrlist.add("intel_extension_for_pytorch.cpu.launch");
+        arrlist.add("torch.backends.xeon.run_cpu");
+
         if (launcherArgs != null && launcherArgs.length() > 1) {
             String[] argarray = launcherArgs.split(" ");
             for (int i = 0; i < argarray.length; i++) {
@@ -63,22 +63,25 @@ public ArrayList<String> launcherArgsToList() {
         return arrlist;
     }
 
-    public boolean isLauncherAvailable()
+    public boolean isLauncherAvailable(String launcherArgs)
             throws WorkerInitializationException, InterruptedException {
         boolean launcherAvailable = false;
-        try {
-            ArrayList<String> cmd = new ArrayList<String>();
-            cmd.add("python");
-            ArrayList<String> args = launcherArgsToList();
-            cmd.addAll(args);
-            cmd.add("--no_python");
-            // try launching dummy command to check launcher availability
-            String dummyCmd = "hostname";
-            cmd.add(dummyCmd);
-
-            String[] cmdList = new String[cmd.size()];
-            cmdList = cmd.toArray(cmdList);
 
+        ArrayList<String> cmd = new ArrayList<String>();
+        cmd.add("python");
+        ArrayList<String> args = launcherArgsToList(launcherArgs);
+        cmd.addAll(args);
+        cmd.add("--no_python");
+        // try launching dummy command to check launcher availability
+        String dummyCmd = "hostname";
+        cmd.add(dummyCmd);
+
+        String[] cmdList = new String[cmd.size()];
+        cmdList = cmd.toArray(cmdList);
+
+        logger.debug("launcherAvailable cmdline: {}", cmd.toString());
+
+        try {
             Process processLauncher = Runtime.getRuntime().exec(cmdList);
             int ret = processLauncher.waitFor();
             launcherAvailable = (ret == 0);
@@ -115,23 +118,24 @@ public void startWorker(int port, String deviceIds)
         }
 
         if (configManager.isCPULauncherEnabled()) {
-            launcherArgs = configManager.getCPULauncherArgs();
-            boolean launcherAvailable = isLauncherAvailable();
+            String launcherArgs = configManager.getCPULauncherArgs();
+            boolean launcherAvailable = isLauncherAvailable(launcherArgs);
             if (launcherAvailable) {
-                ArrayList<String> args = launcherArgsToList();
+                ArrayList<String> args = launcherArgsToList(launcherArgs);
                 argl.addAll(args);
 
                 // multi-worker core pinning
                 if (this.numWorker > 1) {
                     argl.add("--ninstances");
                     argl.add(String.valueOf(this.numWorker));
-                    argl.add("--instance_idx");
+                    argl.add("--rank");
+                    // instance_idx is 0-indexed
                     argl.add(String.valueOf(this.currNumRunningWorkers));
                 }
 
             } else {
                 logger.warn(
-                        "CPU launcher is enabled but launcher is not available. Proceeding without launcher.");
+                        "torch.backends.xeon.run_cpu is not available. Proceeding without worker core pinning. For better performance, please make sure torch.backends.xeon.run_cpu is available.");
             }
         }
 

ts_scripts/install_dependencies.py

@@ -74,6 +74,9 @@ def install_jmeter(self):
     def install_wget(self):
         pass
 
+    def install_numactl(self):
+        pass
+
 
 class Linux(Common):
     def __init__(self):
@@ -107,6 +110,10 @@ def install_libgit2(self):
         os.system(f"cd libgit2-1.3.0 && cmake . && make && sudo make install && cd ..")
         os.system(f"rm -rf libgit2-1.3.0 && rm libgit2-1.3.0.tar.gz")
 
+    def install_numactl(self):
+        if os.system("numactl --show") != 0 or args.force:
+            os.system(f"{self.sudo_cmd}apt-get install -y numactl")
+
 
 class Windows(Common):
     def __init__(self):
@@ -122,6 +129,9 @@ def install_nodejs(self):
     def install_wget(self):
         pass
 
+    def install_numactl(self):
+        pass
+
 
 class Darwin(Common):
     def __init__(self):
@@ -146,6 +156,10 @@ def install_wget(self):
         if os.system("wget --version") != 0 or args.force:
             os.system("brew install wget")
 
+    def install_numactl(self):
+        if os.system("numactl --show") != 0 or args.force:
+            os.system("brew install numactl")
+
 
 def install_dependencies(cuda_version=None, nightly=False):
     os_map = {"Linux": Linux, "Windows": Windows, "Darwin": Darwin}