Line data Source code
1 : #include "fd_accdb_admin_v1.h"
2 : #include "../fd_flamenco_base.h"
3 :
4 : FD_STATIC_ASSERT( alignof(fd_accdb_admin_v1_t)<=alignof(fd_accdb_admin_t), layout );
5 : FD_STATIC_ASSERT( sizeof (fd_accdb_admin_v1_t)<=sizeof(fd_accdb_admin_t), layout );
6 :
7 : fd_accdb_admin_t *
8 : fd_accdb_admin_v1_init( fd_accdb_admin_t * admin_,
9 72 : void * shfunk ) {
10 72 : if( FD_UNLIKELY( !admin_ ) ) {
11 0 : FD_LOG_WARNING(( "NULL ljoin" ));
12 0 : return NULL;
13 0 : }
14 72 : if( FD_UNLIKELY( !shfunk ) ) {
15 0 : FD_LOG_WARNING(( "NULL shfunk" ));
16 0 : return NULL;
17 0 : }
18 :
19 72 : fd_accdb_admin_v1_t * admin = (fd_accdb_admin_v1_t *)admin_;
20 72 : memset( admin, 0, sizeof(fd_accdb_admin_t) );
21 72 : admin->base.accdb_type = FD_ACCDB_TYPE_V1;
22 72 : admin->base.vt = &fd_accdb_admin_v1_vt;
23 :
24 72 : if( FD_UNLIKELY( !fd_funk_join( admin->funk, shfunk ) ) ) {
25 0 : FD_LOG_CRIT(( "fd_funk_join failed" ));
26 0 : }
27 :
28 72 : return admin_;
29 72 : }
30 :
31 : static fd_accdb_admin_v1_t *
32 796962 : downcast( fd_accdb_admin_t * admin ) {
33 796962 : if( FD_UNLIKELY( !admin ) ) {
34 0 : FD_LOG_CRIT(( "NULL admin" ));
35 0 : }
36 796962 : if( FD_UNLIKELY( admin->base.accdb_type!=FD_ACCDB_TYPE_V1 ) ) {
37 0 : FD_LOG_CRIT(( "corrupt accdb_admin handle" ));
38 0 : }
39 796962 : return (fd_accdb_admin_v1_t *)admin;
40 796962 : }
41 :
42 : void
43 66 : fd_accdb_admin_v1_fini( fd_accdb_admin_t * admin_ ) {
44 66 : fd_accdb_admin_v1_t * admin = downcast( admin_ );
45 66 : if( FD_UNLIKELY( !fd_funk_leave( admin->funk, NULL ) ) ) FD_LOG_CRIT(( "fd_funk_leave failed" ));
46 66 : memset( admin, 0, sizeof(fd_accdb_admin_base_t) );
47 66 : }
48 :
49 : fd_funk_t *
50 12 : fd_accdb_admin_v1_funk( fd_accdb_admin_t * admin ) {
51 12 : fd_accdb_admin_v1_t * a = downcast( admin );
52 12 : return a->funk;
53 12 : }
54 :
55 : fd_funk_txn_xid_t
56 24 : fd_accdb_v1_root_get( fd_accdb_admin_t const * admin_ ) {
57 24 : fd_accdb_admin_v1_t const * admin = (fd_accdb_admin_v1_t const *)admin_;
58 24 : return *fd_funk_last_publish( admin->funk );
59 24 : }
60 :
61 : /* Begin transaction-level operations. It is assumed that funk_txn data
62 : structures are not concurrently modified. This includes txn_pool and
63 : txn_map. */
64 :
65 : void
66 : fd_accdb_v1_attach_child( fd_accdb_admin_t * db_,
67 : fd_funk_txn_xid_t const * xid_parent,
68 571872 : fd_funk_txn_xid_t const * xid_new ) {
69 571872 : fd_accdb_admin_v1_t * db = downcast( db_ );
70 571872 : FD_LOG_INFO(( "accdb txn xid %lu:%lu: created with parent %lu:%lu",
71 571872 : xid_new ->ul[0], xid_new ->ul[1],
72 571872 : xid_parent->ul[0], xid_parent->ul[1] ));
73 571872 : fd_funk_txn_prepare( db->funk, xid_parent, xid_new );
74 571872 : }
75 :
76 : static void
77 : fd_accdb_txn_cancel_one( fd_accdb_admin_v1_t * admin,
78 347103 : fd_funk_txn_t * txn ) {
79 347103 : FD_LOG_INFO(( "accdb txn laddr=%p xid %lu:%lu: cancel", (void *)txn, txn->xid.ul[0], txn->xid.ul[1] ));
80 :
81 347103 : if( FD_UNLIKELY( txn->state!=FD_FUNK_TXN_STATE_ACTIVE ) ) {
82 0 : FD_LOG_CRIT(( "cannot cancel xid %lu:%lu: unxpected state %u-%s",
83 0 : txn->xid.ul[0], txn->xid.ul[1],
84 0 : txn->state, fd_funk_txn_state_str( txn->state ) ));
85 0 : }
86 347103 : fd_funk_t * funk = admin->funk;
87 347103 : if( FD_UNLIKELY( !fd_funk_txn_idx_is_null( txn->child_head_cidx ) ||
88 347103 : !fd_funk_txn_idx_is_null( txn->child_tail_cidx ) ) ) {
89 0 : FD_LOG_CRIT(( "fd_accdb_txn_cancel failed: txn at %p with xid %lu:%lu has children (data corruption?)",
90 0 : (void *)txn, txn->xid.ul[0], txn->xid.ul[1] ));
91 0 : }
92 :
93 : /* Phase 1: Drain users from transaction */
94 :
95 347103 : fd_rwlock_write( txn->lock );
96 347103 : FD_COMPILER_MFENCE();
97 347103 : FD_VOLATILE( txn->state ) = FD_FUNK_TXN_STATE_CANCEL;
98 :
99 : /* Phase 2: Detach all records */
100 :
101 347103 : FD_COMPILER_MFENCE();
102 347103 : uint const rec_head_idx = txn->rec_head_idx;
103 347103 : txn->rec_head_idx = FD_FUNK_REC_IDX_NULL;
104 347103 : txn->rec_tail_idx = FD_FUNK_REC_IDX_NULL;
105 347103 : FD_COMPILER_MFENCE();
106 :
107 : /* Phase 3: Remove records */
108 :
109 347103 : ulong rec_cnt = 0UL;
110 347103 : uint rec_idx = rec_head_idx;
111 1030830 : while( !fd_funk_rec_idx_is_null( rec_idx ) ) {
112 683727 : fd_funk_rec_t * rec = &funk->rec_pool->ele[ rec_idx ];
113 683727 : fd_funk_xid_key_pair_t pair = FD_VOLATILE_CONST( rec->pair );
114 :
115 683727 : uint next_idx = rec->next_idx;
116 683727 : if( FD_UNLIKELY( !fd_funk_txn_xid_eq( pair.xid, &txn->xid ) ) ) {
117 0 : FD_LOG_CRIT(( "Record does not belong to txn being cancelled (data corruption?): rec_idx=%u", rec_idx ));
118 0 : }
119 :
120 : /* Phase 3.1: Hide record */
121 :
122 683727 : fd_funk_rec_query_t query[1];
123 683727 : int remove_err = fd_funk_rec_map_remove( funk->rec_map, &pair, NULL, query, FD_MAP_FLAG_BLOCKING );
124 683727 : if( FD_UNLIKELY( remove_err ) ) FD_LOG_CRIT(( "fd_funk_rec_map_remove failed: %i-%s", remove_err, fd_map_strerror( remove_err ) ));
125 683727 : if( FD_UNLIKELY( query->ele!=rec ) ) FD_LOG_CRIT(( "Found duplicate record in map idx[0]=%p idx[1]=%p", (void *)query->ele, (void *)rec ));
126 :
127 : /* Phase 3.2: Mark record as invalid */
128 :
129 683727 : FD_COMPILER_MFENCE();
130 683727 : memset( &rec->pair, 0, sizeof(fd_funk_xid_key_pair_t) );
131 683727 : FD_COMPILER_MFENCE();
132 :
133 : /* Phase 3.3: Free record */
134 :
135 683727 : fd_funk_val_flush( rec, funk->alloc, funk->wksp );
136 683727 : rec->next_idx = FD_FUNK_REC_IDX_NULL;
137 683727 : rec->prev_idx = FD_FUNK_REC_IDX_NULL;
138 683727 : rec->ver_lock = fd_funk_rec_ver_lock( fd_funk_rec_ver_inc( fd_funk_rec_ver_bits( rec->ver_lock ) ), 0UL );
139 683727 : fd_funk_rec_pool_release( funk->rec_pool, rec, 1 );
140 683727 : rec_idx = next_idx;
141 683727 : rec_cnt++;
142 683727 : }
143 347103 : admin->base.revert_cnt += rec_cnt;
144 347103 : FD_LOG_INFO(( "accdb freed %lu records while cancelling txn %lu:%lu",
145 347103 : rec_cnt, txn->xid.ul[0], txn->xid.ul[1] ));
146 :
147 : /* Phase 4: Remove transaction from fork graph */
148 :
149 347103 : uint self_cidx = fd_funk_txn_cidx( (ulong)( txn-funk->txn_pool->ele ) );
150 347103 : uint prev_cidx = txn->sibling_prev_cidx; ulong prev_idx = fd_funk_txn_idx( prev_cidx );
151 347103 : uint next_cidx = txn->sibling_next_cidx; ulong next_idx = fd_funk_txn_idx( next_cidx );
152 347103 : if( !fd_funk_txn_idx_is_null( next_idx ) ) {
153 156813 : funk->txn_pool->ele[ next_idx ].sibling_prev_cidx = prev_cidx;
154 156813 : }
155 347103 : if( !fd_funk_txn_idx_is_null( prev_idx ) ) {
156 109587 : funk->txn_pool->ele[ prev_idx ].sibling_next_cidx = next_cidx;
157 109587 : }
158 347103 : if( !fd_funk_txn_idx_is_null( fd_funk_txn_idx( txn->parent_cidx ) ) ) {
159 169809 : fd_funk_txn_t * parent = &funk->txn_pool->ele[ fd_funk_txn_idx( txn->parent_cidx ) ];
160 169809 : if( parent->child_head_cidx==self_cidx ) parent->child_head_cidx = next_cidx;
161 169809 : if( parent->child_tail_cidx==self_cidx ) parent->child_tail_cidx = prev_cidx;
162 177294 : } else {
163 177294 : if( funk->shmem->child_head_cidx==self_cidx ) funk->shmem->child_head_cidx = next_cidx;
164 177294 : if( funk->shmem->child_tail_cidx==self_cidx ) funk->shmem->child_tail_cidx = prev_cidx;
165 177294 : }
166 :
167 : /* Phase 5: Remove transcation from index */
168 :
169 347103 : fd_funk_txn_map_query_t query[1];
170 347103 : int remove_err = fd_funk_txn_map_remove( funk->txn_map, &txn->xid, NULL, query, FD_MAP_FLAG_BLOCKING );
171 347103 : if( FD_UNLIKELY( remove_err!=FD_MAP_SUCCESS ) ) {
172 0 : FD_LOG_CRIT(( "fd_accdb_txn_cancel failed: fd_funk_txn_map_remove(%lu:%lu) failed: %i-%s",
173 0 : txn->xid.ul[0], txn->xid.ul[1], remove_err, fd_map_strerror( remove_err ) ));
174 0 : }
175 :
176 : /* Phase 6: Free transaction object */
177 :
178 347103 : txn->parent_cidx = UINT_MAX;
179 347103 : txn->sibling_prev_cidx = UINT_MAX;
180 347103 : txn->sibling_next_cidx = UINT_MAX;
181 347103 : fd_rwlock_unwrite( txn->lock );
182 347103 : FD_VOLATILE( txn->state ) = FD_FUNK_TXN_STATE_FREE;
183 347103 : fd_funk_txn_pool_release( funk->txn_pool, txn, 1 );
184 347103 : }
185 :
186 : /* Cancels txn and all children */
187 :
188 : static void
189 : fd_accdb_txn_cancel_tree( fd_accdb_admin_v1_t * accdb,
190 347103 : fd_funk_txn_t * txn ) {
191 516900 : for(;;) {
192 516900 : ulong child_idx = fd_funk_txn_idx( txn->child_head_cidx );
193 516900 : if( fd_funk_txn_idx_is_null( child_idx ) ) break;
194 169797 : fd_funk_txn_t * child = &accdb->funk->txn_pool->ele[ child_idx ];
195 169797 : fd_accdb_txn_cancel_tree( accdb, child );
196 169797 : }
197 347103 : fd_accdb_txn_cancel_one( accdb, txn );
198 347103 : }
199 :
200 : /* Cancels all left/right siblings */
201 :
202 : static void
203 : fd_accdb_txn_cancel_prev_list( fd_accdb_admin_v1_t * accdb,
204 224742 : fd_funk_txn_t * txn ) {
205 224742 : ulong self_idx = (ulong)( txn - accdb->funk->txn_pool->ele );
206 313809 : for(;;) {
207 313809 : ulong prev_idx = fd_funk_txn_idx( txn->sibling_prev_cidx );
208 313809 : if( FD_UNLIKELY( prev_idx==self_idx ) ) FD_LOG_CRIT(( "detected cycle in fork graph" ));
209 313809 : if( fd_funk_txn_idx_is_null( prev_idx ) ) break;
210 89067 : fd_funk_txn_t * sibling = &accdb->funk->txn_pool->ele[ prev_idx ];
211 89067 : fd_accdb_txn_cancel_tree( accdb, sibling );
212 89067 : }
213 224742 : }
214 :
215 : static void
216 : fd_accdb_txn_cancel_next_list( fd_accdb_admin_v1_t * accdb,
217 224793 : fd_funk_txn_t * txn ) {
218 224793 : ulong self_idx = (ulong)( txn - accdb->funk->txn_pool->ele );
219 312981 : for(;;) {
220 312981 : ulong next_idx = fd_funk_txn_idx( txn->sibling_next_cidx );
221 312981 : if( FD_UNLIKELY( next_idx==self_idx ) ) FD_LOG_CRIT(( "detected cycle in fork graph" ));
222 312981 : if( fd_funk_txn_idx_is_null( next_idx ) ) break;
223 88188 : fd_funk_txn_t * sibling = &accdb->funk->txn_pool->ele[ next_idx ];
224 88188 : fd_accdb_txn_cancel_tree( accdb, sibling );
225 88188 : }
226 224793 : }
227 :
228 : void
229 : fd_accdb_txn_cancel_siblings( fd_accdb_admin_v1_t * accdb,
230 224742 : fd_funk_txn_t * txn ) {
231 224742 : fd_accdb_txn_cancel_prev_list( accdb, txn );
232 224742 : fd_accdb_txn_cancel_next_list( accdb, txn );
233 224742 : txn->sibling_prev_cidx = UINT_MAX;
234 224742 : txn->sibling_next_cidx = UINT_MAX;
235 224742 : }
236 :
237 : void
238 : fd_accdb_v1_cancel( fd_accdb_admin_t * accdb_,
239 51 : fd_funk_txn_xid_t const * xid ) {
240 51 : fd_accdb_admin_v1_t * accdb = downcast( accdb_ );
241 51 : fd_funk_t * funk = accdb->funk;
242 :
243 : /* Assume no concurrent access to txn_map */
244 :
245 51 : fd_funk_txn_map_query_t query[1];
246 51 : int query_err = fd_funk_txn_map_query_try( funk->txn_map, xid, NULL, query, 0 );
247 51 : if( FD_UNLIKELY( query_err ) ) {
248 0 : FD_LOG_CRIT(( "fd_accdb_cancel failed: fd_funk_txn_map_query_try(xid=%lu:%lu) returned (%i-%s)",
249 0 : xid->ul[0], xid->ul[1], query_err, fd_map_strerror( query_err ) ));
250 0 : }
251 51 : fd_funk_txn_t * txn = fd_funk_txn_map_query_ele( query );
252 :
253 51 : fd_accdb_txn_cancel_next_list( accdb, txn );
254 51 : fd_accdb_txn_cancel_tree( accdb, txn );
255 51 : }
256 :
257 : /* fd_accdb_chain_reclaim "reclaims" a zero-lamport account by removing
258 : its underlying record. */
259 :
260 : static void
261 : fd_accdb_chain_reclaim( fd_accdb_admin_v1_t * accdb,
262 53511 : fd_funk_rec_t * rec ) {
263 53511 : fd_funk_t * funk = accdb->funk;
264 :
265 : /* Phase 1: Remove record from map */
266 :
267 53511 : fd_funk_xid_key_pair_t pair = rec->pair;
268 53511 : fd_funk_rec_query_t query[1];
269 53511 : int rm_err = fd_funk_rec_map_remove( funk->rec_map, &pair, NULL, query, FD_MAP_FLAG_BLOCKING );
270 53511 : if( FD_UNLIKELY( rm_err!=FD_MAP_SUCCESS ) ) FD_LOG_CRIT(( "fd_funk_rec_map_remove failed (%i-%s)", rm_err, fd_map_strerror( rm_err ) ));
271 53511 : FD_COMPILER_MFENCE();
272 :
273 : /* Phase 2: Invalidate record */
274 :
275 53511 : fd_funk_rec_t * old_rec = query->ele;
276 53511 : memset( &old_rec->pair, 0, sizeof(fd_funk_xid_key_pair_t) );
277 53511 : FD_COMPILER_MFENCE();
278 :
279 : /* Phase 3: Free record */
280 :
281 53511 : old_rec->map_next = FD_FUNK_REC_IDX_NULL;
282 53511 : fd_funk_val_flush( old_rec, funk->alloc, funk->wksp );
283 53511 : fd_funk_rec_pool_release( funk->rec_pool, old_rec, 1 );
284 53511 : accdb->base.reclaim_cnt++;
285 53511 : }
286 :
287 : /* fd_accdb_chain_gc_root cleans up a stale "rooted" version of a
288 : record. */
289 :
290 : static void
291 : fd_accdb_chain_gc_root( fd_accdb_admin_v1_t * accdb,
292 593460 : fd_funk_xid_key_pair_t const * pair ) {
293 593460 : fd_funk_t * funk = accdb->funk;
294 :
295 : /* Phase 1: Remove record from map if found */
296 :
297 593460 : fd_funk_rec_query_t query[1];
298 593460 : int rm_err = fd_funk_rec_map_remove( funk->rec_map, pair, NULL, query, FD_MAP_FLAG_BLOCKING );
299 593460 : if( rm_err==FD_MAP_ERR_KEY ) return;
300 539736 : if( FD_UNLIKELY( rm_err!=FD_MAP_SUCCESS ) ) FD_LOG_CRIT(( "fd_funk_rec_map_remove failed (%i-%s)", rm_err, fd_map_strerror( rm_err ) ));
301 539736 : FD_COMPILER_MFENCE();
302 :
303 : /* Phase 2: Invalidate record */
304 :
305 539736 : fd_funk_rec_t * old_rec = query->ele;
306 539736 : memset( &old_rec->pair, 0, sizeof(fd_funk_xid_key_pair_t) );
307 539736 : FD_COMPILER_MFENCE();
308 :
309 : /* Phase 3: Free record */
310 :
311 539736 : old_rec->map_next = FD_FUNK_REC_IDX_NULL;
312 539736 : fd_funk_val_flush( old_rec, funk->alloc, funk->wksp );
313 539736 : fd_funk_rec_pool_release( funk->rec_pool, old_rec, 1 );
314 539736 : accdb->base.gc_root_cnt++;
315 539736 : }
316 :
317 : /* fd_accdb_publish_recs moves all records in a transaction to the DB
318 : root. Currently, the DB root is stored by funk, which might change
319 : in the future.
320 :
321 : It is assumed at this point that the txn has no more concurrent
322 : users. */
323 :
324 : static void
325 : fd_accdb_publish_recs( fd_accdb_admin_v1_t * accdb,
326 224742 : fd_funk_txn_t * txn ) {
327 : /* Iterate record list */
328 224742 : uint head = txn->rec_head_idx;
329 224742 : txn->rec_head_idx = FD_FUNK_REC_IDX_NULL;
330 224742 : txn->rec_tail_idx = FD_FUNK_REC_IDX_NULL;
331 224742 : fd_wksp_t * funk_wksp = accdb->funk->wksp;
332 818202 : while( !fd_funk_rec_idx_is_null( head ) ) {
333 593460 : fd_funk_rec_t * rec = &accdb->funk->rec_pool->ele[ head ];
334 :
335 : /* Evict previous value from hash chain */
336 593460 : fd_funk_xid_key_pair_t pair[1];
337 593460 : fd_funk_rec_key_copy( pair->key, rec->pair.key );
338 593460 : fd_funk_txn_xid_set_root( pair->xid );
339 593460 : fd_accdb_chain_gc_root( accdb, pair );
340 :
341 : /* Root or reclaim record */
342 593460 : uint next = rec->next_idx;
343 593460 : fd_account_meta_t const * meta = fd_funk_val( rec, funk_wksp );
344 593460 : FD_CRIT( meta && rec->val_sz>=sizeof(fd_account_meta_t), "invalid funk record value" );
345 593460 : if( !meta->lamports ) {
346 : /* Remove record */
347 53511 : fd_accdb_chain_reclaim( accdb, rec );
348 539949 : } else {
349 : /* Migrate record to root */
350 539949 : rec->prev_idx = FD_FUNK_REC_IDX_NULL;
351 539949 : rec->next_idx = FD_FUNK_REC_IDX_NULL;
352 539949 : fd_funk_txn_xid_t const root = { .ul = { ULONG_MAX, ULONG_MAX } };
353 539949 : fd_funk_txn_xid_st_atomic( rec->pair.xid, &root );
354 539949 : accdb->base.root_cnt++;
355 539949 : }
356 :
357 593460 : head = next; /* next record */
358 593460 : }
359 224742 : }
360 :
361 : /* fd_accdb_txn_publish_one merges an in-prep transaction whose
362 : parent is the last published, into the parent. */
363 :
364 : static void
365 : fd_accdb_txn_publish_one( fd_accdb_admin_v1_t * accdb,
366 224742 : fd_funk_txn_t * txn ) {
367 224742 : fd_funk_t * funk = accdb->funk;
368 :
369 : /* Phase 1: Mark transaction as "last published" */
370 :
371 224742 : fd_funk_txn_xid_t xid[1]; fd_funk_txn_xid_copy( xid, fd_funk_txn_xid( txn ) );
372 224742 : if( FD_UNLIKELY( !fd_funk_txn_idx_is_null( fd_funk_txn_idx( txn->parent_cidx ) ) ) ) {
373 0 : FD_LOG_CRIT(( "fd_accdb_publish failed: txn with xid %lu:%lu is not a child of the last published txn", xid->ul[0], xid->ul[1] ));
374 0 : }
375 224742 : fd_funk_txn_xid_st_atomic( funk->shmem->last_publish, xid );
376 224742 : FD_LOG_INFO(( "accdb txn laddr=%p xid %lu:%lu: publish", (void *)txn, txn->xid.ul[0], txn->xid.ul[1] ));
377 :
378 : /* Phase 2: Drain users from transaction */
379 :
380 224742 : fd_rwlock_write( txn->lock );
381 224742 : FD_VOLATILE( txn->state ) = FD_FUNK_TXN_STATE_PUBLISH;
382 :
383 : /* Phase 3: Migrate records */
384 :
385 224742 : fd_accdb_publish_recs( accdb, txn );
386 :
387 : /* Phase 4: Remove transaction from fork graph
388 :
389 : Because the transaction has no more records, removing it from the
390 : fork graph has no visible side effects to concurrent query ops
391 : (always return "no found") or insert ops (refuse to write to a
392 : "publish" state txn). */
393 :
394 224742 : { /* Adjust the parent pointers of the children to point to "last published" */
395 224742 : ulong child_idx = fd_funk_txn_idx( txn->child_head_cidx );
396 372297 : while( FD_UNLIKELY( !fd_funk_txn_idx_is_null( child_idx ) ) ) {
397 147555 : funk->txn_pool->ele[ child_idx ].parent_cidx = fd_funk_txn_cidx( FD_FUNK_TXN_IDX_NULL );
398 147555 : child_idx = fd_funk_txn_idx( funk->txn_pool->ele[ child_idx ].sibling_next_cidx );
399 147555 : }
400 224742 : }
401 :
402 : /* Phase 5: Remove transaction from index
403 :
404 : The transaction is now an orphan and won't get any new records. */
405 :
406 224742 : fd_funk_txn_map_query_t query[1];
407 224742 : int remove_err = fd_funk_txn_map_remove( funk->txn_map, xid, NULL, query, 0 );
408 224742 : if( FD_UNLIKELY( remove_err!=FD_MAP_SUCCESS ) ) {
409 0 : FD_LOG_CRIT(( "fd_accdb_publish failed: fd_funk_txn_map_remove failed: %i-%s", remove_err, fd_map_strerror( remove_err ) ));
410 0 : }
411 :
412 : /* Phase 6: Free transaction object */
413 :
414 224742 : fd_rwlock_unwrite( txn->lock );
415 224742 : FD_VOLATILE( txn->state ) = FD_FUNK_TXN_STATE_FREE;
416 224742 : txn->parent_cidx = UINT_MAX;
417 224742 : txn->sibling_prev_cidx = UINT_MAX;
418 224742 : txn->sibling_next_cidx = UINT_MAX;
419 224742 : txn->child_head_cidx = UINT_MAX;
420 224742 : txn->child_tail_cidx = UINT_MAX;
421 224742 : fd_funk_txn_pool_release( funk->txn_pool, txn, 1 );
422 224742 : }
423 :
424 : void
425 : fd_accdb_v1_advance_root( fd_accdb_admin_t * accdb_,
426 224742 : fd_funk_txn_xid_t const * xid ) {
427 224742 : fd_accdb_admin_v1_t * accdb = downcast( accdb_ );
428 224742 : fd_funk_t * funk = accdb->funk;
429 :
430 : /* Assume no concurrent access to txn_map */
431 :
432 224742 : fd_funk_txn_map_query_t query[1];
433 224742 : int query_err = fd_funk_txn_map_query_try( funk->txn_map, xid, NULL, query, 0 );
434 224742 : if( FD_UNLIKELY( query_err ) ) {
435 0 : FD_LOG_CRIT(( "fd_accdb_advance_root failed: fd_funk_txn_map_query_try(xid=%lu:%lu) returned (%i-%s)",
436 0 : xid->ul[0], xid->ul[1], query_err, fd_map_strerror( query_err ) ));
437 0 : }
438 224742 : fd_funk_txn_t * txn = fd_funk_txn_map_query_ele( query );
439 :
440 224742 : if( FD_UNLIKELY( !fd_funk_txn_idx_is_null( fd_funk_txn_idx( txn->parent_cidx ) ) ) ) {
441 0 : FD_LOG_CRIT(( "fd_accdb_txn_advance_root: parent of txn %lu:%lu is not root", xid->ul[0], xid->ul[1] ));
442 0 : }
443 :
444 224742 : FD_LOG_INFO(( "accdb txn laddr=%p xid %lu:%lu: advancing root",
445 224742 : (void *)txn,
446 224742 : xid->ul[0], xid->ul[1] ));
447 :
448 224742 : fd_accdb_txn_cancel_siblings( accdb, txn );
449 :
450 : /* Children of transaction are now children of root */
451 224742 : funk->shmem->child_head_cidx = txn->child_head_cidx;
452 224742 : funk->shmem->child_tail_cidx = txn->child_tail_cidx;
453 :
454 224742 : fd_accdb_txn_publish_one( accdb, txn );
455 224742 : }
456 :
457 : /* reset_rec_map frees all records in a funk instance. */
458 :
459 : static void
460 21 : reset_rec_map( fd_funk_t * funk ) {
461 21 : fd_wksp_t * wksp = funk->wksp;
462 21 : fd_alloc_t * alloc = funk->alloc;
463 21 : fd_funk_rec_map_t * rec_map = funk->rec_map;
464 21 : fd_funk_rec_pool_t * rec_pool = funk->rec_pool;
465 :
466 21 : ulong chain_cnt = fd_funk_rec_map_chain_cnt( rec_map );
467 1077 : for( ulong chain_idx=0UL; chain_idx<chain_cnt; chain_idx++ ) {
468 1056 : for(
469 1056 : fd_funk_rec_map_iter_t iter = fd_funk_rec_map_iter( rec_map, chain_idx );
470 1287 : !fd_funk_rec_map_iter_done( iter );
471 1056 : ) {
472 231 : fd_funk_rec_t * rec = fd_funk_rec_map_iter_ele( iter );
473 231 : ulong next = fd_funk_rec_map_private_idx( rec->map_next );;
474 :
475 : /* Remove rec object from map */
476 231 : fd_funk_rec_map_query_t rec_query[1];
477 231 : int err = fd_funk_rec_map_remove( rec_map, fd_funk_rec_pair( rec ), NULL, rec_query, FD_MAP_FLAG_BLOCKING );
478 231 : fd_funk_rec_key_t key; fd_funk_rec_key_copy( &key, rec->pair.key );
479 231 : if( FD_UNLIKELY( err!=FD_MAP_SUCCESS ) ) FD_LOG_CRIT(( "fd_funk_rec_map_remove failed (%i-%s)", err, fd_map_strerror( err ) ));
480 :
481 : /* Free rec resources */
482 231 : rec->map_next = FD_FUNK_REC_IDX_NULL;
483 231 : rec->next_idx = FD_FUNK_REC_IDX_NULL;
484 231 : rec->prev_idx = FD_FUNK_REC_IDX_NULL;
485 231 : memset( &rec->pair, 0, sizeof(fd_funk_xid_key_pair_t) );
486 231 : fd_funk_val_flush( rec, alloc, wksp );
487 231 : fd_funk_rec_pool_release( rec_pool, rec, 1 );
488 231 : iter.ele_idx = next;
489 231 : }
490 1056 : }
491 21 : }
492 :
493 : /* clear_txn_list does a depth-first traversal of the txn tree.
494 : Removes all txns. */
495 :
496 : static void
497 : clear_txn_list( fd_funk_t * funk,
498 33 : ulong txn_head_idx ) {
499 33 : fd_funk_txn_pool_t * txn_pool = funk->txn_pool;
500 33 : fd_funk_txn_map_t * txn_map = funk->txn_map;
501 33 : for( ulong idx = txn_head_idx;
502 45 : !fd_funk_txn_idx_is_null( idx );
503 33 : ) {
504 12 : fd_funk_txn_t * txn = &txn_pool->ele[ idx ];
505 12 : fd_funk_txn_state_assert( txn, FD_FUNK_TXN_STATE_ACTIVE );
506 12 : ulong next_idx = fd_funk_txn_idx( txn->sibling_next_cidx );
507 12 : ulong child_idx = fd_funk_txn_idx( txn->child_head_cidx );
508 12 : txn->rec_head_idx = FD_FUNK_REC_IDX_NULL;
509 12 : txn->rec_tail_idx = FD_FUNK_REC_IDX_NULL;
510 12 : txn->child_head_cidx = UINT_MAX;
511 12 : txn->child_tail_cidx = UINT_MAX;
512 12 : txn->parent_cidx = UINT_MAX;
513 12 : txn->sibling_prev_cidx = UINT_MAX;
514 12 : txn->sibling_next_cidx = UINT_MAX;
515 12 : clear_txn_list( funk, child_idx );
516 12 : fd_funk_txn_map_query_t query[1];
517 12 : int rm_err = fd_funk_txn_map_remove( txn_map, &txn->xid, NULL, query, FD_MAP_FLAG_BLOCKING );
518 12 : if( FD_UNLIKELY( rm_err!=FD_MAP_SUCCESS ) ) FD_LOG_CRIT(( "fd_funk_txn_map_remove failed (%i-%s)", rm_err, fd_map_strerror( rm_err ) ));
519 12 : txn->state = FD_FUNK_TXN_STATE_FREE;
520 12 : int free_err = fd_funk_txn_pool_release( txn_pool, txn, 1 );
521 12 : if( FD_UNLIKELY( free_err!=FD_POOL_SUCCESS ) ) FD_LOG_CRIT(( "fd_funk_txn_pool_release failed (%i)", free_err ));
522 12 : idx = next_idx;
523 12 : }
524 33 : funk->shmem->child_head_cidx = UINT_MAX;
525 33 : funk->shmem->child_tail_cidx = UINT_MAX;
526 33 : }
527 :
528 : void
529 21 : fd_accdb_v1_clear( fd_accdb_admin_t * accdb_ ) {
530 21 : fd_accdb_admin_v1_t * accdb = downcast( accdb_ );
531 21 : fd_funk_t * funk = accdb->funk;
532 21 : clear_txn_list( funk, fd_funk_txn_idx( funk->shmem->child_head_cidx ) );
533 21 : reset_rec_map( funk );
534 21 : }
535 :
536 : void
537 198 : fd_accdb_v1_verify( fd_accdb_admin_t * accdb_ ) {
538 198 : fd_accdb_admin_v1_t * accdb = downcast( accdb_ );
539 198 : FD_TEST( fd_funk_verify( accdb->funk )==FD_FUNK_SUCCESS );
540 198 : }
541 :
542 : fd_accdb_admin_vt_t const fd_accdb_admin_v1_vt = {
543 : .fini = fd_accdb_admin_v1_fini,
544 : .root_get = fd_accdb_v1_root_get,
545 : .attach_child = fd_accdb_v1_attach_child,
546 : .advance_root = fd_accdb_v1_advance_root,
547 : .cancel = fd_accdb_v1_cancel
548 : };
|
