← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Vincent Ladeuil
  • Date: 2017-01-17 13:48:10 UTC
  • mfrom: (6615.3.6 merges)
  • mto: This revision was merged to the branch mainline in revision 6620.
  • Revision ID: v.ladeuil+lp@free.fr-20170117134810-j9p3lidfy6pfyfsc
Merge 2.7, resolving conflicts

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/medium.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
24
24
bzrlib/transport/smart/__init__.py.
25
25
"""
26
26
 
 
27
from __future__ import absolute_import
 
28
 
 
29
import errno
27
30
import os
28
31
import sys
29
 
import urllib
 
32
import time
30
33
 
31
34
import bzrlib
32
35
from bzrlib.lazy_import import lazy_import
33
36
lazy_import(globals(), """
 
37
import select
34
38
import socket
35
39
import thread
36
40
import weakref
38
42
from bzrlib import (
39
43
    debug,
40
44
    errors,
41
 
    symbol_versioning,
42
45
    trace,
 
46
    transport,
43
47
    ui,
44
48
    urlutils,
45
49
    )
46
 
from bzrlib.smart import client, protocol, request, vfs
 
50
from bzrlib.i18n import gettext
 
51
from bzrlib.smart import client, protocol, request, signals, vfs
47
52
from bzrlib.transport import ssh
48
53
""")
49
54
from bzrlib import osutils
176
181
        ui.ui_factory.report_transport_activity(self, bytes, direction)
177
182
 
178
183
 
 
184
_bad_file_descriptor = (errno.EBADF,)
 
185
if sys.platform == 'win32':
 
186
    # Given on Windows if you pass a closed socket to select.select. Probably
 
187
    # also given if you pass a file handle to select.
 
188
    WSAENOTSOCK = 10038
 
189
    _bad_file_descriptor += (WSAENOTSOCK,)
 
190
 
 
191
 
179
192
class SmartServerStreamMedium(SmartMedium):
180
193
    """Handles smart commands coming over a stream.
181
194
 
194
207
        the stream.  See also the _push_back method.
195
208
    """
196
209
 
197
 
    def __init__(self, backing_transport, root_client_path='/'):
 
210
    _timer = time.time
 
211
 
 
212
    def __init__(self, backing_transport, root_client_path='/', timeout=None):
198
213
        """Construct new server.
199
214
 
200
215
        :param backing_transport: Transport for the directory served.
203
218
        self.backing_transport = backing_transport
204
219
        self.root_client_path = root_client_path
205
220
        self.finished = False
 
221
        if timeout is None:
 
222
            raise AssertionError('You must supply a timeout.')
 
223
        self._client_timeout = timeout
 
224
        self._client_poll_timeout = min(timeout / 10.0, 1.0)
206
225
        SmartMedium.__init__(self)
207
226
 
208
227
    def serve(self):
214
233
            while not self.finished:
215
234
                server_protocol = self._build_protocol()
216
235
                self._serve_one_request(server_protocol)
 
236
        except errors.ConnectionTimeout, e:
 
237
            trace.note('%s' % (e,))
 
238
            trace.log_exception_quietly()
 
239
            self._disconnect_client()
 
240
            # We reported it, no reason to make a big fuss.
 
241
            return
217
242
        except Exception, e:
218
243
            stderr.write("%s terminating on exception %s\n" % (self, e))
219
244
            raise
 
245
        self._disconnect_client()
 
246
 
 
247
    def _stop_gracefully(self):
 
248
        """When we finish this message, stop looking for more."""
 
249
        trace.mutter('Stopping %s' % (self,))
 
250
        self.finished = True
 
251
 
 
252
    def _disconnect_client(self):
 
253
        """Close the current connection. We stopped due to a timeout/etc."""
 
254
        # The default implementation is a no-op, because that is all we used to
 
255
        # do when disconnecting from a client. I suppose we never had the
 
256
        # *server* initiate a disconnect, before
 
257
 
 
258
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
259
        """Wait for more bytes to be read, but timeout if none available.
 
260
 
 
261
        This allows us to detect idle connections, and stop trying to read from
 
262
        them, without setting the socket itself to non-blocking. This also
 
263
        allows us to specify when we watch for idle timeouts.
 
264
 
 
265
        :return: Did we timeout? (True if we timed out, False if there is data
 
266
            to be read)
 
267
        """
 
268
        raise NotImplementedError(self._wait_for_bytes_with_timeout)
