Source · fairmq/
TransportFactory.hpublic header
Defines class fair::mq::TransportFactory, struct fair::mq::TransportFactoryError.
Described in Messages and multipart, Transports, region: unmanaged regions, Use channels without a device, Use std::pmr memory resources, Poll several channels, Send your own data types, Test your devices, Overview.
1/********************************************************************************2 * Copyright (C) 2021-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *3 * *4 * This software is distributed under the terms of the *5 * GNU Lesser General Public Licence (LGPL) version 3, *6 * copied verbatim in the file "LICENSE" *7 ********************************************************************************/8 9#ifndef FAIR_MQ_TRANSPORTFACTORY_H10#define FAIR_MQ_TRANSPORTFACTORY_H11 12#include <cstddef> // size_t13#include <fairmq/MemoryResources.h>14#include <fairmq/Message.h>15#include <fairmq/Poller.h>16#include <fairmq/Socket.h>17#include <fairmq/TransportEnum.h>18#include <fairmq/UnmanagedRegion.h>19#include <memory> // shared_ptr20#include <stdexcept>21#include <string>22#include <unordered_map>23#include <vector>24 25namespace fair::mq {26 27class Channel;28class ProgOptions;29 30class TransportFactory31{32 private:33 /// Topology wide unique id34 const std::string fkId;35 36 /// The polymorphic memory resource associated with the transport37 ChannelResource fMemoryResource{this};38 39 public:40 /// ctor41 /// @param id Topology wide unique id, usually the device id.42 TransportFactory(std::string id)43 : fkId(std::move(id))44 {}45 46 TransportFactory(const TransportFactory&) = delete;47 TransportFactory(TransportFactory&&) = delete;48 TransportFactory& operator=(const TransportFactory&) = delete;49 TransportFactory& operator=(TransportFactory&&) = delete;50 51 auto GetId() const -> const std::string { return fkId; };52 53 /// Get a pointer to the associated polymorphic memory resource54 ChannelResource* GetMemoryResource() { return &fMemoryResource; }55 operator ChannelResource*() { return &fMemoryResource; }56 57 /// @brief Create empty Message (for receiving)58 /// @return pointer to Message59 virtual MessagePtr CreateMessage() = 0;60 /// @brief Create empty Message (for receiving), align received buffer to specified alignment61 /// @param alignment alignment to align received buffer to62 /// @return pointer to Message63 virtual MessagePtr CreateMessage(Alignment alignment) = 0;64 /// @brief Create new Message of specified size65 /// @param size message size66 /// @return pointer to Message67 virtual MessagePtr CreateMessage(size_t size) = 0;68 /// @brief Create new Message of specified size and alignment69 /// @param size message size70 /// @param alignment message alignment71 /// @return pointer to Message72 virtual MessagePtr CreateMessage(size_t size, Alignment alignment) = 0;73 /// @brief Create new Message with user provided buffer and size74 /// @param data pointer to user provided buffer75 /// @param size size of the user provided buffer76 /// @param ffn callback, called when the message is transfered (and can be deleted)77 /// @param obj optional helper pointer that can be used in the callback78 /// @return pointer to Message79 virtual MessagePtr CreateMessage(void* data,80 size_t size,81 FreeFn* ffn,82 void* hint = nullptr) = 0;83 /// @brief create a message with the buffer located within the corresponding unmanaged region84 /// @param unmanagedRegion the unmanaged region that this message buffer belongs to85 /// @param data message buffer (must be within the region - checked at runtime by the transport)86 /// @param size size of the message87 /// @param hint optional parameter, returned to the user in the RegionCallback88 virtual MessagePtr CreateMessage(UnmanagedRegionPtr& unmanagedRegion,89 void* data,90 size_t size,91 void* hint = nullptr) = 0;92 93 /// @brief Create a socket94 virtual SocketPtr CreateSocket(const std::string& type, const std::string& name) = 0;95 96 /// @brief Create a poller for a single channel (all subchannels)97 virtual PollerPtr CreatePoller(const std::vector<Channel>& channels) const = 0;98 /// @brief Create a poller for specific channels99 virtual PollerPtr CreatePoller(const std::vector<Channel*>& channels) const = 0;100 /// @brief Create a poller for specific channels (all subchannels)101 virtual PollerPtr CreatePoller(102 const std::unordered_map<std::string, std::vector<Channel>>& channelsMap,103 const std::vector<std::string>& channelList) const = 0;104 105 /// @brief Create new UnmanagedRegion106 /// @param size size of the region107 /// @param callback callback to be called when a message belonging to this region is no longer108 /// needed by the transport109 /// @param path optional parameter to pass to the underlying transport110 /// @param flags optional parameter to pass to the underlying transport111 /// @return pointer to UnmanagedRegion112 // [[deprecated("Use CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg)")]]113 virtual UnmanagedRegionPtr CreateUnmanagedRegion(size_t size,114 RegionCallback callback = nullptr,115 const std::string& path = "",116 int flags = 0,117 RegionConfig cfg = RegionConfig()) = 0;118 // [[deprecated("Use CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg)")]]119 virtual UnmanagedRegionPtr CreateUnmanagedRegion(size_t size,120 RegionBulkCallback bulkCallback = nullptr,121 const std::string& path = "",122 int flags = 0,123 RegionConfig cfg = RegionConfig()) = 0;124 /// @brief Create new UnmanagedRegion125 /// @param size size of the region126 /// @param userFlags flags to be stored with the region, have no effect on the transport, but127 /// can be retrieved from the region by the user128 /// @param callback callback to be called when a message belonging to this region is no longer129 /// needed by the transport130 /// @param path optional parameter to pass to the underlying transport131 /// @param flags optional parameter to pass to the underlying transport132 /// @return pointer to UnmanagedRegion133 // [[deprecated("Use CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg)")]]134 virtual UnmanagedRegionPtr CreateUnmanagedRegion(size_t size,135 int64_t userFlags,136 RegionCallback callback = nullptr,137 const std::string& path = "",138 int flags = 0,139 RegionConfig cfg = RegionConfig()) = 0;140 // [[deprecated("Use CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg)")]]141 virtual UnmanagedRegionPtr CreateUnmanagedRegion(size_t size,142 int64_t userFlags,143 RegionBulkCallback bulkCallback = nullptr,144 const std::string& path = "",145 int flags = 0,146 RegionConfig cfg = RegionConfig()) = 0;147 148 /// @brief Create new UnmanagedRegion149 /// @param size size of the region150 /// @param callback callback to be called when a message belonging to this region is no longer needed by the transport151 /// @param cfg region configuration152 /// @return pointer to UnmanagedRegion153 virtual UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionCallback callback, RegionConfig cfg) = 0;154 155 /// @brief Create new UnmanagedRegion156 /// @param size size of the region157 /// @param bulkCallback callback to be called when message(s) belonging to this region is no longer needed by the transport158 /// @param cfg region configuration159 /// @return pointer to UnmanagedRegion160 virtual UnmanagedRegionPtr CreateUnmanagedRegion(size_t size, RegionBulkCallback bulkCallback, RegionConfig cfg) = 0;161 162 /// @brief Subscribe to region events (creation, destruction, ...)163 /// @param callback the callback that is called when a region event occurs164 virtual void SubscribeToRegionEvents(RegionEventCallback callback) = 0;165 /// @brief Check if there is an active subscription to region events166 /// @return true/false167 virtual bool SubscribedToRegionEvents() = 0;168 /// @brief Unsubscribe from region events169 virtual void UnsubscribeFromRegionEvents() = 0;170 171 virtual std::vector<RegionInfo> GetRegionInfo() = 0;172 173 /// Get transport type174 virtual Transport GetType() const = 0;175 176 virtual void Interrupt() = 0;177 virtual void Resume() = 0;178 virtual void Reset() = 0;179 180 virtual ~TransportFactory() = default;181 182 static auto CreateTransportFactory(const std::string& type,183 const std::string& id = "",184 const ProgOptions* config = nullptr)185 -> std::shared_ptr<TransportFactory>;186 187 static void NoCleanup(void* /*data*/, void* /*obj*/) {}188 189 template<typename T>190 static void SimpleMsgCleanup(void* /*data*/, void* obj)191 {192 delete static_cast<T*>(obj);193 }194 195 template<typename T>196 MessagePtr NewSimpleMessage(const T& data)197 {198 // todo: is_trivially_copyable not available on gcc < 5, workaround?199 // static_assert(std::is_trivially_copyable<T>::value, "The argument type for200 // NewSimpleMessage has to be trivially copyable!");201 T* dataCopy = new T(data);202 return CreateMessage(dataCopy, sizeof(T), SimpleMsgCleanup<T>, dataCopy);203 }204 205 template<std::size_t N>206 MessagePtr NewSimpleMessage(const char (&data)[N])207 {208 std::string* msgStr = new std::string(data);209 return CreateMessage(const_cast<char*>(msgStr->c_str()),210 msgStr->length(),211 SimpleMsgCleanup<std::string>,212 msgStr);213 }214 215 MessagePtr NewSimpleMessage(const std::string& str)216 {217 218 std::string* msgStr = new std::string(str);219 return CreateMessage(const_cast<char*>(msgStr->c_str()),220 msgStr->length(),221 SimpleMsgCleanup<std::string>,222 msgStr);223 }224 225 template<typename T>226 MessagePtr NewStaticMessage(const T& data)227 {228 return CreateMessage(data, sizeof(T), NoCleanup, nullptr);229 }230 231 MessagePtr NewStaticMessage(const std::string& str)232 {233 return CreateMessage(const_cast<char*>(str.c_str()), str.length(), NoCleanup, nullptr);234 }235};236 237struct TransportFactoryError : std::runtime_error238{239 using std::runtime_error::runtime_error;240};241 242} // namespace fair::mq243 244using FairMQTransportFactory [[deprecated("Use fair::mq::TransportFactory")]] = fair::mq::TransportFactory;245using FairMQTransportFactoryError [[deprecated("Use fair::mq::TransportFactoryError")]] = fair::mq::TransportFactoryError;246 247#endif // FAIR_MQ_TRANSPORTFACTORY_H