JANA/CLI/JBenchmarker.cc

0001
0002 // Copyright 2020, Jefferson Science Associates, LLC.
0003 // Subject to the terms in the LICENSE file found in the top-level directory.
0004
0005 #include "JBenchmarker.h"
0006
0007 #include <JANA/Utils/JCpuInfo.h>
0008
0009 #include <fstream>
0010 #include <cmath>
0011 #include <sys/stat.h>
0012 #include <iostream>
0013
0014 JBenchmarker::JBenchmarker(JApplication* app) : m_app(app) {
0015
0016     m_max_threads = JCpuInfo::GetNumCpus();
0017
0018     auto params = app->GetJParameterManager();
0019
0020     m_logger = params->GetLogger("JBenchmarker");
0021
0022     params->SetParameter("jana:nevents", 0);
0023     // Prevent users' choice of nevents from interfering with everything
0024
0025     params->SetDefaultParameter(
0026             "benchmark:nsamples",
0027             m_nsamples,
0028             "Number of samples for each benchmark test");
0029
0030     params->SetDefaultParameter(
0031             "benchmark:minthreads",
0032             m_min_threads,
0033             "Minimum number of threads for benchmark test");
0034
0035     params->SetDefaultParameter(
0036             "benchmark:maxthreads",
0037             m_max_threads,
0038             "Maximum number of threads for benchmark test");
0039
0040     params->SetDefaultParameter(
0041             "benchmark:threadstep",
0042             m_thread_step,
0043             "Delta number of threads between each benchmark test");
0044
0045     params->SetDefaultParameter(
0046             "benchmark:resultsdir",
0047             m_output_dir,
0048             "Output directory name for benchmark test results");
0049
0050     params->SetDefaultParameter(
0051             "benchmark:copyscript",
0052             m_copy_script,
0053             "Copy plotting script to results dir");
0054
0055     params->SetParameter("nthreads", m_max_threads);
0056     // Otherwise JApplication::Scale() doesn't scale up. This is an interesting bug. TODO: Remove me when fixed.
0057 }
0058
0059
0060 JBenchmarker::~JBenchmarker() {}
0061
0062
0063 void JBenchmarker::RunUntilFinished() {
0064
0065     LOG_INFO(m_logger) << "Running benchmarker with the following settings:\n"
0066                        << "    benchmark:minthreads = " << m_min_threads << "\n"
0067                        << "    benchmark:maxthreads = " << m_max_threads << "\n"
0068                        << "    benchmark:threadstep = " << m_thread_step << "\n"
0069                        << "    benchmark:nsamples = " << m_nsamples << "\n"
0070                        << "    benchmark:resultsdir = " << m_output_dir << LOG_END;
0071
0072     m_app->SetTicker(false);
0073     m_app->Run(false);
0074
0075     // Wait for events to start flowing indicating the source is primed
0076     for (int i = 0; i < 5; i++) {
0077         LOG_INFO(m_logger) << "Waiting for event source to start producing ... rate: " << m_app->GetInstantaneousRate() << LOG_END;
0078         std::this_thread::sleep_for(std::chrono::milliseconds(1000));
0079         auto rate = m_app->GetInstantaneousRate();
0080         if (rate > 10.0) {
0081             LOG_INFO(m_logger) << "Rate: " << rate << "Hz   -  ready to begin test" << LOG_END;
0082             break;
0083         }
0084     }
0085     // Loop over all thread settings in set
0086     std::map<uint32_t, std::vector<double> > samples;
0087     std::map<uint32_t, std::pair<double, double> > rates; // key=nthreads  val.first=rate in Hz, val.second=rms of rate in Hz
0088     for (uint32_t nthreads = m_min_threads; nthreads <= m_max_threads && !m_app->IsQuitting(); nthreads += m_thread_step) {
0089
0090         LOG_INFO(m_logger) << "Setting nthreads = " << nthreads << " ..." << LOG_END;
0091         m_app->Scale(nthreads);
0092
0093         // Loop for at most 60 seconds waiting for the number of threads to update
0094         for (int i = 0; i < 60; i++) {
0095             std::this_thread::sleep_for(std::chrono::milliseconds(1000));
0096             if (m_app->GetNThreads() == nthreads) break;
0097         }
0098
0099         // Acquire mNsamples instantaneous rate measurements. The
0100         // GetInstantaneousRate method will only update every 0.5
0101         // seconds so we just wait for 1 second between samples to
0102         // ensure independent measurements.
0103         double sum = 0;
0104         double sum2 = 0;
0105         for (uint32_t isample = 0; isample < m_nsamples && !m_app->IsQuitting(); isample++) {
0106             std::this_thread::sleep_for(std::chrono::milliseconds(1000));
0107             auto rate = m_app->GetInstantaneousRate();
0108             samples[nthreads].push_back(rate);
0109
0110             sum += rate;
0111             sum2 += rate * rate;
0112             double N = (double) (isample + 1);
0113             double avg = sum / N;
0114             double rms = sqrt((sum2 + N * avg * avg - 2.0 * avg * sum) / N);
0115             rates[nthreads].first = avg;  // overwrite with updated value after each sample
0116             rates[nthreads].second = rms;  // overwrite with updated value after each sample
0117
0118             LOG_INFO(m_logger) << "nthreads=" << m_app->GetNThreads() << "  rate=" << rate << "Hz"
0119                        << "  (avg = " << avg << " +/- " << rms / sqrt(N) << " Hz)" << LOG_END;
0120         }
0121     }
0122
0123     // Write results to files
0124     LOG_INFO(m_logger) << "Writing test results to: " << m_output_dir << LOG_END;
0125     mkdir(m_output_dir.c_str(), S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
0126
0127     std::ofstream ofs1(m_output_dir + "/samples.dat");
0128     ofs1 << "# nthreads     rate" << std::endl;
0129     for (auto p : samples) {
0130         auto nthreads = p.first;
0131         for (auto rate: p.second)
0132             ofs1 << std::setw(7) << nthreads << " " << std::setw(12) << std::setprecision(1) << std::fixed << rate
0133                  << std::endl;
0134     }
0135     ofs1.close();
0136
0137     std::ofstream ofs2(m_output_dir + "/rates.dat");
0138     ofs2 << "# nthreads  avg_rate       rms" << std::endl;
0139     for (auto p : rates) {
0140         auto nthreads = p.first;
0141         auto avg_rate = p.second.first;
0142         auto rms = p.second.second;
0143         ofs2 << std::setw(7) << nthreads << " ";
0144         ofs2 << std::setw(12) << std::setprecision(1) << std::fixed << avg_rate << " ";
0145         ofs2 << std::setw(10) << std::setprecision(1) << std::fixed << rms << std::endl;
0146     }
0147     ofs2.close();
0148
0149     if (m_copy_script) {
0150         copy_to_output_dir("${JANA_HOME}/bin/jana-plot-scaletest.py");
0151         LOG_INFO(m_logger)
0152             << "Testing finished. To view a plot of test results:\n"
0153             << "    cd " << m_output_dir
0154             << "\n    ./jana-plot-scaletest.py\n" << LOG_END;
0155     }
0156     else {
0157         LOG_INFO(m_logger)
0158             << "Testing finished. To view a plot of test results:\n"
0159             << "    cd " << m_output_dir << "\n"
0160             << "    $JANA_HOME/bin/jana-plot-scaletest.py\n" << LOG_END;
0161     }
0162     m_app->Stop(true);
0163 }
0164
0165
0166 void JBenchmarker::copy_to_output_dir(std::string filename) {
0167
0168     // Substitute environment variables in given filename
0169     std::string new_fname = filename;
0170     while (auto pos_start = new_fname.find("${") != new_fname.npos) {
0171         auto pos_end = new_fname.find("}", pos_start + 3);
0172         if (pos_end != new_fname.npos) {
0173
0174             std::string envar_name = new_fname.substr(pos_start + 1, pos_end - pos_start - 1);
0175             LOG_DEBUG(m_logger) << "Looking for env var '" << envar_name
0176                                 << "'" << LOG_END;
0177
0178             auto envar = getenv(envar_name.c_str());
0179             if (envar) {
0180                 new_fname.replace(pos_start - 1, pos_end + 2 - pos_start, envar);
0181             } else {
0182                 LOG_ERROR(m_logger) << "Environment variable '"
0183                                     << envar_name
0184                                     << "' not set. Cannot copy "
0185                                     << filename << LOG_END;
0186                 return;
0187             }
0188         } else {
0189             LOG_ERROR(m_logger) << "Error in string format: "
0190                                 << filename << LOG_END;
0191         }
0192     }
0193
0194     // Extract filename without path
0195     std::string base_fname = new_fname;
0196     if (auto pos = base_fname.rfind("/")) base_fname.erase(0, pos);
0197     auto out_name = m_output_dir + "/" + base_fname;
0198
0199     // Copy file
0200     LOG_INFO(m_logger) << "Copying " << new_fname << " -> " << m_output_dir << LOG_END;
0201     std::ifstream src(new_fname, std::ios::binary);
0202     std::ofstream dst(out_name, std::ios::binary);
0203     dst << src.rdbuf();
0204
0205     // Change permissions to match source
0206     struct stat st;
0207     stat(new_fname.c_str(), &st);
0208     chmod(out_name.c_str(), st.st_mode);
0209 }
0210
0211
0212
0213
0214
0215
0216
0217
0218
0219
0220
0221
0222