Skip to content

Commit 14d10eb

Browse files
ravyu-jumpravyu-jump
committed
gossip: weighted sampler API
1 parent c3e0731 commit 14d10eb

File tree

3 files changed

+186
-1
lines changed

3 files changed

+186
-1
lines changed

‎src/flamenco/gossip/Local.mk‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ $(call add-objs,fd_gossip fd_gossip_msg_parse fd_gossip_msg_ser fd_gossip_out fd
33

44
$(call add-hdrs,fd_bloom.h)
55
$(call add-hdrs,fd_gossip_types.h)
6-
$(call add-objs,fd_bloom fd_active_set fd_ping_tracker,fd_flamenco)
6+
$(call add-objs,fd_bloom fd_active_set fd_ping_tracker fd_gossip_wpeer_sampler,fd_flamenco)
77

88
$(call make-unit-test,test_bloom,test_bloom,fd_flamenco fd_util)
99
$(call run-unit-test,test_bloom)

‎src/flamenco/gossip/fd_gossip_wpeer_sampler.c‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
#include "fd_gossip_wpeer_sampler.h"
2+
#include "crds/fd_crds.h"
3+
4+
/* This is a very rudimentary implementation of a weighted sampler. We
5+
will eventually want to use a modified fd_wsample. */
6+
struct wpeer_sampler_private {
7+
/* Cumulative weight for each peer.
8+
Individual peer weight can be derived with
9+
cumul_weight[i]-cumul_weight[i-1] */
10+
ulong * cumul_weight;
11+
ulong max_idx;
12+
};
13+
14+
#define PREV_PEER_WEIGHT( ps, idx ) \
15+
( (idx) ? (ps)->cumul_weight[(idx)-1] : 0UL )
16+
17+
18+
FD_FN_CONST ulong
19+
wpeer_sampler_footprint( ulong max_peers ) {
20+
ulong l;
21+
l = FD_LAYOUT_INIT;
22+
l = FD_LAYOUT_APPEND( l, wpeer_sampler_align(), sizeof(wpeer_sampler_t) );
23+
l = FD_LAYOUT_APPEND( l, alignof(ulong), sizeof(ulong)*max_peers );
24+
return FD_LAYOUT_FINI( l, wpeer_sampler_align() );
25+
}
26+
27+
void *
28+
wpeer_sampler_new( void * shmem, ulong max_peers ) {
29+
if( FD_UNLIKELY( !shmem ) ){
30+
FD_LOG_WARNING(( "NULL shmem" ));
31+
return NULL;
32+
}
33+
34+
if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, wpeer_sampler_align() ) ) ) {
35+
FD_LOG_WARNING(( "misaligned shmem" ));
36+
return NULL;
37+
}
38+
39+
FD_SCRATCH_ALLOC_INIT( l, shmem );
40+
wpeer_sampler_t * ws = FD_SCRATCH_ALLOC_APPEND( l, wpeer_sampler_align(), sizeof(wpeer_sampler_t) );
41+
ws->cumul_weight = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), sizeof(ulong)*max_peers );
42+
FD_TEST( FD_SCRATCH_ALLOC_FINI( l, wpeer_sampler_align() )==(ulong)shmem + wpeer_sampler_footprint( max_peers ) );
43+
44+
fd_memset( ws->cumul_weight, 0, sizeof(ulong)*max_peers );
45+
ws->max_idx = ULONG_MAX;
46+
return (void *)ws;
47+
}
48+
49+
wpeer_sampler_t *
50+
wpeer_sampler_join( void * shmem ) {
51+
52+
if( FD_UNLIKELY( !shmem ) ){
53+
FD_LOG_WARNING(( "null shmem" ));
54+
return NULL;
55+
}
56+
57+
if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, wpeer_sampler_align() ) ) ) {
58+
FD_LOG_WARNING(( "misaligned shmem" ));
59+
return NULL;
60+
}
61+
62+
wpeer_sampler_t * ws = (wpeer_sampler_t *) shmem;
63+
64+
return ws;
65+
66+
}
67+
68+
ulong
69+
wpeer_sampler_sample( wpeer_sampler_t const * ps,
70+
fd_rng_t * rng ) {
71+
if( FD_UNLIKELY( ps->max_idx == ULONG_MAX || !ps->cumul_weight[ps->max_idx] ) ) {
72+
return SAMPLE_IDX_SENTINEL;
73+
}
74+
75+
ulong sample = fd_rng_ulong_roll( rng, ps->cumul_weight[ps->max_idx] );
76+
/* avoid sampling 0 */
77+
sample = fd_ulong_min( sample+1UL, ps->cumul_weight[ps->max_idx] );
78+
79+
/* Binary search for the smallest cumulative weight >= sample */
80+
ulong left = 0UL;
81+
ulong right = ps->max_idx+1;
82+
while( left < right ) {
83+
ulong mid = left + (right - left) / 2UL;
84+
if( ps->cumul_weight[mid]<sample ) {
85+
left = mid + 1UL;
86+
} else {
87+
right = mid;
88+
}
89+
}
90+
return left;
91+
}
92+
93+
int
94+
wpeer_sampler_upd( wpeer_sampler_t * ps,
95+
ulong weight,
96+
ulong idx ) {
97+
if( FD_UNLIKELY( !ps ) ) return -1;
98+
99+
/* Special case weight = 0 and idx == max_idx */
100+
if( FD_UNLIKELY( weight==0UL && idx==ps->max_idx ) ) {
101+
ps->cumul_weight[idx] = 0UL;
102+
ps->max_idx = ( idx==0UL ) ? ULONG_MAX : idx-1;
103+
return 0;
104+
}
105+
/* Handle edge case where ps->max_idx is ULONG_MAX (sampler is empty) */
106+
if( FD_UNLIKELY( ps->max_idx == ULONG_MAX ) ) {
107+
ps->max_idx = idx;
108+
} else {
109+
ps->max_idx = fd_ulong_max( ps->max_idx, idx );
110+
}
111+
112+
ulong old_weight = ps->cumul_weight[idx] - PREV_PEER_WEIGHT( ps, idx );
113+
if( FD_UNLIKELY( old_weight==weight ) ) return 0;
114+
115+
if( weight>old_weight ) {
116+
for( ulong i=idx; i<ps->max_idx+1; i++ ) {
117+
ps->cumul_weight[i] += (weight - old_weight);
118+
}
119+
} else {
120+
for( ulong i=idx; i<ps->max_idx+1; i++ ) {
121+
ps->cumul_weight[i] -= (old_weight - weight);
122+
}
123+
}
124+
return 0;
125+
}

