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.
58
def __init__(self, backing_transport):
59
"""Construct new server.
61
:param backing_transport: Transport for the directory served.
63
# backing_transport could be passed to serve instead of __init__
64
self.backing_transport = backing_transport
68
"""Serve requests until the client disconnects."""
69
# Keep a reference to stderr because the sys module's globals get set to
70
# None during interpreter shutdown.
71
from sys import stderr
73
while not self.finished:
74
server_protocol = self._build_protocol()
75
self._serve_one_request(server_protocol)
77
stderr.write("%s terminating on exception %s\n" % (self, e))
80
def _build_protocol(self):
81
"""Identifies the version of the incoming request, and returns an
82
a protocol object that can interpret it.
84
If more bytes than the version prefix of the request are read, they will
85
be fed into the protocol before it is returned.
87
:returns: a SmartServerRequestProtocol.
89
# Identify the protocol version.
90
bytes = self._get_line()
91
if bytes.startswith(REQUEST_VERSION_TWO):
92
protocol_class = SmartServerRequestProtocolTwo
93
bytes = bytes[len(REQUEST_VERSION_TWO):]
95
protocol_class = SmartServerRequestProtocolOne
96
protocol = protocol_class(self.backing_transport, self._write_out)
97
protocol.accept_bytes(bytes)
100
def _serve_one_request(self, protocol):
101
"""Read one request from input, process, send back a response.
103
:param protocol: a SmartServerRequestProtocol.
106
self._serve_one_request_unguarded(protocol)
107
except KeyboardInterrupt:
110
self.terminate_due_to_error()
112
def terminate_due_to_error(self):
113
"""Called when an unhandled exception from the protocol occurs."""
114
raise NotImplementedError(self.terminate_due_to_error)
116
def _get_bytes(self, desired_count):
117
"""Get some bytes from the medium.
119
:param desired_count: number of bytes we want to read.
121
raise NotImplementedError(self._get_bytes)
124
"""Read bytes from this request's response until a newline byte.
126
This isn't particularly efficient, so should only be used when the
127
expected size of the line is quite short.
129
:returns: a string of bytes ending in a newline (byte 0x0A).
131
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
133
while not line or line[-1] != '\n':
134
new_char = self._get_bytes(1)
137
# Ran out of bytes before receiving a complete line.
142
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
144
def __init__(self, sock, backing_transport):
147
:param sock: the socket the server will read from. It will be put
150
SmartServerStreamMedium.__init__(self, backing_transport)
152
sock.setblocking(True)
155
def _serve_one_request_unguarded(self, protocol):
156
while protocol.next_read_size():
158
protocol.accept_bytes(self.push_back)
161
bytes = self._get_bytes(4096)
165
protocol.accept_bytes(bytes)
167
self.push_back = protocol.excess_buffer
169
def _get_bytes(self, desired_count):
170
# We ignore the desired_count because on sockets it's more efficient to
172
return self.socket.recv(4096)
174
def terminate_due_to_error(self):
175
"""Called when an unhandled exception from the protocol occurs."""
176
# TODO: This should log to a server log file, but no such thing
177
# exists yet. Andrew Bennetts 2006-09-29.
181
def _write_out(self, bytes):
182
osutils.send_all(self.socket, bytes)
185
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
187
def __init__(self, in_file, out_file, backing_transport):
188
"""Construct new server.
190
:param in_file: Python file from which requests can be read.
191
:param out_file: Python file to write responses.
192
:param backing_transport: Transport for the directory served.
194
SmartServerStreamMedium.__init__(self, backing_transport)
195
if sys.platform == 'win32':
196
# force binary mode for files
198
for f in (in_file, out_file):
199
fileno = getattr(f, 'fileno', None)
201
msvcrt.setmode(fileno(), os.O_BINARY)
205
def _serve_one_request_unguarded(self, protocol):
207
bytes_to_read = protocol.next_read_size()
208
if bytes_to_read == 0:
209
# Finished serving this request.
212
bytes = self._get_bytes(bytes_to_read)
214
# Connection has been closed.
218
protocol.accept_bytes(bytes)
220
def _get_bytes(self, desired_count):
221
return self._in.read(desired_count)
223
def terminate_due_to_error(self):
224
# TODO: This should log to a server log file, but no such thing
225
# exists yet. Andrew Bennetts 2006-09-29.
229
def _write_out(self, bytes):
230
self._out.write(bytes)
233
class SmartClientMediumRequest(object):
234
"""A request on a SmartClientMedium.
236
Each request allows bytes to be provided to it via accept_bytes, and then
237
the response bytes to be read via read_bytes.
240
request.accept_bytes('123')
241
request.finished_writing()
242
result = request.read_bytes(3)
243
request.finished_reading()
245
It is up to the individual SmartClientMedium whether multiple concurrent
246
requests can exist. See SmartClientMedium.get_request to obtain instances
247
of SmartClientMediumRequest, and the concrete Medium you are using for
248
details on concurrency and pipelining.
251
def __init__(self, medium):
252
"""Construct a SmartClientMediumRequest for the medium medium."""
253
self._medium = medium
254
# we track state by constants - we may want to use the same
255
# pattern as BodyReader if it gets more complex.
256
# valid states are: "writing", "reading", "done"
257
self._state = "writing"
259
def accept_bytes(self, bytes):
260
"""Accept bytes for inclusion in this request.
262
This method may not be be called after finished_writing() has been
263
called. It depends upon the Medium whether or not the bytes will be
264
immediately transmitted. Message based Mediums will tend to buffer the
265
bytes until finished_writing() is called.
267
:param bytes: A bytestring.
269
if self._state != "writing":
270
raise errors.WritingCompleted(self)
271
self._accept_bytes(bytes)
273
def _accept_bytes(self, bytes):
274
"""Helper for accept_bytes.
276
Accept_bytes checks the state of the request to determing if bytes
277
should be accepted. After that it hands off to _accept_bytes to do the
280
raise NotImplementedError(self._accept_bytes)
282
def finished_reading(self):
283
"""Inform the request that all desired data has been read.
285
This will remove the request from the pipeline for its medium (if the
286
medium supports pipelining) and any further calls to methods on the
287
request will raise ReadingCompleted.
289
if self._state == "writing":
290
raise errors.WritingNotComplete(self)
291
if self._state != "reading":
292
raise errors.ReadingCompleted(self)
294
self._finished_reading()
296
def _finished_reading(self):
297
"""Helper for finished_reading.
299
finished_reading checks the state of the request to determine if
300
finished_reading is allowed, and if it is hands off to _finished_reading
301
to perform the action.
303
raise NotImplementedError(self._finished_reading)
305
def finished_writing(self):
306
"""Finish the writing phase of this request.
308
This will flush all pending data for this request along the medium.
309
After calling finished_writing, you may not call accept_bytes anymore.
311
if self._state != "writing":
312
raise errors.WritingCompleted(self)
313
self._state = "reading"
314
self._finished_writing()
316
def _finished_writing(self):
317
"""Helper for finished_writing.
319
finished_writing checks the state of the request to determine if
320
finished_writing is allowed, and if it is hands off to _finished_writing
321
to perform the action.
323
raise NotImplementedError(self._finished_writing)
325
def read_bytes(self, count):
326
"""Read bytes from this requests response.
328
This method will block and wait for count bytes to be read. It may not
329
be invoked until finished_writing() has been called - this is to ensure
330
a message-based approach to requests, for compatibility with message
331
based mediums like HTTP.
333
if self._state == "writing":
334
raise errors.WritingNotComplete(self)
335
if self._state != "reading":
336
raise errors.ReadingCompleted(self)
337
return self._read_bytes(count)
339
def _read_bytes(self, count):
340
"""Helper for read_bytes.
342
read_bytes checks the state of the request to determing if bytes
343
should be read. After that it hands off to _read_bytes to do the
346
raise NotImplementedError(self._read_bytes)
349
"""Read bytes from this request's response until a newline byte.
351
This isn't particularly efficient, so should only be used when the
352
expected size of the line is quite short.
354
:returns: a string of bytes ending in a newline (byte 0x0A).
356
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
358
while not line or line[-1] != '\n':
359
new_char = self.read_bytes(1)
362
# end of file encountered reading from server
363
raise errors.ConnectionReset(
364
"please check connectivity and permissions",
365
"(and try -Dhpss if further diagnosis is required)")
369
class SmartClientMedium(object):
370
"""Smart client is a medium for sending smart protocol requests over."""
373
super(SmartClientMedium, self).__init__()
374
self._protocol_version_error = None
375
self._protocol_version = None
377
def protocol_version(self):
378
"""Find out the best protocol version to use."""
379
if self._protocol_version_error is not None:
380
raise self._protocol_version_error
381
if self._protocol_version is None:
383
medium_request = self.get_request()
384
# Send a 'hello' request in protocol version one, for maximum
385
# backwards compatibility.
386
client_protocol = SmartClientRequestProtocolOne(medium_request)
387
self._protocol_version = client_protocol.query_version()
388
except errors.SmartProtocolError, e:
389
# Cache the error, just like we would cache a successful
391
self._protocol_version_error = e
393
return self._protocol_version
395
def disconnect(self):
396
"""If this medium maintains a persistent connection, close it.
398
The default implementation does nothing.
402
class SmartClientStreamMedium(SmartClientMedium):
403
"""Stream based medium common class.
405
SmartClientStreamMediums operate on a stream. All subclasses use a common
406
SmartClientStreamMediumRequest for their requests, and should implement
407
_accept_bytes and _read_bytes to allow the request objects to send and
412
SmartClientMedium.__init__(self)
413
self._current_request = None
414
# Be optimistic: we assume the remote end can accept new remote
415
# requests until we get an error saying otherwise. (1.2 adds some
416
# requests that send bodies, which confuses older servers.)
417
self._remote_is_at_least_1_2 = True
419
def accept_bytes(self, bytes):
420
self._accept_bytes(bytes)
423
"""The SmartClientStreamMedium knows how to close the stream when it is
429
"""Flush the output stream.
431
This method is used by the SmartClientStreamMediumRequest to ensure that
432
all data for a request is sent, to avoid long timeouts or deadlocks.
434
raise NotImplementedError(self._flush)
436
def get_request(self):
437
"""See SmartClientMedium.get_request().
439
SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
442
return SmartClientStreamMediumRequest(self)
444
def read_bytes(self, count):
445
return self._read_bytes(count)
448
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
449
"""A client medium using simple pipes.
451
This client does not manage the pipes: it assumes they will always be open.
454
def __init__(self, readable_pipe, writeable_pipe):
455
SmartClientStreamMedium.__init__(self)
456
self._readable_pipe = readable_pipe
457
self._writeable_pipe = writeable_pipe
459
def _accept_bytes(self, bytes):
460
"""See SmartClientStreamMedium.accept_bytes."""
461
self._writeable_pipe.write(bytes)
464
"""See SmartClientStreamMedium._flush()."""
465
self._writeable_pipe.flush()
467
def _read_bytes(self, count):
468
"""See SmartClientStreamMedium._read_bytes."""
469
return self._readable_pipe.read(count)
472
class SmartSSHClientMedium(SmartClientStreamMedium):
473
"""A client medium using SSH."""
475
def __init__(self, host, port=None, username=None, password=None,
476
vendor=None, bzr_remote_path=None):
477
"""Creates a client that will connect on the first use.
479
:param vendor: An optional override for the ssh vendor to use. See
480
bzrlib.transport.ssh for details on ssh vendors.
482
SmartClientStreamMedium.__init__(self)
483
self._connected = False
485
self._password = password
487
self._username = username
488
self._read_from = None
489
self._ssh_connection = None
490
self._vendor = vendor
491
self._write_to = None
492
self._bzr_remote_path = bzr_remote_path
493
if self._bzr_remote_path is None:
494
symbol_versioning.warn(
495
'bzr_remote_path is required as of bzr 0.92',
496
DeprecationWarning, stacklevel=2)
497
self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
499
def _accept_bytes(self, bytes):
500
"""See SmartClientStreamMedium.accept_bytes."""
501
self._ensure_connection()
502
self._write_to.write(bytes)
504
def disconnect(self):
505
"""See SmartClientMedium.disconnect()."""
506
if not self._connected:
508
self._read_from.close()
509
self._write_to.close()
510
self._ssh_connection.close()
511
self._connected = False
513
def _ensure_connection(self):
514
"""Connect this medium if not already connected."""
517
if self._vendor is None:
518
vendor = ssh._get_ssh_vendor()
520
vendor = self._vendor
521
self._ssh_connection = vendor.connect_ssh(self._username,
522
self._password, self._host, self._port,
523
command=[self._bzr_remote_path, 'serve', '--inet',
524
'--directory=/', '--allow-writes'])
525
self._read_from, self._write_to = \
526
self._ssh_connection.get_filelike_channels()
527
self._connected = True
530
"""See SmartClientStreamMedium._flush()."""
531
self._write_to.flush()
533
def _read_bytes(self, count):
534
"""See SmartClientStreamMedium.read_bytes."""
535
if not self._connected:
536
raise errors.MediumNotConnected(self)
537
return self._read_from.read(count)
540
# Port 4155 is the default port for bzr://, registered with IANA.
541
BZR_DEFAULT_INTERFACE = '0.0.0.0'
542
BZR_DEFAULT_PORT = 4155
545
class SmartTCPClientMedium(SmartClientStreamMedium):
546
"""A client medium using TCP."""
548
def __init__(self, host, port):
549
"""Creates a client that will connect on the first use."""
550
SmartClientStreamMedium.__init__(self)
551
self._connected = False
556
def _accept_bytes(self, bytes):
557
"""See SmartClientMedium.accept_bytes."""
558
self._ensure_connection()
559
osutils.send_all(self._socket, bytes)
561
def disconnect(self):
562
"""See SmartClientMedium.disconnect()."""
563
if not self._connected:
567
self._connected = False
569
def _ensure_connection(self):
570
"""Connect this medium if not already connected."""
573
self._socket = socket.socket()
574
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
575
if self._port is None:
576
port = BZR_DEFAULT_PORT
578
port = int(self._port)
580
self._socket.connect((self._host, port))
581
except socket.error, err:
582
# socket errors either have a (string) or (errno, string) as their
584
if type(err.args) is str:
587
err_msg = err.args[1]
588
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
589
(self._host, port, err_msg))
590
self._connected = True
593
"""See SmartClientStreamMedium._flush().
595
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
596
add a means to do a flush, but that can be done in the future.
599
def _read_bytes(self, count):
600
"""See SmartClientMedium.read_bytes."""
601
if not self._connected:
602
raise errors.MediumNotConnected(self)
603
return self._socket.recv(count)
606
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
607
"""A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
609
def __init__(self, medium):
610
SmartClientMediumRequest.__init__(self, medium)
611
# check that we are safe concurrency wise. If some streams start
612
# allowing concurrent requests - i.e. via multiplexing - then this
613
# assert should be moved to SmartClientStreamMedium.get_request,
614
# and the setting/unsetting of _current_request likewise moved into
615
# that class : but its unneeded overhead for now. RBC 20060922
616
if self._medium._current_request is not None:
617
raise errors.TooManyConcurrentRequests(self._medium)
618
self._medium._current_request = self
620
def _accept_bytes(self, bytes):
621
"""See SmartClientMediumRequest._accept_bytes.
623
This forwards to self._medium._accept_bytes because we are operating
624
on the mediums stream.
626
self._medium._accept_bytes(bytes)
628
def _finished_reading(self):
629
"""See SmartClientMediumRequest._finished_reading.
631
This clears the _current_request on self._medium to allow a new
632
request to be created.
634
assert self._medium._current_request is self
635
self._medium._current_request = None
637
def _finished_writing(self):
638
"""See SmartClientMediumRequest._finished_writing.
640
This invokes self._medium._flush to ensure all bytes are transmitted.
642
self._medium._flush()
644
def _read_bytes(self, count):
645
"""See SmartClientMediumRequest._read_bytes.
647
This forwards to self._medium._read_bytes because we are operating
648
on the mediums stream.
650
return self._medium._read_bytes(count)
added
removed
bzrlib/smart/medium.py
