|
|
|
|
@ -40,7 +40,7 @@ struct stream_optimizer {
|
|
|
|
|
struct stream_selection_settings settings; |
|
|
|
|
struct stream_selection_info info; |
|
|
|
|
struct stream_probe_schedule schedule; |
|
|
|
|
uint64_t roster_ms, progress_ms, next_seq; |
|
|
|
|
uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms; |
|
|
|
|
int progress_valid, draining; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
@ -113,6 +113,11 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
|
|
|
|
|
s->offers[s->probe_sub].info.info_ms = 0; |
|
|
|
|
if (info.reason == STREAM_REASON_TRANSPORT || info.reason == STREAM_REASON_MEMBER || info.reason == STREAM_REASON_PROTOCOL) { |
|
|
|
|
s->failures++; s->retry_ms = now + backoff(s->failures); |
|
|
|
|
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe closed source=%llx peer=%llx sub=%u reason=%d" |
|
|
|
|
" info_ready=%d local_epoch=%llx peer_epoch=%llx failures=%u retry_in_ms=%llu cycle=%llu", |
|
|
|
|
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->probe_sub, info.reason, |
|
|
|
|
info.info_ready, (unsigned long long)info.local_epoch, (unsigned long long)info.peer_epoch, s->failures, |
|
|
|
|
(unsigned long long)(s->retry_ms - now), (unsigned long long)s->cycle); |
|
|
|
|
stream_receive_close(s->probe); s->probe = NULL; continue; |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
@ -421,4 +426,22 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
|
|
|
|
|
if (o->selected[sub].replacement) o->info.overlaps++; |
|
|
|
|
} |
|
|
|
|
if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0; |
|
|
|
|
/* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */ |
|
|
|
|
if (!local.anchored && now >= o->diagnostic_ms) { |
|
|
|
|
o->diagnostic_ms = now + STREAM_PROBE_PERIOD_MS; |
|
|
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stream bootstrap waits group=%llx source=%llx epoch=%llx candidates=%u probes=%u" |
|
|
|
|
" subflows=%u cycle=%llu cycle_age_ms=%llu", |
|
|
|
|
(unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, |
|
|
|
|
(unsigned long long)o->descriptor.source_epoch, o->info.candidates, o->info.probes_active, o->info.subflows, |
|
|
|
|
(unsigned long long)o->schedule.cycle, (unsigned long long)(now - o->schedule.started_ms)); |
|
|
|
|
for (struct stream_supplier *s = o->suppliers; s; s = s->next) { |
|
|
|
|
struct stream_receive_info probe = {0}; |
|
|
|
|
if (s->probe) stream_receive_get_info(s->probe, &probe); |
|
|
|
|
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "stream bootstrap supplier source=%llx peer=%llx online=%d compatible=%d source_peer=%d" |
|
|
|
|
" cycle=%llu failures=%u retry_in_ms=%llu probe=%d sub=%u state=%d info_ready=%d reason=%d", |
|
|
|
|
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->online, s->compatible, s->source, |
|
|
|
|
(unsigned long long)s->cycle, s->failures, (unsigned long long)(s->retry_ms > now ? s->retry_ms - now : 0), |
|
|
|
|
s->probe != NULL, s->probe_sub, probe.state, probe.info_ready, probe.reason); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|