59
101
from socket import error as socket_error
60
102
self._socket_error = socket_error
61
103
self._socket_timeout = socket_timeout
62
self._server_socket = socket.socket()
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)
63
110
# SO_REUSERADDR has a different meaning on Windows
64
111
if sys.platform != 'win32':
65
112
self._server_socket.setsockopt(socket.SOL_SOCKET,
66
113
socket.SO_REUSEADDR, 1)
68
self._server_socket.bind((host, port))
115
self._server_socket.bind(sockaddr)
69
116
except self._socket_error, message:
70
117
raise errors.CannotBindAddress(host, port, message)
71
118
self._sockname = self._server_socket.getsockname()
72
119
self.port = self._sockname[1]
73
120
self._server_socket.listen(1)
74
self._server_socket.settimeout(1)
75
self.backing_transport = backing_transport
121
self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
122
# Once we start accept()ing connections, we set started.
76
123
self._started = threading.Event()
124
# Once we stop accept()ing connections (and are closing the socket) we
77
126
self._stopped = threading.Event()
78
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()
80
def serve(self, thread_name_suffix=''):
81
self._should_terminate = False
82
# for hooks we are letting code know that a server has started (and
131
def _backing_urls(self):
84
132
# There are three interesting urls:
85
133
# The URL the server can be contacted on. (e.g. bzr://host/)
86
134
# The URL that a commit done on the same machine as the server will
87
135
# have within the servers space. (e.g. file:///home/user/source)
88
136
# The URL that will be given to other hooks in the same process -
89
# the URL of the backing transport itself. (e.g. chroot+:///)
137
# the URL of the backing transport itself. (e.g. filtered-36195:///)
90
138
# We need all three because:
91
139
# * other machines see the first
92
140
# * local commits on this machine should be able to be mapped to
94
142
# * commits the server does itself need to be mapped across to this
96
144
# The latter two urls are different aliases to the servers url,
97
# so we group those in a list - as there might be more aliases
145
# so we group those in a list - as there might be more aliases
99
backing_urls = [self.backing_transport.base]
147
urls = [self.backing_transport.base]
101
backing_urls.append(self.backing_transport.external_url())
149
urls.append(self.backing_transport.external_url())
102
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()
104
157
for hook in SmartTCPServer.hooks['server_started']:
105
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()
106
204
self._started.set()
115
213
except self._socket_error, e:
116
214
# if the socket is closed by stop_background_thread
117
# we might get a EBADF here, any other socket errors
119
if e.args[0] != errno.EBADF:
120
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:
122
225
self.serve_conn(conn, thread_name_suffix)
226
# Cleanout any threads that have finished processing.
227
self._poll_active_connections()
123
228
except KeyboardInterrupt:
124
229
# dont log when CTRL-C'd.
126
231
except Exception, e:
127
trace.error("Unhandled smart server error.")
128
trace.log_exception_quietly()
232
trace.report_exception(sys.exc_info(), sys.stderr)
133
236
# ensure the server socket is closed.
134
237
self._server_socket.close()
135
238
except self._socket_error:
136
239
# ignore errors on close
138
for hook in SmartTCPServer.hooks['server_stopped']:
139
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()
141
248
def get_url(self):
142
249
"""Return the url of the server"""
143
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
145
274
def serve_conn(self, conn, thread_name_suffix):
146
275
# For WIN32, where the timeout value from the listening socket
147
# propogates to the newly accepted socket.
276
# propagates to the newly accepted socket.
148
277
conn.setblocking(True)
149
278
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
150
handler = SmartServerSocketStreamMedium(
151
conn, self.backing_transport, self.root_client_path)
152
279
thread_name = 'smart-server-child' + thread_name_suffix
280
handler = self._make_handler(conn)
153
281
connection_thread = threading.Thread(
154
282
None, handler.serve, name=thread_name)
283
self._active_connections.append((handler, connection_thread))
155
284
connection_thread.setDaemon(True)
156
285
connection_thread.start()
286
return connection_thread
158
288
def start_background_thread(self, thread_name_suffix=''):
159
289
self._started.clear()
199
329
These are all empty initially, because by default nothing should get
203
# Introduced in 0.16:
204
# invoked whenever the server starts serving a directory.
205
# The api signature is (backing urls, public url).
206
self['server_started'] = []
207
# Introduced in 0.16:
208
# invoked whenever the server stops serving a directory.
209
# The api signature is (backing urls, public url).
210
self['server_stopped'] = []
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))
212
353
SmartTCPServer.hooks = SmartServerHooks()
215
class SmartTCPServer_for_testing(SmartTCPServer):
216
"""Server suitable for use by transport tests.
218
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.
221
def __init__(self, thread_name_suffix=''):
222
SmartTCPServer.__init__(self, None)
223
self.client_path_extra = None
224
self.thread_name_suffix = thread_name_suffix
226
def get_backing_transport(self, backing_transport_server):
227
"""Get a backing transport from a server we are decorating."""
228
return transport.get_transport(backing_transport_server.get_url())
230
def setUp(self, backing_transport_server=None,
231
client_path_extra='/extra/'):
232
"""Set up server for testing.
234
:param backing_transport_server: backing server to use. If not
235
specified, a LocalURLServer at the current working directory will
237
:param client_path_extra: a path segment starting with '/' to append to
238
the root URL for this server. For instance, a value of '/foo/bar/'
239
will mean the root of the backing transport will be published at a
240
URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
241
`bzr://127.0.0.1:nnnn/`. Default value is `extra`, so that tests
242
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.
244
if not client_path_extra.startswith('/'):
245
raise ValueError(client_path_extra)
246
from bzrlib.transport.chroot import ChrootServer
247
if backing_transport_server is None:
248
from bzrlib.transport.local import LocalURLServer
249
backing_transport_server = LocalURLServer()
250
self.chroot_server = ChrootServer(
251
self.get_backing_transport(backing_transport_server))
252
self.chroot_server.setUp()
253
self.backing_transport = transport.get_transport(
254
self.chroot_server.get_url())
255
self.root_client_path = self.client_path_extra = client_path_extra
256
self.start_background_thread(self.thread_name_suffix)
259
self.stop_background_thread()
260
self.chroot_server.tearDown()
263
url = super(SmartTCPServer_for_testing, self).get_url()
264
return url[:-1] + self.client_path_extra
266
def get_bogus_url(self):
267
"""Return a URL which will fail to connect"""
268
return 'bzr://127.0.0.1:1/'
271
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
272
"""Get a readonly server for testing."""
274
def get_backing_transport(self, backing_transport_server):
275
"""Get a backing transport from a server we are decorating."""
276
url = 'readonly+' + backing_transport_server.get_url()
277
return transport.get_transport(url)
280
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
281
"""A variation of SmartTCPServer_for_testing that limits the client to
282
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
286
url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
287
url = 'bzr-v2://' + url[len('bzr://'):]
291
class ReadonlySmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing_v2_only):
292
"""Get a readonly server for testing."""
294
def get_backing_transport(self, backing_transport_server):
295
"""Get a backing transport from a server we are decorating."""
296
url = 'readonly+' + backing_transport_server.get_url()
297
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()