Source · fairmq/shmem/
UnmanagedRegion.hpublic header
Defines struct fair::mq::shmem::UnmanagedRegion.
Described in region: unmanaged regions, Unmanaged regions.
1/********************************************************************************2 * Copyright (C) 2014-2023 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_SHMEM_UNMANAGEDREGION_H_10#define FAIR_MQ_SHMEM_UNMANAGEDREGION_H_11 12#include <fairmq/shmem/Common.h>13#include <fairmq/shmem/Monitor.h>14#include <fairmq/tools/Strings.h>15#include <fairmq/UnmanagedRegion.h>16#include <fairmq/Transports.h>17 18#include <fairlogger/Logger.h>19 20#include <boost/date_time/posix_time/posix_time.hpp>21#include <boost/interprocess/managed_shared_memory.hpp>22#include <boost/interprocess/file_mapping.hpp>23#include <boost/interprocess/ipc/message_queue.hpp>24 25#include <algorithm> // min26#include <atomic>27#include <thread>28#include <memory> // make_unique29#include <mutex>30#include <condition_variable>31#include <unordered_map>32#include <cerrno>33#include <chrono>34#include <ios>35#include <utility> // move36 37namespace fair::mq::shmem38{39 40struct UnmanagedRegion41{42 friend class Message;43 friend class Manager;44 friend class Monitor;45 46 UnmanagedRegion(const std::string& shmId, uint16_t id, uint64_t size)47 : UnmanagedRegion(shmId, size, true, makeRegionConfig(id))48 {}49 50 UnmanagedRegion(const std::string& shmId, uint64_t size, RegionConfig cfg)51 : UnmanagedRegion(shmId, size, true, std::move(cfg))52 {}53 54 UnmanagedRegion(const std::string& shmId, RegionConfig cfg)55 : UnmanagedRegion(shmId, cfg.size, true, std::move(cfg))56 {}57 58 UnmanagedRegion(const std::string& shmId, uint64_t size, bool controlling, RegionConfig cfg)59 : fControlling(controlling)60 , fRemoveOnDestruction(cfg.removeOnDestruction)61 , fLinger(cfg.linger)62 , fStopAcks(false)63 , fName(MakeShmName(shmId, "rg", cfg.id.value()))64 , fQueueName(MakeShmName(shmId, "rgq", cfg.id.value()))65 , fRefCountSegmentName(MakeShmName(shmId, "rrc", cfg.id.value()))66 , fFile(nullptr)67 , fRcSegmentSize(cfg.rcSegmentSize)68 , fQueue(nullptr)69 , fCallback(nullptr)70 , fBulkCallback(nullptr)71 {72 using namespace boost::interprocess;73 74 // TODO: refactor this75 cfg.size = size;76 const uint16_t id = cfg.id.value();77 bool created = false;78 79 LOG(debug) << "UnmanagedRegion(): " << fName << " (" << (fControlling ? "controller" : "viewer") << ")";80 81 if (!cfg.path.empty()) {82 fName = std::string(cfg.path + fName);83 84 if (fControlling) {85 // create a file86 std::filebuf fbuf;87 if (fbuf.open(fName, std::ios_base::in | std::ios_base::out | std::ios_base::trunc | std::ios_base::binary)) {88 // set the size89 fbuf.pubseekoff(size - 1, std::ios_base::beg);90 fbuf.sputc(0);91 }92 }93 94 fFile = fopen(fName.c_str(), "r+");95 96 if (!fFile) {97 LOG(error) << "Failed to initialize file: " << fName;98 LOG(error) << "errno: " << errno << ": " << strerror(errno);99 throw TransportError(tools::ToString("Failed to initialize file for shared memory region: ", strerror(errno)));100 }101 fFileMapping = file_mapping(fName.c_str(), read_write);102 LOG(debug) << "UnmanagedRegion(): initialized file: " << fName;103 fRegion = mapped_region(fFileMapping, read_write, 0, size, nullptr, cfg.creationFlags);104 } else {105 try {106 // if opening fails, create107 try {108 fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write);109 created = false;110 } catch (interprocess_exception& e) {111 if (fControlling) {112 LOG(debug) << "Could not open controlling shared_memory_object for region " << id << ": " << e.what() << ", creating...";113 fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write);114 fShmemObject.truncate(size);115 created = true;116 } else {117 LOG(error) << "Could not open view for shared_memory_object for region " << id << ": " << e.what();118 throw TransportError(tools::ToString("Could not open view for shared_memory_object for region ", id, ": ", e.what()));119 }120 }121 } catch (interprocess_exception& e) {122 LOG(error) << "Failed initializing shared_memory_object for region id " << id << ": " << e.what();123 throw;124 }125 126 try {127 fRegion = mapped_region(fShmemObject, read_write, 0, 0, nullptr, cfg.creationFlags);128 if (size != 0 && size != fRegion.get_size()) {129 LOG(error) << "Created/opened region size (" << fRegion.get_size() << ") does not match configured size (" << size << ")";130 throw TransportError(tools::ToString("Created/opened region size (", fRegion.get_size(), ") does not match configured size (", size, ")"));131 }132 } catch (interprocess_exception& e) {133 LOG(error) << "Failed mapping shared_memory_object for region id " << id << ": " << e.what();134 throw;135 }136 }137 138 if (cfg.lock) {139 LOG(debug) << "Locking region " << id << "...";140 Lock();141 LOG(debug) << "Successfully locked region " << id << ".";142 }143 if (cfg.zero) {144 LOG(debug) << "Zeroing free memory of region " << id << "...";145 Zero();146 LOG(debug) << "Successfully zeroed free memory of region " << id << ".";147 }148 149 InitializeRefCountSegment(fRcSegmentSize);150 151 if (fControlling && created) {152 Register(shmId, cfg);153 }154 155 LOG(debug) << (created ? "Created" : "Opened") << " unmanaged shared memory region: " << fName << " (" << (fControlling ? "controller" : "viewer") << "), refCount segment size: " << fRcSegmentSize;156 }157 158 UnmanagedRegion() = delete;159 160 UnmanagedRegion(const UnmanagedRegion&) = delete;161 UnmanagedRegion(UnmanagedRegion&&) = delete;162 UnmanagedRegion& operator=(const UnmanagedRegion&) = delete;163 UnmanagedRegion& operator=(UnmanagedRegion&&) = delete;164 165 void BecomeController(RegionConfig& cfg)166 {167 fControlling = true;168 fLinger = cfg.linger;169 fRemoveOnDestruction = cfg.removeOnDestruction;170 }171 172 void Zero()173 {174 memset(fRegion.get_address(), 0x00, fRegion.get_size());175 }176 void Lock()177 {178 if (mlock(fRegion.get_address(), fRegion.get_size()) == -1) {179 LOG(error) << "Could not lock region " << fName << ". Code: " << errno << ", reason: " << strerror(errno);180 throw TransportError(tools::ToString("Could not lock region ", fName, ": ", strerror(errno)));181 }182 }183 184 void* GetData() const { return fRegion.get_address(); }185 size_t GetSize() const { return fRegion.get_size(); }186 187 void SetLinger(uint32_t linger) { fLinger = linger; }188 uint32_t GetLinger() const { return fLinger; }189 190 bool RemoveOnDestruction() { return fRemoveOnDestruction; }191 192 RefCount& MakeRefCount(uint16_t initialCount = 1)193 {194 RefCount* refCount = fRefCountPool->allocate_one().get();195 new (refCount) RefCount(initialCount);196 return *refCount;197 }198 199 void RemoveRefCount(RefCount& refCount)200 {201 refCount.~RefCount();202 fRefCountPool->deallocate_one(&refCount);203 }204 205 ~UnmanagedRegion()206 {207 LOG(debug) << "~UnmanagedRegion(): " << fName << " (" << (fControlling ? "controller" : "viewer") << ")";208 SetStopAcks();209 210 if (fAcksSender.joinable()) {211 fAcksSender.join();212 }213 214 if (fControlling) {215 if (fAcksReceiver.joinable()) {216 fAcksReceiver.join();217 }218 219 if (fRemoveOnDestruction) {220 if (Monitor::RemoveObject(fName.c_str())) {221 LOG(trace) << "Region '" << fName << "' destroyed.";222 }223 if (Monitor::RemoveFileMapping(fName.c_str())) {224 LOG(trace) << "File mapping '" << fName << "' destroyed.";225 }226 if (fRefCountSegment) {227 if (Monitor::RemoveObject(fRefCountSegmentName)) {228 LOG(trace) << "Ref Count Segment '" << fRefCountSegmentName << "' destroyed.";229 }230 }231 } else {232 LOG(debug) << "Skipping removal of " << fName << " unmanaged region, because RegionConfig::removeOnDestruction is false";233 }234 235 if (boost::interprocess::message_queue::remove(fQueueName.c_str())) {236 LOG(trace) << "Region queue '" << fQueueName << "' destroyed.";237 } else {238 LOG(debug) << "Region queue '" << fQueueName << "' not destroyed.";239 }240 241 if (fFile) {242 fclose(fFile);243 }244 } else {245 // LOG(debug) << "Region queue '" << fQueueName << "' is viewer, no cleanup necessary";246 }247 248 // LOG(debug) << "Region '" << fName << "' (" << (fControlling ? "controller" : "viewer") << ") destructed.";249 }250 251 private:252 bool fControlling;253 bool fRemoveOnDestruction;254 uint32_t fLinger;255 std::atomic<bool> fStopAcks;256 std::string fName;257 std::string fQueueName;258 std::string fRefCountSegmentName;259 boost::interprocess::shared_memory_object fShmemObject;260 FILE* fFile;261 boost::interprocess::file_mapping fFileMapping;262 boost::interprocess::mapped_region fRegion;263 264 std::mutex fBlockMtx;265 std::condition_variable fBlockSendCV;266 std::vector<RegionBlock> fBlocksToFree;267 const std::size_t fAckBunchSize = 256;268 uint64_t fRcSegmentSize;269 std::unique_ptr<boost::interprocess::message_queue> fQueue;270 std::unique_ptr<boost::interprocess::managed_shared_memory> fRefCountSegment;271 std::unique_ptr<RefCountPool> fRefCountPool;272 273 std::thread fAcksReceiver;274 std::thread fAcksSender;275 RegionCallback fCallback;276 RegionBulkCallback fBulkCallback;277 278 static RegionConfig makeRegionConfig(uint16_t id)279 {280 RegionConfig regionCfg;281 regionCfg.id = id;282 return regionCfg;283 }284 285 static void Register(const std::string& shmId, const RegionConfig& cfg)286 {287 using namespace boost::interprocess;288 LOG(debug) << "Registering unmanaged shared memory region with id " << cfg.id.value();289 managed_shared_memory mngSegment(open_or_create, MakeShmName(shmId, "mng").c_str(), kManagementSegmentSize);290 VoidAlloc alloc(mngSegment.get_segment_manager());291 292 Uint16RegionInfoHashMap* shmRegions = mngSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(alloc);293 294 EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);295 296 auto it = shmRegions->find(cfg.id.value());297 if (it != shmRegions->end()) {298 LOG(error) << "Unmanaged Region with id " << cfg.id.value() << " has already been registered. Only unique IDs per session are allowed.";299 throw TransportError(tools::ToString("Unmanaged Region with id ", cfg.id.value(), " has already been registered. Only unique IDs per session are allowed."));300 }301 302 shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, cfg.rcSegmentSize, alloc));303 (eventCounter->fCount)++;304 }305 306 void SetCallbacks(RegionCallback callback, RegionBulkCallback bulkCallback)307 {308 fCallback = std::move(callback);309 fBulkCallback = std::move(bulkCallback);310 }311 312 void InitializeQueues()313 {314 using namespace boost::interprocess;315 if (!fQueue) {316 fQueue = std::make_unique<message_queue>(open_or_create, fQueueName.c_str(), 1024, fAckBunchSize * sizeof(RegionBlock));317 LOG(trace) << "shmem: initialized region queue: " << fQueueName;318 }319 }320 321 void InitializeRefCountSegment(uint64_t size)322 {323 using namespace boost::interprocess;324 if (!fRefCountSegment && size > 0) {325 fRefCountSegment = std::make_unique<managed_shared_memory>(open_or_create, fRefCountSegmentName.c_str(), size);326 LOG(trace) << "shmem: initialized ref count segment: " << fRefCountSegmentName;327 fRefCountPool = std::make_unique<RefCountPool>(fRefCountSegment->get_segment_manager());328 }329 }330 331 RefCount* GetRefCountAddressFromHandle(const boost::interprocess::managed_shared_memory::handle_t handle)332 {333 if (fRefCountPool) {334 return reinterpret_cast<RefCount*>(fRefCountSegment->get_address_from_handle(handle));335 }336 return nullptr;337 };338 339 boost::interprocess::managed_shared_memory::handle_t HandleFromAddress(const void* ptr)340 {341 return fRefCountSegment->get_handle_from_address(ptr);342 }343 344 void StartAckSender()345 {346 if (!fAcksSender.joinable()) {347 fAcksSender = std::thread(&UnmanagedRegion::SendAcks, this);348 }349 }350 void SendAcks()351 {352 std::unique_ptr<RegionBlock[]> blocks = std::make_unique<RegionBlock[]>(fAckBunchSize);353 size_t blocksToSend = 0;354 355 while (true) {356 {357 std::unique_lock<std::mutex> lock(fBlockMtx);358 359 // try to get <fAckBunchSize> blocks, but do not wait when stopping360 fBlockSendCV.wait_for(lock, std::chrono::milliseconds(500), [&] { return fBlocksToFree.size() >= fAckBunchSize || fStopAcks; });361 362 // send whatever blocks we have363 blocksToSend = std::min(fBlocksToFree.size(), fAckBunchSize);364 365 copy_n(fBlocksToFree.end() - blocksToSend, blocksToSend, blocks.get());366 fBlocksToFree.resize(fBlocksToFree.size() - blocksToSend);367 }368 369 if (blocksToSend > 0) {370 while (!fQueue->try_send(blocks.get(), blocksToSend * sizeof(RegionBlock), 0) && !fStopAcks) {371 // receiver slow? yield and try again...372 std::this_thread::yield();373 }374 // LOG(debug) << "Sent " << blocksToSend << " blocks.";375 } else { // blocksToSend == 0376 if (fStopAcks) {377 break;378 }379 }380 }381 382 LOG(trace) << "AcksSender for " << fName << " leaving " << "(blocks left to free: " << fBlocksToFree.size() << ", "383 << " blocks left to send: " << blocksToSend << ").";384 }385 386 void StartAckReceiver()387 {388 if (!fAcksReceiver.joinable()) {389 fAcksReceiver = std::thread(&UnmanagedRegion::ReceiveAcks, this);390 }391 }392 void ReceiveAcks()393 {394 unsigned int priority = 0;395 boost::interprocess::message_queue::size_type recvdSize = 0;396 std::unique_ptr<RegionBlock[]> blocks = std::make_unique<RegionBlock[]>(fAckBunchSize);397 std::vector<fair::mq::RegionBlock> result;398 result.reserve(fAckBunchSize);399 400 while (true) {401 uint32_t timeout = 100;402 bool leave = false;403 if (fStopAcks) {404 timeout = fLinger;405 leave = true;406 }407 auto rcvTill = boost::posix_time::microsec_clock::universal_time() + boost::posix_time::milliseconds(timeout);408 409 while (fQueue->timed_receive(blocks.get(), fAckBunchSize * sizeof(RegionBlock), recvdSize, priority, rcvTill)) {410 const auto numBlocks = recvdSize / sizeof(RegionBlock);411 // LOG(debug) << "Received " << numBlocks << " blocks (recvdSize: " << recvdSize << "). (remaining queue size: " << fQueue->get_num_msg() << ").";412 if (fBulkCallback) {413 result.clear();414 for (size_t i = 0; i < numBlocks; i++) {415 result.emplace_back(reinterpret_cast<char*>(fRegion.get_address()) + blocks[i].fHandle, blocks[i].fSize, reinterpret_cast<void*>(blocks[i].fHint));416 }417 fBulkCallback(result);418 } else if (fCallback) {419 for (size_t i = 0; i < numBlocks; i++) {420 fCallback(reinterpret_cast<char*>(fRegion.get_address()) + blocks[i].fHandle, blocks[i].fSize, reinterpret_cast<void*>(blocks[i].fHint));421 }422 }423 }424 425 if (leave) {426 break;427 }428 }429 430 LOG(trace) << "AcksReceiver for " << fName << " leaving (remaining queue size: " << fQueue->get_num_msg() << ").";431 }432 433 void ReleaseBlock(const RegionBlock& block)434 {435 std::unique_lock<std::mutex> lock(fBlockMtx);436 437 fBlocksToFree.emplace_back(block);438 439 if (fBlocksToFree.size() >= fAckBunchSize) {440 lock.unlock();441 fBlockSendCV.notify_one();442 }443 }444 445 void SetStopAcks()446 {447 {448 // set under the lock, so that the ack sender cannot miss the notification449 std::lock_guard<std::mutex> lock(fBlockMtx);450 fStopAcks = true;451 }452 fBlockSendCV.notify_one();453 }454 455 void StopAcks()456 {457 SetStopAcks();458 459 if (fAcksSender.joinable()) {460 fAcksSender.join();461 }462 463 if (fAcksReceiver.joinable()) {464 fAcksReceiver.join();465 }466 }467};468 469} // namespace fair::mq::shmem470 471#endif /* FAIR_MQ_SHMEM_UNMANAGEDREGION_H_ */