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

Support region callbacks when no channel is specified

Browse Source
This commit is contained in:
Alexey Rybalchenko 2019-03-08 11:06:30 +01:00
parent bd5105d609
commit 92632a022c
3 changed files with 16 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
examples/readout/Sampler.cxx
View File

@ -35,28 +35,23 @@ void Sampler::InitTask()
fMsgSize = fConfig->GetValue<int>("msg-size"); fMsgSize = fConfig->GetValue<int>("msg-size");
fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations"); fMaxIterations = fConfig->GetValue<uint64_t>("max-iterations");
fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegionFor("data1", fRegion = FairMQUnmanagedRegionPtr(NewUnmanagedRegion(10000000,
0, [this](void* /*data*/, size_t /*size*/, void* /*hint*/) { // callback to be called when message buffers no longer needed by transport
10000000, --fNumUnackedMsgs;
[this](void* /*data*/, size_t /*size*/, void* /*hint*/) { // callback to be called when message buffers no longer needed by transport if (fMaxIterations > 0)
--fNumUnackedMsgs; {
if (fMaxIterations > 0) LOG(debug) << "Received ack";
{ }
LOG(debug) << "Received ack"; }));
}
}
));
} }
bool Sampler::ConditionalRun() bool Sampler::ConditionalRun()
{ {
FairMQMessagePtr msg(NewMessageFor("data1", // channel FairMQMessagePtr msg(NewMessage(fRegion, // region
0, // sub-channel fRegion->GetData(), // ptr within region
fRegion, // region fMsgSize, // offset from ptr
fRegion->GetData(), // ptr within region nullptr // hint
fMsgSize, // offset from ptr ));
nullptr // hint
));
if (Send(msg, "data1", 0) > 0) if (Send(msg, "data1", 0) > 0)
{ {

1
examples/readout/fairmq-start-ex-readout.sh.in
View File

@ -11,6 +11,7 @@ fi
SAMPLER="fairmq-ex-readout-sampler" SAMPLER="fairmq-ex-readout-sampler"
SAMPLER+=" --id sampler1" SAMPLER+=" --id sampler1"
SAMPLER+=" --severity debug" SAMPLER+=" --severity debug"
SAMPLER+=" --transport shmem"
SAMPLER+=" --msg-size $msgSize" SAMPLER+=" --msg-size $msgSize"
# SAMPLER+=" --rate 10" # SAMPLER+=" --rate 10"
SAMPLER+=" --channel-config name=data1,type=pair,method=bind,address=tcp://127.0.0.1:7777,transport=shmem" SAMPLER+=" --channel-config name=data1,type=pair,method=bind,address=tcp://127.0.0.1:7777,transport=shmem"

4
fairmq/FairMQDevice.h
View File

@ -227,9 +227,9 @@ class FairMQDevice
return GetChannel(channel, index).NewSimpleMessage(data); return GetChannel(channel, index).NewSimpleMessage(data);
} }
FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size) FairMQUnmanagedRegionPtr NewUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr)
{ {
return Transport()->CreateUnmanagedRegion(size); return Transport()->CreateUnmanagedRegion(size, callback);
} }
FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr) FairMQUnmanagedRegionPtr NewUnmanagedRegionFor(const std::string& channel, int index, const size_t size, FairMQRegionCallback callback = nullptr)