Line data Source code
1 : #include "fd_prune_finder.h"
2 : #include "../../util/log/fd_log.h"
3 :
4 : /* Maximum number of origins tracked in the outer map. Matches
5 : Agave's ReceivedCache capacity of 2 * CRDS_UNIQUE_PUBKEY_CAPACITY. */
6 0 : #define FD_PRUNE_FINDER_ORIGIN_MAX (16384UL)
7 :
8 : /* Maximum number of relayers tracked per origin. Matches Agave's
9 : ReceivedCacheEntry::CAPACITY. When the inner map is full, new
10 : relayers are silently dropped (score-0 entries are pruned first,
11 : so an attacker cannot displace good relayers). */
12 : #define FD_PRUNE_FINDER_RELAYER_MAX (50UL)
13 :
14 : /* Minimum number of fresh (num_dups==0) messages recorded for an
15 : origin before a prune decision is made. Matches Agave's
16 : ReceivedCache::MIN_NUM_UPSERTS. */
17 : #define FD_PRUNE_FINDER_MIN_NUM_UPSERTS (20UL)
18 :
19 : /* Minimum number of relayers to keep (never pruned) per origin,
20 : regardless of score or stake. Matches Agave's
21 : CRDS_GOSSIP_PRUNE_MIN_INGRESS_NODES. */
22 : #define FD_PRUNE_FINDER_MIN_INGRESS_NODES (2UL)
23 :
24 : /* Fraction of stake that must be covered before additional relayers
25 : can be pruned. min_ingress_stake = min(identity_stake, origin_stake)
26 : * FD_PRUNE_FINDER_STAKE_THRESHOLD_PCT / 100. Matches Agave's
27 : CRDS_GOSSIP_PRUNE_STAKE_THRESHOLD_PCT = 0.15. */
28 0 : #define FD_PRUNE_FINDER_STAKE_THRESHOLD_PCT (15UL)
29 :
30 : /* Number of duplicates below which a relayer is considered timely
31 : (its score is incremented). Matches Agave's
32 : ReceivedCacheEntry::NUM_DUPS_THRESHOLD. */
33 : #define FD_PRUNE_FINDER_NUM_DUPS_THRESHOLD (2UL)
34 :
35 : struct pubkey_private {
36 : uchar b[ 32UL ];
37 : };
38 :
39 : typedef struct pubkey_private pubkey_private_t;
40 :
41 : /* fd_prune_relayer stores the score for a single relayer within an
42 : origin entry. The score counts how many times this relayer was
43 : among the first NUM_DUPS_THRESHOLD (2) to deliver a message from
44 : this origin. relayer_stake is cached at insertion time. */
45 :
46 : struct fd_prune_relayer {
47 : uchar pubkey[ 32UL ];
48 : ulong score;
49 : ulong stake;
50 : };
51 :
52 : typedef struct fd_prune_relayer fd_prune_relayer_t;
53 :
54 : /* fd_prune_origin is an entry in the outer map, keyed by origin
55 : pubkey. It tracks all known relayers for that origin and the
56 : number of fresh (num_dups==0) messages received. */
57 :
58 : struct fd_prune_origin {
59 : pubkey_private_t origin_pubkey;
60 :
61 : ulong num_upserts;
62 : ulong origin_stake;
63 : ulong relayers_cnt;
64 : fd_prune_relayer_t relayers[ FD_PRUNE_FINDER_RELAYER_MAX ];
65 :
66 : ulong pool_next;
67 :
68 : ulong map_next;
69 : ulong map_prev;
70 :
71 : ulong lru_prev;
72 : ulong lru_next;
73 : };
74 :
75 : typedef struct fd_prune_origin fd_prune_origin_t;
76 :
77 : #define POOL_NAME pool
78 0 : #define POOL_NEXT pool_next
79 0 : #define POOL_T fd_prune_origin_t
80 : #include "../../util/tmpl/fd_pool.c"
81 :
82 : #pragma GCC diagnostic push
83 : #pragma GCC diagnostic ignored "-Wunused-value"
84 : #define DLIST_NAME lru_list
85 : #define DLIST_ELE_T fd_prune_origin_t
86 0 : #define DLIST_PREV lru_prev
87 0 : #define DLIST_NEXT lru_next
88 : #include "../../util/tmpl/fd_dlist.c"
89 : #pragma GCC diagnostic pop
90 :
91 : #define MAP_NAME origin_map
92 : #define MAP_ELE_T fd_prune_origin_t
93 : #define MAP_KEY_T pubkey_private_t
94 0 : #define MAP_KEY origin_pubkey
95 0 : #define MAP_IDX_T ulong
96 0 : #define MAP_NEXT map_next
97 0 : #define MAP_PREV map_prev
98 0 : #define MAP_KEY_HASH(k,s) ((s) ^ fd_ulong_load_8( (k)->b ))
99 0 : #define MAP_KEY_EQ(k0,k1) (!memcmp((k0)->b, (k1)->b, 32UL))
100 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
101 : #include "../../util/tmpl/fd_map_chain.c"
102 :
103 : /* Maximum number of (destination, origin) pairs that can be buffered
104 : between record and pop_prune calls. 17 push values * 50 relayers
105 : per origin = 850 worst case. */
106 : #define FD_PRUNE_FINDER_PENDING_MAX (850UL)
107 :
108 : struct fd_prune_pending {
109 : uchar relayer[ 32UL ];
110 : uchar origin[ 32UL ];
111 : };
112 :
113 : struct fd_prune_finder_private {
114 : fd_prune_origin_t * pool;
115 : origin_map_t * origins;
116 : lru_list_t * lru;
117 :
118 : uchar identity_pubkey[ 32UL ];
119 : ulong identity_stake;
120 :
121 : ulong pending_cnt;
122 : ulong pending_read;
123 : struct fd_prune_pending pending[ FD_PRUNE_FINDER_PENDING_MAX ];
124 : };
125 :
126 : FD_FN_CONST ulong
127 0 : fd_prune_finder_align( void ) {
128 0 : return 128UL;
129 0 : }
130 :
131 : FD_FN_CONST ulong
132 0 : fd_prune_finder_footprint( void ) {
133 0 : ulong chain_cnt = origin_map_chain_cnt_est( FD_PRUNE_FINDER_ORIGIN_MAX );
134 0 : ulong l;
135 0 : l = FD_LAYOUT_INIT;
136 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_prune_finder_t), sizeof(fd_prune_finder_t) );
137 0 : l = FD_LAYOUT_APPEND( l, pool_align(), pool_footprint( FD_PRUNE_FINDER_ORIGIN_MAX ) );
138 0 : l = FD_LAYOUT_APPEND( l, origin_map_align(), origin_map_footprint( chain_cnt ) );
139 0 : l = FD_LAYOUT_APPEND( l, lru_list_align(), lru_list_footprint() );
140 0 : l = FD_LAYOUT_FINI( l, fd_prune_finder_align() );
141 0 : return l;
142 0 : }
143 :
144 : void *
145 : fd_prune_finder_new( void * shmem,
146 0 : ulong seed ) {
147 0 : if( FD_UNLIKELY( !shmem ) ) return NULL;
148 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_prune_finder_align() ) ) ) return NULL;
149 :
150 0 : ulong chain_cnt = origin_map_chain_cnt_est( FD_PRUNE_FINDER_ORIGIN_MAX );
151 :
152 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
153 0 : fd_prune_finder_t * pf = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_prune_finder_t), sizeof(fd_prune_finder_t) );
154 0 : void * pool_mem = FD_SCRATCH_ALLOC_APPEND( l, pool_align(), pool_footprint( FD_PRUNE_FINDER_ORIGIN_MAX ) );
155 0 : void * map_mem = FD_SCRATCH_ALLOC_APPEND( l, origin_map_align(), origin_map_footprint( chain_cnt ) );
156 0 : void * lru_mem = FD_SCRATCH_ALLOC_APPEND( l, lru_list_align(), lru_list_footprint() );
157 :
158 0 : pf->pool = pool_join( pool_new( pool_mem, FD_PRUNE_FINDER_ORIGIN_MAX ) );
159 0 : FD_TEST( pf->pool );
160 0 : pf->origins = origin_map_join( origin_map_new( map_mem, chain_cnt, seed ) );
161 0 : FD_TEST( pf->origins );
162 0 : pf->lru = lru_list_join( lru_list_new( lru_mem ) );
163 0 : FD_TEST( pf->lru );
164 :
165 0 : fd_memset( pf->identity_pubkey, 0, 32UL );
166 0 : pf->identity_stake = 0UL;
167 :
168 0 : pf->pending_cnt = 0UL;
169 0 : pf->pending_read = 0UL;
170 :
171 0 : return pf;
172 0 : }
173 :
174 : fd_prune_finder_t *
175 0 : fd_prune_finder_join( void * shpf ) {
176 0 : return (fd_prune_finder_t *)shpf;
177 0 : }
178 :
179 : void
180 : fd_prune_finder_set_identity( fd_prune_finder_t * pf,
181 : uchar const * identity_pubkey,
182 0 : ulong identity_stake ) {
183 0 : fd_memcpy( pf->identity_pubkey, identity_pubkey, 32UL );
184 0 : pf->identity_stake = identity_stake;
185 0 : }
186 :
187 : static inline fd_prune_relayer_t *
188 : find_relayer( fd_prune_origin_t * origin,
189 0 : uchar const * relayer_pubkey ) {
190 0 : for( ulong i=0UL; i<origin->relayers_cnt; i++ ) {
191 0 : if( FD_UNLIKELY( !memcmp( origin->relayers[ i ].pubkey, relayer_pubkey, 32UL ) ) ) {
192 0 : return &origin->relayers[ i ];
193 0 : }
194 0 : }
195 0 : return NULL;
196 0 : }
197 :
198 : static inline fd_prune_relayer_t *
199 : insert_relayer( fd_prune_origin_t * origin,
200 : uchar const * relayer_pubkey,
201 : ulong score,
202 0 : ulong relayer_stake ) {
203 0 : if( FD_UNLIKELY( origin->relayers_cnt>=FD_PRUNE_FINDER_RELAYER_MAX ) ) return NULL;
204 0 : fd_prune_relayer_t * r = &origin->relayers[ origin->relayers_cnt++ ];
205 0 : fd_memcpy( r->pubkey, relayer_pubkey, 32UL );
206 0 : r->score = score;
207 0 : r->stake = relayer_stake;
208 0 : return r;
209 0 : }
210 :
211 : #define SORT_NAME sort_relayers_desc
212 0 : #define SORT_KEY_T fd_prune_relayer_t
213 0 : #define SORT_BEFORE(a,b) ((a).score>(b).score || ((a).score==(b).score && (a).stake>(b).stake))
214 : #define SORT_QUICK_SWAP_MINIMIZE 1
215 : #include "../../util/tmpl/fd_sort.c"
216 :
217 : /* do_prune executes the prune decision for a single origin entry.
218 : Sorts relayers by (score, stake) descending, keeps at least
219 : min_ingress_nodes (2) and enough stake to cover min_ingress_stake,
220 : then appends (destination, origin) pairs for the remaining relayers
221 : to the pending buffer. Resets the origin entry afterward. */
222 :
223 : static void
224 : do_prune( fd_prune_finder_t * pf,
225 0 : fd_prune_origin_t * origin ) {
226 0 : ulong cnt = origin->relayers_cnt;
227 0 : if( FD_UNLIKELY( !cnt ) ) {
228 0 : origin->num_upserts = 0UL;
229 0 : origin->relayers_cnt = 0UL;
230 0 : return;
231 0 : }
232 :
233 0 : sort_relayers_desc_insert( origin->relayers, cnt );
234 :
235 : /* Compute min_ingress_stake = min(identity_stake, origin_stake) * 15/100.
236 : Relayers covering the top min_ingress_nodes and enough cumulative
237 : stake to meet min_ingress_stake are kept; the rest are pruned. */
238 0 : ulong min_base = fd_ulong_min( pf->identity_stake, origin->origin_stake );
239 : /* Keep floor(min_base * pct / 100) semantics while avoiding overflow:
240 : min_base = 100*q + r, so floor(min_base*pct/100)=q*pct+floor(r*pct/100).
241 : With pct<=100, both products are bounded and do not overflow ulong. */
242 0 : ulong min_ingress_stake = ( min_base / 100UL ) * FD_PRUNE_FINDER_STAKE_THRESHOLD_PCT
243 0 : + ( ( min_base % 100UL ) * FD_PRUNE_FINDER_STAKE_THRESHOLD_PCT ) / 100UL;
244 :
245 0 : ulong cum_stake = 0UL;
246 :
247 0 : for( ulong i=0UL; i<cnt; i++ ) {
248 0 : fd_prune_relayer_t * r = &origin->relayers[ i ];
249 :
250 0 : if( FD_LIKELY( i<FD_PRUNE_FINDER_MIN_INGRESS_NODES ) ) {
251 0 : cum_stake += r->stake;
252 0 : continue;
253 0 : }
254 :
255 0 : if( FD_LIKELY( cum_stake<min_ingress_stake ) ) {
256 0 : cum_stake += r->stake;
257 0 : continue;
258 0 : }
259 :
260 : /* Filter out origin == relayer (per Agave) */
261 0 : if( FD_UNLIKELY( !memcmp( r->pubkey, origin->origin_pubkey.b, 32UL ) ) ) continue;
262 :
263 0 : FD_TEST( pf->pending_cnt<FD_PRUNE_FINDER_PENDING_MAX );
264 0 : struct fd_prune_pending * p = &pf->pending[ pf->pending_cnt++ ];
265 0 : fd_memcpy( p->relayer, r->pubkey, 32UL );
266 0 : fd_memcpy( p->origin, origin->origin_pubkey.b, 32UL );
267 0 : }
268 :
269 : /* Reset the origin entry (matching Agave's std::mem::take). */
270 0 : origin->num_upserts = 0UL;
271 0 : origin->relayers_cnt = 0UL;
272 0 : }
273 :
274 : void
275 : fd_prune_finder_record( fd_prune_finder_t * pf,
276 : uchar const * origin_pubkey,
277 : ulong origin_stake,
278 : uchar const * relayer_pubkey,
279 : ulong relayer_stake,
280 0 : ulong num_dups ) {
281 0 : fd_prune_origin_t * origin = origin_map_ele_query( pf->origins,
282 0 : fd_type_pun_const( origin_pubkey ),
283 0 : NULL,
284 0 : pf->pool );
285 :
286 0 : if( FD_UNLIKELY( !origin ) ) {
287 0 : if( FD_LIKELY( pool_free( pf->pool ) ) ) {
288 0 : origin = pool_ele_acquire( pf->pool );
289 0 : } else {
290 0 : origin = lru_list_ele_pop_head( pf->lru, pf->pool );
291 0 : origin_map_ele_remove( pf->origins, &origin->origin_pubkey, NULL, pf->pool );
292 0 : }
293 :
294 0 : origin->num_upserts = 0UL;
295 0 : origin->relayers_cnt = 0UL;
296 0 : origin->origin_stake = origin_stake;
297 0 : fd_memcpy( origin->origin_pubkey.b, origin_pubkey, 32UL );
298 :
299 0 : origin_map_ele_insert( pf->origins, origin, pf->pool );
300 0 : lru_list_ele_push_tail( pf->lru, origin, pf->pool );
301 0 : } else {
302 0 : lru_list_ele_remove( pf->lru, origin, pf->pool );
303 0 : lru_list_ele_push_tail( pf->lru, origin, pf->pool );
304 0 : origin->origin_stake = origin_stake;
305 0 : }
306 :
307 0 : if( FD_UNLIKELY( !num_dups ) ) origin->num_upserts++;
308 :
309 0 : if( FD_LIKELY( num_dups<FD_PRUNE_FINDER_NUM_DUPS_THRESHOLD ) ) {
310 0 : fd_prune_relayer_t * r = find_relayer( origin, relayer_pubkey );
311 0 : if( FD_LIKELY( r ) ) {
312 0 : r->score++;
313 0 : r->stake = relayer_stake;
314 0 : } else {
315 0 : insert_relayer( origin, relayer_pubkey, 1UL, relayer_stake );
316 0 : }
317 0 : } else {
318 : /* Late delivery (num_dups >= 2): insert with score 0 if room.
319 : Do not increment score — prevents spoofed addresses from
320 : penalizing a good relayer. But do ensure the relayer is in
321 : the map so it can be pruned later. */
322 0 : fd_prune_relayer_t * r = find_relayer( origin, relayer_pubkey );
323 0 : if( FD_UNLIKELY( !r ) ) {
324 0 : insert_relayer( origin, relayer_pubkey, 0UL, relayer_stake );
325 0 : } else {
326 0 : r->stake = relayer_stake;
327 0 : }
328 0 : }
329 :
330 0 : if( FD_UNLIKELY( origin->num_upserts>=FD_PRUNE_FINDER_MIN_NUM_UPSERTS ) ) {
331 0 : do_prune( pf, origin );
332 0 : }
333 0 : }
334 :
335 : int
336 : fd_prune_finder_pop_prune( fd_prune_finder_t * pf,
337 : uchar const ** out_relayer,
338 0 : uchar const ** out_origin ) {
339 0 : if( FD_UNLIKELY( pf->pending_read>=pf->pending_cnt ) ) {
340 0 : pf->pending_read = 0UL;
341 0 : pf->pending_cnt = 0UL;
342 0 : return 0;
343 0 : }
344 :
345 0 : struct fd_prune_pending * p = &pf->pending[ pf->pending_read++ ];
346 0 : *out_relayer = p->relayer;
347 0 : *out_origin = p->origin;
348 0 : return 1;
349 0 : }
|
