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

start experimenting with root transport mechanism.

Browse Source 
enable efficient message copying for zeromq (currently used only in benchmarkSampler).
cleanup some code.
This commit is contained in:
Alexey Rybalchenko 2014-04-26 23:33:57 +02:00
parent 68d51d8ed5
commit 622e9e9c09
1 changed files with 15 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
fairmq/zeromq/FairMQMessageZMQ.cxx
View File

@ -31,7 +31,7 @@ FairMQMessageZMQ::FairMQMessageZMQ(size_t size)
FairMQMessageZMQ::FairMQMessageZMQ(void* data, size_t size) FairMQMessageZMQ::FairMQMessageZMQ(void* data, size_t size)
{ {
int rc = zmq_msg_init_data(&fMessage, data, size, &CleanUp, NULL); int rc = zmq_msg_init_data(&fMessage, data, size, &CleanUp, NULL); // TODO: expose the cleanup function part in the interface?
if (rc != 0) if (rc != 0)
{ {
LOG(ERROR) << "failed initializing message with data, reason: " << zmq_strerror(errno); LOG(ERROR) << "failed initializing message with data, reason: " << zmq_strerror(errno);
@ -61,7 +61,7 @@ void FairMQMessageZMQ::Rebuild(size_t size)
void FairMQMessageZMQ::Rebuild(void* data, size_t size) void FairMQMessageZMQ::Rebuild(void* data, size_t size)
{ {
CloseMessage(); CloseMessage();
int rc = zmq_msg_init_data(&fMessage, data, size, &CleanUp, NULL); int rc = zmq_msg_init_data(&fMessage, data, size, &CleanUp, NULL); // TODO: expose the cleanup function part in the interface?
if (rc != 0) if (rc != 0)
{ {
LOG(ERROR) << "failed initializing message with data, reason: " << zmq_strerror(errno); LOG(ERROR) << "failed initializing message with data, reason: " << zmq_strerror(errno);
@ -90,10 +90,19 @@ void FairMQMessageZMQ::SetMessage(void* data, size_t size)
void FairMQMessageZMQ::Copy(FairMQMessage* msg) void FairMQMessageZMQ::Copy(FairMQMessage* msg)
{ {
CloseMessage(); // Shares the message buffer between msg and this fMessage.
size_t size = msg->GetSize(); int rc = zmq_msg_copy(&fMessage, (zmq_msg_t*)msg->GetMessage());
zmq_msg_init_size(&fMessage, size); if (rc != 0)
std::memcpy(zmq_msg_data(&fMessage), msg->GetData(), size); {
LOG(ERROR) << "failed copying message, reason: " << zmq_strerror(errno);
}
// Alternatively, following code does a hard copy of the message, which allows to modify the original after making a copy, without affecting the new msg.
// CloseMessage();
// size_t size = msg->GetSize();
// zmq_msg_init_size(&fMessage, size);
// std::memcpy(zmq_msg_data(&fMessage), msg->GetData(), size);
} }
inline void FairMQMessageZMQ::CloseMessage() inline void FairMQMessageZMQ::CloseMessage()