diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 26c6bd2..6453965 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -155,6 +155,10 @@ jobs: run: python3 unittest/io_timeout_check.py - name: Check addressed UDP with bounded children run: python3 unittest/udp_datagram_check.py + - name: Check TCP resolver lifecycle and deadlines + run: | + python3 unittest/socket_glue_check.py + python3 unittest/tcp_connect_check.py - name: Compare every corpus program run: sh run_corpus.sh - name: Check YAML store round trips diff --git a/.gitignore b/.gitignore index f51bd3c..ff6a746 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ !vendor/prism/src/*.c !vendor/prism/src/util/*.c !tools/prism_glue.c +!tools/socket_glue.c vendor/prism/obj/ mere_extern.h *.o diff --git a/KNOWN_GAPS.md b/KNOWN_GAPS.md index 64291da..646c910 100644 --- a/KNOWN_GAPS.md +++ b/KNOWN_GAPS.md @@ -1070,11 +1070,15 @@ on a chosen address and answers getsockname / getpeername itself: whatever the kernel named -- for pipes and files too, since the layer is shared. The messages are the class's own text; ruby's sometimes add the call site (`@ io_fillbuf - fd:11`), which is not reproduced. -- **TCP, either family.** `TCPSocket`/`Socket#connect` go through - `tcp_connect`. `TCPSocket.new` resolves the host itself and tries each - address in turn, as ruby does ("localhost" is ::1, then 127.0.0.1); a name - the files half cannot resolve is left to `tcp_connect`'s own resolver, - which tries only its first address. `TCPServer.new` is the runtime's +- **TCP, either family.** `TCPSocket.new` resolves IPv4 and IPv6 on native + workers and polls nonblocking connections while yielding to Ruby threads. + It enforces open, resolution, and connection deadlines, binds a requested + local address/port before connecting, and supports the resolver delay/error + controls used by Ruby's socket tests. Fast fallback prefers IPv6 within a + 50 ms resolution window and starts another attempt after 250 ms; disabling + it serializes connection attempts. Resolver workers own libc data only and + release it when a cancelled resolution eventually returns. `Socket#connect` + still uses the older blocking `tcp_connect` path. `TCPServer.new` uses tcp_listen_at, which is init_inetsock_internal's loop: every AI_PASSIVE answer in turn gets socket(2), SO_REUSEADDR and bind(2), the first that binds is the listener, and a refusal is the Errno the kernel named diff --git a/corpus/503_tcp_connect_deadlines.rb b/corpus/503_tcp_connect_deadlines.rb new file mode 100644 index 0000000..601c8fd --- /dev/null +++ b/corpus/503_tcp_connect_deadlines.rb @@ -0,0 +1,66 @@ +require 'socket' + +# A listening socket can queue a connection without a Ruby accept thread. +# Failures therefore leave no thread blocked in an ensure clause. +def exchange(host, **options) + server = TCPServer.new(host, 0) + client = TCPSocket.new('localhost', server.addr[1], **options) + accepted = server.accept + accepted.write("a\0b") + p [client.remote_address.ip_address == host, client.recvfrom(3)] + accepted.close + accepted = nil + p client.recvfrom(1) # EOF is nil, not a two-element packet. +ensure + accepted&.close + client&.close + server&.close +end + +exchange('127.0.0.1', test_mode_settings: {delay: {ipv6: 1000}}) +exchange('127.0.0.1', test_mode_settings: {error: {ipv6: Socket::EAI_FAIL}}) +# A deadline for the unresolved family must preserve the IPv4 answer and +# release the preference wait, even before that answer has been attempted. +exchange('127.0.0.1', resolv_timeout: 0.01, + test_mode_settings: {delay: {ipv6: 1000}}) + +# A failed connection does not hide the unresolved family's timeout when +# there is no usable address or active attempt left. +closed = TCPServer.new('127.0.0.1', 0) +port = closed.addr[1] +closed.close +begin + TCPSocket.new('localhost', port, resolv_timeout: 0.01, + test_mode_settings: {delay: {ipv4: 1000}}) +rescue IO::TimeoutError + p :resolution_timeout_after_connection_failure +end + +[:open_timeout, :resolv_timeout].each do |kind| + begin + TCPSocket.new('localhost', 1, **{kind => 0.02}, + test_mode_settings: {delay: {ipv4: 1000, ipv6: 1000}}) + rescue IO::TimeoutError + p kind + end +end + +begin + TCPSocket.new('localhost', 1, open_timeout: 1, connect_timeout: 1) +rescue ArgumentError + p :conflicting_timeouts +end + +server = TCPServer.new('127.0.0.1', 0) +client = TCPSocket.new('127.0.0.1', server.addr[1], '127.0.0.1', 0, + connect_timeout: 1, fast_fallback: false) +accepted = server.accept +p client.local_address.ip_address == '127.0.0.1' +client.close +accepted.close +server.close + +# Each address comes from getifaddrs; IPv6 scope IDs must survive conversion. +addresses = Socket.ip_address_list +p [addresses.any?(&:ipv4_loopback?), addresses.all?(&:ip?), + addresses.all? { |a| a.ip_port == 0 && a.socktype == 0 }] diff --git a/m_eval_leaf.mere b/m_eval_leaf.mere index eb9b056..4a70bf4 100644 --- a/m_eval_leaf.mere +++ b/m_eval_leaf.mere @@ -6432,6 +6432,10 @@ and stdlib_src = fn (name: str) -> "end\n" ++ "\n" ++ "class TCPSocket < IPSocket\n" ++ + " def recvfrom(maxlen, flags = 0)\n" ++ + " data = recv(maxlen, flags)\n" ++ + " data.nil? ? nil : [data, nil]\n" ++ + " end\n" ++ " def self.gethostbyname(host)\n" ++ " list = Addrinfo.__getaddrinfo(host, nil, 0, Socket::SOCK_STREAM, 0, Socket::AI_CANONNAME, false)\n" ++ " names = [list[0].canonname || host, []]\n" ++ @@ -6456,23 +6460,117 @@ and stdlib_src = fn (name: str) -> " end\n" ++ " -1\n" ++ " end\n" ++ - " # A connect here is the runtime's blocking connect(2): it cannot be given\n" ++ - " # a deadline. A deadline of zero is the one that is known in advance -- no\n" ++ - " # connect completes in no time, and ruby's own nonblocking connect times out\n" ++ - " # on it even over loopback -- so it is answered without connecting. A\n" ++ - " # positive one is not enforced (KNOWN_GAPS).\n" ++ - " def initialize(host, port, local_host = nil, local_port = nil, connect_timeout: nil, open_timeout: nil, resolv_timeout: nil, fast_fallback: true)\n" ++ + " # Resolution workers never run Ruby. This loop owns their lifetime and yields\n" ++ + " # while DNS or connect(2) is pending, including during timeout cleanup.\n" ++ + " def self.__connect_with_deadlines(host, port, local_host, local_port, connect_timeout, open_timeout, resolv_timeout, fast_fallback, settings)\n" ++ + " start = Process.clock_gettime(Process::CLOCK_MONOTONIC)\n" ++ + " jobs = \{}\n" ++ + " ready = \{6 => [], 4 => []}\n" ++ + " active = []\n" ++ + " last_error = nil\n" ++ + " resolution_error = nil\n" ++ + " resolution_timeout_error = nil\n" ++ + " first_v4 = nil\n" ++ + " next_attempt = start\n" ++ + " last_family = 4\n" ++ + " local = local_host.nil? ? '' : Addrinfo.__getaddrinfo(local_host, nil, 0, 0, 0, 0, false)[0].ip_address\n" ++ + " families = host.include?(':') ? [6] : (host.match?(/\\A[0-9.]+\\z/) ? [4] : [6, 4])\n" ++ + " settings = \{} unless settings.is_a?(Hash)\n" ++ + " delays = settings[:delay] || \{}\n" ++ + " errors = settings[:error] || \{}\n" ++ + " families.each do |family|\n" ++ + " key = family == 6 ? :ipv6 : :ipv4\n" ++ + " job = Socket.__resolve_start(host, port, family, Integer(delays[key] || 0), Integer(errors[key] || 0))\n" ++ + " raise SystemCallError.new('getaddrinfo worker', -job) if job < 0\n" ++ + " jobs[family] = job\n" ++ + " end\n" ++ + " loop do\n" ++ + " now = Process.clock_gettime(Process::CLOCK_MONOTONIC)\n" ++ + " raise IO::TimeoutError, 'user specified open timeout' if open_timeout && now - start >= open_timeout\n" ++ + " jobs.keys.each do |family|\n" ++ + " job = jobs[family]\n" ++ + " next if Socket.__resolve_done(job) == 0\n" ++ + " error = Socket.__resolve_error(job)\n" ++ + " if error != 0\n" ++ + " resolution_error = Socket::ResolutionError.new('getaddrinfo failed')\n" ++ + " else\n" ++ + " index = 0\n" ++ + " loop do\n" ++ + " address = Socket.__resolve_address(job, index)\n" ++ + " break if address.empty?\n" ++ + " ready[family] << address unless ready[family].include?(address)\n" ++ + " index += 1\n" ++ + " end\n" ++ + " first_v4 ||= now if family == 4\n" ++ + " end\n" ++ + " Socket.__resolve_release(job)\n" ++ + " jobs.delete(family)\n" ++ + " end\n" ++ + " active.dup.each do |entry|\n" ++ + " fd, since = entry\n" ++ + " if (IO.__poll([fd], [2], 0)[0] & 6) != 0\n" ++ + " error = Socket.__connect_error(fd)\n" ++ + " active.delete(entry)\n" ++ + " return fd if error == 0\n" ++ + " IO.__fdclose(fd)\n" ++ + " last_error = SystemCallError.new(\"connect(2) for #\{host.inspect} port #\{port}\", error)\n" ++ + " next_attempt = now\n" ++ + " elsif connect_timeout && now - since >= connect_timeout\n" ++ + " active.delete(entry)\n" ++ + " IO.__fdclose(fd)\n" ++ + " last_error = IO::TimeoutError.new('user specified connect timeout')\n" ++ + " next_attempt = now\n" ++ + " end\n" ++ + " end\n" ++ + " if resolv_timeout && !jobs.empty? && now - start >= resolv_timeout\n" ++ + " jobs.each_value \{ |job| Socket.__resolve_release(job) }\n" ++ + " jobs.clear\n" ++ + " resolution_timeout_error = IO::TimeoutError.new('user specified resolution timeout')\n" ++ + " # Resolved addresses and active attempts can still connect. Removing\n" ++ + " # the pending jobs also ends the IPv6 preference wait.\n" ++ + " end\n" ++ + " if now >= next_attempt && (fast_fallback || active.empty?)\n" ++ + " family = last_family == 6 ? 4 : 6\n" ++ + " family = 10 - family if ready[family].empty?\n" ++ + " wait_v6 = family == 4 && jobs.key?(6) && first_v4 && now - first_v4 < 0.05\n" ++ + " unless ready[family].empty? || wait_v6\n" ++ + " address = ready[family].shift\n" ++ + " last_family = family\n" ++ + " fd = Socket.__connect_start(address, port, local, local_port || 0)\n" ++ + " if fd < 0\n" ++ + " context = local_host || local_port ? \"bind/connect(2) for #\{local_host.inspect} port #\{local_port}\" : \"connect(2) for #\{host.inspect} port #\{port}\"\n" ++ + " last_error = SystemCallError.new(context, -fd)\n" ++ + " next_attempt = now\n" ++ + " else\n" ++ + " active << [fd, now]\n" ++ + " next_attempt = now + 0.25\n" ++ + " end\n" ++ + " next\n" ++ + " end\n" ++ + " end\n" ++ + " if jobs.empty? && active.empty? && ready[4].empty? && ready[6].empty?\n" ++ + " raise(resolution_timeout_error || last_error || resolution_error || Socket::ResolutionError.new('getaddrinfo returned no addresses'))\n" ++ + " end\n" ++ + " sleep 0.001\n" ++ + " end\n" ++ + " ensure\n" ++ + " jobs&.each_value \{ |job| Socket.__resolve_release(job) }\n" ++ + " active&.each \{ |entry| IO.__fdclose(entry[0]) }\n" ++ + " end\n" ++ + " def initialize(host, port, local_host = nil, local_port = nil, connect_timeout: nil, open_timeout: nil, resolv_timeout: nil, fast_fallback: true, test_mode_settings: nil)\n" ++ " __new_block_warn if block_given?\n" ++ + " if open_timeout && (connect_timeout || resolv_timeout)\n" ++ + " raise ArgumentError, 'open_timeout cannot be combined with connect_timeout or resolv_timeout'\n" ++ + " end\n" ++ + " raise Socket::ResolutionError, 'getaddrinfo: no service specified' if port.nil?\n" ++ " @host = host\n" ++ " @port = port\n" ++ - " t = open_timeout || connect_timeout\n" ++ - " raise IO::TimeoutError, \"user specified timeout for #\{host}:#\{port}\" if t && t <= 0\n" ++ - " fd = TCPSocket.__connect_any(host, Addrinfo.__port_of(port, \"tcp\"))\n" ++ - " if fd < 0\n" ++ - " raise Errno::ECONNREFUSED, \"connect(2) for #\{host.inspect} port #\{port}\"\n" ++ + " [open_timeout, connect_timeout, resolv_timeout].each do |timeout|\n" ++ + " raise IO::TimeoutError, 'user specified timeout' if timeout && timeout <= 0\n" ++ " end\n" ++ + " host = host.nil? ? '' : host.to_str\n" ++ + " fd = TCPSocket.__connect_with_deadlines(host, Addrinfo.__port_of(port, 'tcp'), local_host, local_port && Addrinfo.__port_of(local_port, 'tcp'), connect_timeout, open_timeout, resolv_timeout, fast_fallback, test_mode_settings)\n" ++ " __setup(fd)\n" ++ - " Socket::Endpoint__.check_local(self, local_host, local_port) if local_host || local_port\n" ++ " end\n" ++ "end\n" ++ "\n" ++ @@ -6732,6 +6830,9 @@ and stdlib_src = fn (name: str) -> "end\n" ++ "\n" ++ "class Socket < BasicSocket\n" ++ + " def self.ip_address_list\n" ++ + " Socket.__ip_addresses.split.map \{ |address| Addrinfo.new([address].pack('H*'), nil, 0, 0) }\n" ++ + " end\n" ++ " module Constants\n" ++ " def self.__table\n" ++ " @__table ||= begin\n" ++ diff --git a/m_state.mere b/m_state.mere index 7e43de4..e8a6c9a 100644 --- a/m_state.mere +++ b/m_state.mere @@ -17,6 +17,15 @@ // writes cross through the shared byte arena (`mem_*`), so a payload with a // zero byte survives. The Wasm host has no sockets and stubs these to -1. extern fn tcp_connect: str -> int -> int; +// tools/socket_glue.c: resolver workers and nonblocking numeric connects. +extern fn mr_resolve_start: str -> int -> int -> int -> int -> int; +extern fn mr_resolve_done: int -> int; +extern fn mr_resolve_error: int -> int; +extern fn mr_resolve_address: int -> int -> str; +extern fn mr_resolve_release: int -> int; +extern fn mr_connect_start: str -> int -> str -> int -> int; +extern fn mr_connect_error: int -> int; +extern fn mr_ip_addresses: unit -> str; extern fn tcp_listen: int -> int; extern fn tcp_accept: int -> int; extern fn tcp_close: int -> unit; diff --git a/main.mere b/main.mere index b713f6d..1d00851 100644 --- a/main.mere +++ b/main.mere @@ -20850,6 +20850,25 @@ and invoke_val_c0 = fn env -> fn world -> fn blk -> fn (recv0: Val) -> fn (name0 (match args with | Cons (VStr h, Cons (VInt port, _)) -> VInt (tcp_connect (sget h) port) | _ -> VInt (0 - 1)) + else if str_eq name "__ip_addresses" then mkstr (mr_ip_addresses ()) + else if str_eq name "__resolve_start" then + (match args with + | Cons (VStr h, Cons (VInt p, Cons (VInt f, Cons (VInt d, Cons (VInt e, _))))) -> VInt (mr_resolve_start (sget h) p f d e) + | _ -> VInt (0 - 22)) + else if str_eq name "__resolve_address" then + (match args with Cons (VInt j, Cons (VInt i, _)) -> mkstr (mr_resolve_address j i) | _ -> mkstr "") + else if str_eq name "__resolve_done" then + (match args with Cons (VInt j, _) -> VInt (mr_resolve_done j) | _ -> VInt 1) + else if str_eq name "__resolve_error" then + (match args with Cons (VInt j, _) -> VInt (mr_resolve_error j) | _ -> VInt 1) + else if str_eq name "__resolve_release" then + (match args with Cons (VInt j, _) -> VInt (mr_resolve_release j) | _ -> VInt 0) + else if str_eq name "__connect_start" then + (match args with + | Cons (VStr h, Cons (VInt p, Cons (VStr l, Cons (VInt lp, _)))) -> VInt (mr_connect_start (sget h) p (sget l) lp) + | _ -> VInt (0 - 22)) + else if str_eq name "__connect_error" then + (match args with Cons (VInt fd, _) -> VInt (mr_connect_error fd) | _ -> VInt 22) else if (str_len name == 8 && str_eq name "__listen") then (match args with Cons (VInt port, _) -> VInt (tcp_listen port) | _ -> VInt (0 - 1)) // compiler v0.1.555: a listener on the host asked for (TCPServer.new's diff --git a/tools/build.sh b/tools/build.sh index 79a07a2..d32ce96 100755 --- a/tools/build.sh +++ b/tools/build.sh @@ -113,12 +113,12 @@ if [ "$pneed" = 1 ]; then ar rcs "$plib" "$pobj"/*.o || { rm -rf "$pobj"; exit 1; } fi -printf 'cc clang %s %s mr.c tools/prism_glue.c libprism.a -o mere-ruby\n' "$opt" "$plat" +printf 'cc clang %s %s mr.c tools/prism_glue.c tools/socket_glue.c libprism.a -o mere-ruby\n' "$opt" "$plat" # ⚠ NEVER `cp` OVER A RUNNING BINARY -- the text pages of a process that is # mid-run are backed by that file, and overwriting them SIGKILLs it or, worse, # gives it another build's instructions. Build beside it and rename. # shellcheck disable=SC2086 -clang $opt -w $plat -I "$root/vendor/prism/include" mr.c "$root/tools/prism_glue.c" "$plib" -o mere-ruby.new || { rm -f mere-ruby.new; exit 1; } +clang $opt -w $plat -I "$root/vendor/prism/include" mr.c "$root/tools/prism_glue.c" "$root/tools/socket_glue.c" -pthread "$plib" -o mere-ruby.new || { rm -f mere-ruby.new; exit 1; } /bin/mv -f mere-ruby.new mere-ruby # ...and say which build this is, where something that must not be misled can diff --git a/tools/socket_glue.c b/tools/socket_glue.c new file mode 100644 index 0000000..780f02a --- /dev/null +++ b/tools/socket_glue.c @@ -0,0 +1,219 @@ +/* Async DNS and numeric nonblocking TCP connection primitives. Workers own + * only libc data; they never enter Mere or access its regions. Releasing a + * resolver detaches the caller immediately, even while getaddrinfo runs. */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +struct resolver { + int id, refs, cancelled, done, error, family, delay, injected_error; + char *host, service[16]; + struct addrinfo *addresses; + struct resolver *next; +}; +static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t changed = PTHREAD_COND_INITIALIZER; +static struct resolver *jobs; +static int next_id; +static pthread_once_t fork_hooks = PTHREAD_ONCE_INIT; +static void before_fork(void) { pthread_mutex_lock(&lock); } +static void after_fork(void) { pthread_mutex_unlock(&lock); } +static void child_after_fork(void) { + /* The worker threads do not survive fork. Their inherited requests must + * neither wait forever nor alias a new request in the child. The small + * inherited libc allocations are left to process teardown. */ + jobs = NULL; + pthread_mutex_unlock(&lock); + changed = (pthread_cond_t)PTHREAD_COND_INITIALIZER; +} +static void install_fork_hooks(void) { + pthread_atfork(before_fork, after_fork, child_after_fork); +} + +static struct resolver *find_job(int id) { + struct resolver *j; + for (j = jobs; j && j->id != id; j = j->next) {} + return j; +} +static void unref(struct resolver *j) { + if (--j->refs) return; + if (j->addresses) freeaddrinfo(j->addresses); + free(j->host); + free(j); +} +static void *resolve_worker(void *arg) { + struct resolver *j = arg; + pthread_mutex_lock(&lock); + if (j->delay > 0) { + struct timespec end; + clock_gettime(CLOCK_REALTIME, &end); + end.tv_sec += j->delay / 1000; + end.tv_nsec += (j->delay % 1000) * 1000000L; + if (end.tv_nsec >= 1000000000L) { end.tv_sec++; end.tv_nsec -= 1000000000L; } + while (!j->cancelled && pthread_cond_timedwait(&changed, &lock, &end) != ETIMEDOUT) {} + } + if (j->cancelled) { unref(j); pthread_mutex_unlock(&lock); return NULL; } + pthread_mutex_unlock(&lock); + struct addrinfo hints = {0}, *addresses = NULL; + hints.ai_family = j->family; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_NUMERICSERV; + int error = j->injected_error ? j->injected_error : + getaddrinfo(j->host[0] ? j->host : NULL, j->service, &hints, &addresses); + pthread_mutex_lock(&lock); + j->addresses = addresses; + j->error = error; + j->done = 1; + unref(j); + pthread_mutex_unlock(&lock); + return NULL; +} +int mr_resolve_start(const char *host, int port, int family, int delay, int error) { + pthread_once(&fork_hooks, install_fork_hooks); + struct resolver *j = calloc(1, sizeof(*j)); + if (!j) return -ENOMEM; + j->host = strdup(host); + if (!j->host) { free(j); return -ENOMEM; } + snprintf(j->service, sizeof(j->service), "%d", port); + j->family = family == 6 ? AF_INET6 : AF_INET; + j->delay = delay; + j->injected_error = error; + j->refs = 2; + pthread_mutex_lock(&lock); + do { next_id = next_id == INT_MAX ? 1 : next_id + 1; } while (find_job(next_id)); + j->id = next_id; + j->next = jobs; + jobs = j; + pthread_t thread; + int e = pthread_create(&thread, NULL, resolve_worker, j); + if (e) { jobs = j->next; j->refs = 1; unref(j); pthread_mutex_unlock(&lock); return -e; } + pthread_detach(thread); + int id = j->id; + pthread_mutex_unlock(&lock); + return id; +} +int mr_resolve_done(int id) { + pthread_mutex_lock(&lock); + struct resolver *j = find_job(id); + int done = j ? j->done : 1; + pthread_mutex_unlock(&lock); + return done; +} +int mr_resolve_error(int id) { + pthread_mutex_lock(&lock); + struct resolver *j = find_job(id); + int error = j ? j->error : EAI_FAIL; + pthread_mutex_unlock(&lock); + return error; +} +const char *mr_resolve_address(int id, int index) { + /* The FFI copies str results before the next call. */ + static char address[NI_MAXHOST]; + address[0] = 0; + pthread_mutex_lock(&lock); + struct resolver *j = find_job(id); + struct addrinfo *a = j && j->done ? j->addresses : NULL; + while (a && index-- > 0) a = a->ai_next; + if (a && getnameinfo(a->ai_addr, a->ai_addrlen, address, sizeof(address), NULL, 0, NI_NUMERICHOST)) address[0] = 0; + pthread_mutex_unlock(&lock); + return address; +} +int mr_resolve_release(int id) { + pthread_mutex_lock(&lock); + struct resolver **p = &jobs; + while (*p && (*p)->id != id) p = &(*p)->next; + if (*p) { + struct resolver *j = *p; + *p = j->next; + j->cancelled = 1; + pthread_cond_broadcast(&changed); + unref(j); + } + pthread_mutex_unlock(&lock); + return 0; +} +int mr_connect_start(const char *host, int port, const char *local, int local_port) { + struct addrinfo hints = {0}, *a = NULL, *bound = NULL; + char service[16]; + snprintf(service, sizeof(service), "%d", port); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV; + if (getaddrinfo(host, service, &hints, &a)) return -EINVAL; + int fd = socket(a->ai_family, a->ai_socktype, a->ai_protocol), e = errno; + if (fd >= 0) { + if (fcntl(fd, F_SETFD, FD_CLOEXEC) < 0 || fcntl(fd, F_SETFL, O_NONBLOCK) < 0) e = errno; + else if (local[0] || local_port) { + snprintf(service, sizeof(service), "%d", local_port); + hints.ai_family = a->ai_family; + hints.ai_flags |= AI_PASSIVE; + if (getaddrinfo(local[0] ? local : NULL, service, &hints, &bound)) e = EADDRNOTAVAIL; + else e = bind(fd, bound->ai_addr, bound->ai_addrlen) < 0 ? errno : 0; + } else e = 0; + if (!e && connect(fd, a->ai_addr, a->ai_addrlen) < 0 && errno != EINPROGRESS && errno != EWOULDBLOCK) e = errno; + if (e) { close(fd); fd = -1; } + } + if (bound) freeaddrinfo(bound); + freeaddrinfo(a); + return fd < 0 ? -e : fd; +} +int mr_connect_error(int fd) { + int e = 0; + socklen_t n = sizeof(e); + if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &e, &n)) return errno; + return e; +} +const char *mr_ip_addresses(void) { + static char *out; + struct ifaddrs *addresses; + free(out); + out = NULL; + if (getifaddrs(&addresses)) return ""; + size_t size = 1; + out = calloc(size, 1); + if (!out) { freeifaddrs(addresses); return ""; } + for (struct ifaddrs *a = addresses; a; a = a->ifa_next) { + if (!a->ifa_addr) continue; + int family = a->ifa_addr->sa_family; + if (family != AF_INET && family != AF_INET6) continue; + /* Addrinfo's internal sockaddr representation is BSD layout on + * every host. Preserve IPv6 scope IDs instead of reparsing names. */ + unsigned char raw[28] = {0}; + size_t bytes = family == AF_INET ? 16 : 28; + raw[0] = (unsigned char)bytes; + raw[1] = (unsigned char)family; + if (family == AF_INET) { + struct sockaddr_in *in = (struct sockaddr_in *)a->ifa_addr; + memcpy(raw + 4, &in->sin_addr, 4); + } else { + struct sockaddr_in6 *in = (struct sockaddr_in6 *)a->ifa_addr; + memcpy(raw + 8, &in->sin6_addr, 16); + memcpy(raw + 24, &in->sin6_scope_id, 4); + } + char name[57]; + static const char hex[] = "0123456789abcdef"; + for (size_t i = 0; i < bytes; i++) { + name[i * 2] = hex[raw[i] >> 4]; + name[i * 2 + 1] = hex[raw[i] & 15]; + } + size_t length = bytes * 2; + char *grown = realloc(out, size + length + 1); + if (!grown) break; + out = grown; + memcpy(out + size - 1, name, length); + out[size + length - 1] = '\n'; + out[size + length] = 0; + size += length + 1; + } + freeifaddrs(addresses); + return out; +} diff --git a/unittest/socket_glue_check.py b/unittest/socket_glue_check.py new file mode 100644 index 0000000..4b73721 --- /dev/null +++ b/unittest/socket_glue_check.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Exercise native resolver ownership and numeric nonblocking connections.""" +import ctypes +import errno +import fcntl +import os +from pathlib import Path +import select +import socket +import subprocess +import tempfile +import time + +ROOT = Path(__file__).resolve().parent.parent + + +def wait_for(test): + deadline = time.monotonic() + 5 + while not test(): + assert time.monotonic() < deadline, 'native operation did not finish' + time.sleep(0.001) + + +with tempfile.TemporaryDirectory() as directory: + library = Path(directory) / 'socket_glue.so' + subprocess.run([os.environ.get('CC', 'clang'), '-shared', '-fPIC', '-pthread', + '-Wall', '-Wextra', '-Werror', str(ROOT / 'tools/socket_glue.c'), + '-o', str(library)], check=True, timeout=30) + lib = ctypes.CDLL(str(library)) + lib.mr_resolve_start.argtypes = [ctypes.c_char_p] + [ctypes.c_int] * 4 + lib.mr_resolve_address.argtypes = [ctypes.c_int, ctypes.c_int] + lib.mr_resolve_address.restype = ctypes.c_char_p + lib.mr_connect_start.argtypes = [ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_int] + lib.mr_ip_addresses.restype = ctypes.c_char_p + + job = lib.mr_resolve_start(b'localhost', 80, 4, 0, 0) + assert job > 0 + wait_for(lambda: lib.mr_resolve_done(job)) + assert lib.mr_resolve_error(job) == 0 + assert lib.mr_resolve_address(job, 0) == b'127.0.0.1' + lib.mr_resolve_release(job) + assert lib.mr_resolve_address(job, 0) == b'' + + job = lib.mr_resolve_start(b'localhost', 80, 6, 20, socket.EAI_FAIL) + wait_for(lambda: lib.mr_resolve_done(job)) + assert lib.mr_resolve_error(job) == socket.EAI_FAIL + lib.mr_resolve_release(job) + + # Release before a resolver starts, during its delay, and after completion. + # Each worker must release its remaining reference without a caller join. + for _ in range(100): + job = lib.mr_resolve_start(b'localhost', 80, 4, 5000, 0) + assert job > 0 and lib.mr_resolve_done(job) == 0 + before = time.monotonic() + lib.mr_resolve_release(job) + assert time.monotonic() - before < 1 + lib.mr_resolve_release(job) # already released; no stale pointer + time.sleep(0.05) + + # A forked child has no resolver workers from its parent. It must be able + # to start a fresh resolution, and old request IDs must not become valid. + job = lib.mr_resolve_start(b'localhost', 80, 4, 5000, 0) + child = os.fork() + if child == 0: + try: + assert lib.mr_resolve_address(job, 0) == b'' + fresh = lib.mr_resolve_start(b'localhost', 80, 4, 0, 0) + assert fresh > 0 and fresh != job + wait_for(lambda: lib.mr_resolve_done(fresh)) + assert lib.mr_resolve_error(fresh) == 0 + lib.mr_resolve_release(fresh) + os._exit(0) + except BaseException: + os._exit(1) + lib.mr_resolve_release(job) + status = [] + def child_finished(): + pid, value = os.waitpid(child, os.WNOHANG) + if pid: + status.append(value) + return bool(pid) + try: + wait_for(child_finished) + except BaseException: + os.kill(child, 9) + os.waitpid(child, 0) + raise + assert status == [0] + + with socket.socket() as server: + server.bind(('127.0.0.1', 0)) + server.listen() + fd = lib.mr_connect_start(b'127.0.0.1', server.getsockname()[1], b'127.0.0.1', 0) + assert fd >= 0 + try: + assert fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_NONBLOCK + assert fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC + assert select.select([], [fd], [], 5)[1] == [fd] + assert lib.mr_connect_error(fd) == 0 + server.settimeout(5) + accepted, _ = server.accept() + accepted.close() + finally: + os.close(fd) + bad = lib.mr_connect_start(b'127.0.0.1', server.getsockname()[1], b'192.0.2.1', 8000) + assert bad == -errno.EADDRNOTAVAIL + + addresses = lib.mr_ip_addresses().splitlines() + assert addresses + for address in addresses: + raw = bytes.fromhex(address.decode()) + assert len(raw) == raw[0] and raw[1] in (socket.AF_INET, socket.AF_INET6) + print('native socket resolver lifecycle, nonblocking connect, binding and interface addresses: PASS') diff --git a/unittest/tcp_connect_check.py b/unittest/tcp_connect_check.py new file mode 100644 index 0000000..f1daa98 --- /dev/null +++ b/unittest/tcp_connect_check.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python3 +"""Bound connection regression children, including timeout cleanup.""" +import os +from pathlib import Path +import subprocess + +ROOT = Path(__file__).resolve().parent.parent +program = ROOT / 'corpus/503_tcp_connect_deadlines.rb' + + +def run(binary, env): + return subprocess.run([str(binary), str(program)], env=env, + capture_output=True, check=True, timeout=20).stdout + + +expected = run(os.environ.get('REF_RUBY_BIN', 'ruby'), os.environ.copy()) +for mode in ('on', 'off'): + env = os.environ.copy() + env.pop('MERE_RUBY_NO_HASH_INDEX', None) + if mode == 'off': + env['MERE_RUBY_NO_HASH_INDEX'] = '1' + actual = run(ROOT / 'mere-ruby', env) + assert actual == expected, f'TCP regression differs with hash index {mode}' + print(f'TCP deadlines: matches reference, hash index {mode}', flush=True)