/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/server.py

  • Committer: John Arbash Meinel
  • Date: 2011-09-23 20:24:51 UTC
  • mto: This revision was merged to the branch mainline in revision 6170.
  • Revision ID: john@arbash-meinel.com-20110923202451-5hrc3s8zmmh9tnoi
Add a nicer repr for shutting down.

Also, if we disconnect the stdin/stdout handles, it looks like we get ValueError
trying to flush them.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/server.py
1
 
# Copyright (C) 2006-2010 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
20
20
import os.path
21
21
import socket
22
22
import sys
 
23
import time
23
24
import threading
24
25
 
25
 
from bzrlib.hooks import HookPoint, Hooks
 
26
from bzrlib.hooks import Hooks
26
27
from bzrlib import (
27
28
    errors,
28
29
    trace,
29
 
    transport,
 
30
    transport as _mod_transport,
30
31
)
31
32
from bzrlib.lazy_import import lazy_import
32
33
lazy_import(globals(), """
33
 
from bzrlib.smart import medium
 
34
from bzrlib.smart import (
 
35
    medium,
 
36
    signals,
 
37
    )
34
38
from bzrlib.transport import (
35
39
    chroot,
36
 
    get_transport,
37
40
    pathfilter,
38
41
    )
39
42
from bzrlib import (
 
43
    config,
40
44
    urlutils,
41
45
    )
42
46
""")
51
55
    hooks: An instance of SmartServerHooks.
52
56
    """
53
57
 
54
 
    def __init__(self, backing_transport, host='127.0.0.1', port=0,
55
 
                 root_client_path='/'):
 
58
    # This is the timeout on the socket we use .accept() on. It is exposed here
 
59
    # so the test suite can set it faster. (It thread.interrupt_main() will not
 
60
    # fire a KeyboardInterrupt during socket.accept)
 
61
    _ACCEPT_TIMEOUT = 1.0
 
62
    _SHUTDOWN_POLL_TIMEOUT = 1.0
 
63
    _LOG_WAITING_TIMEOUT = 10.0
 
64
 
 
65
    _timer = time.time
 
66
 
 
67
    def __init__(self, backing_transport, root_client_path='/',
 
68
                 client_timeout=None):
56
69
        """Construct a new server.
57
70
 
58
71
        To actually start it running, call either start_background_thread or
59
72
        serve.
60
73
 
61
74
        :param backing_transport: The transport to serve.
 
75
        :param root_client_path: The client path that will correspond to root
 
76
            of backing_transport.
 
77
        :param client_timeout: See SmartServerSocketStreamMedium's timeout
 
78
            parameter.
 
79
        """
 
80
        self.backing_transport = backing_transport
 
81
        self.root_client_path = root_client_path
 
82
        self._client_timeout = client_timeout
 
83
        self._active_connections = []
 
84
        # This is set to indicate we want to wait for clients to finish before
 
85
        # we disconnect.
 
86
        self._gracefully_stopping = False
 
87
 
 
88
    def start_server(self, host, port):
 
89
        """Create the server listening socket.
 
90
 
62
91
        :param host: Name of the interface to listen on.
63
92
        :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
