Merge remote-tracking branch 'origin/master'

* origin/master:
  Restore CARP_MALLOC NULL checks in kqueue and epoll Poll_wait_
  Add NULL check after CARP_MALLOC in Poll_copy
  Guard CARP_MALLOC returns in Poll_wait_ to prevent NULL dereference
  Add CARP_FORCE_POLL override, CARP_MALLOC NULL checks, and poll2 test wrapper
  Implement POSIX poll(2) fallback for non-epoll/kqueue platforms
  Make Poll.wait propagate errors from epoll_wait/kevent
  Fix send_nb_ negative offset UB and send_len_ OOB read
  Add non-blocking I/O support to UnixStream

Author: hellerve

src/poll.carp

@@ -37,7 +37,8 @@ Check `readable`, `writable`, and `error?` to see what happened on `fd`.")
 (register-type Poll)
 
 (doc Poll "is an I/O event multiplexer that monitors multiple file descriptors
-for readiness. Uses kqueue on macOS/BSD and epoll on Linux.
+for readiness. Uses kqueue on macOS/BSD, epoll on Linux, and POSIX poll(2)
+as a fallback on other platforms.
 
 ## Usage
 ```
@@ -77,6 +78,9 @@ for readiness. Uses kqueue on macOS/BSD and epoll on Linux.
   (private wait-)
   (hidden wait-)
   (register wait- (Fn [&Poll Int] (Array PollEvent)) "Poll_wait_")
+  (private wait-failed?)
+  (hidden wait-failed?)
+  (register wait-failed? (Fn [&(Array PollEvent)] Bool) "Poll_wait_failed_")
 
   ; -- public API --
 
@@ -121,7 +125,11 @@ Use `poll-read`, `poll-write`, or `poll-read-write`.")
 
   (doc wait "waits for events, blocking up to `timeout-ms` milliseconds.
 Pass -1 to block indefinitely. Returns `(Result (Array PollEvent) String)`.")
-  (defn wait [poll timeout-ms] (Result.Success (wait- poll timeout-ms)))
+  (defn wait [poll timeout-ms]
+    (let [events (wait- poll timeout-ms)]
+      (if (wait-failed? &events)
+        (Result.Error (System.error-text))
+        (Result.Success events))))
 
   (register close (Fn [Poll] ()))
   (doc close "closes the poll instance.")

src/poll.h

@@ -35,7 +35,10 @@ PollEvent PollEvent_copy(PollEvent* e) {
  * Platform-specific Poll implementation
  * -------------------------------------------------------------------------- */
 
-#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
+#if defined(CARP_FORCE_POLL)
+#define CARP_USE_POLL 1
+#include <poll.h>
+#elif defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
 #define CARP_USE_KQUEUE 1
 #include <sys/event.h>
 #elif defined(__linux__)
@@ -120,7 +123,16 @@ Array Poll_wait_(Poll* p, int timeout_ms) {
 
   if (n > 0) {
     fds = (int*)CARP_MALLOC(n * sizeof(int));
+    if (!fds) {
+      result.len = 0; result.capacity = 0; result.data = NULL;
+      return result;
+    }
     flags = (int*)CARP_MALLOC(n * sizeof(int));
+    if (!flags) {
+      CARP_FREE(fds);
+      result.len = 0; result.capacity = 0; result.data = NULL;
+      return result;
+    }
 
     for (int i = 0; i < n; i++) {
       int fd = (int)events[i].ident;
@@ -147,6 +159,12 @@ Array Poll_wait_(Poll* p, int timeout_ms) {
   result.len = unique;
   result.capacity = unique > 0 ? unique : 1;
   result.data = CARP_MALLOC(result.capacity * sizeof(PollEvent));
+  if (!result.data) {
+    if (fds) CARP_FREE(fds);
+    if (flags) CARP_FREE(flags);
+    result.len = 0; result.capacity = 0;
+    return result;
+  }
 
   for (int i = 0; i < unique; i++) {
     PollEvent* e = &((PollEvent*)result.data)[i];
@@ -216,6 +234,10 @@ Array Poll_wait_(Poll* p, int timeout_ms) {
   result.len = n;
   result.capacity = n > 0 ? n : 1;
   result.data = CARP_MALLOC(result.capacity * sizeof(PollEvent));
+  if (!result.data) {
+    result.len = 0; result.capacity = 0;
+    return result;
+  }
 
   for (int i = 0; i < n; i++) {
     PollEvent* e = &((PollEvent*)result.data)[i];
@@ -233,12 +255,146 @@ void Poll_close(Poll p) {
 
 #endif /* CARP_USE_EPOLL */
 
+#ifdef CARP_USE_POLL
+
+typedef struct {
+  struct pollfd *fds;
+  int count;
+  int capacity;
+} Poll;
+
+Poll Poll_create_() {
+  Poll p;
+  p.capacity = 16;
+  p.count = 0;
+  p.fds = (struct pollfd*)CARP_MALLOC(p.capacity * sizeof(struct pollfd));
+  if (!p.fds) {
+    p.capacity = 0;
+  }
+  return p;
+}
+
+int Poll_fd_(Poll* p) {
+  /* No kernel fd — return 0 if valid, -1 if not */
+  return (p->fds != NULL) ? 0 : -1;
+}
+
+int Poll_add_(Poll* p, int fd, int interest) {
+  if (p->count >= p->capacity) {
+    int new_cap = p->capacity * 2;
+    struct pollfd *new_fds = (struct pollfd*)CARP_MALLOC(new_cap * sizeof(struct pollfd));
+    if (!new_fds) return -1;
+    memcpy(new_fds, p->fds, (size_t)p->count * sizeof(struct pollfd));
+    CARP_FREE(p->fds);
+    p->fds = new_fds;
+    p->capacity = new_cap;
+  }
+  struct pollfd *pfd = &p->fds[p->count];
+  pfd->fd = fd;
+  pfd->events = 0;
+  if (interest & POLL_READ) pfd->events |= POLLIN;
+  if (interest & POLL_WRITE) pfd->events |= POLLOUT;
+  pfd->revents = 0;
+  p->count++;
+  return 0;
+}
+
+int Poll_modify_(Poll* p, int fd, int interest) {
+  for (int i = 0; i < p->count; i++) {
+    if (p->fds[i].fd == fd) {
+      p->fds[i].events = 0;
+      if (interest & POLL_READ) p->fds[i].events |= POLLIN;
+      if (interest & POLL_WRITE) p->fds[i].events |= POLLOUT;
+      return 0;
+    }
+  }
+  errno = ENOENT;
+  return -1;
+}
+
+int Poll_remove_(Poll* p, int fd) {
+  for (int i = 0; i < p->count; i++) {
+    if (p->fds[i].fd == fd) {
+      p->fds[i] = p->fds[p->count - 1];
+      p->count--;
+      return 0;
+    }
+  }
+  errno = ENOENT;
+  return -1;
+}
+
+Array Poll_wait_(Poll* p, int timeout_ms) {
+  Array result;
+  if (!p->fds) {
+    result.len = 0; result.capacity = 0; result.data = NULL;
+    return result;
+  }
+
+  int n = poll(p->fds, (nfds_t)p->count, timeout_ms);
+  if (n < 0) {
+    result.len = 0; result.capacity = 0; result.data = NULL;
+    return result;
+  }
+
+  int ready = 0;
+  for (int i = 0; i < p->count; i++) {
+    if (p->fds[i].revents != 0) ready++;
+  }
+
+  result.len = ready;
+  result.capacity = ready > 0 ? ready : 1;
+  result.data = CARP_MALLOC(result.capacity * sizeof(PollEvent));
+  if (!result.data) {
+    result.len = 0; result.capacity = 0;
+    return result;
+  }
+
+  int j = 0;
+  for (int i = 0; i < p->count && j < ready; i++) {
+    if (p->fds[i].revents == 0) continue;
+    PollEvent* e = &((PollEvent*)result.data)[j];
+    e->fd = p->fds[i].fd;
+    e->readable = (p->fds[i].revents & POLLIN) != 0;
+    e->writable = (p->fds[i].revents & POLLOUT) != 0;
+    e->error = (p->fds[i].revents & (POLLERR | POLLHUP | POLLNVAL)) != 0;
+    j++;
+  }
+
+  result.len = j;
+  return result;
+}
+
+void Poll_close(Poll p) {
+  if (p.fds) CARP_FREE(p.fds);
+}
+
+#endif /* CARP_USE_POLL */
+
+bool Poll_wait_failed_(Array* a) {
+  return a->data == NULL;
+}
+
 Poll Poll_copy(Poll* p) {
   Poll c;
 #ifdef CARP_USE_KQUEUE
   c.kq = p->kq;
 #elif defined(CARP_USE_EPOLL)
   c.epfd = p->epfd;
+#elif defined(CARP_USE_POLL)
+  c.count = p->count;
+  c.capacity = p->capacity;
+  if (p->fds && p->capacity > 0) {
+    c.fds = (struct pollfd*)CARP_MALLOC((size_t)p->capacity * sizeof(struct pollfd));
+    if (!c.fds) {
+      c.count = 0;
+      c.capacity = 0;
+    } else {
+      memcpy(c.fds, p->fds, (size_t)p->count * sizeof(struct pollfd));
+    }
+  } else {
+    c.fds = NULL;
+  }
 #endif
   return c;
 }

src/unix_stream.carp

@@ -67,6 +67,19 @@
     (let [n (send-bytes- stream data)]
       (if (= n -1) (Result.Error (System.error-text)) (Result.Success n))))
 
+  (private send-len-)
+  (hidden send-len-)
+  (register send-len-
+    (Fn [&UnixStream &String Int] Int)
+    "UnixStream_send_MINUS_len_")
+
+  (doc send-len "sends a string with known length (avoids strlen). `len` must be
+between 0 and the string’s length; returns an error otherwise.
+Returns bytes sent or an error.")
+  (defn send-len [stream msg len]
+    (let [n (send-len- stream msg len)]
+      (if (= n -1) (Result.Error (System.error-text)) (Result.Success n))))
+
   (doc read
     "reads up to 4096 bytes from the stream. Returns the data as a string, or an error.
 Returns an empty string on connection close.")
@@ -83,7 +96,11 @@ Returns bytes read (0 = connection closed), or an error.")
       (if (= n -1) (Result.Error (System.error-text)) (Result.Success n))))
 
   (register close (Fn [UnixStream] ()))
-  (doc close "closes the stream, releasing the file descriptor.")
+  (doc close "closes the stream, consuming it.")
+
+  (doc close! "closes the stream by reference, for use when the stream lives
+in a collection. Sets the fd to -1 to prevent double-close.")
+  (register close! (Fn [&UnixStream] ()) "UnixStream_close_MINUS_ref")
 
   (doc shutdown "shuts down the stream. 0 = reads, 1 = writes, 2 = both.")
   (register shutdown- (Fn [&UnixStream Int] ()) "UnixStream_shutdown_")
@@ -98,12 +115,59 @@ Returns bytes read (0 = connection closed), or an error.")
   (doc set-timeout "sets read and write timeouts in seconds.")
   (register set-timeout (Fn [&UnixStream Int] ()))
 
+  (doc set-nonblocking "puts the socket into non-blocking mode. After this
+call, `send-nb` and `read-append-nb` are the appropriate I/O entry points;
+the blocking variants will return `EAGAIN` instead of waiting.")
+  (register set-nonblocking (Fn [&UnixStream] ()))
+
+  (private send-nb-)
+  (hidden send-nb-)
+  (register send-nb-
+    (Fn [&UnixStream &(Array Byte) Int] Int)
+    "UnixStream_send_MINUS_nb_")
+
+  (doc send-nb "non-blocking send. Sends as many bytes as the kernel will
+accept right now from `data`, starting at `offset`.
+
+Returns `(Result Int String)`. The `Int` is the number of bytes actually
+written, which may be `0` if the socket is not currently writable. Re-arm
+write interest on the next event-loop iteration in that case.")
+  (defn send-nb [stream data offset]
+    (let [n (send-nb- stream data offset)]
+      (if (= n -1) (Result.Error (System.error-text)) (Result.Success n))))
+
+  (private read-append-nb-)
+  (hidden read-append-nb-)
+  (register read-append-nb-
+    (Fn [&UnixStream &(Array Byte)] Int)
+    "UnixStream_read_MINUS_append_MINUS_nb_")
+
+  (doc read-blocked "sentinel returned by `read-append-nb` when no data is
+currently available on a non-blocking socket.")
+  (def read-blocked -2)
+
+  (doc read-append-nb "non-blocking append-read. Reads whatever the kernel
+has buffered into `buf`, growing it as needed.
+
+Returns `(Result Int String)`. The `Int` is one of:
+
+- `> 0` bytes appended to `buf`,
+- `0` peer closed cleanly (EOF),
+- `read-blocked` (-2) socket has no data right now, retry on the next
+  readable event.")
+  (defn read-append-nb [stream buf]
+    (let [n (read-append-nb- stream buf)]
+      (if (= n -1) (Result.Error (System.error-text)) (Result.Success n))))
+
   (doc peer-path "returns the path of the remote peer socket.")
   (register peer-path (Fn [&UnixStream] String))
 
   (register copy (Fn [&UnixStream] UnixStream))
   (implements copy UnixStream.copy)
 
+  (register prn (Fn [UnixStream] String) "UnixStream_prn_")
+  (implements prn UnixStream.prn)
+
   (defn poll-fd [s] (fd- s))
   (implements poll-fd UnixStream.poll-fd)
 

src/unix_stream.h

@@ -100,6 +100,49 @@ String UnixStream_peer_MINUS_path(UnixStream* s) {
   return str;
 }
 
+void UnixStream_close_MINUS_ref(UnixStream* s) {
+  if (s->fd >= 0) { close(s->fd); s->fd = -1; }
+}
+
+void UnixStream_set_MINUS_nonblocking(UnixStream* s) {
+  int flags = fcntl(s->fd, F_GETFL, 0);
+  if (flags >= 0) fcntl(s->fd, F_SETFL, flags | O_NONBLOCK);
+}
+
+int UnixStream_send_MINUS_len_(UnixStream* s, String* msg, int len) {
+  if (len < 0 || len > (int)strlen(*msg)) return -1;
+  return (int)send_all(s->fd, *msg, (size_t)len);
+}
+
+int UnixStream_send_MINUS_nb_(UnixStream* s, Array* data, int offset) {
+  if (offset < 0 || offset >= data->len) return 0;
+  ssize_t n = send(s->fd, (char*)data->data + offset,
+                   (size_t)(data->len - offset), 0);
+  if (n >= 0) return (int)n;
+  if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return 0;
+  return -1;
+}
+
+int UnixStream_read_MINUS_append_MINUS_nb_(UnixStream* s, Array* buf) {
+  if ((int)(buf->capacity - buf->len) < SOCK_BUF_SIZE) {
+    int new_cap = (buf->len + SOCK_BUF_SIZE) * 2;
+    buf->data = CARP_REALLOC(buf->data, new_cap);
+    buf->capacity = new_cap;
+  }
+  ssize_t r = read(s->fd, (char*)buf->data + buf->len, SOCK_BUF_SIZE);
+  if (r > 0) { buf->len += (int)r; return (int)r; }
+  if (r == 0) return 0;
+  if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return -2;
+  return -1;
+}
+
+String UnixStream_prn_(UnixStream s) {
+  size_t len = (size_t)snprintf(NULL, 0, "UnixStream(%d)", s.fd);
+  String r = CARP_MALLOC(len + 1);
+  snprintf(r, len + 1, "UnixStream(%d)", s.fd);
+  return r;
+}
+
 UnixStream UnixStream_copy(UnixStream* s) {
   UnixStream c;
   c.fd = s->fd;