[Zope-Checkins] CVS: Zope2 - __init__.py:1.1 asynchat.py:1.1 asyncore.py:1.1
Jim Fulton
jim@digicool.com
Thu, 5 Apr 2001 19:38:18 -0400 (EDT)
Update of /cvs-repository/Zope2/ZServer/async1x
In directory korak:/tmp/cvs-serv24500
Added Files:
__init__.py asynchat.py asyncore.py
Log Message:
Moved Python 1.x compatable asyncx to their own package that takes
care of installing them when necessary (not for Python 2.0)
--- Added File __init__.py in package Zope2 ---
# We want to use updated asynchat and asyncore if using Python1.5
import sys
if sys.version[:1] < '2':
import asyncore, asynchat
sys.modules['asyncore']=asyncore
sys.modules['asynchat']=asynchat
del asyncore
del asynchat
del sys
--- Added File asynchat.py in package Zope2 ---
# -*- Mode: Python; tab-width: 4 -*-
# $Id: asynchat.py,v 1.1 2001/04/05 23:38:16 jim Exp $
# Author: Sam Rushing <rushing@nightmare.com>
# ======================================================================
# Copyright 1996 by Sam Rushing
#
# All Rights Reserved
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose and without fee is hereby
# granted, provided that the above copyright notice appear in all
# copies and that both that copyright notice and this permission
# notice appear in supporting documentation, and that the name of Sam
# Rushing not be used in advertising or publicity pertaining to
# distribution of the software without specific, written prior
# permission.
#
# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR
# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
# ======================================================================
import socket
import asyncore
import string
# This class adds support for 'chat' style protocols - where one side
# sends a 'command', and the other sends a response (examples would be
# the common internet protocols - smtp, nntp, ftp, etc..).
# The handle_read() method looks at the input stream for the current
# 'terminator' (usually '\r\n' for single-line responses, '\r\n.\r\n'
# for multi-line output), calling self.found_terminator() on its
# receipt.
# for example:
# Say you build an async nntp client using this class. At the start
# of the connection, you'll have self.terminator set to '\r\n', in
# order to process the single-line greeting. Just before issuing a
# 'LIST' command you'll set it to '\r\n.\r\n'. The output of the LIST
# command will be accumulated (using your own 'collect_incoming_data'
# method) up to the terminator, and then control will be returned to
# you - by calling your self.found_terminator() method
class async_chat (asyncore.dispatcher):
"""This is an abstract class. You must derive from this class, and add
the two methods collect_incoming_data() and found_terminator()"""
# these are overridable defaults
ac_in_buffer_size = 4096
ac_out_buffer_size = 4096
def __init__ (self, conn=None):
self.ac_in_buffer = ''
self.ac_out_buffer = ''
self.producer_fifo = fifo()
asyncore.dispatcher.__init__ (self, conn)
def set_terminator (self, term):
"Set the input delimiter. Can be a fixed string of any length, an integer, or None"
self.terminator = term
def get_terminator (self):
return self.terminator
# grab some more data from the socket,
# throw it to the collector method,
# check for the terminator,
# if found, transition to the next state.
def handle_read (self):
try:
data = self.recv (self.ac_in_buffer_size)
except socket.error, why:
self.handle_error()
return
self.ac_in_buffer = self.ac_in_buffer + data
# Continue to search for self.terminator in self.ac_in_buffer,
# while calling self.collect_incoming_data. The while loop
# is necessary because we might read several data+terminator
# combos with a single recv(1024).
while self.ac_in_buffer:
lb = len(self.ac_in_buffer)
terminator = self.get_terminator()
if terminator is None:
# no terminator, collect it all
self.collect_incoming_data (self.ac_in_buffer)
self.ac_in_buffer = ''
elif type(terminator) == type(0):
# numeric terminator
n = terminator
if lb < n:
self.collect_incoming_data (self.ac_in_buffer)
self.ac_in_buffer = ''
self.terminator = self.terminator - lb
else:
self.collect_incoming_data (self.ac_in_buffer[:n])
self.ac_in_buffer = self.ac_in_buffer[n:]
self.terminator = 0
self.found_terminator()
else:
# 3 cases:
# 1) end of buffer matches terminator exactly:
# collect data, transition
# 2) end of buffer matches some prefix:
# collect data to the prefix
# 3) end of buffer does not match any prefix:
# collect data
terminator_len = len(terminator)
index = string.find (self.ac_in_buffer, terminator)
if index != -1:
# we found the terminator
if index > 0:
# don't bother reporting the empty string (source of subtle bugs)
self.collect_incoming_data (self.ac_in_buffer[:index])
self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:]
# This does the Right Thing if the terminator is changed here.
self.found_terminator()
else:
# check for a prefix of the terminator
index = find_prefix_at_end (self.ac_in_buffer, terminator)
if index:
if index != lb:
# we found a prefix, collect up to the prefix
self.collect_incoming_data (self.ac_in_buffer[:-index])
self.ac_in_buffer = self.ac_in_buffer[-index:]
break
else:
# no prefix, collect it all
self.collect_incoming_data (self.ac_in_buffer)
self.ac_in_buffer = ''
def handle_write (self):
self.initiate_send ()
def handle_close (self):
self.close()
def push (self, data):
self.producer_fifo.push (simple_producer (data))
self.initiate_send()
def push_with_producer (self, producer):
self.producer_fifo.push (producer)
self.initiate_send()
def readable (self):
"predicate for inclusion in the readable for select()"
return (len(self.ac_in_buffer) <= self.ac_in_buffer_size)
def writable (self):
"predicate for inclusion in the writable for select()"
# return len(self.ac_out_buffer) or len(self.producer_fifo) or (not self.connected)
# this is about twice as fast, though not as clear.
return not (
(self.ac_out_buffer is '') and
self.producer_fifo.is_empty() and
self.connected
)
def close_when_done (self):
"automatically close this channel once the outgoing queue is empty"
self.producer_fifo.push (None)
# refill the outgoing buffer by calling the more() method
# of the first producer in the queue
def refill_buffer (self):
_string_type = type('')
while 1:
if len(self.producer_fifo):
p = self.producer_fifo.first()
# a 'None' in the producer fifo is a sentinel,
# telling us to close the channel.
if p is None:
if not self.ac_out_buffer:
self.producer_fifo.pop()
self.close()
return
elif type(p) is _string_type:
self.producer_fifo.pop()
self.ac_out_buffer = self.ac_out_buffer + p
return
data = p.more()
if data:
self.ac_out_buffer = self.ac_out_buffer + data
return
else:
self.producer_fifo.pop()
else:
return
def initiate_send (self):
obs = self.ac_out_buffer_size
# try to refill the buffer
if (len (self.ac_out_buffer) < obs):
self.refill_buffer()
if self.ac_out_buffer and self.connected:
# try to send the buffer
try:
num_sent = self.send (self.ac_out_buffer[:obs])
if num_sent:
self.ac_out_buffer = self.ac_out_buffer[num_sent:]
except socket.error, why:
self.handle_error()
return
def discard_buffers (self):
# Emergencies only!
self.ac_in_buffer = ''
self.ac_out_buffer = ''
while self.producer_fifo:
self.producer_fifo.pop()
class simple_producer:
def __init__ (self, data, buffer_size=512):
self.data = data
self.buffer_size = buffer_size
def more (self):
if len (self.data) > self.buffer_size:
result = self.data[:self.buffer_size]
self.data = self.data[self.buffer_size:]
return result
else:
result = self.data
self.data = ''
return result
class fifo:
def __init__ (self, list=None):
if not list:
self.list = []
else:
self.list = list
def __len__ (self):
return len(self.list)
def is_empty (self):
return self.list == []
def first (self):
return self.list[0]
def push (self, data):
self.list.append (data)
def pop (self):
if self.list:
result = self.list[0]
del self.list[0]
return (1, result)
else:
return (0, None)
# Given 'haystack', see if any prefix of 'needle' is at its end. This
# assumes an exact match has already been checked. Return the number of
# characters matched.
# for example:
# f_p_a_e ("qwerty\r", "\r\n") => 1
# f_p_a_e ("qwerty\r\n", "\r\n") => 2
# f_p_a_e ("qwertydkjf", "\r\n") => 0
# this could maybe be made faster with a computed regex?
##def find_prefix_at_end (haystack, needle):
## nl = len(needle)
## result = 0
## for i in range (1,nl):
## if haystack[-(nl-i):] == needle[:(nl-i)]:
## result = nl-i
## break
## return result
# yes, this is about twice as fast, but still seems
# to be neglible CPU. The previous could do about 290
# searches/sec. the new one about 555/sec.
import regex
prefix_cache = {}
def prefix_regex (needle):
if prefix_cache.has_key (needle):
return prefix_cache[needle]
else:
reg = needle[-1]
for i in range(1,len(needle)):
reg = '%c\(%s\)?' % (needle[-(i+1)], reg)
reg = regex.compile (reg+'$')
prefix_cache[needle] = reg, len(needle)
return reg, len(needle)
def find_prefix_at_end (haystack, needle):
reg, length = prefix_regex (needle)
lh = len(haystack)
result = reg.search (haystack, max(0,lh-length))
if result >= 0:
return (lh - result)
else:
return 0
--- Added File asyncore.py in package Zope2 ---
# -*- Mode: Python; tab-width: 4 -*-
# $Id: asyncore.py,v 1.1 2001/04/05 23:38:16 jim Exp $
# Author: Sam Rushing <rushing@nightmare.com>
# ======================================================================
# Copyright 1996 by Sam Rushing
#
# All Rights Reserved
#
# Permission to use, copy, modify, and distribute this software and
# its documentation for any purpose and without fee is hereby
# granted, provided that the above copyright notice appear in all
# copies and that both that copyright notice and this permission
# notice appear in supporting documentation, and that the name of Sam
# Rushing not be used in advertising or publicity pertaining to
# distribution of the software without specific, written prior
# permission.
#
# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR
# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
# ======================================================================
import exceptions
import select
import socket
import string
import sys
import os
if os.name == 'nt':
EWOULDBLOCK = 10035
EINPROGRESS = 10036
EALREADY = 10037
ECONNRESET = 10054
ENOTCONN = 10057
ESHUTDOWN = 10058
EINTR = 0 # what should this be?
else:
from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET
from errno import ENOTCONN, ESHUTDOWN, EINTR
try:
socket_map
except NameError:
socket_map = {}
class ExitNow (exceptions.Exception):
pass
DEBUG = 0
def poll (timeout=0.0, map=None):
global DEBUG
if map is None:
map = socket_map
if map:
r = []; w = []; e = []
for fd, obj in map.items():
if obj.readable():
r.append (fd)
if obj.writable():
w.append (fd)
while 1:
try: r,w,e = select.select (r,w,e, timeout)
except select.error, v:
if v[0] != EINTR: raise
else: break
if DEBUG:
print r,w,e
for fd in r:
try:
obj = map[fd]
try:
obj.handle_read_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
for fd in w:
try:
obj = map[fd]
try:
obj.handle_write_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
def poll2 (timeout=0.0, map=None):
import poll
if map is None:
map=socket_map
# timeout is in milliseconds
timeout = int(timeout*1000)
if map:
l = []
for fd, obj in map.items():
flags = 0
if obj.readable():
flags = poll.POLLIN
if obj.writable():
flags = flags | poll.POLLOUT
if flags:
l.append ((fd, flags))
r = poll.poll (l, timeout)
for fd, flags in r:
try:
obj = map[fd]
try:
if (flags & poll.POLLIN):
obj.handle_read_event()
if (flags & poll.POLLOUT):
obj.handle_write_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
def loop (timeout=30.0, use_poll=0, map=None):
if use_poll:
poll_fun = poll2
else:
poll_fun = poll
if map is None:
map=socket_map
while map:
poll_fun (timeout, map)
class dispatcher:
debug = 0
connected = 0
accepting = 0
closing = 0
addr = None
def __init__ (self, sock=None, map=None):
if sock:
self.set_socket (sock, map)
# I think it should inherit this anyway
self.socket.setblocking (0)
self.connected = 1
def __repr__ (self):
try:
status = []
if self.accepting and self.addr:
status.append ('listening')
elif self.connected:
status.append ('connected')
if self.addr:
status.append ('%s:%d' % self.addr)
return '<%s %s at %x>' % (
self.__class__.__name__,
string.join (status, ' '),
id(self)
)
except:
try:
ar = repr(self.addr)
except:
ar = 'no self.addr!'
return '<__repr__ (self) failed for object at %x (addr=%s)>' % (id(self),ar)
def add_channel (self, map=None):
#self.log_info ('adding channel %s' % self)
if map is None:
map=socket_map
map [self._fileno] = self
def del_channel (self, map=None):
fd = self._fileno
if map is None:
map=socket_map
if map.has_key (fd):
#self.log_info ('closing channel %d:%s' % (fd, self))
del map [fd]
def create_socket (self, family, type):
self.family_and_type = family, type
self.socket = socket.socket (family, type)
self.socket.setblocking(0)
self._fileno = self.socket.fileno()
self.add_channel()
def set_socket (self, sock, map=None):
self.__dict__['socket'] = sock
self._fileno = sock.fileno()
self.add_channel (map)
def set_reuse_addr (self):
# try to re-use a server port if possible
try:
self.socket.setsockopt (
socket.SOL_SOCKET, socket.SO_REUSEADDR,
self.socket.getsockopt (socket.SOL_SOCKET, socket.SO_REUSEADDR) | 1
)
except:
pass
# ==================================================
# predicates for select()
# these are used as filters for the lists of sockets
# to pass to select().
# ==================================================
def readable (self):
return 1
if os.name == 'mac':
# The macintosh will select a listening socket for
# write if you let it. What might this mean?
def writable (self):
return not self.accepting
else:
def writable (self):
return 1
# ==================================================
# socket object methods.
# ==================================================
def listen (self, num):
self.accepting = 1
if os.name == 'nt' and num > 5:
num = 1
return self.socket.listen (num)
def bind (self, addr):
self.addr = addr
return self.socket.bind (addr)
def connect (self, address):
self.connected = 0
try:
self.socket.connect (address)
except socket.error, why:
if why[0] in (EINPROGRESS, EALREADY, EWOULDBLOCK):
return
else:
raise socket.error, why
self.connected = 1
self.handle_connect()
def accept (self):
try:
conn, addr = self.socket.accept()
return conn, addr
except socket.error, why:
if why[0] == EWOULDBLOCK:
pass
else:
raise socket.error, why
def send (self, data):
try:
result = self.socket.send (data)
return result
except socket.error, why:
if why[0] == EWOULDBLOCK:
return 0
else:
raise socket.error, why
return 0
def recv (self, buffer_size):
try:
data = self.socket.recv (buffer_size)
if not data:
# a closed connection is indicated by signaling
# a read condition, and having recv() return 0.
self.handle_close()
return ''
else:
return data
except socket.error, why:
# winsock sometimes throws ENOTCONN
if why[0] in [ECONNRESET, ENOTCONN, ESHUTDOWN]:
self.handle_close()
return ''
else:
raise socket.error, why
def close (self):
self.del_channel()
self.socket.close()
# cheap inheritance, used to pass all other attribute
# references to the underlying socket object.
def __getattr__ (self, attr):
return getattr (self.socket, attr)
# log and log_info maybe overriden to provide more sophisitcated
# logging and warning methods. In general, log is for 'hit' logging
# and 'log_info' is for informational, warning and error logging.
def log (self, message):
sys.stderr.write ('log: %s\n' % str(message))
def log_info (self, message, type='info'):
if __debug__ or type != 'info':
print '%s: %s' % (type, message)
def handle_read_event (self):
if self.accepting:
# for an accepting socket, getting a read implies
# that we are connected
if not self.connected:
self.connected = 1
self.handle_accept()
elif not self.connected:
self.handle_connect()
self.connected = 1
self.handle_read()
else:
self.handle_read()
def handle_write_event (self):
# getting a write implies that we are connected
if not self.connected:
self.handle_connect()
self.connected = 1
self.handle_write()
def handle_expt_event (self):
self.handle_expt()
def handle_error (self):
(file,fun,line), t, v, tbinfo = compact_traceback()
# sometimes a user repr method will crash.
try:
self_repr = repr (self)
except:
self_repr = '<__repr__ (self) failed for object at %0x>' % id(self)
self.log_info (
'uncaptured python exception, closing channel %s (%s:%s %s)' % (
self_repr,
t,
v,
tbinfo
),
'error'
)
self.close()
def handle_expt (self):
self.log_info ('unhandled exception', 'warning')
def handle_read (self):
self.log_info ('unhandled read event', 'warning')
def handle_write (self):
self.log_info ('unhandled write event', 'warning')
def handle_connect (self):
self.log_info ('unhandled connect event', 'warning')
def handle_accept (self):
self.log_info ('unhandled accept event', 'warning')
def handle_close (self):
self.log_info ('unhandled close event', 'warning')
self.close()
# ---------------------------------------------------------------------------
# adds simple buffered output capability, useful for simple clients.
# [for more sophisticated usage use asynchat.async_chat]
# ---------------------------------------------------------------------------
class dispatcher_with_send (dispatcher):
def __init__ (self, sock=None):
dispatcher.__init__ (self, sock)
self.out_buffer = ''
def initiate_send (self):
num_sent = 0
num_sent = dispatcher.send (self, self.out_buffer[:512])
self.out_buffer = self.out_buffer[num_sent:]
def handle_write (self):
self.initiate_send()
def writable (self):
return (not self.connected) or len(self.out_buffer)
def send (self, data):
if self.debug:
self.log_info ('sending %s' % repr(data))
self.out_buffer = self.out_buffer + data
self.initiate_send()
# ---------------------------------------------------------------------------
# used for debugging.
# ---------------------------------------------------------------------------
def compact_traceback ():
t,v,tb = sys.exc_info()
tbinfo = []
while 1:
tbinfo.append ((
tb.tb_frame.f_code.co_filename,
tb.tb_frame.f_code.co_name,
str(tb.tb_lineno)
))
tb = tb.tb_next
if not tb:
break
# just to be safe
del tb
file, function, line = tbinfo[-1]
info = '[' + string.join (
map (
lambda x: string.join (x, '|'),
tbinfo
),
'] ['
) + ']'
return (file, function, line), t, v, info
def close_all (map=None):
if map is None:
map=socket_map
for x in map.values():
x.socket.close()
map.clear()
# Asynchronous File I/O:
#
# After a little research (reading man pages on various unixen, and
# digging through the linux kernel), I've determined that select()
# isn't meant for doing doing asynchronous file i/o.
# Heartening, though - reading linux/mm/filemap.c shows that linux
# supports asynchronous read-ahead. So _MOST_ of the time, the data
# will be sitting in memory for us already when we go to read it.
#
# What other OS's (besides NT) support async file i/o? [VMS?]
#
# Regardless, this is useful for pipes, and stdin/stdout...
import os
if os.name == 'posix':
import fcntl
import FCNTL
class file_wrapper:
# here we override just enough to make a file
# look like a socket for the purposes of asyncore.
def __init__ (self, fd):
self.fd = fd
def recv (self, *args):
return apply (os.read, (self.fd,)+args)
def write (self, *args):
return apply (os.write, (self.fd,)+args)
def close (self):
return os.close (self.fd)
def fileno (self):
return self.fd
class file_dispatcher (dispatcher):
def __init__ (self, fd):
dispatcher.__init__ (self)
self.connected = 1
# set it to non-blocking mode
flags = fcntl.fcntl (fd, FCNTL.F_GETFL, 0)
flags = flags | FCNTL.O_NONBLOCK
fcntl.fcntl (fd, FCNTL.F_SETFL, flags)
self.set_file (fd)
def set_file (self, fd):
self._fileno = fd
self.socket = file_wrapper (fd)
self.add_channel()