← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/server.py

  • Committer: Patch Queue Manager
  • Date: 2016-02-01 19:13:13 UTC
  • mfrom: (6614.2.2 trunk)
  • Revision ID: pqm@pqm.ubuntu.com-20160201191313-wdfvmfff1djde6oq
(vila) Release 2.7.0 (Vincent Ladeuil)

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/server.py
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
28
from bzrlib.hooks import Hooks
28
31
    trace,
29
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
43
    pathfilter,
37
44
    )
38
45
from bzrlib import (
 
46
    config,
39
47
    urlutils,
40
48
    )
41
49
""")
50
58
    hooks: An instance of SmartServerHooks.
51
59
    """
52
60
 
53
 
    def __init__(self, backing_transport, 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):
54
72
        """Construct a new server.
55
73
 
56
74
        To actually start it running, call either start_background_thread or
59
77
        :param backing_transport: The transport to serve.
60
78
        :param root_client_path: The client path that will correspond to root
61
79
            of backing_transport.
 
80
        :param client_timeout: See SmartServerSocketStreamMedium's timeout
 
81
            parameter.
62
82
        """
63
83
        self.backing_transport = backing_transport
64
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
65
90
 
66
91
    def start_server(self, host, port):
67
92
        """Create the server listening socket.
93
118
        self._sockname = self._server_socket.getsockname()
94
119
        self.port = self._sockname[1]
95
120
        self._server_socket.listen(1)
96
 
        self._server_socket.settimeout(1)
 
121
        self._server_socket.settimeout(self._ACCEPT_TIMEOUT)
 
122
        # Once we start accept()ing connections, we set started.
97
123
        self._started = threading.Event()
 
124
        # Once we stop accept()ing connections (and are closing the socket) we
 
125
        # set _stopped
98
126
        self._stopped = threading.Event()
 
127
        # Once we have finished waiting for all clients, etc. We set
 
128
        # _fully_stopped
 
129
        self._fully_stopped = threading.Event()
99
130
 
100
131
    def _backing_urls(self):
101
132
        # There are three interesting urls:
134
165
        for hook in SmartTCPServer.hooks['server_stopped']:
135
166
            hook(backing_urls, self.get_url())
136
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
 
137
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)
138
200
        self._should_terminate = False
139
201
        # for hooks we are letting code know that a server has started (and
140
202
        # later stopped).
150
212
                        pass
151
213
                    except self._socket_error, e:
152
214
                        # if the socket is closed by stop_background_thread
153
 
                        # we might get a EBADF here, any other socket errors
154
 
                        # should get logged.
155
 
                        if e.args[0] != errno.EBADF:
156
 
                            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,))
157
221
                    else:
158
222
                        if self._should_terminate:
 
223
                            conn.close()
159
224
                            break
160
225
                        self.serve_conn(conn, thread_name_suffix)
 
226
                    # Cleanout any threads that have finished processing.
 
227
                    self._poll_active_connections()
161
228
            except KeyboardInterrupt:
162
229
                # dont log when CTRL-C'd.
163
230
                raise
165
232
                trace.report_exception(sys.exc_info(), sys.stderr)
166
233
                raise
167
234
        finally:
168
 
            self._stopped.set()
169
235
            try:
170
236
                # ensure the server socket is closed.
171
237
                self._server_socket.close()
172
238
            except self._socket_error:
173
239
                # ignore errors on close
174
240
                pass
 
241
            self._stopped.set()
 
242
            signals.unregister_on_hangup(id(self))
175
243
            self.run_server_stopped_hooks()
 
244
        if self._gracefully_stopping:
 
245
            self._wait_for_clients_to_disconnect()
 
246
        self._fully_stopped.set()
176
247
 
177
248
    def get_url(self):
178
249
        """Return the url of the server"""
179
250
        return "bzr://%s:%s/" % (self._sockname[0], self._sockname[1])
180
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
 
273
 
181
274
    def serve_conn(self, conn, thread_name_suffix):
182
275
        # For WIN32, where the timeout value from the listening socket
183
276
        # propagates to the newly accepted socket.
184
277
        conn.setblocking(True)
185
278
        conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
186
 
        handler = medium.SmartServerSocketStreamMedium(
187
 
            conn, self.backing_transport, self.root_client_path)
188
279
        thread_name = 'smart-server-child' + thread_name_suffix
 
280
        handler = self._make_handler(conn)
189
281
        connection_thread = threading.Thread(
190
282
            None, handler.serve, name=thread_name)
191
 
        # FIXME: This thread is never joined, it should at least be collected
192
 
        # somewhere so that tests that want to check for leaked threads can get
193
 
        # rid of them -- vila 20100531
 
283
        self._active_connections.append((handler, connection_thread))
194
284
        connection_thread.setDaemon(True)
195
285
        connection_thread.start()
196
286
        return connection_thread
340
430
            transport = _mod_transport.get_transport_from_url(expand_userdirs.get_url())
341
431
        self.transport = transport
342
432
 
343
 
    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')
344
440
        if inet:
 
441
            stdin, stdout = self._get_stdin_stdout()
345
442
            smart_server = medium.SmartServerPipeStreamMedium(
346
 
                sys.stdin, sys.stdout, self.transport)
 
443
                stdin, stdout, self.transport, timeout=timeout)
347
444
        else:
348
445
            if host is None:
349
446
                host = medium.BZR_DEFAULT_INTERFACE
350
447
            if port is None:
351
448
                port = medium.BZR_DEFAULT_PORT
352
 
            smart_server = SmartTCPServer(self.transport)
 
449
            smart_server = SmartTCPServer(self.transport,
 
450
                                          client_timeout=timeout)
353
451
            smart_server.start_server(host, port)
354
 
            trace.note('listening on port: %s' % smart_server.port)
 
452
            trace.note(gettext('listening on port: %s') % smart_server.port)
355
453
        self.smart_server = smart_server
356
454
 
357
455
    def _change_globals(self):
368
466
        self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
369
467
        ui.ui_factory = ui.SilentUIFactory()
370
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)
371
473
 
372
 
    def set_up(self, transport, host, port, inet):
 
474
    def set_up(self, transport, host, port, inet, timeout):
373
475
        self._make_backing_transport(transport)
374
 
        self._make_smart_server(host, port, inet)
 
476
        self._make_smart_server(host, port, inet, timeout)
375
477
        self._change_globals()
376
478
 
377
479
    def tear_down(self):
378
480
        for cleanup in reversed(self.cleanups):
379
481
            cleanup()
380
482
 
381
 
def serve_bzr(transport, host=None, port=None, inet=False):
 
483
 
 
484
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
382
485
    """This is the default implementation of 'bzr serve'.
383
 
    
 
486
 
384
487
    It creates a TCP or pipe smart server on 'transport, and runs it.  The
385
488
    transport will be decorated with a chroot and pathfilter (using
386
489
    os.path.expanduser).
387
490
    """
388
491
    bzr_server = BzrServerFactory()
389
492
    try:
390
 
        bzr_server.set_up(transport, host, port, inet)
 
493
        bzr_server.set_up(transport, host, port, inet, timeout)
391
494
        bzr_server.smart_server.serve()
392
495
    except:
393
496
        hook_caught_exception = False