| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #define STREAM_INTERNALS |
| 4 | #include "stream-waiting-list.h" |
| 5 | |
| 6 | #define ACCEPT_NODES_EVERY_UT (5 * USEC_PER_SEC) |
| 7 | |
| 8 | static __thread struct { |
| 9 | size_t metadata; |
| 10 | size_t replication; |
| 11 | } throttle = { 0 }; |
| 12 | |
| 13 | void stream_thread_received_metadata(void) { |
| 14 | throttle.metadata++; |
| 15 | } |
| 16 | void stream_thread_received_replication(void) { |
| 17 | throttle.replication++; |
| 18 | } |
| 19 | |
| 20 | static inline size_t normalize_value(size_t v) { |
| 21 | return (v / 100) * 100; |
| 22 | } |
| 23 | |
| 24 | void stream_thread_process_waiting_list_unsafe(struct stream_thread *sth, usec_t now_ut) { |
| 25 | internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ ); |
| 26 | |
| 27 | Word_t idx = 0; |
| 28 | struct receiver_state *rpt = RECEIVERS_FIRST(&sth->queue.receivers, &idx); |
| 29 | if(!rpt) return; |
| 30 | |
| 31 | if(sth->waiting_list.last_accepted_ut + ACCEPT_NODES_EVERY_UT > now_ut || |
| 32 | !stream_control_children_should_be_accepted()) |
| 33 | return; |
| 34 | |
| 35 | size_t n_metadata = normalize_value(throttle.metadata); |
| 36 | size_t n_replication = normalize_value(throttle.replication); |
| 37 | |
| 38 | if(sth->waiting_list.metadata != n_metadata || |
| 39 | sth->waiting_list.replication != n_replication) { |
| 40 | sth->waiting_list.metadata = n_metadata; |
| 41 | sth->waiting_list.replication = n_replication; |
| 42 | return; |
| 43 | } |
| 44 | |
| 45 | RECEIVERS_DEL(&sth->queue.receivers, idx); |
| 46 | stream_receiver_move_to_running_unsafe(sth, rpt); |
| 47 | sth->queue.receivers_waiting--; |
| 48 | } |