← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/server.py

(jelmer) Use the absolute_import feature everywhere in bzrlib,
 and add a source test to make sure it's used everywhere. (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/server.py
1
 
# Copyright (C) 2006, 2007, 2008 Canonical Ltd
 
1
# Copyright (C) 2006-2011 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
16
16
 
17
17
"""Server for smart-server protocol."""
18
18
 
 
19
from __future__ import absolute_import
 
20
 
19
21
import errno
20
22
import os.path
21
23
import socket
22
24
import sys
 
25
import time
23
26
import threading
24
27
 
25
 
from bzrlib.hooks import HookPoint, Hooks
 
28
from bzrlib.hooks import Hooks
26
29
from bzrlib import (
27
30
    errors,
28
31
    trace,
29
 
    transport,
 
32
    transport as _mod_transport,
30
33
)
 
34
from bzrlib.i18n import gettext
31
35
from bzrlib.lazy_import import lazy_import
32
36
lazy_import(globals(), """
33
 
from bzrlib.smart import medium
 
37
from bzrlib.smart import (
 
38
    medium,
 
39
    signals,
 
40
    )
34
41
from bzrlib.transport import (
35
42
    chroot,
36
 
    get_transport,
37
43
    pathfilter,
38
44
    )
39
45
from bzrlib import (
 
46
    config,
40
47
    urlutils,
41
48
    )
42
49
""")
51
58
    hooks: An instance of SmartServerHooks.
52
59
    """
53
60
 
54
 
    def __init__(self, backing_transport, host='127.0.0.1', port=0,
55
 
                 root_client_path='/'):
 
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)
 
64
    _ACCEPT_TIMEOUT = 1.0
 
65
    _SHUTDOWN_POLL_TIMEOUT = 1.0
 
66
    _LOG_WAITING_TIMEOUT = 10.0
 
67
 
 
68
    _timer = time.time
 
69
 
 
70
    def __init__(self, backing_transport, root_client_path='/',
 
71
                 client_timeout=None):
56
72
        """Construct a new server.
57
73
 
58
74
        To actually start it running, call either start_background_thread or
59
75
        serve.
60
76
 
61
77
        :param backing_transport: The transport to serve.
 
78
        :param root_client_path: The client path that will correspond to root
 
79
            of backing_transport.
 
80
        :param client_timeout: See SmartServerSocketStreamMedium's timeout
 
81
            parameter.
 
82
        """
 
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
 
88
        # we disconnect.
 
89
        self._gracefully_stopping = False
 
90
 
 
91
    def start_server(self, host, port):
 
92
        """Create the server listening socket.
 
93
 
62
94
        :param host: Name of the interface to listen on.
63
95
        :param port: TCP port to listen on, or 0 to allocate a transient port.
64
 
        :param root_client_path: The client path that will correspond to root
65
 
            of backing_transport.