93
        """
67
94
        # let connections timeout so that we get a chance to terminate
68
95
        # Keep a reference to the exceptions we want to catch because the socket
88
115
        self._sockname = self._server_socket.getsockname()
89
116
        self.port = self._sockname[1]
90
117
        self._server_socket.listen(1)
91
 
        self._server_socket.settimeout(1)
92
 
        self.backing_transport = backing_transport
 
118
        self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
 
119
        # Once we start accept()ing connections, we set started.
93
120
        self._started = threading.Event()
 
121
        # Once we stop accept()ing connections (and are closing the socket) we
 
122
        # set _stopped
94
123
        self._stopped = threading.Event()
95
 
        self.root_client_path = root_client_path
 
124
        # Once we have finished waiting for all clients, etc. We set
 
125
        # _fully_stopped
 
126
        self._fully_stopped = threading.Event()
96
127
 
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).
 
128
    def _backing_urls(self):
101
129
        # There are three interesting urls:
102
130
        # The URL the server can be contacted on. (e.g. bzr://host/)
103
131
        # The URL that a commit done on the same machine as the server will
104
132
        # have within the servers space. (e.g. file:///home/user/source)
105
133
        # The URL that will be given to other hooks in the same process -
106
 
        # the URL of the backing transport itself. (e.g. chroot+:///)
 
134
        # the URL of the backing transport itself. (e.g. filtered-36195:///)
107
135
        # We need all three because:
108
136
        #  * other machines see the first
109
137
        #  * local commits on this machine should be able to be mapped to
113
141
        # The latter two urls are different aliases to the servers url,
114
142
        # so we group those in a list - as there might be more aliases
115
143
        # in the future.
116
 
        backing_urls = [self.backing_transport.base]
 
144
        urls = [self.backing_transport.base]
117
145
        try:
118
 
            backing_urls.append(self.backing_transport.external_url())
 
146
            urls.append(self.backing_transport.external_url())
119
147
        except errors.InProcessTransport:
120
148
            pass
 
149
        return urls
 
150
 
 
151
    def run_server_started_hooks(self, backing_urls=None):
 
152
        if backing_urls is None:
 
153
            backing_urls = self._backing_urls()
121
154
        for hook in SmartTCPServer.hooks['server_started']:
122
155
            hook(backing_urls, self.get_url())
123
156
        for hook in SmartTCPServer.hooks['server_started_ex']:
124
157
            hook(backing_urls, self)
 
158
 
 
159
    def run_server_stopped_hooks(self, backing_urls=None):
 
160
        if backing_urls is None:
 
161
            backing_urls = self._backing_urls()
 
162
        for hook in SmartTCPServer.hooks['server_stopped']:
 
163
            hook(backing_urls, self.get_url())
 
164
 
 
165
    def _stop_gracefully(self):
 
166
        # XXX: ATM, once we see the self._should_terminate we immediately exit
 
167
        #      without waiting for client threads to shut down. (I don't know
 
168
        #      if we *can* because they are marked as Daemon.)
 
169
        #      Which means that while in-theory this is a graceful shutdown,
 
170
        #      because we don't actively close the connections, etc, we don't
 
171
        #      have a good way (yet) to poll the spawned clients and
 
172
        trace.note('Requested to stop gracefully')
 
173
        self._should_terminate = True
 
174
        self._gracefully_stopping = True
 
175
        for handler, _ in self._active_connections:
 
176
            handler._stop_gracefully()
 
177
 
 
178
    def _wait_for_clients_to_disconnect(self):
 
179
        self._poll_active_connections()
 
180
        if not self._active_connections:
 
181
            return
 
182
        trace.note('Waiting for %d client(s) to finish'
 
183
                   % (len(self._active_connections),))
 
184
        t_next_log = self._timer() + self._LOG_WAITING_TIMEOUT
 
185
        while self._active_connections:
 
186
            now = self._timer()
 
187
            if now >= t_next_log:
 
188
                trace.note('Still waiting for %d client(s) to finish'
 
189
                           % (len(self._active_connections),))
 
190
                t_next_log = now + self._LOG_WAITING_TIMEOUT
 
191
            self._poll_active_connections(self._SHUTDOWN_POLL_TIMEOUT)
 
192
 
 
193
    def serve(self, thread_name_suffix=''):
 
194
        # Note: There is a temptation to do
 
195
        #       signals.register_on_hangup(id(self), self._stop_gracefully)
 
196
        #       However, that creates a temporary object which is a bound
 
197
        #       method. signals._on_sighup is a WeakKeyDictionary so it
 
198
        #       immediately gets garbage collected, because nothing else
 
199
        #       references it. Instead, we need to keep a real reference to the
 
200
        #       bound method for the lifetime of the serve() function.
 
201
        stop_gracefully = self._stop_gracefully
 
202
        signals.register_on_hangup(id(self), stop_gracefully)
 
203
        self._should_terminate = False
 
204
        # for hooks we are letting code know that a server has started (and
 
205
        # later stopped).
 
206
        self.run_server_started_hooks()
125
207
        self._started.set()
126
208
        try:
127
209
            try:
135
217
                        # if the socket is closed by stop_background_thread
136
218
                        # we might get a EBADF here, any other socket errors
137
219
                        # should get logged.
138
 
                        if e.args[0] != errno.EBADF:
 
220
                        if e.args[0] not in (errno.EBADF, errno.EINTR):
139
221
                            trace.warning("listening socket error: %s", e)
140
222
                    else:
141
223
                        if self._should_terminate:
 
224
                            conn.close()
142
225
                            break
143
226
                        self.serve_conn(conn, thread_name_suffix)
 
227
                    # Cleanout any threads that have finished processing.
 
228
                    self._poll_active_connections()
144
229
            except KeyboardInterrupt:
145
230
                # dont log when CTRL-C'd.
146
231
                raise
148
233
                trace.report_exception(sys.exc_info(), sys.stderr)
149
234
                raise
150
235
        finally:
151
 
            self._stopped.set()
152
236
            try:
153
237
                # ensure the server socket is closed.
154
238
                self._server_socket.close()
155
239
            except self._socket_error:
156
240
                # ignore errors on close
157
241
                pass
158
 
            for hook in SmartTCPServer.hooks['server_stopped']:
159
 
                hook(backing_urls, self.get_url())
 
242
            self._stopped.set()
 
243
            signals.unregister_on_hangup(id(self))
 
244
            self.run_server_stopped_hooks()
 
245
        if self._gracefully_stopping:
 
246
            self._wait_for_clients_to_disconnect()
 
247
        self._fully_stopped.set()
160
248
 
161
249
    def get_url(self):
162
250
        """Return the url of the server"""
163
 
        return "bzr://%s:%d/" % self._sockname
 
251
        return "bzr://%s:%s/" % (self._sockname[0], self._sockname[1])
 
252
 
 
253
    def _make_handler(self, conn):
 
254
        return medium.SmartServerSocketStreamMedium(
 
255
            conn, self.backing_transport, self.root_client_path,
 
256
            timeout=self._client_timeout)
 
257
 
 
258
    def _poll_active_connections(self, timeout=0.0):
 
259
        """Check to see if any active connections have finished.
 
