Fixes for zmq_worker_client benchmark app

	Fixed Makefile to link to margo
	Updated config script to include monitoring and the multi-executable launch options
	Improved run script

Author: giltirn

benchmark_suite/zmq_worker_client/Makefile

@@ -5,9 +5,9 @@ INSTALL := ${INSTALL}../
 all: server client
 
 server: server.cpp
-	tau_cxx.sh -std=c++17 -O3 -g -D_USE_ZMQNET -D_PERF_METRIC -I${INSTALL}/include -I${INSTALL}/include/chimbuko/3rdparty -L${INSTALL}/lib server.cpp -o server -lchimbuko  -lstdc++fs
+	tau_cxx.sh -std=c++17 -O3 -g -D_USE_ZMQNET -D_PERF_METRIC -I${INSTALL}/include -I${INSTALL}/include/chimbuko/3rdparty -L${INSTALL}/lib server.cpp -o server -lchimbuko  -lstdc++fs -lmargo
 client: client.cpp
-	tau_cxx.sh -std=c++17 -O3 -g -D_USE_ZMQNET -D_PERF_METRIC -I${INSTALL}/include -I${INSTALL}/include/chimbuko/3rdparty -L${INSTALL}/lib client.cpp -o client -lchimbuko  -lstdc++fs
+	tau_cxx.sh -std=c++17 -O3 -g -D_USE_ZMQNET -D_PERF_METRIC -I${INSTALL}/include -I${INSTALL}/include/chimbuko/3rdparty -L${INSTALL}/lib client.cpp -o client -lchimbuko  -lstdc++fs -lmargo
 
 clean:
 	rm -f client server

benchmark_suite/zmq_worker_client/chimbuko_config.sh

@@ -8,7 +8,7 @@ service_node_iface=eth0 #network interface upon which communication to the servi
 ####################################
 #Options for visualization module
 ####################################
-use_viz=1 #enable or disable the visualization
+use_viz=0 #enable or disable the visualization
 viz_root=/opt/chimbuko/viz    #the root directory of the visualization module  <------------ ***SET ME (if using viz)***
 viz_worker_port=6379  #the port on which to run the redis server for the visualization backend
 viz_port=5002 #the port on which to run the webserver
@@ -28,7 +28,7 @@ provdb_extra_args="" #any extra command line arguments to pass
 provdb_nshards=4  #number of database shards
 provdb_ninstances=1 #number of database server instances. Shards are distributed over instances
 provdb_engine="ofi+tcp;ofi_rxm"  #the OFI libfabric provider used for the Mochi stack
-provdb_port=5000 #the port of the provenance database
+provdb_port=5000 #the port of the provenance database. For >1 instance the port of instance i will be provdb_port+i 
 provdb_writedir=chimbuko/provdb #the directory in which the provenance database is written. Chimbuko creates chimbuko/provdb which can be used as a default
 provdb_commit_freq=10000   #frequency ms at which the provenance database is committed to disk. If set to 0 it will commit only at the end
 
@@ -51,33 +51,35 @@ pserver_nt=2 #number of worker threads
 ####################################
 ad_extra_args="-perf_outputpath chimbuko/logs -perf_step 1" #any extra command line arguments to pass. Note: chimbuko/logs is automatically created by services script
 ad_win_size=5  #number of events around an anomaly to store; provDB entry size is proportional to this so keep it small!
-ad_alg="sstd" #the anomaly detection algorithm. Valid values are "hbos" and "sstd"
+ad_alg="hbos" #the anomaly detection algorithm. Valid values are "hbos" and "sstd"
 ad_outlier_hbos_threshold=0.99  #the percentile of events outside of which are considered anomalies by the HBOS algorithm
 ad_outlier_sstd_sigma=12 #number of standard deviations that defines an outlier in the SSTD algorithm
 ####################################
 #Options for TAU
-#Note: Only the TAU_ADIOS2_PATH, TAU_ADIOS2_FILE_PREFIX, EXE_NAME and TAU_ADIOS2_ENGINE variables are used by the Chimbuko services script and there only to generate the suggested
+#Note: Only the TAU_ADIOS2_PATH, TAU_ADIOS2_FILE_PREFIX, EXE_NAME, TAU_ADIOS2_ENGINE and tau_monitoring_conf variables are used by the Chimbuko services script and there only to generate the suggested
 #      command to launch the AD (output to chimbuko/vars/chimbuko_ad_cmdline.var); they can be overridden by the run script if desired providing the appropriate modifications
 #      are made to the AD launch command. The remainder of the variables are used only by TAU and can be freely overridden.
 ####################################
-export TAU_ADIOS2_ENGINE=SST    #online communication engine (alternative BP4 although this goes through the disk system and may be slower unless the BPfiles are stored on a burst disk)
+export TAU_ADIOS2_ENGINE=BP4    #online communication engine (alternative BP4 although this goes through the disk system and may be slower unless the BPfiles are stored on a burst disk)
 export TAU_ADIOS2_ONE_FILE=FALSE  #a different connection file for each rank
 export TAU_ADIOS2_PERIODIC=1   #enable/disable ADIOS2 periodic output
 export TAU_ADIOS2_PERIOD=1000000  #period in us between ADIOS2 io steps
 export TAU_THREAD_PER_GPU_STREAM=1  #force GPU streams to appear as different TAU virtual threads
 export TAU_THROTTLE=0  #enable/disable throttling of short-running functions
 
 export TAU_MAKEFILE=/opt/tau2/x86_64/lib/Makefile.tau-papi-pthread-python-pdt-adios2   #non-MPI TAU required!
