diff --git a/lib/ble/ble_transport.c b/lib/ble/ble_transport.c index 93f8437fc..d67bf69ff 100644 --- a/lib/ble/ble_transport.c +++ b/lib/ble/ble_transport.c @@ -164,7 +164,8 @@ static int schedule_ack(struct ble_transport* t, struct ble_connection* c) { if (!c->ack_dirty) c->ack_due = c->ack_sent ? c->ack_sent + BLE_ACK_INTERVAL_US : now + BLE_ACK_INTERVAL_US; c->ack_dirty = 1; if (c->ack_timer || (c->pending && now >= c->ack_due)) return 1; - uint64_t delay = c->ack_due > now ? (c->ack_due - now + 99) / 100 : 0; + /* uasync использует усечённые TB: дополнительный TB исключает срабатывание раньше 15ms. */ + uint64_t delay = c->ack_due > now ? (c->ack_due - now + 199) / 100 : 0; c->ack_timer = uasync_set_timeout(t->ua, (int)delay, c, ack_timer, "ble_ack"); if (c->ack_timer) return 1; DEBUG_ERROR(CAT, "BLE ACK timer allocation failed link=%llu rx_last=%u", (unsigned long long)c->id, c->rx_last); diff --git a/lib/ble/tests/test_ble_transport.c b/lib/ble/tests/test_ble_transport.c index fb25e9ac6..fd9030470 100644 --- a/lib/ble/tests/test_ble_transport.c +++ b/lib/ble/tests/test_ble_transport.c @@ -7,11 +7,16 @@ #include #include #define CHECK(v) do { if (!(v)) { fprintf(stderr, "FAIL line=%d: %s\n", __LINE__, #v); exit(1); } } while (0) -struct fake_link { ble_link_id id, other; struct ble_backend* peer; uint8_t data[512]; size_t len; int pending; enum ble_wire_channel channel; }; +struct fake_link { + ble_link_id id, other; struct ble_backend* peer; uint8_t data[512]; size_t len; int pending; + enum ble_wire_channel channel; uint64_t ack_at; uint16_t ack_id; +}; struct ble_backend { struct ble_transport* owner; struct UASYNC* ua; void* task; int index, started; struct fake_link links[16]; }; static struct ble_backend* backends[3]; static unsigned int mtu = 23; static int hold, drop, reject_peer, ready_count, writable_count; +static int hold_ack; +static uint64_t data_wire_bytes[3]; static ble_link_id incoming_ids[3]; static void run(void* arg); static void schedule(struct ble_backend* b) { if (!b->task) b->task = uasync_call_soon(b->ua, b, run); CHECK(b->task); } @@ -44,6 +49,13 @@ void ble_backend_close(struct ble_backend* b, ble_link_id id) { int ble_backend_send(struct ble_backend* b, ble_link_id id, enum ble_wire_channel channel, const uint8_t* data, size_t len) { struct fake_link* f = slot(b, id); CHECK(f && !f->pending && len <= sizeof(f->data)); CHECK(len <= mtu - 3); + if (channel == BLE_WIRE_DATA) data_wire_bytes[b->index] += len; + if (channel == BLE_WIRE_ACK) { + CHECK(len == 2); + uint64_t now = get_time_us(); + CHECK(!f->ack_at || now - f->ack_at >= BLE_ACK_INTERVAL_US); + f->ack_at = now; f->ack_id = (uint16_t)((data[0] << 8) | data[1]); + } f->channel = channel; memcpy(f->data, data, len); f->len = len; f->pending = 1; schedule(b); return 0; } static void run(void* arg) { @@ -52,7 +64,7 @@ static void run(void* arg) { for (int i = 0; i < 16; i++) { struct fake_link* f = &b->links[i]; if (!f->id) continue; if (!f->len) { f->len = 1; ble_backend_opened(b->owner, f->id, mtu); } - if (!f->pending || hold) continue; + if (!f->pending || hold || (hold_ack && f->channel == BLE_WIRE_ACK)) continue; ble_link_id id = f->id, other = f->other; struct ble_backend* peer = f->peer; uint8_t data[512]; size_t len = f->len; memcpy(data, f->data, len); f->pending = 0; @@ -138,9 +150,65 @@ static void exercise(unsigned int test_mtu) { for (int i = 0; i < 3; i++) ble_destroy(t[i]); drain(ua); uasync_destroy(ua, 0); } +/* Локальный completion мгновенный, удалённые ACK задержаны: предел относится именно к очереди ниже backend. */ +static void exercise_flow(void) { + mtu = 517; hold = drop = reject_peer = hold_ack = 0; + struct UASYNC* ua = uasync_create(); CHECK(ua); + struct ble_config cfg = {.node_id = 1}; cfg.callbacks.writable = writable; + struct ble_transport* a = ble_create(ua, &cfg); cfg.node_id = 2; + struct ble_transport* b = ble_create(ua, &cfg); CHECK(a && b && !ble_start(a) && !ble_start(b)); drain(ua); + struct ble_endpoint ep = {"1"}; ble_link_id id = ble_connect(a, &ep, 2); CHECK(id); drain(ua); + ble_link_id other = incoming_ids[1]; + uint64_t base[2] = {data_wire_bytes[0], data_wire_bytes[1]}; + unsigned received[2] = {0}; hold_ack = 1; + for (unsigned i = 0; i < BLE_QUEUE_MAX; i++) { + CHECK(!ble_tx_put(a, id, frame(1500, i))); CHECK(!ble_tx_put(b, other, frame(1500, i))); + } + for (unsigned i = 0; i < 10000; i++) uasync_poll(ua, 0); + for (unsigned i = 0; i < 2; i++) { + CHECK(data_wire_bytes[i] - base[i] <= BLE_TX_WINDOW_BYTES); + CHECK(data_wire_bytes[i] - base[i] >= BLE_TX_WINDOW_BYTES - BLE_VALUE_MAX); + } + CHECK(!ble_tx_ready(a, id) && !ble_tx_ready(b, other)); + uint64_t paused = data_wire_bytes[0]; + struct fake_link* receiver = slot(backends[1], other); CHECK(receiver); + uint8_t duplicate[2] = {receiver->ack_id >> 8, receiver->ack_id}; + /* Только уже доставленный ACK: до запуска отложенного ACK последняя запись ещё от HELLO. */ + ble_backend_received(a, id, BLE_WIRE_ACK, duplicate, 2); + for (unsigned i = 0; i < 1000; i++) uasync_poll(ua, 0); + CHECK(data_wire_bytes[0] == paused && !ble_tx_ready(a, id)); + hold_ack = 0; schedule(backends[0]); schedule(backends[1]); + uint64_t deadline = get_time_us() + 5000000; + while ((received[0] < BLE_QUEUE_MAX || received[1] < BLE_QUEUE_MAX) && get_time_us() < deadline) { + uasync_poll(ua, 10); + for (unsigned i = 0; i < 2; i++) { + struct ll_entry* e; + while ((e = ble_rx_get(i ? b : a, i ? other : id))) { + CHECK(e->len == 1500 && e->dgram[0] == received[i] && e->dgram[1499] == received[i]); + received[i]++; release(e); + } + } + } + CHECK(received[0] == BLE_QUEUE_MAX && received[1] == BLE_QUEUE_MAX); drain(ua); + /* Больше полного оборота wire ID; новые данные действительно приходят, старые ACK не открывают окно. */ + unsigned sent = 0, delivered = 0; deadline = get_time_us() + 10000000; + while (delivered < 70000 && get_time_us() < deadline) { + while (sent < 70000 && ble_tx_ready(a, id)) { CHECK(!ble_tx_put(a, id, frame(1, sent))); sent++; } + uasync_poll(ua, 0); + struct ll_entry* e; + while ((e = ble_rx_get(b, other))) { CHECK(e->len == 1 && e->dgram[0] == (uint8_t)delivered); delivered++; release(e); } + } + CHECK(sent == 70000 && delivered == sent); drain(ua); + struct fake_link* sender = slot(backends[0], id); CHECK(sender); + uint8_t future[2] = {(uint8_t)((receiver->ack_id + 1) >> 8), (uint8_t)(receiver->ack_id + 1)}; + ble_backend_received(a, id, BLE_WIRE_ACK, future, 2); CHECK(!ble_tx_ready(a, id)); + ble_destroy(a); ble_destroy(b); uasync_destroy(ua, 0); + puts("BLE flow PASS (12500-byte duplex bound, ACK >=15ms, duplicate ACK, resume, 70000 IDs, invalid ACK, teardown)"); +} int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); size_t before = u_get_allocated_count(); exercise(23); exercise(247); exercise(517); + exercise_flow(); CHECK(u_get_allocated_count() == before); puts("BLE transport tests PASS (MTU 23/247/517, duplex, loss, backpressure, peers, teardown)"); return 0; }