Phase 3: Response streaming for callable Rack bodies

ext/iodine/fio.h

@@ -457,7 +457,7 @@ Logging and testing helpers
 #define FIO_LOG____LENGTH_BORDER FIO_LOG_LENGTH_LIMIT
 #endif
 /** The logging level */
-int __attribute__((weak)) FIO_LOG_LEVEL;
+extern int FIO_LOG_LEVEL;
 
 #pragma weak FIO_LOG2STDERR
 void __attribute__((format(printf, 1, 0), weak))

ext/iodine/http.c

@@ -91,6 +91,13 @@ static inline void remove_content_length(http_s *r) {
   fiobj_hash_delete2(r->private_data.out_headers, cl_hash);
 }
 
+static inline void remove_transfer_encoding(http_s *r) {
+  static uint64_t te_hash = 0;
+  if (!te_hash)
+    te_hash = fiobj_hash_string("transfer-encoding", 17);
+  fiobj_hash_delete2(r->private_data.out_headers, te_hash);
+}
+
 static inline void add_content_type(http_s *r) {
   static uint64_t ct_hash = 0;
   if (!ct_hash)
@@ -378,6 +385,64 @@ intptr_t http_uuid(http_s *h) {
   return ((http_fio_protocol_s *)h->private_data.flag)->uuid;
 }
 
+/**
+ * Marks the response as a streaming response: the connection's protocol will
+ * not auto-finalize it when the request callback returns, and the `http_s`
+ * handle remains valid until `http_streaming_end` completes the response.
+ */
+void http_streaming_start(http_s *h) {
+  if (HTTP_INVALID_HANDLE(h))
+    return;
+  remove_content_length(h);
+  remove_transfer_encoding(h);
+  ((http_vtable_s *)h->private_data.vtbl)->http_streaming_start(h);
+}
+
+/**
+ * Completes a streaming response: sends the terminating chunk via
+ * `http_finish` and resumes normal request handling on the connection.
+ *
+ * AFTER THIS FUNCTION IS CALLED, THE `http_s` OBJECT IS NO LONGER VALID.
+ */
+void http_streaming_end(http_s *h) {
+  if (HTTP_INVALID_HANDLE(h))
+    return;
+  ((http_vtable_s *)h->private_data.vtbl)->http_streaming_end(h);
+}
+
+/**
+ * Arms a one-shot wake when the outgoing queue drains or the connection closes.
+ * Re-arm after each blocked write.
+ */
+void http_streaming_arm_wake(http_s *h) {
+  if (HTTP_INVALID_HANDLE(h))
+    return;
+  ((http_vtable_s *)h->private_data.vtbl)->http_streaming_arm_wake(h);
+}
+
+/**
+ * Copies the current streaming response's NUL-terminated wake channel name to
+ * `dest`, which must hold at least `HTTP_WAKE_CHANNEL_MAX` bytes. Returns its
+ * length, or 0 if there's no active streaming response.
+ */
+size_t http_streaming_wake_channel(http_s *h, char dest[HTTP_WAKE_CHANNEL_MAX]) {
+  static const char prefix[] = "iodine:stream:";
+  if (HTTP_INVALID_HANDLE(h) || !dest)
+    return 0;
+
+  http_fio_protocol_s *p = (http_fio_protocol_s *)h->private_data.flag;
+  if (!p->stream_generation)
+    return 0;
+
+  memcpy(dest, prefix, sizeof(prefix) - 1);
+  size_t len = sizeof(prefix) - 1;
+  len += fio_ltoa(dest + len, (int64_t)p->uuid, 16);
+  dest[len++] = ':';
+  len += fio_ltoa(dest + len, (int64_t)p->stream_generation, 16);
+  dest[len] = 0;
+  return len;
+}
+
 /**
  * Sends the response headers and the specified file (the response's body).
  *

ext/iodine/http.h

@@ -230,6 +230,53 @@ int http_stream(http_s *h, void *data, uintptr_t length);
  */
 intptr_t http_uuid(http_s *h);
 
+/**
+ * Marks the response as a streaming response.
+ *
+ * The connection's protocol will not auto-finalize the response when the
+ * request callback returns, and the `http_s` handle remains valid for
+ * repeated `http_stream` calls until `http_streaming_end` completes the
+ * response.
+ *
+ * Any application-supplied `Content-Length` or `Transfer-Encoding` header is
+ * removed: the streaming transport owns the response framing (it adds
+ * `Transfer-Encoding: chunked` on the first write, or `Content-Length: 0`
+ * when the stream closes without writing).
+ */
+void http_streaming_start(http_s *h);
+
+/**
+ * Completes a streaming response: sends the terminating chunk via
+ * `http_finish` and resumes normal request handling on the connection.
+ *
+ * AFTER THIS FUNCTION IS CALLED, THE `http_s` OBJECT IS NO LONGER VALID.
+ */
+void http_streaming_end(http_s *h);
+
+/**
+ * Arms a one-shot wake for the streaming response.
+ *
+ * The protocol publishes "drain" or "close" to the process-local wake channel
+ * when the socket queue drains or the connection closes. Re-arm after each
+ * blocked write.
+ */
+void http_streaming_arm_wake(http_s *h);
+
+/** Upper bound (including the NUL) for a wake channel name: the
+ * "iodine:stream:" prefix plus two hex numbers of up to 20 characters each. */
+#define HTTP_WAKE_CHANNEL_MAX 64
+
+/**
+ * Copies the current streaming response's NUL-terminated wake channel name to
+ * `dest`, which must hold at least `HTTP_WAKE_CHANNEL_MAX` bytes.
+ *
+ * The process-local name stays the same for this response and changes for later
+ * responses on the same keep-alive connection.
+ *
+ * Returns its length, or 0 if there's no active streaming response.
+ */
+size_t http_streaming_wake_channel(http_s *h, char dest[HTTP_WAKE_CHANNEL_MAX]);
+
 /**
  * Sends the response headers and the specified file (the response's body).
  *

ext/iodine/http1.c

@@ -28,6 +28,8 @@ typedef struct http1pr_s {
   uint8_t close;
   uint8_t is_client;
   uint8_t stop;
+  uint8_t streaming;
+  uint8_t stream_wake;
   http_stream_state_e stream_state;
   uint8_t buf[];
 } http1pr_s;
@@ -46,10 +48,18 @@ inline static void h1_reset(http1pr_s *p) { p->header_size = 0; }
 #define http1_pr2handle(pr) (((http1pr_s *)(pr))->request)
 #define handle2pr(h) ((http1pr_s *)h->private_data.flag)
 
+#define HTTP1_STREAM_WAKE_DRAIN "drain"
+#define HTTP1_STREAM_WAKE_CLOSE "close"
+
+static void http1_stream_wake_publish(http1pr_s *p, const char *msg,
+                                      size_t len);
+
 /* cleanup an HTTP/1.1 handler object */
 static inline void http1_after_finish(http_s *h) {
   http1pr_s *p = handle2pr(h);
   p->stop = p->stop & (~1UL);
+  p->streaming = 0;
+  p->stream_wake = 0;
   p->stream_state = HTTP_STREAM_IDLE;
   if (h != &p->request) {
     http_s_destroy(h, 0);
@@ -289,6 +299,40 @@ static int http1_stream(http_s *h, void *data, uintptr_t length) {
   return 0;
 }
 
+/** Marks the in-progress response as streaming: the parser must not
+ * auto-finish it and must not parse further pipelined requests until the
+ * stream completes. */
+static void http1_streaming_start(http_s *h) {
+  http1pr_s *p = handle2pr(h);
+  if (!p->streaming) {
+    if (++p->p.stream_generation == 0)
+      p->p.stream_generation = 1; /* zero means no active stream */
+  }
+  p->streaming = 1;
+  p->stream_wake = 0;
+}
+
+/* Arm one wake notification for the next drain or disconnect. */
+static void http1_streaming_arm_wake(http_s *h) {
+  handle2pr(h)->stream_wake = 1;
+}
+
+/** Completes a streaming response: sends the terminating chunk via
+ * `http_finish`, then re-arms the parser for any buffered pipelined data
+ * that was suspended while the stream was active. */
+static void http1_streaming_end(http_s *h) {
+  http1pr_s *p = handle2pr(h);
+  const intptr_t uuid = p->p.uuid;
+  /* Explicit close wakes a blocked producer before http_finish resets it. */
+  if (p->streaming && p->stream_wake) {
+    p->stream_wake = 0;
+    http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_CLOSE, sizeof(HTTP1_STREAM_WAKE_CLOSE) - 1);
+  }
+  p->streaming = 0;
+  http_finish(h);
+  fio_force_event(uuid, FIO_EVENT_ON_DATA);
+}
+
 /** Push for data - unsupported. */
 static int http1_push_data(http_s *h, void *data, uintptr_t length,
                            FIOBJ mime_type) {
@@ -319,7 +363,8 @@ static void http1_on_pause(http_s *h, http_fio_protocol_s *pr) {
  * called after the resume task had completed.
  */
 static void http1_on_resume(http_s *h, http_fio_protocol_s *pr) {
-  if (!((http1pr_s *)pr)->stop) {
+  http1pr_s *p = (http1pr_s *)pr;
+  if (!p->stop || p->streaming) {
     fio_resume(pr->uuid);
   }
   (void)h;
@@ -595,6 +640,9 @@ struct http_vtable_s HTTP1_VTABLE = {
     .http_send_body = http1_send_body,
     .http_sendfile = http1_sendfile,
     .http_stream = http1_stream,
+    .http_streaming_start = http1_streaming_start,
+    .http_streaming_end = http1_streaming_end,
+    .http_streaming_arm_wake = http1_streaming_arm_wake,
     .http_finish = htt1p_finish,
     .http_push_data = http1_push_data,
     .http_push_file = http1_push_file,
@@ -617,7 +665,7 @@ Parser Callbacks
 static int http1_on_request(http1_parser_s *parser) {
   http1pr_s *p = parser2http(parser);
   http_on_request_handler______internal(&http1_pr2handle(p), p->p.settings);
-  if (p->request.method && !p->stop)
+  if (p->request.method && !p->stop && !p->streaming)
     http_finish(&p->request);
   h1_reset(p);
   return fio_is_closed(p->p.uuid);
@@ -626,7 +674,7 @@ static int http1_on_request(http1_parser_s *parser) {
 static int http1_on_response(http1_parser_s *parser) {
   http1pr_s *p = parser2http(parser);
   http_on_response_handler______internal(&http1_pr2handle(p), p->p.settings);
-  if (p->request.status_str && !p->stop)
+  if (p->request.status_str && !p->stop && !p->streaming)
     http_finish(&p->request);
   h1_reset(p);
   return fio_is_closed(p->p.uuid);
@@ -755,7 +803,7 @@ static inline void http1_consume_data(intptr_t uuid, http1pr_s *p) {
     i = http1_parse(&p->parser, p->buf + (org_len - p->buf_len), p->buf_len);
     p->buf_len -= i;
     --pipeline_limit;
-  } while (i && p->buf_len && pipeline_limit && !p->stop);
+  } while (i && p->buf_len && pipeline_limit && !p->stop && !p->streaming);
 
   if (p->buf_len && org_len != p->buf_len) {
     memmove(p->buf, p->buf + (org_len - p->buf_len), p->buf_len);
@@ -787,7 +835,7 @@ static inline void http1_consume_data(intptr_t uuid, http1pr_s *p) {
 /** called when a data is available, but will not run concurrently */
 static void http1_on_data(intptr_t uuid, fio_protocol_s *protocol) {
   http1pr_s *p = (http1pr_s *)protocol;
-  if (p->stop) {
+  if (p->stop || p->streaming) {
     fio_suspend(uuid);
     return;
   }
@@ -801,21 +849,43 @@ static void http1_on_data(intptr_t uuid, fio_protocol_s *protocol) {
   http1_consume_data(uuid, p);
 }
 
+/* Notify a blocked producer without calling Ruby here. */
+static void http1_stream_wake_publish(http1pr_s *p, const char *msg,
+                                      size_t len) {
+  char channel[HTTP_WAKE_CHANNEL_MAX];
+  size_t channel_len = http_streaming_wake_channel(&p->request, channel);
+  if (!channel_len)
+    return;
+  fio_publish(.engine = FIO_PUBSUB_PROCESS,
+              .channel = {.len = channel_len, .data = channel},
+              .message = {.len = len, .data = (char *)msg});
+}
+
 /** called when the connection was closed, but will not run concurrently */
 static void http1_on_close(intptr_t uuid, fio_protocol_s *protocol) {
+  http1pr_s *p = (http1pr_s *)protocol;
+  /* Wake blocked producers on disconnect. The protocol keeps the original
+   * UUID; the callback UUID may be newer. The generation separates streams. */
+  if (p->streaming && p->stream_wake) {
+    p->stream_wake = 0;
+    http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_CLOSE, sizeof(HTTP1_STREAM_WAKE_CLOSE) - 1);
+  }
   http1_destroy(protocol);
-  (void)uuid;
 }
 
-/** called when the connection was closed, but will not run concurrently */
+/** called when all pending socket data was sent (the queue drained) */
 static void http1_on_ready(intptr_t uuid, fio_protocol_s *protocol) {
   /* resume slow clients from suspension */
   http1pr_s *p = (http1pr_s *)protocol;
   if (p->stop & 4) {
     p->stop ^= 4; /* flip back the bit, so it's zero */
     fio_force_event(uuid, FIO_EVENT_ON_DATA);
   }
-  (void)protocol;
+  /* Wake a blocked producer after the socket queue drains. */
+  if (p->streaming && p->stream_wake) {
+    p->stream_wake = 0;
+    http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_DRAIN, sizeof(HTTP1_STREAM_WAKE_DRAIN) - 1);
+  }
 }
 
 /** called when a data is available for the first time */

ext/iodine/http_internal.h

@@ -46,6 +46,12 @@ struct http_vtable_s {
                              uintptr_t offset);
   /** Should send existing headers and data and prepare for streaming */
   int (*const http_stream)(http_s *h, void *data, uintptr_t length);
+  /** Should mark the response as streaming, preventing auto-finalization */
+  void (*const http_streaming_start)(http_s *h);
+  /** Should complete a streaming response and resume request handling */
+  void (*const http_streaming_end)(http_s *h);
+  /** Arms a one-shot drain or disconnect wake for a blocked stream */
+  void (*const http_streaming_arm_wake)(http_s *h);
   /** Should send existing headers or complete streaming */
   void (*const http_finish)(http_s *h);
   /** Push for data. */
@@ -75,6 +81,7 @@ struct http_fio_protocol_s {
   fio_protocol_s protocol;   /* facil.io protocol */
   intptr_t uuid;             /* socket uuid */
   http_settings_s *settings; /* pointer to HTTP settings */
+  uint64_t stream_generation; /* streaming response generation */
 };
 
 #define http2protocol(h) ((http_fio_protocol_s *)h->private_data.flag)

ext/iodine/iodine_http.c

@@ -589,6 +589,14 @@ static inline int ruby2c_response_send(iodine_http_request_handle_s *handle,
     if (rb_respond_to(body, close_method_id))
       IodineCaller.call(body, close_method_id);
     return 0;
+  } else if (rb_respond_to(body, iodine_call_proc_id)) {
+    // Rage owns producer scheduling. Iodine invokes the callable
+    VALUE stream = IodineRackStream.create(handle->h);
+    if (stream == Qnil)
+      return -1;
+    IodineCaller.call2(body, iodine_call_proc_id, 1, &stream);
+    handle->type = IODINE_HTTP_NONE;
+    return 0;
   }
   return -1;
 }