THRIFT-6434: Release exhausted input buffers in Ruby transports

[kpumuk]

BufferedTransport, FramedTransport, HeaderTransport and HTTPClientTransport retain their last input buffer after the caller consumes it. A long-lived idle transport can therefore keep a large frame or HTTP response alive until the next refill or response.

Release the backing storage when input is exhausted, including when a later read operation raises. Replace the backing string so returned data stays valid. Preserve unread and pipelined bytes, Header reply metadata, HTTP status and existing close/error behavior. Use a direct path for ordinary Buffered/Framed byte reads and StringIO#getbyte for buffered Header/HTTP bytes. Framed refills release the exhausted previous frame before attempting the next read. Regression coverage exercises ownership, read entry points, failures, unsigned bytes, frame transitions and real HTTP/TCP round trips.

Benchmarks

Ruby 4.0.7 / PRISM, aarch64 Linux, YJIT disabled, ordinary GC enabled. Workers were pinned to Linux virtual CPU 0; each workload ran separately, with three warmup batches per worker and a balanced rotating/reversed measurement order. Setup and explicit GC are outside timing. CPU pinning does not reserve a physical host core.

The following pure-Ruby microbenchmark compares master 926dcc971c97aa935f00bc98848352e905c4c755 with that runtime plus the proposed transport changes. Each variant has 24 measured rounds. Byte batches repeat 262,144 read_byte calls six times; chunk batches repeat 8 MiB consumed in 16 KiB read_all calls 24 times. Every read is checked. Values are median milliseconds per repetition; time delta is the ratio of the displayed medians.

Transport / workload	Master ms	Proposed ms	Time delta
Buffered, 262,144 byte reads	32.344	33.378	+3.20%
Framed, 262,144 byte reads	34.017	34.818	+2.35%
Header, 262,144 byte reads	88.528	35.858	−59.50%
HTTP, 262,144 byte reads	81.254	33.606	−58.64%
Buffered, 8 MiB in 16 KiB reads	4.307	4.285	−0.51%
Framed, 8 MiB in 16 KiB reads	2.137	2.139	+0.09%
Header, 8 MiB in 16 KiB reads	2.541	2.551	+0.39%
HTTP, 8 MiB in 16 KiB reads	2.079	2.108	+1.39%

Buffered/Framed retain a small byte-read cost for prompt reclamation. Header/HTTP avoid temporary one-byte Strings: allocated objects per 262,144 byte reads fall from 262,164 to 22 and from 262,145 to 2, respectively. Chunk differences near one percent should be treated as noise-sensitive. HTTP input is a seeded StringIO response body; network/request processing is excluded.

A separate generated-struct benchmark isolates the byte-read optimization against the same input-reclamation patch before that optimization (fixed), rather than against master. It decodes 128 CompactProtoTestStruct records per repetition, with two repetitions per batch and 12 paired rounds. Every decoded object is checked against the fixture. Both variants use this checkout's identical native extension and remaining runtime. The native Compact results are:

Transport	Before byte optimization ms	Proposed ms	Median paired time change
Buffered	15.810	14.551	−7.23%
Framed	16.200	15.105	−6.74%
Header	29.902	15.615	−47.86%
HTTP	28.243	16.367	−41.81%

These decoding changes are incremental gains within this patch, not total gains over master. The full decoding matrix also covered Binary and pure Ruby. This is one fixture on one CRuby environment, without network/application scheduling costs; balanced execution reduces order bias but does not rule out host/VM interference. The earlier seven-process timings are superseded by these measurements.

Benchmark commands, from /thrift/src/lib/rb in the Linux container with this checkout mounted at /thrift/src, generated Ruby spec fixtures available, and its native extension built:

ruby /tmp/thrift-6434-options/bench_run.rb micro
ruby /tmp/thrift-6434-options/bench_run.rb decode

Reproduction harness and source setup

