Merge pull request #435 from perftool-incubator/dev-kmr

Dev kmr
perftool-incubator · Dec 8, 2023 · df0be05 · df0be05
2 parents 27a131d + b4e60b6
commit df0be05
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 10 deletions.
diff --git a/rickshaw-post-process-bench b/rickshaw-post-process-bench
@@ -209,6 +209,21 @@ foreach my $benchmark_and_id ( split(/,/, $run{'bench-ids'}) ) {
     }
 }
 
+# determine the max number of forked jobs
+my $max_forked_jobs = 0;
+open(PROCCPUINFO, "<", "/proc/cpuinfo") || die("[ERROR] Could not open /proc/cpuinfo for reading\n");
+while(<PROCCPUINFO>) {
+    if ($_ =~ /^processor/) {
+        $max_forked_jobs++;
+    }
+}
+close(PROCCPUINFO);
+
+if ($max_forked_jobs > 0) {
+    printf "Will fork a maximum of %d jobs at a time\n", $max_forked_jobs;
+} else {
+    die("[ERROR] Could not determine the maximum number of jobs to fork at a time\n");
+}
 
 printf "Launching a post-process job for each iteration x sample x [client|server] for %s\n", $run{'benchmark'};
 for (my $i = 1; $i <= scalar @{ $run{'iterations'} }; $i++) {
@@ -229,7 +244,7 @@ for (my $i = 1; $i <= scalar @{ $run{'iterations'} }; $i++) {
         my @samples;
         my $primary_metric;
         my $primary_period;
-            for my $samp_dir (@samp_dirs) {
+        for my $samp_dir (@samp_dirs) {
             my @pids;
             my %sample; # sample data from all clients/servers
             my @cons_periods; # consolidated periods across clients/servers get merged here
@@ -252,6 +267,12 @@ for (my $i = 1; $i <= scalar @{ $run{'iterations'} }; $i++) {
                             my $iter_params = dump_params($run{'iterations'}[$i - 1]{'params'}, $cs_id, $cs_name);
                             my $cs_id_dir = $cs_name_dir . "/" . $cs_id;
                             if (-d $run_dir . "/" . $cs_id_dir) {
+                                if (scalar @pids >= $max_forked_jobs) {
+                                    printf "Waiting for %d post-processing jobs to complete before starting more\n", scalar @pids;
+                                    while(wait() > -1) {}
+                                    printf "Starting more jobs...\n";
+                                    @pids = ();
+                                }
                                 if (my $pid = fork) {
                                     push(@pids, $pid);
                                 } else {
@@ -262,11 +283,11 @@ for (my $i = 1; $i <= scalar @{ $run{'iterations'} }; $i++) {
                                     if ($rc == -1) {
                                         printf "Failed to execute '%s%s'!\n", $pp_cmd, $iter_params;
                                     } elsif ($rc & 127) {
-                                        printf "'%s%s' died with signal %d, %s coredump!\n", $pp_cmd, $iter_params, ($rc & 127), ($rc & 128) ? 'with' : 'without';
+                                        printf "'%s %s' died with signal %d, %s coredump!\n", $pp_cmd, $iter_params, ($rc & 127), ($rc & 128) ? 'with' : 'without';
                                     } else {
                                         $rc = $rc >> 8;
                                         if ($rc != 0) {
-                                        printf "'%s%s' exited with non-zero value %d\n", $pp_cmd, $iter_params, $rc;
+                                        printf "'%s %s' exited with non-zero value %d\n", $pp_cmd, $iter_params, $rc;
                                         }
                                     }
                                     exit;
@@ -276,8 +297,10 @@ for (my $i = 1; $i <= scalar @{ $run{'iterations'} }; $i++) {
                     }
                 }
             }
-            printf "Waiting for %d post-processing jobs to complete\n", scalar @pids;
-            while (wait() > -1) {}
+            if (wait() > -1) {
+                printf "Waiting for %d post-processing jobs to complete\n", scalar @pids;
+                while (wait() > -1) {}
+            }
             print "Post-processing complete\n";
         }
     }

diff --git a/rickshaw-run b/rickshaw-run
@@ -90,6 +90,7 @@ my $iterations_dir;
 my $rickshaw_project_dir;
 my $roadblock_msgs_dir;
 my $roadblock_logs_dir;
+my $roadblock_followers_dir;
 my $endpoint_roadblock_opt = "";
 my $workshop_roadblock_opt = "";
 my %utility_configs;
@@ -189,12 +190,12 @@ sub do_roadblock {
     my $timeout = shift;
     # $_[0] is for a reference to the messages data structure
 
-    my $follower;
-    my $follower_param = "";
+    my $rb_followers_file = $roadblock_followers_dir . "/" . $label . ".txt";
+    open(RB_FOLLOWERS, ">", $rb_followers_file) || die("[ERROR] Could not open the roadblock followers file for writing [" . $rb_followers_file . "]!\n");
     for (my $i=1; $i<scalar(@_); $i++) {
-        $follower = $_[$i];
-        $follower_param .= " --followers=" . $follower;
+        printf RB_FOLLOWERS "%s\n", $_[$i];
     }
+    close(RB_FOLLOWERS);
 
     my $attempts = 0;
     my $rc = 99;
@@ -219,7 +220,7 @@ sub do_roadblock {
             " --uuid=" . $this_uuid .
             " --timeout=" . $timeout .
             " --message-log=" . $msgs_log_file .
-            $follower_param;
+            " --followers-file=" . $rb_followers_file;
         if ($abort_via_roadblock) {
             $cmd .= " --abort";
         }
@@ -1489,6 +1490,8 @@ sub make_run_dirs() {
     mkdir($roadblock_msgs_dir);
     $roadblock_logs_dir = $run_dir . "/roadblock-logs";
     mkdir($roadblock_logs_dir);
+    $roadblock_followers_dir = $run_dir . "/roadblock-followers";
+    mkdir($roadblock_followers_dir);
     # If there are no endpoints, assume 1 endpoint using the 'local' extension
     if (scalar @endpoints == 0) {
         printf "ERROR: you must declare endpoints\n";