220
269
 
221
270
    def _build_protocol(self):
222
271
        """Identifies the version of the incoming request, and returns an
227
276
 
228
277
        :returns: a SmartServerRequestProtocol.
229
278
        """
 
279
        self._wait_for_bytes_with_timeout(self._client_timeout)
 
280
        if self.finished:
 
281
            # We're stopping, so don't try to do any more work
 
282
            return None
230
283
        bytes = self._get_line()
231
284
        protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
232
285
        protocol = protocol_factory(
234
287
        protocol.accept_bytes(unused_bytes)
235
288
        return protocol
236
289
 
 
290
    def _wait_on_descriptor(self, fd, timeout_seconds):
 
291
        """select() on a file descriptor, waiting for nonblocking read()
 
292
 
 
293
        This will raise a ConnectionTimeout exception if we do not get a
 
294
        readable handle before timeout_seconds.
 
295
        :return: None
 
296
        """
 
297
        t_end = self._timer() + timeout_seconds
 
298
        poll_timeout = min(timeout_seconds, self._client_poll_timeout)
 
299
        rs = xs = None
 
300
        while not rs and not xs and self._timer() < t_end:
 
301
            if self.finished:
 
302
                return
 
303
            try:
 
304
                rs, _, xs = select.select([fd], [], [fd], poll_timeout)
 
305
            except (select.error, socket.error) as e:
 
306
                err = getattr(e, 'errno', None)
 
307
                if err is None and getattr(e, 'args', None) is not None:
 
308
                    # select.error doesn't have 'errno', it just has args[0]
 
309
                    err = e.args[0]
 
310
                if err in _bad_file_descriptor:
 
311
                    return # Not a socket indicates read() will fail
 
312
                elif err == errno.EINTR:
 
313
                    # Interrupted, keep looping.
 
314
                    continue
 
315
                raise
 
316
        if rs or xs:
 
317
            return
 
318
        raise errors.ConnectionTimeout('disconnecting client after %.1f seconds'
 
319
                                       % (timeout_seconds,))
 
320
 
237
321
    def _serve_one_request(self, protocol):
238
322
        """Read one request from input, process, send back a response.
239
323
 
240
324
        :param protocol: a SmartServerRequestProtocol.
241
325
        """
 
326
        if protocol is None:
 
327
            return
242
328
        try:
243
329
            self._serve_one_request_unguarded(protocol)
244
330
        except KeyboardInterrupt:
260
346
 
261
347
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
262
348
 
263
 
    def __init__(self, sock, backing_transport, root_client_path='/'):
 
349
    def __init__(self, sock, backing_transport, root_client_path='/',
 
350
                 timeout=None):
264
351
        """Constructor.
265
352
 
266
353
        :param sock: the socket the server will read from.  It will be put
267
354
            into blocking mode.
268
355
        """
269
356
        SmartServerStreamMedium.__init__(
270
 
            self, backing_transport, root_client_path=root_client_path)
 
357
            self, backing_transport, root_client_path=root_client_path,
 
358
            timeout=timeout)
271
359
        sock.setblocking(True)
272
360
        self.socket = sock
 
361
        # Get the getpeername now, as we might be closed later when we care.
 
362
        try:
 
363
            self._client_info = sock.getpeername()
 
364
        except socket.error:
 
365
            self._client_info = '<unknown>'
 
366
 
 
367
    def __str__(self):
 
368
        return '%s(client=%s)' % (self.__class__.__name__, self._client_info)
 
369
 
 
370
    def __repr__(self):
 
371
        return '%s.%s(client=%s)' % (self.__module__, self.__class__.__name__,
 
372
            self._client_info)
273
373
 
274
374
    def _serve_one_request_unguarded(self, protocol):
275
375
        while protocol.next_read_size():
284
384
 
285
385
        self._push_back(protocol.unused_data)
286
386
 
 
387
    def _disconnect_client(self):
 
388
        """Close the current connection. We stopped due to a timeout/etc."""
 
389
        self.socket.close()
 
390
 
 
391
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
392
        """Wait for more bytes to be read, but timeout if none available.
 
393
 
 
394
        This allows us to detect idle connections, and stop trying to read from
 
395
        them, without setting the socket itself to non-blocking. This also
 
396
        allows us to specify when we watch for idle timeouts.
 
397
 
 
398
        :return: None, this will raise ConnectionTimeout if we time out before
 
399
            data is available.
 
400
        """
 
