From 6fb3ce34b7ee6309356fc3366d5e3724871a98b3 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Tue, 7 Jul 2026 14:23:25 +0530 Subject: [PATCH 01/14] detection seam and fiber execution --- ext/iodine/iodine_http.c | 24 +++++++++ spec/integration/response_streaming_spec.rb | 51 +++++++++++++++++++ spec/support/apps/response_streaming.ru | 14 +++++ .../apps/response_streaming_backpressure.ru | 20 ++++++++ 4 files changed, 109 insertions(+) create mode 100644 spec/integration/response_streaming_spec.rb create mode 100644 spec/support/apps/response_streaming.ru create mode 100644 spec/support/apps/response_streaming_backpressure.ru diff --git a/ext/iodine/iodine_http.c b/ext/iodine/iodine_http.c index 4686b9c4..554a0842 100644 --- a/ext/iodine/iodine_http.c +++ b/ext/iodine/iodine_http.c @@ -548,6 +548,21 @@ static VALUE for_each_body_string(VALUE str, VALUE body_, int argc, (void)blockarg; } +// Runs a streaming body (`body.call(stream)`) inside a dedicated fiber, +// `pair` is [body, stream]. +static VALUE iodine_stream_run_producer(RB_BLOCK_CALL_FUNC_ARGLIST(first, pair)) { + VALUE body = RARRAY_AREF(pair, 0); + VALUE stream = RARRAY_AREF(pair, 1); + IodineCaller.call2(body, iodine_call_proc_id, 1, &stream); + // idempotent auto-close so the response finalizes even if the app didn't. + IodineCaller.call(stream, close_method_id); + return Qnil; + (void)first; + (void)argc; + (void)argv; + (void)blockarg; +} + static inline int ruby2c_response_send(iodine_http_request_handle_s *handle, VALUE rbresponse, VALUE env) { (void)(env); @@ -589,6 +604,15 @@ 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)) { + // Rack streaming body: run body.call(stream) inside a managed fiber + // (pause/resume wiring lands in a follow-up step.) + VALUE stream = IodineRackStream.create(handle->h, Qnil); + VALUE pair = rb_ary_new_from_args(2, body, stream); + VALUE fiber = rb_fiber_new(iodine_stream_run_producer, pair); + rb_fiber_resume(fiber, 0, NULL); + handle->type = IODINE_HTTP_NONE; // fully handled here; nothing left to send + return 0; } return -1; } diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb new file mode 100644 index 00000000..b8268cf0 --- /dev/null +++ b/spec/integration/response_streaming_spec.rb @@ -0,0 +1,51 @@ +require 'spec_helper' +require 'socket' + +# Functional tests for HTTP response streaming: a real Iodine server runs the +# `response_streaming` app and we assert on the wire behavior end to end. +RSpec.describe 'HTTP response streaming', with_app: :response_streaming do + let(:expected) { "chunk-0\nchunk-1\nchunk-2\nchunk-3\nchunk-4\n" } + + # Sends a raw GET and returns [raw_headers_string, chunk_arrival_times]. + def raw_stream_get + times = [] + headers = +"" + Socket.tcp('localhost', server_port, connect_timeout: 1) do |sock| + sock.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + buf = +"" + loop do + begin + data = sock.read_nonblock(4096) + times << Time.now + buf << data + rescue IO::WaitReadable + break unless IO.select([sock], nil, nil, 2) + retry + rescue EOFError + break + end + end + headers = buf.split("\r\n\r\n", 2).first.to_s + end + [headers, times] + end + + it 'responds 200 and reassembles the full streamed body' do + response = http_get("/") + expect(response.status).to eq(200) + expect(response.body.to_s).to eq(expected) + end + + it 'uses chunked transfer encoding, not Content-Length' do + headers, = raw_stream_get + expect(headers.downcase).to match(/transfer-encoding:\s*chunked/) + expect(headers.downcase).not_to include("content-length:") + end + + it 'delivers chunks incrementally rather than buffering the whole response' do + _, times = raw_stream_get + expect(times.length).to be > 1 + # 5 chunks written 0.05s apart -> arrivals must span well beyond a single read + expect(times.last - times.first).to be > 0.1 + end +end diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru new file mode 100644 index 00000000..d7f8c5e0 --- /dev/null +++ b/spec/support/apps/response_streaming.ru @@ -0,0 +1,14 @@ +# Test app for HTTP response streaming. +# The body responds to `call(stream)` (Rack streaming body), so Iodine should +# hand it a RackStream writer and stream each chunk incrementally. +run ->(env) do + body = proc do |stream| + 5.times do |i| + stream.write("chunk-#{i}\n") + sleep 0.05 # a gap so incremental delivery is observable + end + stream.close + end + + [200, {}, body] +end diff --git a/spec/support/apps/response_streaming_backpressure.ru b/spec/support/apps/response_streaming_backpressure.ru new file mode 100644 index 00000000..f7e7ab65 --- /dev/null +++ b/spec/support/apps/response_streaming_backpressure.ru @@ -0,0 +1,20 @@ +# Test app that produces faster than a slow client can read, so the producer is +# pushed past the HIGH watermark and must handle :would_block by yielding. +run ->(env) do + body = proc do |stream| + payload = "x" * 16_000 + total = 128 + total.times do + loop do + case stream.write(payload) + when :ok then break + when :would_block then Fiber.yield + else break + end + end + end + stream.close + end + + [200, {}, body] +end From 11b85a6147d3f926971ded9ed969f5d12fab02d9 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Wed, 5 Aug 2026 01:04:41 +0530 Subject: [PATCH 02/14] spec: verify response streaming with HTTP gem --- spec/integration/response_streaming_spec.rb | 57 +++++++++------------ 1 file changed, 25 insertions(+), 32 deletions(-) diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index b8268cf0..26c1759c 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -1,51 +1,44 @@ require 'spec_helper' -require 'socket' # Functional tests for HTTP response streaming: a real Iodine server runs the -# `response_streaming` app and we assert on the wire behavior end to end. +# `response_streaming` app and the HTTP gem consumes the response incrementally. RSpec.describe 'HTTP response streaming', with_app: :response_streaming do let(:expected) { "chunk-0\nchunk-1\nchunk-2\nchunk-3\nchunk-4\n" } - # Sends a raw GET and returns [raw_headers_string, chunk_arrival_times]. - def raw_stream_get - times = [] - headers = +"" - Socket.tcp('localhost', server_port, connect_timeout: 1) do |sock| - sock.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") - buf = +"" - loop do - begin - data = sock.read_nonblock(4096) - times << Time.now - buf << data - rescue IO::WaitReadable - break unless IO.select([sock], nil, nil, 2) - retry - rescue EOFError - break - end - end - headers = buf.split("\r\n\r\n", 2).first.to_s - end - [headers, times] + def consume_body(response) + body = +"" + response.body.each { |fragment| body << fragment } + body end it 'responds 200 and reassembles the full streamed body' do response = http_get("/") expect(response.status).to eq(200) - expect(response.body.to_s).to eq(expected) + expect(consume_body(response)).to eq(expected) end it 'uses chunked transfer encoding, not Content-Length' do - headers, = raw_stream_get - expect(headers.downcase).to match(/transfer-encoding:\s*chunked/) - expect(headers.downcase).not_to include("content-length:") + response = http_get("/") + expect(response.chunked?).to be(true) + expect(response.headers).not_to include('Content-Length') + expect(consume_body(response)).to eq(expected) end it 'delivers chunks incrementally rather than buffering the whole response' do - _, times = raw_stream_get - expect(times.length).to be > 1 - # 5 chunks written 0.05s apart -> arrivals must span well beyond a single read - expect(times.last - times.first).to be > 0.1 + first_seen = {} + buf = +"" + + response = http_get("/") + response.body.each do |fragment| + observed_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + buf << fragment + 5.times do |i| + first_seen[i] ||= observed_at if buf.include?("chunk-#{i}\n") + end + end + + expect(buf).to eq(expected) + expect(first_seen.keys.sort).to eq([0, 1, 2, 3, 4]) + expect(first_seen[4] - first_seen[0]).to be > 0.1 end end From 32684320d1e57d68fa92386dca149f7c35291b16 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Wed, 5 Aug 2026 14:30:16 +0530 Subject: [PATCH 03/14] fix: run streaming body in the caller's fiber --- ext/iodine/iodine_http.c | 31 ++++++------------- ext/iodine/iodine_rack_stream.c | 15 +++------ ext/iodine/iodine_rack_stream.h | 2 +- spec/integration/response_streaming_spec.rb | 8 +++++ spec/support/apps/response_streaming.ru | 8 +++++ .../apps/response_streaming_backpressure.ru | 20 ------------ 6 files changed, 30 insertions(+), 54 deletions(-) delete mode 100644 spec/support/apps/response_streaming_backpressure.ru diff --git a/ext/iodine/iodine_http.c b/ext/iodine/iodine_http.c index 554a0842..9b9ca96a 100644 --- a/ext/iodine/iodine_http.c +++ b/ext/iodine/iodine_http.c @@ -548,21 +548,6 @@ static VALUE for_each_body_string(VALUE str, VALUE body_, int argc, (void)blockarg; } -// Runs a streaming body (`body.call(stream)`) inside a dedicated fiber, -// `pair` is [body, stream]. -static VALUE iodine_stream_run_producer(RB_BLOCK_CALL_FUNC_ARGLIST(first, pair)) { - VALUE body = RARRAY_AREF(pair, 0); - VALUE stream = RARRAY_AREF(pair, 1); - IodineCaller.call2(body, iodine_call_proc_id, 1, &stream); - // idempotent auto-close so the response finalizes even if the app didn't. - IodineCaller.call(stream, close_method_id); - return Qnil; - (void)first; - (void)argc; - (void)argv; - (void)blockarg; -} - static inline int ruby2c_response_send(iodine_http_request_handle_s *handle, VALUE rbresponse, VALUE env) { (void)(env); @@ -605,13 +590,15 @@ static inline int ruby2c_response_send(iodine_http_request_handle_s *handle, IodineCaller.call(body, close_method_id); return 0; } else if (rb_respond_to(body, iodine_call_proc_id)) { - // Rack streaming body: run body.call(stream) inside a managed fiber - // (pause/resume wiring lands in a follow-up step.) - VALUE stream = IodineRackStream.create(handle->h, Qnil); - VALUE pair = rb_ary_new_from_args(2, body, stream); - VALUE fiber = rb_fiber_new(iodine_stream_run_producer, pair); - rb_fiber_resume(fiber, 0, NULL); - handle->type = IODINE_HTTP_NONE; // fully handled here; nothing left to send + // 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); + // Transitional safety: until the pause/token lifecycle lands, close before + // the raw HTTP handle can escape this request callback. + IodineCaller.call(stream, close_method_id); + handle->type = IODINE_HTTP_NONE; return 0; } return -1; diff --git a/ext/iodine/iodine_rack_stream.c b/ext/iodine/iodine_rack_stream.c index 18540a7f..49872f69 100644 --- a/ext/iodine/iodine_rack_stream.c +++ b/ext/iodine/iodine_rack_stream.c @@ -6,7 +6,7 @@ typedef enum { IODINE_STREAM_IDLE = 0, /* created, nothing written yet */ IODINE_STREAM_HEADERS_SENT, /* first write flushed the response headers */ IODINE_STREAM_STREAMING, /* chunks flowing */ - IODINE_STREAM_BLOCKED, /* paused on backpressure, fiber yielded */ + IODINE_STREAM_BLOCKED, /* caller must wait for readiness and retry */ IODINE_STREAM_CLOSING, /* close requested, finishing safely */ IODINE_STREAM_CLOSED, /* terminal: completed */ IODINE_STREAM_ERROR, /* terminal: write failure / disconnect */ @@ -17,7 +17,6 @@ typedef struct { http_s *h; intptr_t uuid; /* socket uuid, for fio_pending / fio_is_valid */ iodine_stream_state_e state; - VALUE fiber; /* producer fiber; also held as an ivar so the GC marks it */ size_t high_watermark; /* pause threshold */ size_t low_watermark; /* resume threshold */ int blocked; /* backpressure flag */ @@ -38,7 +37,6 @@ Core data / helpers static VALUE rRackStream; static ID ctx_var_id; /* ivar holding the stream_ctx_t pointer */ -static ID fiber_var_id; /* ivar holding the producer fiber */ static ID iodine_new_func_id; /* write() return values (cached symbols) */ @@ -64,7 +62,6 @@ static void stream_teardown(VALUE stream) { ctx->freed = 1; ctx->state = IODINE_STREAM_CLOSED; set_ctx(stream, NULL); - rb_ivar_set(stream, fiber_var_id, Qnil); free(ctx); } @@ -103,8 +100,8 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { return SYM_error; } - /* 6. backpressure -> yield/retry (write would overflow HARD, or past HIGH) - * TODO: register http_pause here and resume from http1_on_ready at LOW. */ + /* 6. backpressure -> caller-owned wait/retry (would overflow HARD or past HIGH) + * TODO(phase-3): publish readiness from http1_on_ready at LOW. */ if (pending + packets_needed >= IODINE_STREAM_HARD_MAX || pending >= ctx->high_watermark) { ctx->blocked = 1; @@ -159,7 +156,7 @@ static VALUE rack_stream_is_closed(VALUE self) { C land API ***************************************************************************** */ -static VALUE new_rack_stream(http_s *h, VALUE fiber) { +static VALUE new_rack_stream(http_s *h) { stream_ctx_t *ctx = malloc(sizeof(*ctx)); if (!ctx) return Qnil; @@ -167,7 +164,6 @@ static VALUE new_rack_stream(http_s *h, VALUE fiber) { .h = h, .uuid = http_uuid(h), /* stable connection id; cached for the write path */ .state = IODINE_STREAM_IDLE, - .fiber = fiber, .high_watermark = IODINE_STREAM_HIGH_WATERMARK, .low_watermark = IODINE_STREAM_LOW_WATERMARK, .blocked = 0, @@ -176,8 +172,6 @@ static VALUE new_rack_stream(http_s *h, VALUE fiber) { VALUE stream = rb_funcall2(rRackStream, iodine_new_func_id, 0, NULL); set_ctx(stream, ctx); - /* hold the fiber as an ivar so Ruby's GC keeps it alive while blocked. */ - rb_ivar_set(stream, fiber_var_id, fiber); return stream; } @@ -191,7 +185,6 @@ static void init_rack_stream(void) { rRackStream = rb_define_class_under(IodineBaseModule, "RackStream", rb_cObject); ctx_var_id = rb_intern("stream_ctx"); - fiber_var_id = rb_intern("stream_fiber"); iodine_new_func_id = rb_intern("new"); SYM_ok = ID2SYM(rb_intern("ok")); diff --git a/ext/iodine/iodine_rack_stream.h b/ext/iodine/iodine_rack_stream.h index a66bd005..90d01d2b 100644 --- a/ext/iodine/iodine_rack_stream.h +++ b/ext/iodine/iodine_rack_stream.h @@ -6,7 +6,7 @@ #include "http.h" extern struct IodineRackStream { - VALUE (*create)(http_s *h, VALUE fiber); + VALUE (*create)(http_s *h); void (*close)(VALUE stream); void (*init)(void); diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index 26c1759c..9f3122d8 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -17,6 +17,14 @@ def consume_body(response) expect(consume_body(response)).to eq(expected) end + it 'invokes the callable body in the current Fiber' do + response = http_get("/") + expect(consume_body(response)).to eq(expected) + + state = http_get('/stream-state') + expect(consume_body(state)).to eq('same_fiber=true') + end + it 'uses chunked transfer encoding, not Content-Length' do response = http_get("/") expect(response.chunked?).to be(true) diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru index d7f8c5e0..1ecd076d 100644 --- a/spec/support/apps/response_streaming.ru +++ b/spec/support/apps/response_streaming.ru @@ -1,8 +1,16 @@ # Test app for HTTP response streaming. # The body responds to `call(stream)` (Rack streaming body), so Iodine should # hand it a RackStream writer and stream each chunk incrementally. +same_fiber = nil + run ->(env) do + if env['PATH_INFO'] == '/stream-state' + next [200, {}, ["same_fiber=#{same_fiber}"]] + end + + request_fiber = Fiber.current body = proc do |stream| + same_fiber = Fiber.current.equal?(request_fiber) 5.times do |i| stream.write("chunk-#{i}\n") sleep 0.05 # a gap so incremental delivery is observable diff --git a/spec/support/apps/response_streaming_backpressure.ru b/spec/support/apps/response_streaming_backpressure.ru deleted file mode 100644 index f7e7ab65..00000000 --- a/spec/support/apps/response_streaming_backpressure.ru +++ /dev/null @@ -1,20 +0,0 @@ -# Test app that produces faster than a slow client can read, so the producer is -# pushed past the HIGH watermark and must handle :would_block by yielding. -run ->(env) do - body = proc do |stream| - payload = "x" * 16_000 - total = 128 - total.times do - loop do - case stream.write(payload) - when :ok then break - when :would_block then Fiber.yield - else break - end - end - end - stream.close - end - - [200, {}, body] -end From 6eede4d06fa5cd76803f8140dc6bdb2815b25eb0 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Wed, 5 Aug 2026 15:55:36 +0530 Subject: [PATCH 04/14] spec: prove streaming continues after the callable body returns --- spec/integration/response_streaming_spec.rb | 17 +++++++++++++++++ spec/support/apps/response_streaming.ru | 19 +++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index 9f3122d8..cc06816c 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -49,4 +49,21 @@ def consume_body(response) expect(first_seen.keys.sort).to eq([0, 1, 2, 3, 4]) expect(first_seen[4] - first_seen[0]).to be > 0.1 end + + it 'keeps streaming after the callable returns' do + body = +"" + released = false + + response = http_get('/async') + response.body.each do |fragment| + body << fragment + next if released || !body.include?("marker-a\n") + + released = true + expect(http_get('/release').status).to eq(204) + end + + expect(released).to be(true) + expect(body).to eq("marker-a\nmarker-b\n") + end end diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru index 1ecd076d..d6b1355f 100644 --- a/spec/support/apps/response_streaming.ru +++ b/spec/support/apps/response_streaming.ru @@ -2,12 +2,31 @@ # The body responds to `call(stream)` (Rack streaming body), so Iodine should # hand it a RackStream writer and stream each chunk incrementally. same_fiber = nil +release_channel = "response-streaming-release" run ->(env) do if env['PATH_INFO'] == '/stream-state' next [200, {}, ["same_fiber=#{same_fiber}"]] end + if env['PATH_INFO'] == '/release' + Iodine.publish(release_channel, '', Iodine::PubSub::PROCESS) + next [204, {}, []] + end + + if env['PATH_INFO'] == '/async' + body = proc do |stream| + Iodine.subscribe(release_channel) do + stream.write("marker-b\n") + stream.close + Iodine.defer { Iodine.unsubscribe(release_channel) } + end + stream.write("marker-a\n") + end + + next [200, {}, body] + end + request_fiber = Fiber.current body = proc do |stream| same_fiber = Fiber.current.equal?(request_fiber) From ecc74b0f6ca580d973b00809b7e1d75a29ee9394 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Wed, 5 Aug 2026 22:43:51 +0530 Subject: [PATCH 05/14] fix: preserve async Rack streaming lifecycle --- ext/iodine/http.c | 34 ++++ ext/iodine/http.h | 16 ++ ext/iodine/http1.c | 27 ++- ext/iodine/iodine_http.c | 4 +- ext/iodine/iodine_rack_stream.c | 289 ++++++++++++++++++++++++++++---- ext/iodine/iodine_rack_stream.h | 2 +- 6 files changed, 330 insertions(+), 42 deletions(-) diff --git a/ext/iodine/http.c b/ext/iodine/http.c index 64257c02..4ca13448 100644 --- a/ext/iodine/http.c +++ b/ext/iodine/http.c @@ -831,6 +831,40 @@ void http_resume(http_pause_handle_s *http, void (*task)(http_s *h), .fallback = http_resume_fallback_wrapper); } +/** + * Attempts to resume a paused request synchronously. + */ +int http_resume_try(http_pause_handle_s *http, + void (*task)(http_s *h, void *udata), void *udata, + void (*fallback)(void *udata)) { + if (!http) + return -1; + + fio_protocol_s *protocol = + fio_protocol_try_lock(http->uuid, FIO_PR_LOCK_TASK); + if (!protocol) { + if (errno != EBADF) + return 1; + if (fallback) + fallback(http->udata); + fio_free(http); + return -1; + } + + http_fio_protocol_s *p = (http_fio_protocol_s *)protocol; + http_s *h = http->h; + h->udata = http->udata; + h->fiber = http->fiber; + h->subscription = http->subscription; + http_vtable_s *vtbl = (http_vtable_s *)h->private_data.vtbl; + if (task) + task(h, udata); + vtbl->http_on_resume(h, p); + fio_free(http); + fio_protocol_unlock(protocol, FIO_PR_LOCK_TASK); + return 0; +} + /** * Hijacks the socket away from the HTTP protocol and away from facil.io. */ diff --git a/ext/iodine/http.h b/ext/iodine/http.h index 90bc680d..e58a1f1a 100644 --- a/ext/iodine/http.h +++ b/ext/iodine/http.h @@ -345,6 +345,22 @@ void http_pause(http_s *h, void (*task)(http_pause_handle_s *http)); void http_resume(http_pause_handle_s *http, void (*task)(http_s *h), void (*fallback)(void *udata)); +/** + * Attempts to resume a paused request synchronously without waiting for the + * connection task lock. + * + * Returns 0 after consuming `http` and running `task`, 1 when the connection + * lock is busy (the handle remains valid for a later retry), and -1 when the + * connection was closed (the handle is consumed and `fallback` is called). + * + * As with `http_resume`, `task` MUST send, finish, or pause the response before + * returning. `udata` is passed only to `task`; `fallback` receives the paused + * response's stored `udata`. + */ +int http_resume_try(http_pause_handle_s *http, + void (*task)(http_s *h, void *udata), void *udata, + void (*fallback)(void *udata)); + /** Returns the `udata` associated with the paused opaque handle */ void *http_paused_udata_get(http_pause_handle_s *http); diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index b8f5a70f..2b16f3e3 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -28,6 +28,8 @@ typedef struct http1pr_s { uint8_t close; uint8_t is_client; uint8_t stop; + uint8_t paused; + uint8_t pause_counted; http_stream_state_e stream_state; uint8_t buf[]; } http1pr_s; @@ -50,6 +52,7 @@ inline static void h1_reset(http1pr_s *p) { p->header_size = 0; } static inline void http1_after_finish(http_s *h) { http1pr_s *p = handle2pr(h); p->stop = p->stop & (~1UL); + p->paused = 0; p->stream_state = HTTP_STREAM_IDLE; if (h != &p->request) { http_s_destroy(h, 0); @@ -310,8 +313,12 @@ static int http1_push_file(http_s *h, FIOBJ filename, FIOBJ mime_type) { * Called befor a pause task, */ static void http1_on_pause(http_s *h, http_fio_protocol_s *pr) { - ((http1pr_s *)pr)->stop = 1; - fio_pause(pr->uuid); + http1pr_s *p = (http1pr_s *)pr; + if (!p->pause_counted) { + fio_pause(pr->uuid); + p->pause_counted = 1; + } + p->paused = 1; (void)h; } @@ -319,7 +326,9 @@ 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->paused && p->pause_counted) { + p->pause_counted = 0; fio_resume(pr->uuid); } (void)h; @@ -617,7 +626,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->paused) http_finish(&p->request); h1_reset(p); return fio_is_closed(p->p.uuid); @@ -626,7 +635,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->paused) http_finish(&p->request); h1_reset(p); return fio_is_closed(p->p.uuid); @@ -755,7 +764,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->paused); if (p->buf_len && org_len != p->buf_len) { memmove(p->buf, p->buf + (org_len - p->buf_len), p->buf_len); @@ -787,7 +796,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->paused) { fio_suspend(uuid); return; } @@ -882,6 +891,10 @@ fio_protocol_s *http1_new(uintptr_t uuid, http_settings_s *settings, /** Manually destroys the HTTP1 protocol object. */ void http1_destroy(fio_protocol_s *pr) { http1pr_s *p = (http1pr_s *)pr; + if (p->pause_counted) { + p->pause_counted = 0; + fio_resume(p->p.uuid); + } http1_pr2handle(p).status = 0; http_s_destroy(&http1_pr2handle(p), 0); // FIO_LOG_DEBUG("Deallocating HTTP/1.1 protocol %p(%d)=>%p", (void diff --git a/ext/iodine/iodine_http.c b/ext/iodine/iodine_http.c index 9b9ca96a..ce847d2c 100644 --- a/ext/iodine/iodine_http.c +++ b/ext/iodine/iodine_http.c @@ -595,9 +595,7 @@ static inline int ruby2c_response_send(iodine_http_request_handle_s *handle, if (stream == Qnil) return -1; IodineCaller.call2(body, iodine_call_proc_id, 1, &stream); - // Transitional safety: until the pause/token lifecycle lands, close before - // the raw HTTP handle can escape this request callback. - IodineCaller.call(stream, close_method_id); + IodineRackStream.pause(stream); handle->type = IODINE_HTTP_NONE; return 0; } diff --git a/ext/iodine/iodine_rack_stream.c b/ext/iodine/iodine_rack_stream.c index 49872f69..23829577 100644 --- a/ext/iodine/iodine_rack_stream.c +++ b/ext/iodine/iodine_rack_stream.c @@ -12,17 +12,35 @@ typedef enum { IODINE_STREAM_ERROR, /* terminal: write failure / disconnect */ } iodine_stream_state_e; +typedef enum { + IODINE_STREAM_TRANSPORT_ACTIVE = 0, + IODINE_STREAM_TRANSPORT_PAUSING, + IODINE_STREAM_TRANSPORT_PAUSED, + IODINE_STREAM_TRANSPORT_RESUMING, + IODINE_STREAM_TRANSPORT_TERMINAL, +} iodine_stream_transport_state_e; + typedef struct { - /* TODO(phase-3): don't persist across http_pause/http_resume (invalidates h). */ - http_s *h; + http_s *h; /* valid only while transport_state is ACTIVE */ + http_pause_handle_s *pause_handle; intptr_t uuid; /* socket uuid, for fio_pending / fio_is_valid */ iodine_stream_state_e state; + iodine_stream_transport_state_e transport_state; + fio_lock_i lock; size_t high_watermark; /* pause threshold */ size_t low_watermark; /* resume threshold */ int blocked; /* backpressure flag */ + int close_requested; int freed; /* terminal guard: teardown runs exactly once */ } stream_ctx_t; +typedef struct { + stream_ctx_t *ctx; + const char *data; + size_t length; + VALUE result; +} stream_write_args_s; + /* Watermarks are queued-packet counts ; each write is sliced * into CHUNK_SIZE packets, so 1 packet ~= 16KB. */ #define IODINE_STREAM_CHUNK_SIZE (16 * 1024) @@ -46,6 +64,11 @@ static VALUE SYM_disconnected; static VALUE SYM_would_block; static VALUE SYM_error; +static void stream_on_paused(http_pause_handle_s *pause_handle); +static void stream_finish_resumed(http_s *h); +static void stream_finish_fallback(void *udata); +static VALUE rack_stream_close(VALUE self); + #define set_ctx(object, ctx) \ rb_ivar_set((object), ctx_var_id, ULL2NUM((uintptr_t)(ctx))) @@ -54,17 +77,105 @@ inline static stream_ctx_t *get_ctx(VALUE obj) { return (stream_ctx_t *)NUM2ULL(i); } -/* Frees the context exactly once and detaches it from the Ruby object. */ -static void stream_teardown(VALUE stream) { - stream_ctx_t *ctx = get_ctx(stream); - if (!ctx || ctx->freed) +/* Frees native state after the final handle or pause token is consumed. */ +static void stream_ctx_free(stream_ctx_t *ctx) { + if (!ctx) return; + + fio_lock(&ctx->lock); + if (ctx->freed) { + fio_unlock(&ctx->lock); + return; + } ctx->freed = 1; ctx->state = IODINE_STREAM_CLOSED; - set_ctx(stream, NULL); + ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; + fio_unlock(&ctx->lock); free(ctx); } +static void stream_finish_resumed(http_s *h) { + stream_ctx_t *ctx = h->udata; + http_finish(h); + stream_ctx_free(ctx); +} + +static void stream_finish_fallback(void *udata) { + stream_ctx_free(udata); +} + +static void stream_resume_finish(http_pause_handle_s *pause_handle) { + http_resume(pause_handle, stream_finish_resumed, stream_finish_fallback); +} + +static void stream_on_paused(http_pause_handle_s *pause_handle) { + stream_ctx_t *ctx = http_paused_udata_get(pause_handle); + int finish = 0; + + fio_lock(&ctx->lock); + if (ctx->close_requested) { + ctx->transport_state = IODINE_STREAM_TRANSPORT_RESUMING; + finish = 1; + } else { + ctx->pause_handle = pause_handle; + ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSED; + } + fio_unlock(&ctx->lock); + + if (finish) + stream_resume_finish(pause_handle); +} + +/* Sends one complete application chunk through a currently valid HTTP handle. */ +static VALUE stream_write_with_handle(stream_ctx_t *ctx, http_s *h, + const char *data, size_t length) { + const char *p = data; + size_t remaining = length; + + do { + size_t n = + remaining < IODINE_STREAM_CHUNK_SIZE ? remaining : IODINE_STREAM_CHUNK_SIZE; + if (http_stream(h, (void *)p, n) < 0) { + ctx->state = IODINE_STREAM_ERROR; + return SYM_error; + } + p += n; + remaining -= n; + } while (remaining); + + if (ctx->state < IODINE_STREAM_CLOSING) { + ctx->state = IODINE_STREAM_STREAMING; + ctx->blocked = 0; + } + return SYM_ok; +} + +static void stream_write_resumed(http_s *h, void *udata) { + stream_write_args_s *args = udata; + stream_ctx_t *ctx = args->ctx; + int close_requested = 0; + + args->result = stream_write_with_handle(ctx, h, args->data, args->length); + + fio_lock(&ctx->lock); + close_requested = ctx->close_requested; + if (args->result == SYM_ok && !close_requested) + ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSING; + else + ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; + fio_unlock(&ctx->lock); + + if (args->result == SYM_ok && !close_requested) { + h->udata = ctx; + http_pause(h, stream_on_paused); + } else { + if (http_uuid(h) != -1) + http_finish(h); + if (close_requested) + stream_ctx_free(ctx); + } +} + /* ***************************************************************************** Ruby API ***************************************************************************** */ @@ -81,7 +192,7 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { /* 2. socket disconnected -> disconnected */ if (!fio_is_valid(ctx->uuid)) { - ctx->state = IODINE_STREAM_ERROR; + rack_stream_close(self); return SYM_disconnected; } @@ -108,24 +219,65 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { ctx->state = IODINE_STREAM_BLOCKED; return SYM_would_block; } - - /* 7. send in <= CHUNK_SIZE slices; empty chunk runs once to flush headers. */ - const char *p = RSTRING_PTR(data); - size_t remaining = RSTRING_LEN(data); - do { - size_t n = - remaining < IODINE_STREAM_CHUNK_SIZE ? remaining : IODINE_STREAM_CHUNK_SIZE; - if (http_stream(ctx->h, (void *)p, n) < 0) { - ctx->state = IODINE_STREAM_ERROR; - return SYM_error; - } - p += n; - remaining -= n; - } while (remaining); - if (ctx->state < IODINE_STREAM_STREAMING) - ctx->state = IODINE_STREAM_STREAMING; /* first write flushed the headers */ - return SYM_ok; + /* 7. send through the active handle, or try to consume the paused handle. + * A busy/missing pause token accepts no bytes and is safe to retry. */ + http_s *h = NULL; + http_pause_handle_s *pause_handle = NULL; + + fio_lock(&ctx->lock); + if (ctx->transport_state == IODINE_STREAM_TRANSPORT_ACTIVE) { + h = ctx->h; + } else if (ctx->transport_state == IODINE_STREAM_TRANSPORT_PAUSED) { + pause_handle = ctx->pause_handle; + ctx->pause_handle = NULL; + ctx->transport_state = IODINE_STREAM_TRANSPORT_RESUMING; + } + fio_unlock(&ctx->lock); + + if (h) + return stream_write_with_handle(ctx, h, RSTRING_PTR(data), RSTRING_LEN(data)); + + if (!pause_handle) { + ctx->blocked = 1; + ctx->state = IODINE_STREAM_BLOCKED; + return SYM_would_block; + } + + stream_write_args_s args = { + .ctx = ctx, + .data = RSTRING_PTR(data), + .length = RSTRING_LEN(data), + .result = SYM_error, + }; + int resume_result = + http_resume_try(pause_handle, stream_write_resumed, &args, NULL); + + if (resume_result > 0) { + fio_lock(&ctx->lock); + ctx->pause_handle = pause_handle; + ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSED; + fio_unlock(&ctx->lock); + ctx->blocked = 1; + ctx->state = IODINE_STREAM_BLOCKED; + return SYM_would_block; + } + + if (resume_result < 0) { + fio_lock(&ctx->lock); + ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; + fio_unlock(&ctx->lock); + ctx->state = IODINE_STREAM_ERROR; + set_ctx(self, NULL); + stream_ctx_free(ctx); + return SYM_disconnected; + } + + if (args.result != SYM_ok) { + set_ctx(self, NULL); + stream_ctx_free(ctx); + } + return args.result; } /* Closes the stream. Idempotent in every state. Sends the terminating @@ -136,11 +288,47 @@ static VALUE rack_stream_close(VALUE self) { if (!ctx || ctx->freed) return Qnil; /* already closed -> no-op */ - if (ctx->state < IODINE_STREAM_CLOSING && fio_is_valid(ctx->uuid)) { - ctx->state = IODINE_STREAM_CLOSING; - http_finish(ctx->h); /* invalidates the http_s handle */ + http_s *h = NULL; + http_pause_handle_s *pause_handle = NULL; + int free_now = 0; + + /* Detach immediately so repeated Ruby close calls are idempotent. Native + * state remains alive until any outstanding pause token is consumed. */ + set_ctx(self, NULL); + + fio_lock(&ctx->lock); + ctx->close_requested = 1; + ctx->state = IODINE_STREAM_CLOSING; + switch (ctx->transport_state) { + case IODINE_STREAM_TRANSPORT_ACTIVE: + h = ctx->h; + ctx->h = NULL; + ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; + free_now = 1; + break; + case IODINE_STREAM_TRANSPORT_PAUSED: + pause_handle = ctx->pause_handle; + ctx->pause_handle = NULL; + ctx->transport_state = IODINE_STREAM_TRANSPORT_RESUMING; + break; + case IODINE_STREAM_TRANSPORT_TERMINAL: + free_now = 1; + break; + case IODINE_STREAM_TRANSPORT_PAUSING: + case IODINE_STREAM_TRANSPORT_RESUMING: + break; + } + fio_unlock(&ctx->lock); + + if (h) { + if (fio_is_valid(ctx->uuid) && http_uuid(h) != -1) + http_finish(h); + stream_ctx_free(ctx); + } else if (pause_handle) { + stream_resume_finish(pause_handle); + } else if (free_now) { + stream_ctx_free(ctx); } - stream_teardown(self); return Qnil; } @@ -162,11 +350,15 @@ static VALUE new_rack_stream(http_s *h) { return Qnil; *ctx = (stream_ctx_t){ .h = h, + .pause_handle = NULL, .uuid = http_uuid(h), /* stable connection id; cached for the write path */ .state = IODINE_STREAM_IDLE, + .transport_state = IODINE_STREAM_TRANSPORT_ACTIVE, + .lock = FIO_LOCK_INIT, .high_watermark = IODINE_STREAM_HIGH_WATERMARK, .low_watermark = IODINE_STREAM_LOW_WATERMARK, .blocked = 0, + .close_requested = 0, .freed = 0, }; @@ -175,7 +367,42 @@ static VALUE new_rack_stream(http_s *h) { return stream; } -static void close_rack_stream(VALUE stream) { stream_teardown(stream); } +static void pause_rack_stream(VALUE stream) { + stream_ctx_t *ctx = get_ctx(stream); + http_s *h = NULL; + int terminal = 0; + + if (!ctx) + return; + + fio_lock(&ctx->lock); + if (!ctx->close_requested && + ctx->transport_state == IODINE_STREAM_TRANSPORT_ACTIVE) { + h = ctx->h; + ctx->h = NULL; + if (!h || ctx->state >= IODINE_STREAM_CLOSED || http_uuid(h) == -1) { + ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; + terminal = 1; + } else { + ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSING; + } + } + fio_unlock(&ctx->lock); + + if (!h) + return; + + if (terminal) { + set_ctx(stream, NULL); + if (fio_is_valid(ctx->uuid) && http_uuid(h) != -1) + http_finish(h); + stream_ctx_free(ctx); + return; + } + + h->udata = ctx; + http_pause(h, stream_on_paused); +} /* ***************************************************************************** Initialization @@ -200,6 +427,6 @@ static void init_rack_stream(void) { struct IodineRackStream IodineRackStream = { .create = new_rack_stream, - .close = close_rack_stream, + .pause = pause_rack_stream, .init = init_rack_stream, }; diff --git a/ext/iodine/iodine_rack_stream.h b/ext/iodine/iodine_rack_stream.h index 90d01d2b..7f828368 100644 --- a/ext/iodine/iodine_rack_stream.h +++ b/ext/iodine/iodine_rack_stream.h @@ -7,7 +7,7 @@ extern struct IodineRackStream { VALUE (*create)(http_s *h); - void (*close)(VALUE stream); + void (*pause)(VALUE stream); void (*init)(void); } IodineRackStream; From ab75f5020d880a4d77d8189a5d360db14c4da3c7 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Fri, 7 Aug 2026 02:03:37 +0530 Subject: [PATCH 06/14] implement http_streaming_start/end streaming lifecycle --- ext/iodine/http.c | 57 ++-- ext/iodine/http.h | 34 +-- ext/iodine/http1.c | 49 ++-- ext/iodine/http_internal.h | 4 + ext/iodine/iodine_http.c | 1 - ext/iodine/iodine_rack_stream.c | 303 +++----------------- ext/iodine/iodine_rack_stream.h | 1 - spec/integration/response_streaming_spec.rb | 9 + spec/support/apps/response_streaming.ru | 14 + 9 files changed, 132 insertions(+), 340 deletions(-) diff --git a/ext/iodine/http.c b/ext/iodine/http.c index 4ca13448..576b7807 100644 --- a/ext/iodine/http.c +++ b/ext/iodine/http.c @@ -378,6 +378,29 @@ 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; + ((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); +} + /** * Sends the response headers and the specified file (the response's body). * @@ -831,40 +854,6 @@ void http_resume(http_pause_handle_s *http, void (*task)(http_s *h), .fallback = http_resume_fallback_wrapper); } -/** - * Attempts to resume a paused request synchronously. - */ -int http_resume_try(http_pause_handle_s *http, - void (*task)(http_s *h, void *udata), void *udata, - void (*fallback)(void *udata)) { - if (!http) - return -1; - - fio_protocol_s *protocol = - fio_protocol_try_lock(http->uuid, FIO_PR_LOCK_TASK); - if (!protocol) { - if (errno != EBADF) - return 1; - if (fallback) - fallback(http->udata); - fio_free(http); - return -1; - } - - http_fio_protocol_s *p = (http_fio_protocol_s *)protocol; - http_s *h = http->h; - h->udata = http->udata; - h->fiber = http->fiber; - h->subscription = http->subscription; - http_vtable_s *vtbl = (http_vtable_s *)h->private_data.vtbl; - if (task) - task(h, udata); - vtbl->http_on_resume(h, p); - fio_free(http); - fio_protocol_unlock(protocol, FIO_PR_LOCK_TASK); - return 0; -} - /** * Hijacks the socket away from the HTTP protocol and away from facil.io. */ diff --git a/ext/iodine/http.h b/ext/iodine/http.h index e58a1f1a..37278bdb 100644 --- a/ext/iodine/http.h +++ b/ext/iodine/http.h @@ -230,6 +230,24 @@ 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. + */ +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); + /** * Sends the response headers and the specified file (the response's body). * @@ -345,22 +363,6 @@ void http_pause(http_s *h, void (*task)(http_pause_handle_s *http)); void http_resume(http_pause_handle_s *http, void (*task)(http_s *h), void (*fallback)(void *udata)); -/** - * Attempts to resume a paused request synchronously without waiting for the - * connection task lock. - * - * Returns 0 after consuming `http` and running `task`, 1 when the connection - * lock is busy (the handle remains valid for a later retry), and -1 when the - * connection was closed (the handle is consumed and `fallback` is called). - * - * As with `http_resume`, `task` MUST send, finish, or pause the response before - * returning. `udata` is passed only to `task`; `fallback` receives the paused - * response's stored `udata`. - */ -int http_resume_try(http_pause_handle_s *http, - void (*task)(http_s *h, void *udata), void *udata, - void (*fallback)(void *udata)); - /** Returns the `udata` associated with the paused opaque handle */ void *http_paused_udata_get(http_pause_handle_s *http); diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index 2b16f3e3..6daf4c3b 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -28,8 +28,7 @@ typedef struct http1pr_s { uint8_t close; uint8_t is_client; uint8_t stop; - uint8_t paused; - uint8_t pause_counted; + uint8_t streaming; http_stream_state_e stream_state; uint8_t buf[]; } http1pr_s; @@ -52,7 +51,7 @@ inline static void h1_reset(http1pr_s *p) { p->header_size = 0; } static inline void http1_after_finish(http_s *h) { http1pr_s *p = handle2pr(h); p->stop = p->stop & (~1UL); - p->paused = 0; + p->streaming = 0; p->stream_state = HTTP_STREAM_IDLE; if (h != &p->request) { http_s_destroy(h, 0); @@ -292,6 +291,24 @@ 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) { + handle2pr(h)->streaming = 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; + 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) { @@ -313,12 +330,8 @@ static int http1_push_file(http_s *h, FIOBJ filename, FIOBJ mime_type) { * Called befor a pause task, */ static void http1_on_pause(http_s *h, http_fio_protocol_s *pr) { - http1pr_s *p = (http1pr_s *)pr; - if (!p->pause_counted) { - fio_pause(pr->uuid); - p->pause_counted = 1; - } - p->paused = 1; + ((http1pr_s *)pr)->stop = 1; + fio_pause(pr->uuid); (void)h; } @@ -326,9 +339,7 @@ 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) { - http1pr_s *p = (http1pr_s *)pr; - if (!p->paused && p->pause_counted) { - p->pause_counted = 0; + if (!((http1pr_s *)pr)->stop) { fio_resume(pr->uuid); } (void)h; @@ -604,6 +615,8 @@ 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_finish = htt1p_finish, .http_push_data = http1_push_data, .http_push_file = http1_push_file, @@ -626,7 +639,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 && !p->paused) + if (p->request.method && !p->stop && !p->streaming) http_finish(&p->request); h1_reset(p); return fio_is_closed(p->p.uuid); @@ -635,7 +648,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 && !p->paused) + if (p->request.status_str && !p->stop && !p->streaming) http_finish(&p->request); h1_reset(p); return fio_is_closed(p->p.uuid); @@ -764,7 +777,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 && !p->paused); + } 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); @@ -796,7 +809,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 || p->paused) { + if (p->stop || p->streaming) { fio_suspend(uuid); return; } @@ -891,10 +904,6 @@ fio_protocol_s *http1_new(uintptr_t uuid, http_settings_s *settings, /** Manually destroys the HTTP1 protocol object. */ void http1_destroy(fio_protocol_s *pr) { http1pr_s *p = (http1pr_s *)pr; - if (p->pause_counted) { - p->pause_counted = 0; - fio_resume(p->p.uuid); - } http1_pr2handle(p).status = 0; http_s_destroy(&http1_pr2handle(p), 0); // FIO_LOG_DEBUG("Deallocating HTTP/1.1 protocol %p(%d)=>%p", (void diff --git a/ext/iodine/http_internal.h b/ext/iodine/http_internal.h index 53182977..6ca766ab 100644 --- a/ext/iodine/http_internal.h +++ b/ext/iodine/http_internal.h @@ -46,6 +46,10 @@ 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); /** Should send existing headers or complete streaming */ void (*const http_finish)(http_s *h); /** Push for data. */ diff --git a/ext/iodine/iodine_http.c b/ext/iodine/iodine_http.c index ce847d2c..e674569c 100644 --- a/ext/iodine/iodine_http.c +++ b/ext/iodine/iodine_http.c @@ -595,7 +595,6 @@ static inline int ruby2c_response_send(iodine_http_request_handle_s *handle, if (stream == Qnil) return -1; IodineCaller.call2(body, iodine_call_proc_id, 1, &stream); - IodineRackStream.pause(stream); handle->type = IODINE_HTTP_NONE; return 0; } diff --git a/ext/iodine/iodine_rack_stream.c b/ext/iodine/iodine_rack_stream.c index 23829577..ef17dda4 100644 --- a/ext/iodine/iodine_rack_stream.c +++ b/ext/iodine/iodine_rack_stream.c @@ -12,35 +12,13 @@ typedef enum { IODINE_STREAM_ERROR, /* terminal: write failure / disconnect */ } iodine_stream_state_e; -typedef enum { - IODINE_STREAM_TRANSPORT_ACTIVE = 0, - IODINE_STREAM_TRANSPORT_PAUSING, - IODINE_STREAM_TRANSPORT_PAUSED, - IODINE_STREAM_TRANSPORT_RESUMING, - IODINE_STREAM_TRANSPORT_TERMINAL, -} iodine_stream_transport_state_e; - typedef struct { - http_s *h; /* valid only while transport_state is ACTIVE */ - http_pause_handle_s *pause_handle; + http_s *h; /* stays valid while the response is in streaming mode */ intptr_t uuid; /* socket uuid, for fio_pending / fio_is_valid */ iodine_stream_state_e state; - iodine_stream_transport_state_e transport_state; - fio_lock_i lock; - size_t high_watermark; /* pause threshold */ - size_t low_watermark; /* resume threshold */ - int blocked; /* backpressure flag */ - int close_requested; int freed; /* terminal guard: teardown runs exactly once */ } stream_ctx_t; -typedef struct { - stream_ctx_t *ctx; - const char *data; - size_t length; - VALUE result; -} stream_write_args_s; - /* Watermarks are queued-packet counts ; each write is sliced * into CHUNK_SIZE packets, so 1 packet ~= 16KB. */ #define IODINE_STREAM_CHUNK_SIZE (16 * 1024) @@ -64,11 +42,6 @@ static VALUE SYM_disconnected; static VALUE SYM_would_block; static VALUE SYM_error; -static void stream_on_paused(http_pause_handle_s *pause_handle); -static void stream_finish_resumed(http_s *h); -static void stream_finish_fallback(void *udata); -static VALUE rack_stream_close(VALUE self); - #define set_ctx(object, ctx) \ rb_ivar_set((object), ctx_var_id, ULL2NUM((uintptr_t)(ctx))) @@ -77,105 +50,17 @@ inline static stream_ctx_t *get_ctx(VALUE obj) { return (stream_ctx_t *)NUM2ULL(i); } -/* Frees native state after the final handle or pause token is consumed. */ -static void stream_ctx_free(stream_ctx_t *ctx) { - if (!ctx) - return; - - fio_lock(&ctx->lock); - if (ctx->freed) { - fio_unlock(&ctx->lock); +/* Frees the context exactly once and detaches it from the Ruby object. */ +static void stream_teardown(VALUE stream) { + stream_ctx_t *ctx = get_ctx(stream); + if (!ctx || ctx->freed) return; - } ctx->freed = 1; ctx->state = IODINE_STREAM_CLOSED; - ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; - fio_unlock(&ctx->lock); + set_ctx(stream, NULL); free(ctx); } -static void stream_finish_resumed(http_s *h) { - stream_ctx_t *ctx = h->udata; - http_finish(h); - stream_ctx_free(ctx); -} - -static void stream_finish_fallback(void *udata) { - stream_ctx_free(udata); -} - -static void stream_resume_finish(http_pause_handle_s *pause_handle) { - http_resume(pause_handle, stream_finish_resumed, stream_finish_fallback); -} - -static void stream_on_paused(http_pause_handle_s *pause_handle) { - stream_ctx_t *ctx = http_paused_udata_get(pause_handle); - int finish = 0; - - fio_lock(&ctx->lock); - if (ctx->close_requested) { - ctx->transport_state = IODINE_STREAM_TRANSPORT_RESUMING; - finish = 1; - } else { - ctx->pause_handle = pause_handle; - ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSED; - } - fio_unlock(&ctx->lock); - - if (finish) - stream_resume_finish(pause_handle); -} - -/* Sends one complete application chunk through a currently valid HTTP handle. */ -static VALUE stream_write_with_handle(stream_ctx_t *ctx, http_s *h, - const char *data, size_t length) { - const char *p = data; - size_t remaining = length; - - do { - size_t n = - remaining < IODINE_STREAM_CHUNK_SIZE ? remaining : IODINE_STREAM_CHUNK_SIZE; - if (http_stream(h, (void *)p, n) < 0) { - ctx->state = IODINE_STREAM_ERROR; - return SYM_error; - } - p += n; - remaining -= n; - } while (remaining); - - if (ctx->state < IODINE_STREAM_CLOSING) { - ctx->state = IODINE_STREAM_STREAMING; - ctx->blocked = 0; - } - return SYM_ok; -} - -static void stream_write_resumed(http_s *h, void *udata) { - stream_write_args_s *args = udata; - stream_ctx_t *ctx = args->ctx; - int close_requested = 0; - - args->result = stream_write_with_handle(ctx, h, args->data, args->length); - - fio_lock(&ctx->lock); - close_requested = ctx->close_requested; - if (args->result == SYM_ok && !close_requested) - ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSING; - else - ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; - fio_unlock(&ctx->lock); - - if (args->result == SYM_ok && !close_requested) { - h->udata = ctx; - http_pause(h, stream_on_paused); - } else { - if (http_uuid(h) != -1) - http_finish(h); - if (close_requested) - stream_ctx_free(ctx); - } -} - /* ***************************************************************************** Ruby API ***************************************************************************** */ @@ -192,7 +77,7 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { /* 2. socket disconnected -> disconnected */ if (!fio_is_valid(ctx->uuid)) { - rack_stream_close(self); + ctx->state = IODINE_STREAM_ERROR; return SYM_disconnected; } @@ -214,121 +99,44 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { /* 6. backpressure -> caller-owned wait/retry (would overflow HARD or past HIGH) * TODO(phase-3): publish readiness from http1_on_ready at LOW. */ if (pending + packets_needed >= IODINE_STREAM_HARD_MAX || - pending >= ctx->high_watermark) { - ctx->blocked = 1; + pending >= IODINE_STREAM_HIGH_WATERMARK) { ctx->state = IODINE_STREAM_BLOCKED; return SYM_would_block; } - /* 7. send through the active handle, or try to consume the paused handle. - * A busy/missing pause token accepts no bytes and is safe to retry. */ - http_s *h = NULL; - http_pause_handle_s *pause_handle = NULL; - - fio_lock(&ctx->lock); - if (ctx->transport_state == IODINE_STREAM_TRANSPORT_ACTIVE) { - h = ctx->h; - } else if (ctx->transport_state == IODINE_STREAM_TRANSPORT_PAUSED) { - pause_handle = ctx->pause_handle; - ctx->pause_handle = NULL; - ctx->transport_state = IODINE_STREAM_TRANSPORT_RESUMING; - } - fio_unlock(&ctx->lock); - - if (h) - return stream_write_with_handle(ctx, h, RSTRING_PTR(data), RSTRING_LEN(data)); - - if (!pause_handle) { - ctx->blocked = 1; - ctx->state = IODINE_STREAM_BLOCKED; - return SYM_would_block; - } - - stream_write_args_s args = { - .ctx = ctx, - .data = RSTRING_PTR(data), - .length = RSTRING_LEN(data), - .result = SYM_error, - }; - int resume_result = - http_resume_try(pause_handle, stream_write_resumed, &args, NULL); - - if (resume_result > 0) { - fio_lock(&ctx->lock); - ctx->pause_handle = pause_handle; - ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSED; - fio_unlock(&ctx->lock); - ctx->blocked = 1; - ctx->state = IODINE_STREAM_BLOCKED; - return SYM_would_block; - } - - if (resume_result < 0) { - fio_lock(&ctx->lock); - ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; - fio_unlock(&ctx->lock); - ctx->state = IODINE_STREAM_ERROR; - set_ctx(self, NULL); - stream_ctx_free(ctx); - return SYM_disconnected; - } + /* 7. send in <= CHUNK_SIZE slices; empty chunk runs once to flush headers. */ + const char *p = RSTRING_PTR(data); + size_t remaining = RSTRING_LEN(data); + do { + size_t n = + remaining < IODINE_STREAM_CHUNK_SIZE ? remaining : IODINE_STREAM_CHUNK_SIZE; + if (http_stream(ctx->h, (void *)p, n) < 0) { + ctx->state = IODINE_STREAM_ERROR; + return SYM_error; + } + p += n; + remaining -= n; + } while (remaining); - if (args.result != SYM_ok) { - set_ctx(self, NULL); - stream_ctx_free(ctx); - } - return args.result; + if (ctx->state < IODINE_STREAM_STREAMING) + ctx->state = IODINE_STREAM_STREAMING; /* first write flushed the headers */ + return SYM_ok; } /* Closes the stream. Idempotent in every state. Sends the terminating - * zero-length chunk via http_finish exactly once when the connection is still - * alive, then frees the context. */ + * zero-length chunk via http_streaming_end exactly once when the connection is + * still alive, then frees the context. */ static VALUE rack_stream_close(VALUE self) { stream_ctx_t *ctx = get_ctx(self); if (!ctx || ctx->freed) return Qnil; /* already closed -> no-op */ - http_s *h = NULL; - http_pause_handle_s *pause_handle = NULL; - int free_now = 0; - - /* Detach immediately so repeated Ruby close calls are idempotent. Native - * state remains alive until any outstanding pause token is consumed. */ - set_ctx(self, NULL); - - fio_lock(&ctx->lock); - ctx->close_requested = 1; - ctx->state = IODINE_STREAM_CLOSING; - switch (ctx->transport_state) { - case IODINE_STREAM_TRANSPORT_ACTIVE: - h = ctx->h; - ctx->h = NULL; - ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; - free_now = 1; - break; - case IODINE_STREAM_TRANSPORT_PAUSED: - pause_handle = ctx->pause_handle; - ctx->pause_handle = NULL; - ctx->transport_state = IODINE_STREAM_TRANSPORT_RESUMING; - break; - case IODINE_STREAM_TRANSPORT_TERMINAL: - free_now = 1; - break; - case IODINE_STREAM_TRANSPORT_PAUSING: - case IODINE_STREAM_TRANSPORT_RESUMING: - break; - } - fio_unlock(&ctx->lock); - - if (h) { - if (fio_is_valid(ctx->uuid) && http_uuid(h) != -1) - http_finish(h); - stream_ctx_free(ctx); - } else if (pause_handle) { - stream_resume_finish(pause_handle); - } else if (free_now) { - stream_ctx_free(ctx); + /* End the stream whenever the connection is still alive */ + if (fio_is_valid(ctx->uuid)) { + ctx->state = IODINE_STREAM_CLOSING; + http_streaming_end(ctx->h); /* invalidates the http_s handle */ } + stream_teardown(self); return Qnil; } @@ -350,60 +158,20 @@ static VALUE new_rack_stream(http_s *h) { return Qnil; *ctx = (stream_ctx_t){ .h = h, - .pause_handle = NULL, .uuid = http_uuid(h), /* stable connection id; cached for the write path */ .state = IODINE_STREAM_IDLE, - .transport_state = IODINE_STREAM_TRANSPORT_ACTIVE, - .lock = FIO_LOCK_INIT, - .high_watermark = IODINE_STREAM_HIGH_WATERMARK, - .low_watermark = IODINE_STREAM_LOW_WATERMARK, - .blocked = 0, - .close_requested = 0, .freed = 0, }; + /* the response now outlives the request callback; only an explicit close + * (http_streaming_end) finishes it. */ + http_streaming_start(h); + VALUE stream = rb_funcall2(rRackStream, iodine_new_func_id, 0, NULL); set_ctx(stream, ctx); return stream; } -static void pause_rack_stream(VALUE stream) { - stream_ctx_t *ctx = get_ctx(stream); - http_s *h = NULL; - int terminal = 0; - - if (!ctx) - return; - - fio_lock(&ctx->lock); - if (!ctx->close_requested && - ctx->transport_state == IODINE_STREAM_TRANSPORT_ACTIVE) { - h = ctx->h; - ctx->h = NULL; - if (!h || ctx->state >= IODINE_STREAM_CLOSED || http_uuid(h) == -1) { - ctx->transport_state = IODINE_STREAM_TRANSPORT_TERMINAL; - terminal = 1; - } else { - ctx->transport_state = IODINE_STREAM_TRANSPORT_PAUSING; - } - } - fio_unlock(&ctx->lock); - - if (!h) - return; - - if (terminal) { - set_ctx(stream, NULL); - if (fio_is_valid(ctx->uuid) && http_uuid(h) != -1) - http_finish(h); - stream_ctx_free(ctx); - return; - } - - h->udata = ctx; - http_pause(h, stream_on_paused); -} - /* ***************************************************************************** Initialization ***************************************************************************** */ @@ -427,6 +195,5 @@ static void init_rack_stream(void) { struct IodineRackStream IodineRackStream = { .create = new_rack_stream, - .pause = pause_rack_stream, .init = init_rack_stream, }; diff --git a/ext/iodine/iodine_rack_stream.h b/ext/iodine/iodine_rack_stream.h index 7f828368..d652f6ad 100644 --- a/ext/iodine/iodine_rack_stream.h +++ b/ext/iodine/iodine_rack_stream.h @@ -7,7 +7,6 @@ extern struct IodineRackStream { VALUE (*create)(http_s *h); - void (*pause)(VALUE stream); void (*init)(void); } IodineRackStream; diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index cc06816c..1dc98ee4 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -50,6 +50,15 @@ def consume_body(response) expect(first_seen[4] - first_seen[0]).to be > 0.1 end + it 'completes the response cleanly after an oversized write fails the stream' do + response = http_get('/oversized') + expect(response.status).to eq(200) + expect(consume_body(response)).to eq("") + + result = http_get('/oversized-result') + expect(consume_body(result)).to eq('result=error') + end + it 'keeps streaming after the callable returns' do body = +"" released = false diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru index d6b1355f..3e081c4f 100644 --- a/spec/support/apps/response_streaming.ru +++ b/spec/support/apps/response_streaming.ru @@ -2,6 +2,7 @@ # The body responds to `call(stream)` (Rack streaming body), so Iodine should # hand it a RackStream writer and stream each chunk incrementally. same_fiber = nil +oversized_result = nil release_channel = "response-streaming-release" run ->(env) do @@ -14,6 +15,19 @@ run ->(env) do next [204, {}, []] end + if env['PATH_INFO'] == '/oversized' + body = proc do |stream| + oversized_result = stream.write("x" * (2 * 1024 * 1024)) + stream.close + end + + next [200, {}, body] + end + + if env['PATH_INFO'] == '/oversized-result' + next [200, {}, ["result=#{oversized_result}"]] + end + if env['PATH_INFO'] == '/async' body = proc do |stream| Iodine.subscribe(release_channel) do From b34ffe7d57aeafb1c431663b80609c0709d2d054 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Fri, 7 Aug 2026 02:30:50 +0530 Subject: [PATCH 07/14] fix: release pause count when resumed request starts streaming --- ext/iodine/http1.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index 6daf4c3b..a28ec7be 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -339,7 +339,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; From dff7a46c98b059416daa5c68d428aded0f423046 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Sat, 8 Aug 2026 01:18:56 +0530 Subject: [PATCH 08/14] fix: strip the content-length --- ext/iodine/http.c | 9 ++++ ext/iodine/http.h | 5 ++ spec/integration/response_streaming_spec.rb | 50 +++++++++++++++++++ spec/support/apps/response_streaming.ru | 53 +++++++++++++++++++++ 4 files changed, 117 insertions(+) diff --git a/ext/iodine/http.c b/ext/iodine/http.c index 576b7807..e12a7bf7 100644 --- a/ext/iodine/http.c +++ b/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) @@ -386,6 +393,8 @@ intptr_t http_uuid(http_s *h) { 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); } diff --git a/ext/iodine/http.h b/ext/iodine/http.h index 37278bdb..1c3b95d8 100644 --- a/ext/iodine/http.h +++ b/ext/iodine/http.h @@ -237,6 +237,11 @@ intptr_t http_uuid(http_s *h); * 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); diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index 1dc98ee4..afb45326 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -50,6 +50,56 @@ def consume_body(response) expect(first_seen[4] - first_seen[0]).to be > 0.1 end + it 'strips a conflicting application-supplied Content-Length' do + response = http_get('/conflicting-length') + expect(response.status).to eq(200) + expect(response.chunked?).to be(true) + expect(response.headers).not_to include('Content-Length') + expect(consume_body(response)).to eq('hello world') + end + + it 'strips an application-supplied Transfer-Encoding when the stream closes without writing' do + response = http_get('/te-no-write') + expect(response.status).to eq(200) + expect(response.chunked?).to be(false) + expect(response.headers.get('Transfer-Encoding')).to be_empty + expect(response.headers['Content-Length']).to eq('0') + expect(consume_body(response)).to eq("") + end + + it 'strips application-supplied framing headers when an oversized write fails the stream' do + response = http_get('/framing-oversized') + expect(response.status).to eq(200) + expect(response.chunked?).to be(false) + expect(response.headers.get('Transfer-Encoding')).to be_empty + expect(response.headers['Content-Length']).to eq('0') + expect(consume_body(response)).to eq("") + end + + it 'sends exactly one Transfer-Encoding: chunked on a successful write' do + response = http_get('/te-write') + expect(response.status).to eq(200) + expect(response.headers.get('Transfer-Encoding')).to eq(['chunked']) + expect(response.headers.get('Content-Length')).to be_empty + expect(consume_body(response)).to eq('hello') + end + + it 'completes a stream closed without writing as a normal empty response' do + response = http_get('/no-write') + expect(response.status).to eq(200) + expect(response.chunked?).to be(false) + expect(response.headers['Content-Length']).to eq('0') + expect(consume_body(response)).to eq("") + end + + it 'starts chunked framing on an explicit empty first write' do + response = http_get('/empty-write') + expect(response.status).to eq(200) + expect(response.chunked?).to be(true) + expect(response.headers).not_to include('Content-Length') + expect(consume_body(response)).to eq("") + end + it 'completes the response cleanly after an oversized write fails the stream' do response = http_get('/oversized') expect(response.status).to eq(200) diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru index 3e081c4f..b99bf0e8 100644 --- a/spec/support/apps/response_streaming.ru +++ b/spec/support/apps/response_streaming.ru @@ -15,6 +15,59 @@ run ->(env) do next [204, {}, []] end + if env['PATH_INFO'] == '/conflicting-length' + body = proc do |stream| + stream.write("hello ") + stream.write("world") + stream.close + end + + next [200, { 'Content-Length' => '999' }, body] + end + + if env['PATH_INFO'] == '/no-write' + body = proc do |stream| + stream.close + end + + next [200, { 'Content-Length' => '5' }, body] + end + + if env['PATH_INFO'] == '/te-no-write' + body = proc do |stream| + stream.close + end + + next [200, { 'Transfer-Encoding' => 'chunked' }, body] + end + + if env['PATH_INFO'] == '/te-write' + body = proc do |stream| + stream.write("hello") + stream.close + end + + next [200, { 'Transfer-Encoding' => 'chunked' }, body] + end + + if env['PATH_INFO'] == '/framing-oversized' + body = proc do |stream| + stream.write("x" * (2 * 1024 * 1024)) + stream.close + end + + next [200, { 'Transfer-Encoding' => 'chunked', 'Content-Length' => '999' }, body] + end + + if env['PATH_INFO'] == '/empty-write' + body = proc do |stream| + stream.write("") + stream.close + end + + next [200, {}, body] + end + if env['PATH_INFO'] == '/oversized' body = proc do |stream| oversized_result = stream.write("x" * (2 * 1024 * 1024)) From 0d7b08314a949a4563c9ef1e71ef861d5a620dd1 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Sun, 9 Aug 2026 14:31:36 +0530 Subject: [PATCH 09/14] implement wake bridge for blocked streaming producers --- ext/iodine/http.c | 37 ++++++ ext/iodine/http.h | 20 ++++ ext/iodine/http1.c | 51 +++++++- ext/iodine/http_internal.h | 3 + ext/iodine/iodine_rack_stream.c | 25 +++- spec/integration/response_streaming_spec.rb | 126 ++++++++++++++++++++ spec/support/apps/response_streaming.ru | 99 +++++++++++++++ spec/unit/rack_stream_spec.rb | 6 +- 8 files changed, 357 insertions(+), 10 deletions(-) diff --git a/ext/iodine/http.c b/ext/iodine/http.c index e12a7bf7..27e5a8e6 100644 --- a/ext/iodine/http.c +++ b/ext/iodine/http.c @@ -410,6 +410,43 @@ void http_streaming_end(http_s *h) { ((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`. Returns its length, or 0 if `limit` is too small. + */ +size_t http_streaming_wake_channel(http_s *h, char *dest, size_t limit) { + static const char prefix[] = "iodine:stream:"; + char channel[64]; + 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(channel, prefix, sizeof(prefix) - 1); + size_t len = sizeof(prefix) - 1; + len += fio_ltoa(channel + len, (int64_t)p->uuid, 16); + channel[len++] = ':'; + len += fio_ltoa(channel + len, (int64_t)p->stream_generation, 16); + + if (limit <= len) + return 0; + memcpy(dest, channel, len); + dest[len] = 0; + return len; +} + /** * Sends the response headers and the specified file (the response's body). * diff --git a/ext/iodine/http.h b/ext/iodine/http.h index 1c3b95d8..c5e22da2 100644 --- a/ext/iodine/http.h +++ b/ext/iodine/http.h @@ -253,6 +253,26 @@ void http_streaming_start(http_s *h); */ 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); + +/** + * Copies the current streaming response's NUL-terminated wake channel name to + * `dest`. + * + * 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 `limit` is too small. A 64-byte buffer is enough. + */ +size_t http_streaming_wake_channel(http_s *h, char *dest, size_t limit); + /** * Sends the response headers and the specified file (the response's body). * diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index a28ec7be..b537de26 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -29,6 +29,7 @@ typedef struct http1pr_s { 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; @@ -47,11 +48,15 @@ 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) +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); @@ -295,7 +300,16 @@ static int http1_stream(http_s *h, void *data, uintptr_t length) { * auto-finish it and must not parse further pipelined requests until the * stream completes. */ static void http1_streaming_start(http_s *h) { - handle2pr(h)->streaming = 1; + http1pr_s *p = handle2pr(h); + if (!p->streaming && !(++p->p.stream_generation)) + ++p->p.stream_generation; /* 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 @@ -304,6 +318,11 @@ static void http1_streaming_start(http_s *h) { 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, "close", 5); + } p->streaming = 0; http_finish(h); fio_force_event(uuid, FIO_EVENT_ON_DATA); @@ -618,6 +637,7 @@ struct http_vtable_s HTTP1_VTABLE = { .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, @@ -824,13 +844,32 @@ 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[64]; + size_t channel_len = + http_streaming_wake_channel(&p->request, channel, sizeof(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, "close", 5); + } 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; @@ -838,7 +877,11 @@ static void http1_on_ready(intptr_t uuid, fio_protocol_s *protocol) { 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, "drain", 5); + } } /** called when a data is available for the first time */ diff --git a/ext/iodine/http_internal.h b/ext/iodine/http_internal.h index 6ca766ab..5d4a2fca 100644 --- a/ext/iodine/http_internal.h +++ b/ext/iodine/http_internal.h @@ -50,6 +50,8 @@ struct http_vtable_s { 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. */ @@ -79,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) diff --git a/ext/iodine/iodine_rack_stream.c b/ext/iodine/iodine_rack_stream.c index ef17dda4..4e105802 100644 --- a/ext/iodine/iodine_rack_stream.c +++ b/ext/iodine/iodine_rack_stream.c @@ -15,6 +15,8 @@ typedef enum { typedef struct { http_s *h; /* stays valid while the response is in streaming mode */ intptr_t uuid; /* socket uuid, for fio_pending / fio_is_valid */ + char wake_channel[64]; /* fixed wake channel for this response */ + size_t wake_channel_len; iodine_stream_state_e state; int freed; /* terminal guard: teardown runs exactly once */ } stream_ctx_t; @@ -96,11 +98,12 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { return SYM_error; } - /* 6. backpressure -> caller-owned wait/retry (would overflow HARD or past HIGH) - * TODO(phase-3): publish readiness from http1_on_ready at LOW. */ + /* 6. The caller waits and retries this chunk. A blocked write arms one wake + * for the next drain or disconnect; another blocked retry re-arms it. */ if (pending + packets_needed >= IODINE_STREAM_HARD_MAX || pending >= IODINE_STREAM_HIGH_WATERMARK) { ctx->state = IODINE_STREAM_BLOCKED; + http_streaming_arm_wake(ctx->h); return SYM_would_block; } @@ -118,11 +121,22 @@ static VALUE rack_stream_write(VALUE self, VALUE data) { remaining -= n; } while (remaining); - if (ctx->state < IODINE_STREAM_STREAMING) - ctx->state = IODINE_STREAM_STREAMING; /* first write flushed the headers */ + /* The first write flushes headers; any successful write clears BLOCKED. */ + ctx->state = IODINE_STREAM_STREAMING; return SYM_ok; } +/* Process-local channel for "drain" and "close" after :would_block. Ruby owns + * Fiber resumption; a retry re-arms the wake if it still blocks. */ +static VALUE rack_stream_wake_channel(VALUE self) { + stream_ctx_t *ctx = get_ctx(self); + if (!ctx || ctx->state >= IODINE_STREAM_CLOSED) + return Qnil; + if (!ctx->wake_channel_len) + return Qnil; + return rb_str_new(ctx->wake_channel, (long)ctx->wake_channel_len); +} + /* Closes the stream. Idempotent in every state. Sends the terminating * zero-length chunk via http_streaming_end exactly once when the connection is * still alive, then frees the context. */ @@ -166,6 +180,8 @@ static VALUE new_rack_stream(http_s *h) { /* the response now outlives the request callback; only an explicit close * (http_streaming_end) finishes it. */ http_streaming_start(h); + ctx->wake_channel_len = http_streaming_wake_channel( + h, ctx->wake_channel, sizeof(ctx->wake_channel)); VALUE stream = rb_funcall2(rRackStream, iodine_new_func_id, 0, NULL); set_ctx(stream, ctx); @@ -191,6 +207,7 @@ static void init_rack_stream(void) { rb_define_method(rRackStream, "write", rack_stream_write, 1); rb_define_method(rRackStream, "close", rack_stream_close, 0); rb_define_method(rRackStream, "closed?", rack_stream_is_closed, 0); + rb_define_method(rRackStream, "wake_channel", rack_stream_wake_channel, 0); } struct IodineRackStream IodineRackStream = { diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index afb45326..ee6c4881 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -1,4 +1,5 @@ require 'spec_helper' +require 'timeout' # Functional tests for HTTP response streaming: a real Iodine server runs the # `response_streaming` app and the HTTP gem consumes the response incrementally. @@ -11,6 +12,64 @@ def consume_body(response) body end + def backpressure_result(path = '/backpressure-result') + consume_body(http_get(path)) + end + + def wait_for_backpressure(deadline: 10, result_path: '/backpressure-result') + start = Process.clock_gettime(Process::CLOCK_MONOTONIC) + loop do + result = backpressure_result(result_path) + return result if yield(result) + if Process.clock_gettime(Process::CLOCK_MONOTONIC) - start > deadline + raise "timed out waiting for backpressure state, last: #{result}" + end + sleep 0.05 + end + end + + def read_chunked_response(socket) + status = socket.gets("\r\n") + raise EOFError, 'connection closed before response status' unless status + + headers = {} + while (line = socket.gets("\r\n")) && line != "\r\n" + name, value = line.delete_suffix("\r\n").split(':', 2) + headers[name.downcase] = value&.strip + end + raise EOFError, 'connection closed before response headers completed' unless line + unless headers['transfer-encoding'] == 'chunked' + raise "expected chunked response, got #{headers.inspect}" + end + + body = +"" + loop do + size_line = socket.gets("\r\n") + raise EOFError, 'connection closed before chunk size' unless size_line + + size = size_line.split(';', 2).first.to_i(16) + if size.zero? + loop do + trailer = socket.gets("\r\n") + unless trailer + raise EOFError, 'connection closed before chunk trailers completed' + end + break if trailer == "\r\n" + end + break + end + + chunk = socket.read(size) + unless chunk&.bytesize == size + raise EOFError, 'connection closed inside response chunk' + end + raise 'invalid chunk terminator' unless socket.read(2) == "\r\n" + body << chunk + end + + [status, body] + end + it 'responds 200 and reassembles the full streamed body' do response = http_get("/") expect(response.status).to eq(200) @@ -109,6 +168,73 @@ def consume_body(response) expect(consume_body(result)).to eq('result=error') end + it 'wakes a blocked producer when the outgoing queue drains' do + response = http_get('/backpressure') + + wait_for_backpressure { |r| r =~ /would_blocks=[1-9]/ } + + expect(consume_body(response)).to eq('x' * (256 * 16_384)) + + result = wait_for_backpressure { |r| r.include?('result=completed') } + expect(result).to match(/result=completed sent=256 would_blocks=[1-9]\d* wakes=[1-9]\d*/) + end + + it 'wakes a blocked producer when another callback closes the stream' do + socket = Socket.tcp('localhost', server_port) + socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 16_384) + socket.write("GET /backpressure-close HTTP/1.1\r\nHost: localhost\r\n\r\n") + + result = wait_for_backpressure(result_path: '/backpressure-close-result') do |r| + r.include?('result=closed') && r.include?('unsubscribed=true') + end + expect(result).to match( + %r{\Aresult=closed\s+sent=\d+\s+would_blocks=1\s+ + wakes=1\s+last_wake=close\s+close_scheduled=true\s+ + closed_externally=true\s+finished=true\s+unsubscribed=true\z}x + ) + + status, body = Timeout.timeout(10) { read_chunked_response(socket) } + sent = result[/sent=(\d+)/, 1].to_i + expect(status).to start_with('HTTP/1.1 200') + expect(body).to eq('x' * (sent * 16_384)) + ensure + socket&.close + end + + it 'wakes a parked producer when the client disconnects mid-stream' do + sock = Socket.tcp('localhost', server_port) + begin + sock.write("GET /backpressure HTTP/1.1\r\nHost: localhost\r\n\r\n") + + wait_for_backpressure { |r| r =~ /would_blocks=[1-9]/ } + ensure + sock.close + end + + result = wait_for_backpressure { |r| r.include?('result=disconnected') } + expect(result).to match(/result=disconnected/) + + response = http_get('/te-write') + expect(consume_body(response)).to eq('hello') + end + + it 'isolates wake subscriptions for pipelined streams on one connection' do + socket = Socket.tcp('localhost', server_port) + responses = Timeout.timeout(15) do + socket.write( + "GET /backpressure HTTP/1.1\r\nHost: localhost\r\n\r\n" \ + "GET /backpressure HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" + ) + [read_chunked_response(socket), read_chunked_response(socket)] + end + + expected = 'x' * (256 * 16_384) + expect(responses.map(&:first)).to all(start_with('HTTP/1.1 200')) + expect(responses.map(&:last)).to eq([expected, expected]) + ensure + socket&.close + end + it 'keeps streaming after the callable returns' do body = +"" released = false diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru index b99bf0e8..9c861303 100644 --- a/spec/support/apps/response_streaming.ru +++ b/spec/support/apps/response_streaming.ru @@ -4,6 +4,8 @@ same_fiber = nil oversized_result = nil release_channel = "response-streaming-release" +backpressure = nil +backpressure_close = nil run ->(env) do if env['PATH_INFO'] == '/stream-state' @@ -81,6 +83,103 @@ run ->(env) do next [200, {}, ["result=#{oversized_result}"]] end + if ['/backpressure', '/backpressure-close'].include?(env['PATH_INFO']) + close_while_blocked = env['PATH_INFO'] == '/backpressure-close' + state = { + result: nil, + sent: 0, + would_blocks: 0, + wakes: 0, + last_wake: nil, + finished: false, + unsubscribed: false + } + if close_while_blocked + backpressure_close = state + else + backpressure = state + end + + body = proc do |stream| + payload = "x" * 16_384 + total = 256 + + producer = Fiber.new do + sent = 0 + while sent < total + case (status = stream.write(payload)) + when :ok + sent += 1 + state[:sent] = sent + when :would_block + state[:would_blocks] += 1 + Fiber.yield + state[:wakes] += 1 + else + state[:result] = status + break + end + end + state[:result] ||= :completed + stream.close + state[:finished] = true + end + + channel = stream.wake_channel + Iodine.subscribe(channel) do |_, message| + state[:last_wake] = message + producer.resume if producer.alive? + unless producer.alive? + Iodine.defer do + removed = Iodine.unsubscribe(channel) + state[:unsubscribed] = removed && !Iodine.subscribed?(channel) + end + end + end + + if close_while_blocked + state[:close_scheduled] = true + Iodine.defer do + producer.resume + if producer.alive? && state[:would_blocks] > 0 + state[:closed_externally] = true + stream.close + elsif !producer.alive? + Iodine.defer do + removed = Iodine.unsubscribe(channel) + state[:unsubscribed] = removed && !Iodine.subscribed?(channel) + end + end + end + else + producer.resume + unless producer.alive? + Iodine.defer do + removed = Iodine.unsubscribe(channel) + state[:unsubscribed] = removed && !Iodine.subscribed?(channel) + end + end + end + end + + next [200, {}, body] + end + + if env['PATH_INFO'] == '/backpressure-result' + s = backpressure || {} + next [200, {}, ["result=#{s[:result]} sent=#{s[:sent]} would_blocks=#{s[:would_blocks]} wakes=#{s[:wakes]}"]] + end + + if env['PATH_INFO'] == '/backpressure-close-result' + s = backpressure_close || {} + next [200, {}, [ + "result=#{s[:result]} sent=#{s[:sent]} would_blocks=#{s[:would_blocks]} " \ + "wakes=#{s[:wakes]} last_wake=#{s[:last_wake]} " \ + "close_scheduled=#{s[:close_scheduled]} closed_externally=#{s[:closed_externally]} " \ + "finished=#{s[:finished]} unsubscribed=#{s[:unsubscribed]}" + ]] + end + if env['PATH_INFO'] == '/async' body = proc do |stream| Iodine.subscribe(release_channel) do diff --git a/spec/unit/rack_stream_spec.rb b/spec/unit/rack_stream_spec.rb index 67a8f1a8..50624626 100644 --- a/spec/unit/rack_stream_spec.rb +++ b/spec/unit/rack_stream_spec.rb @@ -5,7 +5,9 @@ expect(defined?(Iodine::Base::RackStream)).to eq('constant') end - it 'exposes the writer API: write, close, closed?' do - expect(Iodine::Base::RackStream.instance_methods(false)).to include(:write, :close, :closed?) + it 'exposes the writer and wake-channel API' do + expect(Iodine::Base::RackStream.instance_methods(false)).to include( + :write, :close, :closed?, :wake_channel + ) end end From a262f87f3c9a555db4fadeddfd0c911e6aaf81f1 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Sun, 9 Aug 2026 16:57:07 +0530 Subject: [PATCH 10/14] spec: cover streaming tests --- spec/integration/response_streaming_spec.rb | 161 +++++++++++--------- spec/support/apps/response_streaming.ru | 40 ++++- 2 files changed, 124 insertions(+), 77 deletions(-) diff --git a/spec/integration/response_streaming_spec.rb b/spec/integration/response_streaming_spec.rb index ee6c4881..43cdc892 100644 --- a/spec/integration/response_streaming_spec.rb +++ b/spec/integration/response_streaming_spec.rb @@ -1,5 +1,4 @@ require 'spec_helper' -require 'timeout' # Functional tests for HTTP response streaming: a real Iodine server runs the # `response_streaming` app and the HTTP gem consumes the response incrementally. @@ -28,48 +27,6 @@ def wait_for_backpressure(deadline: 10, result_path: '/backpressure-result') end end - def read_chunked_response(socket) - status = socket.gets("\r\n") - raise EOFError, 'connection closed before response status' unless status - - headers = {} - while (line = socket.gets("\r\n")) && line != "\r\n" - name, value = line.delete_suffix("\r\n").split(':', 2) - headers[name.downcase] = value&.strip - end - raise EOFError, 'connection closed before response headers completed' unless line - unless headers['transfer-encoding'] == 'chunked' - raise "expected chunked response, got #{headers.inspect}" - end - - body = +"" - loop do - size_line = socket.gets("\r\n") - raise EOFError, 'connection closed before chunk size' unless size_line - - size = size_line.split(';', 2).first.to_i(16) - if size.zero? - loop do - trailer = socket.gets("\r\n") - unless trailer - raise EOFError, 'connection closed before chunk trailers completed' - end - break if trailer == "\r\n" - end - break - end - - chunk = socket.read(size) - unless chunk&.bytesize == size - raise EOFError, 'connection closed inside response chunk' - end - raise 'invalid chunk terminator' unless socket.read(2) == "\r\n" - body << chunk - end - - [status, body] - end - it 'responds 200 and reassembles the full streamed body' do response = http_get("/") expect(response.status).to eq(200) @@ -168,6 +125,61 @@ def read_chunked_response(socket) expect(consume_body(result)).to eq('result=error') end + it 'serves a multi-part each body through the buffered non-streaming path' do + response = http_get('/each-body') + expect(response.status).to eq(200) + expect(response.chunked?).to be(false) + expect(response.headers['Content-Length']).to eq('12') + expect(consume_body(response)).to eq('each-body-ok') + end + + it 'serves a status-only response through the empty non-streaming path' do + response = http_get('/status-only') + expect(response.status).to eq(204) + expect(response.headers.get('Transfer-Encoding')).to be_empty + # Iodine's http_finish adds Content-Length: 0 to every header-only + # response, including 204. RFC 9110 forbids it on 204; worth raising + # upstream separately. + expect(response.headers['Content-Length']).to eq('0') + expect(response.headers.get('Content-Type')).to be_empty + expect(consume_body(response)).to eq("") + end + + it 'serves a plain request after a streamed response on the same connection' do + http_client.persistent("http://localhost:#{server_port}") do |client| + streamed = client.get('/') + expect(streamed.status).to eq(200) + expect(streamed.chunked?).to be(true) + expect(streamed.headers['Connection']).to eq('keep-alive') + expect(consume_body(streamed)).to eq(expected) + + plain = client.get('/each-body') + expect(plain.status).to eq(200) + expect(plain.chunked?).to be(false) + expect(plain.headers['Content-Length']).to eq('12') + expect(consume_body(plain)).to eq('each-body-ok') + end + end + + it 'treats close as idempotent and rejects writes after the terminal state' do + http_client.persistent("http://localhost:#{server_port}") do |client| + response = client.get('/double-close') + expect(response.status).to eq(200) + expect(response.chunked?).to be(true) + expect(consume_body(response)).to eq('payload') + + followup = client.get('/each-body') + expect(followup.status).to eq(200) + expect(consume_body(followup)).to eq('each-body-ok') + end + + result = consume_body(http_get('/double-close-result')) + expect(result).to eq( + 'first_close=nil second_close=nil closed=true ' \ + 'write_after_close=closed wake_channel=nil' + ) + end + it 'wakes a blocked producer when the outgoing queue drains' do response = http_get('/backpressure') @@ -175,14 +187,18 @@ def read_chunked_response(socket) expect(consume_body(response)).to eq('x' * (256 * 16_384)) - result = wait_for_backpressure { |r| r.include?('result=completed') } - expect(result).to match(/result=completed sent=256 would_blocks=[1-9]\d* wakes=[1-9]\d*/) + result = wait_for_backpressure do |r| + r.include?('result=completed') && r.include?('unsubscribed=true') + end + expect(result).to match( + /result=completed sent=256 would_blocks=[1-9]\d* wakes=[1-9]\d* finished=true unsubscribed=true/ + ) end it 'wakes a blocked producer when another callback closes the stream' do - socket = Socket.tcp('localhost', server_port) - socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 16_384) - socket.write("GET /backpressure-close HTTP/1.1\r\nHost: localhost\r\n\r\n") + # The body stays unconsumed until the server reports the close, so the + # outgoing queue backs up and the producer blocks on its own. + response = http_get('/backpressure-close') result = wait_for_backpressure(result_path: '/backpressure-close-result') do |r| r.include?('result=closed') && r.include?('unsubscribed=true') @@ -193,46 +209,39 @@ def read_chunked_response(socket) closed_externally=true\s+finished=true\s+unsubscribed=true\z}x ) - status, body = Timeout.timeout(10) { read_chunked_response(socket) } sent = result[/sent=(\d+)/, 1].to_i - expect(status).to start_with('HTTP/1.1 200') - expect(body).to eq('x' * (sent * 16_384)) - ensure - socket&.close + expect(response.status).to eq(200) + expect(consume_body(response)).to eq('x' * (sent * 16_384)) end it 'wakes a parked producer when the client disconnects mid-stream' do - sock = Socket.tcp('localhost', server_port) - begin - sock.write("GET /backpressure HTTP/1.1\r\nHost: localhost\r\n\r\n") + client = http_client + client.get("http://localhost:#{server_port}/backpressure") - wait_for_backpressure { |r| r =~ /would_blocks=[1-9]/ } - ensure - sock.close - end + wait_for_backpressure { |r| r =~ /would_blocks=[1-9]/ } + client.close - result = wait_for_backpressure { |r| r.include?('result=disconnected') } - expect(result).to match(/result=disconnected/) + result = wait_for_backpressure do |r| + r.include?('result=disconnected') && r.include?('unsubscribed=true') + end + expect(result).to match(/result=disconnected .*finished=true unsubscribed=true/) response = http_get('/te-write') expect(consume_body(response)).to eq('hello') end - it 'isolates wake subscriptions for pipelined streams on one connection' do - socket = Socket.tcp('localhost', server_port) - responses = Timeout.timeout(15) do - socket.write( - "GET /backpressure HTTP/1.1\r\nHost: localhost\r\n\r\n" \ - "GET /backpressure HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n" - ) - [read_chunked_response(socket), read_chunked_response(socket)] - end + it 'isolates wake subscriptions for sequential streams on one connection' do + expected_stream = 'x' * (256 * 16_384) - expected = 'x' * (256 * 16_384) - expect(responses.map(&:first)).to all(start_with('HTTP/1.1 200')) - expect(responses.map(&:last)).to eq([expected, expected]) - ensure - socket&.close + http_client.persistent("http://localhost:#{server_port}") do |client| + first = client.get('/backpressure') + expect(first.status).to eq(200) + expect(consume_body(first)).to eq(expected_stream) + + second = client.get('/backpressure') + expect(second.status).to eq(200) + expect(consume_body(second)).to eq(expected_stream) + end end it 'keeps streaming after the callable returns' do diff --git a/spec/support/apps/response_streaming.ru b/spec/support/apps/response_streaming.ru index 9c861303..812ace8d 100644 --- a/spec/support/apps/response_streaming.ru +++ b/spec/support/apps/response_streaming.ru @@ -6,6 +6,7 @@ oversized_result = nil release_channel = "response-streaming-release" backpressure = nil backpressure_close = nil +double_close = nil run ->(env) do if env['PATH_INFO'] == '/stream-state' @@ -83,6 +84,40 @@ run ->(env) do next [200, {}, ["result=#{oversized_result}"]] end + # Non-streaming response regressions. + if env['PATH_INFO'] == '/each-body' + next [200, {}, ['each-', 'body-', 'ok']] + end + + if env['PATH_INFO'] == '/status-only' + next [204, {}, []] + end + + if env['PATH_INFO'] == '/double-close' + state = {} + double_close = state + + body = proc do |stream| + stream.write("payload") + state[:first_close] = stream.close.inspect + state[:closed_after_first] = stream.closed? + state[:second_close] = stream.close.inspect + state[:write_after_close] = stream.write("late") + state[:wake_channel_after_close] = stream.wake_channel.inspect + end + + next [200, {}, body] + end + + if env['PATH_INFO'] == '/double-close-result' + s = double_close || {} + next [200, {}, [ + "first_close=#{s[:first_close]} second_close=#{s[:second_close]} " \ + "closed=#{s[:closed_after_first]} write_after_close=#{s[:write_after_close]} " \ + "wake_channel=#{s[:wake_channel_after_close]}" + ]] + end + if ['/backpressure', '/backpressure-close'].include?(env['PATH_INFO']) close_while_blocked = env['PATH_INFO'] == '/backpressure-close' state = { @@ -167,7 +202,10 @@ run ->(env) do if env['PATH_INFO'] == '/backpressure-result' s = backpressure || {} - next [200, {}, ["result=#{s[:result]} sent=#{s[:sent]} would_blocks=#{s[:would_blocks]} wakes=#{s[:wakes]}"]] + next [200, {}, [ + "result=#{s[:result]} sent=#{s[:sent]} would_blocks=#{s[:would_blocks]} " \ + "wakes=#{s[:wakes]} finished=#{s[:finished]} unsubscribed=#{s[:unsubscribed]}" + ]] end if env['PATH_INFO'] == '/backpressure-close-result' From b6fcdf13a8e87dcfa6c0daeecf68207f66f9f9ba Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Fri, 14 Aug 2026 15:33:25 +0530 Subject: [PATCH 11/14] fixed logging --- ext/iodine/fio.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ext/iodine/fio.h b/ext/iodine/fio.h index 42daf96c..3ca1746e 100644 --- a/ext/iodine/fio.h +++ b/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)) From 2b9812f6d6266420885b941bda78d3f6bfca0640 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Fri, 14 Aug 2026 15:54:01 +0530 Subject: [PATCH 12/14] fixed HTTP_WAKE_CHANNEL_MAX buffer in http_streaming_wake_channel --- ext/iodine/http.c | 18 +++++++----------- ext/iodine/http.h | 10 +++++++--- ext/iodine/http1.c | 5 ++--- ext/iodine/iodine_rack_stream.c | 5 ++--- 4 files changed, 18 insertions(+), 20 deletions(-) diff --git a/ext/iodine/http.c b/ext/iodine/http.c index 27e5a8e6..dfc6c5f1 100644 --- a/ext/iodine/http.c +++ b/ext/iodine/http.c @@ -422,11 +422,11 @@ void http_streaming_arm_wake(http_s *h) { /** * Copies the current streaming response's NUL-terminated wake channel name to - * `dest`. Returns its length, or 0 if `limit` is too small. + * `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, size_t limit) { +size_t http_streaming_wake_channel(http_s *h, char dest[HTTP_WAKE_CHANNEL_MAX]) { static const char prefix[] = "iodine:stream:"; - char channel[64]; if (HTTP_INVALID_HANDLE(h) || !dest) return 0; @@ -434,15 +434,11 @@ size_t http_streaming_wake_channel(http_s *h, char *dest, size_t limit) { if (!p->stream_generation) return 0; - memcpy(channel, prefix, sizeof(prefix) - 1); + memcpy(dest, prefix, sizeof(prefix) - 1); size_t len = sizeof(prefix) - 1; - len += fio_ltoa(channel + len, (int64_t)p->uuid, 16); - channel[len++] = ':'; - len += fio_ltoa(channel + len, (int64_t)p->stream_generation, 16); - - if (limit <= len) - return 0; - memcpy(dest, channel, len); + 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; } diff --git a/ext/iodine/http.h b/ext/iodine/http.h index c5e22da2..5d6b7642 100644 --- a/ext/iodine/http.h +++ b/ext/iodine/http.h @@ -262,16 +262,20 @@ void http_streaming_end(http_s *h); */ 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`. + * `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 `limit` is too small. A 64-byte buffer is enough. + * Returns its length, or 0 if there's no active streaming response. */ -size_t http_streaming_wake_channel(http_s *h, char *dest, size_t limit); +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). diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index b537de26..513337b2 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -847,9 +847,8 @@ static void http1_on_data(intptr_t uuid, fio_protocol_s *protocol) { /* 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[64]; - size_t channel_len = - http_streaming_wake_channel(&p->request, channel, sizeof(channel)); + 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, diff --git a/ext/iodine/iodine_rack_stream.c b/ext/iodine/iodine_rack_stream.c index 4e105802..d1b27559 100644 --- a/ext/iodine/iodine_rack_stream.c +++ b/ext/iodine/iodine_rack_stream.c @@ -15,7 +15,7 @@ typedef enum { typedef struct { http_s *h; /* stays valid while the response is in streaming mode */ intptr_t uuid; /* socket uuid, for fio_pending / fio_is_valid */ - char wake_channel[64]; /* fixed wake channel for this response */ + char wake_channel[HTTP_WAKE_CHANNEL_MAX]; /* fixed wake channel for this response */ size_t wake_channel_len; iodine_stream_state_e state; int freed; /* terminal guard: teardown runs exactly once */ @@ -180,8 +180,7 @@ static VALUE new_rack_stream(http_s *h) { /* the response now outlives the request callback; only an explicit close * (http_streaming_end) finishes it. */ http_streaming_start(h); - ctx->wake_channel_len = http_streaming_wake_channel( - h, ctx->wake_channel, sizeof(ctx->wake_channel)); + ctx->wake_channel_len = http_streaming_wake_channel(h, ctx->wake_channel); VALUE stream = rb_funcall2(rRackStream, iodine_new_func_id, 0, NULL); set_ctx(stream, ctx); From eced05cdeb884017f6e12bdef0eeaeec53e72fda Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Fri, 14 Aug 2026 15:59:24 +0530 Subject: [PATCH 13/14] refactor the function --- ext/iodine/http1.c | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index 513337b2..0585ad82 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -301,8 +301,10 @@ static int http1_stream(http_s *h, void *data, uintptr_t length) { * stream completes. */ static void http1_streaming_start(http_s *h) { http1pr_s *p = handle2pr(h); - if (!p->streaming && !(++p->p.stream_generation)) - ++p->p.stream_generation; /* zero means no active stream */ + 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; } From 53ee221d4a22bf750640b6bc29fc60141ecacc98 Mon Sep 17 00:00:00 2001 From: Piyush-Goenka Date: Fri, 14 Aug 2026 16:09:01 +0530 Subject: [PATCH 14/14] refactored close and drain into constants --- ext/iodine/http1.c | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/ext/iodine/http1.c b/ext/iodine/http1.c index 0585ad82..0e7d1791 100644 --- a/ext/iodine/http1.c +++ b/ext/iodine/http1.c @@ -48,6 +48,9 @@ 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); @@ -323,7 +326,7 @@ static void http1_streaming_end(http_s *h) { /* 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, "close", 5); + http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_CLOSE, sizeof(HTTP1_STREAM_WAKE_CLOSE) - 1); } p->streaming = 0; http_finish(h); @@ -865,7 +868,7 @@ static void http1_on_close(intptr_t uuid, fio_protocol_s *protocol) { * 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, "close", 5); + http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_CLOSE, sizeof(HTTP1_STREAM_WAKE_CLOSE) - 1); } http1_destroy(protocol); } @@ -881,7 +884,7 @@ static void http1_on_ready(intptr_t uuid, fio_protocol_s *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, "drain", 5); + http1_stream_wake_publish(p, HTTP1_STREAM_WAKE_DRAIN, sizeof(HTTP1_STREAM_WAKE_DRAIN) - 1); } }