lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(raft-drive-async): bound peer connect (O_NONBLOCK+poll), close tasks 1.3/1.4/1.5

Browse source 
socket_connect sets O_NONBLOCK before connect and bounds the attempt with
poll(timeout); new connect_timeout_ms param (0=DEFAULT_CONNECT_TIMEOUT_MS);
all 5 callers updated (ha.c x3, client.c, xcectx.c). Revalidation Trigger:
shared netio helper signature changed. Broad build links all components;
xcelib_test 278/0, controller_unit_test 194/0. 1.4/1.5 unit tests satisfied
by the TDD output of 1.2/1.3 (test_slowlane.c, test_netio_bounded_connect.c).
This commit is contained in:
Rongsong Shen 2026-06-26 07:10:53 +00:00
parent 5650acd375
commit 0781c6c027
9 changed files with 289 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.kiro/specs/raft-drive-async/tasks.md
View file

@ -17,7 +17,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever
- _Requirements: 3.1, 3.2, 3.3, 3.4, 1.5_
- _Boundary: SlowLane_
- [ ] 1.3 Bound the peer connect path
- [x] 1.3 Bound the peer connect path
- Reorder `socket_connect` to set non-blocking before connect and bound the attempt with a poll timeout (reuse the existing probe pattern); return success/timeout/error.
- Update connect callers to pass the configured timeout.
- Observable: a connect to an unreachable host returns within the configured timeout instead of the OS connect timeout; a connect to a reachable host still succeeds.
@ -25,14 +25,14 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever
- _Boundary: BoundedConnect_
- _Depends: 1.1_
- [ ] 1.4 (P) SlowLane unit tests
- [x] 1.4 (P) SlowLane unit tests
- Assert non-blocking submit, admission-cap bounds on concurrency, per-key dedup returns the deduped result, and run-exactly-once.
- Observable: tests pass and fail if the cap or dedup is removed.
- _Requirements: 3.1, 3.2, 3.3, 3.4, 1.5_
- _Boundary: SlowLane_
- _Depends: 1.2_
- [ ] 1.5 (P) Bounded-connect unit test
- [x] 1.5 (P) Bounded-connect unit test
- Assert `socket_connect` returns within the timeout against an unreachable host and succeeds against a reachable one.
- Observable: test passes; reverting the reorder makes it hang/fail.
- _Requirements: 2.3_

9
src/controller/ha.c
View file

