| 80 | } |
| 81 | |
| 82 | bool RuntimeFilter::WaitForArrival(int32_t timeout_ms) const { |
| 83 | unique_lock<mutex> l(arrival_mutex_); |
| 84 | while (arrival_time_.Load() == 0) { |
| 85 | int64_t ms_since_registration = MonotonicMillis() - registration_time_; |
| 86 | int64_t ms_remaining = timeout_ms - ms_since_registration; |
| 87 | if (ms_remaining <= 0) break; |
| 88 | if (injection_delay_ > 0) SleepForMs(injection_delay_); |
| 89 | arrival_cv_.WaitFor(l, ms_remaining * MICROS_PER_MILLI); |
| 90 | } |
| 91 | return arrival_time_.Load() != 0; |
| 92 | } |
| 93 | |
| 94 | void RuntimeFilter::SetIntermediateAggregation(bool is_intermediate_aggregator, |
| 95 | std::string intermediate_krpc_hostname, NetworkAddressPB intermediate_krpc_backend) { |