The measured schedule also interleaved an experimental cached-index variant. It is reconstructed below only to reproduce the schedule; it is not included in this PR. The scripts write raw JSONL samples. The fast variant is the proposed change; fixed removes only the byte-read optimization.

# From a checkout containing the proposed commit and the pinned master commit:
bench=/tmp/thrift-6434-options
mkdir -p "$bench/master" "$bench/fixed" "$bench/fast" "$bench/cached"
for path in transport/memory_buffer_transport.rb protocol/json_protocol.rb transport/buffered_transport.rb transport/framed_transport.rb transport/header_transport.rb transport/http_client_transport.rb; do
  git show "926dcc971c97aa935f00bc98848352e905c4c755:lib/rb/lib/thrift/$path" > "$bench/master/${path##*/}"
done
for kind in buffered framed header http_client; do
  git show "9f527c30f7813917eac08cc41f9d14cfeab7cbec:lib/rb/lib/thrift/transport/${kind}_transport.rb" > "$bench/fast/${kind}_transport.rb"
  cp "$bench/fast/${kind}_transport.rb" "$bench/fixed/"
done
# Save the reverseable patch below as "$bench/fast.patch".
git -C "$bench" apply --directory=fixed --reverse -p1 fast.patch
cp "$bench"/fast/*_transport.rb "$bench/cached/"
# Reconstruct the additional experimental variant used to balance the measured runs.
# It is not part of the proposed change.
BENCH="$bench" ruby <<'RUBY'
%w[buffered framed].each do |kind|
  path = File.join(ENV.fetch('BENCH'), 'cached', kind + '_transport.rb')
  source = File.read(path).sub('@index < @rbuf.size - 1', '@index < @rbuf_last_index')
  source = source.lines.flat_map do |line|
    line.match?(/^\s*@rbuf\s*=/) ? [line, line[/^\s*/] + "@rbuf_last_index = @rbuf.size - 1\n"] : [line]
  end.join
  File.write(path, source)
end
RUBY
# Save loader.rb, bench_worker.rb and bench_run.rb below into "$bench".
# Inside the same Ruby environment, from lib/rb:
ruby /tmp/thrift-6434-options/bench_run.rb micro
ruby /tmp/thrift-6434-options/bench_run.rb decode

Save as /tmp/thrift-6434-options/fast.patch:

--- fixed/buffered_transport.rb
+++ fast/buffered_transport.rb
@@ -59,18 +59,25 @@
     end
 
     def read_byte
-      # If the read buffer is exhausted, try to read up to DEFAULT_BUFFER more bytes into it.
-      if @index >= @rbuf.size
-        @rbuf = @transport.read(DEFAULT_BUFFER)
-        @index = 0
+      if @index < @rbuf.size - 1
+        @index += 1
+        return Bytes.get_string_byte(@rbuf, @index - 1)
       end
 
-      # The read buffer has some data now, read a single byte. Using get_string_byte() avoids
-      # allocating a temp string of size 1 unnecessarily.
-      @index += 1
-      Bytes.get_string_byte(@rbuf, @index - 1)
-    ensure
-      clear_consumed_read_buffer if @rbuf && @index >= @rbuf.size
+      begin
+        # If the read buffer is exhausted, try to read up to DEFAULT_BUFFER more bytes into it.
+        if @index >= @rbuf.size
+          @rbuf = @transport.read(DEFAULT_BUFFER)
+          @index = 0
+        end
+
+        # The read buffer has some data now, read a single byte. Using get_string_byte() avoids
+        # allocating a temp string of size 1 unnecessarily.
+        @index += 1
+        Bytes.get_string_byte(@rbuf, @index - 1)
+      ensure
+        clear_consumed_read_buffer if @rbuf && @index >= @rbuf.size
+      end
     end
 
     # Reads a number of bytes from the transport into the buffer passed.
