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
SmartServerRequestProtocolOne,
39
SmartServerRequestProtocolTwo,
41
from bzrlib.transport import ssh
44
class SmartServerStreamMedium(object):
45
"""Handles smart commands coming over a stream.
47
The stream may be a pipe connected to sshd, or a tcp socket, or an
48
in-process fifo for testing.
50
One instance is created for each connected client; it can serve multiple
51
requests in the lifetime of the connection.
53
The server passes requests through to an underlying backing transport,
54
which will typically be a LocalTransport looking at the server's filesystem.
57
def __init__(self, backing_transport):
58
"""Construct new server.
60
:param backing_transport: Transport for the directory served.
62
# backing_transport could be passed to serve instead of __init__
63
self.backing_transport = backing_transport
67
"""Serve requests until the client disconnects."""
68
# Keep a reference to stderr because the sys module's globals get set to
69
# None during interpreter shutdown.
70
from sys import stderr
72
while not self.finished:
73
server_protocol = self._build_protocol()
74
self._serve_one_request(server_protocol)
76
stderr.write("%s terminating on exception %s\n" % (self, e))
79
def _build_protocol(self):
80
"""Identifies the version of the incoming request, and returns an
81
a protocol object that can interpret it.
83
If more bytes than the version prefix of the request are read, they will
84
be fed into the protocol before it is returned.
86
:returns: a SmartServerRequestProtocol.
88
# Identify the protocol version.
89
bytes = self._get_line()
90
if bytes.startswith(REQUEST_VERSION_TWO):
91
protocol_class = SmartServerRequestProtocolTwo
92
bytes = bytes[len(REQUEST_VERSION_TWO):]
94
protocol_class = SmartServerRequestProtocolOne
95
protocol = protocol_class(self.backing_transport, self._write_out)
96
protocol.accept_bytes(bytes)
99
def _serve_one_request(self, protocol):
100
"""Read one request from input, process, send back a response.
102
:param protocol: a SmartServerRequestProtocol.
105
self._serve_one_request_unguarded(protocol)
106
except KeyboardInterrupt:
109
self.terminate_due_to_error()
111
def terminate_due_to_error(self):
112
"""Called when an unhandled exception from the protocol occurs."""
113
raise NotImplementedError(self.terminate_due_to_error)
115
def _get_bytes(self, desired_count):
116
"""Get some bytes from the medium.
118
:param desired_count: number of bytes we want to read.
120
raise NotImplementedError(self._get_bytes)
123
"""Read bytes from this request's response until a newline byte.
125
This isn't particularly efficient, so should only be used when the
126
expected size of the line is quite short.
128
:returns: a string of bytes ending in a newline (byte 0x0A).
130
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
132
while not line or line[-1] != '\n':
133
new_char = self._get_bytes(1)
136
# Ran out of bytes before receiving a complete line.
141
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
143
def __init__(self, sock, backing_transport):
146
:param sock: the socket the server will read from. It will be put
149
SmartServerStreamMedium.__init__(self, backing_transport)
151
sock.setblocking(True)
154
def _serve_one_request_unguarded(self, protocol):
155
while protocol.next_read_size():
157
protocol.accept_bytes(self.push_back)
160
bytes = self._get_bytes(4096)
164
protocol.accept_bytes(bytes)
166
self.push_back = protocol.excess_buffer
168
def _get_bytes(self, desired_count):
169
# We ignore the desired_count because on sockets it's more efficient to
171
return self.socket.recv(4096)
173
def terminate_due_to_error(self):
174
"""Called when an unhandled exception from the protocol occurs."""
175
# TODO: This should log to a server log file, but no such thing
176
# exists yet. Andrew Bennetts 2006-09-29.
180
def _write_out(self, bytes):
181
osutils.send_all(self.socket, bytes)
184
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
186
def __init__(self, in_file, out_file, backing_transport):
187
"""Construct new server.
189
:param in_file: Python file from which requests can be read.
190
:param out_file: Python file to write responses.
191
:param backing_transport: Transport for the directory served.
193
SmartServerStreamMedium.__init__(self, backing_transport)
194
if sys.platform == 'win32':
195
# force binary mode for files
197
for f in (in_file, out_file):
198
fileno = getattr(f, 'fileno', None)
200
msvcrt.setmode(fileno(), os.O_BINARY)
204
def _serve_one_request_unguarded(self, protocol):
206
bytes_to_read = protocol.next_read_size()
207
if bytes_to_read == 0:
208
# Finished serving this request.
211
bytes = self._get_bytes(bytes_to_read)
213
# Connection has been closed.
217
protocol.accept_bytes(bytes)
219
def _get_bytes(self, desired_count):
220
return self._in.read(desired_count)
222
def terminate_due_to_error(self):
223
# TODO: This should log to a server log file, but no such thing
224
# exists yet. Andrew Bennetts 2006-09-29.
228
def _write_out(self, bytes):
229
self._out.write(bytes)
232
class SmartClientMediumRequest(object):
233
"""A request on a SmartClientMedium.
235
Each request allows bytes to be provided to it via accept_bytes, and then
236
the response bytes to be read via read_bytes.
239
request.accept_bytes('123')
240
request.finished_writing()
241
result = request.read_bytes(3)
242
request.finished_reading()
244
It is up to the individual SmartClientMedium whether multiple concurrent
245
requests can exist. See SmartClientMedium.get_request to obtain instances
246
of SmartClientMediumRequest, and the concrete Medium you are using for
247
details on concurrency and pipelining.
250
def __init__(self, medium):
251
"""Construct a SmartClientMediumRequest for the medium medium."""
252
self._medium = medium
253
# we track state by constants - we may want to use the same
254
# pattern as BodyReader if it gets more complex.
255
# valid states are: "writing", "reading", "done"
256
self._state = "writing"
258
def accept_bytes(self, bytes):
259
"""Accept bytes for inclusion in this request.
261
This method may not be be called after finished_writing() has been
262
called. It depends upon the Medium whether or not the bytes will be
263
immediately transmitted. Message based Mediums will tend to buffer the
264
bytes until finished_writing() is called.
266
:param bytes: A bytestring.
268
if self._state != "writing":
269
raise errors.WritingCompleted(self)
270
self._accept_bytes(bytes)
272
def _accept_bytes(self, bytes):
273
"""Helper for accept_bytes.
275
Accept_bytes checks the state of the request to determing if bytes
276
should be accepted. After that it hands off to _accept_bytes to do the
279
raise NotImplementedError(self._accept_bytes)
281
def finished_reading(self):
282
"""Inform the request that all desired data has been read.
284
This will remove the request from the pipeline for its medium (if the
285
medium supports pipelining) and any further calls to methods on the
286
request will raise ReadingCompleted.
288
if self._state == "writing":
289
raise errors.WritingNotComplete(self)
290
if self._state != "reading":
291
raise errors.ReadingCompleted(self)
293
self._finished_reading()
295
def _finished_reading(self):
296
"""Helper for finished_reading.
298
finished_reading checks the state of the request to determine if
299
finished_reading is allowed, and if it is hands off to _finished_reading
300
to perform the action.
302
raise NotImplementedError(self._finished_reading)
304
def finished_writing(self):
305
"""Finish the writing phase of this request.
307
This will flush all pending data for this request along the medium.
308
After calling finished_writing, you may not call accept_bytes anymore.
310
if self._state != "writing":
311
raise errors.WritingCompleted(self)
312
self._state = "reading"
313
self._finished_writing()
315
def _finished_writing(self):
316
"""Helper for finished_writing.
318
finished_writing checks the state of the request to determine if
319
finished_writing is allowed, and if it is hands off to _finished_writing
320
to perform the action.
322
raise NotImplementedError(self._finished_writing)
324
def read_bytes(self, count):
325
"""Read bytes from this requests response.
327
This method will block and wait for count bytes to be read. It may not
328
be invoked until finished_writing() has been called - this is to ensure
329
a message-based approach to requests, for compatibility with message
330
based mediums like HTTP.
332
if self._state == "writing":
333
raise errors.WritingNotComplete(self)
334
if self._state != "reading":
335
raise errors.ReadingCompleted(self)
336
return self._read_bytes(count)
338
def _read_bytes(self, count):
339
"""Helper for read_bytes.
341
read_bytes checks the state of the request to determing if bytes
342
should be read. After that it hands off to _read_bytes to do the
345
raise NotImplementedError(self._read_bytes)
348
"""Read bytes from this request's response until a newline byte.
350
This isn't particularly efficient, so should only be used when the
351
expected size of the line is quite short.
353
:returns: a string of bytes ending in a newline (byte 0x0A).
355
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
357
while not line or line[-1] != '\n':
358
new_char = self.read_bytes(1)
361
# end of file encountered reading from server
362
raise errors.ConnectionReset(
363
"please check connectivity and permissions",
364
"(and try -Dhpss if further diagnosis is required)")
368
class SmartClientMedium(object):
369
"""Smart client is a medium for sending smart protocol requests over."""
371
def disconnect(self):
372
"""If this medium maintains a persistent connection, close it.
374
The default implementation does nothing.
378
class SmartClientStreamMedium(SmartClientMedium):
379
"""Stream based medium common class.
381
SmartClientStreamMediums operate on a stream. All subclasses use a common
382
SmartClientStreamMediumRequest for their requests, and should implement
383
_accept_bytes and _read_bytes to allow the request objects to send and
388
self._current_request = None
389
# Be optimistic: we assume the remote end can accept new remote
390
# requests until we get an error saying otherwise. (1.2 adds some
391
# requests that send bodies, which confuses older servers.)
392
self._remote_is_at_least_1_2 = True
394
def accept_bytes(self, bytes):
395
self._accept_bytes(bytes)
398
"""The SmartClientStreamMedium knows how to close the stream when it is
404
"""Flush the output stream.
406
This method is used by the SmartClientStreamMediumRequest to ensure that
407
all data for a request is sent, to avoid long timeouts or deadlocks.
409
raise NotImplementedError(self._flush)
411
def get_request(self):
412
"""See SmartClientMedium.get_request().
414
SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
417
return SmartClientStreamMediumRequest(self)
419
def read_bytes(self, count):
420
return self._read_bytes(count)
423
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
424
"""A client medium using simple pipes.
426
This client does not manage the pipes: it assumes they will always be open.
429
def __init__(self, readable_pipe, writeable_pipe):
430
SmartClientStreamMedium.__init__(self)
431
self._readable_pipe = readable_pipe
432
self._writeable_pipe = writeable_pipe
434
def _accept_bytes(self, bytes):
435
"""See SmartClientStreamMedium.accept_bytes."""
436
self._writeable_pipe.write(bytes)
439
"""See SmartClientStreamMedium._flush()."""
440
self._writeable_pipe.flush()
442
def _read_bytes(self, count):
443
"""See SmartClientStreamMedium._read_bytes."""
444
return self._readable_pipe.read(count)
447
class SmartSSHClientMedium(SmartClientStreamMedium):
448
"""A client medium using SSH."""
450
def __init__(self, host, port=None, username=None, password=None,
451
vendor=None, bzr_remote_path=None):
452
"""Creates a client that will connect on the first use.
454
:param vendor: An optional override for the ssh vendor to use. See
455
bzrlib.transport.ssh for details on ssh vendors.
457
SmartClientStreamMedium.__init__(self)
458
self._connected = False
460
self._password = password
462
self._username = username
463
self._read_from = None
464
self._ssh_connection = None
465
self._vendor = vendor
466
self._write_to = None
467
self._bzr_remote_path = bzr_remote_path
468
if self._bzr_remote_path is None:
469
symbol_versioning.warn(
470
'bzr_remote_path is required as of bzr 0.92',
471
DeprecationWarning, stacklevel=2)
472
self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
474
def _accept_bytes(self, bytes):
475
"""See SmartClientStreamMedium.accept_bytes."""
476
self._ensure_connection()
477
self._write_to.write(bytes)
479
def disconnect(self):
480
"""See SmartClientMedium.disconnect()."""
481
if not self._connected:
483
self._read_from.close()
484
self._write_to.close()
485
self._ssh_connection.close()
486
self._connected = False
488
def _ensure_connection(self):
489
"""Connect this medium if not already connected."""
492
if self._vendor is None:
493
vendor = ssh._get_ssh_vendor()
495
vendor = self._vendor
496
self._ssh_connection = vendor.connect_ssh(self._username,
497
self._password, self._host, self._port,
498
command=[self._bzr_remote_path, 'serve', '--inet',
499
'--directory=/', '--allow-writes'])
500
self._read_from, self._write_to = \
501
self._ssh_connection.get_filelike_channels()
502
self._connected = True
505
"""See SmartClientStreamMedium._flush()."""
506
self._write_to.flush()
508
def _read_bytes(self, count):
509
"""See SmartClientStreamMedium.read_bytes."""
510
if not self._connected:
511
raise errors.MediumNotConnected(self)
512
return self._read_from.read(count)
515
# Port 4155 is the default port for bzr://, registered with IANA.
516
BZR_DEFAULT_INTERFACE = '0.0.0.0'
517
BZR_DEFAULT_PORT = 4155
520
class SmartTCPClientMedium(SmartClientStreamMedium):
521
"""A client medium using TCP."""
523
def __init__(self, host, port):
524
"""Creates a client that will connect on the first use."""
525
SmartClientStreamMedium.__init__(self)
526
self._connected = False
531
def _accept_bytes(self, bytes):
532
"""See SmartClientMedium.accept_bytes."""
533
self._ensure_connection()
534
osutils.send_all(self._socket, bytes)
536
def disconnect(self):
537
"""See SmartClientMedium.disconnect()."""
538
if not self._connected:
542
self._connected = False
544
def _ensure_connection(self):
545
"""Connect this medium if not already connected."""
548
self._socket = socket.socket()
549
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
550
if self._port is None:
551
port = BZR_DEFAULT_PORT
553
port = int(self._port)
555
self._socket.connect((self._host, port))
556
except socket.error, err:
557
# socket errors either have a (string) or (errno, string) as their
559
if type(err.args) is str:
562
err_msg = err.args[1]
563
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
564
(self._host, port, err_msg))
565
self._connected = True
568
"""See SmartClientStreamMedium._flush().
570
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
571
add a means to do a flush, but that can be done in the future.
574
def _read_bytes(self, count):
575
"""See SmartClientMedium.read_bytes."""
576
if not self._connected:
577
raise errors.MediumNotConnected(self)
578
return self._socket.recv(count)
581
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
582
"""A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
584
def __init__(self, medium):
585
SmartClientMediumRequest.__init__(self, medium)
586
# check that we are safe concurrency wise. If some streams start
587
# allowing concurrent requests - i.e. via multiplexing - then this
588
# assert should be moved to SmartClientStreamMedium.get_request,
589
# and the setting/unsetting of _current_request likewise moved into
590
# that class : but its unneeded overhead for now. RBC 20060922
591
if self._medium._current_request is not None:
592
raise errors.TooManyConcurrentRequests(self._medium)
593
self._medium._current_request = self
595
def _accept_bytes(self, bytes):
596
"""See SmartClientMediumRequest._accept_bytes.
598
This forwards to self._medium._accept_bytes because we are operating
599
on the mediums stream.
601
self._medium._accept_bytes(bytes)
603
def _finished_reading(self):
604
"""See SmartClientMediumRequest._finished_reading.
606
This clears the _current_request on self._medium to allow a new
607
request to be created.
609
assert self._medium._current_request is self
610
self._medium._current_request = None
612
def _finished_writing(self):
613
"""See SmartClientMediumRequest._finished_writing.
615
This invokes self._medium._flush to ensure all bytes are transmitted.
617
self._medium._flush()
619
def _read_bytes(self, count):
620
"""See SmartClientMediumRequest._read_bytes.
622
This forwards to self._medium._read_bytes because we are operating
623
on the mediums stream.
625
return self._medium._read_bytes(count)
added
removed
bzrlib/smart/medium.py