260
 
 
261
        This will iterate through self._active_connections, and update any
 
262
        connections that are finished.
 
263
 
 
264
        :param timeout: The timeout to pass to thread.join(). By default, we
 
265
            set it to 0, so that we don't hang if threads are not done yet.
 
266
        :return: None
 
267
        """
 
268
        still_active = []
 
269
        for handler, thread in self._active_connections:
 
270
            thread.join(timeout)
 
271
            if thread.isAlive():
 
272
                still_active.append((handler, thread))
 
273
        self._active_connections = still_active
164
274
 
165
275
    def serve_conn(self, conn, thread_name_suffix):
166
276
        # For WIN32, where the timeout value from the listening socket
167
277
        # propagates to the newly accepted socket.
168
278
        conn.setblocking(True)
169
279
        conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
170
 
        handler = medium.SmartServerSocketStreamMedium(
171
 
            conn, self.backing_transport, self.root_client_path)
172
280
        thread_name = 'smart-server-child' + thread_name_suffix
 
281
        handler = self._make_handler(conn)
173
282
        connection_thread = threading.Thread(
174
283
            None, handler.serve, name=thread_name)
 
284
        self._active_connections.append((handler, connection_thread))
175
285
        connection_thread.setDaemon(True)
176
286
        connection_thread.start()
 
287
        return connection_thread
177
288
 
178
289
    def start_background_thread(self, thread_name_suffix=''):
179
290
        self._started.clear()
219
330
        These are all empty initially, because by default nothing should get
220
331
        notified.
221
332
        """
222
 
        Hooks.__init__(self)
223
 
        self.create_hook(HookPoint('server_started',
 
333
        Hooks.__init__(self, "bzrlib.smart.server", "SmartTCPServer.hooks")
 
334
        self.add_hook('server_started',
224
335
            "Called by the bzr server when it starts serving a directory. "
225
336
            "server_started is called with (backing urls, public url), "
226
337
            "where backing_url is a list of URLs giving the "
227
338
            "server-specific directory locations, and public_url is the "
228
 
            "public URL for the directory being served.", (0, 16), None))
229
 
        self.create_hook(HookPoint('server_started_ex',
 
339
            "public URL for the directory being served.", (0, 16))
 
340
        self.add_hook('server_started_ex',
230
341
            "Called by the bzr server when it starts serving a directory. "
231
342
            "server_started is called with (backing_urls, server_obj).",
232
 
            (1, 17), None))
233
 
        self.create_hook(HookPoint('server_stopped',
 
343
            (1, 17))
 
344
        self.add_hook('server_stopped',
234
345
            "Called by the bzr server when it stops serving a directory. "
235
346
            "server_stopped is called with the same parameters as the "
236
 
            "server_started hook: (backing_urls, public_url).", (0, 16), None))
 
347
            "server_started hook: (backing_urls, public_url).", (0, 16))
 
