Skip to content

Commit 5be38fb

Browse files
committed
Parallelize merge of per-thread changes into global map.
When we commit the changes from Soroban apply threads to the global state map, we can shard the output such that every worker only writes into a single shard (thus avoiding any synchronization). Besides the parallelization, this includes a few micro-optimizations that help us avoid extra hash re-computations and copies.
1 parent 4cdc91b commit 5be38fb

7 files changed

Lines changed: 541 additions & 229 deletions

File tree

src/ledger/LedgerEntryScope.cpp

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -417,6 +417,14 @@ LedgerEntryScope<S>::scopeAdoptEntryOpt(
417417
return ScopedLedgerEntryOpt(mScopeID, entry);
418418
}
419419

420+
template <StaticLedgerEntryScope S>
421+
ScopedLedgerEntryOpt<S>
422+
LedgerEntryScope<S>::scopeAdoptEntryOpt(
423+
std::optional<LedgerEntry>&& entry) const
424+
{
425+
return ScopedLedgerEntryOpt(mScopeID, std::move(entry));
426+
}
427+
420428
template <StaticLedgerEntryScope S>
421429
template <StaticLedgerEntryScope OtherScope>
422430
ScopedLedgerEntry<S>
@@ -456,6 +464,41 @@ LedgerEntryScope<S>::scopeAdoptEntryOptFromImpl(
456464
return ScopedLedgerEntryOpt<S>{mScopeID, entry.mEntry};
457465
}
458466

467+
template <StaticLedgerEntryScope S>
468+
template <StaticLedgerEntryScope OtherScope>
469+
ScopedLedgerEntry<S>
470+
LedgerEntryScope<S>::scopeAdoptEntryFromImpl(
471+
ScopedLedgerEntry<OtherScope>&& entry,
472+
LedgerEntryScope<OtherScope> const& scope) const
473+
{
474+
if (scope.mActive)
475+
{
476+
throw std::runtime_error(fmt::format(
477+
"scopeAdoptEntryFrom: adopting entry with scope ID {} from "
478+
"still-active scope ID '{}'",
479+
entry.mScopeID, scope.mScopeID));
480+
}
481+
return EntryT{mScopeID, std::move(entry.mEntry)};
482+
}
483+
484+
template <StaticLedgerEntryScope S>
485+
template <StaticLedgerEntryScope OtherScope>
486+
ScopedLedgerEntryOpt<S>
487+
LedgerEntryScope<S>::scopeAdoptEntryOptFromImpl(
488+
ScopedLedgerEntryOpt<OtherScope>&& entry,
489+
LedgerEntryScope<OtherScope> const& scope) const
490+
{
491+
if (scope.mActive)
492+
{
493+
throw std::runtime_error(
494+
fmt::format("scopeAdoptEntryOptFrom: adopting entry with "
495+
"scope ID {} from "
496+
"still-active scope ID '{}'",
497+
entry.mScopeID, scope.mScopeID));
498+
}
499+
return ScopedLedgerEntryOpt<S>{mScopeID, std::move(entry.mEntry)};
500+
}
501+
459502
/////////////////////////////////
460503
// DeactivateScopeGuard
461504
/////////////////////////////////
@@ -495,6 +538,20 @@ FOREACH_STATIC_LEDGER_ENTRY_SCOPE(INSTANTIATE_SCOPE_CLASSES)
495538
scopeAdoptEntryOptFromImpl<StaticLedgerEntryScope::SOURCE_SCOPE>( \
496539
ScopedLedgerEntryOpt<StaticLedgerEntryScope::SOURCE_SCOPE> const&, \
497540
LedgerEntryScope<StaticLedgerEntryScope::SOURCE_SCOPE> const&) \
541+
const; \
542+
\
543+
template ScopedLedgerEntry<StaticLedgerEntryScope::DEST_SCOPE> \
544+
LedgerEntryScope<StaticLedgerEntryScope::DEST_SCOPE>:: \
545+
scopeAdoptEntryFromImpl<StaticLedgerEntryScope::SOURCE_SCOPE>( \
546+
ScopedLedgerEntry<StaticLedgerEntryScope::SOURCE_SCOPE>&&, \
547+
LedgerEntryScope<StaticLedgerEntryScope::SOURCE_SCOPE> const&) \
548+
const; \
549+
\
550+
template ScopedLedgerEntryOpt<StaticLedgerEntryScope::DEST_SCOPE> \
551+
LedgerEntryScope<StaticLedgerEntryScope::DEST_SCOPE>:: \
552+
scopeAdoptEntryOptFromImpl<StaticLedgerEntryScope::SOURCE_SCOPE>( \
553+
ScopedLedgerEntryOpt<StaticLedgerEntryScope::SOURCE_SCOPE>&&, \
554+
LedgerEntryScope<StaticLedgerEntryScope::SOURCE_SCOPE> const&) \
498555
const;
499556

