/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 breezy/bzr/smart/server.py

  • Committer: Breezy landing bot
  • Author(s): Jelmer Vernooij
  • Date: 2019-10-13 17:31:55 UTC
  • mfrom: (7397.4.9 remove-unused)
  • Revision ID: breezy.the.bot@gmail.com-20191013173155-yoiokny4mknxb3um
Remove Tree.has_id.

Merged from https://code.launchpad.net/~jelmer/brz/remove-unused/+merge/373320

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/bzr/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
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
26
 
from bzrlib import (
 
28
from ...hooks import Hooks
 
29
from ... import (
27
30
    errors,
28
31
    trace,
29
 
    transport,
 
32
    transport as _mod_transport,
30
33
)
31
 
from bzrlib.lazy_import import lazy_import
 
34
from ...i18n import gettext
 
35
from ...lazy_import import lazy_import
32
36
lazy_import(globals(), """
33
 
from bzrlib.smart import medium
34
 
from bzrlib.transport import (
 
37
from breezy.bzr.smart import (
 
38
    medium,
 
39
    signals,
 
40
    )
 
41
from breezy.transport import (
35
42
    chroot,
36
 
    get_transport,
37
43
    pathfilter,
38
44
    )
39
 
from bzrlib import (
 
45
from breezy 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
72
102
        self._socket_error = socket_error
73
103
        self._socket_timeout = socket_timeout
74
104
        addrs = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
75
 
            socket.SOCK_STREAM, 0, socket.AI_PASSIVE)[0]
 
105
                                   socket.SOCK_STREAM, 0, socket.AI_PASSIVE)[0]
76
106
 
77
107
        (family, socktype, proto, canonname, sockaddr) = addrs
78
108
 
80
110
        # SO_REUSERADDR has a different meaning on Windows
81
111
        if sys.platform != 'win32':
82
112
            self._server_socket.setsockopt(socket.SOL_SOCKET,
83
 
                socket.SO_REUSEADDR, 1)
 
113
                                           socket.SO_REUSEADDR, 1)
84
114
        try:
85
115
            self._server_socket.bind(sockaddr)
86
 
        except self._socket_error, message:
 
116
        except self._socket_error as message:
87
117
            raise errors.CannotBindAddress(host, port, message)
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:
131
210
                    except self._socket_timeout:
132
211
                        # just check if we're asked to stop
133
212
                        pass
134
 
                    except self._socket_error, e:
 
213
                    except self._socket_error as 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:
141
222
                        if self._should_terminate:
 
223
                            conn.close()
142
224
                            break
143
225
                        self.serve_conn(conn, thread_name_suffix)
 
226
                    # Cleanout any threads that have finished processing.
 
227
                    self._poll_active_connections()
144
228
            except KeyboardInterrupt:
145
229
                # dont log when CTRL-C'd.
146
230
                raise
147
 
            except Exception, e:
 
231
            except Exception as e:
148
232
                trace.report_exception(sys.exc_info(), sys.stderr)
149
233
                raise
150
234
        finally:
151
 
            self._stopped.set()
152
235
            try:
153
236
                # ensure the server socket is closed.
154
237
                self._server_socket.close()
155
238
            except self._socket_error:
156
239
                # ignore errors on close
157
240
                pass
158
 
            for hook in SmartTCPServer.hooks['server_stopped']:
159
 
                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()
160
247
 
161
248
    def get_url(self):
162
249
        """Return the url of the server"""
163
 
        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
164
273
 
165
274
    def serve_conn(self, conn, thread_name_suffix):
166
275
        # For WIN32, where the timeout value from the listening socket
167
276
        # propagates to the newly accepted socket.
168
277
        conn.setblocking(True)
169
278
        conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
170
 
        handler = medium.SmartServerSocketStreamMedium(
171
 
            conn, self.backing_transport, self.root_client_path)
172
279
        thread_name = 'smart-server-child' + thread_name_suffix
 
280
        handler = self._make_handler(conn)
173
281
        connection_thread = threading.Thread(
174
282
            None, handler.serve, name=thread_name)
 
283
        self._active_connections.append((handler, connection_thread))
175
284
        connection_thread.setDaemon(True)
176
285
        connection_thread.start()
 
286
        return connection_thread
177
287
 
178
288
    def start_background_thread(self, thread_name_suffix=''):
179
289
        self._started.clear()
180
290
        self._server_thread = threading.Thread(None,
181
 
                self.serve, args=(thread_name_suffix,),
182
 
                name='server-' + self.get_url())
 
291
                                               self.serve, args=(
 
292
                                                   thread_name_suffix,),
 
293
                                               name='server-' + self.get_url())
183
294
        self._server_thread.setDaemon(True)
184
295
        self._server_thread.start()
185
296
        self._started.wait()
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',
224
 
            "Called by the bzr server when it starts serving a directory. "
225
 
            "server_started is called with (backing urls, public url), "
226
 
            "where backing_url is a list of URLs giving the "
227
 
            "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',
230
 
            "Called by the bzr server when it starts serving a directory. "
231
 
            "server_started is called with (backing_urls, server_obj).",
232
 
            (1, 17), None))
233
 
        self.create_hook(HookPoint('server_stopped',
234
 
            "Called by the bzr server when it stops serving a directory. "
235
 
            "server_stopped is called with the same parameters as the "
236
 
            "server_started hook: (backing_urls, public_url).", (0, 16), None))
 
333
        Hooks.__init__(self, "breezy.bzr.smart.server", "SmartTCPServer.hooks")
 
334
        self.add_hook('server_started',
 
335
                      "Called by the bzr server when it starts serving a directory. "
 
336
                      "server_started is called with (backing urls, public url), "
 
337
                      "where backing_url is a list of URLs giving the "
 
338
                      "server-specific directory locations, and public_url is the "
 
339
                      "public URL for the directory being served.", (0, 16))
 
340
        self.add_hook('server_started_ex',
 
341
                      "Called by the bzr server when it starts serving a directory. "
 
342
                      "server_started is called with (backing_urls, server_obj).",
 
343
                      (1, 17))
 
344
        self.add_hook('server_stopped',
 
345
                      "Called by the bzr server when it stops serving a directory. "
 
346
                      "server_stopped is called with the same parameters as the "
 
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))
 
353
 
237
354
 
238
355
SmartTCPServer.hooks = SmartServerHooks()
239
356
 
240
357
 
241
358
def _local_path_for_transport(transport):
242
359
    """Return a local path for transport, if reasonably possible.
243
 
    
 
360
 
244
361
    This function works even if transport's url has a "readonly+" prefix,
245
362
    unlike local_path_from_url.
246
 
    
 
363
 
247
364
    This essentially recovers the --directory argument the user passed to "bzr
248
365
    serve" from the transport passed to serve_bzr.
249
366
    """
257
374
            base_url = base_url[len('readonly+'):]
258
375
        try:
259
376
            return urlutils.local_path_from_url(base_url)
260
 
        except errors.InvalidURL:
 
377
        except urlutils.InvalidURL:
261
378
            return None
262
379
 
263
380
 
305
422
        chroot_server = chroot.ChrootServer(transport)
306
423
        chroot_server.start_server()
307
424
        self.cleanups.append(chroot_server.stop_server)
308
 
        transport = get_transport(chroot_server.get_url())
 
425
        transport = _mod_transport.get_transport_from_url(
 
426
            chroot_server.get_url())
309
427
        if self.base_path is not None:
310
428
            # Decorate the server's backing transport with a filter that can
311
429
            # expand homedirs.
312
430
            expand_userdirs = self._make_expand_userdirs_filter(transport)
313
431
            expand_userdirs.start_server()
314
432
            self.cleanups.append(expand_userdirs.stop_server)
315
 
            transport = get_transport(expand_userdirs.get_url())
 
433
            transport = _mod_transport.get_transport_from_url(
 
434
                expand_userdirs.get_url())
316
435
        self.transport = transport
317
436
 
318
 
    def _make_smart_server(self, host, port, inet):
 
437
    def _get_stdin_stdout(self):
 
438
        if sys.version_info[0] < 3:
 
439
            return sys.stdin, sys.stdout
 
440
        else:
 
441
            return sys.stdin.buffer, sys.stdout.buffer
 
442
 
 
443
    def _make_smart_server(self, host, port, inet, timeout):
 
444
        if timeout is None:
 
445
            c = config.GlobalStack()
 
446
            timeout = c.get('serve.client_timeout')
319
447
        if inet:
 
448
            stdin, stdout = self._get_stdin_stdout()
320
449
            smart_server = medium.SmartServerPipeStreamMedium(
321
 
                sys.stdin, sys.stdout, self.transport)
 
450
                stdin, stdout, self.transport, timeout=timeout)
322
451
        else:
323
452
            if host is None:
324
453
                host = medium.BZR_DEFAULT_INTERFACE
325
454
            if port is None:
326
455
                port = medium.BZR_DEFAULT_PORT
327
 
            smart_server = SmartTCPServer(self.transport, host=host, port=port)
328
 
            trace.note('listening on port: %s' % smart_server.port)
 
456
            smart_server = SmartTCPServer(self.transport,
 
457
                                          client_timeout=timeout)
 
458
            smart_server.start_server(host, port)
 
459
            trace.note(gettext('listening on port: %s'),
 
460
                       str(smart_server.port))
329
461
        self.smart_server = smart_server
330
462
 
331
463
    def _change_globals(self):
332
 
        from bzrlib import lockdir, ui
 
464
        from breezy import lockdir, ui
333
465
        # For the duration of this server, no UI output is permitted. note
334
466
        # that this may cause problems with blackbox tests. This should be
335
467
        # changed with care though, as we dont want to use bandwidth sending
336
468
        # progress over stderr to smart server clients!
337
469
        old_factory = ui.ui_factory
338
470
        old_lockdir_timeout = lockdir._DEFAULT_TIMEOUT_SECONDS
 
471
 
339
472
        def restore_default_ui_factory_and_lockdir_timeout():
340
473
            ui.ui_factory = old_factory
341
474
            lockdir._DEFAULT_TIMEOUT_SECONDS = old_lockdir_timeout
342
475
        self.cleanups.append(restore_default_ui_factory_and_lockdir_timeout)
343
476
        ui.ui_factory = ui.SilentUIFactory()
344
477
        lockdir._DEFAULT_TIMEOUT_SECONDS = 0
345
 
 
346
 
    def set_up(self, transport, host, port, inet):
 
478
        orig = signals.install_sighup_handler()
 
479
 
 
480
        def restore_signals():
 
481
            signals.restore_sighup_handler(orig)
 
482
        self.cleanups.append(restore_signals)
 
483
 
 
484
    def set_up(self, transport, host, port, inet, timeout):
347
485
        self._make_backing_transport(transport)
348
 
        self._make_smart_server(host, port, inet)
 
486
        self._make_smart_server(host, port, inet, timeout)
349
487
        self._change_globals()
350
488
 
351
489
    def tear_down(self):
353
491
            cleanup()
354
492
 
355
493
 
356
 
def serve_bzr(transport, host=None, port=None, inet=False):
 
494
def serve_bzr(transport, host=None, port=None, inet=False, timeout=None):
357
495
    """This is the default implementation of 'bzr serve'.
358
 
    
 
496
 
359
497
    It creates a TCP or pipe smart server on 'transport, and runs it.  The
360
498
    transport will be decorated with a chroot and pathfilter (using
361
499
    os.path.expanduser).
362
500
    """
363
501
    bzr_server = BzrServerFactory()
364
502
    try:
365
 
        bzr_server.set_up(transport, host, port, inet)
 
503
        bzr_server.set_up(transport, host, port, inet, timeout)
366
504
        bzr_server.smart_server.serve()
 
505
    except:
 
506
        hook_caught_exception = False
 
507
        for hook in SmartTCPServer.hooks['server_exception']:
 
508
            hook_caught_exception = hook(sys.exc_info())
 
509
        if not hook_caught_exception:
 
510
            raise
367
511
    finally:
368
512
        bzr_server.tear_down()
369