--- fixed/framed_transport.rb
+++ fast/framed_transport.rb
@@ -73,14 +73,21 @@
     def read_byte
       return @transport.read_byte unless @read
 
-      read_frame if @index >= @rbuf.length
+      if @index < @rbuf.size - 1
+        @index += 1
+        return Bytes.get_string_byte(@rbuf, @index - 1)
+      end
 
-      # The read buffer has some data now, read a single byte. Using get_string_byte() avoids
-      # allocating a temp string of size 1 unnecessarily.
-      @index += 1
-      Bytes.get_string_byte(@rbuf, @index - 1)
-    ensure
-      clear_consumed_read_buffer if @rbuf && @index >= @rbuf.size
+      begin
+        read_frame if @index >= @rbuf.length
+
+        # The read buffer has some data now, read a single byte. Using get_string_byte() avoids
+        # allocating a temp string of size 1 unnecessarily.
+        @index += 1
+        Bytes.get_string_byte(@rbuf, @index - 1)
+      ensure
+        clear_consumed_read_buffer if @rbuf && @index >= @rbuf.size
+      end
     end
 
     def read_into_buffer(buffer, size)
--- fixed/header_transport.rb
+++ fast/header_transport.rb
@@ -229,6 +229,14 @@
       if @read_buffer.eof? && !@read_buffer.string.empty?
         @read_buffer.string = Bytes.empty_byte_buffer
       end
+    end
+
+    def read_byte
+      return super unless @read_buffer && !@read_buffer.eof?
+
+      byte = @read_buffer.getbyte
+      @read_buffer.string = Bytes.empty_byte_buffer if @read_buffer.eof?
+      byte
     end
 
     def write(buf)
--- fixed/http_client_transport.rb
+++ fast/http_client_transport.rb
@@ -48,6 +48,14 @@
       end
     end
 
+    def read_byte
+      return super unless @inbuf && !@inbuf.eof?
+
+      byte = @inbuf.getbyte
+      @inbuf.string = Bytes.empty_byte_buffer if @inbuf.eof?
+      byte
+    end
+
     def write(buf); @outbuf << Bytes.force_binary_encoding(buf) end
 
     def add_headers(headers)

Save as /tmp/thrift-6434-options/loader.rb:

$LOADED_FEATURES << 'thrift_native.rb' if ENV['PURE']=='1'
require 'thrift'
$VERBOSE=nil
root=File.dirname(__FILE__)
variant=ENV.fetch('VARIANT')
if ENV['CURRENT_MASTER']=='1'
  %w[memory_buffer json buffered framed header http_client].each do |name|
    suffix=name=='json' ? '_protocol.rb' : '_transport.rb'
    load File.join(root,'master',name+suffix)
  end
end
unless variant=='master'
  %w[buffered framed header http_client].each { |name|load File.join(root,variant,name+'_transport.rb') }
end

Save as /tmp/thrift-6434-options/bench_worker.rb:

require 'json'
require 'stringio'
variant,kind,workload,mode=ARGV
ENV['VARIANT']=variant
ENV['PURE']=mode=='pure' ? '1' : '0'
ENV['CURRENT_MASTER']=workload.start_with?('decode_') ? '0' : '1'
require_relative 'loader'
def make_reader(kind,payload,compact=false)
  case kind
  when 'buffered'
    source=if ENV['DECODE_STREAM_SOURCE']=='1'
    Thrift::IOStreamTransport.new(StringIO.new(payload),StringIO.new(''.b))
  else
    Thrift::MemoryBufferTransport.new(payload)
  end
  Thrift::BufferedTransport.new(source)
  when 'framed'
    Thrift::FramedTransport.new(Thrift::MemoryBufferTransport.new([payload.bytesize].pack('N')+payload))
  when 'header'
    wire=Thrift::MemoryBufferTransport.new
    writer=Thrift::HeaderTransport.new(wire)
    writer.protocol_id=Thrift::HeaderSubprotocolID::COMPACT if compact
    writer.write(payload); writer.flush
    Thrift::HeaderTransport.new(wire)
  when 'http'
    reader=Thrift::HTTPClientTransport.new('http://localhost/')
    reader.instance_variable_set(:@inbuf,StringIO.new(payload))
    reader.instance_variable_set(:@response_code,200)
    reader
  end
