101
59
from socket import error as socket_error
102
60
self._socket_error = socket_error
103
61
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)
62
self._server_socket = socket.socket()
110
63
# SO_REUSERADDR has a different meaning on Windows
111
64
if sys.platform != 'win32':
112
65
self._server_socket.setsockopt(socket.SOL_SOCKET,
113
66
socket.SO_REUSEADDR, 1)
115
self._server_socket.bind(sockaddr)
68
self._server_socket.bind((host, port))
116
69
except self._socket_error, message:
117
70
raise errors.CannotBindAddress(host, port, message)
118
71
self._sockname = self._server_socket.getsockname()
119
72
self.port = self._sockname[1]
120
73
self._server_socket.listen(1)
121
self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
122
# Once we start accept()ing connections, we set started.
74
self._server_socket.settimeout(1)
75
self.backing_transport = backing_transport
123
76
self._started = threading.Event()
124
# Once we stop accept()ing connections (and are closing the socket) we
126
77
self._stopped = threading.Event()
127
# Once we have finished waiting for all clients, etc. We set
129
self._fully_stopped = threading.Event()
78
self.root_client_path = root_client_path
131
def _backing_urls(self):
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
132
84
# There are three interesting urls:
133
85
# The URL the server can be contacted on. (e.g. bzr://host/)
134
86
# The URL that a commit done on the same machine as the server will
135
87
# have within the servers space. (e.g. file:///home/user/source)
136
88
# 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:///)
89
# the URL of the backing transport itself. (e.g. chroot+:///)
138
90
# We need all three because:
139
91
# * other machines see the first
140
92
# * local commits on this machine should be able to be mapped to
142
94
# * commits the server does itself need to be mapped across to this
144
96
# 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
97
# so we group those in a list - as there might be more aliases
147
urls = [self.backing_transport.base]
99
backing_urls = [self.backing_transport.base]
149
urls.append(self.backing_transport.external_url())
101
backing_urls.append(self.backing_transport.external_url())
150
102
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
104
for hook in SmartTCPServer.hooks['server_started']:
158
105
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()
204
106
self._started.set()
213
115
except self._socket_error, e:
214
116
# 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")
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)
222
if self._should_terminate:
225
122
self.serve_conn(conn, thread_name_suffix)
226
# Cleanout any threads that have finished processing.
227
self._poll_active_connections()
228
123
except KeyboardInterrupt:
229
124
# dont log when CTRL-C'd.
231
126
except Exception, e:
232
trace.report_exception(sys.exc_info(), sys.stderr)
127
trace.error("Unhandled smart server error.")
128
trace.log_exception_quietly()
236
133
# ensure the server socket is closed.
237
134
self._server_socket.close()
238
135
except self._socket_error:
239
136
# 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()
138
for hook in SmartTCPServer.hooks['server_stopped']:
139
hook(backing_urls, self.get_url())
248
141
def get_url(self):
249
142
"""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
143
return "bzr://%s:%d/" % self._sockname
274
145
def serve_conn(self, conn, thread_name_suffix):
275
146
# For WIN32, where the timeout value from the listening socket
276
# propagates to the newly accepted socket.
147
# propogates to the newly accepted socket.
277
148
conn.setblocking(True)
278
149
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
150
handler = SmartServerSocketStreamMedium(
151
conn, self.backing_transport, self.root_client_path)
279
152
thread_name = 'smart-server-child' + thread_name_suffix
280
handler = self._make_handler(conn)
281
153
connection_thread = threading.Thread(
282
154
None, handler.serve, name=thread_name)
283
self._active_connections.append((handler, connection_thread))
284
155
connection_thread.setDaemon(True)
285
156
connection_thread.start()
286
return connection_thread
288
158
def start_background_thread(self, thread_name_suffix=''):
289
159
self._started.clear()
329
199
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))
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'] = []
353
212
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.
215
class SmartTCPServer_for_testing(SmartTCPServer):
216
"""Server suitable for use by transport tests.
218
This server is backed by the process's cwd.
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.
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.
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()
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)