Skip to content

Commit 25df1a7

Browse files
lterraclterrac
committed
tests: add mpi tests for variable size spsc
1 parent 3c0a068 commit 25df1a7

10 files changed

Lines changed: 672 additions & 286 deletions

File tree

‎tests/frontends/channel/variableSize/spsc/include/channelFixture.hpp‎

Lines changed: 21 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -167,51 +167,45 @@ class ChannelFixture : public ::testing::Test
167167
channelCapacity);
168168
}
169169

170-
std::unique_ptr<HiCR::CommunicationManager> _communicationManager;
171-
std::unique_ptr<HiCR::InstanceManager> _instanceManager;
172-
std::unique_ptr<HiCR::MemoryManager> _memoryManager;
173-
std::unique_ptr<HiCR::TopologyManager> _topologyManager;
174-
std::unique_ptr<HiCR::backend::pthreads::ComputeManager> _computeManager;
170+
std::unique_ptr<HiCR::CommunicationManager> communicationManager;
171+
std::unique_ptr<HiCR::InstanceManager> instanceManager;
172+
std::unique_ptr<HiCR::MemoryManager> memoryManager;
173+
std::unique_ptr<HiCR::TopologyManager> topologyManager;
174+
std::unique_ptr<HiCR::backend::pthreads::ComputeManager> computeManager;
175175

176-
std::unique_ptr<HiCR::channel::variableSize::SPSC::Consumer> _consumer;
177-
std::unique_ptr<HiCR::channel::variableSize::SPSC::Producer> _producer;
176+
std::unique_ptr<HiCR::channel::variableSize::SPSC::Consumer> consumer;
177+
std::unique_ptr<HiCR::channel::variableSize::SPSC::Producer> producer;
178178

179-
std::shared_ptr<HiCR::MemorySpace> _memorySpace;
179+
std::shared_ptr<HiCR::MemorySpace> memorySpace;
180180

181181
protected:
182182

183183
void SetUp() override
184184
{
185-
_instanceManager = std::make_unique<HiCR::backend::mpi::InstanceManager>(MPI_COMM_WORLD);
185+
instanceManager = std::make_unique<HiCR::backend::mpi::InstanceManager>(MPI_COMM_WORLD);
186186

187187
// Sanity Check
188-
if (_instanceManager->getInstances().size() != 2)
188+
if (instanceManager->getInstances().size() != 2)
189189
{
190-
if (_instanceManager->getCurrentInstance()->isRootInstance()) fprintf(stderr, "Launch error: MPI process count must be equal to 2\n");
190+
if (instanceManager->getCurrentInstance()->isRootInstance()) fprintf(stderr, "Launch error: MPI process count must be equal to 2\n");
191191
MPI_Finalize();
192192
}
193193

194-
_communicationManager = std::make_unique<HiCR::backend::mpi::CommunicationManager>(MPI_COMM_WORLD);
195-
_memoryManager = std::make_unique<HiCR::backend::mpi::MemoryManager>();
196-
_computeManager = std::make_unique<HiCR::backend::pthreads::ComputeManager>();
197-
_topologyManager = HiCR::backend::hwloc::TopologyManager::createDefault();
194+
communicationManager = std::make_unique<HiCR::backend::mpi::CommunicationManager>(MPI_COMM_WORLD);
195+
memoryManager = std::make_unique<HiCR::backend::mpi::MemoryManager>();
196+
computeManager = std::make_unique<HiCR::backend::pthreads::ComputeManager>();
197+
topologyManager = HiCR::backend::hwloc::TopologyManager::createDefault();
198198

199-
_topology = _topologyManager->queryTopology();
200-
_memorySpace = _topology.getDevices().begin().operator*()->getMemorySpaceList().begin().operator*();
201-
202-
if (_instanceManager->getCurrentInstance()->isRootInstance())
203-
{
204-
_producer = createProducer(*_memoryManager, *_memoryManager, *_communicationManager, *_communicationManager, _memorySpace, _memorySpace, CHANNEL_CAPACITY);
205-
}
206-
else { _consumer = createConsumer(*_memoryManager, *_memoryManager, *_communicationManager, *_communicationManager, _memorySpace, _memorySpace, CHANNEL_CAPACITY); }
199+
_topology = topologyManager->queryTopology();
200+
memorySpace = _topology.getDevices().begin().operator*()->getMemorySpaceList().begin().operator*();
207201
}
208202

