1 # Copyright (c) 2010, 2011, 2012 Nicira, Inc.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at:
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
21 import ovs.socket_util
24 vlog = ovs.vlog.Vlog("stream")
27 def stream_or_pstream_needs_probes(name):
28 """ 1 if the stream or pstream specified by 'name' needs periodic probes to
29 verify connectivity. For [p]streams which need probes, it can take a long
30 time to notice the connection was dropped. Returns 0 if probes aren't
31 needed, and -1 if 'name' is invalid"""
33 if PassiveStream.is_valid_name(name) or Stream.is_valid_name(name):
34 # Only unix and punix are supported currently.
41 """Bidirectional byte stream. Currently only Unix domain sockets
49 # Kinds of events that one might wait for.
50 W_CONNECT = 0 # Connect complete (success or failure).
51 W_RECV = 1 # Data received.
52 W_SEND = 2 # Send buffer room available.
55 def is_valid_name(name):
56 """Returns True if 'name' is a stream name in the form "TYPE:ARGS" and
57 TYPE is a supported stream type (currently only "unix:"), otherwise
59 return name.startswith("unix:")
61 def __init__(self, socket, name, status):
64 if status == errno.EAGAIN:
65 self.state = Stream.__S_CONNECTING
67 self.state = Stream.__S_CONNECTED
69 self.state = Stream.__S_DISCONNECTED
75 """Attempts to connect a stream to a remote peer. 'name' is a
76 connection name in the form "TYPE:ARGS", where TYPE is an active stream
77 class's name and ARGS are stream class-specific. Currently the only
78 supported TYPE is "unix".
80 Returns (error, stream): on success 'error' is 0 and 'stream' is the
81 new Stream, on failure 'error' is a positive errno value and 'stream'
84 Never returns errno.EAGAIN or errno.EINPROGRESS. Instead, returns 0
85 and a new Stream. The connect() method can be used to check for
86 successful connection completion."""
87 if not Stream.is_valid_name(name):
88 return errno.EAFNOSUPPORT, None
90 connect_path = name[5:]
91 error, sock = ovs.socket_util.make_unix_socket(socket.SOCK_STREAM,
97 status = ovs.socket_util.check_connection_completion(sock)
98 return 0, Stream(sock, name, status)
101 def open_block((error, stream)):
102 """Blocks until a Stream completes its connection attempt, either
103 succeeding or failing. (error, stream) should be the tuple returned by
104 Stream.open(). Returns a tuple of the same form.
107 error, stream = Stream.open_block(Stream.open("unix:/tmp/socket"))"""
111 error = stream.connect()
112 if error != errno.EAGAIN:
115 poller = ovs.poller.Poller()
116 stream.run_wait(poller)
117 stream.connect_wait(poller)
119 assert error != errno.EINPROGRESS
129 def __scs_connecting(self):
130 retval = ovs.socket_util.check_connection_completion(self.socket)
131 assert retval != errno.EINPROGRESS
133 self.state = Stream.__S_CONNECTED
134 elif retval != errno.EAGAIN:
135 self.state = Stream.__S_DISCONNECTED
139 """Tries to complete the connection on this stream. If the connection
140 is complete, returns 0 if the connection was successful or a positive
141 errno value if it failed. If the connection is still in progress,
142 returns errno.EAGAIN."""
143 last_state = -1 # Always differs from initial self.state
144 while self.state != last_state:
145 last_state = self.state
146 if self.state == Stream.__S_CONNECTING:
147 self.__scs_connecting()
148 elif self.state == Stream.__S_CONNECTED:
150 elif self.state == Stream.__S_DISCONNECTED:
154 """Tries to receive up to 'n' bytes from this stream. Returns a
155 (error, string) tuple:
157 - If successful, 'error' is zero and 'string' contains between 1
158 and 'n' bytes of data.
160 - On error, 'error' is a positive errno value.
162 - If the connection has been closed in the normal fashion or if 'n'
163 is 0, the tuple is (0, "").
165 The recv function will not block waiting for data to arrive. If no
166 data have been received, it returns (errno.EAGAIN, "") immediately."""
168 retval = self.connect()
175 return (0, self.socket.recv(n))
176 except socket.error, e:
177 return (ovs.socket_util.get_exception_errno(e), "")
180 """Tries to send 'buf' on this stream.
182 If successful, returns the number of bytes sent, between 1 and
183 len(buf). 0 is only a valid return value if len(buf) is 0.
185 On error, returns a negative errno value.
187 Will not block. If no bytes can be immediately accepted for
188 transmission, returns -errno.EAGAIN immediately."""
190 retval = self.connect()
197 return self.socket.send(buf)
198 except socket.error, e:
199 return -ovs.socket_util.get_exception_errno(e)
204 def run_wait(self, poller):
207 def wait(self, poller, wait):
208 assert wait in (Stream.W_CONNECT, Stream.W_RECV, Stream.W_SEND)
210 if self.state == Stream.__S_DISCONNECTED:
211 poller.immediate_wake()
214 if self.state == Stream.__S_CONNECTING:
215 wait = Stream.W_CONNECT
216 if wait == Stream.W_RECV:
217 poller.fd_wait(self.socket, select.POLLIN)
219 poller.fd_wait(self.socket, select.POLLOUT)
221 def connect_wait(self, poller):
222 self.wait(poller, Stream.W_CONNECT)
224 def recv_wait(self, poller):
225 self.wait(poller, Stream.W_RECV)
227 def send_wait(self, poller):
228 self.wait(poller, Stream.W_SEND)
231 # Don't delete the file: we might have forked.
235 class PassiveStream(object):
237 def is_valid_name(name):
238 """Returns True if 'name' is a passive stream name in the form
239 "TYPE:ARGS" and TYPE is a supported passive stream type (currently only
240 "punix:"), otherwise False."""
241 return name.startswith("punix:")
243 def __init__(self, sock, name, bind_path):
246 self.bind_path = bind_path
250 """Attempts to start listening for remote stream connections. 'name'
251 is a connection name in the form "TYPE:ARGS", where TYPE is an passive
252 stream class's name and ARGS are stream class-specific. Currently the
253 only supported TYPE is "punix".
255 Returns (error, pstream): on success 'error' is 0 and 'pstream' is the
256 new PassiveStream, on failure 'error' is a positive errno value and
257 'pstream' is None."""
258 if not PassiveStream.is_valid_name(name):
259 return errno.EAFNOSUPPORT, None
262 error, sock = ovs.socket_util.make_unix_socket(socket.SOCK_STREAM,
263 True, bind_path, None)
269 except socket.error, e:
270 vlog.err("%s: listen: %s" % (name, os.strerror(e.error)))
274 return 0, PassiveStream(sock, name, bind_path)
277 """Closes this PassiveStream."""
279 if self.bind_path is not None:
280 ovs.fatal_signal.unlink_file_now(self.bind_path)
281 self.bind_path = None
284 """Tries to accept a new connection on this passive stream. Returns
285 (error, stream): if successful, 'error' is 0 and 'stream' is the new
286 Stream object, and on failure 'error' is a positive errno value and
289 Will not block waiting for a connection. If no connection is ready to
290 be accepted, returns (errno.EAGAIN, None) immediately."""
294 sock, addr = self.socket.accept()
295 ovs.socket_util.set_nonblocking(sock)
296 return 0, Stream(sock, "unix:%s" % addr, 0)
297 except socket.error, e:
298 error = ovs.socket_util.get_exception_errno(e)
299 if error != errno.EAGAIN:
301 vlog.dbg("accept: %s" % os.strerror(error))
304 def wait(self, poller):
305 poller.fd_wait(self.socket, select.POLLIN)
308 # Don't delete the file: we might have forked.
314 Active %s connection methods:
315 unix:FILE Unix domain socket named FILE
317 Passive %s connection methods:
318 punix:FILE Listen on Unix domain socket FILE""" % (name, name)