end
if workload.start_with?('decode_')
  ENV['DECODE_STREAM_SOURCE']='1'
  $LOAD_PATH.unshift('/thrift/src/lib/rb/test/debug_proto/gen-rb')
  require 'debug_proto_test_constants'
  fixture=Thrift::Test::COMPACT_TEST.dup
  fixture.a_binary=[0,1,2,3,4,5,6,7,8].pack('c*')
  fixture.set_byte_map=nil; fixture.map_byte_map=nil
  compact=workload=='decode_compact'
  protocol_class=compact ? Thrift::CompactProtocol : (mode=='native' ? Thrift::BinaryProtocolAccelerated : Thrift::BinaryProtocol)
  wire=Thrift::MemoryBufferTransport.new
  fixture.write(protocol_class.new(wire))
  unit=wire.read_all(wire.available)
  records=128
  payload=unit*records
  repeats=2
  probe=protocol_class.new(make_reader(kind,payload,compact))
  raise 'wrong protocol native mode' unless probe.native? == (mode=='native')
else
  payload='x'.b*(workload=='byte' ? 262_144 : 8_388_608)
  expected='x'.b*16_384
  repeats=workload=='byte' ? 6 : 24
end
$stdout.sync=true
puts JSON.generate(ready:true,variant:variant,kind:kind,workload:workload,mode:mode,payload_bytes:payload.bytesize)
while (request=$stdin.gets)
  break if JSON.parse(request)['stop']
  result=Thread.new do
    wall=cpu=0.0
    allocations=gc=0
    repeats.times do
      reader=make_reader(kind,payload,compact)
      protocol=protocol_class.new(reader) if protocol_class
      GC.start
      before_objects=GC.stat(:total_allocated_objects)
      before_gc=GC.stat(:count)
      cpu_start=Process.clock_gettime(Process::CLOCK_PROCESS_CPUTIME_ID)
      wall_start=Process.clock_gettime(Process::CLOCK_MONOTONIC)
      if protocol
        decoded=Array.new(records) { fixture.class.new.tap { |value|value.read(protocol) } }
      elsif workload=='byte'
        sum=0
        payload.bytesize.times { sum+=reader.read_byte }
        raise 'wrong bytes' unless sum==120*payload.bytesize
      else
        (payload.bytesize/16_384).times { raise 'wrong chunks' unless reader.read_all(16_384)==expected }
      end
      wall+=Process.clock_gettime(Process::CLOCK_MONOTONIC)-wall_start
      cpu+=Process.clock_gettime(Process::CLOCK_PROCESS_CPUTIME_ID)-cpu_start
      allocations+=GC.stat(:total_allocated_objects)-before_objects
      gc+=GC.stat(:count)-before_gc
      raise 'wrong decoded data' if decoded && !decoded.all? { |value|value==fixture }
      reader.close
    end
    {wall_ms:wall*1000/repeats,cpu_ms:cpu*1000/repeats,objects:allocations.to_f/repeats,timed_gc:gc}
  end.value
  puts JSON.generate(result.merge(variant:variant,kind:kind,workload:workload,mode:mode,repeats:repeats))
end

Save as /tmp/thrift-6434-options/bench_run.rb:

require 'json'
require 'open3'
root=File.dirname(__FILE__)
phase=ARGV.fetch(0)
cpu=File.read('/proc/self/status')[/^Cpus_allowed_list:\s+(.*)/,1].split(/[,-]/).first
$stdout.sync=true
cases=if phase=='micro'
  %w[buffered framed header http].product(%w[byte chunk],['pure'])
