Module: Uma::ServerControl

Extended by:
ServerControl
Included in:
ServerControl
Defined in:
lib/uma/server.rb

Constant Summary collapse

BIND_SPEC_RE = 
/^(.+)\:(\d+)$/.freeze

Instance Method Summary collapse

Instance Method Details

#await_process_termination(machine) ⇒ Object 



143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/uma/server.rb', line 143

def await_process_termination(machine)
  sig_queue = UM::Queue.new

  old_term_handler = trap('SIGTERM') {
    machine.push(sig_queue, :term)
    trap('SIGTERM', old_term_handler)
  }
  old_int_handler = trap('SIGINT')  {
    machine.push(sig_queue, :int)
    trap('SIGINT', old_int_handler)
  }

  machine.shift(sig_queue)
end

#bind_entries(bind_value) ⇒ Object 



47
48
49
50
51
52
53
54
55
56
 
# File 'lib/uma/server.rb', line 47

def bind_entries(bind_value)
  case bind_value
  when Array
    bind_value.map { parse_bind_spec(it) }
  when String
    [parse_bind_spec(bind_value)]
  else
    raise ArgumentError, "invalid bind value"
  end
end

#parse_bind_spec(spec) ⇒ Object 



60
61
62
63
64
65
66
 
# File 'lib/uma/server.rb', line 60

def parse_bind_spec(spec)
  if (m = spec.match(BIND_SPEC_RE))
    [m[1], m[2].to_i]
  else
    raise ArgumentError, "Invalid bind spec"
  end
end

#prepare_listening_socket(machine, host, port) ⇒ Object 



107
108
109
110
111
112
113
 
# File 'lib/uma/server.rb', line 107

def prepare_listening_socket(machine, host, port)
  fd = machine.socket(UM::AF_INET, UM::SOCK_STREAM, 0, 0)
  machine.setsockopt(fd, UM::SOL_SOCKET, UM::SO_REUSEPORT, true)
  machine.bind(fd, host, port)
  machine.listen(fd, UM::SOMAXCONN)
  fd
end

#server_config(env) ⇒ Object 



38
39
40
41
42
43
44
45
 
# File 'lib/uma/server.rb', line 38

def server_config(env)
  {
    thread_count: 2,
    bind_entries: env[:bind] ? bind_entries(env[:bind]) : [],
    connection_proc: env[:connection_proc],
    error_stream: env[:error_stream]
  }
end

#start_acceptors(machine, config, connection_fibers) ⇒ Set<Fiber>

Returns a set of accept fibers.

Returns:

  • (Set<Fiber>) —

    a set of accept fibers



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/uma/server.rb', line 88

def start_acceptors(machine, config, connection_fibers)
  set = Set.new
  return set if !config[:bind_entries] || config[:bind_entries].empty?

  config[:bind_entries].each do
    host, port = it
    set << machine.spin do
      fd = prepare_listening_socket(machine, host, port)
      machine.accept_each(fd) { |fd|
        start_connection(machine, config, connection_fibers, fd)
      }
    rescue UM::Terminate
    ensure
      machine.close(fd)
    end
  end
  set
end

#start_connection(machine, config, connection_fibers, fd) ⇒ Fiber

Returns:

  • (Fiber) 


116
117
118
119
120
121
122
123
124
 
# File 'lib/uma/server.rb', line 116

def start_connection(machine, config, connection_fibers, fd)
  f = machine.spin do
    connection_fibers << f
    config[:connection_proc]&.(machine, fd)
  ensure
    machine.close(fd) rescue nil
    connection_fibers.delete(f)
  end
end

#start_worker_thread(config, stop_queue) ⇒ Object 



68
69
70
71
72
73
74
75
76
 
# File 'lib/uma/server.rb', line 68

def start_worker_thread(config, stop_queue)
  Thread.new do
    machine = UM.new
    scheduler = UM::FiberScheduler.new(machine)
    Fiber.set_scheduler(scheduler)

    worker_thread(machine, config, stop_queue)
  end
end

#worker_thread(machine, config, stop_queue) ⇒ Object 



78
79
80
81
82
83
84
85
 
# File 'lib/uma/server.rb', line 78

def worker_thread(machine, config, stop_queue)
  connection_fibers = Set.new
  accept_fibers = start_acceptors(machine, config, connection_fibers)

  machine.shift(stop_queue)

  worker_thread_graceful_stop(machine, accept_fibers, connection_fibers)
end

#worker_thread_graceful_stop(machine, accept_fibers, connection_fibers) ⇒ Object 



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/uma/server.rb', line 126

def worker_thread_graceful_stop(machine, accept_fibers, connection_fibers)
  # stop accepting connections
  machine.terminate(accept_fibers)
  machine.await_fibers(accept_fibers)

  machine.terminate(connection_fibers)

  # graceful stop with a timeout of 10 seconds
  machine.timeout(10, UM::Terminate) do
    machine.await_fibers(connection_fibers)
  rescue UM::Terminate
    alive = connection_fibers.reject(&:done?)
    machine.terminate(alive)
    machine.await_fibers(alive)
  end
end