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 ...hooks import Hooks
32
transport as _mod_transport,
34
from ...i18n import gettext
35
from ...lazy_import import lazy_import
36
lazy_import(globals(), """
37
from breezy.bzr.smart import (
41
from breezy.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 as 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 as 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.
231
except Exception as e:
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,
292
thread_name_suffix,),
293
name='server-' + self.get_url())
294
self._server_thread.setDaemon(True)
295
self._server_thread.start()
298
def stop_background_thread(self):
299
self._stopped.clear()
300
# tell the main loop to quit on the next iteration.
301
self._should_terminate = True
302
# close the socket - gives error to connections from here on in,
303
# rather than a connection reset error to connections made during
304
# the period between setting _should_terminate = True and
305
# the current request completing/aborting. It may also break out the
306
# main loop if it was currently in accept() (on some platforms).
308
self._server_socket.close()
309
except self._socket_error:
310
# ignore errors on close
312
if not self._stopped.isSet():
313
# server has not stopped (though it may be stopping)
314
# its likely in accept(), so give it a connection
315
temp_socket = socket.socket()
316
temp_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
317
if not temp_socket.connect_ex(self._sockname):
318
# and close it immediately: we dont choose to send any requests.
321
self._server_thread.join()
324
class SmartServerHooks(Hooks):
325
"""Hooks for the smart server."""
328
"""Create the default hooks.
330
These are all empty initially, because by default nothing should get
333
Hooks.__init__(self, "breezy.bzr.smart.server", "SmartTCPServer.hooks")
334
self.add_hook('server_started',
335
"Called by the bzr server when it starts serving a directory. "
336
"server_started is called with (backing urls, public url), "
337
"where backing_url is a list of URLs giving the "
338
"server-specific directory locations, and public_url is the "
339
"public URL for the directory being served.", (0, 16))
340
self.add_hook('server_started_ex',
341
"Called by the bzr server when it starts serving a directory. "
342
"server_started is called with (backing_urls, server_obj).",
344
self.add_hook('server_stopped',
345
"Called by the bzr server when it stops serving a directory. "
346
"server_stopped is called with the same parameters as the "
347
"server_started hook: (backing_urls, public_url).", (0, 16))
348
self.add_hook('server_exception',
349
"Called by the bzr server when an exception occurs. "
350
"server_exception is called with the sys.exc_info() tuple "
351
"return true for the hook if the exception has been handled, "
352
"in which case the server will exit normally.", (2, 4))
355
SmartTCPServer.hooks = SmartServerHooks()
358
def _local_path_for_transport(transport):
359
"""Return a local path for transport, if reasonably possible.
361
This function works even if transport's url has a "readonly+" prefix,
362
unlike local_path_from_url.
364
This essentially recovers the --directory argument the user passed to "bzr
365
serve" from the transport passed to serve_bzr.
368
base_url = transport.external_url()
369
except (errors.InProcessTransport, NotImplementedError):
372
# Strip readonly prefix
373
if base_url.startswith('readonly+'):
374
base_url = base_url[len('readonly+'):]
376
return urlutils.local_path_from_url(base_url)
377
except urlutils.InvalidURL:
381
class BzrServerFactory(object):
382
"""Helper class for serve_bzr."""
384
def __init__(self, userdir_expander=None, get_base_path=None):
386
self.base_path = None
387
self.backing_transport = None
388
if userdir_expander is None:
389
userdir_expander = os.path.expanduser
390
self.userdir_expander = userdir_expander
391
if get_base_path is None:
392
get_base_path = _local_path_for_transport
393
self.get_base_path = get_base_path
395
def _expand_userdirs(self, path):
396
"""Translate /~/ or /~user/ to e.g. /home/foo, using
397
self.userdir_expander (os.path.expanduser by default).
399
If the translated path would fall outside base_path, or the path does
400
not start with ~, then no translation is applied.
402
If the path is inside, it is adjusted to be relative to the base path.
404
e.g. if base_path is /home, and the expanded path is /home/joe, then
405
the translated path is joe.
408
if path.startswith('~'):
409
expanded = self.userdir_expander(path)
410
if not expanded.endswith('/'):
412
if expanded.startswith(self.base_path):
413
result = expanded[len(self.base_path):]
416
def _make_expand_userdirs_filter(self, transport):
417
return pathfilter.PathFilteringServer(transport, self._expand_userdirs)
419
def _make_backing_transport(self, transport):
420
"""Chroot transport, and decorate with userdir expander."""
421
self.base_path = self.get_base_path(transport)
422
chroot_server = chroot.ChrootServer(transport)
423
chroot_server.start_server()
424
self.cleanups.append(chroot_server.stop_server)
425
transport = _mod_transport.get_transport_from_url(
426
chroot_server.get_url())
427
if self.base_path is not None:
428
# Decorate the server's backing transport with a filter that can
430
expand_userdirs = self._make_expand_userdirs_filter(transport)
431
expand_userdirs.start_server()
432
self.cleanups.append(expand_userdirs.stop_server)
433
transport = _mod_transport.get_transport_from_url(
434
expand_userdirs.get_url())
435
self.transport = transport
437
def _get_stdin_stdout(self):
438
if sys.version_info[0] < 3:
439
return sys.stdin, sys.stdout
441
return sys.stdin.buffer, sys.stdout.buffer
443
def _make_smart_server(self, host, port, inet, timeout):
445
c = config.GlobalStack()
446
timeout = c.get('serve.client_timeout')
448
stdin, stdout = self._get_stdin_stdout()
449
smart_server = medium.SmartServerPipeStreamMedium(
450
stdin, stdout, self.transport, timeout=timeout)
453
host = medium.BZR_DEFAULT_INTERFACE
455
port = medium.BZR_DEFAULT_PORT
456
smart_server = SmartTCPServer(self.transport,
457
client_timeout=timeout)
458
smart_server.start_server(host, port)
459
trace.note(gettext('listening on port: %s'),
460
str(smart_server.port))
461
self.smart_server = smart_server
463
def _change_globals(self):
464
from breezy import lockdir, ui
465
# For the duration of this server, no UI output is permitted. note
466
# that this may cause problems with blackbox tests. This should be
467
# changed with care though, as we dont want to use bandwidth sending
468
# progress over stderr to smart server clients!
469
old_factory = ui.ui_factory
470
old_lockdir_timeout = lockdir._DEFAULT_TIMEOUT_SECONDS
472
def restore_default_ui_factory_and_lockdir_timeout():
473
ui.ui_factory = old_factory
474
lockdir._DEFAULT_TIMEOUT_SECONDS = old_lockdir_timeout
475
self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
476
ui.ui_factory = ui.SilentUIFactory()
477
lockdir._DEFAULT_TIMEOUT_SECONDS = 0
478
orig = signals.install_sighup_handler()
480
def restore_signals():
481
signals.restore_sighup_handler(orig)
482
self.cleanups.append(restore_signals)
484
def set_up(self, transport, host, port, inet, timeout):
485
self._make_backing_transport(transport)
486
self._make_smart_server(host, port, inet, timeout)
487
self._change_globals()
490
for cleanup in reversed(self.cleanups):
494
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
495
"""This is the default implementation of 'bzr serve'.
497
It creates a TCP or pipe smart server on 'transport, and runs it. The
498
transport will be decorated with a chroot and pathfilter (using
501
bzr_server = BzrServerFactory()
503
bzr_server.set_up(transport, host, port, inet, timeout)
504
bzr_server.smart_server.serve()
506
hook_caught_exception = False
507
for hook in SmartTCPServer.hooks['server_exception']:
508
hook_caught_exception = hook(sys.exc_info())
509
if not hook_caught_exception:
512
bzr_server.tear_down()
added
removed
breezy/bzr/smart/server.py
