[stacklessexamples] r190 committed - - Clean up listen/accept to work correctly....
stackles...@googlecode.com
Author: richard.m.tew
Date: Thu Mar 1 18:08:21 2012
Log: - Clean up listen/accept to work correctly.
- Added 'recv' and 'send' suppport.
http://code.google.com/p/stacklessexamples/source/detail?r=190
Modified:
/trunk/libraries/stacklesslib/COPYING.txt
/trunk/libraries/stacklesslib/stacklesslib/replacements/socket_pyuv.py
=======================================
--- /trunk/libraries/stacklesslib/COPYING.txt Mon Dec 19 00:10:28 2011
+++ /trunk/libraries/stacklesslib/COPYING.txt Thu Mar 1 18:08:21 2012
@@ -1,5 +1,6 @@
Copyright (c) 2011 CCP Hf.
+Copyright (c) 2012 Richard Tew.
Permission is hereby granted, free of charge, to any person obtaining a
copy
of this software and associated documentation files (the "Software"), to
deal
=======================================
--- /trunk/libraries/stacklesslib/stacklesslib/replacements/socket_pyuv.py
Thu Mar 1 15:31:05 2012
+++ /trunk/libraries/stacklesslib/stacklesslib/replacements/socket_pyuv.py
Thu Mar 1 18:08:21 2012
@@ -2,6 +2,7 @@
Stackless compatible socket module (pyuv based).
Author: Richard Tew <richar...@gmail.com>
+This is licensed under the MIT open source license.
Feel free to email me with any questions, comments, or suggestions for
improvement.
@@ -12,22 +13,32 @@
- loop.run_once does some strange arbitrary calculations to determine
whether it should block forever waiting for events. There is no way
for
the caller to pass in a max timeout.
+- 'errno' error codes. There are special Winsock variants which are
expected
+ on Windows. In the case of EWOULDBLOCK, this is okay as this value is
the
+ correct value (the Winsock value) on Windows. But in the case of EINVAL,
+ the value is not that of WSAEINVAL. So we need special casing..
+- os.strerror(). This does not return useful messages on my computer.
+ So messages are hard-coded for now.
Todo list:
- TODO: Exceptions raised out of pyuv are custom ones that present custom
libuv error codes. We have a mapping of some of these to Windows
error codes below, but the exceptions need to be caught and
translated to the standard Python socket ones.
-- TODO: The listening needs to be made more standard. The actual 'listen'
- call should be made in the socket listen function, and accept
- should handle getting a "won't block" exception and only
- cooperate with the listen callback when it has to be awakened.
"""
import errno
import socket as stdsocket # We need the "socket" name for the function we
export.
+import sys
import weakref
+# Locally import and adjust error numbers.
+from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, \
+ ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED, \
+ ECONNREFUSED, EWOULDBLOCK, EINVAL
+if sys.platform == "win32":
+ EINVAL = errno.WSAEINVAL
+
import pyuv
import stackless
@@ -61,6 +72,7 @@
pass
return d
_errno_map = convert_uv_errnos()
+_EWOULDBLOCK_text = "The socket operation could not complete without
blocking"
_poll_interval = 0.05
_pumping = False
@@ -105,13 +117,15 @@
class socket(object):
# Optionally overriden variables.
+ _accept_channel = None
_blocking = True
- _listen_backlog = None
- _listen_channel = None
+ _connected = False
+ _listening = False
_opt_keepalive = DEFAULT_KEEPALIVE_FLAG
_opt_keepalive_delay = DEFAULT_KEEPALIVE_DELAY
_opt_nodelay = DEFAULT_NODELAY_FLAG
_timeout = None
+ _was_connected = False
# Official socket object functions.
def __init__(self, family=AF_INET, type=SOCK_STREAM, proto=0):
global _pyuv_loop
@@ -132,28 +146,34 @@
start_pumping()
def accept(self): # TCP # COMPLETE
"""
- The uv accept operation should be done in the callback. So this is
- the reason that listen() is a token call, and accept() has the
- actual listen initialisation done internally.
-
- This should be called in a loop, so that all values received out
- of it come directly from the 'pyuv.listen' callback and return
- directly back into it to let it exit.
"""
- if self._listen_channel is None:
- self._listen_channel = stackless.channel()
- def listen_callback(_listen_tcp_socket, err):
- if err is None:
- _new_tcp_socket = socket()
- _listen_tcp_socket.accept(_new_tcp_socket._tcp_socket)
- # Emulate the standard 'socket.accept' return value.
- ret = _new_tcp_socket, _new_tcp_socket.getpeername()
- self._listen_channel.send(ret)
- else:
- # TODO: Really should be able to pass multiple
arguments to the exception type..
- self._listen_channel.send_exception(stdsocket.error,
errno_map[err])
- self._tcp_socket.listen(listen_callback, self._listen_backlog)
- return self._listen_channel.receive()
+ def accept_result(_new_tcp_socket):
+ _new_tcp_socket._was_connected = True
+ # Emulate the standard 'socket.accept' return value.
+ return _new_tcp_socket, _new_tcp_socket.getpeername()
+ self._accept_channel = stackless.channel()
+ self._accept_channel.preference = 1
+ try:
+ # First actually try and do an accept.
+ _new_tcp_socket = socket()
+ try:
+ self._tcp_socket.accept(_new_tcp_socket._tcp_socket)
+ return accept_result(_new_tcp_socket)
+ except pyuv.error.TCPError:
+ # If listen has not been called yet.
+ if not self._listening:
+ raise stdsocket.error(EINVAL)
+ # If the socket is set to non-blocking.
+ if not self._blocking:
+ raise stdsocket.error(EWOULDBLOCK, _EWOULDBLOCK_text)
+ # Otherwise, assume all is well and block on a channel.
+ sys.exc_clear()
+ # Block until there is an incoming connection.
+ self._accept_channel.receive()
+ self._tcp_socket.accept(_new_tcp_socket._tcp_socket)
+ return accept_result(_new_tcp_socket)
+ finally:
+ self._accept_channel = None
def bind(self, address): # TCP / UDP # COMPLETE
self._socket.bind(address)
def close(self): # TCP / UDP # COMPLETE
@@ -161,6 +181,7 @@
Blocks until the close has completed.. correct behaviour?
"""
channel = stackless.channel()
+ channel.preference = 1
def close_callback(_socket_handle):
channel.send(None)
self._socket.close(close_callback)
@@ -168,15 +189,18 @@
def connect(self, address): # TCP # COMPLETE
err = self.connect_ex(address)
if err:
- raise stdsocket.error(err, errno.errorcode[err])
+ raise stdsocket.error(err)
def connect_ex(self, address): # COMPLETE
channel = stackless.channel()
+ channel.preference = 1
def connect_callback(_tcp_handle, err):
channel.send(err)
self._tcp_socket.connect(address, connect_callback)
err = channel.receive()
if err is None:
err = 0
+ self._connected = True
+ self._was_connected = True
return err
def fileno(self):
pass
@@ -186,22 +210,69 @@
return self._tcp_socket.getsockname()
def ioctl(self, control, option):
pass
- def listen(self, backlog): # TCP # COMPLETE
- self._listen_backlog = backlog
+ def listen(self, backlog): # TCP
+ def listen_callback(_listen_tcp_socket, err):
+ if self._accept_channel:
+ if err is None:
+ self._accept_channel.send(None)
+ else:
+ # TODO: Really should be able to pass multiple
arguments to the exception type..
+ self._accept_channel.send_exception(stdsocket.error,
errno_map[err])
+ self._tcp_socket.listen(listen_callback, backlog)
+ self._listening = True
def makefile(self, mode, bufsize):
pass
def recv(self, bufsize, flags=0):
- pass
+ if self.type != SOCK_DGRAM and not self._connected:
+ # Sockets which have never been connected do this.
+ if not self._was_connected:
+ raise stdsocket.error(ENOTCONN, 'Socket is not connected')
+
+ channel = stackless.channel()
+ channel.preference = 1
+ if self.type == SOCK_STREAM:
+ def tcp_callback(redundant_handle, data, err):
+ self._tcp_socket.stop_read()
+ if err is None:
+ channel.send(data)
+ else:
+ channel.send_exception(stdsocket.error, errno_map[err])
+ self._tcp_socket.start_read(tcp_callback)
+ else:
+ raise RuntimeError("No UDP handling yet")
+ return channel.receive()
def recvfrom(self, bufsize, flags=0):
- pass
+ channel = stackless.channel()
+ channel.preference = 1
+ if self.type == SOCK_DGRAM:
+ def udp_callback(redundant_handle, address, data, err):
+ self._udp_socket.stop_recv()
+ if err is None:
+ channel.send((data, address))
+ else:
+ channel.send_exception(stdsocket.error, errno_map[err])
+ self._udp_socket.start_recv(udp_callback)
+ else:
+ raise RuntimeError("No TCP handling yet")
+ return channel.receive()
def recvfrom_into(self, buffer, nbytes, flags=0):
pass
def recv_into(self, buffer, nbytes, flags=0):
pass
- def send(self, string, flags=0): # TCP / UDP
- pass
- def sendall(self, string, flags=0):
- pass
+ def send(self, string, flags=0): # TCP / UDP # COMPLETE
+ channel = stackless.channel()
+ channel.preference = 1
+ def write_callback(redundant_tcp_handle, err):
+ if err is None:
+ channel.send(None)
+ else:
+ # TODO: Really should be able to pass multiple arguments
to the exception type..
+ channel.send_exception(stdsocket.error, errno_map[err])
+ self._socket.write(string, write_callback)
+ channel.receive()
+ return len(string)
+ def sendall(self, string, flags=0): # COMPLETE
+ return self.send(string, flags)
def sendto(self, string, flags=0, address=None):
pass
def setblocking(self, flag): # COMPLETE
@@ -237,6 +308,7 @@
if how != stdsocket.SHUT_WR:
raise RuntimeError("Not supported")
channel = stackless.channel()
+ channel.preference = 1
def shutdown_callback(tcp_handle):
channel.send(None)
self._tcp_socket.shutdown(shutdown_callback)
@@ -251,4 +323,6 @@
@property
def proto(self):
return self._proto
-
+ # Custom internal logic.
+ def _is_non_blocking(self):
+ return not self._blocking or self._timeout == 0.0