#include #include #include "../lib/platform_compat.h" #include "../lib/debug_config.h" #include "../lib/mem.h" #include "../lib/u_async.h" #ifdef UTUN_HAVE_STANDBY #include "standby.h" #endif #include "etcp_api.h" #include "etcp_connections.h" #include "etcp.h" #include "topo_node.h" #include "topo_group.h" #include "broadcast.h" struct broadcast_msg { uint64_t local_time_tb; // offset 0 uint8_t uuid[BROADCAST_UUID_SIZE]; // offset 8 — queue index key uint16_t data_len; // offset 24 uint8_t data[]; // offset 26 } __attribute__((packed)); #define BMSG_UUID_OFFSET 8 struct broadcast_ctx { struct TOPO_GROUP* group; struct ll_queue* messages; struct broadcast_cbk_entry* cbks; void* cleanup_timer; void* cleanup_wait; /* standby_wait handle (Android) */ uint32_t send_seq; }; static void broadcast_cleanup_timer_cb(void* arg); int broadcast_init(struct TOPO_GROUP* group) { if (!group) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: null group"); return -1; } if (group->broadcast) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: already initialized grp=%016llx", (unsigned long long)group->group_id); return -1; } struct broadcast_ctx* ctx = u_calloc(1, sizeof(struct broadcast_ctx)); if (!ctx) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: alloc failed"); return -1; } ctx->group = group; ctx->send_seq = 0; ctx->messages = queue_new(group->instance->ua, 64, BMSG_UUID_OFFSET, BROADCAST_UUID_SIZE, "broadcast_msg"); if (!ctx->messages) { u_free(ctx); DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: queue_new failed"); return -1; } group->broadcast = ctx; ctx->cleanup_timer = uasync_set_timeout(group->instance->ua, BROADCAST_CLEANUP_INTERVAL_TB, ctx, broadcast_cleanup_timer_cb, "broadcast_cleanup_timer"); if (!ctx->cleanup_timer) DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_init: cleanup_timer failed grp=%016llx", (unsigned long long)group->group_id); DEBUG_INFO(DEBUG_CATEGORY_BGP, "broadcast_init ok grp=%016llx", (unsigned long long)group->group_id); return 0; } void broadcast_destroy(struct TOPO_GROUP* group) { if (!group || !group->broadcast) return; struct broadcast_ctx* ctx = group->broadcast; DEBUG_INFO(DEBUG_CATEGORY_BGP, "broadcast_destroy grp=%016llx", (unsigned long long)group->group_id); if (ctx->cleanup_timer) { uasync_cancel_timeout(group->instance->ua, ctx->cleanup_timer); ctx->cleanup_timer = NULL; } #ifdef UTUN_HAVE_STANDBY if (ctx->cleanup_wait) { standby_wait_cancel(ctx->cleanup_wait); ctx->cleanup_wait = NULL; } #endif while (ctx->cbks) { struct broadcast_cbk_entry* c = ctx->cbks; ctx->cbks = c->next; u_free(c); } if (ctx->messages) { struct ll_entry* e; while ((e = queue_data_get(ctx->messages)) != NULL) queue_entry_free(e); queue_free(ctx->messages); ctx->messages = NULL; } u_free(ctx); group->broadcast = NULL; } int broadcast_send(struct TOPO_GROUP* group, const uint8_t* data, uint16_t data_len) { if (!group || !group->broadcast || !data || data_len > BROADCAST_MAX_DATA) return -1; struct broadcast_ctx* ctx = group->broadcast; uint8_t uuid[BROADCAST_UUID_SIZE]; { uint64_t tb = get_time_tb(); uint32_t seq = ctx->send_seq++; uint64_t ptr = (uint64_t)(uintptr_t)ctx; memset(uuid, 0, BROADCAST_UUID_SIZE); memcpy(uuid, &tb, 8); memcpy(uuid + 8, &seq, 4); memcpy(uuid + 12, &ptr, 4); } { size_t sz = 26 + (size_t)data_len; // 8(time) + 16(uuid) + 2(data_len) + data struct ll_entry* e = queue_entry_new(sz); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_send: alloc failed"); return -1; } struct broadcast_msg* msg = (struct broadcast_msg*)e->data; msg->local_time_tb = get_time_tb(); memcpy(msg->uuid, uuid, BROADCAST_UUID_SIZE); msg->data_len = data_len; memcpy(msg->data, data, data_len); queue_data_put_with_index(ctx->messages, e); } if (group->senders_list) { struct ll_entry* se = group->senders_list->head; while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (item && item->conn) { uint16_t total = 1 + 8 + BROADCAST_UUID_SIZE + 2 + data_len; uint8_t* pkt = u_malloc(total); if (!pkt) { se = se->next; continue; } pkt[0] = ETCP_ID_BROADCAST; memcpy(pkt + 1, &group->group_id, 8); memcpy(pkt + 9, uuid, BROADCAST_UUID_SIZE); pkt[25] = (uint8_t)(data_len & 0xFF); pkt[26] = (uint8_t)((data_len >> 8) & 0xFF); memcpy(pkt + 27, data, data_len); struct ll_entry* ee = queue_entry_new(0); if (!ee) { u_free(pkt); se = se->next; continue; } ee->dgram = pkt; ee->len = total; if (etcp_send(item->conn, ee) != 0) { u_free(pkt); queue_entry_free(ee); } } se = se->next; } } return 0; } void broadcast_add_cbk(struct TOPO_GROUP* group, broadcast_recv_fn fn, void* arg) { if (!group || !group->broadcast || !fn) return; struct broadcast_ctx* ctx = group->broadcast; struct broadcast_cbk_entry* e = u_malloc(sizeof(*e)); if (!e) return; e->fn = fn; e->arg = arg; e->next = ctx->cbks; ctx->cbks = e; } void broadcast_remove_cbk(struct TOPO_GROUP* group, broadcast_recv_fn fn, void* arg) { if (!group || !group->broadcast || !fn) return; struct broadcast_ctx* ctx = group->broadcast; struct broadcast_cbk_entry** p = &ctx->cbks; while (*p) { if ((*p)->fn == fn && (*p)->arg == arg) { struct broadcast_cbk_entry* r = *p; *p = r->next; u_free(r); return; } p = &(*p)->next; } } static void broadcast_fire_cbks(struct broadcast_ctx* ctx, const uint8_t* uuid, const uint8_t* data, uint16_t data_len) { struct broadcast_cbk_entry* c = ctx->cbks; while (c) { c->fn(uuid, data, data_len, c->arg); c = c->next; } } static void broadcast_cleanup_timer_cb(void* arg) { struct broadcast_ctx* ctx = (struct broadcast_ctx*)arg; if (!ctx || !ctx->group) return; #ifdef UTUN_HAVE_STANDBY if (standby_get_sleep_tb() > 0) { ctx->cleanup_wait = standby_wait(ctx, broadcast_cleanup_timer_cb); return; } ctx->cleanup_wait = NULL; #endif uint64_t now_tb = get_time_tb(); int removed = 0; while (1) { struct ll_entry* e = queue_data_get(ctx->messages); if (!e) break; struct broadcast_msg* msg = (struct broadcast_msg*)e->data; if (now_tb - msg->local_time_tb > BROADCAST_TTL_TB) { queue_entry_free(e); removed++; } else { queue_data_put_first(ctx->messages, e); break; } } if (removed > 0) DEBUG_DEBUG(DEBUG_CATEGORY_BGP, "broadcast_cleanup: removed %d stale msgs grp=%016llx", removed, (unsigned long long)ctx->group->group_id); ctx->cleanup_timer = uasync_set_timeout(ctx->group->instance->ua, BROADCAST_CLEANUP_INTERVAL_TB, ctx, broadcast_cleanup_timer_cb, "broadcast_cleanup_timer"); } void broadcast_recv(struct TOPO_GROUP* group, struct ETCP_CONN* from_conn, const uint8_t* payload, size_t payload_len) { if (!group || !group->broadcast || !payload || payload_len < BROADCAST_UUID_SIZE + 2) return; struct broadcast_ctx* ctx = group->broadcast; const uint8_t* uuid = payload; uint16_t data_len = (uint16_t)payload[BROADCAST_UUID_SIZE] | ((uint16_t)payload[BROADCAST_UUID_SIZE + 1] << 8); const uint8_t* data = payload + BROADCAST_UUID_SIZE + 2; if (data_len > BROADCAST_MAX_DATA || BROADCAST_UUID_SIZE + 2 + (size_t)data_len > payload_len) { DEBUG_WARN(DEBUG_CATEGORY_BGP, "broadcast_recv: invalid size data_len=%u payload_len=%zu", data_len, payload_len); return; } if (queue_find_data_by_index(ctx->messages, uuid) != NULL) return; { size_t sz = 26 + (size_t)data_len; struct ll_entry* e = queue_entry_new(sz); if (!e) { DEBUG_ERROR(DEBUG_CATEGORY_BGP, "broadcast_recv: alloc failed"); return; } struct broadcast_msg* msg = (struct broadcast_msg*)e->data; msg->local_time_tb = get_time_tb(); memcpy(msg->uuid, uuid, BROADCAST_UUID_SIZE); msg->data_len = data_len; memcpy(msg->data, data, data_len); queue_data_put_with_index(ctx->messages, e); } if (group->senders_list) { struct ll_entry* se = group->senders_list->head; while (se) { struct TOPO_GROUP_CONN_ITEM* item = (struct TOPO_GROUP_CONN_ITEM*)se->data; if (item && item->conn && item->conn != from_conn) { uint16_t total = 1 + 8 + BROADCAST_UUID_SIZE + 2 + data_len; uint8_t* pkt = u_malloc(total); if (!pkt) { se = se->next; continue; } pkt[0] = ETCP_ID_BROADCAST; memcpy(pkt + 1, &group->group_id, 8); memcpy(pkt + 9, uuid, BROADCAST_UUID_SIZE); pkt[25] = (uint8_t)(data_len & 0xFF); pkt[26] = (uint8_t)((data_len >> 8) & 0xFF); memcpy(pkt + 27, data, data_len); struct ll_entry* ee = queue_entry_new(0); if (!ee) { u_free(pkt); se = se->next; continue; } ee->dgram = pkt; ee->len = total; if (etcp_send(item->conn, ee) != 0) { u_free(pkt); queue_entry_free(ee); } } se = se->next; } } broadcast_fire_cbks(ctx, uuid, data, data_len); } static void broadcast_recv_dispatcher(struct ETCP_CONN* from_conn, struct ll_entry* entry) { if (!from_conn || !entry || entry->len < 1 + 8) { if (entry) { queue_dgram_free(entry); queue_entry_free(entry); } return; } struct UTUN_INSTANCE* instance = from_conn->instance; if (!instance) { queue_dgram_free(entry); queue_entry_free(entry); return; } uint8_t* dgram = entry->dgram; size_t dgram_len = entry->len; if (dgram[0] != ETCP_ID_BROADCAST) { queue_dgram_free(entry); queue_entry_free(entry); return; } uint64_t group_id; memcpy(&group_id, dgram + 1, 8); size_t payload_len = dgram_len - 1 - 8; const uint8_t* payload = dgram + 1 + 8; if (instance->topo_groups) { struct TOPO_GROUP* group = topo_groups_find(instance->topo_groups, group_id); if (group) broadcast_recv(group, from_conn, payload, payload_len); } queue_dgram_free(entry); queue_entry_free(entry); } int broadcast_init_instance(struct UTUN_INSTANCE* instance) { if (!instance) return -1; DEBUG_INFO(DEBUG_CATEGORY_BGP, "broadcast_init_instance: binding ETCP_ID_BROADCAST"); return etcp_bind(instance, ETCP_ID_BROADCAST, broadcast_recv_dispatcher); } void broadcast_destroy_instance(struct UTUN_INSTANCE* instance) { if (!instance) return; DEBUG_INFO(DEBUG_CATEGORY_BGP, "broadcast_destroy_instance: unbinding ETCP_ID_BROADCAST"); etcp_unbind(instance, ETCP_ID_BROADCAST); }