1
# Copyright (C) 2006-2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Server for smart-server protocol."""
19
from __future__ import absolute_import
28
from bzrlib.hooks import Hooks
32
transport as _mod_transport,
34
from bzrlib.i18n import gettext
35
from bzrlib.lazy_import import lazy_import
36
lazy_import(globals(), """
37
from bzrlib.smart import (
41
from bzrlib.transport import (
52
class SmartTCPServer(object):
53
"""Listens on a TCP socket and accepts connections from smart clients.
55
Each connection will be served by a SmartServerSocketStreamMedium running in
58
hooks: An instance of SmartServerHooks.
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='/',
72
"""Construct a new server.
74
To actually start it running, call either start_background_thread or
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.
94
:param host: Name of the interface to listen on.
95
:param port: TCP port to listen on, or 0 to allocate a transient port.
97
# let connections timeout so that we get a chance to terminate
98
# Keep a reference to the exceptions we want to catch because the socket
99
# module's globals get set to None during interpreter shutdown.
100
from socket import timeout as socket_timeout
101
from socket import error as socket_error
102
self._socket_error = socket_error
103
self._socket_timeout = socket_timeout
104
addrs = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
105
socket.SOCK_STREAM, 0, socket.AI_PASSIVE)[0]
107
(family, socktype, proto, canonname, sockaddr) = addrs
109
self._server_socket = socket.socket(family, socktype, proto)
110
# SO_REUSERADDR has a different meaning on Windows
111
if sys.platform != 'win32':
112
self._server_socket.setsockopt(socket.SOL_SOCKET,
113
socket.SO_REUSEADDR, 1)
115
self._server_socket.bind(sockaddr)
116
except self._socket_error, message:
117
raise errors.CannotBindAddress(host, port, message)
118
self._sockname = self._server_socket.getsockname()
119
self.port = self._sockname[1]
120
self._server_socket.listen(1)
121
self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
122
# Once we start accept()ing connections, we set started.
123
self._started = threading.Event()
124
# Once we stop accept()ing connections (and are closing the socket) we
126
self._stopped = threading.Event()
127
# Once we have finished waiting for all clients, etc. We set
129
self._fully_stopped = threading.Event()
131
def _backing_urls(self):
132
# There are three interesting urls:
133
# The URL the server can be contacted on. (e.g. bzr://host/)
134
# The URL that a commit done on the same machine as the server will
135
# have within the servers space. (e.g. file:///home/user/source)
136
# The URL that will be given to other hooks in the same process -
137
# the URL of the backing transport itself. (e.g. filtered-36195:///)
138
# We need all three because:
139
# * other machines see the first
140
# * local commits on this machine should be able to be mapped to
142
# * commits the server does itself need to be mapped across to this
144
# The latter two urls are different aliases to the servers url,
145
# so we group those in a list - as there might be more aliases
147
urls = [self.backing_transport.base]
149
urls.append(self.backing_transport.external_url())
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()
157
for hook in SmartTCPServer.hooks['server_started']:
158
hook(backing_urls, self.get_url())
159
for hook in SmartTCPServer.hooks['server_started_ex']:
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()
207
while not self._should_terminate:
209
conn, client_addr = self._server_socket.accept()
210
except self._socket_timeout:
211
# just check if we're asked to stop
213
except self._socket_error, e:
214
# if the socket is closed by stop_background_thread
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")
222
if self._should_terminate:
225
self.serve_conn(conn, thread_name_suffix)
226
# Cleanout any threads that have finished processing.
227
self._poll_active_connections()
228
except KeyboardInterrupt:
229
# dont log when CTRL-C'd.
232
trace.report_exception(sys.exc_info(), sys.stderr)
236
# ensure the server socket is closed.
237
self._server_socket.close()
238
except self._socket_error:
239
# ignore errors on close
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()
249
"""Return the url of the server"""
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
274
def serve_conn(self, conn, thread_name_suffix):
275
# For WIN32, where the timeout value from the listening socket
276
# propagates to the newly accepted socket.
277
conn.setblocking(True)
278
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
279
thread_name = 'smart-server-child' + thread_name_suffix
280
handler = self._make_handler(conn)
281
connection_thread = threading.Thread(
282
None, handler.serve, name=thread_name)
283
self._active_connections.append((handler, connection_thread))
284
connection_thread.setDaemon(True)
285
connection_thread.start()
286
return connection_thread
288
def start_background_thread(self, thread_name_suffix=''):
289
self._started.clear()
290
self._server_thread = threading.Thread(None,
291
self.serve, args=(thread_name_suffix,),
292
name='server-' + self.get_url())
293
self._server_thread.setDaemon(True)
294
self._server_thread.start()
297
def stop_background_thread(self):
298
self._stopped.clear()
299
# tell the main loop to quit on the next iteration.
300
self._should_terminate = True
301
# close the socket - gives error to connections from here on in,
302
# rather than a connection reset error to connections made during
303
# the period between setting _should_terminate = True and
304
# the current request completing/aborting. It may also break out the
305
# main loop if it was currently in accept() (on some platforms).
307
self._server_socket.close()
308
except self._socket_error:
309
# ignore errors on close
311
if not self._stopped.isSet():
312
# server has not stopped (though it may be stopping)
313
# its likely in accept(), so give it a connection
314
temp_socket = socket.socket()
315
temp_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
316
if not temp_socket.connect_ex(self._sockname):
317
# and close it immediately: we dont choose to send any requests.
320
self._server_thread.join()
323
class SmartServerHooks(Hooks):
324
"""Hooks for the smart server."""
327
"""Create the default hooks.
329
These are all empty initially, because by default nothing should get
332
Hooks.__init__(self, "bzrlib.smart.server", "SmartTCPServer.hooks")
333
self.add_hook('server_started',
334
"Called by the bzr server when it starts serving a directory. "
335
"server_started is called with (backing urls, public url), "
336
"where backing_url is a list of URLs giving the "
337
"server-specific directory locations, and public_url is the "
338
"public URL for the directory being served.", (0, 16))
339
self.add_hook('server_started_ex',
340
"Called by the bzr server when it starts serving a directory. "
341
"server_started is called with (backing_urls, server_obj).",
343
self.add_hook('server_stopped',
344
"Called by the bzr server when it stops serving a directory. "
345
"server_stopped is called with the same parameters as the "
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))
353
SmartTCPServer.hooks = SmartServerHooks()
356
def _local_path_for_transport(transport):
357
"""Return a local path for transport, if reasonably possible.
359
This function works even if transport's url has a "readonly+" prefix,
360
unlike local_path_from_url.
362
This essentially recovers the --directory argument the user passed to "bzr
363
serve" from the transport passed to serve_bzr.
366
base_url = transport.external_url()
367
except (errors.InProcessTransport, NotImplementedError):
370
# Strip readonly prefix
371
if base_url.startswith('readonly+'):
372
base_url = base_url[len('readonly+'):]
374
return urlutils.local_path_from_url(base_url)
375
except errors.InvalidURL:
379
class BzrServerFactory(object):
380
"""Helper class for serve_bzr."""
382
def __init__(self, userdir_expander=None, get_base_path=None):
384
self.base_path = None
385
self.backing_transport = None
386
if userdir_expander is None:
387
userdir_expander = os.path.expanduser
388
self.userdir_expander = userdir_expander
389
if get_base_path is None:
390
get_base_path = _local_path_for_transport
391
self.get_base_path = get_base_path
393
def _expand_userdirs(self, path):
394
"""Translate /~/ or /~user/ to e.g. /home/foo, using
395
self.userdir_expander (os.path.expanduser by default).
397
If the translated path would fall outside base_path, or the path does
398
not start with ~, then no translation is applied.
400
If the path is inside, it is adjusted to be relative to the base path.
402
e.g. if base_path is /home, and the expanded path is /home/joe, then
403
the translated path is joe.
406
if path.startswith('~'):
407
expanded = self.userdir_expander(path)
408
if not expanded.endswith('/'):
410
if expanded.startswith(self.base_path):
411
result = expanded[len(self.base_path):]
414
def _make_expand_userdirs_filter(self, transport):
415
return pathfilter.PathFilteringServer(transport, self._expand_userdirs)
417
def _make_backing_transport(self, transport):
418
"""Chroot transport, and decorate with userdir expander."""
419
self.base_path = self.get_base_path(transport)
420
chroot_server = chroot.ChrootServer(transport)
421
chroot_server.start_server()
422
self.cleanups.append(chroot_server.stop_server)
423
transport = _mod_transport.get_transport_from_url(chroot_server.get_url())
424
if self.base_path is not None:
425
# Decorate the server's backing transport with a filter that can
427
expand_userdirs = self._make_expand_userdirs_filter(transport)
428
expand_userdirs.start_server()
429
self.cleanups.append(expand_userdirs.stop_server)
430
transport = _mod_transport.get_transport_from_url(expand_userdirs.get_url())
431
self.transport = transport
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()
442
smart_server = medium.SmartServerPipeStreamMedium(
443
stdin, stdout, self.transport, timeout=timeout)
446
host = medium.BZR_DEFAULT_INTERFACE
448
port = medium.BZR_DEFAULT_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)
453
self.smart_server = smart_server
455
def _change_globals(self):
456
from bzrlib import lockdir, ui
457
# For the duration of this server, no UI output is permitted. note
458
# that this may cause problems with blackbox tests. This should be
459
# changed with care though, as we dont want to use bandwidth sending
460
# progress over stderr to smart server clients!
461
old_factory = ui.ui_factory
462
old_lockdir_timeout = lockdir._DEFAULT_TIMEOUT_SECONDS
463
def restore_default_ui_factory_and_lockdir_timeout():
464
ui.ui_factory = old_factory
465
lockdir._DEFAULT_TIMEOUT_SECONDS = old_lockdir_timeout
466
self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
467
ui.ui_factory = ui.SilentUIFactory()
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)
474
def set_up(self, transport, host, port, inet, timeout):
475
self._make_backing_transport(transport)
476
self._make_smart_server(host, port, inet, timeout)
477
self._change_globals()
480
for cleanup in reversed(self.cleanups):
484
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
485
"""This is the default implementation of 'bzr serve'.
487
It creates a TCP or pipe smart server on 'transport, and runs it. The
488
transport will be decorated with a chroot and pathfilter (using
491
bzr_server = BzrServerFactory()
493
bzr_server.set_up(transport, host, port, inet, timeout)
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:
502
bzr_server.tear_down()
added
removed
bzrlib/smart/server.py
