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

Implement parallel ofi::Socket::Send

Browse Source
This commit is contained in:
Dennis Klein 2018-11-13 22:26:38 +01:00 committed by Dennis Klein
parent 9ae48c21f5
commit 46e2420547
7 changed files with 229 additions and 348 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
fairmq/ofi/Context.cxx
View File

@ -21,7 +21,6 @@
#include <string> #include <string>
#include <string.h> #include <string.h>
#include <sys/socket.h> #include <sys/socket.h>
#include <zmq.h>
namespace fair namespace fair
{ {
@ -33,15 +32,11 @@ namespace ofi
using namespace std; using namespace std;
Context::Context(int numberIoThreads) Context::Context(int numberIoThreads)
: fZmqContext(zmq_ctx_new()) : fOfiInfo(nullptr)
, fOfiInfo(nullptr)
, fOfiFabric(nullptr) , fOfiFabric(nullptr)
, fOfiDomain(nullptr) , fOfiDomain(nullptr)
, fIoWork(fIoContext) , fIoWork(fIoContext)
{ {
if (!fZmqContext)
throw ContextError{tools::ToString("Failed creating zmq context, reason: ", zmq_strerror(errno))};
InitThreadPool(numberIoThreads); InitThreadPool(numberIoThreads);
} }
@ -63,16 +58,6 @@ Context::~Context()
fIoContext.stop(); fIoContext.stop();
for (auto& thread : fThreadPool) for (auto& thread : fThreadPool)
thread.join(); thread.join();
if (zmq_ctx_term(fZmqContext) != 0)
LOG(error) << "Failed closing zmq context, reason: " << zmq_strerror(errno);
}
auto Context::GetZmqVersion() const -> string
{
int major, minor, patch;
zmq_version(&major, &minor, &patch);
return tools::ToString(major, ".", minor, ".", patch);
} }
auto Context::GetAsiofiVersion() const -> string auto Context::GetAsiofiVersion() const -> string

3
fairmq/ofi/Context.h
View File

