17
17
"""Server for smart-server protocol."""
19
from __future__ import absolute_import
25
from bzrlib.hooks import HookPoint, Hooks
28
from bzrlib.hooks import Hooks
26
29
from bzrlib import (
32
transport as _mod_transport,
34
from bzrlib.i18n import gettext
31
35
from bzrlib.lazy_import import lazy_import
32
36
lazy_import(globals(), """
33
from bzrlib.smart import medium
37
from bzrlib.smart import (
34
41
from bzrlib.transport import (
39
45
from bzrlib import (
51
58
hooks: An instance of SmartServerHooks.
54
def __init__(self, backing_transport, host='127.0.0.1', port=0,
55
root_client_path='/'):
61
# This is the timeout on the socket we use .accept() on. It is exposed here
62
# so the test suite can set it faster. (It thread.interrupt_main() will not
63
# fire a KeyboardInterrupt during socket.accept)
65
_SHUTDOWN_POLL_TIMEOUT = 1.0
66
_LOG_WAITING_TIMEOUT = 10.0
70
def __init__(self, backing_transport, root_client_path='/',
56
72
"""Construct a new server.
58
74
To actually start it running, call either start_background_thread or
61
77
:param backing_transport: The transport to serve.
78
:param root_client_path: The client path that will correspond to root
80
:param client_timeout: See SmartServerSocketStreamMedium's timeout
83
self.backing_transport = backing_transport
84
self.root_client_path = root_client_path
85
self._client_timeout = client_timeout
86
self._active_connections = []
87
# This is set to indicate we want to wait for clients to finish before
89
self._gracefully_stopping = False
91
def start_server(self, host, port):
92
"""Create the server listening socket.
62
94
:param host: Name of the interface to listen on.
63
95
:param port: TCP port to listen on, or 0 to allocate a transient port.
64
:param root_client_path: The client path that will correspond to root
67
97
# let connections timeout so that we get a chance to terminate
68
98
# Keep a reference to the exceptions we want to catch because the socket
88
118
self._sockname = self._server_socket.getsockname()
89
119
self.port = self._sockname[1]
90
120
self._server_socket.listen(1)
91
self._server_socket.settimeout(1)
92
self.backing_transport = backing_transport
121
self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
122
# Once we start accept()ing connections, we set started.
93
123
self._started = threading.Event()
124
# Once we stop accept()ing connections (and are closing the socket) we
94
126
self._stopped = threading.Event()
95
self.root_client_path = root_client_path
127
# Once we have finished waiting for all clients, etc. We set
129
self._fully_stopped = threading.Event()
97
def serve(self, thread_name_suffix=''):
98
self._should_terminate = False
99
# for hooks we are letting code know that a server has started (and
131
def _backing_urls(self):
101
132
# There are three interesting urls:
102
133
# The URL the server can be contacted on. (e.g. bzr://host/)
103
134
# The URL that a commit done on the same machine as the server will
104
135
# have within the servers space. (e.g. file:///home/user/source)
105
136
# The URL that will be given to other hooks in the same process -
106
# the URL of the backing transport itself. (e.g. chroot+:///)
137
# the URL of the backing transport itself. (e.g. filtered-36195:///)
107
138
# We need all three because:
108
139
# * other machines see the first
109
140
# * local commits on this machine should be able to be mapped to
113
144
# The latter two urls are different aliases to the servers url,
114
145
# so we group those in a list - as there might be more aliases
116
backing_urls = [self.backing_transport.base]
147
urls = [self.backing_transport.base]
118
backing_urls.append(self.backing_transport.external_url())
149
urls.append(self.backing_transport.external_url())
119
150
except errors.InProcessTransport:
154
def run_server_started_hooks(self, backing_urls=None):
155
if backing_urls is None:
156
backing_urls = self._backing_urls()
121
157
for hook in SmartTCPServer.hooks['server_started']:
122
158
hook(backing_urls, self.get_url())
123
159
for hook in SmartTCPServer.hooks['server_started_ex']:
124
160
hook(backing_urls, self)
162
def run_server_stopped_hooks(self, backing_urls=None):
163
if backing_urls is None:
164
backing_urls = self._backing_urls()
165
for hook in SmartTCPServer.hooks['server_stopped']:
166
hook(backing_urls, self.get_url())
168
def _stop_gracefully(self):
169
trace.note(gettext('Requested to stop gracefully'))
170
self._should_terminate = True
171
self._gracefully_stopping = True
172
for handler, _ in self._active_connections:
173
handler._stop_gracefully()
175
def _wait_for_clients_to_disconnect(self):
176
self._poll_active_connections()
177
if not self._active_connections:
179
trace.note(gettext('Waiting for %d client(s) to finish')
180
% (len(self._active_connections),))
181
t_next_log = self._timer() + self._LOG_WAITING_TIMEOUT
182
while self._active_connections:
184
if now >= t_next_log:
185
trace.note(gettext('Still waiting for %d client(s) to finish')
186
% (len(self._active_connections),))
187
t_next_log = now + self._LOG_WAITING_TIMEOUT
188
self._poll_active_connections(self._SHUTDOWN_POLL_TIMEOUT)
190
def serve(self, thread_name_suffix=''):
191
# Note: There is a temptation to do
192
# signals.register_on_hangup(id(self), self._stop_gracefully)
193
# However, that creates a temporary object which is a bound
194
# method. signals._on_sighup is a WeakKeyDictionary so it
195
# immediately gets garbage collected, because nothing else
196
# references it. Instead, we need to keep a real reference to the
197
# bound method for the lifetime of the serve() function.
198
stop_gracefully = self._stop_gracefully
199
signals.register_on_hangup(id(self), stop_gracefully)
200
self._should_terminate = False
201
# for hooks we are letting code know that a server has started (and
203
self.run_server_started_hooks()
125
204
self._started.set()
134
213
except self._socket_error, e:
135
214
# if the socket is closed by stop_background_thread
136
# we might get a EBADF here, any other socket errors
138
if e.args[0] != errno.EBADF:
139
trace.warning("listening socket error: %s", e)
215
# we might get a EBADF here, or if we get a signal we
216
# can get EINTR, any other socket errors should get
218
if e.args[0] not in (errno.EBADF, errno.EINTR):
219
trace.warning(gettext("listening socket error: %s")
141
222
if self._should_terminate:
143
225
self.serve_conn(conn, thread_name_suffix)
226
# Cleanout any threads that have finished processing.
227
self._poll_active_connections()
144
228
except KeyboardInterrupt:
145
229
# dont log when CTRL-C'd.
148
232
trace.report_exception(sys.exc_info(), sys.stderr)
153
236
# ensure the server socket is closed.
154
237
self._server_socket.close()
155
238
except self._socket_error:
156
239
# ignore errors on close
158
for hook in SmartTCPServer.hooks['server_stopped']:
159
hook(backing_urls, self.get_url())
242
signals.unregister_on_hangup(id(self))
243
self.run_server_stopped_hooks()
244
if self._gracefully_stopping:
245
self._wait_for_clients_to_disconnect()
246
self._fully_stopped.set()
161
248
def get_url(self):
162
249
"""Return the url of the server"""
163
return "bzr://%s:%d/" % self._sockname
250
return "bzr://%s:%s/" % (self._sockname[0], self._sockname[1])
252
def _make_handler(self, conn):
253
return medium.SmartServerSocketStreamMedium(
254
conn, self.backing_transport, self.root_client_path,
255
timeout=self._client_timeout)
257
def _poll_active_connections(self, timeout=0.0):
258
"""Check to see if any active connections have finished.
260
This will iterate through self._active_connections, and update any
261
connections that are finished.
263
:param timeout: The timeout to pass to thread.join(). By default, we
264
set it to 0, so that we don't hang if threads are not done yet.
268
for handler, thread in self._active_connections:
271
still_active.append((handler, thread))
272
self._active_connections = still_active
165
274
def serve_conn(self, conn, thread_name_suffix):
166
275
# For WIN32, where the timeout value from the listening socket
167
276
# propagates to the newly accepted socket.
168
277
conn.setblocking(True)
169
278
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
170
handler = medium.SmartServerSocketStreamMedium(
171
conn, self.backing_transport, self.root_client_path)
172
279
thread_name = 'smart-server-child' + thread_name_suffix
280
handler = self._make_handler(conn)
173
281
connection_thread = threading.Thread(
174
282
None, handler.serve, name=thread_name)
283
self._active_connections.append((handler, connection_thread))
175
284
connection_thread.setDaemon(True)
176
285
connection_thread.start()
286
return connection_thread
178
288
def start_background_thread(self, thread_name_suffix=''):
179
289
self._started.clear()
219
329
These are all empty initially, because by default nothing should get
223
self.create_hook(HookPoint('server_started',
332
Hooks.__init__(self, "bzrlib.smart.server", "SmartTCPServer.hooks")
333
self.add_hook('server_started',
224
334
"Called by the bzr server when it starts serving a directory. "
225
335
"server_started is called with (backing urls, public url), "
226
336
"where backing_url is a list of URLs giving the "
227
337
"server-specific directory locations, and public_url is the "
228
"public URL for the directory being served.", (0, 16), None))
229
self.create_hook(HookPoint('server_started_ex',
338
"public URL for the directory being served.", (0, 16))
339
self.add_hook('server_started_ex',
230
340
"Called by the bzr server when it starts serving a directory. "
231
341
"server_started is called with (backing_urls, server_obj).",
233
self.create_hook(HookPoint('server_stopped',
343
self.add_hook('server_stopped',
234
344
"Called by the bzr server when it stops serving a directory. "
235
345
"server_stopped is called with the same parameters as the "
236
"server_started hook: (backing_urls, public_url).", (0, 16), None))
346
"server_started hook: (backing_urls, public_url).", (0, 16))
347
self.add_hook('server_exception',
348
"Called by the bzr server when an exception occurs. "
349
"server_exception is called with the sys.exc_info() tuple "
350
"return true for the hook if the exception has been handled, "
351
"in which case the server will exit normally.", (2, 4))
238
353
SmartTCPServer.hooks = SmartServerHooks()
305
420
chroot_server = chroot.ChrootServer(transport)
306
421
chroot_server.start_server()
307
422
self.cleanups.append(chroot_server.stop_server)
308
transport = get_transport(chroot_server.get_url())
423
transport = _mod_transport.get_transport_from_url(chroot_server.get_url())
309
424
if self.base_path is not None:
310
425
# Decorate the server's backing transport with a filter that can
311
426
# expand homedirs.
312
427
expand_userdirs = self._make_expand_userdirs_filter(transport)
313
428
expand_userdirs.start_server()
314
429
self.cleanups.append(expand_userdirs.stop_server)
315
transport = get_transport(expand_userdirs.get_url())
430
transport = _mod_transport.get_transport_from_url(expand_userdirs.get_url())
316
431
self.transport = transport
318
def _make_smart_server(self, host, port, inet):
433
def _get_stdin_stdout(self):
434
return sys.stdin, sys.stdout
436
def _make_smart_server(self, host, port, inet, timeout):
438
c = config.GlobalStack()
439
timeout = c.get('serve.client_timeout')
441
stdin, stdout = self._get_stdin_stdout()
320
442
smart_server = medium.SmartServerPipeStreamMedium(
321
sys.stdin, sys.stdout, self.transport)
443
stdin, stdout, self.transport, timeout=timeout)
324
446
host = medium.BZR_DEFAULT_INTERFACE
326
448
port = medium.BZR_DEFAULT_PORT
327
smart_server = SmartTCPServer(self.transport, host=host, port=port)
328
trace.note('listening on port: %s' % smart_server.port)
449
smart_server = SmartTCPServer(self.transport,
450
client_timeout=timeout)
451
smart_server.start_server(host, port)
452
trace.note(gettext('listening on port: %s') % smart_server.port)
329
453
self.smart_server = smart_server
331
455
def _change_globals(self):
342
466
self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
343
467
ui.ui_factory = ui.SilentUIFactory()
344
468
lockdir._DEFAULT_TIMEOUT_SECONDS = 0
469
orig = signals.install_sighup_handler()
470
def restore_signals():
471
signals.restore_sighup_handler(orig)
472
self.cleanups.append(restore_signals)
346
def set_up(self, transport, host, port, inet):
474
def set_up(self, transport, host, port, inet, timeout):
347
475
self._make_backing_transport(transport)
348
self._make_smart_server(host, port, inet)
476
self._make_smart_server(host, port, inet, timeout)
349
477
self._change_globals()
351
479
def tear_down(self):
356
def serve_bzr(transport, host=None, port=None, inet=False):
484
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
357
485
"""This is the default implementation of 'bzr serve'.
359
487
It creates a TCP or pipe smart server on 'transport, and runs it. The
360
488
transport will be decorated with a chroot and pathfilter (using
361
489
os.path.expanduser).
363
491
bzr_server = BzrServerFactory()
365
bzr_server.set_up(transport, host, port, inet)
493
bzr_server.set_up(transport, host, port, inet, timeout)
366
494
bzr_server.smart_server.serve()
496
hook_caught_exception = False
497
for hook in SmartTCPServer.hooks['server_exception']:
498
hook_caught_exception = hook(sys.exc_info())
499
if not hook_caught_exception:
368
502
bzr_server.tear_down()
added
removed
bzrlib/smart/server.py
