Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-18 11:01:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

FairMQ: Introduce callbacks for the FairMQUnmanagedRegion.

Browse Source 
Callbacks are called when the data buffer of the message assiciated
with the corresponding region is no longer needed by the transport.
Example in examples/advanced/Region/
This commit is contained in:
Alexey Rybalchenko
2017-11-14 17:00:37 +01:00
parent 6eddf50a47
commit 15479a58d1
4 changed files with 45 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
examples/advanced/Region/FairMQExampleRegionSampler.cxx
View File

@@ -16,29 +16,54 @@
#include "FairMQLogger.h"
#include "FairMQProgOptions.h" // device->fConfig
#include <thread>
#include <atomic>
using namespace std;
FairMQExampleRegionSampler::FairMQExampleRegionSampler()
: fMsgSize(10000)
, fRegion(nullptr)
, fNumUnackedMsgs(0)
{
}
void FairMQExampleRegionSampler::InitTask()
{
fMsgSize = fConfig->GetValue<int>("msg-size");
fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("data",
0,
10000000,
[this](void* data, size_t size) { --fNumUnackedMsgs; } // callback to be called when message buffers no longer needed by transport
));
}
void FairMQExampleRegionSampler::Run()
bool FairMQExampleRegionSampler::ConditionalRun()
{
FairMQChannel& dataOutChannel = fChannels.at("data").at(0);
FairMQUnmanagedRegionPtr region(NewUnmanagedRegionFor("data", 0, 10000000));
while (CheckCurrentState(RUNNING))
FairMQMessagePtr msg(NewMessageFor("data", // channel
0, // sub-channel
fRegion, // region
fRegion->GetData(), // ptr within region
fMsgSize // offset from ptr
));
if (Send(msg, "data", 0) > 0)
{
FairMQMessagePtr msg(NewMessageFor("data", 0, region, region->GetData(), fMsgSize));
dataOutChannel.Send(msg);
++fNumUnackedMsgs;
}
return true;
}
void FairMQExampleRegionSampler::ResetTask()
{
// if not all messages acknowledged, wait for a bit. But only once, since receiver could be already dead.
if (fNumUnackedMsgs != 0)
{
LOG(DEBUG) << "waiting for all acknowledgements... (" << fNumUnackedMsgs << ")";
this_thread::sleep_for(chrono::milliseconds(500));
}
fRegion.reset();
}
FairMQExampleRegionSampler::~FairMQExampleRegionSampler()