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.
31
from bzrlib import errors
32
from bzrlib.smart.protocol import (
33
SmartServerRequestProtocolOne,
34
SmartServerRequestProtocolTwo,
38
from bzrlib.transport import ssh
39
except errors.ParamikoNotPresent:
40
# no paramiko. SmartSSHClientMedium will break.
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
# Identify the protocol version.
81
bytes = self._get_bytes(2)
82
if bytes.startswith('2\n'):
83
protocol_class = SmartServerRequestProtocolTwo
86
protocol_class = SmartServerRequestProtocolOne
87
protocol = protocol_class(self.backing_transport, self._write_out)
88
protocol.accept_bytes(bytes)
91
def _serve_one_request(self, protocol):
92
"""Read one request from input, process, send back a response.
94
:param protocol: a SmartServerRequestProtocol.
97
self._serve_one_request_unguarded(protocol)
98
except KeyboardInterrupt:
101
self.terminate_due_to_error()
103
def terminate_due_to_error(self):
104
"""Called when an unhandled exception from the protocol occurs."""
105
raise NotImplementedError(self.terminate_due_to_error)
107
def _get_bytes(self, desired_count):
108
"""Get some bytes from the medium.
110
:param desired_count: number of bytes we want to read.
112
raise NotImplementedError(self._get_bytes)
115
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
117
def __init__(self, sock, backing_transport):
120
:param sock: the socket the server will read from. It will be put
123
SmartServerStreamMedium.__init__(self, backing_transport)
125
sock.setblocking(True)
128
def _serve_one_request_unguarded(self, protocol):
129
while protocol.next_read_size():
131
protocol.accept_bytes(self.push_back)
134
bytes = self._get_bytes(4096)
138
protocol.accept_bytes(bytes)
140
self.push_back = protocol.excess_buffer
142
def _get_bytes(self, desired_count):
143
# We ignore the desired_count because on sockets it's more efficient to
145
return self.socket.recv(4096)
147
def terminate_due_to_error(self):
148
"""Called when an unhandled exception from the protocol occurs."""
149
# TODO: This should log to a server log file, but no such thing
150
# exists yet. Andrew Bennetts 2006-09-29.
154
def _write_out(self, bytes):
155
self.socket.sendall(bytes)
158
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
160
def __init__(self, in_file, out_file, backing_transport):
161
"""Construct new server.
163
:param in_file: Python file from which requests can be read.
164
:param out_file: Python file to write responses.
165
:param backing_transport: Transport for the directory served.
167
SmartServerStreamMedium.__init__(self, backing_transport)
168
if sys.platform == 'win32':
169
# force binary mode for files
171
for f in (in_file, out_file):
172
fileno = getattr(f, 'fileno', None)
174
msvcrt.setmode(fileno(), os.O_BINARY)
178
def _serve_one_request_unguarded(self, protocol):
180
bytes_to_read = protocol.next_read_size()
181
if bytes_to_read == 0:
182
# Finished serving this request.
185
bytes = self._get_bytes(bytes_to_read)
187
# Connection has been closed.
191
protocol.accept_bytes(bytes)
193
def _get_bytes(self, desired_count):
194
return self._in.read(desired_count)
196
def terminate_due_to_error(self):
197
# TODO: This should log to a server log file, but no such thing
198
# exists yet. Andrew Bennetts 2006-09-29.
202
def _write_out(self, bytes):
203
self._out.write(bytes)
206
class SmartClientMediumRequest(object):
207
"""A request on a SmartClientMedium.
209
Each request allows bytes to be provided to it via accept_bytes, and then
210
the response bytes to be read via read_bytes.
213
request.accept_bytes('123')
214
request.finished_writing()
215
result = request.read_bytes(3)
216
request.finished_reading()
218
It is up to the individual SmartClientMedium whether multiple concurrent
219
requests can exist. See SmartClientMedium.get_request to obtain instances
220
of SmartClientMediumRequest, and the concrete Medium you are using for
221
details on concurrency and pipelining.
224
def __init__(self, medium):
225
"""Construct a SmartClientMediumRequest for the medium medium."""
226
self._medium = medium
227
# we track state by constants - we may want to use the same
228
# pattern as BodyReader if it gets more complex.
229
# valid states are: "writing", "reading", "done"
230
self._state = "writing"
232
def accept_bytes(self, bytes):
233
"""Accept bytes for inclusion in this request.
235
This method may not be be called after finished_writing() has been
236
called. It depends upon the Medium whether or not the bytes will be
237
immediately transmitted. Message based Mediums will tend to buffer the
238
bytes until finished_writing() is called.
240
:param bytes: A bytestring.
242
if self._state != "writing":
243
raise errors.WritingCompleted(self)
244
self._accept_bytes(bytes)
246
def _accept_bytes(self, bytes):
247
"""Helper for accept_bytes.
249
Accept_bytes checks the state of the request to determing if bytes
250
should be accepted. After that it hands off to _accept_bytes to do the
253
raise NotImplementedError(self._accept_bytes)
255
def finished_reading(self):
256
"""Inform the request that all desired data has been read.
258
This will remove the request from the pipeline for its medium (if the
259
medium supports pipelining) and any further calls to methods on the
260
request will raise ReadingCompleted.
262
if self._state == "writing":
263
raise errors.WritingNotComplete(self)
264
if self._state != "reading":
265
raise errors.ReadingCompleted(self)
267
self._finished_reading()
269
def _finished_reading(self):
270
"""Helper for finished_reading.
272
finished_reading checks the state of the request to determine if
273
finished_reading is allowed, and if it is hands off to _finished_reading
274
to perform the action.
276
raise NotImplementedError(self._finished_reading)
278
def finished_writing(self):
279
"""Finish the writing phase of this request.
281
This will flush all pending data for this request along the medium.
282
After calling finished_writing, you may not call accept_bytes anymore.
284
if self._state != "writing":
285
raise errors.WritingCompleted(self)
286
self._state = "reading"
287
self._finished_writing()
289
def _finished_writing(self):
290
"""Helper for finished_writing.
292
finished_writing checks the state of the request to determine if
293
finished_writing is allowed, and if it is hands off to _finished_writing
294
to perform the action.
296
raise NotImplementedError(self._finished_writing)
298
def read_bytes(self, count):
299
"""Read bytes from this requests response.
301
This method will block and wait for count bytes to be read. It may not
302
be invoked until finished_writing() has been called - this is to ensure
303
a message-based approach to requests, for compatability with message
304
based mediums like HTTP.
306
if self._state == "writing":
307
raise errors.WritingNotComplete(self)
308
if self._state != "reading":
309
raise errors.ReadingCompleted(self)
310
return self._read_bytes(count)
312
def _read_bytes(self, count):
313
"""Helper for read_bytes.
315
read_bytes checks the state of the request to determing if bytes
316
should be read. After that it hands off to _read_bytes to do the
319
raise NotImplementedError(self._read_bytes)
322
class SmartClientMedium(object):
323
"""Smart client is a medium for sending smart protocol requests over."""
325
def disconnect(self):
326
"""If this medium maintains a persistent connection, close it.
328
The default implementation does nothing.
332
class SmartClientStreamMedium(SmartClientMedium):
333
"""Stream based medium common class.
335
SmartClientStreamMediums operate on a stream. All subclasses use a common
336
SmartClientStreamMediumRequest for their requests, and should implement
337
_accept_bytes and _read_bytes to allow the request objects to send and
342
self._current_request = None
344
def accept_bytes(self, bytes):
345
self._accept_bytes(bytes)
348
"""The SmartClientStreamMedium knows how to close the stream when it is
354
"""Flush the output stream.
356
This method is used by the SmartClientStreamMediumRequest to ensure that
357
all data for a request is sent, to avoid long timeouts or deadlocks.
359
raise NotImplementedError(self._flush)
361
def get_request(self):
362
"""See SmartClientMedium.get_request().
364
SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
367
return SmartClientStreamMediumRequest(self)
369
def read_bytes(self, count):
370
return self._read_bytes(count)
373
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
374
"""A client medium using simple pipes.
376
This client does not manage the pipes: it assumes they will always be open.
379
def __init__(self, readable_pipe, writeable_pipe):
380
SmartClientStreamMedium.__init__(self)
381
self._readable_pipe = readable_pipe
382
self._writeable_pipe = writeable_pipe
384
def _accept_bytes(self, bytes):
385
"""See SmartClientStreamMedium.accept_bytes."""
386
self._writeable_pipe.write(bytes)
389
"""See SmartClientStreamMedium._flush()."""
390
self._writeable_pipe.flush()
392
def _read_bytes(self, count):
393
"""See SmartClientStreamMedium._read_bytes."""
394
return self._readable_pipe.read(count)
397
class SmartSSHClientMedium(SmartClientStreamMedium):
398
"""A client medium using SSH."""
400
def __init__(self, host, port=None, username=None, password=None,
402
"""Creates a client that will connect on the first use.
404
:param vendor: An optional override for the ssh vendor to use. See
405
bzrlib.transport.ssh for details on ssh vendors.
407
SmartClientStreamMedium.__init__(self)
408
self._connected = False
410
self._password = password
412
self._username = username
413
self._read_from = None
414
self._ssh_connection = None
415
self._vendor = vendor
416
self._write_to = None
418
def _accept_bytes(self, bytes):
419
"""See SmartClientStreamMedium.accept_bytes."""
420
self._ensure_connection()
421
self._write_to.write(bytes)
423
def disconnect(self):
424
"""See SmartClientMedium.disconnect()."""
425
if not self._connected:
427
self._read_from.close()
428
self._write_to.close()
429
self._ssh_connection.close()
430
self._connected = False
432
def _ensure_connection(self):
433
"""Connect this medium if not already connected."""
436
executable = os.environ.get('BZR_REMOTE_PATH', 'bzr')
437
if self._vendor is None:
438
vendor = ssh._get_ssh_vendor()
440
vendor = self._vendor
441
self._ssh_connection = vendor.connect_ssh(self._username,
442
self._password, self._host, self._port,
443
command=[executable, 'serve', '--inet', '--directory=/',
445
self._read_from, self._write_to = \
446
self._ssh_connection.get_filelike_channels()
447
self._connected = True
450
"""See SmartClientStreamMedium._flush()."""
451
self._write_to.flush()
453
def _read_bytes(self, count):
454
"""See SmartClientStreamMedium.read_bytes."""
455
if not self._connected:
456
raise errors.MediumNotConnected(self)
457
return self._read_from.read(count)
460
class SmartTCPClientMedium(SmartClientStreamMedium):
461
"""A client medium using TCP."""
463
def __init__(self, host, port):
464
"""Creates a client that will connect on the first use."""
465
SmartClientStreamMedium.__init__(self)
466
self._connected = False
471
def _accept_bytes(self, bytes):
472
"""See SmartClientMedium.accept_bytes."""
473
self._ensure_connection()
474
self._socket.sendall(bytes)
476
def disconnect(self):
477
"""See SmartClientMedium.disconnect()."""
478
if not self._connected:
482
self._connected = False
484
def _ensure_connection(self):
485
"""Connect this medium if not already connected."""
488
self._socket = socket.socket()
489
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
490
result = self._socket.connect_ex((self._host, int(self._port)))
492
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
493
(self._host, self._port, os.strerror(result)))
494
self._connected = True
497
"""See SmartClientStreamMedium._flush().
499
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
500
add a means to do a flush, but that can be done in the future.
503
def _read_bytes(self, count):
504
"""See SmartClientMedium.read_bytes."""
505
if not self._connected:
506
raise errors.MediumNotConnected(self)
507
return self._socket.recv(count)
510
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
511
"""A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
513
def __init__(self, medium):
514
SmartClientMediumRequest.__init__(self, medium)
515
# check that we are safe concurrency wise. If some streams start
516
# allowing concurrent requests - i.e. via multiplexing - then this
517
# assert should be moved to SmartClientStreamMedium.get_request,
518
# and the setting/unsetting of _current_request likewise moved into
519
# that class : but its unneeded overhead for now. RBC 20060922
520
if self._medium._current_request is not None:
521
raise errors.TooManyConcurrentRequests(self._medium)
522
self._medium._current_request = self
524
def _accept_bytes(self, bytes):
525
"""See SmartClientMediumRequest._accept_bytes.
527
This forwards to self._medium._accept_bytes because we are operating
528
on the mediums stream.
530
self._medium._accept_bytes(bytes)
532
def _finished_reading(self):
533
"""See SmartClientMediumRequest._finished_reading.
535
This clears the _current_request on self._medium to allow a new
536
request to be created.
538
assert self._medium._current_request is self
539
self._medium._current_request = None
541
def _finished_writing(self):
542
"""See SmartClientMediumRequest._finished_writing.
544
This invokes self._medium._flush to ensure all bytes are transmitted.
546
self._medium._flush()
548
def _read_bytes(self, count):
549
"""See SmartClientMediumRequest._read_bytes.
551
This forwards to self._medium._read_bytes because we are operating
552
on the mediums stream.
554
return self._medium._read_bytes(count)
added
removed
bzrlib/smart/medium.py