401
        return self._wait_on_descriptor(self.socket, timeout_seconds)
 
402
 
287
403
    def _read_bytes(self, desired_count):
288
404
        return osutils.read_bytes_from_socket(
289
405
            self.socket, self._report_activity)
306
422
 
307
423
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
308
424
 
309
 
    def __init__(self, in_file, out_file, backing_transport):
 
425
    def __init__(self, in_file, out_file, backing_transport, timeout=None):
310
426
        """Construct new server.
311
427
 
312
428
        :param in_file: Python file from which requests can be read.
313
429
        :param out_file: Python file to write responses.
314
430
        :param backing_transport: Transport for the directory served.
315
431
        """
316
 
        SmartServerStreamMedium.__init__(self, backing_transport)
 
432
        SmartServerStreamMedium.__init__(self, backing_transport,
 
433
            timeout=timeout)
317
434
        if sys.platform == 'win32':
318
435
            # force binary mode for files
319
436
            import msvcrt
324
441
        self._in = in_file
325
442
        self._out = out_file
326
443
 
 
444
    def serve(self):
 
445
        """See SmartServerStreamMedium.serve"""
 
446
        # This is the regular serve, except it adds signal trapping for soft
 
447
        # shutdown.
 
448
        stop_gracefully = self._stop_gracefully
 
449
        signals.register_on_hangup(id(self), stop_gracefully)
 
450
        try:
 
451
            return super(SmartServerPipeStreamMedium, self).serve()
 
452
        finally:
 
453
            signals.unregister_on_hangup(id(self))
 
454
 
327
455
    def _serve_one_request_unguarded(self, protocol):
328
456
        while True:
329
457
            # We need to be careful not to read past the end of the current
342
470
                return
343
471
            protocol.accept_bytes(bytes)
344
472
 
 
473
    def _disconnect_client(self):
 
474
        self._in.close()
 
475
        self._out.flush()
 
476
        self._out.close()
 
477
 
 
478
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
479
        """Wait for more bytes to be read, but timeout if none available.
 
480
 
 
481
        This allows us to detect idle connections, and stop trying to read from
 
482
        them, without setting the socket itself to non-blocking. This also
 
483
        allows us to specify when we watch for idle timeouts.
 
484
 
 
485
        :return: None, this will raise ConnectionTimeout if we time out before
 
486
            data is available.
 
487
        """
 
488
        if (getattr(self._in, 'fileno', None) is None
 
489
            or sys.platform == 'win32'):
 
490
            # You can't select() file descriptors on Windows.
 
491
            return
 
492
        return self._wait_on_descriptor(self._in, timeout_seconds)
 
493
 
345
494
    def _read_bytes(self, desired_count):
346
495
        return self._in.read(desired_count)
347
496
 
491
640
        return self._medium._get_line()
492
641
 
493
642
 
 
643
class _VfsRefuser(object):
 
644
    """An object that refuses all VFS requests.
 
645
 
 
646
    """
 
647
 
 
648
    def __init__(self):
 
649
        client._SmartClient.hooks.install_named_hook(
 
650
            'call', self.check_vfs, 'vfs refuser')
 
651
 
 
652
    def check_vfs(self, params):
 
653
        try:
 
654
            request_method = request.request_handlers.get(params.method)
 
655
        except KeyError:
 
656
            # A method we don't know about doesn't count as a VFS method.
 
657
            return
 
658
        if issubclass(request_method, vfs.VfsRequest):
 
659
            raise errors.HpssVfsRequestNotAllowed(params.method, params.args)
 
660
 
 
661
 
494
662
class _DebugCounter(object):
495
663
    """An object that counts the HPSS calls made to each client medium.
496
664
 
543
711
        value['count'] = 0
544
712
        value['vfs_count'] = 0
545
713
        if count != 0:
546
 
            trace.note('HPSS calls: %d (%d vfs) %s',
547
 
                       count, vfs_count, medium_repr)
 
714
            trace.note(gettext('HPSS calls: {0} ({1} vfs) {2}').format(
 
715
                       count, vfs_count, medium_repr))
548
716
 
549
717
    def flush_all(self):
550
718
        for ref in list(self.counts.keys()):
551
719
            self.done(ref)
552
720
 
553
721
_debug_counter = None
 
722
_vfs_refuser = None
554
723
 
555
724
 
556
725
class SmartClientMedium(SmartMedium):
573
742
            if _debug_counter is None:
574
743
                _debug_counter = _DebugCounter()
575
744
            _debug_counter.track(self)
 
745
        if 'hpss_client_no_vfs' in debug.debug_flags:
 
746
            global _vfs_refuser
 
747
            if _vfs_refuser is None:
 
748
                _vfs_refuser = _VfsRefuser()
576
749
 
577
750
    def _is_remote_before(self, version_tuple):
578
751
        """Is it possible the remote side supports RPCs for a given version?
