* Dispatch the commit callbacks registered on this txg to worker threads. * * If no callbacks are registered for a given TXG, nothing happens. * This function creates a taskq for the associated pool, if needed. */
| 423 | * This function creates a taskq for the associated pool, if needed. |
| 424 | */ |
| 425 | static void |
| 426 | txg_dispatch_callbacks(dsl_pool_t *dp, uint64_t txg) |
| 427 | { |
| 428 | int c; |
| 429 | tx_state_t *tx = &dp->dp_tx; |
| 430 | list_t *cb_list; |
| 431 | |
| 432 | for (c = 0; c < max_ncpus; c++) { |
| 433 | tx_cpu_t *tc = &tx->tx_cpu[c]; |
| 434 | /* |
| 435 | * No need to lock tx_cpu_t at this point, since this can |
| 436 | * only be called once a txg has been synced. |
| 437 | */ |
| 438 | |
| 439 | int g = txg & TXG_MASK; |
| 440 | |
| 441 | if (list_is_empty(&tc->tc_callbacks[g])) |
| 442 | continue; |
| 443 | |
| 444 | if (tx->tx_commit_cb_taskq == NULL) { |
| 445 | /* |
| 446 | * Commit callback taskq hasn't been created yet. |
| 447 | */ |
| 448 | tx->tx_commit_cb_taskq = taskq_create("tx_commit_cb", |
| 449 | 100, defclsyspri, boot_ncpus, boot_ncpus * 2, |
| 450 | TASKQ_PREPOPULATE | TASKQ_DYNAMIC | |
| 451 | TASKQ_THREADS_CPU_PCT); |
| 452 | } |
| 453 | |
| 454 | cb_list = kmem_alloc(sizeof (list_t), KM_SLEEP); |
| 455 | list_create(cb_list, sizeof (dmu_tx_callback_t), |
| 456 | offsetof(dmu_tx_callback_t, dcb_node)); |
| 457 | |
| 458 | list_move_tail(cb_list, &tc->tc_callbacks[g]); |
| 459 | |
| 460 | (void) taskq_dispatch(tx->tx_commit_cb_taskq, (task_func_t *) |
| 461 | txg_do_callbacks, cb_list, TQ_SLEEP); |
| 462 | } |
| 463 | } |
| 464 | |
| 465 | /* |
| 466 | * Wait for pending commit callbacks of already-synced transactions to finish |
no test coverage detected