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.
35
from bzrlib.smart.protocol import (
37
SmartServerRequestProtocolOne,
38
SmartServerRequestProtocolTwo,
40
from bzrlib.transport import ssh
43
class SmartServerStreamMedium(object):
44
"""Handles smart commands coming over a stream.
46
The stream may be a pipe connected to sshd, or a tcp socket, or an
47
in-process fifo for testing.
49
One instance is created for each connected client; it can serve multiple
50
requests in the lifetime of the connection.
52
The server passes requests through to an underlying backing transport,
53
which will typically be a LocalTransport looking at the server's filesystem.
56
def __init__(self, backing_transport, root_client_path='/'):
57
"""Construct new server.
59
:param backing_transport: Transport for the directory served.
61
# backing_transport could be passed to serve instead of __init__
62
self.backing_transport = backing_transport
63
self.root_client_path = root_client_path
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(
96
self.backing_transport, self._write_out, self.root_client_path)
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, root_client_path='/'):
147
:param sock: the socket the server will read from. It will be put
150
SmartServerStreamMedium.__init__(
151
self, backing_transport, root_client_path=root_client_path)
153
sock.setblocking(True)
156
def _serve_one_request_unguarded(self, protocol):
157
while protocol.next_read_size():
159
protocol.accept_bytes(self.push_back)
162
bytes = self._get_bytes(4096)
166
protocol.accept_bytes(bytes)
168
self.push_back = protocol.excess_buffer
170
def _get_bytes(self, desired_count):
171
# We ignore the desired_count because on sockets it's more efficient to
173
return self.socket.recv(4096)
175
def terminate_due_to_error(self):
176
"""Called when an unhandled exception from the protocol occurs."""
177
# TODO: This should log to a server log file, but no such thing
178
# exists yet. Andrew Bennetts 2006-09-29.
182
def _write_out(self, bytes):
183
self.socket.sendall(bytes)
186
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
188
def __init__(self, in_file, out_file, backing_transport):
189
"""Construct new server.
191
:param in_file: Python file from which requests can be read.
192
:param out_file: Python file to write responses.
193
:param backing_transport: Transport for the directory served.
195
SmartServerStreamMedium.__init__(self, backing_transport)
196
if sys.platform == 'win32':
197
# force binary mode for files
199
for f in (in_file, out_file):
200
fileno = getattr(f, 'fileno', None)
202
msvcrt.setmode(fileno(), os.O_BINARY)
206
def _serve_one_request_unguarded(self, protocol):
208
bytes_to_read = protocol.next_read_size()
209
if bytes_to_read == 0:
210
# Finished serving this request.
213
bytes = self._get_bytes(bytes_to_read)
215
# Connection has been closed.
219
protocol.accept_bytes(bytes)
221
def _get_bytes(self, desired_count):
222
return self._in.read(desired_count)
224
def terminate_due_to_error(self):
225
# TODO: This should log to a server log file, but no such thing
226
# exists yet. Andrew Bennetts 2006-09-29.
230
def _write_out(self, bytes):
231
self._out.write(bytes)
234
class SmartClientMediumRequest(object):
235
"""A request on a SmartClientMedium.
237
Each request allows bytes to be provided to it via accept_bytes, and then
238
the response bytes to be read via read_bytes.
241
request.accept_bytes('123')
242
request.finished_writing()
243
result = request.read_bytes(3)
244
request.finished_reading()
246
It is up to the individual SmartClientMedium whether multiple concurrent
247
requests can exist. See SmartClientMedium.get_request to obtain instances
248
of SmartClientMediumRequest, and the concrete Medium you are using for
249
details on concurrency and pipelining.
252
def __init__(self, medium):
253
"""Construct a SmartClientMediumRequest for the medium medium."""
254
self._medium = medium
255
# we track state by constants - we may want to use the same
256
# pattern as BodyReader if it gets more complex.
257
# valid states are: "writing", "reading", "done"
258
self._state = "writing"
260
def accept_bytes(self, bytes):
261
"""Accept bytes for inclusion in this request.
263
This method may not be be called after finished_writing() has been
264
called. It depends upon the Medium whether or not the bytes will be
265
immediately transmitted. Message based Mediums will tend to buffer the
266
bytes until finished_writing() is called.
268
:param bytes: A bytestring.
270
if self._state != "writing":
271
raise errors.WritingCompleted(self)
272
self._accept_bytes(bytes)
274
def _accept_bytes(self, bytes):
275
"""Helper for accept_bytes.
277
Accept_bytes checks the state of the request to determing if bytes
278
should be accepted. After that it hands off to _accept_bytes to do the
281
raise NotImplementedError(self._accept_bytes)
283
def finished_reading(self):
284
"""Inform the request that all desired data has been read.
286
This will remove the request from the pipeline for its medium (if the
287
medium supports pipelining) and any further calls to methods on the
288
request will raise ReadingCompleted.
290
if self._state == "writing":
291
raise errors.WritingNotComplete(self)
292
if self._state != "reading":
293
raise errors.ReadingCompleted(self)
295
self._finished_reading()
297
def _finished_reading(self):
298
"""Helper for finished_reading.
300
finished_reading checks the state of the request to determine if
301
finished_reading is allowed, and if it is hands off to _finished_reading
302
to perform the action.
304
raise NotImplementedError(self._finished_reading)
306
def finished_writing(self):
307
"""Finish the writing phase of this request.
309
This will flush all pending data for this request along the medium.
310
After calling finished_writing, you may not call accept_bytes anymore.
312
if self._state != "writing":
313
raise errors.WritingCompleted(self)
314
self._state = "reading"
315
self._finished_writing()
317
def _finished_writing(self):
318
"""Helper for finished_writing.
320
finished_writing checks the state of the request to determine if
321
finished_writing is allowed, and if it is hands off to _finished_writing
322
to perform the action.
324
raise NotImplementedError(self._finished_writing)
326
def read_bytes(self, count):
327
"""Read bytes from this requests response.
329
This method will block and wait for count bytes to be read. It may not
330
be invoked until finished_writing() has been called - this is to ensure
331
a message-based approach to requests, for compatibility with message
332
based mediums like HTTP.
334
if self._state == "writing":
335
raise errors.WritingNotComplete(self)
336
if self._state != "reading":
337
raise errors.ReadingCompleted(self)
338
return self._read_bytes(count)
340
def _read_bytes(self, count):
341
"""Helper for read_bytes.
343
read_bytes checks the state of the request to determing if bytes
344
should be read. After that it hands off to _read_bytes to do the
347
raise NotImplementedError(self._read_bytes)
350
"""Read bytes from this request's response until a newline byte.
352
This isn't particularly efficient, so should only be used when the
353
expected size of the line is quite short.
355
:returns: a string of bytes ending in a newline (byte 0x0A).
357
# XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
359
while not line or line[-1] != '\n':
360
new_char = self.read_bytes(1)
363
raise errors.SmartProtocolError(
364
'unexpected end of file reading from server')
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
390
def accept_bytes(self, bytes):
391
self._accept_bytes(bytes)
394
"""The SmartClientStreamMedium knows how to close the stream when it is
400
"""Flush the output stream.
402
This method is used by the SmartClientStreamMediumRequest to ensure that
403
all data for a request is sent, to avoid long timeouts or deadlocks.
405
raise NotImplementedError(self._flush)
407
def get_request(self):
408
"""See SmartClientMedium.get_request().
410
SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
413
return SmartClientStreamMediumRequest(self)
415
def read_bytes(self, count):
416
return self._read_bytes(count)
419
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
420
"""A client medium using simple pipes.
422
This client does not manage the pipes: it assumes they will always be open.
425
def __init__(self, readable_pipe, writeable_pipe):
426
SmartClientStreamMedium.__init__(self)
427
self._readable_pipe = readable_pipe
428
self._writeable_pipe = writeable_pipe
430
def _accept_bytes(self, bytes):
431
"""See SmartClientStreamMedium.accept_bytes."""
432
self._writeable_pipe.write(bytes)
435
"""See SmartClientStreamMedium._flush()."""
436
self._writeable_pipe.flush()
438
def _read_bytes(self, count):
439
"""See SmartClientStreamMedium._read_bytes."""
440
return self._readable_pipe.read(count)
443
class SmartSSHClientMedium(SmartClientStreamMedium):
444
"""A client medium using SSH."""
446
def __init__(self, host, port=None, username=None, password=None,
447
vendor=None, bzr_remote_path=None):
448
"""Creates a client that will connect on the first use.
450
:param vendor: An optional override for the ssh vendor to use. See
451
bzrlib.transport.ssh for details on ssh vendors.
453
SmartClientStreamMedium.__init__(self)
454
self._connected = False
456
self._password = password
458
self._username = username
459
self._read_from = None
460
self._ssh_connection = None
461
self._vendor = vendor
462
self._write_to = None
463
self._bzr_remote_path = bzr_remote_path
464
if self._bzr_remote_path is None:
465
symbol_versioning.warn(
466
'bzr_remote_path is required as of bzr 0.92',
467
DeprecationWarning, stacklevel=2)
468
self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
470
def _accept_bytes(self, bytes):
471
"""See SmartClientStreamMedium.accept_bytes."""
472
self._ensure_connection()
473
self._write_to.write(bytes)
475
def disconnect(self):
476
"""See SmartClientMedium.disconnect()."""
477
if not self._connected:
479
self._read_from.close()
480
self._write_to.close()
481
self._ssh_connection.close()
482
self._connected = False
484
def _ensure_connection(self):
485
"""Connect this medium if not already connected."""
488
if self._vendor is None:
489
vendor = ssh._get_ssh_vendor()
491
vendor = self._vendor
492
self._ssh_connection = vendor.connect_ssh(self._username,
493
self._password, self._host, self._port,
494
command=[self._bzr_remote_path, 'serve', '--inet',
495
'--directory=/', '--allow-writes'])
496
self._read_from, self._write_to = \
497
self._ssh_connection.get_filelike_channels()
498
self._connected = True
501
"""See SmartClientStreamMedium._flush()."""
502
self._write_to.flush()
504
def _read_bytes(self, count):
505
"""See SmartClientStreamMedium.read_bytes."""
506
if not self._connected:
507
raise errors.MediumNotConnected(self)
508
return self._read_from.read(count)
511
# Port 4155 is the default port for bzr://, registered with IANA.
512
BZR_DEFAULT_INTERFACE = '0.0.0.0'
513
BZR_DEFAULT_PORT = 4155
516
class SmartTCPClientMedium(SmartClientStreamMedium):
517
"""A client medium using TCP."""
519
def __init__(self, host, port):
520
"""Creates a client that will connect on the first use."""
521
SmartClientStreamMedium.__init__(self)
522
self._connected = False
527
def _accept_bytes(self, bytes):
528
"""See SmartClientMedium.accept_bytes."""
529
self._ensure_connection()
530
self._socket.sendall(bytes)
532
def disconnect(self):
533
"""See SmartClientMedium.disconnect()."""
534
if not self._connected:
538
self._connected = False
540
def _ensure_connection(self):
541
"""Connect this medium if not already connected."""
544
self._socket = socket.socket()
545
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
546
if self._port is None:
547
port = BZR_DEFAULT_PORT
549
port = int(self._port)
550
result = self._socket.connect_ex((self._host, port))
552
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
553
(self._host, port, os.strerror(result)))
554
self._connected = True
557
"""See SmartClientStreamMedium._flush().
559
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
560
add a means to do a flush, but that can be done in the future.
563
def _read_bytes(self, count):
564
"""See SmartClientMedium.read_bytes."""
565
if not self._connected:
566
raise errors.MediumNotConnected(self)
567
return self._socket.recv(count)
570
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
571
"""A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
573
def __init__(self, medium):
574
SmartClientMediumRequest.__init__(self, medium)
575
# check that we are safe concurrency wise. If some streams start
576
# allowing concurrent requests - i.e. via multiplexing - then this
577
# assert should be moved to SmartClientStreamMedium.get_request,
578
# and the setting/unsetting of _current_request likewise moved into
579
# that class : but its unneeded overhead for now. RBC 20060922
580
if self._medium._current_request is not None:
581
raise errors.TooManyConcurrentRequests(self._medium)
582
self._medium._current_request = self
584
def _accept_bytes(self, bytes):
585
"""See SmartClientMediumRequest._accept_bytes.
587
This forwards to self._medium._accept_bytes because we are operating
588
on the mediums stream.
590
self._medium._accept_bytes(bytes)
592
def _finished_reading(self):
593
"""See SmartClientMediumRequest._finished_reading.
595
This clears the _current_request on self._medium to allow a new
596
request to be created.
598
assert self._medium._current_request is self
599
self._medium._current_request = None
601
def _finished_writing(self):
602
"""See SmartClientMediumRequest._finished_writing.
604
This invokes self._medium._flush to ensure all bytes are transmitted.
606
self._medium._flush()
608
def _read_bytes(self, count):
609
"""See SmartClientMediumRequest._read_bytes.
611
This forwards to self._medium._read_bytes because we are operating
612
on the mediums stream.
614
return self._medium._read_bytes(count)
added
removed
bzrlib/smart/medium.py
