Source · fairmq/devices/
BenchmarkSampler.hheader
Described in builtin-devices: a topology of built-in devices, Run a benchmark, Built-in devices.
1/********************************************************************************2 * Copyright (C) 2014-2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_BENCHMARKSAMPLER_H10#define FAIR_MQ_BENCHMARKSAMPLER_H11 12#include <fairmq/Device.h>13#include <fairmq/tools/RateLimit.h>14 15#include <chrono>16#include <cstddef> // size_t17#include <cstdint> // uint64_t18#include <cstring> // memset19#include <fairlogger/Logger.h>20#include <string>21 22namespace fair::mq23{24 25/**26 * Sampler to generate traffic for benchmarking.27 */28 29class BenchmarkSampler : public Device30{31 public:32 void InitTask() override33 {34 fMultipart = fConfig->GetProperty<bool>("multipart");35 fMemSet = fConfig->GetProperty<bool>("memset");36 fNumParts = fConfig->GetProperty<size_t>("num-parts");37 fMsgSize = fConfig->GetProperty<size_t>("msg-size");38 fMsgAlignment = fConfig->GetProperty<size_t>("msg-alignment");39 fMsgRate = fConfig->GetProperty<float>("msg-rate");40 fMaxIterations = fConfig->GetProperty<uint64_t>("max-iterations");41 fOutChannelName = fConfig->GetProperty<std::string>("out-channel");42 }43 44 void Run() override45 {46 // store the channel reference to avoid traversing the map on every loop iteration47 Channel& dataOutChannel = GetChannel(fOutChannelName, 0);48 49 LOG(info) << "Starting the benchmark with message size of " << fMsgSize << " and " << fMaxIterations << " iterations.";50 auto tStart = std::chrono::high_resolution_clock::now();51 52 fair::mq::tools::RateLimiter rateLimiter(fMsgRate);53 54 while (!NewStatePending()) {55 if (fMultipart) {56 Parts parts;57 58 for (size_t i = 0; i < fNumParts; ++i) {59 parts.AddPart(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment}));60 if (fMemSet) {61 std::memset(parts.At(i)->GetData(), 0, parts.At(i)->GetSize());62 }63 }64 65 if (dataOutChannel.Send(parts) >= 0) {66 if (fMaxIterations > 0) {67 if (fNumIterations >= fMaxIterations) {68 break;69 }70 }71 ++fNumIterations;72 }73 } else {74 MessagePtr msg(dataOutChannel.NewMessage(fMsgSize, fair::mq::Alignment{fMsgAlignment}));75 if (fMemSet) {76 std::memset(msg->GetData(), 0, msg->GetSize());77 }78 79 if (dataOutChannel.Send(msg) >= 0) {80 if (fMaxIterations > 0) {81 if (fNumIterations >= fMaxIterations) {82 break;83 }84 }85 ++fNumIterations;86 }87 }88 89 if (fMsgRate > 0) {90 rateLimiter.maybe_sleep();91 }92 }93 94 auto tEnd = std::chrono::high_resolution_clock::now();95 96 LOG(info) << "Done " << fNumIterations << " iterations in " << std::chrono::duration<double, std::milli>(tEnd - tStart).count() << "ms.";97 }98 99 protected:100 bool fMultipart = false;101 bool fMemSet = false;102 size_t fNumParts = 1;103 size_t fMsgSize = 10000;104 size_t fMsgAlignment = 0;105 float fMsgRate = 0;106 uint64_t fNumIterations = 0;107 uint64_t fMaxIterations = 0;108 std::string fOutChannelName;109};110 111} // namespace fair::mq112 113#endif /* FAIR_MQ_BENCHMARKSAMPLER_H */