1
# Copyright (C) 2006 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""The 'medium' layer for the smart servers and clients.
19
"Medium" here is the noun meaning "a means of transmission", not the adjective
20
for "the quality between big and small."
22
Media carry the bytes of the requests somehow (e.g. via TCP, wrapped in HTTP, or
23
over SSH), and pass them to and from the protocol logic. See the overview in
24
bzrlib/transport/smart/__init__.py.
36
from bzrlib.smart.protocol import (
38
SmartClientRequestProtocolOne,
39
SmartServerRequestProtocolOne,
40
SmartServerRequestProtocolTwo,
42
from bzrlib.transport import ssh
45
class SmartServerStreamMedium(object):
46
"""Handles smart commands coming over a stream.
48
The stream may be a pipe connected to sshd, or a tcp socket, or an
49
in-process fifo for testing.
51
One instance is created for each connected client; it can serve multiple
52
requests in the lifetime of the connection.
54
The server passes requests through to an underlying backing transport,
55
which will typically be a LocalTransport looking at the server's filesystem.
57
:ivar _push_back_buffer: a str of bytes that have been read from the stream
58
but not used yet, or None if there are no buffered bytes. Subclasses
59
should make sure to exhaust this buffer before reading more bytes from
60
the stream. See also the _push_back method.
63
def __init__(self, backing_transport, root_client_path='/'):
64
"""Construct new server.
66
:param backing_transport: Transport for the directory served.
68
# backing_transport could be passed to serve instead of __init__
69
self.backing_transport = backing_transport
70
self.root_client_path = root_client_path
72
self._push_back_buffer = None
74
def _push_back(self, bytes):
75
"""Return unused bytes to the medium, because they belong to the next
78
This sets the _push_back_buffer to the given bytes.
80
if self._push_back_buffer is not None:
82
"_push_back called when self._push_back_buffer is %r"
83
% (self._push_back_buffer,))
86
self._push_back_buffer = bytes
88
def _get_push_back_buffer(self):
89
if self._push_back_buffer == '':
91
'%s._push_back_buffer should never be the empty string, '
92
'which can be confused with EOF' % (self,))
93
bytes = self._push_back_buffer
94
self._push_back_buffer = None
98
"""Serve requests until the client disconnects."""
99
# Keep a reference to stderr because the sys module's globals get set to
100
# None during interpreter shutdown.
101
from sys import stderr
103
while not self.finished:
104
server_protocol = self._build_protocol()
105
self._serve_one_request(server_protocol)
107
stderr.write("%s terminating on exception %s\n" % (self, e))
110
def _build_protocol(self):
111
"""Identifies the version of the incoming request, and returns an
112
a protocol object that can interpret it.
114
If more bytes than the version prefix of the request are read, they will
115
be fed into the protocol before it is returned.
117
:returns: a SmartServerRequestProtocol.
119
# Identify the protocol version.
120
bytes = self._get_line()
121
if bytes.startswith(REQUEST_VERSION_TWO):
122
protocol_class = SmartServerRequestProtocolTwo
123
bytes = bytes[len(REQUEST_VERSION_TWO):]
125
protocol_class = SmartServerRequestProtocolOne
126
protocol = protocol_class(
127
self.backing_transport, self._write_out, self.root_client_path)
128
protocol.accept_bytes(bytes)
131
def _serve_one_request(self, protocol):
132
"""Read one request from input, process, send back a response.
134
:param protocol: a SmartServerRequestProtocol.
137
self._serve_one_request_unguarded(protocol)
138
except KeyboardInterrupt:
141
self.terminate_due_to_error()
143
def terminate_due_to_error(self):
144
"""Called when an unhandled exception from the protocol occurs."""
145
raise NotImplementedError(self.terminate_due_to_error)
147
def _get_bytes(self, desired_count):
148
"""Get some bytes from the medium.
150
:param desired_count: number of bytes we want to read.
152
raise NotImplementedError(self._get_bytes)
155
"""Read bytes from this request's response until a newline byte.
157
This isn't particularly efficient, so should only be used when the
158
expected size of the line is quite short.
160
:returns: a string of bytes ending in a newline (byte 0x0A).
164
while newline_pos == -1:
165
new_bytes = self._get_bytes(1)
168
# Ran out of bytes before receiving a complete line.
170
newline_pos = bytes.find('\n')
171
line = bytes[:newline_pos+1]
172
self._push_back(bytes[newline_pos+1:])
176
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
178
def __init__(self, sock, backing_transport, root_client_path='/'):
181
:param sock: the socket the server will read from. It will be put
184
SmartServerStreamMedium.__init__(
185
self, backing_transport, root_client_path=root_client_path)
186
sock.setblocking(True)
189
def _serve_one_request_unguarded(self, protocol):
190
while protocol.next_read_size():
191
bytes = self._get_bytes(4096)
195
protocol.accept_bytes(bytes)
197
self._push_back(protocol.excess_buffer)
199
def _get_bytes(self, desired_count):
200
if self._push_back_buffer is not None:
201
return self._get_push_back_buffer()
202
# We ignore the desired_count because on sockets it's more efficient to
204
return self.socket.recv(4096)
206
def terminate_due_to_error(self):
207
"""Called when an unhandled exception from the protocol occurs."""
208
# TODO: This should log to a server log file, but no such thing
209
# exists yet. Andrew Bennetts 2006-09-29.
213
def _write_out(self, bytes):
214
osutils.send_all(self.socket, bytes)
217
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
219
def __init__(self, in_file, out_file, backing_transport):
220
"""Construct new server.
222
:param in_file: Python file from which requests can be read.
223
:param out_file: Python file to write responses.
224
:param backing_transport: Transport for the directory served.
226
SmartServerStreamMedium.__init__(self, backing_transport)
227
if sys.platform == 'win32':
228
# force binary mode for files
230
for f in (in_file, out_file):
231
fileno = getattr(f, 'fileno', None)
233
msvcrt.setmode(fileno(), os.O_BINARY)
237
def _serve_one_request_unguarded(self, protocol):
239
bytes_to_read = protocol.next_read_size()
240
if bytes_to_read == 0:
241
# Finished serving this request.
244
bytes = self._get_bytes(bytes_to_read)
246
# Connection has been closed.
250
protocol.accept_bytes(bytes)
252
def _get_bytes(self, desired_count):
253
if self._push_back_buffer is not None:
254
return self._get_push_back_buffer()
255
return self._in.read(desired_count)
257
def terminate_due_to_error(self):
258
# TODO: This should log to a server log file, but no such thing
259
# exists yet. Andrew Bennetts 2006-09-29.
263
def _write_out(self, bytes):
264
self._out.write(bytes)
267
class SmartClientMediumRequest(object):
268
"""A request on a SmartClientMedium.
270
Each request allows bytes to be provided to it via accept_bytes, and then
271
the response bytes to be read via read_bytes.
274
request.accept_bytes('123')
275
request.finished_writing()
276
result = request.read_bytes(3)
277
request.finished_reading()
279
It is up to the individual SmartClientMedium whether multiple concurrent
280
requests can exist. See SmartClientMedium.get_request to obtain instances
281
of SmartClientMediumRequest, and the concrete Medium you are using for
282
details on concurrency and pipelining.
285
def __init__(self, medium):
286
"""Construct a SmartClientMediumRequest for the medium medium."""
287
self._medium = medium
288
# we track state by constants - we may want to use the same
289
# pattern as BodyReader if it gets more complex.
290
# valid states are: "writing", "reading", "done"
291
self._state = "writing"
293
def accept_bytes(self, bytes):
294
"""Accept bytes for inclusion in this request.
296
This method may not be be called after finished_writing() has been
297
called. It depends upon the Medium whether or not the bytes will be
298
immediately transmitted. Message based Mediums will tend to buffer the
299
bytes until finished_writing() is called.
301
:param bytes: A bytestring.
303
if self._state != "writing":
304
raise errors.WritingCompleted(self)
305
self._accept_bytes(bytes)
307
def _accept_bytes(self, bytes):
308
"""Helper for accept_bytes.
310
Accept_bytes checks the state of the request to determing if bytes
311
should be accepted. After that it hands off to _accept_bytes to do the
314
raise NotImplementedError(self._accept_bytes)
316
def finished_reading(self):
317
"""Inform the request that all desired data has been read.
319
This will remove the request from the pipeline for its medium (if the
320
medium supports pipelining) and any further calls to methods on the
321
request will raise ReadingCompleted.
323
if self._state == "writing":
324
raise errors.WritingNotComplete(self)
325
if self._state != "reading":
326
raise errors.ReadingCompleted(self)
328
self._finished_reading()
330
def _finished_reading(self):
331
"""Helper for finished_reading.
333
finished_reading checks the state of the request to determine if
334
finished_reading is allowed, and if it is hands off to _finished_reading
335
to perform the action.
337
raise NotImplementedError(self._finished_reading)
339
def finished_writing(self):
340
"""Finish the writing phase of this request.
342
This will flush all pending data for this request along the medium.
343
After calling finished_writing, you may not call accept_bytes anymore.
345
if self._state != "writing":
346
raise errors.WritingCompleted(self)
347
self._state = "reading"
348
self._finished_writing()
350
def _finished_writing(self):
351
"""Helper for finished_writing.
353
finished_writing checks the state of the request to determine if
354
finished_writing is allowed, and if it is hands off to _finished_writing
355
to perform the action.
357
raise NotImplementedError(self._finished_writing)
359
def read_bytes(self, count):
360
"""Read bytes from this requests response.
362
This method will block and wait for count bytes to be read. It may not
363
be invoked until finished_writing() has been called - this is to ensure
364
a message-based approach to requests, for compatibility with message
365
based mediums like HTTP.
367
if self._state == "writing":
368
raise errors.WritingNotComplete(self)
369
if self._state != "reading":
370
raise errors.ReadingCompleted(self)
371
return self._read_bytes(count)
373
def _read_bytes(self, count):
374
"""Helper for read_bytes.
376
read_bytes checks the state of the request to determing if bytes
377
should be read. After that it hands off to _read_bytes to do the
380
raise NotImplementedError(self._read_bytes)
383
"""Read bytes from this request's response until a newline byte.
385
This isn't particularly efficient, so should only be used when the
386
expected size of the line is quite short.
388
:returns: a string of bytes ending in a newline (byte 0x0A).
390
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
392
while not line or line[-1] != '\n':
393
new_char = self.read_bytes(1)
396
# end of file encountered reading from server
397
raise errors.ConnectionReset(
398
"please check connectivity and permissions",
399
"(and try -Dhpss if further diagnosis is required)")
403
class SmartClientMedium(object):
404
"""Smart client is a medium for sending smart protocol requests over."""
407
super(SmartClientMedium, self).__init__()
408
self._protocol_version_error = None
409
self._protocol_version = None
411
def protocol_version(self):
412
"""Find out the best protocol version to use."""
413
if self._protocol_version_error is not None:
414
raise self._protocol_version_error
415
if self._protocol_version is None:
417
medium_request = self.get_request()
418
# Send a 'hello' request in protocol version one, for maximum
419
# backwards compatibility.
420
client_protocol = SmartClientRequestProtocolOne(medium_request)
421
self._protocol_version = client_protocol.query_version()
422
except errors.SmartProtocolError, e:
423
# Cache the error, just like we would cache a successful
425
self._protocol_version_error = e
427
return self._protocol_version
429
def disconnect(self):
430
"""If this medium maintains a persistent connection, close it.
432
The default implementation does nothing.
436
class SmartClientStreamMedium(SmartClientMedium):
437
"""Stream based medium common class.
439
SmartClientStreamMediums operate on a stream. All subclasses use a common
440
SmartClientStreamMediumRequest for their requests, and should implement
441
_accept_bytes and _read_bytes to allow the request objects to send and
446
SmartClientMedium.__init__(self)
447
self._current_request = None
448
# Be optimistic: we assume the remote end can accept new remote
449
# requests until we get an error saying otherwise. (1.2 adds some
450
# requests that send bodies, which confuses older servers.)
451
self._remote_is_at_least_1_2 = True
453
def accept_bytes(self, bytes):
454
self._accept_bytes(bytes)
457
"""The SmartClientStreamMedium knows how to close the stream when it is
463
"""Flush the output stream.
465
This method is used by the SmartClientStreamMediumRequest to ensure that
466
all data for a request is sent, to avoid long timeouts or deadlocks.
468
raise NotImplementedError(self._flush)
470
def get_request(self):
471
"""See SmartClientMedium.get_request().
473
SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
476
return SmartClientStreamMediumRequest(self)
478
def read_bytes(self, count):
479
return self._read_bytes(count)
482
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
483
"""A client medium using simple pipes.
485
This client does not manage the pipes: it assumes they will always be open.
488
def __init__(self, readable_pipe, writeable_pipe):
489
SmartClientStreamMedium.__init__(self)
490
self._readable_pipe = readable_pipe
491
self._writeable_pipe = writeable_pipe
493
def _accept_bytes(self, bytes):
494
"""See SmartClientStreamMedium.accept_bytes."""
495
self._writeable_pipe.write(bytes)
498
"""See SmartClientStreamMedium._flush()."""
499
self._writeable_pipe.flush()
501
def _read_bytes(self, count):
502
"""See SmartClientStreamMedium._read_bytes."""
503
return self._readable_pipe.read(count)
506
class SmartSSHClientMedium(SmartClientStreamMedium):
507
"""A client medium using SSH."""
509
def __init__(self, host, port=None, username=None, password=None,
510
vendor=None, bzr_remote_path=None):
511
"""Creates a client that will connect on the first use.
513
:param vendor: An optional override for the ssh vendor to use. See
514
bzrlib.transport.ssh for details on ssh vendors.
516
SmartClientStreamMedium.__init__(self)
517
self._connected = False
519
self._password = password
521
self._username = username
522
self._read_from = None
523
self._ssh_connection = None
524
self._vendor = vendor
525
self._write_to = None
526
self._bzr_remote_path = bzr_remote_path
527
if self._bzr_remote_path is None:
528
symbol_versioning.warn(
529
'bzr_remote_path is required as of bzr 0.92',
530
DeprecationWarning, stacklevel=2)
531
self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
533
def _accept_bytes(self, bytes):
534
"""See SmartClientStreamMedium.accept_bytes."""
535
self._ensure_connection()
536
self._write_to.write(bytes)
538
def disconnect(self):
539
"""See SmartClientMedium.disconnect()."""
540
if not self._connected:
542
self._read_from.close()
543
self._write_to.close()
544
self._ssh_connection.close()
545
self._connected = False
547
def _ensure_connection(self):
548
"""Connect this medium if not already connected."""
551
if self._vendor is None:
552
vendor = ssh._get_ssh_vendor()
554
vendor = self._vendor
555
self._ssh_connection = vendor.connect_ssh(self._username,
556
self._password, self._host, self._port,
557
command=[self._bzr_remote_path, 'serve', '--inet',
558
'--directory=/', '--allow-writes'])
559
self._read_from, self._write_to = \
560
self._ssh_connection.get_filelike_channels()
561
self._connected = True
564
"""See SmartClientStreamMedium._flush()."""
565
self._write_to.flush()
567
def _read_bytes(self, count):
568
"""See SmartClientStreamMedium.read_bytes."""
569
if not self._connected:
570
raise errors.MediumNotConnected(self)
571
return self._read_from.read(count)
574
# Port 4155 is the default port for bzr://, registered with IANA.
575
BZR_DEFAULT_INTERFACE = '0.0.0.0'
576
BZR_DEFAULT_PORT = 4155
579
class SmartTCPClientMedium(SmartClientStreamMedium):
580
"""A client medium using TCP."""
582
def __init__(self, host, port):
583
"""Creates a client that will connect on the first use."""
584
SmartClientStreamMedium.__init__(self)
585
self._connected = False
590
def _accept_bytes(self, bytes):
591
"""See SmartClientMedium.accept_bytes."""
592
self._ensure_connection()
593
osutils.send_all(self._socket, bytes)
595
def disconnect(self):
596
"""See SmartClientMedium.disconnect()."""
597
if not self._connected:
601
self._connected = False
603
def _ensure_connection(self):
604
"""Connect this medium if not already connected."""
607
self._socket = socket.socket()
608
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
609
if self._port is None:
610
port = BZR_DEFAULT_PORT
612
port = int(self._port)
614
self._socket.connect((self._host, port))
615
except socket.error, err:
616
# socket errors either have a (string) or (errno, string) as their
618
if type(err.args) is str:
621
err_msg = err.args[1]
622
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
623
(self._host, port, err_msg))
624
self._connected = True
627
"""See SmartClientStreamMedium._flush().
629
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
630
add a means to do a flush, but that can be done in the future.
633
def _read_bytes(self, count):
634
"""See SmartClientMedium.read_bytes."""
635
if not self._connected:
636
raise errors.MediumNotConnected(self)
637
return self._socket.recv(count)
640
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
641
"""A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
643
def __init__(self, medium):
644
SmartClientMediumRequest.__init__(self, medium)
645
# check that we are safe concurrency wise. If some streams start
646
# allowing concurrent requests - i.e. via multiplexing - then this
647
# assert should be moved to SmartClientStreamMedium.get_request,
648
# and the setting/unsetting of _current_request likewise moved into
649
# that class : but its unneeded overhead for now. RBC 20060922
650
if self._medium._current_request is not None:
651
raise errors.TooManyConcurrentRequests(self._medium)
652
self._medium._current_request = self
654
def _accept_bytes(self, bytes):
655
"""See SmartClientMediumRequest._accept_bytes.
657
This forwards to self._medium._accept_bytes because we are operating
658
on the mediums stream.
660
self._medium._accept_bytes(bytes)
662
def _finished_reading(self):
663
"""See SmartClientMediumRequest._finished_reading.
665
This clears the _current_request on self._medium to allow a new
666
request to be created.
668
if self._medium._current_request is not self:
669
raise AssertionError()
670
self._medium._current_request = None
672
def _finished_writing(self):
673
"""See SmartClientMediumRequest._finished_writing.
675
This invokes self._medium._flush to ensure all bytes are transmitted.
677
self._medium._flush()
679
def _read_bytes(self, count):
680
"""See SmartClientMediumRequest._read_bytes.
682
This forwards to self._medium._read_bytes because we are operating
683
on the mediums stream.
685
return self._medium._read_bytes(count)
added
removed
bzrlib/smart/medium.py