else
  %w[buffered framed header http].product(%w[decode_binary decode_compact],%w[pure native])
end
count=phase=='micro' ? 24 : 12
File.open(File.join(root,phase+'_samples.jsonl'),'w') do |output|
  cases.each do |kind,workload,mode|
    variants=phase=='micro' ? %w[master fixed fast cached] : %w[fixed fast cached]
    variants.delete('cached') if %w[header http].include?(kind)
    workers={}
    begin
      variants.each do |variant|
        input,stream,errors,wait=Open3.popen3('taskset','-c',cpu,'ruby','--disable-yjit','-I/thrift/src/lib/rb/lib','-I/thrift/src/lib/rb/ext',File.join(root,'bench_worker.rb'),variant,kind,workload,mode)
        err_thread=Thread.new { errors.read }
        ready=stream.gets
        abort "startup failed #{[variant,kind,workload,mode]}: #{err_thread.value}" unless ready
        workers[variant]=[input,stream,wait,err_thread]
        3.times { input.puts('{}');input.flush;abort 'warmup failed' unless stream.gets }
      end
      count.times do |round|
        order=variants.rotate(round%variants.size)
        order=order.reverse if (round/variants.size).odd?
        order.each_with_index do |variant,position|
          input,stream=workers.fetch(variant)
          input.puts('{}');input.flush
          line=stream.gets
          abort "sample failed #{[variant,kind,workload,mode]}" unless line
          row=JSON.parse(line).merge(round:round+1,position:position)
          output.puts(JSON.generate(row));output.flush
        end
      end
      puts JSON.generate(done:[kind,workload,mode],trials:count,variants:variants)
    ensure
      workers.each_value do |input,stream,wait,errors|
        input.puts('{"stop":true}') rescue nil
        input.close rescue nil
        status=wait.value
        warn errors.value unless status.success?
      end
    end
  end
end

• Did you create an Apache Jira ticket? THRIFT-6434. (Request account here, not required for trivial changes)
• If a ticket exists: Does your pull request title follow the pattern "THRIFT-NNNN: describe my issue"?
• Did you squash your changes to a single commit? (not required, but preferred)
• Did you do your best to avoid breaking changes? If one was needed, did you label the Jira ticket with "Breaking-Change"?
• Documentation-only CI exemption applies (not applicable: code and tests changed).

[Copilot]

🟡 Changes recommended

A failed cross-frame refill can still retain and replay an exhausted FramedTransport buffer.

1 open issue

• Clear consumed buffer when refill fails after frame boundary · New

What changed in this PR

Releases exhausted input storage in Ruby transports so idle connections do not retain large consumed payloads.

Changes:

• Adds exhaustion cleanup to four transports.
• Adds ownership, failure-path, metadata, and HTTP/TCP round-trip regression coverage.

File	Description
lib/​rb/​spec/​http_client_spec.rb	Tests real HTTP response lifetime and failed requests.
lib/​rb/​spec/​header_transport_spec.rb	Tests payload cleanup while preserving metadata and pipelined frames.
lib/​rb/​spec/​base_transport_spec.rb	Adds shared lifetime and TCP regression tests.
lib/​rb/​lib/​thrift/​transport/​http_client_transport.rb	Releases exhausted response storage.
lib/​rb/​lib/​thrift/​transport/​header_transport.rb	Releases exhausted payload storage, preserving metadata.
lib/​rb/​lib/​thrift/​transport/​framed_transport.rb	Adds cleanup across read entry points.
lib/​rb/​lib/​thrift/​transport/​buffered_transport.rb	Adds cleanup across read entry points.

_{Review effort: Balanced}

[Copilot]

🟢 Approval recommended

The scoped buffer-lifecycle changes have thorough regression coverage and no unresolved findings.

No open issues 🎉

1 resolved since last review

• Clear consumed buffer when refill fails after frame boundary

_{Review effort: Balanced}