@ -48,9 +48,7 @@ class Context
~Context(); ~Context();
// auto CreateOfiEndpoint() -> fid_ep*; // auto CreateOfiEndpoint() -> fid_ep*;
auto GetZmqVersion() const -> std::string;
auto GetAsiofiVersion() const -> std::string; auto GetAsiofiVersion() const -> std::string;
auto GetZmqContext() const -> void* { return fZmqContext; }
auto GetIoContext() -> boost::asio::io_context& { return fIoContext; } auto GetIoContext() -> boost::asio::io_context& { return fIoContext; }
struct Address { struct Address {
std::string Protocol; std::string Protocol;
@ -70,7 +68,6 @@ class Context
auto Resume() -> void { LOG(debug) << "OFI transport: Resumed (NOOP - not implemented)."; } auto Resume() -> void { LOG(debug) << "OFI transport: Resumed (NOOP - not implemented)."; }
private: private:
void* fZmqContext;
std::unique_ptr<asiofi::info> fOfiInfo; std::unique_ptr<asiofi::info> fOfiInfo;
std::unique_ptr<asiofi::fabric> fOfiFabric; std::unique_ptr<asiofi::fabric> fOfiFabric;
std::unique_ptr<asiofi::domain> fOfiDomain; std::unique_ptr<asiofi::domain> fOfiDomain;

72
fairmq/ofi/ControlMessages.h
View File

@ -9,17 +9,28 @@
#ifndef FAIR_MQ_OFI_CONTROLMESSAGES_H #ifndef FAIR_MQ_OFI_CONTROLMESSAGES_H
#define FAIR_MQ_OFI_CONTROLMESSAGES_H #define FAIR_MQ_OFI_CONTROLMESSAGES_H
#include <FairMQLogger.h>
#include <boost/asio/buffer.hpp>
#include <cstdint> #include <cstdint>
#include <functional> #include <functional>
#include <memory> #include <memory>
#include <type_traits> #include <type_traits>
namespace fair namespace boost {
{ namespace asio {
namespace mq
{ template<typename PodType>
namespace ofi auto buffer(const PodType& obj) -> boost::asio::const_buffer
{ {
return boost::asio::const_buffer(static_cast<const void*>(&obj), sizeof(PodType));
}
} // namespace asio
} // namespace boost
namespace fair {
namespace mq {
namespace ofi {
enum class ControlMessageType enum class ControlMessageType
{ {
@ -28,59 +39,38 @@ enum class ControlMessageType
PostBufferAcknowledgement PostBufferAcknowledgement
}; };
struct ControlMessage { struct ControlMessage
{
ControlMessageType type; ControlMessageType type;
}; };
struct DataAddressAnnouncement : ControlMessage { struct DataAddressAnnouncement : ControlMessage
{
uint32_t ipv4; // in_addr_t from <netinet/in.h> uint32_t ipv4; // in_addr_t from <netinet/in.h>
uint32_t port; // in_port_t from <netinet/in.h> uint32_t port; // in_port_t from <netinet/in.h>
}; };
struct PostBuffer : ControlMessage { struct PostBuffer : ControlMessage
{
uint64_t size; // buffer size (size_t) uint64_t size; // buffer size (size_t)
}; };
struct PostBufferAcknowledgement { template<typename T, typename... Args>
uint64_t size; // size_t auto MakeControlMessage(Args&&... args) -> T
};
template<typename T>
using CtrlMsgPtr = std::unique_ptr<T, std::function<void(T*)>>;
template<typename T, typename A, typename ... Args>
auto MakeControlMessage(A* pmr, Args&& ... args) -> CtrlMsgPtr<T>
{ {
void* raw_mem = pmr->allocate(sizeof(T)); T ctrl = T(std::forward<Args>(args)...);
T* raw_ptr = new (raw_mem) T(std::forward<Args>(args)...);
if (std::is_same<T, DataAddressAnnouncement>::value) { if (std::is_same<T, DataAddressAnnouncement>::value) {
raw_ptr->type = ControlMessageType::DataAddressAnnouncement; ctrl.type = ControlMessageType::DataAddressAnnouncement;
} else if (std::is_same<T, PostBuffer>::value) { } else if (std::is_same<T, PostBuffer>::value) {
raw_ptr->type = ControlMessageType::PostBuffer; ctrl.type = ControlMessageType::PostBuffer;
} }
return {raw_ptr, [=](T* p) { pmr->deallocate(p, sizeof(T)); }}; return ctrl;
} }
template<typename Derived, typename Base, typename Del> } // namespace ofi
auto StaticUniquePtrDowncast(std::unique_ptr<Base, Del>&& p) -> std::unique_ptr<Derived, Del> } // namespace mq
{ } // namespace fair
auto down = static_cast<Derived*>(p.release());
return std::unique_ptr<Derived, Del>(down, std::move(p.get_deleter()));
}
template<typename Base, typename Derived, typename Del>
auto StaticUniquePtrUpcast(std::unique_ptr<Derived, Del>&& p) -> std::unique_ptr<Base, std::function<void(Base*)>>
{
auto up = static_cast<Base*>(p.release());
return {up, [deleter = std::move(p.get_deleter())](Base* ptr) {
deleter(static_cast<Derived*>(ptr));
}};
}
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
#endif /* FAIR_MQ_OFI_CONTROLMESSAGES_H */ #endif /* FAIR_MQ_OFI_CONTROLMESSAGES_H */

12
fairmq/ofi/Poller.cxx
View File

@ -28,13 +28,13 @@ Poller::Poller(const vector<FairMQChannel>& channels)
fItems = new zmq_pollitem_t[fNumItems]; fItems = new zmq_pollitem_t[fNumItems];
for (int i = 0; i < fNumItems; ++i) { for (int i = 0; i < fNumItems; ++i) {
fItems[i].socket = static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket(); fItems[i].socket = static_cast<Socket*>(&(channels.at(i).GetSocket()))->GetSocket();
fItems[i].fd = 0; fItems[i].fd = 0;
fItems[i].revents = 0; fItems[i].revents = 0;
int type = 0; int type = 0;
size_t size = sizeof(type); size_t size = sizeof(type);
zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size); zmq_getsockopt(static_cast<Socket*>(&(channels.at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
SetItemEvents(fItems[i], type); SetItemEvents(fItems[i], type);
} }
@ -46,13 +46,13 @@ Poller::Poller(const vector<const FairMQChannel*>& channels)
fItems = new zmq_pollitem_t[fNumItems]; fItems = new zmq_pollitem_t[fNumItems];
for (int i = 0; i < fNumItems; ++i) { for (int i = 0; i < fNumItems; ++i) {
fItems[i].socket = static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket(); fItems[i].socket = static_cast<Socket*>(&(channels.at(i)->GetSocket()))->GetSocket();
fItems[i].fd = 0; fItems[i].fd = 0;
fItems[i].revents = 0; fItems[i].revents = 0;
int type = 0; int type = 0;
size_t size = sizeof(type); size_t size = sizeof(type);
zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size); zmq_getsockopt(static_cast<Socket*>(&(channels.at(i)->GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
SetItemEvents(fItems[i], type); SetItemEvents(fItems[i], type);
} }
@ -76,13 +76,13 @@ Poller::Poller(const unordered_map<string, vector<FairMQChannel>>& channelsMap,
for (unsigned int i = 0; i < channelsMap.at(channel).size(); ++i) { for (unsigned int i = 0; i < channelsMap.at(channel).size(); ++i) {
index = fOffsetMap[channel] + i; index = fOffsetMap[channel] + i;
fItems[index].socket = static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(); fItems[index].socket = static_cast<Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket();
fItems[index].fd = 0; fItems[index].fd = 0;
fItems[index].revents = 0; fItems[index].revents = 0;
int type = 0; int type = 0;
size_t size = sizeof(type); size_t size = sizeof(type);
zmq_getsockopt(static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size); zmq_getsockopt(static_cast<Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
SetItemEvents(fItems[index], type); SetItemEvents(fItems[index], type);
} }

424
fairmq/ofi/Socket.cxx
View File

@ -12,15 +12,17 @@
#include <fairmq/Tools.h> #include <fairmq/Tools.h>
#include <FairMQLogger.h> #include <FairMQLogger.h>
#include <arpa/inet.h>
#include <asiofi.hpp> #include <asiofi.hpp>
#include <azmq/message.hpp>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/buffer.hpp> #include <boost/asio/buffer.hpp>
#include <boost/asio/post.hpp>
#include <cstring> #include <cstring>
#include <netinet/in.h> #include <functional>
#include <memory>
#include <sstream> #include <sstream>
#include <string.h> #include <string.h>
#include <sys/socket.h> #include <sys/socket.h>
#include <zmq.h>
#include <mutex> #include <mutex>
#include <condition_variable> #include <condition_variable>
@ -35,8 +37,7 @@ namespace ofi
using namespace std; using namespace std;
Socket::Socket(Context& context, const string& type, const string& name, const string& id /*= ""*/) Socket::Socket(Context& context, const string& type, const string& name, const string& id /*= ""*/)
: fControlSocket(nullptr) : fContext(context)
// , fMonitorSocket(nullptr)
, fPassiveDataEndpoint(nullptr) , fPassiveDataEndpoint(nullptr)
, fDataEndpoint(nullptr) , fDataEndpoint(nullptr)
, fId(id + "." + name + "." + type) , fId(id + "." + name + "." + type)
@ -44,46 +45,37 @@ Socket::Socket(Context& context, const string& type, const string& name, const s
, fBytesRx(0) , fBytesRx(0)
, fMessagesTx(0) , fMessagesTx(0)
, fMessagesRx(0) , fMessagesRx(0)
, fContext(context)
, fIoStrand(fContext.GetIoContext()) , fIoStrand(fContext.GetIoContext())
, fControlEndpoint(fIoStrand.context(), ZMQ_PAIR)
, fSndTimeout(100) , fSndTimeout(100)
, fRcvTimeout(100) , fRcvTimeout(100)
, fQueue1(fIoStrand.context())
, fQueue2(fIoStrand.context())
{ {
if (type != "pair") { if (type != "pair") {
throw SocketError{tools::ToString("Socket type '", type, "' not implemented for ofi transport.")}; throw SocketError{tools::ToString("Socket type '", type, "' not implemented for ofi transport.")};
} else { } else {
fControlSocket = zmq_socket(fContext.GetZmqContext(), ZMQ_PAIR); fControlEndpoint.set_option(azmq::socket::identity(fId));
if (fControlSocket == nullptr)
throw SocketError{tools::ToString("Failed creating zmq meta socket ", fId, ", reason: ", zmq_strerror(errno))};
if (zmq_setsockopt(fControlSocket, ZMQ_IDENTITY, fId.c_str(), fId.length()) != 0)
throw SocketError{tools::ToString("Failed setting ZMQ_IDENTITY socket option, reason: ", zmq_strerror(errno))};
// Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating. // Tell socket to try and send/receive outstanding messages for <linger> milliseconds before terminating.
// Default value for ZeroMQ is -1, which is to wait forever. // Default value for ZeroMQ is -1, which is to wait forever.
int linger = 1000; fControlEndpoint.set_option(azmq::socket::linger(1000));
if (zmq_setsockopt(fControlSocket, ZMQ_LINGER, &linger, sizeof(linger)) != 0)
throw SocketError{tools::ToString("Failed setting ZMQ_LINGER socket option, reason: ", zmq_strerror(errno))};
// TODO enable again and implement retries // Setup internal queue
// if (zmq_setsockopt(fControlSocket, ZMQ_SNDTIMEO, &fSndTimeout, sizeof(fSndTimeout)) != 0) auto hashed_id = std::hash<std::string>()(fId);
// throw SocketError{tools::ToString("Failed setting ZMQ_SNDTIMEO socket option, reason: ", zmq_strerror(errno))}; auto queue_id = tools::ToString("inproc://QUEUE", hashed_id);
// LOG(debug) << "OFI transport (" << fId << "): " << "Binding Q1: " << queue_id;
// if (zmq_setsockopt(fControlSocket, ZMQ_RCVTIMEO, &fRcvTimeout, sizeof(fRcvTimeout)) != 0) fQueue1.bind(queue_id);
// throw SocketError{tools::ToString("Failed setting ZMQ_RCVTIMEO socket option, reason: ", zmq_strerror(errno))}; LOG(debug) << "OFI transport (" << fId << "): " << "Connecting Q2: " << queue_id;
fQueue2.connect(queue_id);
// fMonitorSocket = zmq_socket(fContext.GetZmqContext(), ZMQ_PAIR); azmq::socket::snd_hwm send_max(100);
// azmq::socket::rcv_hwm recv_max(100);
// if (fMonitorSocket == nullptr) fQueue1.set_option(send_max);
// throw SocketError{tools::ToString("Failed creating zmq monitor socket ", fId, ", reason: ", zmq_strerror(errno))}; fQueue1.set_option(recv_max);
// fQueue2.set_option(send_max);
// auto mon_addr = tools::ToString("inproc://", fId); fQueue2.set_option(recv_max);
// if (zmq_socket_monitor(fControlSocket, mon_addr.c_str(), ZMQ_EVENT_ACCEPTED | ZMQ_EVENT_CONNECTED) < 0) fControlEndpoint.set_option(send_max);
// throw SocketError{tools::ToString("Failed setting up monitor on meta socket, reason: ", zmq_strerror(errno))}; fControlEndpoint.set_option(recv_max);
//
// if (zmq_connect(fMonitorSocket, mon_addr.c_str()) != 0)
// throw SocketError{tools::ToString("Failed connecting monitor socket to meta socket, reason: ", zmq_strerror(errno))};
} }
} }
@ -91,13 +83,15 @@ auto Socket::Bind(const string& address) -> bool
try { try {
auto addr = Context::VerifyAddress(address); auto addr = Context::VerifyAddress(address);
BindControlSocket(addr); BindControlEndpoint(addr);
// TODO make data port choice more robust // TODO make data port choice more robust
addr.Port += 555; addr.Port += 555;
fLocalDataAddr = addr; fLocalDataAddr = addr;
BindDataEndpoint(); BindDataEndpoint();
boost::asio::post(fIoStrand, std::bind(&Socket::SendQueueReader, this));
return true; return true;
} }
catch (const SilentSocketError& e) catch (const SilentSocketError& e)
@ -115,23 +109,24 @@ catch (const SocketError& e)
auto Socket::Connect(const string& address) -> bool auto Socket::Connect(const string& address) -> bool
{ {
auto addr = Context::VerifyAddress(address); auto addr = Context::VerifyAddress(address);
fRemoteDataAddr = addr;
ConnectControlSocket(addr); ConnectControlEndpoint(addr);
ProcessControlMessage( ReceiveDataAddressAnnouncement();
StaticUniquePtrDowncast<DataAddressAnnouncement>(ReceiveControlMessage()));
ConnectDataEndpoint(); ConnectDataEndpoint();
} }
auto Socket::BindControlSocket(Context::Address address) -> void auto Socket::BindControlEndpoint(Context::Address address) -> void
{ {
auto addr = tools::ToString("tcp://", address.Ip, ":", address.Port); auto addr = tools::ToString("tcp://", address.Ip, ":", address.Port);
if (zmq_bind(fControlSocket, addr.c_str()) != 0) { fControlEndpoint.bind(addr);
if (errno == EADDRINUSE) throw SilentSocketError("EADDRINUSE"); // if (zmq_bind(fControlSocket, addr.c_str()) != 0) {
throw SocketError(tools::ToString("Failed binding control socket ", fId, ", reason: ", zmq_strerror(errno))); // TODO if (errno == EADDRINUSE) throw SilentSocketError("EADDRINUSE");
} // throw SocketError(tools::ToString("Failed binding control socket ", fId, ", reason: ", zmq_strerror(errno)));
// }
LOG(debug) << "OFI transport (" << fId << "): control band bound to " << address; LOG(debug) << "OFI transport (" << fId << "): control band bound to " << address;
} }
@ -170,12 +165,13 @@ auto Socket::BindDataEndpoint() -> void
LOG(debug) << "OFI transport (" << fId << "): data band connection accepted."; LOG(debug) << "OFI transport (" << fId << "): data band connection accepted.";
} }
auto Socket::ConnectControlSocket(Context::Address address) -> void auto Socket::ConnectControlEndpoint(Context::Address address) -> void
{ {
auto addr = tools::ToString("tcp://", address.Ip, ":", address.Port); auto addr = tools::ToString("tcp://", address.Ip, ":", address.Port);
if (zmq_connect(fControlSocket, addr.c_str()) != 0) fControlEndpoint.connect(addr);
throw SocketError(tools::ToString("Failed connecting control socket ", fId, ", reason: ", zmq_strerror(errno)));
LOG(debug) << "OFI transport (" << fId << "): control band connected to " << address;
} }
auto Socket::ConnectDataEndpoint() -> void auto Socket::ConnectDataEndpoint() -> void
@ -191,8 +187,12 @@ auto Socket::ConnectDataEndpoint() -> void
}); });
} }
auto Socket::ProcessControlMessage(CtrlMsgPtr<DataAddressAnnouncement> daa) -> void auto Socket::ReceiveDataAddressAnnouncement() -> void
{ {
azmq::message ctrl;
auto recv = fControlEndpoint.receive(ctrl);
assert(recv == sizeof(DataAddressAnnouncement)); (void)recv;
auto daa(static_cast<const DataAddressAnnouncement*>(ctrl.data()));
assert(daa->type == ControlMessageType::DataAddressAnnouncement); assert(daa->type == ControlMessageType::DataAddressAnnouncement);
sockaddr_in remoteAddr; sockaddr_in remoteAddr;
@ -201,191 +201,130 @@ auto Socket::ProcessControlMessage(CtrlMsgPtr<DataAddressAnnouncement> daa) -> v
remoteAddr.sin_addr.s_addr = daa->ipv4; remoteAddr.sin_addr.s_addr = daa->ipv4;
auto addr = Context::ConvertAddress(remoteAddr); auto addr = Context::ConvertAddress(remoteAddr);
addr.Protocol = fRemoteDataAddr.Protocol;
LOG(debug) << "OFI transport (" << fId << "): Data address announcement of remote endpoint received: " << addr; LOG(debug) << "OFI transport (" << fId << "): Data address announcement of remote endpoint received: " << addr;
fRemoteDataAddr = addr; fRemoteDataAddr = addr;
} }
auto Socket::AnnounceDataAddress() -> void auto Socket::AnnounceDataAddress() -> void
try { {
// fLocalDataAddr = fDataEndpoint->get_local_address(); // fLocalDataAddr = fDataEndpoint->get_local_address();
// LOG(debug) << "Address of local ofi endpoint in socket " << fId << ": " << Context::ConvertAddress(fLocalDataAddr); // LOG(debug) << "Address of local ofi endpoint in socket " << fId << ": " << Context::ConvertAddress(fLocalDataAddr);
// Create new data address announcement message // Create new data address announcement message
auto daa = MakeControlMessage<DataAddressAnnouncement>(&fCtrlMemPool); auto daa = MakeControlMessage<DataAddressAnnouncement>();
auto addr = Context::ConvertAddress(fLocalDataAddr); auto addr = Context::ConvertAddress(fLocalDataAddr);
daa->ipv4 = addr.sin_addr.s_addr; daa.ipv4 = addr.sin_addr.s_addr;
daa->port = addr.sin_port; daa.port = addr.sin_port;
SendControlMessage(StaticUniquePtrUpcast<ControlMessage>(std::move(daa))); auto sent = fControlEndpoint.send(boost::asio::buffer(daa));
assert(sent == sizeof(addr)); (void)sent;
LOG(debug) << "OFI transport (" << fId << "): data address announced."; LOG(debug) << "OFI transport (" << fId << "): data band address " << fLocalDataAddr << " announced.";
} catch (const SocketError& e) {
throw SocketError(tools::ToString("Failed to announce data address, reason: ", e.what()));
} }
auto Socket::SendControlMessage(CtrlMsgPtr<ControlMessage> ctrl) -> void auto Socket::Send(MessagePtr& msg, const int /*timeout*/) -> int
{ {
assert(fControlSocket); LOG(debug) << "OFI transport (" << fId << "): ENTER Send: size=" << msg->GetSize();
// LOG(debug) << "About to send control message: " << ctrl->DebugString();
// Serialize MessagePtr* msgptr(new std::unique_ptr<Message>(std::move(msg)));
struct ZmqMsg try {
{ auto res = fQueue1.send(boost::asio::const_buffer(msgptr, sizeof(MessagePtr)), 0);
zmq_msg_t msg;
~ZmqMsg() { zmq_msg_close(&msg); }
operator zmq_msg_t*() { return &msg; }
} msg;
switch (ctrl->type) { LOG(debug) << "OFI transport (" << fId << "): LEAVE Send";
case ControlMessageType::DataAddressAnnouncement: return res;
{ } catch (const std::exception& e) {
auto ret = zmq_msg_init_size(msg, sizeof(DataAddressAnnouncement)); msg = std::move(*msgptr);
(void)ret; LOG(error) << e.what();
assert(ret == 0); return -1;
std::memcpy(zmq_msg_data(msg), ctrl.get(), sizeof(DataAddressAnnouncement)); } catch (const boost::system::error_code& e) {
} msg = std::move(*msgptr);
break; LOG(error) << e;
case ControlMessageType::PostBuffer: return -1;
{
auto ret = zmq_msg_init_size(msg, sizeof(PostBuffer));
(void)ret;
assert(ret == 0);
std::memcpy(zmq_msg_data(msg), ctrl.get(), sizeof(PostBuffer));
}
break;
default:
throw SocketError(tools::ToString("Cannot send control message of unknown type."));
}
// Send
if (zmq_msg_send(msg, fControlSocket, 0) == -1) {
throw SocketError(
tools::ToString("Failed to send control message, reason: ", zmq_strerror(errno)));
} }
} }
auto Socket::ReceiveControlMessage() -> CtrlMsgPtr<ControlMessage> auto Socket::Receive(MessagePtr& msg, const int timeout) -> int { return 0; /*ReceiveImpl(msg, 0, timeout);*/ }
{
assert(fControlSocket);
// Receive
struct ZmqMsg
{
zmq_msg_t msg;
~ZmqMsg() { zmq_msg_close(&msg); }
operator zmq_msg_t*() { return &msg; }
} msg;
auto ret = zmq_msg_init(msg);
(void)ret;
assert(ret == 0);
if (zmq_msg_recv(msg, fControlSocket, 0) == -1) {
throw SocketError(
tools::ToString("Failed to receive control message, reason: ", zmq_strerror(errno)));
}
// Deserialize and sanity check
const void* msg_data = zmq_msg_data(msg);
const size_t msg_size = zmq_msg_size(msg);
(void)msg_size;
assert(msg_size >= sizeof(ControlMessage));
switch (static_cast<const ControlMessage*>(msg_data)->type) {
case ControlMessageType::DataAddressAnnouncement: {
assert(msg_size == sizeof(DataAddressAnnouncement));
auto daa = MakeControlMessage<DataAddressAnnouncement>(&fCtrlMemPool);
std::memcpy(daa.get(), msg_data, sizeof(DataAddressAnnouncement));
// LOG(debug) << "Received control message: " << ctrl->DebugString();
return StaticUniquePtrUpcast<ControlMessage>(std::move(daa));
}
case ControlMessageType::PostBuffer: {
assert(msg_size == sizeof(PostBuffer));
auto pb = MakeControlMessage<PostBuffer>(&fCtrlMemPool);
std::memcpy(pb.get(), msg_data, sizeof(PostBuffer));
// LOG(debug) << "Received control message: " << ctrl->DebugString();
return StaticUniquePtrUpcast<ControlMessage>(std::move(pb));
}
default:
throw SocketError(tools::ToString("Received control message of unknown type."));
}
}
auto Socket::Send(MessagePtr& msg, const int timeout) -> int { return SendImpl(msg, 0, timeout); }
auto Socket::Receive(MessagePtr& msg, const int timeout) -> int { return ReceiveImpl(msg, 0, timeout); }
auto Socket::Send(std::vector<MessagePtr>& msgVec, const int timeout) -> int64_t { return SendImpl(msgVec, 0, timeout); } auto Socket::Send(std::vector<MessagePtr>& msgVec, const int timeout) -> int64_t { return SendImpl(msgVec, 0, timeout); }
auto Socket::Receive(std::vector<MessagePtr>& msgVec, const int timeout) -> int64_t { return ReceiveImpl(msgVec, 0, timeout); } auto Socket::Receive(std::vector<MessagePtr>& msgVec, const int timeout) -> int64_t { return ReceiveImpl(msgVec, 0, timeout); }
auto Socket::SendImpl(FairMQMessagePtr& msg, const int /*flags*/, const int /*timeout*/) -> int auto Socket::SendQueueReader() -> void
try { {
fQueue2.async_receive(boost::asio::bind_executor(
fIoStrand,
[&](const boost::system::error_code& ec, azmq::message& zmsg, size_t bytes_transferred) {
if (!ec) {
OnSend(zmsg, bytes_transferred);
}
}));
}
auto Socket::OnSend(azmq::message& zmsg, size_t bytes_transferred) -> void
{
LOG(debug) << "OFI transport (" << fId << "): ENTER OnSend: bytes_transferred=" << bytes_transferred;
MessagePtr msg(std::move(*(static_cast<MessagePtr*>(zmsg.buffer().data()))));
auto size = msg->GetSize(); auto size = msg->GetSize();
// LOG(debug) << "OFI transport (" << fId << "): ENTER SendImpl";
LOG(debug) << "OFI transport (" << fId << "): >>>>> OnSend: size=" << size;
// Create and send control message // Create and send control message
auto pb = MakeControlMessage<PostBuffer>(&fCtrlMemPool); auto pb = MakeControlMessage<PostBuffer>();
pb->size = size; pb.size = size;
SendControlMessage(StaticUniquePtrUpcast<ControlMessage>(std::move(pb))); fControlEndpoint.async_send(
// LOG(debug) << "OFI transport (" << fId << "): >>>>> SendImpl: Control message sent, size=" << size; azmq::message(boost::asio::buffer(pb)),
// LOG(debug) << "OFI transport (" << fId << "): >>>>> SendImpl: msg->GetData()=" << msg->GetData() << ",msg->GetSize()=" << msg->GetSize(); [&, msg2 = std::move(msg)](const boost::system::error_code& ec, size_t bytes_transferred2) mutable {
if (!ec) {
OnControlMessageSent(bytes_transferred2, std::move(msg2));
}
});
LOG(debug) << "OFI transport (" << fId << "): LEAVE OnSend";
}
auto Socket::OnControlMessageSent(size_t bytes_transferred, MessagePtr msg) -> void
{
LOG(debug) << "OFI transport (" << fId << "): ENTER OnControlMessageSent: bytes_transferred=" << bytes_transferred;
assert(bytes_transferred == sizeof(PostBuffer));
auto size = msg->GetSize();
if (size) { if (size) {
// Receive ack // Receive ack
auto ack = StaticUniquePtrDowncast<PostBuffer>(ReceiveControlMessage()); // azmq::message ctrl;
assert(ack.get()); // auto recv = fControlEndpoint.receive(ctrl);
auto size_ack = ack->size; // assert(recv == sizeof(PostBuffer));
assert(size == size_ack); // (void)recv;
// LOG(debug) << "OFI transport (" << fId << "): >>>>> SendImpl: Control ack received, size_ack=" << size_ack; // auto ack(static_cast<const PostBuffer*>(ctrl.data()));
// assert(ack->type == ControlMessageType::PostBuffer);
// (void)ack;
// LOG(debug) << "OFI transport (" << fId << "): >>>>> SendImpl: Control ack
// received, size_ack=" << size_ack;
boost::asio::mutable_buffer buffer(msg->GetData(), size); boost::asio::mutable_buffer buffer(msg->GetData(), size);
asiofi::memory_region mr(fContext.GetDomain(), buffer, asiofi::mr::access::send); asiofi::memory_region mr(fContext.GetDomain(), buffer, asiofi::mr::access::send);
std::mutex m; fDataEndpoint->send(buffer, mr.desc(), [&, mr2 = std::move(mr)](boost::asio::mutable_buffer) {
std::condition_variable cv; LOG(debug) << "OFI transport (" << fId << "): >>>>> Data buffer sent";
bool completed(false);
fDataEndpoint->send(
buffer,
mr.desc(),
[&](boost::asio::mutable_buffer) {
{
std::unique_lock<std::mutex> lk(m);
completed = true;
}
cv.notify_one();
// LOG(debug) << "OFI transport (" << fId << "): > SendImpl: Data buffer sent";
}
);
{
std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [&](){ return completed; });
}
// LOG(debug) << "OFI transport (" << fId << "): >>>>> SendImpl: Data send buffer posted";
}
msg.reset(nullptr);
fBytesTx += size; fBytesTx += size;
fMessagesTx++; fMessagesTx++;
});
}
// LOG(debug) << "OFI transport (" << fId << "): LEAVE SendImpl"; LOG(debug) << "OFI transport (" << fId << "): LEAVE OnControlMessageSent";
return size;
}
catch (const SilentSocketError& e)
{
return -2;
}
catch (const std::exception& e)
{
LOG(error) << e.what();
return -1;
} }
auto Socket::ReceiveImpl(FairMQMessagePtr& msg, const int /*flags*/, const int /*timeout*/) -> int auto Socket::ReceiveImpl(FairMQMessagePtr& msg, const int /*flags*/, const int /*timeout*/) -> int
try { try {
// LOG(debug) << "OFI transport (" << fId << "): ENTER ReceiveImpl"; LOG(debug) << "OFI transport (" << fId << "): ENTER ReceiveImpl";
// Receive and process control message // Receive and process control message
auto pb = StaticUniquePtrDowncast<PostBuffer>(ReceiveControlMessage()); azmq::message ctrl;
assert(pb.get()); auto recv = fControlEndpoint.receive(ctrl);
assert(recv == sizeof(PostBuffer)); (void)recv;
auto pb(static_cast<const PostBuffer*>(ctrl.data()));
assert(pb->type == ControlMessageType::PostBuffer);
auto size = pb->size; auto size = pb->size;
// LOG(debug) << "OFI transport (" << fId << "): <<<<< ReceiveImpl: Control message received, size=" << size; LOG(debug) << "OFI transport (" << fId << "): <<<<< ReceiveImpl: Control message received, size=" << size;
// Receive data // Receive data
if (size) { if (size) {
@ -407,9 +346,10 @@ try {
); );
// LOG(debug) << "OFI transport (" << fId << "): <<<<< ReceiveImpl: Data buffer posted"; // LOG(debug) << "OFI transport (" << fId << "): <<<<< ReceiveImpl: Data buffer posted";
auto ack = MakeControlMessage<PostBuffer>(&fCtrlMemPool); auto ack = MakeControlMessage<PostBuffer>();
ack->size = size; ack.size = size;
SendControlMessage(StaticUniquePtrUpcast<ControlMessage>(std::move(ack))); auto sent = fControlEndpoint.send(boost::asio::buffer(ack));
assert(sent == sizeof(PostBuffer)); (void)sent;
// LOG(debug) << "OFI transport (" << fId << "): <<<<< ReceiveImpl: Control Ack sent"; // LOG(debug) << "OFI transport (" << fId << "): <<<<< ReceiveImpl: Control Ack sent";
{ {
@ -606,113 +546,85 @@ auto Socket::ReceiveImpl(vector<FairMQMessagePtr>& /*msgVec*/, const int /*flags
// } // }
} }
auto Socket::Close() -> void auto Socket::Close() -> void {}
{
if (zmq_close(fControlSocket) != 0)
throw SocketError(tools::ToString("Failed closing zmq meta socket, reason: ", zmq_strerror(errno)));
// if (zmq_close(fMonitorSocket) != 0)
// throw SocketError(tools::ToString("Failed closing zmq monitor socket, reason: ", zmq_strerror(errno)));
}
auto Socket::SetOption(const string& option, const void* value, size_t valueSize) -> void auto Socket::SetOption(const string& option, const void* value, size_t valueSize) -> void
{ {
if (zmq_setsockopt(fControlSocket, GetConstant(option), value, valueSize) < 0) { // if (zmq_setsockopt(fControlSocket, GetConstant(option), value, valueSize) < 0) {
throw SocketError{tools::ToString("Failed setting socket option, reason: ", zmq_strerror(errno))}; // throw SocketError{tools::ToString("Failed setting socket option, reason: ", zmq_strerror(errno))};
} // }
} }
auto Socket::GetOption(const string& option, void* value, size_t* valueSize) -> void auto Socket::GetOption(const string& option, void* value, size_t* valueSize) -> void
{ {
if (zmq_getsockopt(fControlSocket, GetConstant(option), value, valueSize) < 0) { // if (zmq_getsockopt(fControlSocket, GetConstant(option), value, valueSize) < 0) {
throw SocketError{tools::ToString("Failed getting socket option, reason: ", zmq_strerror(errno))}; // throw SocketError{tools::ToString("Failed getting socket option, reason: ", zmq_strerror(errno))};
} // }
}
int Socket::GetLinger() const
{
int value = 0;
size_t valueSize;
if (zmq_getsockopt(fControlSocket, ZMQ_LINGER, &value, &valueSize) < 0) {
throw SocketError(tools::ToString("failed getting ZMQ_LINGER, reason: ", zmq_strerror(errno)));
}
return value;
} }
void Socket::SetLinger(const int value) void Socket::SetLinger(const int value)
{ {
if (zmq_setsockopt(fControlSocket, ZMQ_LINGER, &value, sizeof(value)) < 0) { azmq::socket::linger opt(value);
throw SocketError(tools::ToString("failed setting ZMQ_LINGER, reason: ", zmq_strerror(errno))); fControlEndpoint.set_option(opt);
}
} }
int Socket::GetLinger() const
{
azmq::socket::linger opt(0);
fControlEndpoint.get_option(opt);
return opt.value();
}
void Socket::SetSndBufSize(const int value) void Socket::SetSndBufSize(const int value)
{ {
if (zmq_setsockopt(fControlSocket, ZMQ_SNDHWM, &value, sizeof(value)) < 0) { azmq::socket::snd_hwm opt(value);
throw SocketError(tools::ToString("failed setting ZMQ_SNDHWM, reason: ", zmq_strerror(errno))); fControlEndpoint.set_option(opt);
}
} }
int Socket::GetSndBufSize() const int Socket::GetSndBufSize() const
{ {
int value = 0; azmq::socket::snd_hwm opt(0);
size_t valueSize; fControlEndpoint.get_option(opt);
if (zmq_getsockopt(fControlSocket, ZMQ_SNDHWM, &value, &valueSize) < 0) { return opt.value();
throw SocketError(tools::ToString("failed getting ZMQ_SNDHWM, reason: ", zmq_strerror(errno)));
}
return value;
} }
void Socket::SetRcvBufSize(const int value) void Socket::SetRcvBufSize(const int value)
{ {
if (zmq_setsockopt(fControlSocket, ZMQ_RCVHWM, &value, sizeof(value)) < 0) { azmq::socket::rcv_hwm opt(value);
throw SocketError(tools::ToString("failed setting ZMQ_RCVHWM, reason: ", zmq_strerror(errno))); fControlEndpoint.set_option(opt);
}
} }
int Socket::GetRcvBufSize() const int Socket::GetRcvBufSize() const
{ {
int value = 0; azmq::socket::rcv_hwm opt(0);
size_t valueSize; fControlEndpoint.get_option(opt);
if (zmq_getsockopt(fControlSocket, ZMQ_RCVHWM, &value, &valueSize) < 0) { return opt.value();
throw SocketError(tools::ToString("failed getting ZMQ_RCVHWM, reason: ", zmq_strerror(errno)));
}
return value;
} }
void Socket::SetSndKernelSize(const int value) void Socket::SetSndKernelSize(const int value)
{ {
if (zmq_setsockopt(fControlSocket, ZMQ_SNDBUF, &value, sizeof(value)) < 0) { azmq::socket::snd_buf opt(value);
throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno))); fControlEndpoint.set_option(opt);
}
} }
int Socket::GetSndKernelSize() const int Socket::GetSndKernelSize() const
{ {
int value = 0; azmq::socket::snd_buf opt(0);
size_t valueSize; fControlEndpoint.get_option(opt);
if (zmq_getsockopt(fControlSocket, ZMQ_SNDBUF, &value, &valueSize) < 0) { return opt.value();
throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno)));
}
return value;
} }
void Socket::SetRcvKernelSize(const int value) void Socket::SetRcvKernelSize(const int value)
{ {
if (zmq_setsockopt(fControlSocket, ZMQ_RCVBUF, &value, sizeof(value)) < 0) { azmq::socket::rcv_buf opt(value);
throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno))); fControlEndpoint.set_option(opt);
}
} }
int Socket::GetRcvKernelSize() const int Socket::GetRcvKernelSize() const
{ {
int value = 0; azmq::socket::rcv_buf opt(0);
size_t valueSize; fControlEndpoint.get_option(opt);
if (zmq_getsockopt(fControlSocket, ZMQ_RCVBUF, &value, &valueSize) < 0) { return opt.value();
throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno)));
}
return value;
} }
auto Socket::GetConstant(const string& constant) -> int auto Socket::GetConstant(const string& constant) -> int

25
fairmq/ofi/Socket.h
View File

@ -15,8 +15,8 @@
#include <fairmq/ofi/ControlMessages.h> #include <fairmq/ofi/ControlMessages.h>
#include <asiofi/connected_endpoint.hpp> #include <asiofi/connected_endpoint.hpp>
#include <azmq/socket.hpp>
#include <boost/asio.hpp> #include <boost/asio.hpp>
#include <boost/container/pmr/unsynchronized_pool_resource.hpp>
#include <memory> // unique_ptr #include <memory> // unique_ptr
#include <netinet/in.h> #include <netinet/in.h>
class FairMQTransportFactory; class FairMQTransportFactory;
@ -51,7 +51,7 @@ class Socket final : public fair::mq::Socket
auto Send(std::vector<MessagePtr>& msgVec, int timeout = 0) -> int64_t override; auto Send(std::vector<MessagePtr>& msgVec, int timeout = 0) -> int64_t override;
auto Receive(std::vector<MessagePtr>& msgVec, int timeout = 0) -> int64_t override; auto Receive(std::vector<MessagePtr>& msgVec, int timeout = 0) -> int64_t override;
auto GetSocket() const -> void* { return fControlSocket; } auto GetSocket() const -> void* { return fControlEndpoint.native_handle(); }
void SetLinger(const int value) override; void SetLinger(const int value) override;
int GetLinger() const override; int GetLinger() const override;
@ -79,8 +79,7 @@ class Socket final : public fair::mq::Socket
~Socket() override; ~Socket() override;
private: private:
void* fControlSocket; Context& fContext;
// void* fMonitorSocket;
std::unique_ptr<asiofi::passive_endpoint> fPassiveDataEndpoint; std::unique_ptr<asiofi::passive_endpoint> fPassiveDataEndpoint;
std::unique_ptr<asiofi::connected_endpoint> fDataEndpoint; std::unique_ptr<asiofi::connected_endpoint> fDataEndpoint;
std::string fId; std::string fId;
@ -88,30 +87,28 @@ class Socket final : public fair::mq::Socket
std::atomic<unsigned long> fBytesRx; std::atomic<unsigned long> fBytesRx;
std::atomic<unsigned long> fMessagesTx; std::atomic<unsigned long> fMessagesTx;
std::atomic<unsigned long> fMessagesRx; std::atomic<unsigned long> fMessagesRx;
Context& fContext;
Context::Address fRemoteDataAddr; Context::Address fRemoteDataAddr;
Context::Address fLocalDataAddr; Context::Address fLocalDataAddr;
// bool fWaitingForControlPeer;
boost::asio::io_service::strand fIoStrand; boost::asio::io_service::strand fIoStrand;
boost::container::pmr::unsynchronized_pool_resource fCtrlMemPool; mutable azmq::socket fControlEndpoint;
int fSndTimeout; int fSndTimeout;
int fRcvTimeout; int fRcvTimeout;
azmq::pair_socket fQueue1, fQueue2;
auto SendImpl(MessagePtr& msg, const int flags, const int timeout) -> int; auto SendQueueReader() -> void;
auto OnSend(azmq::message& msg, size_t bytes_transferred) -> void;
auto OnControlMessageSent(size_t bytes_transferred, MessagePtr msg) -> void;
auto ReceiveImpl(MessagePtr& msg, const int flags, const int timeout) -> int; auto ReceiveImpl(MessagePtr& msg, const int flags, const int timeout) -> int;
auto SendImpl(std::vector<MessagePtr>& msgVec, const int flags, const int timeout) -> int64_t; auto SendImpl(std::vector<MessagePtr>& msgVec, const int flags, const int timeout) -> int64_t;
auto ReceiveImpl(std::vector<MessagePtr>& msgVec, const int flags, const int timeout) -> int64_t; auto ReceiveImpl(std::vector<MessagePtr>& msgVec, const int flags, const int timeout) -> int64_t;
// auto WaitForControlPeer() -> void; // auto WaitForControlPeer() -> void;
auto AnnounceDataAddress() -> void; auto AnnounceDataAddress() -> void;
auto SendControlMessage(CtrlMsgPtr<ControlMessage> ctrl) -> void; auto ConnectControlEndpoint(Context::Address address) -> void;
auto ReceiveControlMessage() -> CtrlMsgPtr<ControlMessage>; auto BindControlEndpoint(Context::Address address) -> void;
auto ProcessControlMessage(CtrlMsgPtr<DataAddressAnnouncement> ctrl) -> void;
auto ConnectControlSocket(Context::Address address) -> void;
auto BindControlSocket(Context::Address address) -> void;
auto BindDataEndpoint() -> void; auto BindDataEndpoint() -> void;
auto ConnectDataEndpoint() -> void; auto ConnectDataEndpoint() -> void;
auto ReceiveDataAddressAnnouncement() -> void;
}; /* class Socket */ }; /* class Socket */
struct SilentSocketError : SocketError { using SocketError::SocketError; }; struct SilentSocketError : SocketError { using SocketError::SocketError; };

8
fairmq/ofi/TransportFactory.cxx
View File

@ -26,7 +26,7 @@ using namespace std;
TransportFactory::TransportFactory(const string& id, const FairMQProgOptions* /*config*/) TransportFactory::TransportFactory(const string& id, const FairMQProgOptions* /*config*/)
try : FairMQTransportFactory{id} try : FairMQTransportFactory{id}
{ {
LOG(debug) << "OFI transport: Using ZeroMQ (" << fContext.GetZmqVersion() << ") & " LOG(debug) << "OFI transport: Using AZMQ & "
<< "asiofi (" << fContext.GetAsiofiVersion() << ")"; << "asiofi (" << fContext.GetAsiofiVersion() << ")";
} }
catch (ContextError& e) catch (ContextError& e)
@ -65,19 +65,19 @@ auto TransportFactory::CreateSocket(const string& type, const string& name) -> S
return SocketPtr{new Socket(fContext, type, name, GetId(), this)}; return SocketPtr{new Socket(fContext, type, name, GetId(), this)};
} }
auto TransportFactory::CreatePoller(const vector<FairMQChannel>& channels) const -> PollerPtr auto TransportFactory::CreatePoller(const vector<FairMQChannel>& /*channels*/) const -> PollerPtr
{ {
throw runtime_error{"Not yet implemented (Poller)."}; throw runtime_error{"Not yet implemented (Poller)."};
// return PollerPtr{new Poller(channels)}; // return PollerPtr{new Poller(channels)};
} }
auto TransportFactory::CreatePoller(const vector<FairMQChannel*>& channels) const -> PollerPtr auto TransportFactory::CreatePoller(const vector<FairMQChannel*>& /*channels*/) const -> PollerPtr
{ {
throw runtime_error{"Not yet implemented (Poller)."}; throw runtime_error{"Not yet implemented (Poller)."};
// return PollerPtr{new Poller(channels)}; // return PollerPtr{new Poller(channels)};
} }
auto TransportFactory::CreatePoller(const unordered_map<string, vector<FairMQChannel>>& channelsMap, const vector<string>& channelList) const -> PollerPtr auto TransportFactory::CreatePoller(const unordered_map<string, vector<FairMQChannel>>& /*channelsMap*/, const vector<string>& /*channelList*/) const -> PollerPtr
{ {
throw runtime_error{"Not yet implemented (Poller)."}; throw runtime_error{"Not yet implemented (Poller)."};
// return PollerPtr{new Poller(channelsMap, channelList)}; // return PollerPtr{new Poller(channelsMap, channelList)};