66
96
        """
67
97
        # let connections timeout so that we get a chance to terminate
68
98
        # Keep a reference to the exceptions we want to catch because the socket
88
118
        self._sockname = self._server_socket.getsockname()
89
119
        self.port = self._sockname[1]
90
120
        self._server_socket.listen(1)
91
 
        self._server_socket.settimeout(1)
92
 
        self.backing_transport = backing_transport
 
121
        self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
 
122
        # Once we start accept()ing connections, we set started.
93
123
        self._started = threading.Event()
 
124
        # Once we stop accept()ing connections (and are closing the socket) we
 
125
        # set _stopped
94
126
        self._stopped = threading.Event()
95
 
        self.root_client_path = root_client_path
 
127
        # Once we have finished waiting for all clients, etc. We set
 
128
        # _fully_stopped
 
129
        self._fully_stopped = threading.Event()
96
130
 
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
100
 
        # later stopped).
 
131
    def _backing_urls(self):
101
132
        # There are three interesting urls:
102
133
        # The URL the server can be contacted on. (e.g. bzr://host/)
103
134
        # The URL that a commit done on the same machine as the server will
104
135
        # have within the servers space. (e.g. file:///home/user/source)
105
136
        # The URL that will be given to other hooks in the same process -
106
 
        # the URL of the backing transport itself. (e.g. chroot+:///)
 
137
        # the URL of the backing transport itself. (e.g. filtered-36195:///)
107
138
        # We need all three because:
108
139
        #  * other machines see the first
109
140
        #  * local commits on this machine should be able to be mapped to
113
144
        # The latter two urls are different aliases to the servers url,
114
145
        # so we group those in a list - as there might be more aliases
115
146
        # in the future.
116
 
        backing_urls = [self.backing_transport.base]
 
147
        urls = [self.backing_transport.base]
117
148
        try:
118
 
            backing_urls.append(self.backing_transport.external_url())
 
149
            urls.append(self.backing_transport.external_url())
119
150
        except errors.InProcessTransport:
120
151
            pass
 
152
        return urls
 
153
 
 
154
    def run_server_started_hooks(self, backing_urls=None):
 
155
        if backing_urls is None:
 
156
            backing_urls = self._backing_urls()
121
157
        for hook in SmartTCPServer.hooks['server_started']:
122
158
            hook(backing_urls, self.get_url())
123
159
        for hook in SmartTCPServer.hooks['server_started_ex']:
124
160
            hook(backing_urls, self)
 
161
 
 
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())
 
167
 
 
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()
 
174
 
 
175
    def _wait_for_clients_to_disconnect(self):
 
176
        self._poll_active_connections()
 
177
        if not self._active_connections:
 
178
            return
 
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:
 
183
            now = self._timer()
 
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)
 
189
 
 
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
 
202
        # later stopped).
 
203
        self.run_server_started_hooks()
125
204
        self._started.set()
126
205
        try:
127
206
            try:
133
212
                        pass
134
213
                    except self._socket_error, e:
135
214
                        # if the socket is closed by stop_background_thread
136
 
                        # we might get a EBADF here, any other socket errors
137
 
                        # should get logged.
138
 
                        if e.args[0] != errno.EBADF:
139
 
                            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
 
217
                        # logged.
 
218
                        if e.args[0] not in (errno.EBADF, errno.EINTR):
 
219
                            trace.warning(gettext("listening socket error: %s")
 
220
                                          % (e,))
140
221
                    else:
 
222
                        if self._should_terminate:
 
223
                            conn.close()
 
224
                            break
141
225
                        self.serve_conn(conn, thread_name_suffix)
 
226
                    # Cleanout any threads that have finished processing.
 
227
                    self._poll_active_connections()
142
228
            except KeyboardInterrupt:
143
229
                # dont log when CTRL-C'd.
144
230
                raise
146
232
                trace.report_exception(sys.exc_info(), sys.stderr)
147
233
                raise
148
234
        finally:
149
 
            self._stopped.set()
150
235
            try:
151
236
                # ensure the server socket is closed.
152
237
                self._server_socket.close()
153
238
            except self._socket_error:
154
239
                # ignore errors on close
155
240
                pass
156
 
            for hook in SmartTCPServer.hooks['server_stopped']:
157
 
                hook(backing_urls, self.get_url())
 
241
            self._stopped.set()
 
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()
158
247
 
159
248
    def get_url(self):
160
249
        """Return the url of the server"""
161
 
        return "bzr://%s:%d/" % self._sockname
 
250
        return "bzr://%s:%s/" % (self._sockname[0], self._sockname[1])
 
251
 
 
252
    def _make_handler(self, conn):
 
253
        return medium.SmartServerSocketStreamMedium(
 
254
            conn, self.backing_transport, self.root_client_path,
 
255
            timeout=self._client_timeout)
 
256
 
 
257
    def _poll_active_connections(self, timeout=0.0):
 
258
        """Check to see if any active connections have finished.
 
