加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
该仓库未声明开源许可证文件(LICENSE),使用请关注具体项目描述及其代码上游依赖。
克隆/下载
stacklesssocket.py 29.82 KB
一键复制 编辑 原始数据 按行查看 历史
Benjamin Jemlich 提交于 2010-09-13 04:57 . Initial implementation
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794
#
# Stackless compatible socket module.
#
# Author: Richard Tew <richard.m.tew@gmail.com>
#
# Feel free to email me with any questions, comments, or suggestions for
# improvement.
#
# Remaining work:
#
# = Asyncore does not add that much to this module. In fact, its
# limitations and differences between implementations in different Python
# versions just complicate things.
# = Select on Windows only handles 512 sockets at a time. So if there
# are more sockets than that, then they need to be separated and
# batched around this limitation.
# = It should be possible to have this wrap different mechanisms of
# asynchronous IO, from select to IO completion ports.
# = UDP support is mostly there due to the new hands off approach, but
# there are a few spots like handle_write and timeout handling, which need
# to be dealt with.
#
# Python standard library socket unit test state:
#
# - 2.5: Bad.
# - 2.6: Excellent (two UDP failures).
# - 2.7: Excellent (two UDP failures).
#
# This module is otherwise known to generally work for 2.5, 2.6 and 2.7.
#
# Small parts of this code were contributed back with permission from an
# internal version of this module in use at CCP Games.
#
import stackless
import asyncore, weakref, time, select, types
# If you pump the scheduler and wish to prevent the scheduler from staying
# non-empty for prolonged periods of time, If you do not pump the scheduler,
# you may however wish to prevent calls to poll() from running too long.
# Doing so gives all managed sockets a fairer chance at being read from,
# rather than paying prolonged attention to sockets with more incoming data.
#
# These values govern how long a poll() call spends at a given attempt
# of reading the data present on a given socket.
#
VALUE_MAX_NONBLOCKINGREAD_SIZE = 1000000
VALUE_MAX_NONBLOCKINGREAD_CALLS = 100
## Monkey-patching support..
# We need this so that sockets are cleared out when they are no longer in use.
# In fact, it is essential to correct operation of this code.
asyncore.socket_map = weakref.WeakValueDictionary()
import socket as stdsocket # We need the "socket" name for the function we export.
from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, \
ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED
# If we are to masquerade as the socket module, we need to provide the constants.
if "__all__" in stdsocket.__dict__:
__all__ = stdsocket.__dict__
for k, v in stdsocket.__dict__.iteritems():
if k in __all__:
globals()[k] = v
elif k == "EBADF":
globals()[k] = v
else:
for k, v in stdsocket.__dict__.iteritems():
if k.upper() == k:
globals()[k] = v
error = stdsocket.error
timeout = stdsocket.timeout
# WARNING: this function blocks and is not thread safe.
# The only solution is to spawn a thread to handle all
# getaddrinfo requests. Implementing a stackless DNS
# lookup service is only second best as getaddrinfo may
# use other methods.
getaddrinfo = stdsocket.getaddrinfo
# urllib2 apparently uses this directly. We need to cater for that.
_fileobject = stdsocket._fileobject
# Someone needs to invoke asyncore.poll() regularly to keep the socket
# data moving. The "ManageSockets" function here is a simple example
# of such a function. It is started by StartManager(), which uses the
# global "managerRunning" to ensure that no more than one copy is
# running.
#
# If you think you can do this better, register an alternative to
# StartManager using stacklesssocket_manager(). Your function will be
# called every time a new socket is created; it's your responsibility
# to ensure it doesn't start multiple copies of itself unnecessarily.
#
# By Nike: Added poll_interval on install to have it configurable from outside,
managerRunning = False
poll_interval = 0.05
def ManageSockets():
global managerRunning
try:
while len(asyncore.socket_map):
# Check the sockets for activity.
#print "POLL"
asyncore.poll(poll_interval)
# Yield to give other tasklets a chance to be scheduled.
_schedule_func()
finally:
managerRunning = False
def StartManager():
global managerRunning
if not managerRunning:
managerRunning = True
return stackless.tasklet(ManageSockets)()
_schedule_func = stackless.schedule
_manage_sockets_func = StartManager
_sleep_func = None
_timeout_func = None
def can_timeout():
return _sleep_func is not None or _timeout_func is not None
def stacklesssocket_manager(mgr):
global _manage_sockets_func
_manage_sockets_func = mgr
def socket(*args, **kwargs):
import sys
if "socket" in sys.modules and sys.modules["socket"] is not stdsocket:
raise RuntimeError("Use 'stacklesssocket.install' instead of replacing the 'socket' module")
_realsocket_old = stdsocket._realsocket
_socketobject_old = stdsocket._socketobject
class _socketobject_new(_socketobject_old):
def __init__(self, family=AF_INET, type=SOCK_STREAM, proto=0, _sock=None):
# We need to do this here.
if _sock is None:
_sock = _realsocket_old(family, type, proto)
_sock = _fakesocket(_sock)
_manage_sockets_func()
_socketobject_old.__init__(self, family, type, proto, _sock)
if not isinstance(self._sock, _fakesocket):
raise RuntimeError("bad socket")
def accept(self):
sock, addr = self._sock.accept()
sock = _fakesocket(sock)
sock.wasConnected = True
return _socketobject_new(_sock=sock), addr
accept.__doc__ = _socketobject_old.accept.__doc__
def install(pi=None):
global poll_interval
if stdsocket._realsocket is socket:
raise StandardError("Still installed")
stdsocket._realsocket = socket
stdsocket.socket = stdsocket.SocketType = stdsocket._socketobject = _socketobject_new
if pi is not None:
poll_interval = pi
def uninstall():
stdsocket._realsocket = _realsocket_old
stdsocket.socket = stdsocket.SocketType = stdsocket._socketobject = _socketobject_old
class _fakesocket(asyncore.dispatcher):
connectChannel = None
acceptChannel = None
wasConnected = False
_timeout = None
_blocking = True
lastReadChannelRef = None
lastReadArguments = None
lastReadTally = 0
lastReadCalls = 0
def __init__(self, realSocket):
# This is worth doing. I was passing in an invalid socket which
# was an instance of _fakesocket and it was causing tasklet death.
if not isinstance(realSocket, _realsocket_old):
raise StandardError("An invalid socket passed to fakesocket %s" % realSocket.__class__)
# This will register the real socket in the internal socket map.
asyncore.dispatcher.__init__(self, realSocket)
self.readQueue = []
self.writeQueue = []
self.sendToBuffers = []
if can_timeout():
self._timeout = stdsocket.getdefaulttimeout()
def receive_with_timeout(self, channel):
if self._timeout is not None:
# Start a timing out process.
# a) Engage a pre-existing external tasklet to send an exception on our channel if it has a receiver, if we are still there when it times out.
# b) Launch a tasklet that does a sleep, and sends an exception if we are still waiting, when it is awoken.
# Block waiting for a send.
if _timeout_func is not None:
# You will want to use this if you are using sockets in a different thread from your sleep functionality.
_timeout_func(self._timeout, channel, (timeout, "timed out"))
elif _sleep_func is not None:
stackless.tasklet(self._manage_receive_with_timeout)(channel)
else:
raise NotImplementedError("should not be here")
try:
ret = channel.receive()
except BaseException, e:
raise e
return ret
else:
return channel.receive()
def _manage_receive_with_timeout(self, channel):
if channel.balance < 0:
_sleep_func(self._timeout)
if channel.balance < 0:
channel.send_exception(timeout, "timed out")
def __del__(self):
# There are no more users (sockets or files) of this fake socket, we
# are safe to close it fully. If we don't, asyncore will choke on
# the weakref failures.
self.close()
# The asyncore version of this function depends on socket being set
# which is not the case when this fake socket has been closed.
def __getattr__(self, attr):
if not hasattr(self, "socket"):
raise AttributeError("socket attribute unset on '"+ attr +"' lookup")
return getattr(self.socket, attr)
## Asyncore potential activity indicators.
def readable(self):
if self.socket.type == SOCK_DGRAM:
return True
if len(self.readQueue):
return True
if self.acceptChannel is not None and self.acceptChannel.balance < 0:
return True
if self.connectChannel is not None and self.connectChannel.balance < 0:
return True
return False
def writable(self):
if self.socket.type != SOCK_DGRAM and not self.connected:
return True
if len(self.writeQueue):
return True
if len(self.sendToBuffers):
return True
return False
## Overriden socket methods.
def accept(self):
self._ensure_non_blocking_read()
if not self.acceptChannel:
self.acceptChannel = stackless.channel()
return self.receive_with_timeout(self.acceptChannel)
def connect(self, address):
"""
If a timeout is set for the connection attempt, and the timeout occurs
then it is the responsibility of the user to close the socket, should
they not wish the connection to potentially establish anyway.
"""
asyncore.dispatcher.connect(self, address)
# UDP sockets do not connect.
if self.socket.type != SOCK_DGRAM and not self.connected:
if not self.connectChannel:
self.connectChannel = stackless.channel()
# Prefer the sender. Do not block when sending, given that
# there is a tasklet known to be waiting, this will happen.
self.connectChannel.preference = 1
self.receive_with_timeout(self.connectChannel)
def _send(self, data, flags):
self._ensure_connected()
channel = stackless.channel()
channel.preference = 1 # Prefer the sender.
self.writeQueue.append((flags, data, channel))
return self.receive_with_timeout(channel)
def send(self, data, flags=0):
return self._send(data, flags)
def sendall(self, data, flags=0):
while len(data):
nbytes = self._send(data, flags)
if nbytes == 0:
raise Exception("completely unexpected situation, no data sent")
data = data[nbytes:]
def sendto(self, sendData, sendArg1=None, sendArg2=None):
# sendto(data, address)
# sendto(data [, flags], address)
if sendArg2 is not None:
flags = sendArg1
sendAddress = sendArg2
else:
flags = 0
sendAddress = sendArg1
waitChannel = None
for idx, (data, address, channel, sentBytes) in enumerate(self.sendToBuffers):
if address == sendAddress:
self.sendToBuffers[idx] = (data + sendData, address, channel, sentBytes)
waitChannel = channel
break
if waitChannel is None:
waitChannel = stackless.channel()
self.sendToBuffers.append((sendData, sendAddress, waitChannel, 0))
return self.receive_with_timeout(waitChannel)
def _recv(self, methodName, args, sizeIdx=0):
self._ensure_non_blocking_read()
if self._fileno is None:
return ""
if len(args) >= sizeIdx+1:
generalArgs = list(args)
generalArgs[sizeIdx] = 0
generalArgs = tuple(generalArgs)
else:
generalArgs = args
channelKey = methodName, generalArgs
#print self._fileno, "_recv:---ENTER---", channelKey
while True:
channel = None
if self.lastReadChannelRef is not None and self.lastReadTally < VALUE_MAX_NONBLOCKINGREAD_SIZE and self.lastReadCalls < VALUE_MAX_NONBLOCKINGREAD_CALLS:
if channelKey == self.lastReadArguments:
channel = self.lastReadChannelRef()
self.lastReadChannelRef = None
#elif self.lastReadTally >= VALUE_MAX_NONBLOCKINGREAD_SIZE or self.lastReadCalls >= VALUE_MAX_NONBLOCKINGREAD_CALLS:
#print "_recv:FORCE-CHANNEL-CHANGE %d %d" % (self.lastReadTally, self.lastReadCalls)
if channel is None:
channel = stackless.channel()
channel.preference = -1 # Prefer the receiver.
self.lastReadTally = self.lastReadCalls = 0
#print self._fileno, "_recv:NEW-CHANNEL", id(channel)
self.readQueue.append([ channel, methodName, args ])
else:
self.readQueue[0][2] = args
#print self._fileno, "_recv:RECYCLE-CHANNEL", id(channel), self.lastReadTally
try:
ret = self.receive_with_timeout(channel)
except stdsocket.error, e:
if isinstance(e, stdsocket.error) and e.args[0] == EWOULDBLOCK:
#print self._fileno, "_recv:BLOCK-RETRY", id(channel), "-" * 30
continue
else:
raise
break
self.lastReadChannelRef = weakref.ref(channel)
self.lastReadArguments = channelKey
if isinstance(ret, types.StringTypes):
self.lastReadTally += len(ret)
elif methodName == "recvfrom":
self.lastReadTally += len(ret[0])
elif methodName == "recvfrom_into":
self.lastReadTally += ret[0]
else:
self.lastReadTally += ret
self.lastReadCalls += 1
#print self._fileno, "_recv:---EXIT---", channelKey, len(ret), self.lastReadChannelRef()
return ret
def recv(self, *args):
if not self.connected:
# Sockets which have never been connected do this.
if not self.wasConnected:
raise error(10057, 'Socket is not connected')
return self._recv("recv", args)
def recv_into(self, *args):
if not self.connected:
# Sockets which have never been connected do this.
if not self.wasConnected:
raise error(10057, 'Socket is not connected')
return self._recv("recv_into", args, sizeIdx=1)
def recvfrom(self, *args):
return self._recv("recvfrom", args)
def recvfrom_into(self, *args):
return self._recv("recvfrom_into", args, sizeIdx=1)
def close(self):
if self._fileno is None:
return
asyncore.dispatcher.close(self)
self.connected = False
self.accepting = False
self.writeQueue = []
# Clear out all the channels with relevant errors.
while self.acceptChannel and self.acceptChannel.balance < 0:
self.acceptChannel.send_exception(error, 9, 'Bad file descriptor')
while self.connectChannel and self.connectChannel.balance < 0:
self.connectChannel.send_exception(error, 10061, 'Connection refused')
self._clear_read_queue()
def _clear_read_queue(self, *args):
for t in self.readQueue:
if t[0].balance < 0:
if len(args):
t[0].send_exception(*args)
else:
t[0].send("")
self.readQueue = []
# asyncore doesn't support this. Why not?
def fileno(self):
return self.socket.fileno()
def _is_non_blocking(self):
return not self._blocking or self._timeout == 0.0
def _ensure_non_blocking_read(self):
if self._is_non_blocking():
# Ensure there is something on the socket, before fetching it. Otherwise, error complaining.
r, w, e = select.select([ self ], [], [], 0.0)
if not r:
raise stdsocket.error(EWOULDBLOCK, "The socket operation could not complete without blocking")
def _ensure_connected(self):
if not self.connected:
# The socket was never connected.
if not self.wasConnected:
raise error(10057, "Socket is not connected")
# The socket has been closed already.
raise error(EBADF, 'Bad file descriptor')
def setblocking(self, flag):
self._blocking = flag
def gettimeout(self):
return self._timeout
def settimeout(self, value):
if value and not can_timeout():
raise RuntimeError("This is a stackless socket - to have timeout support you need to provide a sleep function")
self._timeout = value
def handle_accept(self):
if self.acceptChannel and self.acceptChannel.balance < 0:
t = asyncore.dispatcher.accept(self)
if t is None:
return
t[0].setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
stackless.tasklet(self.acceptChannel.send)(t)
# Inform the blocked connect call that the connection has been made.
def handle_connect(self):
if self.socket.type != SOCK_DGRAM:
if self.connectChannel and self.connectChannel.balance < 0:
self.wasConnected = True
self.connectChannel.send(None)
# Asyncore says its done but self.readBuffer may be non-empty
# so can't close yet. Do nothing and let 'recv' trigger the close.
def handle_close(self):
# This also gets called in the case that a non-blocking connect gets
# back to us with a no. If we don't reject the connect, then all
# connect calls that do not connect will block indefinitely.
if self.connectChannel is not None:
self.close()
# Some error, just close the channel and let that raise errors to
# blocked calls.
def handle_expt(self):
if False:
import traceback
print "handle_expt: START"
traceback.print_exc()
print "handle_expt: END"
self.close()
def handle_error(self):
self.close()
def handle_read(self):
"""
This will be called once per-poll call per socket with data in its buffer to be read.
If you call poll once every 30th of a second, then you are going to be rate limited
in terms of how fast you can read incoming data by the packet size they arrive in.
In order to deal with the worst case scenario, advantage is taken of how scheduling
works in order to keep reading until there is no more data left to read.
1. This function is called indicating data is present to read.
2. The desired amount is read and a send call is made on the channel with it.
3. The function is blocked on that action and the tasklet it is running in is reinserted into the scheduler.
4. The tasklet that made the read related socket call is awakened with the given data.
5. It returns the data to the function that made that call.
6. The function that made the call makes another read related socket call.
a) If the call is similar enough to the last call, then the previous channel is retrieved.
b) Otherwise, a new channel is created.
7. The tasklet that is making the read related socket call is blocked on the channel.
8. This tasklet that was blocked sending gets scheduled again.
a) If there is a tasklet blocked on the channel that it was using, then goto 2.
b) Otherwise, the function exits.
Note that if this function loops indefinitely, and the scheduler is pumped rather than
continuously run, the pumping application will stay in its pump call for a prolonged
period of time potentially starving the rest of the application for CPU time.
An attempt is made in _recv to limit the amount of data read in this manner to a fixed
amount and it lets this function exit if that amount is exceeded. However, this it is
up to the user of Stackless to understand how their application schedules and blocks,
and there are situations where small reads may still effectively loop indefinitely.
"""
if not len(self.readQueue):
return
channel, methodName, args = self.readQueue[0]
#print self._fileno, "handle_read:---ENTER---", id(channel)
while channel.balance < 0:
args = self.readQueue[0][2]
#print self._fileno, "handle_read:CALL", id(channel), args
try:
result = getattr(self.socket, methodName)(*args)
#print self._fileno, "handle_read:RESULT", id(channel), len(result)
except Exception, e:
# winsock sometimes throws ENOTCONN
#print self._fileno, "handle_read:EXCEPTION", id(channel), len(result)
if isinstance(e, stdsocket.error) and e.args[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED]:
self.handle_close()
result = ''
elif channel.balance < 0:
channel.send_exception(e.__class__, *e.args)
if channel.balance < 0:
#print self._fileno, "handle_read:RETURN-RESULT", id(channel), len(result)
channel.send(result)
if len(self.readQueue) and self.readQueue[0][0] is channel:
del self.readQueue[0]
#print self._fileno, "handle_read:---EXIT---", id(channel)
def handle_write(self):
"""
This function still needs work WRT UDP.
"""
if len(self.writeQueue):
flags, data, channel = self.writeQueue[0]
del self.writeQueue[0]
# asyncore does not expose sending the flags.
def asyncore_send(self, data, flags=0):
try:
result = self.socket.send(data, flags)
return result
except socket.error, why:
if why.args[0] == EWOULDBLOCK:
return 0
elif why.args[0] in (ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED):
self.handle_close()
return 0
else:
raise
nbytes = asyncore_send(self, data, flags)
if channel.balance < 0:
channel.send(nbytes)
elif len(self.sendToBuffers):
data, address, channel, oldSentBytes = self.sendToBuffers[0]
sentBytes = self.socket.sendto(data, address)
totalSentBytes = oldSentBytes + sentBytes
if len(data) > sentBytes:
self.sendToBuffers[0] = data[sentBytes:], address, channel, totalSentBytes
else:
del self.sendToBuffers[0]
stackless.tasklet(channel.send)(totalSentBytes)
if False:
def dump_socket_stack_traces():
import traceback
for skt in asyncore.socket_map.values():
for k, v in skt.__dict__.items():
if isinstance(v, stackless.channel) and v.queue:
i = 0
current = v.queue
while i == 0 or v.queue is not current:
print "%s.%s.%s" % (skt, k, i)
traceback.print_stack(t)
i += 1
if __name__ == '__main__':
import sys
import struct
# Test code goes here.
testAddress = "127.0.0.1", 3000
info = -12345678
data = struct.pack("i", info)
dataLength = len(data)
def TestTCPServer(address):
global info, data, dataLength
print "server listen socket creation"
listenSocket = stdsocket.socket(AF_INET, SOCK_STREAM)
listenSocket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
listenSocket.bind(address)
listenSocket.listen(5)
NUM_TESTS = 2
i = 1
while i < NUM_TESTS + 1:
# No need to schedule this tasklet as the accept should yield most
# of the time on the underlying channel.
print "server connection wait", i
currentSocket, clientAddress = listenSocket.accept()
print "server", i, "listen socket", currentSocket.fileno(), "from", clientAddress
if i == 1:
print "server closing (a)", i, "fd", currentSocket.fileno(), "id", id(currentSocket)
currentSocket.close()
print "server closed (a)", i
elif i == 2:
print "server test", i, "send"
currentSocket.send(data)
print "server test", i, "recv"
if currentSocket.recv(4) != "":
print "server recv(1)", i, "FAIL"
break
# multiple empty recvs are fine
if currentSocket.recv(4) != "":
print "server recv(2)", i, "FAIL"
break
else:
print "server closing (b)", i, "fd", currentSocket.fileno(), "id", id(currentSocket)
currentSocket.close()
print "server test", i, "OK"
i += 1
if i != NUM_TESTS+1:
print "server: FAIL", i
else:
print "server: OK", i
print "Done server"
def TestTCPClient(address):
global info, data, dataLength
# Attempt 1:
clientSocket = stdsocket.socket()
clientSocket.connect(address)
print "client connection (1) fd", clientSocket.fileno(), "id", id(clientSocket._sock), "waiting to recv"
if clientSocket.recv(5) != "":
print "client test", 1, "FAIL"
else:
print "client test", 1, "OK"
# Attempt 2:
clientSocket = stdsocket.socket()
clientSocket.connect(address)
print "client connection (2) fd", clientSocket.fileno(), "id", id(clientSocket._sock), "waiting to recv"
s = clientSocket.recv(dataLength)
if s == "":
print "client test", 2, "FAIL (disconnect)"
else:
t = struct.unpack("i", s)
if t[0] == info:
print "client test", 2, "OK"
else:
print "client test", 2, "FAIL (wrong data)"
print "client exit"
def TestMonkeyPatchUrllib(uri):
# replace the system socket with this module
install()
try:
import urllib # must occur after monkey-patching!
f = urllib.urlopen(uri)
if not isinstance(f.fp._sock, _fakesocket):
raise AssertionError("failed to apply monkeypatch, got %s" % f.fp._sock.__class__)
s = f.read()
if len(s) != 0:
print "Fetched", len(s), "bytes via replaced urllib"
else:
raise AssertionError("no text received?")
finally:
uninstall()
def TestMonkeyPatchUDP(address):
# replace the system socket with this module
install()
try:
def UDPServer(address):
listenSocket = stdsocket.socket(AF_INET, SOCK_DGRAM)
listenSocket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
listenSocket.bind(address)
# Apparently each call to recvfrom maps to an incoming
# packet and if we only ask for part of that packet, the
# rest is lost. We really need a proper unittest suite
# which tests this module against the normal socket
# module.
print "waiting to receive"
rdata = ""
while len(rdata) < 512:
data, address = listenSocket.recvfrom(4096)
print "received", data, len(data)
rdata += data
def UDPClient(address):
clientSocket = stdsocket.socket(AF_INET, SOCK_DGRAM)
# clientSocket.connect(address)
print "sending 512 byte packet"
sentBytes = clientSocket.sendto("-"+ ("*" * 510) +"-", address)
print "sent 512 byte packet", sentBytes
stackless.tasklet(UDPServer)(address)
stackless.tasklet(UDPClient)(address)
stackless.run()
finally:
uninstall()
if len(sys.argv) == 2:
if sys.argv[1] == "client":
print "client started"
TestTCPClient(testAddress)
print "client exited"
elif sys.argv[1] == "slpclient":
print "client started"
stackless.tasklet(TestTCPClient)(testAddress)
stackless.run()
print "client exited"
elif sys.argv[1] == "server":
print "server started"
TestTCPServer(testAddress)
print "server exited"
elif sys.argv[1] == "slpserver":
print "server started"
stackless.tasklet(TestTCPServer)(testAddress)
stackless.run()
print "server exited"
else:
print "Usage:", sys.argv[0], "[client|server|slpclient|slpserver]"
sys.exit(1)
else:
print "* Running client/server test"
install()
try:
stackless.tasklet(TestTCPServer)(testAddress)
stackless.tasklet(TestTCPClient)(testAddress)
stackless.run()
finally:
uninstall()
print "* Running urllib test"
stackless.tasklet(TestMonkeyPatchUrllib)("http://python.org/")
stackless.run()
print "* Running udp test"
TestMonkeyPatchUDP(testAddress)
print "result: SUCCESS"
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化