1
# Copyright (C) 2006,2007 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.
30
from bzrlib import errors
31
from bzrlib.smart.protocol import (
32
SmartServerRequestProtocolOne,
33
SmartServerRequestProtocolTwo,
37
from bzrlib.transport import ssh
38
except errors.ParamikoNotPresent:
39
# no paramiko. SmartSSHClientMedium will break.
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):
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
66
"""Serve requests until the client disconnects."""
67
# Keep a reference to stderr because the sys module's globals get set to
68
# None during interpreter shutdown.
69
from sys import stderr
71
while not self.finished:
72
protocol = self._build_protocol()
73
self._serve_one_request(protocol)
75
stderr.write("%s terminating on exception %s\n" % (self, e))
78
def _build_protocol(self):
79
# Identify the protocol version.
80
bytes = self._get_bytes(2)
81
if bytes.startswith('2\x01'):
82
protocol_class = SmartServerRequestProtocolTwo
85
protocol_class = SmartServerRequestProtocolOne
86
protocol = protocol_class(self.backing_transport, self._write_out)
87
protocol.accept_bytes(bytes)
90
def _serve_one_request(self, protocol):
91
"""Read one request from input, process, send back a response.
93
:param protocol: a SmartServerRequestProtocol.
96
self._serve_one_request_unguarded(protocol)
97
except KeyboardInterrupt:
100
self.terminate_due_to_error()
102
def terminate_due_to_error(self):
103
"""Called when an unhandled exception from the protocol occurs."""
104
raise NotImplementedError(self.terminate_due_to_error)
106
def _get_bytes(self, desired_count):
107
"""Get some bytes from the medium.
109
:param desired_count: number of bytes we want to read.
111
raise NotImplementedError(self._get_bytes)
114
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
116
def __init__(self, sock, backing_transport):
119
:param sock: the socket the server will read from. It will be put
122
SmartServerStreamMedium.__init__(self, backing_transport)
124
sock.setblocking(True)
127
def _serve_one_request_unguarded(self, protocol):
128
while protocol.next_read_size():
130
protocol.accept_bytes(self.push_back)
133
bytes = self._get_bytes(4096)
137
protocol.accept_bytes(bytes)
139
self.push_back = protocol.excess_buffer
141
def _get_bytes(self, desired_count):
142
# We ignore the desired_count because on sockets it's more efficient to
144
return self.socket.recv(4096)
146
def terminate_due_to_error(self):
147
"""Called when an unhandled exception from the protocol occurs."""
148
# TODO: This should log to a server log file, but no such thing
149
# exists yet. Andrew Bennetts 2006-09-29.
153
def _write_out(self, bytes):
154
self.socket.sendall(bytes)
157
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
159
def __init__(self, in_file, out_file, backing_transport):
160
"""Construct new server.
162
:param in_file: Python file from which requests can be read.
163
:param out_file: Python file to write responses.
164
:param backing_transport: Transport for the directory served.
166
SmartServerStreamMedium.__init__(self, backing_transport)
167
if sys.platform == 'win32':
168
# force binary mode for files
170
for f in (in_file, out_file):
171
fileno = getattr(f, 'fileno', None)
173
msvcrt.setmode(fileno(), os.O_BINARY)
177
def _serve_one_request_unguarded(self, protocol):
179
bytes_to_read = protocol.next_read_size()
180
if bytes_to_read == 0:
181
# Finished serving this request.
184
bytes = self._get_bytes(bytes_to_read)
186
# Connection has been closed.
190
protocol.accept_bytes(bytes)
192
def _get_bytes(self, desired_count):
193
return self._in.read(desired_count)
195
def terminate_due_to_error(self):
196
# TODO: This should log to a server log file, but no such thing
197
# exists yet. Andrew Bennetts 2006-09-29.
201
def _write_out(self, bytes):
202
self._out.write(bytes)
205
class SmartClientMediumRequest(object):
206
"""A request on a SmartClientMedium.
208
Each request allows bytes to be provided to it via accept_bytes, and then
209
the response bytes to be read via read_bytes.
212
request.accept_bytes('123')
213
request.finished_writing()
214
result = request.read_bytes(3)
215
request.finished_reading()
217
It is up to the individual SmartClientMedium whether multiple concurrent
218
requests can exist. See SmartClientMedium.get_request to obtain instances
219
of SmartClientMediumRequest, and the concrete Medium you are using for
220
details on concurrency and pipelining.
223
def __init__(self, medium):
224
"""Construct a SmartClientMediumRequest for the medium medium."""
225
self._medium = medium
226
# we track state by constants - we may want to use the same
227
# pattern as BodyReader if it gets more complex.
228
# valid states are: "writing", "reading", "done"
229
self._state = "writing"
231
def accept_bytes(self, bytes):
232
"""Accept bytes for inclusion in this request.
234
This method may not be be called after finished_writing() has been
235
called. It depends upon the Medium whether or not the bytes will be
236
immediately transmitted. Message based Mediums will tend to buffer the
237
bytes until finished_writing() is called.
239
:param bytes: A bytestring.
241
if self._state != "writing":
242
raise errors.WritingCompleted(self)
243
self._accept_bytes(bytes)
245
def _accept_bytes(self, bytes):
246
"""Helper for accept_bytes.
248
Accept_bytes checks the state of the request to determing if bytes
249
should be accepted. After that it hands off to _accept_bytes to do the
252
raise NotImplementedError(self._accept_bytes)
254
def finished_reading(self):
255
"""Inform the request that all desired data has been read.
257
This will remove the request from the pipeline for its medium (if the
258
medium supports pipelining) and any further calls to methods on the
259
request will raise ReadingCompleted.
261
if self._state == "writing":
262
raise errors.WritingNotComplete(self)
263
if self._state != "reading":
264
raise errors.ReadingCompleted(self)
266
self._finished_reading()
268
def _finished_reading(self):
269
"""Helper for finished_reading.
271
finished_reading checks the state of the request to determine if
272
finished_reading is allowed, and if it is hands off to _finished_reading
273
to perform the action.
275
raise NotImplementedError(self._finished_reading)
277
def finished_writing(self):
278
"""Finish the writing phase of this request.
280
This will flush all pending data for this request along the medium.
281
After calling finished_writing, you may not call accept_bytes anymore.
283
if self._state != "writing":
284
raise errors.WritingCompleted(self)
285
self._state = "reading"
286
self._finished_writing()
288
def _finished_writing(self):
289
"""Helper for finished_writing.
291
finished_writing checks the state of the request to determine if
292
finished_writing is allowed, and if it is hands off to _finished_writing
293
to perform the action.
295
raise NotImplementedError(self._finished_writing)
297
def read_bytes(self, count):
298
"""Read bytes from this requests response.
300
This method will block and wait for count bytes to be read. It may not
301
be invoked until finished_writing() has been called - this is to ensure
302
a message-based approach to requests, for compatability with message
303
based mediums like HTTP.
305
if self._state == "writing":
306
raise errors.WritingNotComplete(self)
307
if self._state != "reading":
308
raise errors.ReadingCompleted(self)
309
return self._read_bytes(count)
311
def _read_bytes(self, count):
312
"""Helper for read_bytes.
314
read_bytes checks the state of the request to determing if bytes
315
should be read. After that it hands off to _read_bytes to do the
318
raise NotImplementedError(self._read_bytes)
321
class SmartClientMedium(object):
322
"""Smart client is a medium for sending smart protocol requests over."""
324
def disconnect(self):
325
"""If this medium maintains a persistent connection, close it.
327
The default implementation does nothing.
331
class SmartClientStreamMedium(SmartClientMedium):
332
"""Stream based medium common class.
334
SmartClientStreamMediums operate on a stream. All subclasses use a common
335
SmartClientStreamMediumRequest for their requests, and should implement
336
_accept_bytes and _read_bytes to allow the request objects to send and
341
self._current_request = None
343
def accept_bytes(self, bytes):
344
self._accept_bytes(bytes)
347
"""The SmartClientStreamMedium knows how to close the stream when it is
353
"""Flush the output stream.
355
This method is used by the SmartClientStreamMediumRequest to ensure that
356
all data for a request is sent, to avoid long timeouts or deadlocks.
358
raise NotImplementedError(self._flush)
360
def get_request(self):
361
"""See SmartClientMedium.get_request().
363
SmartClientStreamMedium always returns a SmartClientStreamMediumRequest
366
return SmartClientStreamMediumRequest(self)
368
def read_bytes(self, count):
369
return self._read_bytes(count)
372
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
373
"""A client medium using simple pipes.
375
This client does not manage the pipes: it assumes they will always be open.
378
def __init__(self, readable_pipe, writeable_pipe):
379
SmartClientStreamMedium.__init__(self)
380
self._readable_pipe = readable_pipe
381
self._writeable_pipe = writeable_pipe
383
def _accept_bytes(self, bytes):
384
"""See SmartClientStreamMedium.accept_bytes."""
385
self._writeable_pipe.write(bytes)
388
"""See SmartClientStreamMedium._flush()."""
389
self._writeable_pipe.flush()
391
def _read_bytes(self, count):
392
"""See SmartClientStreamMedium._read_bytes."""
393
return self._readable_pipe.read(count)
396
class SmartSSHClientMedium(SmartClientStreamMedium):
397
"""A client medium using SSH."""
399
def __init__(self, host, port=None, username=None, password=None,
401
"""Creates a client that will connect on the first use.
403
:param vendor: An optional override for the ssh vendor to use. See
404
bzrlib.transport.ssh for details on ssh vendors.
406
SmartClientStreamMedium.__init__(self)
407
self._connected = False
409
self._password = password
411
self._username = username
412
self._read_from = None
413
self._ssh_connection = None
414
self._vendor = vendor
415
self._write_to = None
417
def _accept_bytes(self, bytes):
418
"""See SmartClientStreamMedium.accept_bytes."""
419
self._ensure_connection()
420
self._write_to.write(bytes)
422
def disconnect(self):
423
"""See SmartClientMedium.disconnect()."""
424
if not self._connected:
426
self._read_from.close()
427
self._write_to.close()
428
self._ssh_connection.close()
429
self._connected = False
431
def _ensure_connection(self):
432
"""Connect this medium if not already connected."""
435
executable = os.environ.get('BZR_REMOTE_PATH', 'bzr')
436
if self._vendor is None:
437
vendor = ssh._get_ssh_vendor()
439
vendor = self._vendor
440
self._ssh_connection = vendor.connect_ssh(self._username,
441
self._password, self._host, self._port,
442
command=[executable, 'serve', '--inet', '--directory=/',
444
self._read_from, self._write_to = \
445
self._ssh_connection.get_filelike_channels()
446
self._connected = True
449
"""See SmartClientStreamMedium._flush()."""
450
self._write_to.flush()
452
def _read_bytes(self, count):
453
"""See SmartClientStreamMedium.read_bytes."""
454
if not self._connected:
455
raise errors.MediumNotConnected(self)
456
return self._read_from.read(count)
459
class SmartTCPClientMedium(SmartClientStreamMedium):
460
"""A client medium using TCP."""
462
def __init__(self, host, port):
463
"""Creates a client that will connect on the first use."""
464
SmartClientStreamMedium.__init__(self)
465
self._connected = False
470
def _accept_bytes(self, bytes):
471
"""See SmartClientMedium.accept_bytes."""
472
self._ensure_connection()
473
self._socket.sendall(bytes)
475
def disconnect(self):
476
"""See SmartClientMedium.disconnect()."""
477
if not self._connected:
481
self._connected = False
483
def _ensure_connection(self):
484
"""Connect this medium if not already connected."""
487
self._socket = socket.socket()
488
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
489
result = self._socket.connect_ex((self._host, int(self._port)))
491
raise errors.ConnectionError("failed to connect to %s:%d: %s" %
492
(self._host, self._port, os.strerror(result)))
493
self._connected = True
496
"""See SmartClientStreamMedium._flush().
498
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
499
add a means to do a flush, but that can be done in the future.
502
def _read_bytes(self, count):
503
"""See SmartClientMedium.read_bytes."""
504
if not self._connected:
505
raise errors.MediumNotConnected(self)
506
return self._socket.recv(count)
509
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
510
"""A SmartClientMediumRequest that works with an SmartClientStreamMedium."""
512
def __init__(self, medium):
513
SmartClientMediumRequest.__init__(self, medium)
514
# check that we are safe concurrency wise. If some streams start
515
# allowing concurrent requests - i.e. via multiplexing - then this
516
# assert should be moved to SmartClientStreamMedium.get_request,
517
# and the setting/unsetting of _current_request likewise moved into
518
# that class : but its unneeded overhead for now. RBC 20060922
519
if self._medium._current_request is not None:
520
raise errors.TooManyConcurrentRequests(self._medium)
521
self._medium._current_request = self
523
def _accept_bytes(self, bytes):
524
"""See SmartClientMediumRequest._accept_bytes.
526
This forwards to self._medium._accept_bytes because we are operating
527
on the mediums stream.
529
self._medium._accept_bytes(bytes)
531
def _finished_reading(self):
532
"""See SmartClientMediumRequest._finished_reading.
534
This clears the _current_request on self._medium to allow a new
535
request to be created.
537
assert self._medium._current_request is self
538
self._medium._current_request = None
540
def _finished_writing(self):
541
"""See SmartClientMediumRequest._finished_writing.
543
This invokes self._medium._flush to ensure all bytes are transmitted.
545
self._medium._flush()
547
def _read_bytes(self, count):
548
"""See SmartClientMediumRequest._read_bytes.
550
This forwards to self._medium._read_bytes because we are operating
551
on the mediums stream.
553
return self._medium._read_bytes(count)
added
removed
bzrlib/smart/medium.py