669
842
        """
670
843
        medium_base = urlutils.join(self.base, '/')
671
844
        rel_url = urlutils.relative_url(medium_base, transport.base)
672
 
        return urllib.unquote(rel_url)
 
845
        return urlutils.unquote(rel_url)
673
846
 
674
847
 
675
848
class SmartClientStreamMedium(SmartClientMedium):
710
883
        """
711
884
        return SmartClientStreamMediumRequest(self)
712
885
 
 
886
    def reset(self):
 
887
        """We have been disconnected, reset current state.
 
888
 
 
889
        This resets things like _current_request and connected state.
 
890
        """
 
891
        self.disconnect()
 
892
        self._current_request = None
 
893
 
713
894
 
714
895
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
715
896
    """A client medium using simple pipes.
724
905
 
725
906
    def _accept_bytes(self, bytes):
726
907
        """See SmartClientStreamMedium.accept_bytes."""
727
 
        self._writeable_pipe.write(bytes)
 
908
        try:
 
909
            self._writeable_pipe.write(bytes)
 
910
        except IOError, e:
 
911
            if e.errno in (errno.EINVAL, errno.EPIPE):
 
912
                raise errors.ConnectionReset(
 
913
                    "Error trying to write to subprocess", e)
 
914
            raise
728
915
        self._report_activity(len(bytes), 'write')
729
916
 
730
917
    def _flush(self):
731
918
        """See SmartClientStreamMedium._flush()."""
 
919
        # Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
 
920
        #       However, testing shows that even when the child process is
 
921
        #       gone, this doesn't error.
732
922
        self._writeable_pipe.flush()
733
923
 
734
924
    def _read_bytes(self, count):
753
943
 
754
944
class SmartSSHClientMedium(SmartClientStreamMedium):
755
945
    """A client medium using SSH.
756
 
    
757
 
    It delegates IO to a SmartClientSocketMedium or
 
946
 
 
947
    It delegates IO to a SmartSimplePipesClientMedium or
758
948
    SmartClientAlreadyConnectedSocketMedium (depending on platform).
759
949
    """
760
950
 
782
972
            maybe_port = ''
783
973
        else:
784
974
            maybe_port = ':%s' % self._ssh_params.port
785
 
        return "%s(%s://%s@%s%s/)" % (
 
975
        if self._ssh_params.username is None:
 
976
            maybe_user = ''
 
977
        else:
 
978
            maybe_user = '%s@' % self._ssh_params.username
 
979
        return "%s(%s://%s%s%s/)" % (
786
980
            self.__class__.__name__,
787
981
            self._scheme,
788
 
            self._ssh_params.username,
 
982
            maybe_user,
789
983
            self._ssh_params.host,
790
984
            maybe_port)
791
985
 
828
1022
            raise AssertionError(
829
1023
                "Unexpected io_kind %r from %r"
830
1024
                % (io_kind, self._ssh_connection))
 
1025
        for hook in transport.Transport.hooks["post_connect"]:
 
1026
            hook(self)
831
1027
 
832
1028
    def _flush(self):
833
1029
        """See SmartClientStreamMedium._flush()."""
847
1043
 
848
1044
class SmartClientSocketMedium(SmartClientStreamMedium):
849
1045
    """A client medium using a socket.
850
 
    
 
1046
 
851
1047
    This class isn't usable directly.  Use one of its subclasses instead.
852
1048
    """
853
1049
 
936
1132
            raise errors.ConnectionError("failed to connect to %s:%d: %s" %
937
1133
                    (self._host, port, err_msg))
938
1134
        self._connected = True
 
1135
        for hook in transport.Transport.hooks["post_connect"]:
 
1136
            hook(self)
939
1137
 
940
1138
 
941
1139
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
993
1191
        This invokes self._medium._flush to ensure all bytes are transmitted.
994
1192
        """
995
1193
        self._medium._flush()
996
 
 
997