Class: NewBridge::SessionClient

Inherits:
Object
  • Object
show all
Defined in:
lib/new_bridge/session_client.rb

Overview

NewBridge::SessionClient

Responsibilities (Ruby side, inside one Ruby process):

  • Accept one connection from the R gatekeeper (Unix socket or TCP).
  • Serialize and send framed MsgPack envelopes of type REQ.
  • Wait for matching framed MsgPack envelopes of type RET.
  • Handle R -> Ruby callback requests (CALL) by invoking registered Ruby blocks in background threads and replying with RET.

Protocol basics

  • Transport: Unix domain socket for local R (loopback), TCP when R connects to a non-local host (e.g. Docker). Override with ENV = unix|tcp. Same MsgPack framing.
  • Envelopes are MsgPack maps handled by NewBridge::Envelope.
  • Envelope fields used here:
    • call_id: UUID used to correlate one REQ with one RET
    • type: "REQ" / "RET" / "CALL"
    • session_id: logical session; R uses it to isolate per-session env state
    • instance_id: logical instance routing; used for multi-instance pools
    • parent_id: reserved for nested-call correlation (Phase 4+)

Threading model

  • One reader thread (reader_loop) continuously reads frames and dispatches:
    • RET -> pushes into the per-call Queue (sync eval_r) or schedules async completion (+eval_r_async+)
    • CALL -> spawns a new thread to execute the Ruby callback
  • Callback threads send RET back to R while keeping the reader thread free, which is required to support nested REQ servicing while waiting for a callback.

Failure behavior

  • If the socket closes or a read error happens, pending REQ calls are signaled as closed and subsequent eval calls raise.
  • Infrastructure failures (timeouts / broken pipes / connection reset) are handled at a higher level by RInstanceManager restart+retry logic.

Defined Under Namespace

Classes: Error, RProcessError, TimeoutError

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(source_path:, host: '127.0.0.1', bridge_host: nil, runtime_source_path: nil, r_cmd: 'R', use_unix: nil) ⇒ SessionClient

Create a SessionClient for one R runtime instance.

Parameters:

  • source_path (String) —

    Host-side path to .cpp or .so for Rcpp sourceCpp/dyn.load.

  • host (String) (defaults to: '127.0.0.1') —

    bind/listen address for Ruby TCP server.

  • bridge_host (String, nil) (defaults to: nil) —

    host name/path R uses to connect to this Ruby listener (defaults to host). For containerized R, set to the host reachable from the container (e.g. host.docker.internal).

  • runtime_source_path (String, nil) (defaults to: nil) —

    source path visible from runtime process/container.

  • use_unix (Boolean, nil) (defaults to: nil) —

    force Unix (true) or TCP (false). Default nil: auto — Unix for local loopback targets, TCP for remote/non-loopback bridge_host.

  • r_cmd (String, Array<String>) (defaults to: 'R') —

    executable (or argv prefix ending with executable).



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/new_bridge/session_client.rb', line 86

def initialize(source_path:, host: '127.0.0.1', bridge_host: nil, runtime_source_path: nil, r_cmd: 'R',
               use_unix: nil)
  @source_path = File.expand_path(source_path)
  @use_precompiled = @source_path.end_with?('.so')
  @host = host
  @bridge_host = bridge_host || host
  @runtime_source_path = runtime_source_path || @source_path
  @r_cmd = r_cmd
  @use_unix = if use_unix.nil?
                self.class.default_use_unix?(bridge_target: @bridge_host)
              else
                use_unix
              end
  @unix_path = nil
  @server = nil
  @sock = nil
  @write_mx = Mutex.new
  @pending = {}
  @pending_mx = Mutex.new
  @callbacks = {}
  @callbacks_mx = Mutex.new
  @reader = nil
  @r_stderr = +''
  @r_stderr_mx = Mutex.new
  @r_stderr_limit = 64 * 1024
  @r_exit_status = nil
end

Instance Attribute Details

#r_exit_status ⇒ Object (readonly)

Returns the value of attribute r_exit_status.



74
75
76
 
# File 'lib/new_bridge/session_client.rb', line 74

def r_exit_status
  @r_exit_status
end

#r_stderr ⇒ Object (readonly)

Returns the value of attribute r_stderr.



74
75
76
 
# File 'lib/new_bridge/session_client.rb', line 74

def r_stderr
  @r_stderr
end

Class Method Details

.callback_success_transport_payload(result) ⇒ Object

Gatekeeper parses RET success payload with strtod. Contract:

  • register_callback_proc_stub ends with nil → ACK "1" (semantic value staged in R).
  • Direct register_callback blocks may return a Numeric scalar; pass it through for legacy tests.


323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
 
# File 'lib/new_bridge/session_client.rb', line 323

def self.callback_success_transport_payload(result)
  case result
  when nil
    '1'
  when Numeric
    x = result.to_f
    (x.nan? || x.infinite?) ? '0' : x.to_s
  when true
    '1'
  when false
    '0'
  else
    '1'
  end