+tau_monitoring_conf="default" #Provide a configuration file for the TAU monitoring plugin. It will be copied to the work directory as "tau_monitoring.json" (unless it is already there!). If set to default, Chimbuko will generate one automatically
 
 #Note: the following 2 variables are not used by the service script but are included here for use from the user's run script allowing the application to be launched with either "${TAU_EXEC} <app>" or "${TAU_PYTHON} <app>"
 #Note: the "binding" -T ... is used by Tau to find the appropriate configuration. It can typically be inferred from the name of the Makefile. If using a non-MPI job the 'mpi' should be changed to 'serial' and a non-MPI build of
 #      ADIOS2/TAU must exist
 #Suggestion: It is useful to test the command without Chimbuko first to ensure TAU picks up the correct binding; this can be done by 'export TAU_ADIOS2_ENGINE=BPFile' and then running the application with Tau but without Chimbuko.
-TAU_EXEC="tau_exec -T serial,papi,pthread,python,pdt,adios2 -adios2_trace"   #how to execute tau_exec; the -T arguments should mirror the makefile name  <------------ ***SET ME***
+#TAU_EXEC="tau_exec -T serial,papi,pthread,python,pdt,adios2 -adios2_trace -monitoring"   #how to execute tau_exec; the -T arguments should mirror the makefile name  <------------ ***SET ME***
+TAU_EXEC="tau_exec -T serial,papi,pthread,python,pdt,adios2 -adios2_trace -monitoring"   #how to execute tau_exec; the -T arguments should mirror the makefile name  <------------ ***SET ME***
 TAU_PYTHON="tau_python -T papi,mpi,pthread,pdt,adios2 -tau-python-interpreter=python3 -adios2_trace  -tau-python-args=-u"  #how to execute tau_python. Note that passing -u to python forces it to not buffer stdout so we can pipe it
                                                                                                                            #to tee in realtime <--- SET ME (if !python3)
 
-export EXE_NAME=main  #the name of the executable (without path) <------------ ***SET ME***
+export EXE_NAME=(server client)  #the name of the executable (without path). For multi-component workflows this argument also accepts a list, e.g. (main1 main2)  <------------ ***SET ME***
 
 TAU_ADIOS2_PATH=chimbuko/adios2  #path where the adios2 files are to be stored. Chimbuko services creates the directory chimbuko/adios2 in the working directory and this should be used by default
 TAU_ADIOS2_FILE_PREFIX=tau-metrics  #the prefix of tau adios2 files; full filename is ${TAU_ADIOS2_PREFIX}-${EXE_NAME}-${RANK}.bp

benchmark_suite/zmq_worker_client/client.cpp

@@ -23,30 +23,32 @@ void rpcWaitAndRespond(ADThreadNetClient &client, int wait_ms){
   client.send_and_receive(recv,send); //ignore return
 }
 