209203
void TearDown() override
210204
{
211-
for (auto &g : _globalSlots) { _communicationManager->deregisterGlobalMemorySlot(g); }
212-
for (auto &g : _globalSlotsToDestroy) { _communicationManager->destroyGlobalMemorySlot(g); }
213-
_communicationManager->fence(CHANNEL_TAG);
214-
for (auto &l : _localSlots) { _memoryManager->freeLocalMemorySlot(l); }
205+
for (auto &g : _globalSlots) { communicationManager->deregisterGlobalMemorySlot(g); }
206+
for (auto &g : _globalSlotsToDestroy) { communicationManager->destroyGlobalMemorySlot(g); }
207+
communicationManager->fence(CHANNEL_TAG);
208+
for (auto &l : _localSlots) { memoryManager->freeLocalMemorySlot(l); }
215209
}
216210

217211
private:

tests/frontends/channel/variableSize/spsc/source/common.hpp renamed to tests/frontends/channel/variableSize/spsc/include/common.hpp

File renamed without changes.

‎tests/frontends/channel/variableSize/spsc/source/consumer.hpp‎

Lines changed: 0 additions & 125 deletions
This file was deleted.

‎tests/frontends/channel/variableSize/spsc/source/excessBuffer.cpp‎

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
#include <fstream>
2+
#include <mpi.h>
3+
#include <nlohmann_json/json.hpp>
4+
5+
#include <hicr/backends/mpi/communicationManager.hpp>
6+
#include <hicr/backends/mpi/memoryManager.hpp>
7+
#include <hicr/backends/mpi/instanceManager.hpp>
8+
#include <hicr/backends/hwloc/topologyManager.hpp>
9+
10+
#include "../include/channelFixture.hpp"
11+
12+
void producerFc(ChannelFixture &fixture)
13+
{
14+
// Create producer and pick managers from the fixture
15+
fixture.producer = fixture.createProducer(
16+
*fixture.memoryManager, *fixture.memoryManager, *fixture.communicationManager, *fixture.communicationManager, fixture.memorySpace, fixture.memorySpace, CHANNEL_CAPACITY);
17+
18+
auto &producer = *fixture.producer;
19+
auto &payloadMemoryManager = *fixture.memoryManager;
20+
auto &coordinationCommunicationManager = *fixture.communicationManager;
21+
auto &payloadCommunicationManager = *fixture.communicationManager;
22+
auto payloadMemorySpace = fixture.memorySpace;
23+
24+
////////////////////// Test begin
25+
26+
// Check payload capacity, that buffer is empty, an thus not full
27+
ASSERT_EQ(producer.getPayloadCapacity(), CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE));
28+
producer.updateDepth();
29+
ASSERT_EQ(producer.getCoordinationDepth(), 0);
30+
ASSERT_EQ(producer.getPayloadDepth(), 0);
31+
ASSERT_TRUE(producer.isEmpty());
32+
ASSERT_FALSE(producer.isFull(CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE)));
33+
ASSERT_TRUE(producer.isFull(CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE) + 1));
34+
35+
// Send a buffer big as the buffer channel
36+
ELEMENT_TYPE sendBuffer[CHANNEL_CAPACITY - 1] = {0, 1, 2, 3};
37+
auto sendBufferPtr = &sendBuffer;
38+
auto sendSlot = payloadMemoryManager.registerLocalMemorySlot(payloadMemorySpace, sendBufferPtr, sizeof(sendBuffer));
39+
40+
// Wait for the consumer 1
41+
coordinationCommunicationManager.fence(CHANNEL_TAG);
42+
payloadCommunicationManager.fence(CHANNEL_TAG);
43+
44+
// Push the slot
45+
EXPECT_NO_THROW(producer.push(sendSlot));
46+
47+
// Check that the channel can accept one more element
48+
ASSERT_FALSE(producer.isFull(sizeof(ELEMENT_TYPE)));
49+
ASSERT_TRUE(producer.isFull(2 * sizeof(ELEMENT_TYPE)));
50+
ASSERT_FALSE(producer.isEmpty());
51+
52+
// Check there is only one token, and the payload depth is equal to the capacity of the buffer minus 1 element
53+
producer.updateDepth();
54+
ASSERT_EQ(producer.getCoordinationDepth(), 1);
55+
ASSERT_EQ(producer.getPayloadDepth(), (CHANNEL_CAPACITY - 1) * sizeof(ELEMENT_TYPE));
56+
57+
// Check that trying to push another element throws exception since the channel does not have enough space
58+
EXPECT_THROW(producer.push(sendSlot), HiCR::RuntimeException);
59+
60+
// Wait for the consumer 2
61+
coordinationCommunicationManager.fence(CHANNEL_TAG);
62+
payloadCommunicationManager.fence(CHANNEL_TAG);
63+
64+
// Let the consumer pop
65+
66+
// Wait for the consumer 3
67+
coordinationCommunicationManager.fence(CHANNEL_TAG);
68+
payloadCommunicationManager.fence(CHANNEL_TAG);
69+
70+
// Now that consumer has popped it should succeed, and push to the excess buffer
71+
EXPECT_NO_THROW(producer.push(sendSlot));
72+
73+
// Wait for the consumer 4
74+
coordinationCommunicationManager.fence(CHANNEL_TAG);
75+
payloadCommunicationManager.fence(CHANNEL_TAG);
76+
77+
// Let the consumer do its part of the test
78+
79+
// Wait for the consumer 5
80+
coordinationCommunicationManager.fence(CHANNEL_TAG);
81+
payloadCommunicationManager.fence(CHANNEL_TAG);
82+
}
83+
84+
void consumerFc(ChannelFixture &fixture)
85+
{
86+
// Create producer and pick managers from the fixture
87+
fixture.consumer = fixture.createConsumer(
88+
*fixture.memoryManager, *fixture.memoryManager, *fixture.communicationManager, *fixture.communicationManager, fixture.memorySpace, fixture.memorySpace, CHANNEL_CAPACITY);
89+
90+
auto &consumer = *fixture.consumer;
91+
auto &coordinationCommunicationManager = *fixture.communicationManager;
92+
auto &payloadCommunicationManager = *fixture.communicationManager;
93+
94+
////////////////////// Test begin
95+
96+
// Check payload capacity, that buffer is empty, an thus not full
97+
consumer.updateDepth();
98+
ASSERT_EQ(consumer.getCoordinationDepth(), 0);
99+
ASSERT_EQ(consumer.getPayloadDepth(), 0);
100+
ASSERT_TRUE(consumer.isEmpty());
101+
ASSERT_FALSE(consumer.isFull(CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE)));
102+
ASSERT_TRUE(consumer.isFull(CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE) + 1));
103+
104+
// Wait for producer 1
105+
coordinationCommunicationManager.fence(CHANNEL_TAG);
106+
payloadCommunicationManager.fence(CHANNEL_TAG);
107+
108+
// Let the producer do its part of the test
109+
110+
// Wait for the producer 2
111+
coordinationCommunicationManager.fence(CHANNEL_TAG);
112+
payloadCommunicationManager.fence(CHANNEL_TAG);
113+
114+
// After the push, check there is one token and payload buffer is full
115+
consumer.updateDepth();
116+
ASSERT_EQ(consumer.getCoordinationDepth(), 1);
117+
ASSERT_EQ(consumer.getPayloadDepth(), (CHANNEL_CAPACITY - 1) * sizeof(ELEMENT_TYPE));
118+
ASSERT_FALSE(consumer.isEmpty());
119+
// Check that we still have space to push 1 token
120+
ASSERT_FALSE(consumer.isFull(sizeof(ELEMENT_TYPE)));
121+
ASSERT_TRUE(consumer.isFull(2 * sizeof(ELEMENT_TYPE)));
122+
123+
// Peek and check the token data are correct
124+
auto res = consumer.peek();
125+
ASSERT_EQ(res[0], 0);
126+
ASSERT_EQ(res[1], (CHANNEL_CAPACITY - 1) * sizeof(ELEMENT_TYPE));
127+
128+
// Check the vectory elements corresponds to the ground truth
129+
auto tokenBuffer = (uint8_t *)consumer.getPayloadBufferMemorySlot()->getSourceLocalMemorySlot()->getPointer();
130+
void *tokenPtr = &tokenBuffer[res[0]];
131+
for (ELEMENT_TYPE i = 0; i < (res[1] / sizeof(ELEMENT_TYPE)); ++i) { ASSERT_EQ(i, static_cast<ELEMENT_TYPE *>(tokenPtr)[i]); }
132+
133+
// Pop and check that the channel is empty, the depth are updated
134+
consumer.pop();
135+
ASSERT_TRUE(consumer.isEmpty());
136+
ASSERT_FALSE(consumer.isFull(CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE)));
137+
ASSERT_EQ(consumer.getCoordinationDepth(), 0);
138+
ASSERT_EQ(consumer.getPayloadDepth(), 0);
139+
140+
// Wait for the producer 3
141+
coordinationCommunicationManager.fence(CHANNEL_TAG);
142+
payloadCommunicationManager.fence(CHANNEL_TAG);
143+
144+
// Let the producer push again
145+
146+
// Wait for the producer 4
147+
coordinationCommunicationManager.fence(CHANNEL_TAG);
148+
payloadCommunicationManager.fence(CHANNEL_TAG);
149+
150+
// Wait for the producer 5
151+
coordinationCommunicationManager.fence(CHANNEL_TAG);
152+
payloadCommunicationManager.fence(CHANNEL_TAG);
153+
154+
// After the push, check there is one token
155+
consumer.updateDepth();
156+
ASSERT_EQ(consumer.getCoordinationDepth(), 1);
157+
ASSERT_EQ(consumer.getPayloadDepth(), (CHANNEL_CAPACITY - 1) * sizeof(ELEMENT_TYPE));
158+
ASSERT_FALSE(consumer.isEmpty());
159+
// Check that we still have space to push 1 token
160+
ASSERT_FALSE(consumer.isFull(sizeof(ELEMENT_TYPE)));
161+
ASSERT_TRUE(consumer.isFull(2 * sizeof(ELEMENT_TYPE)));
162+
163+
// Peek and check the token data are correct
164+
res = consumer.peek();
165+
ASSERT_EQ(res[0], (CHANNEL_CAPACITY - 1) * sizeof(ELEMENT_TYPE));
166+
ASSERT_EQ(res[1], (CHANNEL_CAPACITY - 1) * sizeof(ELEMENT_TYPE));
167+
168+
// Check the vectory elements corresponds to the ground truth
169+
tokenBuffer = (uint8_t *)consumer.getPayloadBufferMemorySlot()->getSourceLocalMemorySlot()->getPointer();
170+
tokenPtr = &tokenBuffer[res[0]];
171+
for (ELEMENT_TYPE i = 0; i < (res[1] / sizeof(ELEMENT_TYPE)); ++i) { ASSERT_EQ(i, static_cast<ELEMENT_TYPE *>(tokenPtr)[i]); }
172+
173+
// Pop and check that the channel is empty, the depth are updated
174+
consumer.pop();
175+
ASSERT_TRUE(consumer.isEmpty());
176+
ASSERT_FALSE(consumer.isFull(CHANNEL_CAPACITY * sizeof(ELEMENT_TYPE)));
177+
ASSERT_EQ(consumer.getCoordinationDepth(), 0);
178+
ASSERT_EQ(consumer.getPayloadDepth(), 0);
179+
}
180+
181+
TEST_F(ChannelFixture, UseExcessBuffer)
182+
{
183+
// Rank 0 is producer, Rank 1 is consumer
184+
if (instanceManager->getCurrentInstance()->isRootInstance()) { producerFc(*this); }
185+
else { consumerFc(*this); }
186+
}

0 commit comments

Comments
 (0)