Scan past blocked DATA while prioritizing sendable retries
This commit is contained in:
evgeny committed
1 parent
dcd6cd93c4
commit
ff74ac8295
2 files changed
+63
-34
No files matched your search
+62
-33
@@ -780,30 +780,20 @@ int etcp_int_send(struct ETCP_CONN* etcp, const void* data, uint16_t len) {
|
||||
|
||||
|
||||
|
||||
static void input_queue_try_push(struct ETCP_CONN* etcp) {// пробуем протолкнуть при отправке
|
||||
static int input_queue_try_push(struct ETCP_CONN* etcp) {// подгрузить не более одного нового фрагмента
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
|
||||
|
||||
// когда очередь отправки пуста - пробуем взять новый пакет на обработку
|
||||
size_t wait_ack_bytes = queue_total_bytes(etcp->input_wait_ack);
|
||||
if (wait_ack_bytes <= etcp->optimal_inflight) {
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] resume input queue: inflight_bytes=%d, input_len=%d", etcp->log_name, wait_ack_bytes, etcp->input_queue->total_bytes);
|
||||
if (etcp->input_queue->count > 0) input_queue_cb(etcp->input_queue, etcp);
|
||||
else queue_resume_callback(etcp->input_queue);// и только когда больше нечего отправлять - забираем новый пакет
|
||||
}
|
||||
if (etcp->unacked_bytes >= etcp->optimal_inflight) return 0;
|
||||
if (etcp->input_queue->count > 0) { input_queue_cb(etcp->input_queue, etcp); return 1; }
|
||||
queue_resume_callback(etcp->input_queue);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void input_queue_try_resume(struct ETCP_CONN* etcp) {// при ACK
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "");
|
||||
|
||||
// Сперва отправим всё из очереди отправки
|
||||
size_t send_q_bytes = queue_total_bytes(etcp->input_send_q);
|
||||
// queue_resume_callback(etcp->input_send_q);// вызвать лишний раз resume не страшно.
|
||||
if (send_q_bytes>0) return;
|
||||
|
||||
// когда очередь отправки пуста - пробуем взять новый пакет на обработку
|
||||
size_t wait_ack_bytes = queue_total_bytes(etcp->input_wait_ack);
|
||||
if (wait_ack_bytes <= etcp->optimal_inflight) {
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] resume input queue: inflight_bytes=%d, input_len=%d", etcp->log_name, wait_ack_bytes, etcp->input_queue->total_bytes);
|
||||
if (etcp->unacked_bytes < etcp->optimal_inflight) {
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] resume input queue: unacked_bytes=%u input_len=%zu", etcp->log_name,
|
||||
etcp->unacked_bytes, etcp->input_queue->total_bytes);
|
||||
queue_resume_callback(etcp->input_queue);// и только когда больше нечего отправлять - забираем новый пакет
|
||||
}
|
||||
}
|
||||
@@ -877,6 +867,12 @@ static void input_queue_cb(struct ll_queue* q, void* arg) {
|
||||
etcp->log_name, etcp->next_tx_id, etcp->rx_ack_till, q->count);
|
||||
ack_window_arm(etcp); return;
|
||||
}
|
||||
if (q->head && (q->head->len > etcp->optimal_inflight ||
|
||||
etcp->unacked_bytes > etcp->optimal_inflight - q->head->len)) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] input held by DATA budget: unacked=%u limit=%u next_bytes=%u send_q=%d",
|
||||
etcp->log_name, etcp->unacked_bytes, etcp->optimal_inflight, q->head->len, etcp->input_send_q->count);
|
||||
return;
|
||||
}
|
||||
struct ETCP_FRAGMENT* in_pkt = (struct ETCP_FRAGMENT*)queue_data_get(q);
|
||||
|
||||
if (!in_pkt) {
|
||||
@@ -1054,28 +1050,58 @@ static void write_le(uint8_t* data, uint64_t value, unsigned size) {
|
||||
for (unsigned i = 0; i < size; i++) data[i] = (uint8_t)(value >> (i * 8));
|
||||
}
|
||||
|
||||
/* Один просмотр без извлечения/резервов: первый готовый повтор, иначе первый готовый новый DATA.
|
||||
* Сохранённый кандидат живёт весь просмотр: здесь нет callbacks, poll или изменения очереди. */
|
||||
static struct INFLIGHT_PACKET* send_queue_select(struct ETCP_CONN* conn, struct ETCP_LINK** chosen) {
|
||||
struct ETCP_LINK* burst = NULL;
|
||||
for (struct ETCP_LINK* l = conn->links; l; l = l->next)
|
||||
if (l->burst_active && l->initialized && l->link_status) { burst = l; break; }
|
||||
struct INFLIGHT_PACKET* fresh = NULL;
|
||||
struct ETCP_LINK* fresh_link = NULL;
|
||||
unsigned examined = 0;
|
||||
for (struct ll_entry* e = conn->input_send_q->head; e; e = e->next) {
|
||||
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)e;
|
||||
examined++;
|
||||
if (!packet->last_attempt_id && fresh) continue;
|
||||
struct ETCP_LINK* link;
|
||||
if (burst && !packet->last_attempt_id) {
|
||||
if (conn->active_attempts >= 2 * MAX_INFLIGHT_SIZE || burst->attempts->count >= MAX_INFLIGHT_SIZE ||
|
||||
packet->ll.len > burst->inflight_lim_bytes || burst->inflight_bytes > burst->inflight_lim_bytes - packet->ll.len) {
|
||||
burst->cwnd_limited = 1; continue;
|
||||
}
|
||||
link = burst;
|
||||
} else link = etcp_loadbalancer_select_link(conn, packet->ll.len, packet->last_attempt_id ? packet : NULL);
|
||||
if (!link || (burst && link != burst)) continue;
|
||||
if (packet->last_attempt_id) {
|
||||
*chosen = link;
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] send scan examined=%u retry_seq=%u link=%u", conn->log_name,
|
||||
examined, packet->seq, link->local_link_id);
|
||||
return packet;
|
||||
}
|
||||
fresh = packet; fresh_link = link;
|
||||
}
|
||||
*chosen = fresh ? fresh_link : burst;
|
||||
DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] send scan examined=%u fresh_seq=%u link=%u", conn->log_name, examined,
|
||||
fresh ? fresh->seq : 0, *chosen ? (*chosen)->local_link_id : 0);
|
||||
return fresh;
|
||||
}
|
||||
|
||||
struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
|
||||
if (!conn || conn->state == 2 || conn->close_requested) return NULL;
|
||||
if (!conn->initialized || conn->session_required || conn->reinit_pending) {
|
||||
if (conn->tx_state) conn->tx_state = ETCP_TX_STATE_LINK_WAIT;
|
||||
return NULL;
|
||||
}
|
||||
if (!conn->input_send_q->head) input_queue_try_push(conn);
|
||||
// Callback input_queue мог сменить эпоху при исчерпании seq или OOM; старый проход отправки прекращается.
|
||||
if (conn->close_requested || conn->state == 2 || !conn->initialized || conn->session_required || conn->reinit_pending) return NULL;
|
||||
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)conn->input_send_q->head;
|
||||
struct ETCP_LINK* link = NULL;
|
||||
for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next)
|
||||
if (candidate->burst_active && candidate->initialized && candidate->link_status) { link = candidate; break; }
|
||||
if (link && packet) {
|
||||
if (packet->last_attempt_id && etcp_loadbalancer_select_link(conn, packet->ll.len, packet) != link) packet = NULL;
|
||||
if (packet && (conn->active_attempts >= 2 * MAX_INFLIGHT_SIZE || link->attempts->count >= MAX_INFLIGHT_SIZE ||
|
||||
packet->ll.len > link->inflight_lim_bytes ||
|
||||
link->inflight_bytes > link->inflight_lim_bytes - packet->ll.len)) {
|
||||
link->cwnd_limited = 1; packet = NULL;
|
||||
}
|
||||
struct INFLIGHT_PACKET* packet = send_queue_select(conn, &link);
|
||||
if (!packet) {
|
||||
uint32_t reset = conn->reset_count;
|
||||
int pulled = input_queue_try_push(conn);
|
||||
// Подгрузка может сменить эпоху/закрыть conn либо удалить burst-линк: прежние кандидаты больше не используются.
|
||||
if (conn->close_requested || conn->state == 2 || conn->reset_count != reset || !conn->initialized ||
|
||||
conn->session_required || conn->reinit_pending) return NULL;
|
||||
if (pulled) packet = send_queue_select(conn, &link);
|
||||
}
|
||||
if (!link && packet) link = etcp_loadbalancer_select_link(conn, packet->ll.len, packet->last_attempt_id ? packet : NULL);
|
||||
if (!link) {
|
||||
for (struct ETCP_LINK* candidate = conn->links; candidate; candidate = candidate->next)
|
||||
if (candidate->initialized && candidate->link_status &&
|
||||
@@ -1086,7 +1112,10 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
|
||||
if (!link && (conn->ack_till_pending || conn->ack_request_pending)) {
|
||||
link = etcp_loadbalancer_select_link(conn, 0, NULL); packet = NULL;
|
||||
}
|
||||
if (!link) { conn->tx_state = packet ? ETCP_TX_STATE_LINK_WAIT : ETCP_TX_STATE_DATA_WAIT; return NULL; }
|
||||
if (!link) {
|
||||
conn->tx_state = conn->input_send_q->count || conn->input_queue->count ? ETCP_TX_STATE_LINK_WAIT : ETCP_TX_STATE_DATA_WAIT;
|
||||
return NULL;
|
||||
}
|
||||
conn->tx_state = ETCP_TX_STATE_DATA_WAIT;
|
||||
int max_enc = link->mtu - WIRE_OVERHEAD(link->remote_addr.ss_family);
|
||||
if (max_enc > PACKET_DATA_SIZE) max_enc = PACKET_DATA_SIZE;
|
||||
|
||||
@@ -127,7 +127,7 @@ struct ETCP_CONN {
|
||||
|
||||
// Links (channels) - linked list
|
||||
struct ETCP_LINK* links;
|
||||
struct ETCP_LINK* last_rr_link; // последний линк, выбранный round-robin
|
||||
struct ETCP_LINK* last_rr_link; // последняя принятая DATA-отправка; пробный выбор/ACK курсор не меняет
|
||||
uint8_t last_assigned_link_id; // последний выданный link_id (для циклического перебора)
|
||||
|
||||
struct ETCP_PADDING padding; // диапазон и независимый таймер всех линков этого соединения
|
||||
|
||||
Reference in new issue
Block a user