Class: MorioBridge::Client

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(base_socket_dir = nil) ⇒ Client

Returns a new instance of Client.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/morio_bridge.rb', line 33

def initialize(base_socket_dir = nil)
  @server_path = File.expand_path(File.join(Dir.home, ".morio/bin/bridge_server.js"))
  @bun_path = File.expand_path(File.join(Dir.home, ".bun/bin/bun"))
  @base_socket_dir = base_socket_dir || Dir.mktmpdir("prisma")
  @socket_paths = fetch_socket_paths(6)
  @socket_connections = []
  @socket_status = []
  @socket_index = 0
  @pid = nil
  @started = false
  @starting = false
  @instance = nil
  @logger = Logger.new($stdout)

  return unless !@testing_mode && defined?(Morio)

  # morio global logger
  @logger = Morio.logger
end

Instance Attribute Details

#instance ⇒ Object (readonly)

Returns the value of attribute instance.



31
32
33
 
# File 'lib/morio_bridge.rb', line 31

def instance
  @instance
end

Class Method Details

.with_timing(&proc) ⇒ Object

[ Benchmark helpers ]



281
282
283
284
285
286
287
288
 
# File 'lib/morio_bridge.rb', line 281

def self.with_timing(&proc)
  start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond)
  result = proc.call
  end_ns = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond)
  duration_ms = ((end_ns - start) / 1_000_000.0)

  [duration_ms, result]
end

Instance Method Details

#benchmark(send_rate_per_second: 600_000, requests_to_be_sent: 10_000) ⇒ Object

[ Benchmark ]



227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
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
272
273
274
275
276
 
# File 'lib/morio_bridge.rb', line 227

def benchmark(send_rate_per_second: 600_000, requests_to_be_sent: 10_000)
  basic_test = lambda do
    request({ command: "create", body: { name: "John Doe" } })
  end

  speed_test = lambda do
    error_count = 0
    durations = []

    start = Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond)

    loop_count = 0

    loop do
      break if loop_count > requests_to_be_sent

      begin
        duration, _result = UnixClient.with_timing(&basic_test)
        durations << duration
      rescue Error => _e
        error_count += 1
      end

      loop_count += 1

      sleep(1.0 / send_rate_per_second)
    end

    duration_ms = (Process.clock_gettime(Process::CLOCK_MONOTONIC, :nanosecond) - start) / 1_000_000.0
    rps = requests_to_be_sent / duration_ms * 1000

    {
      requests: requests_to_be_sent,
      send_rate_per_second: send_rate_per_second,
      duration_ms: duration_ms.round(2),
      avg_latency_ms: (durations.sum / durations.size).round(2),
      rps: rps.round(0),
      errors: error_count
    }
  end

  # Run tests in parallel
  speed_thread = Thread.new { UnixClient.with_timing(&speed_test) }
  total_duration, results = speed_thread.value

  result = JSON.pretty_generate({ results:, total_duration: })

  # make result gray and dim and log it
  @logger.info result
end

#request(data, skip_client_start: false) ⇒ Object

[ Message Handler ]



148
149
150
151
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
221
222
 
# File 'lib/morio_bridge.rb', line 148

def request(data, skip_client_start: false)
  # start the client if it's not already started
  start unless @started || skip_client_start

  # set the max number of retries
  max_retries = 3
  retries = 0
  timeout = 0.05

  loop do
    # select a socket in round robin fashion
    socket = select_socket

    begin
      # raise an error if the socket is closed
      raise IOError, "Socket is closed" if socket.closed?

      #=====================================
      #
      #  Send Request
      #
      #=====================================
      # generate the request body
      request = to_socket_format(data)

      # write the request to the socket
      socket.write_nonblock(request)

      #=====================================
      #
      #  Read Response
      #
      #=====================================
      # wait for the socket to be readable
      readable = socket.wait_readable(timeout)

      # raise an error if the socket is not readable
      unless readable
        timeout = 0.002
        raise TimeoutError, "Command timed out"
      end

      # read the response from the socket into a string buffer
      buffer = String.new(capacity: 4096)
      done = false

      while (chunk = socket.readpartial(4096))
        buffer << chunk
        done = true if chunk.include?("\r\n\r\n")
        break if done
      end

      #=====================================
      #
      #  Parse Response
      #
      #=====================================
      body = to_hash(from_socket_format(buffer))
      return body
    rescue Errno::EPIPE, IOError, TimeoutError => _e
      # retry the connection in case of a communication error
      retry_failed_connection(socket)

      # increment the retries
      retries += 1

      # raise an error if the retries exceed the max retries
      raise Error, "No healthy sockets after #{max_retries} retries" if retries >= max_retries

      # sleep for a short period before retrying - exponential backoff
      sleep(0.05 * (2**retries))
      retry
    end
  end
end

#start ⇒ Object

[ Lifecycle ]



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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
113
114
115
116
117
118
119
 
# File 'lib/morio_bridge.rb', line 56

def start
  @starting = true

  begin
    FileUtils.mkdir_p(@base_socket_dir)
    FileUtils.chmod(0o700, @base_socket_dir)

    env = {
      "MORIO_RB_BUN_SOCKETS" => @socket_paths.join(","),
      "MORIO_RB_PWD" => Dir.pwd
    }

    # Start server in background thread immediately
    server_thread = Thread.new do
      # Create pipes for stdout and stderr
      out_read, out_write = IO.pipe
      err_read, err_write = IO.pipe

      @pid = Process.spawn(
        env,
        "#{@bun_path} #{@server_path}",
        err: err_write,
        out: out_write,
        pgroup: true,
        close_others: true
      )

      # Close write ends in parent since they are not needed as we dont write anything to the child process
      out_write.close
      err_write.close

      log_server_output(out_read, err_read)

      Process.detach(@pid)
    end

    # Start health check in parallel
    health_thread = Thread.new { server_healthy? }

    # Wait for both to complete
    server_thread.join
    health_thread.join

    # Initialize sockets...
    threads = @socket_paths.map do |socket_path|
      Thread.new do
        socket_healthy?(socket_path)
        socket = UNIXSocket.new(socket_path)
        socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, 1)
        [socket, true]
      end
    end
    @socket_connections, @socket_status = threads.map(&:value).transpose

    @started = true

    # save instance
    @instance = self

    at_exit { stop }
  ensure
    @starting = false
  end
end

#stop ⇒ Object 



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

def stop
  @logger.debug "Stopping morio_bridge ..."

  # close all the socket connections
  @socket_connections.each do |connection|
    connection.close
  rescue StandardError
    nil
  end

  # kill the unix socket cluster process
  Process.kill("TERM", @pid) if @pid

  # remove the socket directory
  FileUtils.remove_entry(@base_socket_dir) if File.directory?(@base_socket_dir)

  # set the started flag to false
  @started = false

  @logger.debug "Stopped the morio_bridge"
rescue StandardError
  nil
end