500557
FOR_EACH_VALID_SCOPE_ADOPTION(INSTANTIATE_ADOPT_METHODS)

src/ledger/LedgerEntryScope.h

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -387,6 +387,7 @@ template <StaticLedgerEntryScope S> class LedgerEntryScope
387387
EntryT scopeAdoptEntry(LedgerEntry const& entry) const;
388388
OptionalEntryT
389389
scopeAdoptEntryOpt(std::optional<LedgerEntry> const& entry) const;
390+
OptionalEntryT scopeAdoptEntryOpt(std::optional<LedgerEntry>&& entry) const;
390391

391392
template <StaticLedgerEntryScope OtherScope>
392393
EntryT
@@ -414,6 +415,32 @@ template <StaticLedgerEntryScope S> class LedgerEntryScope
414415
return scopeAdoptEntryOptFromImpl(entry, scope);
415416
}
416417

418+
template <StaticLedgerEntryScope OtherScope>
419+
EntryT
420+
scopeAdoptEntryFrom(ScopedLedgerEntry<OtherScope>&& entry,
421+
LedgerEntryScope<OtherScope> const& scope) const
422+
{
423+
static_assert(
424+
IsValidScopeAdoption<S, OtherScope>::value,
425+
"Invalid scope adoption: this transition is not allowed. "
426+
"Check FOR_EACH_VALID_SCOPE_ADOPTION in LedgerEntryScope.h "
427+
"for the list of valid transitions.");
428+
return scopeAdoptEntryFromImpl(std::move(entry), scope);
429+
}
430+
431+
template <StaticLedgerEntryScope OtherScope>
432+
OptionalEntryT
433+
scopeAdoptEntryOptFrom(ScopedLedgerEntryOpt<OtherScope>&& entry,
434+
LedgerEntryScope<OtherScope> const& scope) const
435+
{
436+
static_assert(
437+
IsValidScopeAdoption<S, OtherScope>::value,
438+
"Invalid scope adoption: this transition is not allowed. "
439+
"Check FOR_EACH_VALID_SCOPE_ADOPTION in LedgerEntryScope.h "
440+
"for the list of valid transitions.");
441+
return scopeAdoptEntryOptFromImpl(std::move(entry), scope);
442+
}
443+
417444
private:
418445
template <StaticLedgerEntryScope OtherScope>
419446
EntryT
@@ -424,6 +451,16 @@ template <StaticLedgerEntryScope S> class LedgerEntryScope
424451
OptionalEntryT
425452
scopeAdoptEntryOptFromImpl(ScopedLedgerEntryOpt<OtherScope> const& entry,
426453
LedgerEntryScope<OtherScope> const& scope) const;
454+
455+
template <StaticLedgerEntryScope OtherScope>
456+
EntryT
457+
scopeAdoptEntryFromImpl(ScopedLedgerEntry<OtherScope>&& entry,
458+
LedgerEntryScope<OtherScope> const& scope) const;
459+
460+
template <StaticLedgerEntryScope OtherScope>
461+
OptionalEntryT
462+
scopeAdoptEntryOptFromImpl(ScopedLedgerEntryOpt<OtherScope>&& entry,
463+
LedgerEntryScope<OtherScope> const& scope) const;
427464
};
428465

429466
template <StaticLedgerEntryScope S> class DeactivateScopeGuard

src/ledger/LedgerManagerImpl.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2724,7 +2724,7 @@ LedgerManagerImpl::applySorobanStage(
27242724
txBundle.getResPayload().getRefundableFeeTracker());
27252725
}
27262726