259
 
 
260
        This will iterate through self._active_connections, and update any
 
261
        connections that are finished.
 
262
 
 
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.
 
265
        :return: None
 
266
        """
 
267
        still_active = []
 
268
        for handler, thread in self._active_connections:
 
269
            thread.join(timeout)
 
270
            if thread.isAlive():
 
271
                still_active.append((handler, thread))
 
272
        self._active_connections = still_active
162
273
 
163
274
    def serve_conn(self, conn, thread_name_suffix):
164
275
        # For WIN32, where the timeout value from the listening socket
165
276
        # propagates to the newly accepted socket.
166
277
        conn.setblocking(True)
167
278
        conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
168
 
        handler = medium.SmartServerSocketStreamMedium(
169
 
            conn, self.backing_transport, self.root_client_path)
170
279
        thread_name = 'smart-server-child' + thread_name_suffix
 
280
        handler = self._make_handler(conn)
171
281
        connection_thread = threading.Thread(
172
282
            None, handler.serve, name=thread_name)
 
283
        self._active_connections.append((handler, connection_thread))
173
284
        connection_thread.setDaemon(True)
174
285
        connection_thread.start()
 
286
        return connection_thread
175
287
 
176
288
    def start_background_thread(self, thread_name_suffix=''):
177
289
        self._started.clear()
217
329
        These are all empty initially, because by default nothing should get
218
330
        notified.
219
331
        """
220
 
        Hooks.__init__(self)
221
 
        self.create_hook(HookPoint('server_started',
 
332
        Hooks.__init__(self, "bzrlib.smart.server", "SmartTCPServer.hooks")
 
333
        self.add_hook('server_started',
222
334
            "Called by the bzr server when it starts serving a directory. "
223
335
            "server_started is called with (backing urls, public url), "
224
336
            "where backing_url is a list of URLs giving the "
225
337
            "server-specific directory locations, and public_url is the "
226
 
            "public URL for the directory being served.", (0, 16), None))
227
 
        self.create_hook(HookPoint('server_started_ex',
 
338
            "public URL for the directory being served.", (0, 16))
 
339
        self.add_hook('server_started_ex',
228
340
            "Called by the bzr server when it starts serving a directory. "
229
341
            "server_started is called with (backing_urls, server_obj).",
230
 
            (1, 17), None))
231
 
        self.create_hook(HookPoint('server_stopped',
 
342
            (1, 17))
 
343
        self.add_hook('server_stopped',
232
344
            "Called by the bzr server when it stops serving a directory. "
233
345
            "server_stopped is called with the same parameters as the "
234
 
            "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))
235
352
 
236
353
SmartTCPServer.hooks = SmartServerHooks()
237
354
 
238
355
 
239
 
class SmartTCPServer_for_testing(SmartTCPServer):
240
 
    """Server suitable for use by transport tests.
241
 
 
242
 
    This server is backed by the process's cwd.
243
 
    """
244
 
 
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
249
 
 
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())
253
 
 
254
 
    def setUp(self, backing_transport_server=None,
255
 
              client_path_extra='/extra/'):
256
 
        """Set up server for testing.
257
 
 
258
 
        :param backing_transport_server: backing server to use.  If not
259
 
            specified, a LocalURLServer at the current working directory will
260
 
            be used.
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.
267
 
        """
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)
281
 
 
282
 
    def tearDown(self):
283
 
        self.stop_background_thread()
284
 
        self.chroot_server.tearDown()
285
 
 
286
 
    def get_url(self):
287
 
        url = super(SmartTCPServer_for_testing, self).get_url()
288
 
        return url[:-1] + self.client_path_extra
289
 
 
290
 
    def get_bogus_url(self):
291
 
        """Return a URL which will fail to connect"""
292
 
        return 'bzr://127.0.0.1:1/'
293
 
 
294
 
 
295
 
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
296
 
    """Get a readonly server for testing."""
297
 
 
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)
302
 
 
303
 
 
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).
307
 
    """
308
 
 
309
 
    def get_url(self):
310
 
        url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
311
 
        url = 'bzr-v2://' + url[len('bzr://'):]
312
 
        return url
313
 
 
314
 
 
315
 
class ReadonlySmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing_v2_only):
316
 
    """Get a readonly server for testing."""
317
 
 
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)
322
 
 
323
 
 
324
356
def _local_path_for_transport(transport):
325
357
    """Return a local path for transport, if reasonably possible.
