Skip to content

Commit c835304

Browse files
ironMannironMann
committed
stfsender: optimize stfcopy and improve handling of memory pressure
1 parent 47bd75c commit c835304

6 files changed

Lines changed: 162 additions & 39 deletions

File tree

‎src/StfSender/StfSenderOutput.cxx‎

Lines changed: 73 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,8 @@ void StfSenderOutput::start(std::shared_ptr<ConsulStfSender> pDiscoveryConfig)
6969

7070
// create stf ordering thread
7171
mStfOrderThread = create_thread_member("stfs_order", &StfSenderOutput::StfOrderThread, this);
72+
// create stf alloc thread
73+
mCopyAllocThread = create_thread_member("stfs_alloc", &StfSenderOutput::StfCopyAllocThread, this);
7274

7375
// create stf copy thread
7476
for (auto i = 0; i < 8; i++) {
@@ -125,6 +127,9 @@ void StfSenderOutput::start_standalone(std::shared_ptr<ConsulStfSender> pDiscove
125127
// create stf ordering thread
126128
mStfOrderThread = create_thread_member("stfs_order", &StfSenderOutput::StfOrderThread, this);
127129

130+
// create stf alloc thread
131+
mCopyAllocThread = create_thread_member("stfs_alloc", &StfSenderOutput::StfCopyAllocThread, this);
132+
128133
// create stf copy thread
129134
for (auto i = 0; i < 8; i++) {
130135
mCopyThreads.emplace_back(create_thread_member("stfs_copy", &StfSenderOutput::StfCopyThread, this));
@@ -143,13 +148,19 @@ void StfSenderOutput::stop()
143148
mRunning = false;
144149
mDropQueue.stop();
145150
mScheduleQueue.stop();
151+
mCopyAllocQueue.stop();
146152
mCopyQueue.stop();
147153

148154
// stop the stf ordering: on pipeline interrupt
149155
if (mStfOrderThread.joinable()) {
150156
mStfOrderThread.join();
151157
}
152158

159+
// stop the stf copy alloc threqad
160+
if (mCopyAllocThread.joinable()) {
161+
mCopyAllocThread.join();
162+
}
163+
153164
// stop copy threads
154165
for (auto &lThread : mCopyThreads) {
155166
if (lThread.joinable()) {
@@ -287,7 +298,12 @@ void StfSenderOutput::StfKeepThread()
287298
}
288299

289300
if (lStfOpt) {
290-
std::unique_ptr<SubTimeFrame> lStf = std::move(lStfOpt.value());
301+
std::unique_ptr<SubTimeFrame> lStf = std::move(lStfOpt.value().mStf);
302+
303+
if (lStfOpt.value().mMemoryPressure) {
304+
// release immediately if we could not copy
305+
continue;
306+
}
291307

292308
const auto lStfId = lStf->id();
293309
const auto lStfSize = lStf->getDataSize();
@@ -313,45 +329,77 @@ void StfSenderOutput::StfOrderThread()
313329
{
314330
std::unique_ptr<SubTimeFrame> lStf;
315331
while ((lStf = mPipelineI.dequeue(eSenderIn)) != nullptr) {
316-
317332
DDDLOG_RL(1000, "StfOrderThread: receiving {}", lStf->id() );
318333

319-
std::unique_lock lOrderLock(mScheduledStfMapLock);
320-
mStfOrderingQueue.push(lStf->id());
321-
// push the original Stf for scheduling
322-
mCopyQueue.push(std::move(lStf));
334+
if (mStfCopyBuilder) {
335+
std::unique_lock lOrderLock(mScheduledStfMapLock);
336+
mStfOrderingQueue.push(lStf->id());
337+
// push the original Stf for allocation copy
338+
mCopyAllocQueue.push(std::move(lStf));
339+
} else {
340+
// no copying
341+
StfSchedInfo lSchedInfo;
342+
lSchedInfo.mStf = std::move(lStf);
343+
lSchedInfo.mMemoryPressure = false;
344+
mScheduleQueue.push(std::move(lSchedInfo));
345+
}
323346
}
324347
DDDLOG("StfOrderThread: Exiting.");
325348
}
326349

327-
void StfSenderOutput::StfCopyThread()
350+
void StfSenderOutput::StfCopyAllocThread()
328351
{
329352
while (mRunning) {
330353
std::unique_ptr<SubTimeFrame> lStf;
331-
if (!mCopyQueue.pop(lStf)) {
354+
if (!mCopyAllocQueue.pop(lStf)) {
332355
break;
333356
}
334357

335-
if (mStfCopyBuilder) {
336-
// copy stf
337-
if (!mStfCopyBuilder->copyStfData(lStf)) {
338-
DDMON("stfsender", "stf_region.full", 1);
339-
}
358+
StfCopyInfo lStfCopyInfo;
359+
360+
// allocate memory for the stf
361+
assert (mStfCopyBuilder);
362+
mStfCopyBuilder->allocNewStfData(lStf, lStfCopyInfo.mLinkBuffers);
363+
364+
lStfCopyInfo.mStf = std::move(lStf);
365+
mCopyQueue.push(std::move(lStfCopyInfo));
366+
}
367+
DDDLOG("StfCopyAlloc: Exiting.");
368+
}
369+
370+
void StfSenderOutput::StfCopyThread()
371+
{
372+
while (mRunning) {
373+
StfCopyInfo lStfCopyInfo;
374+
if (!mCopyQueue.pop(lStfCopyInfo)) {
375+
break;
376+
}
377+
378+
assert (mStfCopyBuilder);
379+
380+
// copy stf
381+
const bool lCopyOk = mStfCopyBuilder->copyStfData(lStfCopyInfo.mStf, lStfCopyInfo.mLinkBuffers);
382+
if (!lCopyOk) {
383+
// Copy region is full or fragmented
384+
DDMON("stfsender", "stf_region.full", 1);
340385
}
341386

342387
// wait for our turn to schedule
343388
while (mRunning) {
344389
std::unique_lock lOrderLock(mScheduledStfMapLock);
345390
assert (!mStfOrderingQueue.empty());
346-
assert (mStfOrderingQueue.front() <= lStf->id());
391+
assert (mStfOrderingQueue.front() <= lStfCopyInfo.mStf->id());
347392

348-
if (mStfOrderingQueue.front() != lStf->id()) {
393+
if (mStfOrderingQueue.front() != lStfCopyInfo.mStf->id()) {
349394
mStfOrderingCv.wait_for(lOrderLock, 10ms);
350395
continue;
351396
} else {
352397
mStfOrderingQueue.pop();
353398
// push the original Stf for scheduling
354-
mScheduleQueue.push(std::move(lStf));
399+
StfSchedInfo lSchedInfo;
400+
lSchedInfo.mMemoryPressure = !lCopyOk;
401+
lSchedInfo.mStf = std::move(lStfCopyInfo.mStf);
402+
mScheduleQueue.push(std::move(lSchedInfo));
355403
break;
356404
}
357405
}
@@ -373,11 +421,13 @@ void StfSenderOutput::StfSchedulerThread()
373421
#endif
374422

375423
while (mRunning) {
376-
std::unique_ptr<SubTimeFrame> lStf;
377-
if (!mScheduleQueue.pop(lStf)) {
424+
StfSchedInfo lSchedInfo;
425+
if (!mScheduleQueue.pop(lSchedInfo)) {
378426
break;
379427
}
380428

429+
std::unique_ptr<SubTimeFrame> lStf = std::move(lSchedInfo.mStf);
430+
381431
const auto lStfId = lStf->id();
382432
const auto lStfSize = lStf->getDataSize();
383433

@@ -420,7 +470,7 @@ void StfSenderOutput::StfSchedulerThread()
420470
lStfInfo.set_stf_size(lStfSize);
421471

422472
lStfInfo.mutable_stfs_info()->set_buffer_size(mBufferSize);
423-
lStfInfo.mutable_stfs_info()->set_buffer_used(lCounters.mBuffered.mSize);
473+
lStfInfo.mutable_stfs_info()->set_buffer_used(lSchedInfo.mMemoryPressure ? mBufferSize : lCounters.mBuffered.mSize);
424474
lStfInfo.mutable_stfs_info()->set_num_buffered_stfs(lCounters.mBuffered.mCnt);
425475

426476
switch (lStf->header().mOrigin) {
@@ -487,8 +537,10 @@ void StfSenderOutput::StfSchedulerThread()
487537
mCounters.mValues.mSchedulerStfRejectedCnt += 1;
488538
}
489539

490-
WDDLOG_RL(5000, "TfScheduler rejected the Stf announce. stf_id={} reason={}",
491-
lStfId, SchedulerStfInfoResponse_StfInfoStatus_Name(lSchedResponse.status()));
540+
if (lSchedResponse.status() != SchedulerStfInfoResponse::DROP_STFS_THROTTLING) {
541+
WDDLOG_RL(5000, "TfScheduler rejected the Stf announce. stf_id={} reason={}",
542+
lStfId, SchedulerStfInfoResponse_StfInfoStatus_Name(lSchedResponse.status()));
543+
}
492544
}
493545
DDDLOG_RL(5000, "StfSchedulerThread: Sent an STF announce. stf_id={} stf_size={}", lStfId, lStfInfo.stf_size());
494546
}

‎src/StfSender/StfSenderOutput.h‎

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ class StfSenderOutput
5656
bool running() const;
5757

5858
void StfOrderThread();
59+
void StfCopyAllocThread();
5960
void StfCopyThread();
6061
void StfSchedulerThread();
6162
void StfKeepThread();
@@ -96,6 +97,11 @@ class StfSenderOutput
9697
mScheduledStfMap.clear();
9798
mSchedulingResult.clear();
9899
}
100+
101+
{ // clear the standalone keep map
102+
std::scoped_lock lMapLock(mStfKeepMapLock);
103+
mStfKeepMap.clear();
104+
}
99105
}
100106

101107
private:
@@ -114,8 +120,8 @@ class StfSenderOutput
114120
std::uint64_t mStaleStfTimeoutMs = StaleStfTimeoutMsDefault;
115121

116122
/// Stf keeper thread for standalone runs
117-
std::atomic_uint64_t mKeepTarget = 512ULL << 20;
118-
std::atomic_uint64_t mDeletePercentage = 50;
123+
std::atomic_uint64_t mKeepTarget = StandaloneStfDataBufferSizeMBDefault << 20;
124+
std::atomic_uint64_t mDeletePercentage = StandaloneStfDeleteChanceDefault;
119125
std::mutex mStfKeepMapLock;
120126
std::map<std::uint64_t, std::unique_ptr<SubTimeFrame>> mStfKeepMap;
121127
std::thread mStfKeepThread;
@@ -126,16 +132,30 @@ class StfSenderOutput
126132

127133
/// StfCopy builder; not used if nullptr
128134
std::shared_ptr<SubTimeFrameCopyBuilder> mStfCopyBuilder;
129-
std::vector<std::thread> mCopyThreads;
135+
136+
/// Stf copy threads and queues
137+
struct StfCopyInfo {
138+
std::unique_ptr<SubTimeFrame> mStf;
139+
std::vector<void*> mLinkBuffers;
140+
};
130141

131142
std::thread mStfOrderThread;
132-
ConcurrentFifo<std::unique_ptr<SubTimeFrame>> mCopyQueue;
143+
std::thread mCopyAllocThread;
144+
std::vector<std::thread> mCopyThreads;
145+
146+
ConcurrentFifo<std::unique_ptr<SubTimeFrame>> mCopyAllocQueue;
147+
ConcurrentFifo<StfCopyInfo> mCopyQueue;
148+
133149
std::mutex mStfOrderingLock;
134150
std::condition_variable mStfOrderingCv;
135151
std::queue<std::uint64_t> mStfOrderingQueue;
136152

137153
/// Scheduler threads
138-
ConcurrentFifo<std::unique_ptr<SubTimeFrame>> mScheduleQueue;
154+
struct StfSchedInfo {
155+
std::unique_ptr<SubTimeFrame> mStf;
156+
bool mMemoryPressure = false;
157+
};
158+
ConcurrentFifo<StfSchedInfo> mScheduleQueue;
139159
std::thread mSchedulerThread;
140160
std::uint64_t mLastStfId = 0;
141161
std::mutex mScheduledStfMapLock;

‎src/common/MemoryUtils.h‎

Lines changed: 5 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525

2626
#include <Headers/DataHeader.h>
2727

28+
#include "Utilities.h"
2829
#include "DataDistLogger.h"
2930

3031
#include <vector>
@@ -87,7 +88,7 @@ class RegionAllocatorResource
8788

8889
fair::mq::RegionConfig lRegionCfg;
8990

90-
pSize = align_size_up(pSize);
91+
pSize = align_size_up<ALIGN>(pSize);
9192

9293
int lMapFlags = 0;
9394
std::string lSegmentRoot = "";
@@ -180,7 +181,7 @@ class RegionAllocatorResource
180181
}
181182

182183
// align up the message size for correct interval merging
183-
const auto lASize = align_size_up(lInt.size);
184+
const auto lASize = align_size_up<ALIGN>(lInt.size);
184185

185186
// check for buffer sentinel value
186187
if (lZeroCheckShmMemory && (lASize > lInt.size)) {
@@ -416,7 +417,7 @@ class RegionAllocatorResource
416417
if constexpr (FREE_STRATEGY == eExactRegion) {
417418
auto intervalA = icl::discrete_interval<std::size_t>::right_open(
418419
reinterpret_cast<std::size_t>(pPtr),
419-
reinterpret_cast<std::size_t>(pPtr)+align_size_up(pSize)
420+
reinterpret_cast<std::size_t>(pPtr)+align_size_up<ALIGN>(pSize)
420421
);
421422

422423
if (icl::intersects(mFreeRanges, intervalA)) {
@@ -474,10 +475,6 @@ class RegionAllocatorResource
474475

475476
// NOTE: we align sizes of returned messages, but keep the exact size for allocation
476477
// otherwise the shm messages would be larger than requested
477-
static constexpr inline
478-
std::size_t align_size_up(const std::size_t pSize) {
479-
return (pSize + ALIGN - 1) / ALIGN * ALIGN;
480-
}
481478

482479
void* do_allocate(const std::size_t pSize)
483480
{
@@ -491,7 +488,7 @@ class RegionAllocatorResource
491488
}
492489

493490
// align up
494-
const auto pSizeUp = align_size_up(pSize);
491+
const auto pSizeUp = align_size_up<ALIGN>(pSize);
495492

496493
auto lRet = try_alloc(pSizeUp);
497494

‎src/common/SubTimeFrameBuilder.cxx‎

Lines changed: 52 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -609,20 +609,68 @@ void SubTimeFrameCopyBuilder::allocate_memory(const std::size_t pDataSegSize, co
609609
mMemRes.start();
610610
}
611611

612-
bool SubTimeFrameCopyBuilder::copyStfData(std::unique_ptr<SubTimeFrame> &pStf)
612+
bool SubTimeFrameCopyBuilder::copyStfData(std::unique_ptr<SubTimeFrame> &pStf, const std::vector<void*> &pLinkBuffers)
613613
{
614+
bool lRet = true;
615+
std::size_t lLinkIdx = 0;
616+
std::vector<std::pair<void*, std::size_t>> lDataMsgsBuffers;
617+
614618
for (auto& lDataIdentMapIter : pStf->mData) {
615619
for (auto& lSubSpecMapIter : lDataIdentMapIter.second) {
616620
for (auto& lStfDataIter : lSubSpecMapIter.second) {
617621
// replace and copy data FMQ messages
618-
if (!mMemRes.replaceDataMessages(lStfDataIter.mDataParts)) {
619-
return false;
622+
assert (pLinkBuffers.size() > lLinkIdx);
623+
char* lLinkBuffPtr = reinterpret_cast<char*>(pLinkBuffers[lLinkIdx++]);
624+
if (!lLinkBuffPtr) {
625+
lRet = false;
626+
continue; // we have to check all pointers
620627
}
628+
629+
lDataMsgsBuffers.clear();
630+
631+
assert (lLinkBuffPtr);
632+
for (auto &lDataMsg : lStfDataIter.mDataParts) {
633+
const auto lSize = lDataMsg->GetSize();
634+
std::memcpy(lLinkBuffPtr, lDataMsg->GetData(), lSize);
635+
lDataMsgsBuffers.emplace_back(lLinkBuffPtr, lSize);
636+
lLinkBuffPtr += align_size_up<16>(lSize);
637+
}
638+
639+
// remove old fmq messages
640+
lStfDataIter.mDataParts.clear();
641+
// make new data messages
642+
mMemRes.fmqFromDataBuffers(lDataMsgsBuffers, std::back_inserter(lStfDataIter.mDataParts));
621643
}
622644
}
623645
}
624646

625-
return true;
647+
return lRet;
648+
}
649+
650+
651+
void SubTimeFrameCopyBuilder::allocNewStfData(const std::unique_ptr<SubTimeFrame> &pStf, std::vector<void*> &pLinkBuffers)
652+
{
653+
std::vector<std::uint64_t> lSizes;
654+
655+
// calculate how much to allocate
656+
for (const auto& lDataIdentMapIter : pStf->mData) {
657+
for (const auto& lSubSpecMapIter : lDataIdentMapIter.second) {
658+
for (const auto& lStfDataIter : lSubSpecMapIter.second) {
659+
660+
std::uint64_t lPartSize = 0;
661+
// allocate one buffer for each link
662+
for (const auto &lDataMsg : lStfDataIter.mDataParts) {
663+
lPartSize += align_size_up<16>(lDataMsg->GetSize());
664+
}
665+
666+
// save the size of each link
667+
lSizes.push_back(lPartSize);
668+
}
669+
}
670+
}
671+
672+
// allocate buffers for all links
673+
mMemRes.allocDataBuffers(lSizes, std::back_inserter(pLinkBuffers));
626674
}
627675

628676
} /* o2::DataDistribution */

‎src/common/SubTimeFrameBuilder.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -219,7 +219,8 @@ class SubTimeFrameCopyBuilder
219219
// make allocate the memory here
220220
void allocate_memory(const std::size_t pDataSegSize, const std::optional<std::uint16_t> pDataSegId);
221221

222-
bool copyStfData(std::unique_ptr<SubTimeFrame> &pStf);
222+
void allocNewStfData(const std::unique_ptr<SubTimeFrame> &pStf, std::vector<void*> &pLinkBuffers);
223+
bool copyStfData(std::unique_ptr<SubTimeFrame> &pStf, const std::vector<void*> &pLinkBuffers);
223224

224225
inline auto freeData() const { return mMemRes.freeData(); }
225226

‎src/common/base/Utilities.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,11 @@ void assume(const bool pPred) {
3636
}
3737
}
3838

39+
template<std::size_t ALIGN>
40+
std::size_t align_size_up(const std::size_t pSize) {
41+
return (pSize + ALIGN - 1) / ALIGN * ALIGN;
42+
}
43+
3944
// helper function to get default LHC period
4045
static inline
4146
std::string getDefaultLhcPeriod() {

0 commit comments

Comments
 (0)