Class: HotCell::Connection

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

Overview

One request per connection: the cold side connects, sends one request with its descriptors, reads one response line, and closes. The hot side never initiates.

SOCK_STREAM rather than SOCK_SEQPACKET, and the receiver pays for it. Real message boundaries would remove the framing problem outright, and Darwin has no AF_UNIX SOCK_SEQPACKET, so a stream socket is what there is. A stream socket does not promise that one sendmsg arrives as one recvmsg, and the ancillary data rides on whichever bytes land first. So read to the newline in a loop, and take the descriptors from whichever recvmsg carried them rather than from the one that completes the line.

An implementation that assumes one sendmsg is one recvmsg passes every test small enough not to fragment, which is why this is written down here rather than left to be rediscovered.

Constant Summary collapse

CHUNK_BYTES = 
4096

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket) ⇒ Connection

Returns a new instance of Connection.



22
23
24
 
# File 'lib/hot_cell/connection.rb', line 22

def initialize(socket)
  @socket = socket
end

Instance Attribute Details

#socket ⇒ Object (readonly)

Returns the value of attribute socket.



20
21
22
 
# File 'lib/hot_cell/connection.rb', line 20

def socket
  @socket
end

Instance Method Details

#close ⇒ Object 



110
111
112
 
# File 'lib/hot_cell/connection.rb', line 110

def close
  socket.close unless socket.closed?
end

#read_line(limit: MAX_RESPONSE_BYTES, deadline: nil) ⇒ Object

UTF-8 for the same reason receive_message forces it: a socket read comes back ASCII-8BIT, where every byte is "valid" and nothing downstream can tell a mis-encoded message from a good one. Both directions should behave the same way, and the one that scrubs is Failure.

deadline is an absolute instant covering the whole line rather than a per-read timeout, because a per-read one bounds nothing. Waiting for the socket to be readable and then calling a blocking gets meant a peer that sent a single byte inside the timeout and then stopped held the caller until the cell's own deadline, or forever against a peer that never closed. Returns nil at end of stream, and raises Timeout when the deadline passes with the line incomplete.



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/hot_cell/connection.rb', line 91

def read_line(limit: MAX_RESPONSE_BYTES, deadline: nil)
  buffer = "".b

  until buffer.end_with?("\n")
    chunk = read_chunk(deadline)

    if chunk.nil?
      return nil if buffer.empty?

      raise MessageError, "message ended after #{buffer.bytesize} bytes with no newline"
    end

    buffer << chunk
    raise MessageError, "message passed #{limit} bytes with no newline" if buffer.bytesize > limit
  end

  buffer.force_encoding Encoding::UTF_8
end

#receive_message(limit: MAX_REQUEST_BYTES) ⇒ Object

Returns [line, descriptors]. The line is nil when the peer closed without sending anything.

The caller owns the descriptors and must close every one of them, including any it will not use. A request that is refused still arrives with its descriptors installed in this process.



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/hot_cell/connection.rb', line 56

def receive_message(limit: MAX_REQUEST_BYTES)
  buffer = "".b
  descriptors = []

  loop do
    chunk, _, _, *controls = socket.recvmsg(CHUNK_BYTES, 0, nil, scm_rights: true)
    descriptors.concat controls.flat_map(&:unix_rights)

    break if chunk.nil? || chunk.empty?
    buffer << chunk
    break if buffer.include?("\n")

    raise MessageError, "message passed #{limit} bytes with no newline" if buffer.bytesize > limit
  end

  [ line_from(buffer, limit), descriptors ]
rescue StandardError
  descriptors.each { |descriptor| descriptor.close unless descriptor.closed? }
  raise
end

#send_message(line, descriptors: []) ⇒ Object

The descriptors ride the first sendmsg and the rest of the line follows as ordinary writes, because a stream socket does not promise that one sendmsg sends all of it. The return value used to be ignored: a short send that carried the ancillary data left the receiver holding the descriptors and waiting for a newline that never came, while this side moved on to waiting for a response. Both ends then sat until something else timed them out — and the caller's descriptors were installed in the cell either way, so it is not a case that fails safe.

Ancillary data goes exactly once. Sending it again with a later chunk would install a second copy of every descriptor in the receiver.



42
43
44
45
46
47
48
49
50
 
# File 'lib/hot_cell/connection.rb', line 42

def send_message(line, descriptors: [])
  sent = if descriptors.empty?
    socket.sendmsg line
  else
    socket.sendmsg line, 0, nil, Socket::AncillaryData.unix_rights(*descriptors.map(&:to_io))
  end

  write_all line.byteslice(sent..) if sent < line.bytesize
end

#to_io ⇒ Object

So that a connection can itself be passed over SCM_RIGHTS. That is how the supervisor hands an accepted connection to a worker without reading it: the caller's descriptors stay queued on the connection until somebody calls recvmsg, and the worker is the one who does.



29
30
31
 
# File 'lib/hot_cell/connection.rb', line 29

def to_io
  socket
end

#write_line(line) ⇒ Object

IO#write already loops until everything is written or it raises, which sendmsg does not.



78
79
80
 
# File 'lib/hot_cell/connection.rb', line 78

def write_line(line)
  write_all line
end