diff --git a/lib/protocol/http/executor.rb b/lib/protocol/http/executor.rb index f028032..84dec5b 100644 --- a/lib/protocol/http/executor.rb +++ b/lib/protocol/http/executor.rb @@ -5,6 +5,7 @@ require_relative "executor/version" require_relative "executor/error" +require_relative "executor/codec" require_relative "executor/channel" require_relative "executor/transport" require_relative "executor/request" diff --git a/lib/protocol/http/executor/channel.rb b/lib/protocol/http/executor/channel.rb index 302065f..b1b9ab6 100644 --- a/lib/protocol/http/executor/channel.rb +++ b/lib/protocol/http/executor/channel.rb @@ -5,6 +5,8 @@ require "socket" +require_relative "codec" + module Protocol module HTTP module Executor @@ -17,8 +19,10 @@ class Channel # Initialize a channel over the given IO object. # # @parameter io [IO] The connected stream. - def initialize(io) + # @parameter codec [Class] The message serialization codec. + def initialize(io, codec = Codec::MessagePack) @io = io + @codec = codec.new @write_mutex = Mutex.new @read_closed = false @write_closed = false @@ -29,7 +33,7 @@ def initialize(io) # @parameter type [Symbol] The message type. # @parameter payload [Object] The message payload. def write(type, payload = nil) - data = Marshal.dump([type, payload]) + data = @codec.dump([type, payload]) if data.bytesize > MAXIMUM_FRAME_SIZE raise ArgumentError, "Frame is too large: #{data.bytesize} bytes!" @@ -65,7 +69,7 @@ def read raise ClosedError, "Invalid frame size: #{length} bytes!" end - return Marshal.load(read_exactly(length)) + return @codec.load(read_exactly(length)) rescue EOFError, IOError, SystemCallError => error raise ClosedError, error.message end diff --git a/lib/protocol/http/executor/codec.rb b/lib/protocol/http/executor/codec.rb new file mode 100644 index 0000000..a88f266 --- /dev/null +++ b/lib/protocol/http/executor/codec.rb @@ -0,0 +1,69 @@ +# frozen_string_literal: true + +# Released under the MIT License. +# Copyright, 2026, by Samuel Williams. + +require "msgpack" +require "socket" + +module Protocol + module HTTP + module Executor + # Serialization codecs for execution channel messages. + module Codec + # MessagePack serialization for thread execution. + class MessagePack + SYMBOL_TYPE = 0 + ADDRESS_TYPE = 1 + + # Initialize a MessagePack factory with the protocol's extension types. + def initialize + @factory = ::MessagePack::Factory.new + @factory.register_type(SYMBOL_TYPE, Symbol, packer: ->(symbol){symbol.to_s}, unpacker: ->(string){string.to_sym}) + @factory.register_type( + ADDRESS_TYPE, + Addrinfo, + packer: ->(address){::MessagePack.pack([address.to_sockaddr, address.pfamily, address.socktype, address.protocol])}, + unpacker: ->(data){Addrinfo.new(*::MessagePack.unpack(data))}, + ) + end + + # Serialize a protocol message. + # + # @parameter message [Object] The message to serialize. + # @returns [String] The serialized message. + def dump(message) + @factory.dump(message) + end + + # Deserialize a protocol message. + # + # @parameter data [String] The serialized message. + # @returns [Object] The deserialized message. + def load(data) + @factory.load(data) + end + end + + # Marshal serialization for Ractor execution, where MessagePack's native extension is not available. + class Marshal + # Serialize a protocol message. + # + # @parameter message [Object] The message to serialize. + # @returns [String] The serialized message. + def dump(message) + ::Marshal.dump(message) + end + + # Deserialize a protocol message. + # + # @parameter data [String] The serialized message. + # @returns [Object] The deserialized message. + def load(data) + ::Marshal.load(data) + end + end + end + end + end +end diff --git a/lib/protocol/http/executor/generic.rb b/lib/protocol/http/executor/generic.rb index 820ebc9..93f465b 100644 --- a/lib/protocol/http/executor/generic.rb +++ b/lib/protocol/http/executor/generic.rb @@ -11,14 +11,17 @@ module HTTP module Executor # The common middleware implementation for isolated execution contexts. class Generic < ::Protocol::HTTP::Middleware + CODEC = Codec::MessagePack + # Execute a request using a new isolated execution context. # # @parameter request [Protocol::HTTP::Request] The request to execute. # @returns [Protocol::HTTP::Response] The remote response. def call(request) parent = ::Async::Task.current - endpoint, worker_streams = Transport.pair - backend = spawn(worker_streams) + codec = self.class::CODEC + endpoint, worker_streams = Transport.pair(codec) + backend = spawn(worker_streams, codec) Execution.new(endpoint, backend, request, parent).call rescue @@ -31,8 +34,9 @@ def call(request) # Spawn the isolated execution context. # # @parameter worker_streams [Array(IO)] The worker transport streams. + # @parameter codec [Class] The message serialization codec. # @returns [Thread | Ractor] The isolated execution context. - def spawn(worker_streams) + def spawn(worker_streams, codec) raise NotImplementedError end end diff --git a/lib/protocol/http/executor/ractored.rb b/lib/protocol/http/executor/ractored.rb index 2de4077..871f8aa 100644 --- a/lib/protocol/http/executor/ractored.rb +++ b/lib/protocol/http/executor/ractored.rb @@ -8,6 +8,8 @@ module HTTP module Executor # Executes each request in a dedicated Ruby 4.1 Ractor. class Ractored < Generic + CODEC = Codec::Marshal + # Initialize a Ractored executor. # # @parameter delegate [Interface(:call)] A shareable HTTP application. @@ -34,9 +36,11 @@ def self.supported? private - def spawn(worker_streams) + def spawn(worker_streams, codec) descriptors = worker_streams.map(&:fileno).freeze - worker = ::Ractor.new(@delegate, descriptors, name: "protocol-http-executor"){|application, descriptors| Protocol::HTTP::Executor::Worker.run(application, descriptors.map{|descriptor| ::Socket.for_fd(descriptor)})} + worker = ::Ractor.new(@delegate, descriptors, codec, name: "protocol-http-executor") do |application, descriptors, codec| + Protocol::HTTP::Executor::Worker.run(application, descriptors.map{|descriptor| ::Socket.for_fd(descriptor)}, codec) + end # The worker now owns the descriptors. Close these wrappers without closing # the underlying descriptors: diff --git a/lib/protocol/http/executor/threaded.rb b/lib/protocol/http/executor/threaded.rb index 781927d..f48284a 100644 --- a/lib/protocol/http/executor/threaded.rb +++ b/lib/protocol/http/executor/threaded.rb @@ -10,11 +10,11 @@ module Executor class Threaded < Generic private - def spawn(worker_streams) + def spawn(worker_streams, codec) application = @delegate - return Thread.new(application, worker_streams) do |delegate, streams| - Worker.run(delegate, streams) + return Thread.new(application, worker_streams, codec) do |delegate, streams, codec| + Worker.run(delegate, streams, codec) end end end diff --git a/lib/protocol/http/executor/transport.rb b/lib/protocol/http/executor/transport.rb index 30d9efe..6a945ed 100644 --- a/lib/protocol/http/executor/transport.rb +++ b/lib/protocol/http/executor/transport.rb @@ -16,9 +16,10 @@ class Endpoint # # @parameter control [IO] The bidirectional control stream. # @parameter body [IO] The bidirectional request and response body stream. - def initialize(control, body) - @control = Channel.new(control) - @body = Channel.new(body) + # @parameter codec [Class] The message serialization codec. + def initialize(control, body, codec = Codec::MessagePack) + @control = Channel.new(control, codec) + @body = Channel.new(body, codec) end # @attribute [Channel] The bidirectional control channel. @@ -36,12 +37,13 @@ def close # Create connected client and worker transport endpoints. # + # @parameter codec [Class] The message serialization codec. # @returns [Array(Endpoint, Array(IO))] The client endpoint and worker IO objects. - def self.pair + def self.pair(codec = Codec::MessagePack) client_control, worker_control = Socket.pair(:UNIX, :STREAM, 0) client_body, worker_body = Socket.pair(:UNIX, :STREAM, 0) - client = Endpoint.new(client_control, client_body) + client = Endpoint.new(client_control, client_body, codec) worker = [worker_control, worker_body] return client, worker diff --git a/lib/protocol/http/executor/worker.rb b/lib/protocol/http/executor/worker.rb index f219aee..08a2263 100644 --- a/lib/protocol/http/executor/worker.rb +++ b/lib/protocol/http/executor/worker.rb @@ -17,8 +17,9 @@ module Worker # # @parameter application [Interface(:call)] The HTTP application. # @parameter streams [Array(IO)] The control and bidirectional body streams. - def self.run(application, streams) - endpoint = Transport::Endpoint.new(*streams) + # @parameter codec [Class] The message serialization codec. + def self.run(application, streams, codec = Codec::MessagePack) + endpoint = Transport::Endpoint.new(*streams, codec) Sync do execute(application, endpoint) diff --git a/protocol-http-executor.gemspec b/protocol-http-executor.gemspec index 067304e..683fc5c 100644 --- a/protocol-http-executor.gemspec +++ b/protocol-http-executor.gemspec @@ -24,5 +24,6 @@ Gem::Specification.new do |spec| spec.required_ruby_version = ">= 3.3" spec.add_dependency "async" + spec.add_dependency "msgpack", "~> 1.0" spec.add_dependency "protocol-http", "~> 0.70" end diff --git a/releases.md b/releases.md index 0f9fede..e11c6ab 100644 --- a/releases.md +++ b/releases.md @@ -2,6 +2,7 @@ ## Unreleased + - Use MessagePack serialization for threaded execution channels, avoiding application use of `Marshal`. - Add a Rack example using Falcon's `config/serve.rb`, Protocol::Rack, and Rack's opt-in Ractor support. - Add a Falcon integration example for Shopify's Ractor-safe Writebook experiment. diff --git a/test/protocol/http/executor/channel.rb b/test/protocol/http/executor/channel.rb index 4d46507..779e651 100644 --- a/test/protocol/http/executor/channel.rb +++ b/test/protocol/http/executor/channel.rb @@ -6,9 +6,21 @@ require "protocol/http/executor" describe Protocol::HTTP::Executor::Channel do - def channel_pair + class OversizedCodec + def dump(message) + return Data.new + end + + class Data + def bytesize + return Protocol::HTTP::Executor::Channel::MAXIMUM_FRAME_SIZE + 1 + end + end + end + + def channel_pair(codec = Protocol::HTTP::Executor::Codec::MessagePack) left, right = Socket.pair(:UNIX, :STREAM, 0) - return subject.new(left), right + return subject.new(left, codec), right end it "reports whether both directions are closed" do @@ -25,20 +37,33 @@ def channel_pair end it "rejects oversized outgoing frames" do + channel, peer = channel_pair(OversizedCodec) + + expect do + channel.write(:oversized) + end.to raise_exception(ArgumentError, message: be =~ /too large/) + ensure + channel&.close + peer&.close + end + + it "preserves protocol message types and values" do channel, peer = channel_pair - data = Object.new - maximum_frame_size = subject::MAXIMUM_FRAME_SIZE - data.define_singleton_method(:bytesize){maximum_frame_size + 1} + remote = subject.new(peer) + message = [:response, {status: 200, peer: Addrinfo.tcp("127.0.0.1", 443), body: "\x00\xFF".b}] - mock(Marshal) do |marshal| - marshal.replace(:dump){data} - - expect do - channel.write(:oversized) - end.to raise_exception(ArgumentError, message: be =~ /too large/) - end + channel.write(*message) + + type, payload = remote.read + expect(type).to be == :response + expect(payload[:status]).to be == 200 + expect(payload[:body]).to be == "\x00\xFF".b + expect(payload[:peer].to_sockaddr).to be == message.last[:peer].to_sockaddr + expect(payload[:peer].socktype).to be == message.last[:peer].socktype + expect(payload[:peer].protocol).to be == message.last[:peer].protocol ensure channel&.close + remote&.close peer&.close end diff --git a/test/protocol/http/executor/generic.rb b/test/protocol/http/executor/generic.rb index 797820d..09502c8 100644 --- a/test/protocol/http/executor/generic.rb +++ b/test/protocol/http/executor/generic.rb @@ -10,7 +10,7 @@ executor = subject.new(Protocol::HTTP::Middleware::Okay) expect do - executor.send(:spawn, []) + executor.send(:spawn, [], Protocol::HTTP::Executor::Codec::MessagePack) end.to raise_exception(NotImplementedError) end end