diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 3c8ef820..0fdaafa7 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -628,6 +628,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st stream_buffer_pending(local->buffer, f->info.next, f->subblock, &count, NULL); load_account(f, count, count * stream_block_size(&f->descriptor), 1, 0); f->pending_highest = stream_read_be(payload + 24, 8); + if (f->pending_highest == UINT64_MAX) f->pending_highest = 0; const uint8_t *wire; size_t length; uint64_t seq; if (stream_buffer_read(local->buffer, UINT64_MAX, f->subblock, &wire, &length, &seq) == STREAM_OK) stream_metrics_observe(&f->offered_metrics, seq, (uint16_t)stream_read_be(wire + STREAM_BLOCK_TIMESTAMP_OFF, 2), diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 9c135744..200dc3e8 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -192,11 +192,17 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig double score = problems(&info->supplier_metrics, info->supplier_metrics_valid); double old_score = problems(&previous->supplier_metrics, previous->supplier_metrics_valid); uint64_t rtt = total_rtt(info), old_rtt = total_rtt(previous); - if (score < old_score || (score == old_score && (rtt < old_rtt - || (rtt == old_rtt && (shared_path(o, info, sub) < shared_path(o, previous, sub) - || (shared_path(o, info, sub) == shared_path(o, previous, sub) - && (info->path.count < previous->path.count || (info->path.count == previous->path.count && s->node < best->node)))))))) - best = s; + int better = score < old_score; + if (score == old_score) { + better = rtt < old_rtt; + if (rtt == old_rtt) { + unsigned shared = shared_path(o, info, sub), old_shared = shared_path(o, previous, sub); + better = shared < old_shared; + if (shared == old_shared) better = info->path.count < previous->path.count + || (info->path.count == previous->path.count && s->node < best->node); + } + } + if (better) best = s; } } return best; @@ -277,7 +283,15 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, if (pick->current) { stream_receive_get_info(pick->current, &info); if (info.state == STREAM_RECEIVE_CLOSED || !supplier_find(o, info.peer)) receive_failed(o, &pick->current, sub, now); - else { if (info.blocks) received++; busy++; } + else { + if (info.blocks) received++; busy++; + struct stream_supplier *supplier = supplier_find(o, info.peer); + if (supplier->offers && info.metrics_valid) { + supplier->offers[sub].info.supplier_metrics = info.metrics; + supplier->offers[sub].info.supplier_metrics_valid = 1; + supplier->offers[sub].info.last_available = info.next - 1; + } + } } if (!pick->replacement) continue; struct stream_receive_info replacement; stream_receive_get_info(pick->replacement, &replacement);