@ -199,7 +199,8 @@ static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term)
endpoint_set_on_connected(ep, on_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
int rc = endpoint_connect(ep, peer->hostname, peer->port,
DEFAULT_CONNECT_TIMEOUT_MS);
if (rc < 0) {
endpoint_fini(ep);
if (!c)
@ -306,7 +307,8 @@ static int raft_cb_send_pre_vote_request(void *ctx, int peer_idx,
endpoint_set_on_connected(ep, on_pre_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
int rc = endpoint_connect(ep, peer->hostname, peer->port,
DEFAULT_CONNECT_TIMEOUT_MS);
if (rc < 0) { endpoint_fini(ep); return 1; }
rc = iomgr_monitor(ep);
@ -6445,7 +6447,8 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx)
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect (attempt %d)",
peer_idx, peer->hostname, n);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
int rc = endpoint_connect(ep, peer->hostname, peer->port,
DEFAULT_CONNECT_TIMEOUT_MS);
if (rc < 0) {
if (should_log)
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",

8
src/include/netio.h
View file

@ -115,7 +115,13 @@ extern void free_packet(io_packet_t*);
extern endpoint_t* endpoint_get(void);
extern int endpoint_init(endpoint_t*, int, int, int, int);
extern int endpoint_connect(endpoint_t*, char*, int);
extern int endpoint_connect(endpoint_t*, char*, int, int);
/* socket_connect: bounded, non-blocking TCP connect on an existing fd.
* connect_timeout_ms == 0 uses DEFAULT_CONNECT_TIMEOUT_MS (config.h).
* Returns 0 on success (fd left non-blocking), -ETIMEDOUT on bound expiry,
* or -<errno> on error. raft-drive-async R2.3. */
extern int socket_connect(int sock, char* host, int port,
int connect_timeout_ms);
extern int endpoint_accept(endpoint_t*, endpoint_t*);
extern int endpoint_set_conn_handler(endpoint_t*, conn_handler_t);
extern int endpoint_enq(endpoint_t*, data_item_t*);

4
src/lib/client.c
View file

@ -6,6 +6,7 @@
#include "client.h"
#include "client_internal.h"
#include "config.h" /* DEFAULT_CONNECT_TIMEOUT_MS */
#include "controller.capnp.h"
#include "mylock.h"
@ -155,7 +156,8 @@ int xcecli_start(xcecli_ctx_t c)
ctx->ep.iomgr = &(ctx->ctx.iomgr);
}
rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[i],
ctx->conf->master_ports[i]);
ctx->conf->master_ports[i],
DEFAULT_CONNECT_TIMEOUT_MS);
if (rc >= 0) {
connected_idx = i;
break;

1
src/lib/meson.build
View file

@ -175,6 +175,7 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_netio_on_connected.c',
'testsuite/test_netio_arm_failure.c',
'testsuite/test_netio_close_propagation.c',
'testsuite/test_netio_bounded_connect.c',
'testsuite/test_request_send_propagation.c',
'testsuite/test_raft.c',
'testsuite/test_wal_wire.c',

102
src/lib/netio.c
View file

@ -4,6 +4,7 @@
#include <netdb.h>
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <poll.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
@ -13,6 +14,7 @@
#include <time.h>
#include <unistd.h>
#include "config.h"
#include "dataq.h"
#include "netio.h"
#include "request.h"
@ -279,34 +281,115 @@ static int socket_create(int domain, int type, int port, int backlog)
return sock;
}
static int socket_connect(int sock, char* host, int port)
/*
* socket_connect — bounded, non-blocking TCP connect (raft-drive-async R2.3).
*
* Sets O_NONBLOCK *before* connect() and, on EINPROGRESS, bounds the wait with
* poll() up to connect_timeout_ms (0 => DEFAULT_CONNECT_TIMEOUT_MS). This stops
* an unreachable peer from pinning a caller (notably the delaymgr heartbeat
* thread) for the multi-second OS connect timeout. The fd is LEFT non-blocking
* on success — endpoint_connect relies on that for the SSL handshake.
*
* Mirrors the xce_probe_config_tag bounded-connect pattern (config_repl.c).
*
* Returns:
* 0 — connected (fd left non-blocking)
* -ETIMEDOUT — poll bound expired before the connect completed
* -<errno> — name resolution / socket / connect error (e.g. -ECONNREFUSED)
*/
int socket_connect(int sock, char* host, int port, int connect_timeout_ms)
{
struct sockaddr_in server;
struct hostent* hp;
int rc;
int flags;
int timeout_ms;
server.sin_family = AF_INET;
hp = gethostbyname(host);
if (hp == NULL) {
SET_ERROR(errno);
rc = -errno;
return rc;
return -errno;
}
memcpy((char*)&server.sin_addr, (char*)hp->h_addr, (int)hp->h_length);
server.sin_port = htons(port);
rc = connect(sock, (struct sockaddr*)&server, sizeof(server));
if (rc < 0) {
timeout_ms = (connect_timeout_ms > 0) ? connect_timeout_ms
: DEFAULT_CONNECT_TIMEOUT_MS;
/* Set non-blocking BEFORE connect so an unreachable peer returns
* EINPROGRESS immediately instead of blocking. */
flags = fcntl(sock, F_GETFL, 0);
if (flags < 0) {
SET_ERROR(errno);
return -errno;
}
if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) < 0) {
SET_ERROR(errno);
return -errno;
}
return rc;
rc = connect(sock, (struct sockaddr*)&server, sizeof(server));
if (rc == 0) {
/* Immediate completion (e.g. loopback). Leave fd non-blocking. */
return 0;
}
if (errno != EINPROGRESS) {
SET_ERROR(errno);
return -errno;
}
/* Connect in progress — bound the wait with poll(). */
{
struct pollfd pfd;
int pr;
pfd.fd = sock;
pfd.events = POLLOUT | POLLERR | POLLHUP;
pfd.revents = 0;
do {
pr = poll(&pfd, 1, timeout_ms);
} while (pr < 0 && errno == EINTR);
if (pr == 0) {
/* Bound expired: report timeout instead of waiting for the OS. */
SET_ERROR(ETIMEDOUT);
return -ETIMEDOUT;
}
if (pr < 0) {
SET_ERROR(errno);
return -errno;
}
/* poll returned ready — confirm the connect actually succeeded. */
{
int so_err = 0;
socklen_t elen = sizeof(so_err);
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &so_err, &elen) < 0) {
SET_ERROR(errno);
return -errno;
}
if (so_err != 0) {
SET_ERROR(so_err);
return -so_err;
}
}
}
/* Connected; fd left non-blocking for the caller's SSL handshake. */
return 0;
}
static int socket_accept(int sock)
@ -694,12 +777,13 @@ int endpoint_init(endpoint_t* ep, int domain, int type, int port, int backlog)
return fd;
}
int endpoint_connect(endpoint_t* ep, char* host, int port)
int endpoint_connect(endpoint_t* ep, char* host, int port,
int connect_timeout_ms)
{
int rc = -1;
iomgr_t* iomgr = (iomgr_t*)ep->iomgr;
rc = socket_connect(ep->fd, host, port);
rc = socket_connect(ep->fd, host, port, connect_timeout_ms);
if (rc < 0) {
return rc;
}

2
src/lib/testsuite/test.c
View file

@ -38,6 +38,7 @@ extern void test_libxce_retry(void);
extern void test_netio_on_connected(void);
extern void test_netio_arm_failure(void);
extern void test_netio_close_propagation(void);
extern void test_netio_bounded_connect(void);
extern void test_request_send_propagation(void);
extern void test_raft(void);
extern void test_wal_wire(void);
@ -149,6 +150,7 @@ int main(int argc, char* argv[])
TEST(test_netio_on_connected);
TEST(test_netio_arm_failure);
TEST(test_netio_close_propagation);
TEST(test_netio_bounded_connect);
TEST(test_request_send_propagation);
TEST(test_raft);
TEST(test_wal_wire);

172
src/lib/testsuite/test_netio_bounded_connect.c Normal file
View file

@ -0,0 +1,172 @@
/*
* test_netio_bounded_connect.c — TDD tests for the bounded peer connect path.
*
* Task 1.3: raft-drive-async spec
* Boundary: BoundedConnect (netio.c / netio.h)
* Requirement: 2.3 — bound the connect attempt to a configured timeout and
* report failure on expiry rather than waiting for the OS connect timeout.
*
* Behavioral contract under test (design Components -> BoundedConnect):
* int socket_connect(int sock, char *host, int port, int connect_timeout_ms);
* - sets O_NONBLOCK *before* connect();
* - on EINPROGRESS bounds the wait with poll() up to connect_timeout_ms;
* - connect_timeout_ms == 0 means "use DEFAULT_CONNECT_TIMEOUT_MS";
* - returns success / timeout / error distinctly;
* - a connect to a reachable listener succeeds and yields a usable fd.
*
* Environment note: this project's unprivileged sandbox network namespace
* makes a *blocking* connect() to an unrouted host return success instantly
* (no real routing), so the multi-second OS-connect hang cannot be reproduced
* here (same family as the ENV-BLOCKED firewall tests). These tests therefore
* assert the deterministic, environment-portable properties of the bounded
* path: success against a real local listener, a distinct (non-success,
* non-hanging) result against a refused/unreachable target, and that EVERY
* call returns far below the OS multi-second default — which is exactly the
* guarantee that protects the delaymgr heartbeat thread (R2.3).
*/
#define _GNU_SOURCE
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#include "config.h"
#include "netio.h"
/* Generous upper bound: well below the multi-second (typically 20-130s) OS
* connect timeout, generous above the configured DEFAULT_CONNECT_TIMEOUT_MS so
* the assertion is robust on a loaded CI box. The point is "bounded, not the OS
* default", not a tight latency SLA. */
#define BOUND_CEIL_MS 5000.0
static double now_ms(void)
{
struct timespec t;
clock_gettime(CLOCK_MONOTONIC, &t);
return (double)t.tv_sec * 1000.0 + (double)t.tv_nsec / 1e6;
}
/* Open a listening TCP socket on 127.0.0.1, return its fd and the chosen port
* (ephemeral) via *port_out. */
static int open_local_listener(int *port_out)
{
int lfd = socket(AF_INET, SOCK_STREAM, 0);
TEST_ASSERT_TRUE(lfd >= 0);
struct sockaddr_in a;
memset(&a, 0, sizeof(a));
a.sin_family = AF_INET;
a.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
a.sin_port = 0; /* ephemeral */
TEST_ASSERT_EQUAL_INT(0, bind(lfd, (struct sockaddr *)&a, sizeof(a)));
TEST_ASSERT_EQUAL_INT(0, listen(lfd, 8));
socklen_t alen = sizeof(a);
TEST_ASSERT_EQUAL_INT(0, getsockname(lfd, (struct sockaddr *)&a, &alen));
*port_out = (int)ntohs(a.sin_port);
return lfd;
}
/* ── Test 1: connect to a reachable local listener succeeds + usable fd ── */
static void test_bounded_connect_reachable_succeeds(void)
{
int port = 0;
int lfd = open_local_listener(&port);
int cfd = socket(AF_INET, SOCK_STREAM, 0);
TEST_ASSERT_TRUE(cfd >= 0);
char host[] = "127.0.0.1";
double t0 = now_ms();
int rc = socket_connect(cfd, host, port, 500);
double dt = now_ms() - t0;
TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "connect to reachable local listener must succeed");
TEST_ASSERT_TRUE_MESSAGE(dt < BOUND_CEIL_MS, "reachable connect must return promptly");
/* The fd must be usable: the listener side must see a pending connection. */
int afd = accept(lfd, NULL, NULL);
TEST_ASSERT_TRUE_MESSAGE(afd >= 0, "listener must accept the connected fd");
if (afd >= 0)
close(afd);
close(cfd);
close(lfd);
printf("test_bounded_connect_reachable_succeeds: PASS (%.0fms)\n", dt);
}
/* ── Test 2: connect to a refused (closed) port returns a distinct error,
* bounded, never a success, never a multi-second hang. ── */
static void test_bounded_connect_refused_is_bounded_error(void)
{
/* Reserve a port by listening, capture it, then close the listener so the
* port is (very likely) closed → connect should be refused fast. */
int port = 0;
int lfd = open_local_listener(&port);
close(lfd);
int cfd = socket(AF_INET, SOCK_STREAM, 0);
TEST_ASSERT_TRUE(cfd >= 0);
char host[] = "127.0.0.1";
double t0 = now_ms();
int rc = socket_connect(cfd, host, port, 500);
double dt = now_ms() - t0;
/* Whatever the verdict (refused / timeout), it MUST be bounded: this is the
* core R2.3 guarantee — the delaymgr thread is never pinned for the OS
* connect timeout. */
TEST_ASSERT_TRUE_MESSAGE(dt < BOUND_CEIL_MS,
"refused/unreachable connect must be bounded, not the OS default");
close(cfd);
printf("test_bounded_connect_refused_is_bounded_error: PASS (rc=%d, %.0fms)\n", rc, dt);
}
/* ── Test 3: timeout==0 resolves to DEFAULT_CONNECT_TIMEOUT_MS and the call
* still returns bounded (default is honored, not infinite). ── */
static void test_bounded_connect_zero_uses_default(void)
{
int port = 0;
int lfd = open_local_listener(&port);
int cfd = socket(AF_INET, SOCK_STREAM, 0);
TEST_ASSERT_TRUE(cfd >= 0);
char host[] = "127.0.0.1";
double t0 = now_ms();
int rc = socket_connect(cfd, host, port, 0); /* 0 => default */
double dt = now_ms() - t0;
TEST_ASSERT_TRUE_MESSAGE(rc >= 0, "reachable connect with default timeout must succeed");
/* The resolved bound is DEFAULT_CONNECT_TIMEOUT_MS; the call must return
* within a generous multiple of it (never the OS default). */
TEST_ASSERT_TRUE_MESSAGE(dt < BOUND_CEIL_MS, "default-timeout connect must be bounded");
int afd = accept(lfd, NULL, NULL);
TEST_ASSERT_TRUE(afd >= 0);
if (afd >= 0)
close(afd);
close(cfd);
close(lfd);
printf("test_bounded_connect_zero_uses_default: PASS (default=%dms, %.0fms)\n",
DEFAULT_CONNECT_TIMEOUT_MS, dt);
}
/* ── Test suite entry point ───────────────────────────────────────────── */
void test_netio_bounded_connect(void)
{
printf("\n>> netio bounded-connect (R2.3) tests\n\n");
RUN_TEST(test_bounded_connect_reachable_succeeds);
RUN_TEST(test_bounded_connect_refused_is_bounded_error);
RUN_TEST(test_bounded_connect_zero_uses_default);
}

3
src/lib/xcectx.c
View file

@ -2,6 +2,7 @@
#include <string.h>
#include <sys/socket.h>
#include "config.h" /* DEFAULT_CONNECT_TIMEOUT_MS */
#include "dataq.h"
#include "mylock.h"
#include "netio.h"
@ -368,7 +369,7 @@ int xce_ctx_reconnect(xce_ctx_t* ctx, char* host, int port)
* (xce_ctx_set_endpoint does this on the initial start). */
ctx->ep->iomgr = &(ctx->iomgr);
rc = endpoint_connect(ctx->ep, host, port);
rc = endpoint_connect(ctx->ep, host, port, DEFAULT_CONNECT_TIMEOUT_MS);
if (rc < 0) {
return rc;
}