58
51
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='/',
54
def __init__(self, backing_transport, host='127.0.0.1', port=0,
55
root_client_path='/'):
72
56
"""Construct a new server.
74
58
To actually start it running, call either start_background_thread or
77
61
:param backing_transport: The transport to serve.
62
:param host: Name of the interface to listen on.
63
:param port: TCP port to listen on, or 0 to allocate a transient port.
78
64
:param root_client_path: The client path that will correspond to root
79
65
of backing_transport.
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
67
# let connections timeout so that we get a chance to terminate
98
68
# Keep a reference to the exceptions we want to catch because the socket
118
88
self._sockname = self._server_socket.getsockname()
119
89
self.port = self._sockname[1]
120
90
self._server_socket.listen(1)
121
self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
122
# Once we start accept()ing connections, we set started.
91
self._server_socket.settimeout(1)
92
self.backing_transport = backing_transport
123
93
self._started = threading.Event()
124
# Once we stop accept()ing connections (and are closing the socket) we
126
94
self._stopped = threading.Event()
127
# Once we have finished waiting for all clients, etc. We set
129
self._fully_stopped = threading.Event()
95
self.root_client_path = root_client_path
131
def _backing_urls(self):
97
def serve(self, thread_name_suffix=''):
98
self._should_terminate = False
99
# for hooks we are letting code know that a server has started (and
132
101
# There are three interesting urls:
133
102
# The URL the server can be contacted on. (e.g. bzr://host/)
134
103
# The URL that a commit done on the same machine as the server will
135
104
# have within the servers space. (e.g. file:///home/user/source)
136
105
# 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:///)
106
# the URL of the backing transport itself. (e.g. chroot+:///)
138
107
# We need all three because:
139
108
# * other machines see the first
140
109
# * local commits on this machine should be able to be mapped to
144
113
# The latter two urls are different aliases to the servers url,
145
114
# so we group those in a list - as there might be more aliases
147
urls = [self.backing_transport.base]
116
backing_urls = [self.backing_transport.base]
149
urls.append(self.backing_transport.external_url())
118
backing_urls.append(self.backing_transport.external_url())
150
119
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
121
for hook in SmartTCPServer.hooks['server_started']:
158
122
hook(backing_urls, self.get_url())
159
123
for hook in SmartTCPServer.hooks['server_started_ex']:
160
124
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
125
self._started.set()
232
146
trace.report_exception(sys.exc_info(), sys.stderr)
236
151
# ensure the server socket is closed.
237
152
self._server_socket.close()
238
153
except self._socket_error:
239
154
# 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()
156
for hook in SmartTCPServer.hooks['server_stopped']:
157
hook(backing_urls, self.get_url())
248
159
def get_url(self):
249
160
"""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
161
return "bzr://%s:%d/" % self._sockname
274
163
def serve_conn(self, conn, thread_name_suffix):
275
164
# For WIN32, where the timeout value from the listening socket
276
165
# propagates to the newly accepted socket.
277
166
conn.setblocking(True)
278
167
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
168
handler = medium.SmartServerSocketStreamMedium(
169
conn, self.backing_transport, self.root_client_path)
279
170
thread_name = 'smart-server-child' + thread_name_suffix
280
handler = self._make_handler(conn)
281
171
connection_thread = threading.Thread(
282
172
None, handler.serve, name=thread_name)
283
self._active_connections.append((handler, connection_thread))
284
173
connection_thread.setDaemon(True)
285
174
connection_thread.start()
286
return connection_thread
288
176
def start_background_thread(self, thread_name_suffix=''):
289
177
self._started.clear()
329
217
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',
221
self.create_hook(HookPoint('server_started',
334
222
"Called by the bzr server when it starts serving a directory. "
335
223
"server_started is called with (backing urls, public url), "
336
224
"where backing_url is a list of URLs giving the "
337
225
"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',
226
"public URL for the directory being served.", (0, 16), None))
227
self.create_hook(HookPoint('server_started_ex',
340
228
"Called by the bzr server when it starts serving a directory. "
341
229
"server_started is called with (backing_urls, server_obj).",
343
self.add_hook('server_stopped',
231
self.create_hook(HookPoint('server_stopped',
344
232
"Called by the bzr server when it stops serving a directory. "
345
233
"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))
234
"server_started hook: (backing_urls, public_url).", (0, 16), None))
353
236
SmartTCPServer.hooks = SmartServerHooks()
239
class SmartTCPServer_for_testing(SmartTCPServer):
240
"""Server suitable for use by transport tests.
242
This server is backed by the process's cwd.
245
def __init__(self, thread_name_suffix=''):
246
SmartTCPServer.__init__(self, None)
247
self.client_path_extra = None
248
self.thread_name_suffix = thread_name_suffix
250
def get_backing_transport(self, backing_transport_server):
251
"""Get a backing transport from a server we are decorating."""
252
return transport.get_transport(backing_transport_server.get_url())
254
def setUp(self, backing_transport_server=None,
255
client_path_extra='/extra/'):
256
"""Set up server for testing.
258
:param backing_transport_server: backing server to use. If not
259
specified, a LocalURLServer at the current working directory will
261
:param client_path_extra: a path segment starting with '/' to append to
262
the root URL for this server. For instance, a value of '/foo/bar/'
263
will mean the root of the backing transport will be published at a
264
URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
265
`bzr://127.0.0.1:nnnn/`. Default value is `extra`, so that tests
266
by default will fail unless they do the necessary path translation.
268
if not client_path_extra.startswith('/'):
269
raise ValueError(client_path_extra)
270
from bzrlib.transport.chroot import ChrootServer
271
if backing_transport_server is None:
272
from bzrlib.transport.local import LocalURLServer
273
backing_transport_server = LocalURLServer()
274
self.chroot_server = ChrootServer(
275
self.get_backing_transport(backing_transport_server))
276
self.chroot_server.setUp()
277
self.backing_transport = transport.get_transport(
278
self.chroot_server.get_url())
279
self.root_client_path = self.client_path_extra = client_path_extra
280
self.start_background_thread(self.thread_name_suffix)
283
self.stop_background_thread()
284
self.chroot_server.tearDown()
287
url = super(SmartTCPServer_for_testing, self).get_url()
288
return url[:-1] + self.client_path_extra
290
def get_bogus_url(self):
291
"""Return a URL which will fail to connect"""
292
return 'bzr://127.0.0.1:1/'
295
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
296
"""Get a readonly server for testing."""
298
def get_backing_transport(self, backing_transport_server):
299
"""Get a backing transport from a server we are decorating."""
300
url = 'readonly+' + backing_transport_server.get_url()
301
return transport.get_transport(url)
304
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
305
"""A variation of SmartTCPServer_for_testing that limits the client to
306
using RPCs in protocol v2 (i.e. bzr <= 1.5).
310
url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
311
url = 'bzr-v2://' + url[len('bzr://'):]
315
class ReadonlySmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing_v2_only):
316
"""Get a readonly server for testing."""
318
def get_backing_transport(self, backing_transport_server):
319
"""Get a backing transport from a server we are decorating."""
320
url = 'readonly+' + backing_transport_server.get_url()
321
return transport.get_transport(url)
356
324
def _local_path_for_transport(transport):
357
325
"""Return a local path for transport, if reasonably possible.
418
386
"""Chroot transport, and decorate with userdir expander."""
419
387
self.base_path = self.get_base_path(transport)
420
388
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())
389
chroot_server.setUp()
390
self.cleanups.append(chroot_server.tearDown)
391
transport = get_transport(chroot_server.get_url())
424
392
if self.base_path is not None:
425
393
# Decorate the server's backing transport with a filter that can
426
394
# expand homedirs.
427
395
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())
396
expand_userdirs.setUp()
397
self.cleanups.append(expand_userdirs.tearDown)
398
transport = get_transport(expand_userdirs.get_url())
431
399
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')
401
def _make_smart_server(self, host, port, inet):
441
stdin, stdout = self._get_stdin_stdout()
442
403
smart_server = medium.SmartServerPipeStreamMedium(
443
stdin, stdout, self.transport, timeout=timeout)
404
sys.stdin, sys.stdout, self.transport)
446
407
host = medium.BZR_DEFAULT_INTERFACE
448
409
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)
410
smart_server = SmartTCPServer(self.transport, host=host, port=port)
411
trace.note('listening on port: %s' % smart_server.port)
453
412
self.smart_server = smart_server
455
414
def _change_globals(self):
484
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
439
def serve_bzr(transport, host=None, port=None, inet=False):
485
440
"""This is the default implementation of 'bzr serve'.
487
442
It creates a TCP or pipe smart server on 'transport, and runs it. The
488
443
transport will be decorated with a chroot and pathfilter (using
489
444
os.path.expanduser).
491
446
bzr_server = BzrServerFactory()
493
bzr_server.set_up(transport, host, port, inet, timeout)
448
bzr_server.set_up(transport, host, port, inet)
494
449
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
451
bzr_server.tear_down()
added
removed
bzrlib/smart/server.py