-int main(int argc, char** argv){ 
-  assert(argc >= 6);
+int main(int argc, char** argv){
+  {
+    assert(argc >= 6);
 
-  std::string server_addr = argv[1];
-  int cycles = std::stoi(argv[2]);
-  int cycle_time = std::stoi(argv[3]); //ms
-  int anom_freq =  std::stoi(argv[4]);
-  int anom_mult = std::stoi(argv[5]); //time multiplier for anomalies
-  Log << "Client executing with parameters:  cycles=" << cycles << " cycle_time=" << cycle_time << "ms anom_freq=" << anom_freq << " anom_mult=" << anom_mult << std::endl;
-
-  Log << "connecting to server with address " << server_addr << std::endl;
-  ADThreadNetClient client;
-  client.connect_ps(0,0,server_addr);
-
-  for(int i=0;i<cycles;i++){
-    int ctime = cycle_time;
-    if(i>0 && i % anom_freq == 0) ctime *= anom_mult;
-
-    //Client waits for same time as server
-    clientWait(ctime);
-
-    Log << "calling RPC" << std::endl;
-    rpcWaitAndRespond(client, ctime);
+    std::string server_addr = argv[1];
+    int cycles = std::stoi(argv[2]);
+    int cycle_time = std::stoi(argv[3]); //ms
+    int anom_freq =  std::stoi(argv[4]);
+    int anom_mult = std::stoi(argv[5]); //time multiplier for anomalies
+    Log << "executing with parameters:  cycles=" << cycles << " cycle_time=" << cycle_time << "ms anom_freq=" << anom_freq << " anom_mult=" << anom_mult << std::endl;
+
+    Log << "connecting to server with address " << server_addr << std::endl;
+    ADThreadNetClient client;
+    client.connect_ps(0,0,server_addr);
+
+    for(int i=0;i<cycles;i++){
+      int ctime = cycle_time;
+      if(i>0 && i % anom_freq == 0) ctime *= anom_mult;
+
+      //Client waits for same time as server
+      clientWait(ctime);
+
+      Log << "calling RPC" << std::endl;
+      rpcWaitAndRespond(client, ctime);
+    }
   }
-
+  Log << "finished" << std::endl; 
   return 0;
 }

benchmark_suite/zmq_worker_client/run.sh

@@ -6,25 +6,36 @@ rm -rf chimbuko
 export CHIMBUKO_CONFIG=chimbuko_config.sh
 source ${CHIMBUKO_CONFIG}
 
-if (( 1 )); then
+log_dir=chimbuko/logs
+
+use_chimbuko=1
+use_tau=1
+
+#Launch Chimbuko services
+if (( ${use_chimbuko} == 1 )); then
     echo "Running services"
     ${chimbuko_services} 2>&1 | tee services.log &
     echo "Waiting"
-    while [ ! -f chimbuko/vars/chimbuko_ad_cmdline.var ]; do sleep 1; done
+    while [ ! -f chimbuko/vars/chimbuko_ad_opts.var ]; do sleep 1; done
     ad_opts=$(cat chimbuko/vars/chimbuko_ad_opts.var)
 else
-    mkdir -p chimbuko/logs chimbuko/adios2   
+    mkdir -p chimbuko/logs chimbuko/adios2
 fi
 
-log_dir=chimbuko/logs
+if (( ${use_tau} == 0 )); then
+    TAU_EXEC=
+fi
 
-if (( 1 )); then
-    ad_run_server="driver ${TAU_ADIOS2_ENGINE} ${TAU_ADIOS2_PATH} ${TAU_ADIOS2_FILE_PREFIX}-server -program_idx 0 ${ad_opts} 2>&1 | tee ${log_dir}/ad_server.log"
+#Launch AD on server
+if (( ${use_chimbuko} == 1 )); then
+    ad_run_server=$(cat chimbuko/vars/chimbuko_ad_cmdline.server.var)
     echo "Running AD for server with command:"
     echo ${ad_run_server}
     eval "${ad_run_server} &"
 fi
 
+#Launch server
+export TAU_VERBOSE=0
 ${TAU_EXEC} ./server 9876  2>&1 | tee chimbuko/logs/server.log &
 spid=$!
 
@@ -38,12 +49,15 @@ cycle_time=50 #ms
 anom_freq=30 
 anom_mult=20
 
-if (( 1 )); then
-    ad_run_client="driver ${TAU_ADIOS2_ENGINE} ${TAU_ADIOS2_PATH} ${TAU_ADIOS2_FILE_PREFIX}-client -program_idx 1 ${ad_opts} 2>&1 | tee ${log_dir}/ad_client.log"
+#Launch AD on client
+if (( ${use_chimbuko} == 1 )); then
+    ad_run_client=$(cat chimbuko/vars/chimbuko_ad_cmdline.client.var)
     echo "Running AD for client"
     eval "${ad_run_client} &"
 fi
 
-export TAU_VERBOSE=1
-
+#Launch client
+export TAU_VERBOSE=0
 ${TAU_EXEC} ./client ${ip_port} ${cycles} ${cycle_time} ${anom_freq} ${anom_mult} 2>&1 | tee chimbuko/logs/client.log
+
+wait

benchmark_suite/zmq_worker_client/server.cpp

@@ -38,29 +38,31 @@ class NetPayloadBounce: public NetPayloadBase{
 
 
 int main (int argc, char ** argv){
-  if(argc != 2){ std::cout << "Usage: <binary> <port>" << std::endl; return 0; }
+  {
+    if(argc != 2){ std::cout << "Usage: <binary> <port>" << std::endl; return 0; }
 
-  int port = std::stoi(argv[1]);
-  int threads = 4;
+    int port = std::stoi(argv[1]);
+    int threads = 4;
 
-  ZMQNet net;
-  net.setPort(port);
-  net.setAutoShutdown(true);
+    ZMQNet net;
+    net.setPort(port);
+    net.setAutoShutdown(true);
 
-  Log << "run parameter server on port " << port << std::endl;
+    Log << "run parameter server on port " << port << std::endl;
 
-  net.add_payload(new NetPayloadBounce);
-  net.init(nullptr, nullptr, threads);
+    for(int t=0;t<threads;t++)
+      net.add_payload(new NetPayloadBounce,t);
+    net.init(nullptr, nullptr, threads);
 
-  signal(SIGTERM, termSignalHandler);
+    signal(SIGTERM, termSignalHandler);
 
-  net.run();
+    net.run();
 
-  signal(SIGTERM, SIG_DFL); //restore default signal handling
-
-  Log << "shutdown parameter server ..." << std::endl;
-  net.finalize();
+    signal(SIGTERM, SIG_DFL); //restore default signal handling
 
+    Log << "shutdown parameter server ..." << std::endl;
+    net.finalize();
+  }
   Log << "finished" << std::endl;
   return 0;
 }