348
        self.add_hook('server_exception',
 
349
            "Called by the bzr server when an exception occurs. "
 
350
            "server_exception is called with the sys.exc_info() tuple "
 
351
            "return true for the hook if the exception has been handled, "
 
352
            "in which case the server will exit normally.", (2, 4))
237
353
 
238
354
SmartTCPServer.hooks = SmartServerHooks()
239
355
 
305
421
        chroot_server = chroot.ChrootServer(transport)
306
422
        chroot_server.start_server()
307
423
        self.cleanups.append(chroot_server.stop_server)
308
 
        transport = get_transport(chroot_server.get_url())
 
424
        transport = _mod_transport.get_transport_from_url(chroot_server.get_url())
309
425
        if self.base_path is not None:
310
426
            # Decorate the server's backing transport with a filter that can
311
427
            # expand homedirs.
312
428
            expand_userdirs = self._make_expand_userdirs_filter(transport)
313
429
            expand_userdirs.start_server()
314
430
            self.cleanups.append(expand_userdirs.stop_server)
315
 
            transport = get_transport(expand_userdirs.get_url())
 
431
            transport = _mod_transport.get_transport_from_url(expand_userdirs.get_url())
316
432
        self.transport = transport
317
433
 
318
 
    def _make_smart_server(self, host, port, inet):
 
434
    def _get_stdin_stdout(self):
 
435
        return sys.stdin, sys.stdout
 
436
 
 
437
    def _make_smart_server(self, host, port, inet, timeout):
 
438
        if timeout is None:
 
439
            c = config.GlobalStack()
 
440
            timeout = c.get('serve.client_timeout')
319
441
        if inet:
 
442
            stdin, stdout = self._get_stdin_stdout()
320
443
            smart_server = medium.SmartServerPipeStreamMedium(
321
 
                sys.stdin, sys.stdout, self.transport)
 
444
                stdin, stdout, self.transport, timeout=timeout)
322
445
        else:
323
446
            if host is None:
324
447
                host = medium.BZR_DEFAULT_INTERFACE
325
448
            if port is None:
326
449
                port = medium.BZR_DEFAULT_PORT
327
 
            smart_server = SmartTCPServer(self.transport, host=host, port=port)
 
450
            smart_server = SmartTCPServer(self.transport,
 
451
                                          client_timeout=timeout)
 
452
            smart_server.start_server(host, port)
328
453
            trace.note('listening on port: %s' % smart_server.port)
329
454
        self.smart_server = smart_server
330
455
 
342
467
        self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
343
468
        ui.ui_factory = ui.SilentUIFactory()
344
469
        lockdir._DEFAULT_TIMEOUT_SECONDS = 0
 
470
        orig = signals.install_sighup_handler()
 
471
        def restore_signals():
 
472
            signals.restore_sighup_handler(orig)
 
473
        self.cleanups.append(restore_signals)
345
474
 
346
 
    def set_up(self, transport, host, port, inet):
 
475
    def set_up(self, transport, host, port, inet, timeout):
347
476
        self._make_backing_transport(transport)
348
 
        self._make_smart_server(host, port, inet)
 
477
        self._make_smart_server(host, port, inet, timeout)
349
478
        self._change_globals()
350
479
 
351
480
    def tear_down(self):
353
482
            cleanup()
354
483
 
355
484
 
356
 
def serve_bzr(transport, host=None, port=None, inet=False):
 
485
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
357
486
    """This is the default implementation of 'bzr serve'.
358
 
    
 
487
 
359
488
    It creates a TCP or pipe smart server on 'transport, and runs it.  The
360
489
    transport will be decorated with a chroot and pathfilter (using
361
490
    os.path.expanduser).
362
491
    """
363
492
    bzr_server = BzrServerFactory()
364
493
    try:
365
 
        bzr_server.set_up(transport, host, port, inet)
 
494
        bzr_server.set_up(transport, host, port, inet, timeout)
366
495
        bzr_server.smart_server.serve()
 
496
    except:
 
497
        hook_caught_exception = False
 
498
        for hook in SmartTCPServer.hooks['server_exception']:
 
499
            hook_caught_exception = hook(sys.exc_info())
 
500
        if not hook_caught_exception:
 
501
            raise
367
502
    finally:
368
503
        bzr_server.tear_down()
369