326
358
    
386
418
        """Chroot transport, and decorate with userdir expander."""
387
419
        self.base_path = self.get_base_path(transport)
388
420
        chroot_server = chroot.ChrootServer(transport)
389
 
        chroot_server.setUp()
390
 
        self.cleanups.append(chroot_server.tearDown)
391
 
        transport = get_transport(chroot_server.get_url())
 
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())
392
424
        if self.base_path is not None:
393
425
            # Decorate the server's backing transport with a filter that can
394
426
            # expand homedirs.
395
427
            expand_userdirs = self._make_expand_userdirs_filter(transport)
396
 
            expand_userdirs.setUp()
397
 
            self.cleanups.append(expand_userdirs.tearDown)
398
 
            transport = get_transport(expand_userdirs.get_url())
 
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())
399
431
        self.transport = transport
400
432
 
401
 
    def _make_smart_server(self, host, port, inet):
 
433
    def _get_stdin_stdout(self):
 
434
        return sys.stdin, sys.stdout
 
435
 
 
436
    def _make_smart_server(self, host, port, inet, timeout):
 
437
        if timeout is None:
 
438
            c = config.GlobalStack()
 
439
            timeout = c.get('serve.client_timeout')
402
440
        if inet:
 
441
            stdin, stdout = self._get_stdin_stdout()
403
442
            smart_server = medium.SmartServerPipeStreamMedium(
404
 
                sys.stdin, sys.stdout, self.transport)
 
443
                stdin, stdout, self.transport, timeout=timeout)
405
444
        else:
406
445
            if host is None:
407
446
                host = medium.BZR_DEFAULT_INTERFACE
408
447
            if port is None:
409
448
                port = medium.BZR_DEFAULT_PORT
410
 
            smart_server = SmartTCPServer(self.transport, host=host, port=port)
411
 
            trace.note('listening on port: %s' % smart_server.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)
412
453
        self.smart_server = smart_server
413
454
 
414
455
    def _change_globals(self):
425
466
        self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
426
467
        ui.ui_factory = ui.SilentUIFactory()
427
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)
428
473
 
429
 
    def set_up(self, transport, host, port, inet):
 
474
    def set_up(self, transport, host, port, inet, timeout):
430
475
        self._make_backing_transport(transport)
431
 
        self._make_smart_server(host, port, inet)
 
476
        self._make_smart_server(host, port, inet, timeout)
432
477
        self._change_globals()
433
478
 
434
479
    def tear_down(self):
436
481
            cleanup()
437
482
 
438
483
 
439
 
def serve_bzr(transport, host=None, port=None, inet=False):
 
484
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
440
485
    """This is the default implementation of 'bzr serve'.
441
 
    
 
486
 
442
487
    It creates a TCP or pipe smart server on 'transport, and runs it.  The
443
488
    transport will be decorated with a chroot and pathfilter (using
444
489
    os.path.expanduser).
445
490
    """
446
491
    bzr_server = BzrServerFactory()
447
492
    try:
448
 
        bzr_server.set_up(transport, host, port, inet)
 
493
        bzr_server.set_up(transport, host, port, inet, timeout)
449
494
        bzr_server.smart_server.serve()
 
495
    except:
 
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:
 
500
            raise
450
501
    finally:
451
502
        bzr_server.tear_down()
452