Mirror/FairMQ
1
0
Fork 0
mirror of https://github.com/FairRootGroup/FairMQ.git synced 2025-10-13 00:31:14 +00:00
Code Issues Packages Projects Releases Wiki Activity
9a9c863de1
FairMQ/test/region/_region.cxx

374 lines
12 KiB
C++
Raw Blame History

/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/shmem/Common.h>
#include <fairmq/shmem/Monitor.h>
#include <fairmq/shmem/Segment.h>
#include <fairmq/shmem/UnmanagedRegion.h>
#include <fairmq/TransportFactory.h>
#include <fairmq/ProgOptions.h>
#include <fairmq/tools/Unique.h>
#include <fairmq/tools/Semaphore.h>
#include <fairmq/tools/Strings.h>
#include <fairlogger/Logger.h>
#include <gtest/gtest.h>
#include <cstdint>
#include <map>
#include <memory> // make_unique
#include <string>
#include <utility> // pair
#include <vector> // pair
namespace
{
using namespace std;
using namespace fair::mq;
struct ShmOwner
{
ShmOwner(const string& sessionId,
const vector<pair<uint16_t, size_t>>& segments,
const vector<pair<uint16_t, size_t>>& regions)
: fShmId(fair::mq::shmem::makeShmIdStr(sessionId))
{
LOG(info) << "ShmOwner: creating segments";
for (auto [id, size] : segments) {
fSegments.emplace(id, fair::mq::shmem::Segment(fShmId, id, size, fair::mq::shmem::rbTreeBestFit));
}
LOG(info) << "ShmOwner: creating regions";
for (auto [id, size] : regions) {
fRegions.emplace(id, make_unique<fair::mq::shmem::UnmanagedRegion>(fShmId, id, size));
}
}
~ShmOwner()
{
LOG(info) << "ShmOwner: cleaning up";
fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{fShmId});
}
string fShmId;
map<uint16_t, fair::mq::shmem::Segment> fSegments;
map<uint16_t, unique_ptr<fair::mq::shmem::UnmanagedRegion>> fRegions;
};
void RegionsSizeMismatch()
{
size_t session = tools::UuidHash();
ProgOptions config;
config.SetProperty<string>("session", to_string(session));
config.SetProperty<size_t>("shm-segment-size", 100000000);
auto factory = TransportFactory::CreateTransportFactory("shmem", tools::Uuid(), &config);
fair::mq::RegionConfig rCfg;
rCfg.id = 10;
UnmanagedRegionPtr region1 = nullptr;
ASSERT_NO_THROW(region1 = factory->CreateUnmanagedRegion(10000, [](void*, size_t, void*) {}, rCfg));
ASSERT_NE(region1, nullptr);
UnmanagedRegionPtr region2 = nullptr;
ASSERT_THROW(region2 = factory->CreateUnmanagedRegion(16000, [](void*, size_t, void*) {}, rCfg), fair::mq::TransportError);
ASSERT_EQ(region2, nullptr);
}
void RegionsCache(const string& transport, const string& _address)
{
std::string address(tools::ToString(_address, "_", transport));
size_t session1 = tools::UuidHash();
size_t session2 = tools::UuidHash();
ProgOptions config1;
ProgOptions config2;
config1.SetProperty<string>("session", to_string(session1));
config1.SetProperty<size_t>("shm-segment-size", 100000000);
config2.SetProperty<string>("session", to_string(session2));
config2.SetProperty<size_t>("shm-segment-size", 100000000);
auto factory1 = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config1);
auto factory2 = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config2);
auto region1 = factory1->CreateUnmanagedRegion(1000000, [](void*, size_t, void*) {});
auto region2 = factory2->CreateUnmanagedRegion(1000000, [](void*, size_t, void*) {});
void* r1ptr = region1->GetData();
void* r2ptr = region2->GetData();
Channel push1("Push1", "push", factory1);
Channel pull1("Pull1", "pull", factory1);
push1.Bind(address + to_string(1));
pull1.Connect(address + to_string(1));
Channel push2("Push2", "push", factory2);
Channel pull2("Pull2", "pull", factory2);
push2.Bind(address + to_string(2));
pull2.Connect(address + to_string(2));
{
static_cast<char*>(r1ptr)[0] = 97; // a
static_cast<char*>(static_cast<char*>(r1ptr) + 100)[0] = 98; // b
static_cast<char*>(r2ptr)[0] = 99; // c
static_cast<char*>(static_cast<char*>(r2ptr) + 100)[0] = 100; // d
MessagePtr m1(push1.NewMessage(region1, r1ptr, 100, nullptr));
MessagePtr m2(push1.NewMessage(region1, static_cast<char*>(r1ptr) + 100, 100, nullptr));
push1.Send(m1);
push1.Send(m2);
MessagePtr m3(push2.NewMessage(region2, r2ptr, 100, nullptr));
MessagePtr m4(push2.NewMessage(region2, static_cast<char*>(r2ptr) + 100, 100, nullptr));
push2.Send(m3);
push2.Send(m4);
}
{
MessagePtr m1(pull1.NewMessage());
MessagePtr m2(pull1.NewMessage());
ASSERT_EQ(pull1.Receive(m1), 100);
ASSERT_EQ(pull1.Receive(m2), 100);
ASSERT_EQ(static_cast<char*>(m1->GetData())[0], 'a');
ASSERT_EQ(static_cast<char*>(m2->GetData())[0], 'b');
MessagePtr m3(pull2.NewMessage());
MessagePtr m4(pull2.NewMessage());
ASSERT_EQ(pull2.Receive(m3), 100);
ASSERT_EQ(pull2.Receive(m4), 100);
ASSERT_EQ(static_cast<char*>(m3->GetData())[0], 'c');
ASSERT_EQ(static_cast<char*>(m4->GetData())[0], 'd');
}
}
void RegionEventSubscriptions(const string& transport, bool external)
{
fair::Logger::SetConsoleSeverity(fair::Severity::debug);
unique_ptr<ShmOwner> shmOwner = nullptr;
size_t session{tools::UuidHash()};
constexpr int sSize = 100000000;
constexpr int r1Size = 1000000;
constexpr int r2Size = 5000000;
constexpr uint16_t sId = 0;
constexpr uint16_t r1id = 100;
constexpr uint16_t r2id = 101;
if (external) {
shmOwner = make_unique<ShmOwner>(
to_string(session),
vector<pair<uint16_t, size_t>>{ { sId, sSize } },
vector<pair<uint16_t, size_t>>{ { r1id, r1Size }, { r2id, r2Size } }
);
}
ProgOptions config;
config.SetProperty<string>("session", to_string(session));
config.SetProperty<size_t>("shm-segment-size", sSize);
if (external) {
config.SetProperty<bool>("shm-no-cleanup", true);
config.SetProperty<bool>("shm-monitor", false);
}
auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
constexpr int64_t userFlags = 12345;
tools::Semaphore blocker;
{
fair::mq::RegionConfig r1Cfg;
if (external) {
r1Cfg.id = r1id;
r1Cfg.removeOnDestruction = false;
}
auto region1 = factory->CreateUnmanagedRegion(r1Size, [](void*, size_t, void*) {}, r1Cfg);
void* ptr1 = region1->GetData();
uint64_t id1 = region1->GetId();
if (external) {
ASSERT_EQ(id1, r1id);
}
ASSERT_EQ(region1->GetSize(), r1Size);
fair::mq::RegionConfig r2Cfg;
r2Cfg.userFlags = userFlags;
if (external) {
r2Cfg.id = r2id;
r2Cfg.removeOnDestruction = false;
}
auto region2 = factory->CreateUnmanagedRegion(r2Size, [](void*, size_t, void*) {}, r2Cfg);
void* ptr2 = region2->GetData();
uint64_t id2 = region2->GetId();
if (external) {
ASSERT_EQ(id2, r2id);
}
ASSERT_EQ(region2->GetSize(), r2Size);
ASSERT_EQ(factory->SubscribedToRegionEvents(), false);
factory->SubscribeToRegionEvents([&, id1, id2, ptr1, ptr2](RegionInfo info) {
LOG(info) << ">>> " << info.event << ": "
<< (info.managed ? "managed" : "unmanaged")
<< ", id: " << info.id
<< ", ptr: " << info.ptr
<< ", size: " << info.size
<< ", flags: " << info.flags;
if (info.event == RegionEvent::created) {
if (info.id == id1) {
ASSERT_EQ(info.size, r1Size);
ASSERT_EQ(info.ptr, ptr1);
blocker.Signal();
} else if (info.id == id2) {
ASSERT_EQ(info.size, r2Size);
ASSERT_EQ(info.ptr, ptr2);
if (!external) {
ASSERT_EQ(info.flags, userFlags);
}
blocker.Signal();
}
} else if (info.event == RegionEvent::destroyed) {
if (info.id == id1) {
blocker.Signal();
} else if (info.id == id2) {
blocker.Signal();
}
}
});
ASSERT_EQ(factory->SubscribedToRegionEvents(), true);
LOG(info) << "waiting for blockers...";
blocker.Wait();
LOG(info) << "1 done.";
blocker.Wait();
LOG(info) << "2 done.";
}
if (!external) {
blocker.Wait();
LOG(info) << "3 done.";
blocker.Wait();
LOG(info) << "4 done.";
}
LOG(info) << "All done.";
factory->UnsubscribeFromRegionEvents();
ASSERT_EQ(factory->SubscribedToRegionEvents(), false);
}
void RegionCallbacks(const string& transport, const string& _address)
{
size_t session(tools::UuidHash());
std::string address(tools::ToString(_address, "_", transport));
constexpr size_t sSize = 100000000;
constexpr size_t r1Size = 2000000;
constexpr size_t r2Size = 3000000;
ProgOptions config;
config.SetProperty<string>("session", to_string(session));
config.SetProperty<size_t>("shm-segment-size", sSize);
auto factory = TransportFactory::CreateTransportFactory(transport, tools::Uuid(), &config);
unique_ptr<int> intPtr1 = make_unique<int>(42);
unique_ptr<int> intPtr2 = make_unique<int>(43);
tools::Semaphore blocker;
Channel push("Push", "push", factory);
push.Bind(address);
Channel pull("Pull", "pull", factory);
pull.Connect(address);
void* ptr1 = nullptr;
size_t size1 = 100;
void* ptr2 = nullptr;
size_t size2 = 200;
auto region1 = factory->CreateUnmanagedRegion(r1Size, [&](void* ptr, size_t size, void* hint) {
ASSERT_EQ(ptr, ptr1);
ASSERT_EQ(size, size1);
ASSERT_EQ(hint, intPtr1.get());
ASSERT_EQ(*static_cast<int*>(hint), 42);
blocker.Signal();
});
ptr1 = region1->GetData();
auto region2 = factory->CreateUnmanagedRegion(r2Size, [&](const std::vector<RegionBlock>& blocks) {
ASSERT_EQ(blocks.size(), 1);
ASSERT_EQ(blocks.at(0).ptr, ptr2);
ASSERT_EQ(blocks.at(0).size, size2);
ASSERT_EQ(blocks.at(0).hint, intPtr2.get());
ASSERT_EQ(*static_cast<int*>(blocks.at(0).hint), 43);
blocker.Signal();
});
ptr2 = region2->GetData();
{
MessagePtr msg1out(push.NewMessage(region1, ptr1, size1, intPtr1.get()));
MessagePtr msg2out(push.NewMessage(region2, ptr2, size2, intPtr2.get()));
ASSERT_EQ(push.Send(msg1out), size1);
ASSERT_EQ(push.Send(msg2out), size2);
}
{
MessagePtr msg1in(pull.NewMessage());
MessagePtr msg2in(pull.NewMessage());
ASSERT_EQ(pull.Receive(msg1in), size1);
ASSERT_EQ(pull.Receive(msg2in), size2);
}
LOG(info) << "waiting for blockers...";
blocker.Wait();
LOG(info) << "1 done.";
blocker.Wait();
LOG(info) << "2 done.";
}
TEST(RegionsSizeMismatch, shmem)
{
RegionsSizeMismatch();
}
TEST(Cache, zeromq)
{
RegionsCache("zeromq", "ipc://test_region_cache");
}
TEST(Cache, shmem)
{
RegionsCache("shmem", "ipc://test_region_cache");
}
TEST(EventSubscriptions, zeromq)
{
RegionEventSubscriptions("zeromq", false);
}
TEST(EventSubscriptions, shmem)
{
RegionEventSubscriptions("shmem", false);
}
TEST(Callbacks, zeromq)
{
RegionCallbacks("zeromq", "ipc://test_region_callbacks");
}
TEST(Callbacks, shmem)
{
RegionCallbacks("shmem", "ipc://test_region_callbacks");
}
TEST(EventSubscriptionsExternalRegion, shmem)
{
RegionEventSubscriptions("shmem", true);
}
} // namespace
Reference in New Issue View Git Blame Copy Permalink