/* -- broadd - tcp/udp broadcast server ------ env vars: ---------- MAX_CLIENTS -> maximum number of hypothetical connections ---------- BUFS -> buffer size, values >MTU are not recommended on udp ---------- RATE_LIMIT_S -> rate limit expiry window ---------- CLIENT_TIMEOUT -> how long to wait before evicting a udp client ---------- WILDCARD -> whether or not to enable wildcard. set to 0 or 1. ---------- RELAY_TXT -> read a pre-determined list of relays and connect to them at startup. -------------- format is one relay per line, ` ` ------ difference: ---------- udp mode does not send m_ok or p_ok ---------- send `N\n` to initiate connection in udp mode ---------- you get kicked after TIMEOUT seconds with no packets in udp mode ---------- send an `N\n` packet before TIMEOUT seconds are elapsed, to prevent getting your id reaped ------ protocol: ---------- all server replies except M for incoming messages are suffixed with "\n" ---------- ---------- connect -------------- if in udp mode, server sends `T\n` to tell you when timeout will occur. -------------- server sends `IT` or `IU`, depending on whether it's tcp or udp -------------- this is your id -------------- if server is full, it instead sends "full" ---------- send -------------- `P<8-byte password>` ------------------ server will only broadcast to you messages starting with that password ------------------ the password will be stripped before it is sent to you -------------- `M` ------------------ broadcast message to all clients ------------------ only clients with `password==first_8_bytes` receive your msg ------------------ message shorter than 8 returns "Err" ------------------ default password is "00000000" ASCII -------------- `N` ------------------ does nothing. use this for initiating your UDP connection. -------------- `W` ------------------ returns `W0` or `W1`, depending on whether the server has wildcard enabled or not. ---------- receive ---------------- when you send P for password -------------------- server sends back "p_ok" ---------------- when you send M or S for message -------------------- server sends back "m_ok" ---------------- when server is shutting down -------------------- server sends "bye" -------------------- close the socket when you get this msg ---------------- when an msg is broadcasted -------------------- M:{T} ---------------- when admin sends you a msg -------------------- server sends `A` -------------------- this message is sent to all, regardless of password ------ admin panel: -------- type the commands into server stdin to use -------- spaces are omitted before/after args -------- `E` -> closes the server -------- `C ` -> connects to another server, relaying all its messages to the current server's clients -------- `B ` -> broadcasts message to all currently connected clients */ // ======== change these ======== #if 1 // max buffer size #define UDP_BUFS 1024 #define TCP_BUFS 16384 // rate limit: BUFS bytes / RLM seconds #define RATE_LIMIT_S 1 // max clients count #define MAX_CLIENTS 8192 // client timeout for udp mode in seconds #define CLIENT_TIMEOUT 30 // max age for timestamped message #define DROP_OLD 5 // set to 0 to enable ratelimit #define DISABLE_RATELIMIT 1 #endif // ============================== #if UDP == 1 #define BUFS UDP_BUFS #else #define BUFS TCP_BUFS #endif #include #include #include #include #include #include #include #include #include #include #include #include #include #include #define p_stringify(...) #__VA_ARGS__ #define stringify(...) p_stringify(__VA_ARGS__) #if DEBUG == 1 #define log(...) printf(__VA_ARGS__) #define verb(...) __VA_ARGS__ #else #define log(...) #define verb(...) #endif #if UDP == 1 #define onlyudp(...) __VA_ARGS__ #define onlytcp(...) #else #define onlyudp(...) #define onlytcp(...) __VA_ARGS__ #endif // glob consts using env vars int env_max_clients = MAX_CLIENTS; int env_bufs = BUFS; int env_rate_limit_s = RATE_LIMIT_S; int env_wildcard = 1; onlyudp(int env_client_timeout = CLIENT_TIMEOUT); pthread_mutex_t relay_mutex = PTHREAD_MUTEX_INITIALIZER; #define DEDUP_SIZE 65536 #define DEDUP_MASK (DEDUP_SIZE - 1) #define DEDUP_WINDOW 5 typedef struct { uint32_t hash; time_t seen_at; } dedup_entry_t; dedup_entry_t dedup_table[ DEDUP_SIZE ]; uint32_t hash_data(char* data, int len) { uint32_t hash = 2166136261u; for (int i = 0; i < len; i++) { hash ^= (uint8_t) data[ i ]; hash *= 16777619; } return hash; } int is_duplicate(char* payload, int len) { if (len <= 0) return 0; uint32_t h = hash_data(payload, len); uint32_t idx = h & DEDUP_MASK; time_t now = time(NULL); if (dedup_table[ idx ].hash == h) { if (now - dedup_table[ idx ].seen_at < DEDUP_WINDOW) { return 1; } } dedup_table[ idx ].hash = h; dedup_table[ idx ].seen_at = now; return 0; } #define MAX_RELAYS 64 struct relay { int fd; int is_udp; time_t last_hb; // udp: last heartbeat sent int remote_timeout; struct sockaddr_storage addr; // udp socklen_t addrlen; // udp }; struct relay relays[ MAX_RELAYS ]; int relay_count = 0; // stdin nonblocking #define NB_EOF -1 #define NB_AGAIN -2 static char nb_buffer[ BUFSIZ ]; static size_t nb_buf_pos = 0; static size_t nb_buf_fill = 0; #define fold inline __attribute__((always_inline)) static fold void set_stdin_nonblocking(void) { static int done = 0; if (!done) { int flags = fcntl(STDIN_FILENO, F_GETFL); if (flags != -1) { fcntl(STDIN_FILENO, F_SETFL, flags | O_NONBLOCK); } done = 1; } } fold int nb_getstring(char str[], size_t max_size) { if (max_size <= 1) { if (max_size == 1) str[ 0 ] = '\0'; return 0; } size_t pos = 0; // drain buffer size_t remaining = max_size - 1; size_t buf_avail = nb_buf_fill - nb_buf_pos; size_t copy_len = buf_avail < remaining ? buf_avail : remaining; if (copy_len > 0) { memcpy(str + pos, nb_buffer + nb_buf_pos, copy_len); nb_buf_pos += copy_len; pos += copy_len; } if (pos < max_size - 1 && nb_buf_pos >= nb_buf_fill) { ssize_t n; do { n = read(STDIN_FILENO, nb_buffer, sizeof(nb_buffer)); } while (n == -1 && errno == EINTR); if (n <= 0) { if (n == 0) { if (pos == 0) return NB_EOF; } else { // n == -1 if (errno == EAGAIN || errno == EWOULDBLOCK) { if (pos == 0) return NB_AGAIN; } else { if (pos == 0) return NB_EOF; } } } else { nb_buf_pos = 0; nb_buf_fill = (size_t) n; remaining = max_size - 1 - pos; buf_avail = nb_buf_fill - nb_buf_pos; copy_len = buf_avail < remaining ? buf_avail : remaining; if (copy_len > 0) { memcpy(str + pos, nb_buffer + nb_buf_pos, copy_len); nb_buf_pos += copy_len; pos += copy_len; } } } str[ pos ] = '\0'; return (int) pos; } #define likely(x) __builtin_expect(!!(x), 1) #define unlikely(x) __builtin_expect(!!(x), 0) uint64_t default_pass; // default password for when a client joins. uint64_t wildcard_pass; // a client with this password will receive all messages with any password. onlytcp(struct client { int fd; // descriptor uint64_t id; // identifier uint64_t pass; // password time_t last_rl; // ratelimit timestamp int rl_bytes; // bytes used }); onlyudp(struct client { struct sockaddr_storage addr; // client address socklen_t addrlen; // address len uint64_t id; // identifier uint64_t pass; // password time_t last_rl; // ratelimit timestamp int rl_bytes; // bytes used time_t last_seen; // for kicking }); struct client* clients = NULL; int active = 0; int cap = 0; uint64_t nid = 2; // next id to assign // accept new client onlytcp(void add_client(int fd) { if (active >= env_max_clients) { // refuse: too many clients write(fd, "full\n", 5); close(fd); return; } if (active >= cap) { cap = cap ? cap * 2 : 64; if (cap > env_max_clients) cap = env_max_clients; clients = realloc(clients, cap * sizeof(struct client)); } onlytcp(clients[ active ].fd = fd); clients[ active ].id = nid++; clients[ active ].pass = default_pass; clients[ active ].last_rl = time(NULL); clients[ active ].rl_bytes = 0; char buf[ 64 ]; int len = sprintf(buf, "IT%" PRIu64 "\n", clients[ active ].id); write(fd, buf, len); active++; log("clients:%d\n", active); }); // disconnect client at index i void del_client(int i) { onlytcp(close(clients[ i ].fd)); if (i < active - 1) clients[ i ] = clients[ active - 1 ]; active--; log("clients:%d\n", active); } int sock; // broadcast to all clients whose password matches void broadcast_msg(uint64_t sender_id, char* payload, int msg_len) { if (is_duplicate(payload, msg_len)) return; uint64_t p; // separate if (msg_len < 8) return; memcpy(&p, payload, 8); char* raw_body = payload + 8; int raw_body_len = msg_len - 8; // sign with timestamp char temp_body_buf[ BUFS + 32 ]; char* final_body; int final_body_len; time_t now = time(NULL); int is_timestamped = 0; // dont dupe if (raw_body_len >= 13 && strncmp(raw_body, "{T", 2) == 0 && raw_body[ 12 ] == '}') { // get time char t_str[ 11 ]; memcpy(t_str, raw_body + 2, 10); t_str[ 10 ] = '\0'; long msg_time = atol(t_str); // check if (now - msg_time > DROP_OLD) { // old return; } // valid is_timestamped = 1; final_body = raw_body; final_body_len = raw_body_len; } if (!is_timestamped) { // add stamp if (raw_body_len + 13 > BUFS) { // too big, drop return; } // write sprintf(temp_body_buf, "{T%010ld}", now); // append body memcpy(temp_body_buf + 13, raw_body, raw_body_len); final_body = temp_body_buf; final_body_len = raw_body_len + 13; } int sent = 0; // bake normal char normal_buf[ 64 + BUFS ]; int normal_len = sprintf(normal_buf, "M%" PRIu64 ":", sender_id); memcpy(normal_buf + normal_len, final_body, final_body_len); normal_len += final_body_len; // bake wildcard char wild_buf[ 1 + 8 + BUFS ]; wild_buf[ 0 ] = 'M'; memcpy(wild_buf + 1, &p, 8); memcpy(wild_buf + 9, final_body, final_body_len); int wild_len = 1 + 8 + final_body_len; // send for (int j = 0; j < active; j++) { if (clients[ j ].id == sender_id) continue; int is_wildcard = env_wildcard && clients[ j ].pass == wildcard_pass; if (p != clients[ j ].pass && !is_wildcard) continue; char* send_buf = is_wildcard ? wild_buf : normal_buf; int send_len = is_wildcard ? wild_len : normal_len; onlytcp(write(clients[ j ].fd, send_buf, send_len)); onlyudp(sendto(sock, send_buf, send_len, 0, (struct sockaddr*) &clients[ j ].addr, clients[ j ].addrlen)); sent++; } if (sent > 0) log("sentto:%d\n", sent); } void* connect_relay_thread(void* arg) { char* ip_port = arg; if (relay_count >= MAX_RELAYS) { printf("relay(full)\n"); free(ip_port); return NULL; } char ip[ 64 ], port[ 16 ]; if (sscanf(ip_port, "%63s %15s", ip, port) != 2) { free(ip_port); return NULL; } free(ip_port); // try tcp struct addrinfo hint = {}, *result; hint.ai_family = AF_UNSPEC; hint.ai_socktype = SOCK_STREAM; if (getaddrinfo(ip, port, &hint, &result) != 0) { perror("relay: getaddrinfo failed\n"); return NULL; } int is_udp = 0; int fd = socket(result->ai_family, result->ai_socktype, result->ai_protocol); char gbuf[ 64 ]; if (fd < 0) { perror("relay: socket"); freeaddrinfo(result); return NULL; } // set nonblocking int flags = fcntl(fd, F_GETFL); if (flags == -1 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) { close(fd); freeaddrinfo(result); printf("relay: fcntl failed\n"); return NULL; } int conn_err = connect(fd, result->ai_addr, result->ai_addrlen); if (conn_err == 0) goto tcp_connected; if (errno != EINPROGRESS) { close(fd); freeaddrinfo(result); printf("relay: connect fail\n"); return NULL; } struct pollfd pfd = { .fd = fd, .events = POLLOUT }; int pres = poll(&pfd, 1, 10000); // 10s if (pres <= 0) { // timeout or error close(fd); freeaddrinfo(result); if (pres == 0) printf("relay: tcp timeout\n"); else { perror("relay: poll failed"); } return NULL; } // check success int so_err = 0; socklen_t so_len = sizeof(so_err); if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &so_err, &so_len) == -1 || so_err != 0) { printf("relay: trying udp"); close(fd); freeaddrinfo(result); goto try_udp; } tcp_connected: fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) & ~O_NONBLOCK); // read greeting int n = read(fd, gbuf, sizeof(gbuf) - 1); if (n > 0 && gbuf[ 0 ] == 'I' && gbuf[ 1 ] == 'T') { // it's tcp char pbuf[ 9 ] = "P"; memcpy(pbuf + 1, &wildcard_pass, 8); write(fd, pbuf, 9); read(fd, gbuf, sizeof(gbuf)); // p_ok // set nonblock fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK); } else { close(fd); fd = -1; } freeaddrinfo(result); // udp fallback if (fd < 0) { try_udp: hint.ai_socktype = SOCK_DGRAM; if (getaddrinfo(ip, port, &hint, &result) != 0) { printf("relay: getaddrinfo failed\n"); return NULL; } fd = socket(result->ai_family, result->ai_socktype, result->ai_protocol); if (fd < 0) { freeaddrinfo(result); printf("relay: socket failed\n"); return NULL; } // prevent hanging struct timeval tv; tv.tv_sec = 2; tv.tv_usec = 0; setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*) &tv, sizeof tv); sendto(fd, "N\n", 2, 0, result->ai_addr, result->ai_addrlen); if (connect(fd, result->ai_addr, result->ai_addrlen) < 0) { close(fd); freeaddrinfo(result); printf("relay: udp connect failed\n"); return NULL; } struct sockaddr_storage src; socklen_t srclen = sizeof(src); int rn = read(fd, gbuf, sizeof(gbuf) - 1); if (rn <= 0) { printf("relay: udp timeout\n"); close(fd); freeaddrinfo(result); return NULL; } gbuf[ rn ] = '\0'; int remote_timeout = (gbuf[ 0 ] == 'T') ? atoi(gbuf + 1) : 30; rn = read(fd, gbuf, sizeof(gbuf) - 1); if (rn <= 0 || gbuf[ 0 ] != 'I' || gbuf[ 1 ] != 'U') { close(fd); freeaddrinfo(result); printf("relay: invalid handshake\n"); return NULL; } fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK); char pbuf[ 9 ] = "P"; memcpy(pbuf + 1, &wildcard_pass, 8); sendto(fd, pbuf, 9, 0, result->ai_addr, result->ai_addrlen); is_udp = 1; pthread_mutex_lock(&relay_mutex); if (relay_count >= MAX_RELAYS) { printf("relay(full) %s:%s\n", ip, port); close(fd); } else { struct relay* r = &relays[ relay_count++ ]; r->fd = fd; r->is_udp = 1; r->last_hb = time(NULL); r->remote_timeout = remote_timeout; memcpy(&r->addr, result->ai_addr, result->ai_addrlen); r->addrlen = result->ai_addrlen; printf("relay-connect %s:%s (%s)\n", ip, port, is_udp ? "udp" : "tcp"); } pthread_mutex_unlock(&relay_mutex); freeaddrinfo(result); tv.tv_sec = 0; setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*) &tv, sizeof tv); } else { pthread_mutex_lock(&relay_mutex); if (relay_count >= MAX_RELAYS) { printf("relay(full) %s:%s\n", ip, port); close(fd); } else { struct relay* r = &relays[ relay_count++ ]; r->fd = fd; r->is_udp = 0; r->last_hb = time(NULL); r->remote_timeout = 0; } pthread_mutex_unlock(&relay_mutex); } printf("successful connect\n"); return NULL; } void connect_relay(const char* ip, const char* port) { if (relay_count >= MAX_RELAYS) { printf("relay(full)\n"); return; } char* ip_port = malloc(128); if (!ip_port) return; snprintf(ip_port, 128, "%s %s", ip, port); pthread_t th; if (pthread_create(&th, NULL, connect_relay_thread, ip_port) != 0) { perror("thread spawn failed\n"); free(ip_port); return; } pthread_detach(th); } char* admin_msg; int main(int argc, char** argv) { if (argc < 3) { printf("more args.\n%s \n", argv[ 0 ]); exit(1); } onlytcp(printf("hi from tcp;" verb(" verbose mode") "\n")); onlyudp(printf("hi from udp;" verb(" verbose mode") "\n")); // parse args const char* e; if ((e = getenv("MAX_CLIENTS"))) { int v = atoi(e); if (v > 0 && v <= 65536) env_max_clients = v; else printf("MAX_CLIENTS: ignored (must be 1...65536)\n"); } if ((e = getenv("BUFS"))) { int v = atoi(e); if (v > 0 && v <= 65536) env_bufs = v; else printf("BUFS: ignored (must be 1...65536)\n"); } if ((e = getenv("RATE_LIMIT_S"))) { int v = atoi(e); if (v > 0) env_rate_limit_s = v; else printf("RATE_LIMIT_S: ignored (must be >0)\n"); } onlyudp(if ((e = getenv("CLIENT_TIMEOUT"))) { int v = atoi(e); if (v > 0) env_client_timeout = v; else printf("CLIENT_TIMEOUT: ignored (must be >0)\n"); }); if ((e = getenv("WILDCARD"))) { int v = atoi(e); env_wildcard = (v != 0); } // alloc admin msg admin_msg = malloc(env_bufs); admin_msg[ 0 ] = 'A'; set_stdin_nonblocking(); // fill passwords memcpy(&default_pass, "00000000", 8); memcpy(&wildcard_pass, "00000001", 8); struct addrinfo hint = {}, *result; hint.ai_family = AF_UNSPEC; onlytcp(hint.ai_socktype = SOCK_STREAM); onlyudp(hint.ai_socktype = SOCK_DGRAM); hint.ai_flags = AI_PASSIVE; if (getaddrinfo(argv[ 1 ], argv[ 2 ], &hint, &result) != 0) { perror("getaddrinfo failed"); exit(1); } sock = socket(result->ai_family, result->ai_socktype, result->ai_protocol); int option = 1; setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(option)); if (bind(sock, result->ai_addr, result->ai_addrlen) < 0) { perror("bind failed"); exit(1); } freeaddrinfo(result); onlytcp(listen(sock, 32)); fcntl(sock, F_SETFL, fcntl(sock, F_GETFL, 0) | O_NONBLOCK); // fds[0] = server socket, fds[1] = stdin, fds[2..] = clients struct pollfd fds[ env_max_clients + 2 + MAX_RELAYS ]; fds[ 1 ].fd = STDIN_FILENO; fds[ 1 ].events = POLLIN; // parse relays.txt file const char* rpath = getenv("RELAY_TXT"); if (rpath) { FILE* f = fopen(rpath, "r"); if (f) { char line[ 256 ]; while (fgets(line, sizeof(line), f)) { char ip[ 64 ], port[ 16 ]; if (sscanf(line, "%63s %15s", ip, port) != 2) continue; connect_relay(ip, port); } fclose(f); } else { perror("unable to open RELAY_TXT"); } } while (1) { int run_shutdown = 0; int got_msg = 0; int resval = 0; fds[ 0 ].fd = sock; fds[ 0 ].events = POLLIN; int n = 2; onlytcp(for (int i = 0; i < active; i++) { fds[ n ].fd = clients[ i ].fd; fds[ n ].events = POLLIN; n++; }); pthread_mutex_lock(&relay_mutex); for (int i = 0; i < relay_count; i++) { fds[ n ].fd = relays[ i ].fd; fds[ n ].events = POLLIN; n++; } int relay_base = n - relay_count; int poll_timeout = onlyudp(env_client_timeout * 1000) onlytcp(-1); // find min poll timeout for (int i = 0; i < relay_count; i++) { if (relays[ i ].is_udp && relays[ i ].remote_timeout > 0) { int t = relays[ i ].remote_timeout * 500; if (poll_timeout < 0 || t < poll_timeout) poll_timeout = t; } } pthread_mutex_unlock(&relay_mutex); poll(fds, n, poll_timeout); time_t now = time(NULL); // sweep udp connections onlyudp(for (int i = active - 1; i >= 0; i--) { if (now - clients[ i ].last_seen > env_client_timeout) { log("evict:%" PRIu64 "\n", clients[ i ].id); del_client(i); } }); // check admin stdin if (fds[ 1 ].revents & POLLIN) { char admin_buf[ BUFS ]; resval = nb_getstring(admin_buf, sizeof(admin_buf)); if (resval == NB_EOF || (resval > 0 && admin_buf[ 0 ] == 'E')) { printf("dying\n"); run_shutdown = 1; } else if (resval > 0) { if (admin_buf[ 0 ] == 'B') { // skip 'B' and any spaces char* msg = admin_buf + 1; while (*msg == ' ') msg++; int mlen = resval - (int) (msg - admin_buf); if (mlen > 0) { admin_msg[ 0 ] = 'A'; memcpy(admin_msg + 1, msg, mlen); got_msg = 1; resval = mlen; } } else if (admin_buf[ 0 ] == 'C') { char ip[ 64 ], port[ 16 ]; if (sscanf(admin_buf + 1, " %63s %15s", ip, port) == 2) connect_relay(ip, port); else printf("usage: C \n"); } else { printf("unknown admin command\n"); } } } // shutdown: notify all clients and exit if (unlikely(run_shutdown)) { for (int i = 0; i < active; i++) { onlytcp(write(clients[ i ].fd, "bye\n", 4)); onlytcp(close(clients[ i ].fd)); onlyudp(sendto(sock, "bye\n", 4, 0, (struct sockaddr*) &clients[ i ].addr, clients[ i ].addrlen)); } active = 0; break; } // accept new connections onlytcp(if (fds[ 0 ].revents & POLLIN) { int c; while ((c = accept(sock, 0, 0)) >= 0) add_client(c); }); onlyudp( struct sockaddr_storage udp_src; // socklen_t udp_srclen = sizeof(udp_src); char buf[ env_bufs ]; int rd; int j = -1); // process existing clients onlytcp(for (int i = relay_base - 1; i >= 2; i--)) // onlyudp( while (fds[ 0 ].revents & POLLIN && // (udp_srclen = sizeof(udp_src)) && (rd = recvfrom(sock, buf, sizeof(buf), 0, (struct sockaddr*) &udp_src, &udp_srclen)) > 0)) // { onlytcp(int j = i - 2); onlyudp( // find existing client j = -1; for (int k = 0; k < active; k++) { if (clients[ k ].addrlen == udp_srclen && memcmp(&clients[ k ].addr, &udp_src, udp_srclen) == 0) { j = k; break; } } // register new client if (j < 0) { if (active >= env_max_clients) { sendto(sock, "full\n", 5, 0, (struct sockaddr*) &udp_src, udp_srclen); continue; } if (active >= cap) { cap = cap ? cap * 2 : 64; if (cap > env_max_clients) cap = env_max_clients; clients = realloc(clients, cap * sizeof(struct client)); } j = active; clients[ j ].addr = udp_src; clients[ j ].addrlen = udp_srclen; clients[ j ].id = nid++; clients[ j ].pass = default_pass; clients[ j ].last_rl = now; clients[ j ].rl_bytes = 0; active++; // send timeout amount char tbuf[ 32 ]; int tlen = sprintf(tbuf, "T%d\n", env_client_timeout); sendto(sock, tbuf, tlen, 0, (struct sockaddr*) &udp_src, udp_srclen); char ibuf[ 32 ]; int ilen = sprintf(ibuf, "IU%" PRIu64 "\n", clients[ j ].id); sendto(sock, ibuf, ilen, 0, (struct sockaddr*) &udp_src, udp_srclen); log("clients:%d\n", active); } onlyudp(clients[ j ].last_seen = now); // ); onlytcp(if (likely(fds[ i ].revents & POLLIN))) { onlytcp(char buf[ env_bufs ]; int rd = read(fds[ i ].fd, buf, sizeof(buf)); if (rd <= 0) { del_client(j); if (got_msg && j < active) write(clients[ j ].fd, admin_msg, resval + 1); continue; }); if (buf[ 0 ] == 'P' && rd >= 9) { // set password memcpy(&clients[ j ].pass, buf + 1, 8); onlytcp(write(clients[ j ].fd, "p_ok\n", 5)); } else if (buf[ 0 ] == 'M' && rd >= 9) { // M: broadcast with password if (likely(now - clients[ j ].last_rl >= env_rate_limit_s) && !DISABLE_RATELIMIT) { clients[ j ].last_rl = now; clients[ j ].rl_bytes = 0; } int plen = rd - 1; if (unlikely(clients[ j ].rl_bytes + plen > env_bufs) && !DISABLE_RATELIMIT) { onlytcp(write(clients[ j ].fd, "Rate\n", 5)); onlyudp(sendto(sock, "Rate\n", 5, 0, (struct sockaddr*) &udp_src, udp_srclen)); } else { clients[ j ].rl_bytes += plen; broadcast_msg(clients[ j ].id, buf + 1, plen); onlytcp(write(clients[ j ].fd, "m_ok\n", 5)); } } else if (buf[ 0 ] == 'N') { // no-op } else if (buf[ 0 ] == 'W') { char str_to_send[] = { 'W', env_wildcard + '0', '\n' }; onlytcp(write(clients[ j ].fd, str_to_send, 3)); onlyudp(sendto(sock, str_to_send, 3, 0, (struct sockaddr*) &udp_src, udp_srclen)); } else { onlytcp(write(clients[ j ].fd, "Err\n", 4)); onlyudp(sendto(sock, "Err\n", 4, 0, (struct sockaddr*) &udp_src, udp_srclen)); } } // send admin message onlytcp(if (unlikely(got_msg)) write(clients[ j ].fd, admin_msg, resval + 1)); } // process relay fds pthread_mutex_lock(&relay_mutex); for (int i = relay_count - 1; i >= 0; i--) { struct relay* r = &relays[ i ]; struct pollfd* pf = &fds[ relay_base + i ]; // udp heartbeat if (r->is_udp && r->remote_timeout > 0 && now - r->last_hb >= r->remote_timeout / 2) { write(r->fd, "N\n", 2); r->last_hb = now; } if (!(pf->revents & POLLIN)) continue; char rbuf[ BUFS ]; int rrd; rrd = read(r->fd, rbuf, sizeof(rbuf)); if (rrd <= 0) { printf("leave(relay)\n"); close(r->fd); if (i < relay_count - 1) relays[ i ] = relays[ relay_count - 1 ]; relay_count--; continue; } rbuf[ rrd ] = 0; // broadcast relay msg with id1 if (rbuf[ 0 ] == 'M' && rrd >= 9) { broadcast_msg(1, rbuf + 1, rrd - 1); } } pthread_mutex_unlock(&relay_mutex); onlyudp(if (unlikely(got_msg)) { for (int i = 0; i < active; i++) sendto(sock, admin_msg, resval + 1, 0, (struct sockaddr*) &clients[ i ].addr, clients[ i ].addrlen); }); } close(sock); free(clients); return 0; }