Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 16:46:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

BenchmarkSampler: add memset option

Browse Source
This commit is contained in:
Alexey Rybalchenko 2020-06-29 11:23:28 +02:00
parent aeab9e5407
commit afe2dcaa02
3 changed files with 12 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
fairmq/devices/FairMQBenchmarkSampler.h
View File

@ -17,6 +17,7 @@
#include <chrono>
#include <cstddef> // size_t
#include <cstdint> // uint64_t
#include <cstring> // memset
#include <string>
/**
@ -28,6 +29,7 @@ class FairMQBenchmarkSampler : public FairMQDevice
public:
FairMQBenchmarkSampler()
: fMultipart(false)
, fMemSet(false)
, fNumParts(1)
, fMsgSize(10000)
, fMsgRate(0)
@ -39,6 +41,7 @@ class FairMQBenchmarkSampler : public FairMQDevice
void InitTask() override
{
fMultipart = fConfig->GetProperty<bool>("multipart");
fMemSet = fConfig->GetProperty<bool>("memset");
fNumParts = fConfig->GetProperty<size_t>("num-parts");
fMsgSize = fConfig->GetProperty<size_t>("msg-size");
fMsgRate = fConfig->GetProperty<float>("msg-rate");
@ -64,6 +67,9 @@ class FairMQBenchmarkSampler : public FairMQDevice
for (size_t i = 0; i < fNumParts; ++i) {
parts.AddPart(dataOutChannel.NewMessage(fMsgSize));
if (fMemSet) {
std::memset(parts.At(i)->GetData(), 0, parts.At(i)->GetSize());
}
}
if (dataOutChannel.Send(parts) >= 0) {
@ -76,6 +82,9 @@ class FairMQBenchmarkSampler : public FairMQDevice
}
} else {
FairMQMessagePtr msg(dataOutChannel.NewMessage(fMsgSize));
if (fMemSet) {
std::memset(msg->GetData(), 0, msg->GetSize());
}
if (dataOutChannel.Send(msg) >= 0) {
if (fMaxIterations > 0) {
@ -101,6 +110,7 @@ class FairMQBenchmarkSampler : public FairMQDevice
protected:
bool fMultipart;
bool fMemSet;
size_t fNumParts;
size_t fMsgSize;
std::atomic<int> fMsgCounter;

2
fairmq/run/runBenchmarkSampler.cxx
View File

@ -15,8 +15,8 @@ void addCustomOptions(bpo::options_description& options)
{
options.add_options()
("out-channel", bpo::value<std::string>()->default_value("data"), "Name of the output channel")
("same-msg", bpo::value<bool>()->default_value(false), "Re-send the same message, or recreate for each iteration")
("multipart", bpo::value<bool>()->default_value(false), "Handle multipart payloads")
("memset", bpo::value<bool>()->default_value(false), "Memset allocated buffers to 0")
("num-parts", bpo::value<size_t>()->default_value(1), "Number of parts to send. 1 will send single messages, not parts")
("msg-size", bpo::value<size_t>()->default_value(1000000), "Message size in bytes")
("max-iterations", bpo::value<uint64_t>()->default_value(0), "Number of run iterations (0 - infinite)")

2
fairmq/zeromq/TransportFactory.h
View File

@ -55,7 +55,7 @@ class TransportFactory final : public FairMQTransportFactory
{
return tools::make_unique<Message>(this);
}
MessagePtr CreateMessage(Alignment alignment) override
{
return tools::make_unique<Message>(alignment, this);