Improve mstress local benchmark behavior

benchmarks/mstress/mstress.py

@@ -162,14 +162,26 @@ def PrintMemoryUsage(opts):
             Globals.SERVER_KEYWORD,
         )
 
-    proc = subprocess.Popen(
-        ["ssh", opts.server, psCmd],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-    )
+    if opts.server in ("localhost", "127.0.0.1"):
+        proc = subprocess.Popen(
+            [psCmd],
+            shell=True,
+            executable="/bin/bash",
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+    else:
+        proc = subprocess.Popen(
+            ["ssh", opts.server, psCmd],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
     result = proc.communicate()
     if result and len(result[0].strip()) > 0:
-        print("Memory usage %sKB" % result[0].strip())
+        memory = result[0].strip()
+        if not isinstance(memory, str):
+            memory = memory.decode("utf-8", "replace")
+        print("Memory usage %sKB" % memory)
     else:
         print("Memory usage <unknown> KB")
 
@@ -189,7 +201,7 @@ def RunMStressMaster(opts, hostsList):
     # print 'Master: called with %r, %r' % (opts, hostsList)
 
     startTime = datetime.datetime.now()
-    if RunMStressMasterTest(opts, hostsList, "create"):
+    if not RunMStressMasterTest(opts, hostsList, "create"):
         return False
     deltaTime = datetime.datetime.now() - startTime
     print(
@@ -200,7 +212,7 @@ def RunMStressMaster(opts, hostsList):
     print("==========================================")
 
     startTime = datetime.datetime.now()
-    if RunMStressMasterTest(opts, hostsList, "stat"):
+    if not RunMStressMasterTest(opts, hostsList, "stat"):
         return False
     deltaTime = datetime.datetime.now() - startTime
     print(
@@ -210,7 +222,7 @@ def RunMStressMaster(opts, hostsList):
     print("==========================================")
 
     startTime = datetime.datetime.now()
-    if RunMStressMasterTest(opts, hostsList, "readdir"):
+    if not RunMStressMasterTest(opts, hostsList, "readdir"):
         return False
     deltaTime = datetime.datetime.now() - startTime
     print(
@@ -221,10 +233,10 @@ def RunMStressMaster(opts, hostsList):
 
     if opts.leave_files:
         print("\nNot deleting files because of -l option")
-        return False
+        return True
 
     startTime = datetime.datetime.now()
-    if RunMStressMasterTest(opts, hostsList, "delete"):
+    if not RunMStressMasterTest(opts, hostsList, "delete"):
         return False
     deltaTime = datetime.datetime.now() - startTime
     print(
@@ -273,16 +285,30 @@ def RunMStressMasterTest(opts, hostsList, test):
             + opts.filesystem
             + ".slave.log"
         )
-        p = subprocess.Popen(
-            [
-                "/usr/bin/ssh",
-                client,
-                "%s -c %s -k %s >& %s"
-                % (ssh_cmd, client, clientHostMapping[client], slaveLogfile),
-            ],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
+        slave_cmd = "%s -c %s -k %s >& %s" % (
+            ssh_cmd,
+            client,
+            clientHostMapping[client],
+            slaveLogfile,
         )
+        if client in ("localhost", "127.0.0.1"):
+            p = subprocess.Popen(
+                [slave_cmd],
+                shell=True,
+                executable="/bin/bash",
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+        else:
+            p = subprocess.Popen(
+                [
+                    "/usr/bin/ssh",
+                    client,
+                    slave_cmd,
+                ],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
         running_procs[p] = client
 
     success = True
@@ -323,7 +349,7 @@ def RunMStressMasterTest(opts, hostsList, test):
             else:
                 sys.stdout.write(".")
             sys.stdout.flush()
-            time.sleep(0.5)
+            time.sleep(0.05)
     return success
 
 
@@ -439,7 +465,7 @@ def RunMStressSlave(opts, clientsPerHost):
             else:
                 sys.stdout.write(".")
             sys.stdout.flush()
-            time.sleep(0.5)
+            time.sleep(0.05)
     return success
 
 

benchmarks/mstress/mstress_client.cc

@@ -35,8 +35,8 @@
 #include <sstream>
 #include <string>
 #include <vector>
-#include <queue>
 #include <algorithm>
+#include <deque>
 
 #if __cplusplus >= 201103L
 #include <random>
@@ -185,11 +185,9 @@ void hexout(char* str, int len) {
   printf("\n");
 }
 
-void myitoa(int n, char* buf)
+void myitoa(int n, char* buf, size_t len = 32)
 {
-  static char result[32];
-  snprintf(result, 32, "%d", n);
-  strcpy(buf, result);
+  snprintf(buf, len, "%d", n);
 }
 
 //Return a random permutation of numbers in [0..range).
@@ -343,7 +341,7 @@ int CreateDFSPaths(Client* client, AutoCleanupKfsClient* kfs, int level, int* cr
   char name[512];
   strncpy(name, client->prefix_.c_str(), sizeof(name) / sizeof(name[0]) - 1);
   for (int i = 0; i < client->inodesPerLevel_; i++) {
-    myitoa(i, name + client->prefixLen_);
+    myitoa(i, name + client->prefixLen_, sizeof(name) - client->prefixLen_);
     client->path_.Push(name);
     //hexout(client->path_.actualPath_, client->path_.len_ + 3);
 
@@ -433,16 +431,16 @@ int StatDFSPaths(Client* client, AutoCleanupKfsClient* kfs) {
 
     for (int d = 0; d < client->levels_; d++) {
       int randIdx = rand() % client->inodesPerLevel_;
-      myitoa(randIdx, name + client->prefixLen_);
+      myitoa(randIdx, name + client->prefixLen_, sizeof(name) - client->prefixLen_);
       client->path_.Push(name);
       //fprintf(logFile, "Stat: path now is %s\n", client->path_.actualPath_);
     }
     //fprintf(logFile, "Stat: doing stat on [%s]\n", client->path_.actualPath_);
 
     KFS::KfsFileAttr attr;
-    int err = kfsClient->Stat(os.str().c_str(), attr);
+    int err = kfsClient->Stat(client->path_.String(), attr);
     if (err) {
-      fprintf(logFile, "error doing stat on %s\n", os.str().c_str());
+      fprintf(logFile, "error doing stat on %s\n", client->path_.String());
       return err;
     }
 
@@ -466,14 +464,15 @@ int ListDFSPaths(Client* client, AutoCleanupKfsClient* kfs) {
   gettimeofday(&tvAlpha, NULL);
   int inodeCount = 0;
 
-  queue<string> pending;
+  deque<string> pending;
   ostringstream os;
   os << TEST_BASE_DIR << "/" << client->hostName_ + "_" << client->processName_;
-  pending.push(os.str());
+  pending.push_back(os.str());
 
   while (!pending.empty()) {
-    string parent = pending.front();
-    pending.pop();
+    string parent;
+    parent.swap(pending.front());
+    pending.pop_front();
     //fprintf(logFile, "readdir on parent [%s]\n", parent.c_str());
     vector<KFS::KfsFileAttr> children;
     int err = kfsClient->ReaddirPlus(parent.c_str(), children);
@@ -482,20 +481,19 @@ int ListDFSPaths(Client* client, AutoCleanupKfsClient* kfs) {
       return err;
     }
     while (!children.empty()) {
-      string child = children.back().filename;
-      bool isDir = children.back().isDirectory;
-      children.pop_back();
+      const KFS::KfsFileAttr& childAttr = children.back();
+      const string& child = childAttr.filename;
+      bool isDir = childAttr.isDirectory;
       //fprintf(logFile, "  Child = %s inodeCount=%d\n", child.c_str(), inodeCount);
-      if (child == "." ||
-          child == "..") {
-        continue;
-      }
-      inodeCount ++;
-      if (isDir) {
-        string nextParent = parent + "/" + child;
-        pending.push(nextParent);
-        //fprintf(logFile, "  Adding next parent [%s]\n", nextParent.c_str());
+      if (child != "." && child != "..") {
+        inodeCount ++;
+        if (isDir) {
+          string nextParent = parent + "/" + child;
+          pending.push_back(nextParent);
+          //fprintf(logFile, "  Adding next parent [%s]\n", nextParent.c_str());
+        }
       }
+      children.pop_back();
       if (inodeCount > 0 && inodeCount % COUNT_INCR == 0) {
         fprintf(logFile, "Readdir paths so far: %d\n", inodeCount);
       }
@@ -546,7 +544,7 @@ int RemoveDFSPaths(Client* client, AutoCleanupKfsClient* kfs) {
     while (lev < client->levels_) {
       pos = idx / client->inodesPerLevel_;
       delta = idx - (pos * client->inodesPerLevel_);
-      myitoa(delta, sfx);
+      myitoa(delta, sfx, sizeof(sfx));
       if (pathSoFar.length()) {
         pathSoFar = client->prefix_ + sfx + "/" + pathSoFar;
       } else {

benchmarks/mstress/mstress_plan.py

@@ -183,6 +183,9 @@ def main():
     print("==> Created planfile: %s" % opts.output_file)
     print("copying file %s to all client hosts" % opts.output_file)
     for client in hostlist:
+        if client in ("localhost", "127.0.0.1"):
+            print("available %s on %s" % (opts.output_file, client))
+            continue
         p = subprocess.Popen(
             [
                 "/usr/bin/scp",
@@ -195,7 +198,9 @@ def main():
             if ret is None:
                 time.sleep(0.5)
             else:
-                print("transfered %s to %s" % (opts.output_file, client))
+                if ret != 0:
+                    sys.exit("failed to transfer %s to %s" % (opts.output_file, client))
+                print("transferred %s to %s" % (opts.output_file, client))
                 break
 
 

benchmarks/mstress/mstress_run.py

@@ -45,7 +45,7 @@ def NumFiles2Stat():
             Params.INODES_PER_LEVEL**Params.PATH_LEVELS
             * Params.CLIENTS_PER_HOST
             * len(Params.CLIENT_HOSTS.split(","))
-            / 2
+            // 2
         )
 
     NumFiles2Stat = staticmethod(NumFiles2Stat)
@@ -136,33 +136,30 @@ def Execute(type, args):
         % type
     )
 
-    result = ""
+    result = []
     proc = subprocess.Popen(
-        args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
+        args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+        universal_newlines=True
     )
-    while proc.poll() is None:
-        output = proc.stdout.read(1)
-        result += output
+    for output in iter(proc.stdout.readline, ""):
+        result.append(output)
         sys.stdout.write(output)
         sys.stdout.flush()
-
-    output = proc.stdout.read()
-    result += output
-    sys.stdout.write(output)
-    sys.stdout.flush()
     proc.wait()
 
-    return result
+    return "".join(result)
 
 
 def PrintResult(type, result):
     PrintMsg("\nBenchmark results for '%s':" % type)
     for m in re.findall(r"(\w+) test took (\S+) sec", result):
         PrintMsg("%-10s: %s sec" % (m[0], m[1]))
-    PrintMsg(
-        "\n%s\n=========================================="
-        % re.search(r"Memory usage .*$", result, re.MULTILINE).group(0)
-    )
+    memory = re.search(r"Memory usage .*", result, re.MULTILINE)
+    if memory:
+        PrintMsg(
+            "\n%s\n=========================================="
+            % memory.group(0)
+        )
 
 
 def ParseArgs():