end

.default_use_unix?(bridge_target:) ⇒ Boolean

Whether to use Unix domain sockets for this client when use_unix is nil.

Policy:

  • GALAAZ_BRIDGE_TRANSPORT=tcp — always TCP (remote R, or debugging).
  • GALAAZ_BRIDGE_TRANSPORT=unix — always Unix when supported.
  • Otherwise: Unix only if bridge_target is a local loopback address/name (127.0.0.1, ::1, localhost) or a Unix socket path (+#+). Any other host (e.g. host.docker.internal, LAN IP) uses TCP so R can reach Ruby over the network.

Parameters:

  • bridge_target (String) —

    resolved bridge_host passed to R (+galaaz_run_bridge+).

Returns:

  • (Boolean) 


124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/new_bridge/session_client.rb', line 124

def self.default_use_unix?(bridge_target:)
  return false unless defined?(UNIXServer)

  case ENV['GALAAZ_BRIDGE_TRANSPORT'].to_s.downcase
  when 'tcp'
    false
  when 'unix'
    true
  else
    t = bridge_target.to_s.strip
    if t.empty?
      false
    elsif t.start_with?('/')
      true
    else
      u = t.downcase
      %w[127.0.0.1 ::1 localhost].include?(u)
    end
  end
end

Instance Method Details

#eval_r(code, session_id: 'default', instance_id: 'default', parent_id: nil, timeout: 60) ⇒ Hash

Evaluate one REQ on the connected R runtime and wait for the matching RET.

Parameters:

  • code (String) —

    R expression or snippet to be evaluated by the gatekeeper.

  • session_id (String) (defaults to: 'default') —

    session env isolation key (maps to R's .galaaz_sessions).

  • instance_id (String) (defaults to: 'default') —

    instance routing key (must match RInstanceManager routing).

  • parent_id (String, nil) (defaults to: nil) —

    reserved for nested-call correlation (Phase 4+).

  • timeout (Numeric) (defaults to: 60) —

    how long to wait for RET before raising TimeoutError.

Returns:

  • (Hash) —

    decoded MsgPack payload map returned by gatekeeper on success.

Raises:



244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
 
# File 'lib/new_bridge/session_client.rb', line 244

def eval_r(code, session_id: 'default', instance_id: 'default', parent_id: nil, timeout: 60)
  call_id = SecureRandom.uuid
  q = Queue.new
  @pending_mx.synchronize { @pending[call_id] = q }
  req = Envelope.encode(
    'call_id' => call_id,
    'type' => 'REQ',
    'session_id' => session_id,
    'instance_id' => instance_id,
    'parent_id' => parent_id,
    'payload' => code
  )
  @write_mx.synchronize { Framing.write_frame(@sock, req) }

  ret = wait_for_ret(q, call_id, timeout)
  if ret['instance_id'] && ret['instance_id'] != instance_id
    raise RProcessError, "instance_id mismatch: expected=#{instance_id} got=#{ret['instance_id']}"
  end
  status = ret['status']
  payload = ret['payload']
  unless payload.is_a?(Hash)
    raise RProcessError, "invalid RET payload type=#{payload.class} (expected map)"
  end
  parsed = payload
  raise RProcessError, parsed['message'] || parsed.inspect unless status == 'success'

  parsed
end

#eval_r_async(code, session_id: 'default', instance_id: 'default', parent_id: nil, timeout: nil, &block) ⇒ String

Schedule one REQ and invoke block on a background thread when RET arrives (or on failure).

Unlike eval_r, this method returns immediately with the call_id String; it does not wait for R unless timeout is nil (wait forever on the Ruby side for RET) or a Numeric (raise completion with TimeoutError if RET is late).

Parameters:

  • timeout (nil, Numeric) (defaults to: nil) —

    nil means no limit; a positive number starts a timer thread.

Returns:

  • (String) —

    call_id for correlation / debugging

Raises:

  • (ArgumentError) —

    if no block is given



282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
 
# File 'lib/new_bridge/session_client.rb', line 282

def eval_r_async(code, session_id: 'default', instance_id: 'default', parent_id: nil, timeout: nil, &block)
  raise ArgumentError, 'eval_r_async requires a block' unless block

  call_id = SecureRandom.uuid
  slot = AsyncPendingEntry.new(instance_id, block)
  @pending_mx.synchronize { @pending[call_id] = slot }

  req = Envelope.encode(
    'call_id' => call_id,
    'type' => 'REQ',
    'session_id' => session_id,
    'instance_id' => instance_id,
    'parent_id' => parent_id,
    'payload' => code
  )
  @write_mx.synchronize { Framing.write_frame(@sock, req) }

  if timeout
    slot.timer_thread = Thread.new { async_timeout_wait(call_id, slot, timeout) }
  end

  call_id
end

#register_callback(&block) ⇒ Object

Register a Ruby callback for R->Ruby CALL envelopes.

The gatekeeper will invoke this callback when it receives a CALL with the corresponding callback_call_id, and this client will reply with a RET. The same call_id may be used many times (R stub function is reused); the block stays registered until the client stops — do not delete on first CALL.



312
313
314
315
316
 
# File 'lib/new_bridge/session_client.rb', line 312

def register_callback(&block)
  callback_call_id = SecureRandom.uuid
  @callbacks_mx.synchronize { @callbacks[callback_call_id] = block }
  callback_call_id
end

#start(accept_timeout: 120) ⇒ Object

Start listening for a single TCP connection from one R process and launch that process.

  • Binds to an ephemeral TCP port, or a Unix socket path when use_unix is set.
  • Starts a thread that runs r_cmd and sources the gatekeeper C++ code inside R.
  • Accepts the runtime connection with accept_timeout.

On startup failure (e.g. runtime exits before connecting), it surfaces R stderr and exit status.



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/new_bridge/session_client.rb', line 152

def start(accept_timeout: 120)
  use_u = @use_unix && defined?(UNIXServer)
  if @use_unix && !defined?(UNIXServer)
    warn '[NewBridge::SessionClient] Unix bridge transport selected but UNIXServer is unavailable; falling back to TCP'
    use_u = false
  end

  if use_u
    @unix_path = File.join(Dir.tmpdir, "galaaz_bridge_#{Process.pid}_#{SecureRandom.hex(6)}.sock")
    File.unlink(@unix_path) if File.exist?(@unix_path)
    @server = UNIXServer.new(@unix_path)
    port = 0
    bridge_for_r = @unix_path.gsub("'", "\\\\'")
  else
    @server = TCPServer.new(@host, 0)
    port = @server.addr[1]
    bridge_for_r = @bridge_host.gsub("'", "\\\\'")
  end

  # Use sourceCpp with a persistent cache directory so repeated runtime
  # starts can reuse compiled shared objects when source is unchanged.
  cpp_escaped = @runtime_source_path.gsub("'", "\\\\'")
  rcpp_cache_dir = (ENV['GALAAZ_RCPP_CACHE_DIR'] || File.join(Dir.tmpdir, 'galaaz_rcpp_cache')).gsub("'", "\\\\'")
  r_script = <<~R
    bridge_host <- '#{bridge_for_r}'
    port <- #{port}L
    stopifnot(requireNamespace("Rcpp", quietly = TRUE))
    library(Rcpp)
    dir.create("#{rcpp_cache_dir}", recursive = TRUE, showWarnings = FALSE)
    sourceCpp(file = "#{cpp_escaped}", rebuild = FALSE, cacheDir = "#{rcpp_cache_dir}")
    galaaz_run_bridge(bridge_host, as.integer(port))
  R

  env = if use_u
          { 'GALAAZ_BRIDGE_HOST' => @unix_path, 'GALAAZ_BRIDGE_PORT' => '0', 'GALAAZ_BRIDGE_UNIX' => '1' }
        else
          { 'GALAAZ_BRIDGE_HOST' => @bridge_host, 'GALAAZ_BRIDGE_PORT' => port.to_s }
        end
  @r_thr = Thread.new do
    launch = @r_cmd.is_a?(Array) ? @r_cmd.dup : [@r_cmd]
    _stdin, stdout_err, wait_thr = Open3.popen2e(env, *launch, '--slave', '--no-save', '-e', r_script)
    # When the R process exits or during shutdown, the underlying pipe can
    # close while this background thread is still blocked reading.
    # Treat that as a normal shutdown and avoid JRuby "stream closed in
    # another thread" warnings.
    begin
      stdout_err.each_line { |line| append_r_stderr(line) }
    rescue IOError, Errno::EPIPE, Errno::ECONNRESET
      # normal shutdown / pipe closure
    end
    @r_exit_status = wait_thr.value
  end

  @sock = Timeout.timeout(accept_timeout) { @server.accept }
  @reader = Thread.new { reader_loop }
  self
rescue Timeout::Error
  # If the runtime process failed early (e.g. docker permission/network/source path),
  # surface stderr/exit details instead of a generic accept timeout.
  process_status = @r_exit_status&.exitstatus
  details = current_r_stderr
  msg = "failed to accept runtime connection within #{accept_timeout}s"
  msg += " (runtime exit=#{process_status})" if process_status
  unless details.empty?
    tail = details.length > 2000 ? details[-2000, 2000] : details
    msg += " stderr_tail=#{tail}"
  end
  raise RProcessError, msg
end

#stop ⇒ Object

Stop the client and attempt to join threads cleanly. Best-effort: socket/server are closed, reader thread is joined, runtime thread is joined.



224
225
226
227
228
229
230
231
232
233
 
# File 'lib/new_bridge/session_client.rb', line 224

def stop
  @sock&.close rescue nil
  @server&.close rescue nil
  if @unix_path && File.socket?(@unix_path)
    File.unlink(@unix_path) rescue nil
  end
  @unix_path = nil
  @reader&.join(3)
  @r_thr&.join(15)
end