2727-
globalParState.commitChangesFromThreads(app, threadStates, stage);
2727+
globalParState.commitChangesFromThreads(app, threadStates);
27282728
}
27292729

27302730
void
Lines changed: 189 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,189 @@
1+
// Copyright 2026 Stellar Development Foundation and contributors. Licensed
2+
// under the Apache License, Version 2.0. See the COPYING file at the root
3+
// of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
4+
5+
#pragma once
6+
7+
#include "ledger/LedgerEntryScope.h"
8+
#include "ledger/LedgerHashUtils.h"
9+
#include "ledger/LedgerTxn.h"
10+
#include "util/UnorderedMap.h"
11+
#include "util/UnorderedSet.h"
12+
#include "xdr/Stellar-ledger-entries.h"
13+
14+
namespace stellar
15+
{
16+
17+
// A LedgerKey that caches its own hash.
18+
//
19+
// The parallel apply maps are keyed by LedgerKey and the same key is looked
20+
// up many times as it flows tx map -> thread map -> global map, so caching
21+
// the hash avoids repeated re-hashing of the same key.
22+
class ParallelApplyLedgerKey
23+
{
24+
public:
25+
explicit ParallelApplyLedgerKey(LedgerKey const& ledgerKey)
26+
: mLedgerKey(ledgerKey), mHash(std::hash<LedgerKey>{}(mLedgerKey))
27+
{
28+
}
29+
explicit ParallelApplyLedgerKey(LedgerKey&& ledgerKey)
30+
: mLedgerKey(std::move(ledgerKey))
31+
, mHash(std::hash<LedgerKey>{}(mLedgerKey))
32+
{
33+
}
34+
35+
LedgerKey const&
36+
ledgerKey() const
37+
{
38+
return mLedgerKey;
39+
}
40+
41+
operator LedgerKey const&() const
42+
{
43+
return mLedgerKey;
44+
}
45+
46+
size_t
47+
hash() const
48+
{
49+
return mHash;
50+
}
51+
52+
private:
53+
LedgerKey mLedgerKey;
54+
size_t mHash;
55+
};
56+
57+
inline bool
58+
operator==(ParallelApplyLedgerKey const& lhs, ParallelApplyLedgerKey const& rhs)
59+
{
60+
return lhs.ledgerKey() == rhs.ledgerKey();
61+
}
62+
63+
using ParallelApplyLedgerKeySet = UnorderedSet<ParallelApplyLedgerKey>;
64+
template <typename T>
65+
using ParallelApplyLedgerKeyMap = UnorderedMap<ParallelApplyLedgerKey, T>;
66+
67+
// Tracks entry updates within a transaction during parallel apply phases. If
68+
// the transaction succeeds, the thread's ParallelApplyEntryMap should be
69+
// updated with the entries from the TxModifiedEntryMap.
70+
using TxParApplyLedgerEntry =
71+
ScopedLedgerEntry<StaticLedgerEntryScope::TxParApply>;
72+
using TxModifiedEntryMap = ParallelApplyLedgerKeyMap<TxParApplyLedgerEntryOpt>;
73+
74+
// Used to track the current state of an entry during parallel apply phases. Can
75+
// be updated by successful transactions.
76+
template <StaticLedgerEntryScope S> struct ParallelApplyEntry
77+
{
78+
// Will not be set if the entry doesn't exist, or if no tx was able to load
79+
// it due to hitting read limits.
80+
ScopedLedgerEntryOpt<S> mLedgerEntry;
81+
bool mIsDirty;
82+
static ParallelApplyEntry
83+
clean(ScopedLedgerEntryOpt<S> const& e)
84+
{
85+
return ParallelApplyEntry{e, false};
86+
}
87+
static ParallelApplyEntry
88+
dirty(ScopedLedgerEntryOpt<S> const& e)
89+
{
90+
return ParallelApplyEntry{e, true};
91+
}
92+
template <StaticLedgerEntryScope S2>
93+
ParallelApplyEntry<S2>
94+
rescope(LedgerEntryScope<S> const& s1,
95+
LedgerEntryScope<S2> const& s2) const&
96+
{
97+
auto adoptedEntry = s2.scopeAdoptEntryOptFrom(mLedgerEntry, s1);
98+
return ParallelApplyEntry<S2>{adoptedEntry, mIsDirty};
99+
}
100+
// Moves the entry payload into the new scope and thus makes the current
101+
// entry invalid.
102+
template <StaticLedgerEntryScope S2>
103+
ParallelApplyEntry<S2>
104+
rescope(LedgerEntryScope<S> const& s1, LedgerEntryScope<S2> const& s2) &&
105+
{
106+
auto adoptedEntry =
107+
s2.scopeAdoptEntryOptFrom(std::move(mLedgerEntry), s1);
108+
return ParallelApplyEntry<S2>{std::move(adoptedEntry), mIsDirty};
109+
}
110+
};
111+
using GlobalParallelApplyEntry =
112+
ParallelApplyEntry<StaticLedgerEntryScope::GlobalParApply>;
113+
using ThreadParallelApplyEntry =
114+
ParallelApplyEntry<StaticLedgerEntryScope::ThreadParApply>;
115+
using TxParallelApplyEntry =
116+
ParallelApplyEntry<StaticLedgerEntryScope::TxParApply>;
117+
118+
// This is a map of all entries that will be read and/or written during parallel
119+
// apply phases: there is one such "global" map which disjoint per-thread maps
120+
// get split off of, modified during applyThread, and merged back into. Once all
121+
// threads return, the updates from each threads entry map should be committed
122+
// to LedgerTxn.
123+
template <StaticLedgerEntryScope S>
124+
using ParallelApplyEntryMap = ParallelApplyLedgerKeyMap<ParallelApplyEntry<S>>;
125+
using GlobalParallelApplyEntryMap =
126+
ParallelApplyEntryMap<StaticLedgerEntryScope::GlobalParApply>;
127+
using ThreadParallelApplyEntryMap =
128+
ParallelApplyEntryMap<StaticLedgerEntryScope::ThreadParApply>;
129+
using TxParallelApplyEntryMap =
130+
ParallelApplyEntryMap<StaticLedgerEntryScope::TxParApply>;
131+
132+
// Returned by each parallel transaction on success. It will contain the entries
133+
// modified by the transaction and the keys restored.
134+
class ParallelTxSuccessVal
135+
: public LedgerEntryScope<StaticLedgerEntryScope::TxParApply>
136+
{
137+
public:
138+
ParallelTxSuccessVal(TxModifiedEntryMap&& modifiedEntryMap,
139+
ScopeIdT txScopeID)
140+
: LedgerEntryScope(txScopeID)
141+
, mModifiedEntryMap(std::move(modifiedEntryMap))
142+
{
143+
// The ModifiedEntryMap should not be used for reading entries, only
144+
// to serve as a source for thread state to scopeAdoptEntryFrom. So
145+
// we deactivate ourselves as a LedgerEntryScope on construction, to
146+
// prevent accidental reads.
147+
scopeDeactivate();
148+
}
149+
ParallelTxSuccessVal(TxModifiedEntryMap&& modifiedEntryMap,
150+
RestoredEntries&& restoredEntries, ScopeIdT txScopeID)
151+
: LedgerEntryScope(txScopeID)
152+
, mModifiedEntryMap(std::move(modifiedEntryMap))
153+
, mRestoredEntries(std::move(restoredEntries))
154+
{
155+
scopeDeactivate();
156+
}
157+
158+
TxModifiedEntryMap const&
159+
getModifiedEntryMap() const
160+
{
161+
return mModifiedEntryMap;
162+
}
163+
RestoredEntries const&
164+
getRestoredEntries() const
165+
{
166+
return mRestoredEntries;
167+
}
168+
169+
friend class TxParallelApplyLedgerState;
170+
171+
private:
172+
// This will contain a key for every entry modified by a transaction
173+
TxModifiedEntryMap mModifiedEntryMap;
174+
RestoredEntries mRestoredEntries;
175+
};
176+
} // namespace stellar
177+
178+
namespace std
179+
{
180+
template <> class hash<stellar::ParallelApplyLedgerKey>
181+
{
182+
public:
183+
size_t
184+
operator()(stellar::ParallelApplyLedgerKey const& key) const
185+
{
186+
return key.hash();
187+
}
188+
};
189+
} // namespace std

0 commit comments

Comments
 (0)