‎src/flamenco/gossip/fd_gossip_wpeer_sampler.h‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
#ifndef HEADER_fd_src_flamenco_gossip_fd_gossip_wpeer_sampler_h
2+
#define HEADER_fd_src_flamenco_gossip_fd_gossip_wpeer_sampler_h
3+
4+
#include "../../util/fd_util.h"
5+
6+
/* wpeer_sampler provides a set of APIs to maintain a weighted sampler
7+
with the ability to change weights dynamically on a runtime-bounded
8+
element set. The sampler is designed to be used for sampling peers
9+
in various parts of the gossip protocol. Users supply weights/score
10+
updates in wpeer_sampler_upd and sample with wpeer_sampler_sample.
11+
12+
The sampler works in terms of indices into an array of peers. The
13+
user is responsible for maintaining a mapping between peers and
14+
indices. The sampler does not store any information about the peers
15+
themselves. fd_crds provides a set of APIs to track a peer's contact
16+
info with an index to the contact info sidetable. These are provided
17+
by fd_crds_ci_change_fn callbacks, with an API to lookup the
18+
corresponding Contact Info.
19+
20+
Why not use fd_wsample? The peer population constantly changes
21+
throughout the epoch, as nodes enter, leave, or become
22+
(un)responsive in the cluster. The fd_wsample APIs (currenlty)
23+
do not provide the ability to change individual peer weights without
24+
clearing the sampler and recalculating scores. */
25+
26+
#define SAMPLE_IDX_SENTINEL ULONG_MAX
27+
28+
struct wpeer_sampler_private;
29+
typedef struct wpeer_sampler_private wpeer_sampler_t;
30+
31+
FD_FN_CONST static inline ulong
32+
wpeer_sampler_align( void ) {
33+
return 8UL;
34+
}
35+
36+
FD_FN_CONST ulong
37+
wpeer_sampler_footprint( ulong max_peers );
38+
39+
void *
40+
wpeer_sampler_new( void * shmem, ulong max_peers );
41+
42+
wpeer_sampler_t *
43+
wpeer_sampler_join( void * shmem );
44+
45+
/* wpeer_sampler_sample returns the index of the entry sampled
46+
by the weighted sampler. Does not sample entries with 0 weight. */
47+
ulong
48+
wpeer_sampler_sample( wpeer_sampler_t const * ps,
49+
fd_rng_t * rng );
50+
51+
/* wpeer_sampler_upd updates the weight/score of the entry at index idx
52+
to weight. idx must be in [0, max_peers). weight can be 0, which
53+
effectively disables the entry in the sampler population. */
54+
int
55+
wpeer_sampler_upd( wpeer_sampler_t * ps,
56+
ulong weight,
57+
ulong idx );
58+
59+
60+
#endif

0 commit comments

Comments
 (0)