104
144
# The latter two urls are different aliases to the servers url,
105
145
# so we group those in a list - as there might be more aliases
107
backing_urls = [self.backing_transport.base]
147
urls = [self.backing_transport.base]
109
backing_urls.append(self.backing_transport.external_url())
149
urls.append(self.backing_transport.external_url())
110
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()
112
157
for hook in SmartTCPServer.hooks['server_started']:
113
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()
114
204
self._started.set()
123
213
except self._socket_error, e:
124
214
# if the socket is closed by stop_background_thread
125
# we might get a EBADF here, any other socket errors
127
if e.args[0] != errno.EBADF:
128
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")
222
if self._should_terminate:
130
225
self.serve_conn(conn, thread_name_suffix)
226
# Cleanout any threads that have finished processing.
227
self._poll_active_connections()
131
228
except KeyboardInterrupt:
132
229
# dont log when CTRL-C'd.
134
231
except Exception, e:
135
trace.error("Unhandled smart server error.")
136
trace.log_exception_quietly()
232
trace.report_exception(sys.exc_info(), sys.stderr)
141
236
# ensure the server socket is closed.
142
237
self._server_socket.close()
143
238
except self._socket_error:
144
239
# ignore errors on close
146
for hook in SmartTCPServer.hooks['server_stopped']:
147
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()
149
248
def get_url(self):
150
249
"""Return the url of the server"""
151
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
153
274
def serve_conn(self, conn, thread_name_suffix):
154
275
# For WIN32, where the timeout value from the listening socket
155
276
# propagates to the newly accepted socket.
156
277
conn.setblocking(True)
157
278
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
158
handler = medium.SmartServerSocketStreamMedium(
159
conn, self.backing_transport, self.root_client_path)
160
279
thread_name = 'smart-server-child' + thread_name_suffix
280
handler = self._make_handler(conn)
161
281
connection_thread = threading.Thread(
162
282
None, handler.serve, name=thread_name)
283
self._active_connections.append((handler, connection_thread))
163
284
connection_thread.setDaemon(True)
164
285
connection_thread.start()
286
return connection_thread
166
288
def start_background_thread(self, thread_name_suffix=''):
167
289
self._started.clear()
207
329
These are all empty initially, because by default nothing should get
211
self.create_hook(HookPoint('server_started',
332
Hooks.__init__(self, "bzrlib.smart.server", "SmartTCPServer.hooks")
333
self.add_hook('server_started',
212
334
"Called by the bzr server when it starts serving a directory. "
213
335
"server_started is called with (backing urls, public url), "
214
336
"where backing_url is a list of URLs giving the "
215
337
"server-specific directory locations, and public_url is the "
216
"public URL for the directory being served.", (0, 16), None))
217
self.create_hook(HookPoint('server_stopped',
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',
218
344
"Called by the bzr server when it stops serving a directory. "
219
345
"server_stopped is called with the same parameters as the "
220
"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))
222
353
SmartTCPServer.hooks = SmartServerHooks()
225
class SmartTCPServer_for_testing(SmartTCPServer):
226
"""Server suitable for use by transport tests.
228
This server is backed by the process's cwd.
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.
231
def __init__(self, thread_name_suffix=''):
232
SmartTCPServer.__init__(self, None)
233
self.client_path_extra = None
234
self.thread_name_suffix = thread_name_suffix
236
def get_backing_transport(self, backing_transport_server):
237
"""Get a backing transport from a server we are decorating."""
238
return transport.get_transport(backing_transport_server.get_url())
240
def setUp(self, backing_transport_server=None,
241
client_path_extra='/extra/'):
242
"""Set up server for testing.
244
:param backing_transport_server: backing server to use. If not
245
specified, a LocalURLServer at the current working directory will
247
:param client_path_extra: a path segment starting with '/' to append to
248
the root URL for this server. For instance, a value of '/foo/bar/'
249
will mean the root of the backing transport will be published at a
250
URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
251
`bzr://127.0.0.1:nnnn/`. Default value is `extra`, so that tests
252
by default will fail unless they do the necessary path translation.
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.
254
if not client_path_extra.startswith('/'):
255
raise ValueError(client_path_extra)
256
from bzrlib.transport.chroot import ChrootServer
257
if backing_transport_server is None:
258
from bzrlib.transport.local import LocalURLServer
259
backing_transport_server = LocalURLServer()
260
self.chroot_server = ChrootServer(
261
self.get_backing_transport(backing_transport_server))
262
self.chroot_server.setUp()
263
self.backing_transport = transport.get_transport(
264
self.chroot_server.get_url())
265
self.root_client_path = self.client_path_extra = client_path_extra
266
self.start_background_thread(self.thread_name_suffix)
269
self.stop_background_thread()
270
self.chroot_server.tearDown()
273
url = super(SmartTCPServer_for_testing, self).get_url()
274
return url[:-1] + self.client_path_extra
276
def get_bogus_url(self):
277
"""Return a URL which will fail to connect"""
278
return 'bzr://127.0.0.1:1/'
281
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
282
"""Get a readonly server for testing."""
284
def get_backing_transport(self, backing_transport_server):
285
"""Get a backing transport from a server we are decorating."""
286
url = 'readonly+' + backing_transport_server.get_url()
287
return transport.get_transport(url)
290
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
291
"""A variation of SmartTCPServer_for_testing that limits the client to
292
using RPCs in protocol v2 (i.e. bzr <= 1.5).
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
296
url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
297
url = 'bzr-v2://' + url[len('bzr://'):]
301
class ReadonlySmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing_v2_only):
302
"""Get a readonly server for testing."""
304
def get_backing_transport(self, backing_transport_server):
305
"""Get a backing transport from a server we are decorating."""
306
url = 'readonly+' + backing_transport_server.get_url()
307
return transport.get_transport(url)
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()