Skip to content

Z
Zope
Commit 5e529dcf authored by Amos Latteier's avatar Amos Latteier
Browse files
Options

Synched with latest medusa sources from CVS. The main changes are the... 

Synched with latest medusa sources from CVS. The main changes are the structure of asyncore.socket_map which now maps fds to channels which should provide a slight performance improvement.
parent 09bb46b6
Hide whitespace changes
Inline Side-by-side
Showing with 1308 additions and 1304 deletions
ZServer/medusa/asynchat.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
# $Id: asynchat.py,v 1.9 1999/07/19 17:43:47 amos Exp $
# $Id: asynchat.py,v 1.10 2000/01/14 02:35:56 amos Exp $
# Author: Sam Rushing <rushing@nightmare.com>
# ======================================================================
......@@ -219,7 +219,7 @@ class async_chat (asyncore.dispatcher):
def discard_buffers (self):
# Emergencies only!
self.ac_in_buffer = ''
self.ac_out_buffer == ''
self.ac_out_buffer = ''
while self.producer_fifo:
self.producer_fifo.pop()
......
ZServer/medusa/asyncore.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
# $Id: asyncore.py,v 1.6 1999/07/26 07:06:36 amos Exp $
# $Id: asyncore.py,v 1.7 2000/01/14 02:35:56 amos Exp $
# Author: Sam Rushing <rushing@nightmare.com>
# ======================================================================
......@@ -25,6 +25,7 @@
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
# ======================================================================
import exceptions
import select
import socket
import string
......@@ -41,60 +42,83 @@ if os.name == 'nt':
else:
from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, ENOTCONN, ESHUTDOWN
socket_map = {}
try:
socket_map
except NameError:
socket_map = {}
class ExitNow (exceptions.Exception):
pass
DEBUG = 0
def poll (timeout=0.0):
global DEBUG
if socket_map:
r = []; w = []; e = []
for s in socket_map.keys():
if s.readable():
r.append (s)
if s.writable():
w.append (s)
for fd, obj in socket_map.items():
if obj.readable():
r.append (fd)
if obj.writable():
w.append (fd)
r,w,e = select.select (r,w,e, timeout)
(r,w,e) = select.select (r,w,e, timeout)
if DEBUG:
print r,w,e
for x in r:
for fd in r:
try:
x.handle_read_event()
except:
x.handle_error()
for x in w:
obj = socket_map[fd]
try:
obj.handle_read_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
for fd in w:
try:
x.handle_write_event()
except:
x.handle_error()
obj = socket_map[fd]
try:
obj.handle_write_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
def poll2 (timeout=0.0):
import poll
# timeout is in milliseconds
timeout = int(timeout*1000)
if socket_map:
fd_map = {}
for s in socket_map.keys():
fd_map[s.fileno()] = s
l = []
for fd, s in fd_map.items():
for fd, obj in socket_map.items():
flags = 0
if s.readable():
if obj.readable():
flags = poll.POLLIN
if s.writable():
if obj.writable():
flags = flags | poll.POLLOUT
if flags:
l.append (fd, flags)
r = poll.poll (l, timeout)
for fd, flags in r:
s = fd_map[fd]
try:
if (flags & poll.POLLIN):
s.handle_read_event()
if (flags & poll.POLLOUT):
s.handle_write_event()
if (flags & poll.POLLERR):
s.handle_expt_event()
except:
s.handle_error()
obj = socket_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):
......@@ -143,23 +167,25 @@ class dispatcher:
return '<__repr__ (self) failed for object at %x (addr=%s)>' % (id(self),ar)
def add_channel (self):
self.log_info ('adding channel %s' % self)
socket_map [self] = 1
#self.log_info ('adding channel %s' % self)
socket_map [self._fileno] = self
def del_channel (self):
if socket_map.has_key (self):
self.log_info ('closing channel %d:%s' % (self.fileno(), self))
del socket_map [self]
fd = self._fileno
if socket_map.has_key (fd):
#self.log_info ('closing channel %d:%s' % (fd, self))
del socket_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, socket):
# This is done so we can be called safely from __init__
self.__dict__['socket'] = socket
def set_socket (self, sock):
self.__dict__['socket'] = sock
self._fileno = sock.fileno()
self.add_channel()
def set_reuse_addr (self):
......@@ -261,20 +287,19 @@ class dispatcher:
# cheap inheritance, used to pass all other attribute
# references to the underlying socket object.
# NOTE: this may be removed soon for performance reasons.
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):
print 'log:', 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)
print '%s: %s' % (type, message)
def handle_read_event (self):
if self.accepting:
......@@ -398,7 +423,7 @@ def compact_traceback ():
def close_all ():
global socket_map
for x in socket_map.keys():
for x in socket_map.values():
x.socket.close()
socket_map.clear()
......@@ -449,6 +474,7 @@ if os.name == 'posix':
self.set_file (fd)
def set_file (self, fd):
self._fileno = fd
self.socket = file_wrapper (fd)
self.add_channel()
ZServer/medusa/ftp_server.py
View file @ 5e529dcf
......@@ -8,7 +8,7 @@
# If you are interested in using this software in a commercial context,
# or in purchasing support, please contact the author.
RCS_ID = '$Id: ftp_server.py,v 1.8 1999/11/15 21:53:56 amos Exp $'
RCS_ID = '$Id: ftp_server.py,v 1.9 2000/01/14 02:35:56 amos Exp $'
# An extensible, configurable, asynchronous FTP server.
#
......@@ -336,6 +336,7 @@ class ftp_channel (asynchat.async_chat):
self.current_mode = t
self.respond ('200 Type set to %s.' % self.type_map[t])
def cmd_quit (self, line):
'terminate session'
self.respond ('221 Goodbye.')
......@@ -847,13 +848,16 @@ class xmit_channel (asynchat.async_chat):
self.channel = channel
self.client_addr = client_addr
asynchat.async_chat.__init__ (self)
# def __del__ (self):
# print 'xmit_channel.__del__()'
def readable (self):
return 0
def log(self, *args):
def log (*args):
pass
def readable (self):
return not self.connected
def writable (self):
return 1
......@@ -1054,13 +1058,7 @@ if os.name == 'posix':
self.log_info('FTP server shutting down. (received SIGINT)', 'warning')
# close everything down on SIGINT.
# of course this should be a cleaner shutdown.
sm = socket.socket_map
socket.socket_map = {}
for sock in sm.values():
try:
sock.close()
except:
pass
asyncore.close_all()
if __name__ == '__main__':
test (sys.argv[1])
......
ZServer/medusa/http_server.py
View file @ 5e529dcf
......@@ -9,7 +9,7 @@
# interested in using this software in a commercial context, or in
# purchasing support, please contact the author.
RCS_ID = '$Id: http_server.py,v 1.13 1999/11/15 21:53:56 amos Exp $'
RCS_ID = '$Id: http_server.py,v 1.14 2000/01/14 02:35:56 amos Exp $'
# python modules
import os
......@@ -269,7 +269,7 @@ class http_request:
user_agent=self.get_header('user-agent')
if not user_agent: user_agent=''
referer=self.get_header('referer')
if not referer: referer=''
if not referer: referer=''
self.channel.server.logger.log (
self.channel.addr[0],
' - - [%s] "%s" %d %d "%s" "%s"\n' % (
......@@ -385,7 +385,7 @@ class http_channel (asynchat.async_chat):
def kill_zombies (self):
now = int (time.time())
for channel in asyncore.socket_map.keys():
for channel in asyncore.socket_map.values():
if channel.__class__ == self.__class__:
if (now - channel.creation_time) > channel.zombie_timeout:
channel.close()
......@@ -448,7 +448,7 @@ class http_channel (asynchat.async_chat):
# --------------------------------------------------
# crack the request header
# --------------------------------------------------
while lines and not lines[0]:
# as per the suggestion of http-1.1 section 4.1, (and
# Eric Parker <eparker@zyvex.com>), ignore a leading
......@@ -461,27 +461,18 @@ class http_channel (asynchat.async_chat):
return
request = lines[0]
try:
command, uri, version = crack_request (request)
except:
# deal with broken HTTP requests
try:
# maybe there were spaces in the URL
parts=string.split(request)
command, uri, version = crack_request(
'%s %s %s' % (parts[0], parts[1], parts[-1]))
except:
self.log_info('Bad HTTP request: %s' % request, 'error')
r = http_request (self, request,
None, None, None, join_headers(lines[1:]))
r.error(400)
return
command, uri, version = crack_request (request)
header = join_headers (lines[1:])
r = http_request (self, request, command, uri, version, header)
self.request_counter.increment()
self.server.total_requests.increment()
if command is None:
self.log_info('Bad HTTP request: %s' % request, 'error')
r.error(400)
return
# --------------------------------------------------
# handler selection and dispatch
# --------------------------------------------------
......@@ -612,7 +603,14 @@ class http_server (asyncore.dispatcher):
# accept. socketmodule.c:makesockaddr complains that the
# address family is unknown. We don't want the whole server
# to shut down because of this.
self.log_info('Server accept() threw an exception', 'warning')
self.log_info ('warning: server accept() threw an exception', 'warning')
return
except TypeError:
# unpack non-sequence. this can happen when a read event
# fires on a listening socket, but when we call accept()
# we get EWOULDBLOCK, so dispatcher.accept() returns None.
# Seen on FreeBSD3.
self.log_info ('warning: server accept() threw EWOULDBLOCK', 'warning')
return
self.channel_class (self, conn, addr)
......@@ -689,6 +687,8 @@ def crack_request (r):
else:
version = None
return string.lower (REQUEST.group (1)), REQUEST.group(2), version
else:
return None, None, None
class fifo:
def __init__ (self, list=None):
......
ZServer/medusa/monitor.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
# Author: Sam Rushing <rushing@nightmare.com>
# Author: Sam Rushing <rushing@nightmare.com>
#
# python REPL channel.
#
RCS_ID = '$Id: monitor.py,v 1.5 1999/07/21 17:15:53 klm Exp $'
RCS_ID = '$Id: monitor.py,v 1.6 2000/01/14 02:35:56 amos Exp $'
import md5
import socket
import string
import sys
import time
import traceback
VERSION = string.split(RCS_ID)[2]
......@@ -23,330 +22,325 @@ from counter import counter
import producers
class monitor_channel (asynchat.async_chat):
try_linemode = 1
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = {}
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.push ("[Hint: try 'from __main__ import *']\r\n")
self.prompt()
self.number = server.total_sessions.as_long()
self.line_counter = counter()
self.multi_line = []
def handle_connect (self):
# send IAC DO LINEMODE
self.push ('\377\375\"')
def close (self):
self.server.closed_sessions.increment()
asynchat.async_chat.close(self)
def prompt (self):
self.push ('>>> ')
def collect_incoming_data (self, data):
self.data = self.data + data
if len(self.data) > 1024:
# denial of service.
self.push ('BCNU\r\n')
self.close_when_done()
def found_terminator (self):
line = self.clean_line (self.data)
self.data = ''
self.line_counter.increment()
# check for special case inputs...
if not line and not self.multi_line:
self.prompt()
return
if line in ['\004', 'exit']:
self.push ('BCNU\r\n')
self.close_when_done()
return
oldout = sys.stdout
olderr = sys.stderr
try:
p = output_producer(self, olderr)
sys.stdout = p
sys.stderr = p
try:
# this is, of course, a blocking operation.
# if you wanted to thread this, you would have
# to synchronize, etc... and treat the output
# like a pipe. Not Fun.
#
# try eval first. If that fails, try exec. If that fails,
# hurl.
try:
if self.multi_line:
# oh, this is horrible...
raise SyntaxError
co = compile (line, repr(self), 'eval')
result = eval (co, self.local_env)
method = 'eval'
if result is not None:
self.log_info(repr(result))
print repr(result)
self.local_env['_'] = result
except SyntaxError:
try:
if self.multi_line:
if line and line[0] in [' ','\t']:
self.multi_line.append (line)
self.push ('... ')
return
else:
self.multi_line.append (line)
line = string.join (self.multi_line, '\n')
co = compile (line, repr(self), 'exec')
self.multi_line = []
else:
co = compile (line, repr(self), 'exec')
except SyntaxError, why:
if why[0] == 'unexpected EOF while parsing':
self.push ('... ')
self.multi_line.append (line)
return
exec co in self.local_env
method = 'exec'
except:
method = 'exception'
self.multi_line = []
t, v, tb = sys.exc_info()
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info('%s %s %s' %(t, v, tbinfo), 'warning')
traceback.print_exception(t, v, tb)
del tb
finally:
sys.stdout = oldout
sys.stderr = olderr
self.log_info('%s:%s (%s)> %s' % (
self.number,
self.line_counter,
method,
repr(line))
)
self.push_with_producer (p)
self.prompt()
# for now, we ignore any telnet option stuff sent to
# us, and we process the backspace key ourselves.
# gee, it would be fun to write a full-blown line-editing
# environment, etc...
def clean_line (self, line):
chars = []
for ch in line:
oc = ord(ch)
if oc < 127:
if oc in [8,177]:
# backspace
chars = chars[:-1]
else:
chars.append (ch)
return string.join (chars, '')
try_linemode = 1
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = sys.modules['__main__'].__dict__.copy()
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.push ("[Hint: try 'from __main__ import *']\r\n")
self.prompt()
self.number = server.total_sessions.as_long()
self.line_counter = counter()
self.multi_line = []
def handle_connect (self):
# send IAC DO LINEMODE
self.push ('\377\375\"')
def close (self):
self.server.closed_sessions.increment()
asynchat.async_chat.close(self)
def prompt (self):
self.push ('>>> ')
def collect_incoming_data (self, data):
self.data = self.data + data
if len(self.data) > 1024:
# denial of service.
self.push ('BCNU\r\n')
self.close_when_done()
def found_terminator (self):
line = self.clean_line (self.data)
self.data = ''
self.line_counter.increment()
# check for special case inputs...
if not line and not self.multi_line:
self.prompt()
return
if line in ['\004', 'exit']:
self.push ('BCNU\r\n')
self.close_when_done()
return
oldout = sys.stdout
olderr = sys.stderr
try:
p = output_producer(self, olderr)
sys.stdout = p
sys.stderr = p
try:
# this is, of course, a blocking operation.
# if you wanted to thread this, you would have
# to synchronize, etc... and treat the output
# like a pipe. Not Fun.
#
# try eval first. If that fails, try exec. If that fails,
# hurl.
try:
if self.multi_line:
# oh, this is horrible...
raise SyntaxError
co = compile (line, repr(self), 'eval')
result = eval (co, self.local_env)
method = 'eval'
if result is not None:
print repr(result)
self.local_env['_'] = result
except SyntaxError:
try:
if self.multi_line:
if line and line[0] in [' ','\t']:
self.multi_line.append (line)
self.push ('... ')
return
else:
self.multi_line.append (line)
line = string.join (self.multi_line, '\n')
co = compile (line, repr(self), 'exec')
self.multi_line = []
else:
co = compile (line, repr(self), 'exec')
except SyntaxError, why:
if why[0] == 'unexpected EOF while parsing':
self.push ('... ')
self.multi_line.append (line)
return
exec co in self.local_env
method = 'exec'
except:
method = 'exception'
self.multi_line = []
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info('%s %s %s' %(t, v, tbinfo), 'warning')
finally:
sys.stdout = oldout
sys.stderr = olderr
self.log_info('%s:%s (%s)> %s' % (
self.number,
self.line_counter,
method,
repr(line))
)
self.push_with_producer (p)
self.prompt()
# for now, we ignore any telnet option stuff sent to
# us, and we process the backspace key ourselves.
# gee, it would be fun to write a full-blown line-editing
# environment, etc...
def clean_line (self, line):
chars = []
for ch in line:
oc = ord(ch)
if oc < 127:
if oc in [8,177]:
# backspace
chars = chars[:-1]
else:
chars.append (ch)
return string.join (chars, '')
class monitor_server (asyncore.dispatcher):
SERVER_IDENT = 'Monitor Server (V%s)' % VERSION
channel_class = monitor_channel
def __init__ (self, hostname='127.0.0.1', port=8023):
self.hostname = hostname
self.port = port
self.create_socket (socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind ((hostname, port))
self.log_info('%s started on port %d' % (self.SERVER_IDENT, port))
self.listen (5)
self.closed = 0
self.failed_auths = 0
self.total_sessions = counter()
self.closed_sessions = counter()
def writable (self):
return 0
def handle_accept (self):
conn, addr = self.accept()
self.log_info('Incoming monitor connection from %s:%d' % addr)
self.channel_class (self, conn, addr)
self.total_sessions.increment()
def status (self):
return producers.simple_producer (
'<h2>%s</h2>' % self.SERVER_IDENT
+ '<br><b>Total Sessions:</b> %s' % self.total_sessions
+ '<br><b>Current Sessions:</b> %d' % (
self.total_sessions.as_long()-self.closed_sessions.as_long()
)
)
SERVER_IDENT = 'Monitor Server (V%s)' % VERSION
channel_class = monitor_channel
def __init__ (self, hostname='127.0.0.1', port=8023):
self.hostname = hostname
self.port = port
self.create_socket (socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind ((hostname, port))
self.log_info('%s started on port %d' % (self.SERVER_IDENT, port))
self.listen (5)
self.closed = 0
self.failed_auths = 0
self.total_sessions = counter()
self.closed_sessions = counter()
def writable (self):
return 0
def handle_accept (self):
conn, addr = self.accept()
self.log_info('Incoming monitor connection from %s:%d' % addr)
self.channel_class (self, conn, addr)
self.total_sessions.increment()
def status (self):
return producers.simple_producer (
'<h2>%s</h2>' % self.SERVER_IDENT
+ '<br><b>Total Sessions:</b> %s' % self.total_sessions
+ '<br><b>Current Sessions:</b> %d' % (
self.total_sessions.as_long()-self.closed_sessions.as_long()
)
)
def hex_digest (s):
m = md5.md5()
m.update (s)
return string.joinfields (
map (lambda x: hex (ord (x))[2:], map (None, m.digest())),
'',
)
m = md5.md5()
m.update (s)
return string.joinfields (
map (lambda x: hex (ord (x))[2:], map (None, m.digest())),
'',
)
class secure_monitor_channel (monitor_channel):
authorized = 0
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = {}
# send timestamp string
self.timestamp = str(time.time())
self.count = 0
self.line_counter = counter()
self.number = int(server.total_sessions.as_long())
self.multi_line = []
self.push (self.timestamp + '\r\n')
def found_terminator (self):
if not self.authorized:
if hex_digest ('%s%s' % (self.timestamp, self.server.password)) != self.data:
self.log_info ('%s: failed authorization' % self, 'warning')
self.server.failed_auths = self.server.failed_auths + 1
self.close()
else:
self.authorized = 1
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.push ("[Hint: try 'from __main__ import *']\r\n")
self.prompt()
self.data = ''
else:
monitor_channel.found_terminator (self)
authorized = 0
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = {}
# send timestamp string
self.timestamp = str(time.time())
self.count = 0
self.line_counter = counter()
self.number = int(server.total_sessions.as_long())
self.multi_line = []
self.push (self.timestamp + '\r\n')
def found_terminator (self):
if not self.authorized:
if hex_digest ('%s%s' % (self.timestamp, self.server.password)) != self.data:
self.log_info ('%s: failed authorization' % self, 'warning')
self.server.failed_auths = self.server.failed_auths + 1
self.close()
else:
self.authorized = 1
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.prompt()
self.data = ''
else:
monitor_channel.found_terminator (self)
class secure_encrypted_monitor_channel (secure_monitor_channel):
"Wrap send() and recv() with a stream cipher"
def __init__ (self, server, conn, addr):
key = server.password
self.outgoing = server.cipher.new (key)
self.incoming = server.cipher.new (key)
secure_monitor_channel.__init__ (self, server, conn, addr)
def send (self, data):
# send the encrypted data instead
ed = self.outgoing.encrypt (data)
return secure_monitor_channel.send (self, ed)
def recv (self, block_size):
data = secure_monitor_channel.recv (self, block_size)
if data:
dd = self.incoming.decrypt (data)
return dd
else:
return data
"Wrap send() and recv() with a stream cipher"
def __init__ (self, server, conn, addr):
key = server.password
self.outgoing = server.cipher.new (key)
self.incoming = server.cipher.new (key)
secure_monitor_channel.__init__ (self, server, conn, addr)
def send (self, data):
# send the encrypted data instead
ed = self.outgoing.encrypt (data)
return secure_monitor_channel.send (self, ed)
def recv (self, block_size):
data = secure_monitor_channel.recv (self, block_size)
if data:
dd = self.incoming.decrypt (data)
return dd
else:
return data
class secure_monitor_server (monitor_server):
channel_class = secure_monitor_channel
channel_class = secure_monitor_channel
def __init__ (self, password, hostname='', port=8023):
monitor_server.__init__ (self, hostname, port)
self.password = password
def __init__ (self, password, hostname='', port=8023):
monitor_server.__init__ (self, hostname, port)
self.password = password
def status (self):
p = monitor_server.status (self)
# kludge
p.data = p.data + ('<br><b>Failed Authorizations:</b> %d' % self.failed_auths)
return p
def status (self):
p = monitor_server.status (self)
# kludge
p.data = p.data + ('<br><b>Failed Authorizations:</b> %d' % self.failed_auths)
return p
# don't try to print from within any of the methods
# of this object. 8^)
class output_producer:
def __init__ (self, channel, real_stderr):
self.channel = channel
self.data = ''
# use _this_ for debug output
self.stderr = real_stderr
def check_data (self):
if len(self.data) > 1<<16:
# runaway output, close it.
self.channel.close()
def write (self, data):
lines = string.splitfields (data, '\n')
data = string.join (lines, '\r\n')
self.data = self.data + data
self.check_data()
def writeline (self, line):
self.data = self.data + line + '\r\n'
self.check_data()
def writelines (self, lines):
self.data = self.data + string.joinfields (
lines,
'\r\n'
) + '\r\n'
self.check_data()
def ready (self):
return (len (self.data) > 0)
def flush (self):
pass
def softspace (self, *args):
pass
def more (self):
if self.data:
result = self.data[:512]
self.data = self.data[512:]
return result
else:
return ''
def __init__ (self, channel, real_stderr):
self.channel = channel
self.data = ''
# use _this_ for debug output
self.stderr = real_stderr
def check_data (self):
if len(self.data) > 1<<16:
# runaway output, close it.
self.channel.close()
def write (self, data):
lines = string.splitfields (data, '\n')
data = string.join (lines, '\r\n')
self.data = self.data + data
self.check_data()
def writeline (self, line):
self.data = self.data + line + '\r\n'
self.check_data()
def writelines (self, lines):
self.data = self.data + string.joinfields (
lines,
'\r\n'
) + '\r\n'
self.check_data()
def ready (self):
return (len (self.data) > 0)
def flush (self):
pass
def softspace (self, *args):
pass
def more (self):
if self.data:
result = self.data[:512]
self.data = self.data[512:]
return result
else:
return ''
if __name__ == '__main__':
import string
import sys
if '-s' in sys.argv:
sys.argv.remove ('-s')
print 'Enter password: ',
password = raw_input()
else:
password = None
if '-e' in sys.argv:
sys.argv.remove ('-e')
encrypt = 1
else:
encrypt = 0
print sys.argv
if len(sys.argv) > 1:
port = string.atoi (sys.argv[1])
else:
port = 8023
if password is not None:
s = secure_monitor_server (password, '', port)
if encrypt:
s.channel_class = secure_encrypted_monitor_channel
import sapphire
s.cipher = sapphire
else:
s = monitor_server ('', port)
asyncore.loop()
import string
import sys
if '-s' in sys.argv:
sys.argv.remove ('-s')
print 'Enter password: ',
password = raw_input()
else:
password = None
if '-e' in sys.argv:
sys.argv.remove ('-e')
encrypt = 1
else:
encrypt = 0
print sys.argv
if len(sys.argv) > 1:
port = string.atoi (sys.argv[1])
else:
port = 8023
if password is not None:
s = secure_monitor_server (password, '', port)
if encrypt:
s.channel_class = secure_encrypted_monitor_channel
import sapphire
s.cipher = sapphire
else:
s = monitor_server ('', port)
asyncore.loop()
ZServer/medusa/monitor_client.py
View file @ 5e529dcf
......@@ -60,7 +60,7 @@ class monitor_client (asynchat.async_chat):
def handle_close (self):
# close all the channels, which will make the standard main
# loop exit.
map (lambda x: x.close(), asyncore.socket_map.keys())
map (lambda x: x.close(), asyncore.socket_map.values())
def log (self, *ignore):
pass
......
ZServer/medusa/select_trigger.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
VERSION_STRING = "$Id: select_trigger.py,v 1.8 1999/10/29 15:11:58 brian Exp $"
VERSION_STRING = "$Id: select_trigger.py,v 1.9 2000/01/14 02:35:56 amos Exp $"
import asyncore
import asynchat
......@@ -9,275 +9,259 @@ import os
import socket
import string
import thread
if os.name == 'posix':
class trigger (asyncore.file_dispatcher):
"Wake up a call to select() running in the main thread"
# This is useful in a context where you are using Medusa's I/O
# subsystem to deliver data, but the data is generated by another
# thread. Normally, if Medusa is in the middle of a call to
# select(), new output data generated by another thread will have
# to sit until the call to select() either times out or returns.
# If the trigger is 'pulled' by another thread, it should immediately
# generate a READ event on the trigger object, which will force the
# select() invocation to return.
# A common use for this facility: letting Medusa manage I/O for a
# large number of connections; but routing each request through a
# thread chosen from a fixed-size thread pool. When a thread is
# acquired, a transaction is performed, but output data is
# accumulated into buffers that will be emptied more efficiently
# by Medusa. [picture a server that can process database queries
# rapidly, but doesn't want to tie up threads waiting to send data
# to low-bandwidth connections]
# The other major feature provided by this class is the ability to
# move work back into the main thread: if you call pull_trigger()
# with a thunk argument, when select() wakes up and receives the
# event it will call your thunk from within that thread. The main
# purpose of this is to remove the need to wrap thread locks around
# Medusa's data structures, which normally do not need them. [To see
# why this is true, imagine this scenario: A thread tries to push some
# new data onto a channel's outgoing data queue at the same time that
# the main thread is trying to remove some]
def __init__ (self):
r, w = os.pipe()
self.trigger = w
asyncore.file_dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
def __repr__ (self):
return '<select-trigger (pipe) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
# print 'PULL_TRIGGER: ', len(self.thunks)
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
os.write (self.trigger, 'x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info(
'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo),
'error')
self.thunks = []
finally:
self.lock.release()
class trigger (asyncore.file_dispatcher):
"Wake up a call to select() running in the main thread"
# This is useful in a context where you are using Medusa's I/O
# subsystem to deliver data, but the data is generated by another
# thread. Normally, if Medusa is in the middle of a call to
# select(), new output data generated by another thread will have
# to sit until the call to select() either times out or returns.
# If the trigger is 'pulled' by another thread, it should immediately
# generate a READ event on the trigger object, which will force the
# select() invocation to return.
# A common use for this facility: letting Medusa manage I/O for a
# large number of connections; but routing each request through a
# thread chosen from a fixed-size thread pool. When a thread is
# acquired, a transaction is performed, but output data is
# accumulated into buffers that will be emptied more efficiently
# by Medusa. [picture a server that can process database queries
# rapidly, but doesn't want to tie up threads waiting to send data
# to low-bandwidth connections]
# The other major feature provided by this class is the ability to
# move work back into the main thread: if you call pull_trigger()
# with a thunk argument, when select() wakes up and receives the
# event it will call your thunk from within that thread. The main
# purpose of this is to remove the need to wrap thread locks around
# Medusa's data structures, which normally do not need them. [To see
# why this is true, imagine this scenario: A thread tries to push some
# new data onto a channel's outgoing data queue at the same time that
# the main thread is trying to remove some]
def __init__ (self):
r, w = os.pipe()
self.trigger = w
asyncore.file_dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
def __repr__ (self):
return '<select-trigger (pipe) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
# print 'PULL_TRIGGER: ', len(self.thunks)
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
os.write (self.trigger, 'x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
print 'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo)
self.thunks = []
finally:
self.lock.release()
else:
class trigger (asyncore.dispatcher):
address = ('127.9.9.9', 19999)
def __init__ (self):
a = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
w = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
# set TCP_NODELAY to true to avoid buffering
w.setsockopt(socket.IPPROTO_TCP, 1, 1)
# tricky: get a pair of connected sockets
host='127.9.9.9'
port=19999
while 1:
try:
self.address=(host, port)
a.bind(self.address)
break
except:
if port <= 19950:
raise 'Bind Error', 'Cannot bind trigger!'
port=port - 1
a.listen (1)
w.setblocking (0)
try:
w.connect (self.address)
except:
pass
r, addr = a.accept()
a.close()
w.setblocking (1)
self.trigger = w
asyncore.dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
self._trigger_connected = 0
def __repr__ (self):
return '<select-trigger (loopback) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
self.trigger.send ('x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info(
'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo),
'error')
self.thunks = []
finally:
self.lock.release()
# win32-safe version
class trigger (asyncore.dispatcher):
address = ('127.9.9.9', 19999)
def __init__ (self):
a = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
w = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
# tricky: get a pair of connected sockets
a.bind (self.address)
a.listen (1)
w.setblocking (0)
try:
w.connect (self.address)
except:
pass
r, addr = a.accept()
a.close()
w.setblocking (1)
self.trigger = w
asyncore.dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
self._trigger_connected = 0
def __repr__ (self):
return '<select-trigger (loopback) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
self.trigger.send ('x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
print 'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo)
self.thunks = []
finally:
self.lock.release()
the_trigger = None
class trigger_file:
"A 'triggered' file object"
buffer_size = 4096
def __init__ (self, parent):
global the_trigger
if the_trigger is None:
the_trigger = trigger()
self.parent = parent
self.buffer = ''
def write (self, data):
self.buffer = self.buffer + data
if len(self.buffer) > self.buffer_size:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda d=d,p=self.parent: p.push (d)
)
def writeline (self, line):
self.write (line+'\r\n')
def writelines (self, lines):
self.write (
string.joinfields (
lines,
'\r\n'
) + '\r\n'
)
def flush (self):
if self.buffer:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda p=self.parent,d=d: p.push (d)
)
def softspace (self, *args):
pass
def close (self):
# in a derived class, you may want to call trigger_close() instead.
self.flush()
self.parent = None
def trigger_close (self):
d, self.buffer = self.buffer, ''
p, self.parent = self.parent, None
the_trigger.pull_trigger (
lambda p=p,d=d: (p.push(d), p.close_when_done())
)
"A 'triggered' file object"
buffer_size = 4096
def __init__ (self, parent):
global the_trigger
if the_trigger is None:
the_trigger = trigger()
self.parent = parent
self.buffer = ''
def write (self, data):
self.buffer = self.buffer + data
if len(self.buffer) > self.buffer_size:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda d=d,p=self.parent: p.push (d)
)
def writeline (self, line):
self.write (line+'\r\n')
def writelines (self, lines):
self.write (
string.joinfields (
lines,
'\r\n'
) + '\r\n'
)
def flush (self):
if self.buffer:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda p=self.parent,d=d: p.push (d)
)
def softspace (self, *args):
pass
def close (self):
# in a derived class, you may want to call trigger_close() instead.
self.flush()
self.parent = None
def trigger_close (self):
d, self.buffer = self.buffer, ''
p, self.parent = self.parent, None
the_trigger.pull_trigger (
lambda p=p,d=d: (p.push(d), p.close_when_done())
)
if __name__ == '__main__':
import time
def thread_function (output_file, i, n):
print 'entering thread_function'
while n:
time.sleep (5)
output_file.write ('%2d.%2d %s\r\n' % (i, n, output_file))
output_file.flush()
n = n - 1
output_file.close()
print 'exiting thread_function'
class thread_parent (asynchat.async_chat):
def __init__ (self, conn, addr):
self.addr = addr
asynchat.async_chat.__init__ (self, conn)
self.set_terminator ('\r\n')
self.buffer = ''
self.count = 0
def collect_incoming_data (self, data):
self.buffer = self.buffer + data
def found_terminator (self):
data, self.buffer = self.buffer, ''
if not data:
asyncore.close_all()
print "done"
return
n = string.atoi (string.split (data)[0])
tf = trigger_file (self)
self.count = self.count + 1
thread.start_new_thread (thread_function, (tf, self.count, n))
class thread_server (asyncore.dispatcher):
def __init__ (self, family=socket.AF_INET, address=('', 9003)):
asyncore.dispatcher.__init__ (self)
self.create_socket (family, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind (address)
self.listen (5)
def handle_accept (self):
conn, addr = self.accept()
tp = thread_parent (conn, addr)
thread_server()
#asyncore.loop(1.0, use_poll=1)
try:
asyncore.loop ()
except:
asyncore.close_all()
import time
def thread_function (output_file, i, n):
print 'entering thread_function'
while n:
time.sleep (5)
output_file.write ('%2d.%2d %s\r\n' % (i, n, output_file))
output_file.flush()
n = n - 1
output_file.close()
print 'exiting thread_function'
class thread_parent (asynchat.async_chat):
def __init__ (self, conn, addr):
self.addr = addr
asynchat.async_chat.__init__ (self, conn)
self.set_terminator ('\r\n')
self.buffer = ''
self.count = 0
def collect_incoming_data (self, data):
self.buffer = self.buffer + data
def found_terminator (self):
data, self.buffer = self.buffer, ''
if not data:
asyncore.close_all()
print "done"
return
n = string.atoi (string.split (data)[0])
tf = trigger_file (self)
self.count = self.count + 1
thread.start_new_thread (thread_function, (tf, self.count, n))
class thread_server (asyncore.dispatcher):
def __init__ (self, family=socket.AF_INET, address=('', 9003)):
asyncore.dispatcher.__init__ (self)
self.create_socket (family, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind (address)
self.listen (5)
def handle_accept (self):
conn, addr = self.accept()
tp = thread_parent (conn, addr)
thread_server()
#asyncore.loop(1.0, use_poll=1)
try:
asyncore.loop ()
except:
asyncore.close_all()
lib/python/ZServer/medusa/asynchat.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
# $Id: asynchat.py,v 1.9 1999/07/19 17:43:47 amos Exp $
# $Id: asynchat.py,v 1.10 2000/01/14 02:35:56 amos Exp $
# Author: Sam Rushing <rushing@nightmare.com>
# ======================================================================
......@@ -219,7 +219,7 @@ class async_chat (asyncore.dispatcher):
def discard_buffers (self):
# Emergencies only!
self.ac_in_buffer = ''
self.ac_out_buffer == ''
self.ac_out_buffer = ''
while self.producer_fifo:
self.producer_fifo.pop()
......
lib/python/ZServer/medusa/asyncore.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
# $Id: asyncore.py,v 1.6 1999/07/26 07:06:36 amos Exp $
# $Id: asyncore.py,v 1.7 2000/01/14 02:35:56 amos Exp $
# Author: Sam Rushing <rushing@nightmare.com>
# ======================================================================
......@@ -25,6 +25,7 @@
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
# ======================================================================
import exceptions
import select
import socket
import string
......@@ -41,60 +42,83 @@ if os.name == 'nt':
else:
from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, ENOTCONN, ESHUTDOWN
socket_map = {}
try:
socket_map
except NameError:
socket_map = {}
class ExitNow (exceptions.Exception):
pass
DEBUG = 0
def poll (timeout=0.0):
global DEBUG
if socket_map:
r = []; w = []; e = []
for s in socket_map.keys():
if s.readable():
r.append (s)
if s.writable():
w.append (s)
for fd, obj in socket_map.items():
if obj.readable():
r.append (fd)
if obj.writable():
w.append (fd)
r,w,e = select.select (r,w,e, timeout)
(r,w,e) = select.select (r,w,e, timeout)
if DEBUG:
print r,w,e
for x in r:
for fd in r:
try:
x.handle_read_event()
except:
x.handle_error()
for x in w:
obj = socket_map[fd]
try:
obj.handle_read_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
for fd in w:
try:
x.handle_write_event()
except:
x.handle_error()
obj = socket_map[fd]
try:
obj.handle_write_event()
except ExitNow:
raise ExitNow
except:
obj.handle_error()
except KeyError:
pass
def poll2 (timeout=0.0):
import poll
# timeout is in milliseconds
timeout = int(timeout*1000)
if socket_map:
fd_map = {}
for s in socket_map.keys():
fd_map[s.fileno()] = s
l = []
for fd, s in fd_map.items():
for fd, obj in socket_map.items():
flags = 0
if s.readable():
if obj.readable():
flags = poll.POLLIN
if s.writable():
if obj.writable():
flags = flags | poll.POLLOUT
if flags:
l.append (fd, flags)
r = poll.poll (l, timeout)
for fd, flags in r:
s = fd_map[fd]
try:
if (flags & poll.POLLIN):
s.handle_read_event()
if (flags & poll.POLLOUT):
s.handle_write_event()
if (flags & poll.POLLERR):
s.handle_expt_event()
except:
s.handle_error()
obj = socket_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):
......@@ -143,23 +167,25 @@ class dispatcher:
return '<__repr__ (self) failed for object at %x (addr=%s)>' % (id(self),ar)
def add_channel (self):
self.log_info ('adding channel %s' % self)
socket_map [self] = 1
#self.log_info ('adding channel %s' % self)
socket_map [self._fileno] = self
def del_channel (self):
if socket_map.has_key (self):
self.log_info ('closing channel %d:%s' % (self.fileno(), self))
del socket_map [self]
fd = self._fileno
if socket_map.has_key (fd):
#self.log_info ('closing channel %d:%s' % (fd, self))
del socket_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, socket):
# This is done so we can be called safely from __init__
self.__dict__['socket'] = socket
def set_socket (self, sock):
self.__dict__['socket'] = sock
self._fileno = sock.fileno()
self.add_channel()
def set_reuse_addr (self):
......@@ -261,20 +287,19 @@ class dispatcher:
# cheap inheritance, used to pass all other attribute
# references to the underlying socket object.
# NOTE: this may be removed soon for performance reasons.
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):
print 'log:', 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)
print '%s: %s' % (type, message)
def handle_read_event (self):
if self.accepting:
......@@ -398,7 +423,7 @@ def compact_traceback ():
def close_all ():
global socket_map
for x in socket_map.keys():
for x in socket_map.values():
x.socket.close()
socket_map.clear()
......@@ -449,6 +474,7 @@ if os.name == 'posix':
self.set_file (fd)
def set_file (self, fd):
self._fileno = fd
self.socket = file_wrapper (fd)
self.add_channel()
lib/python/ZServer/medusa/ftp_server.py
View file @ 5e529dcf
......@@ -8,7 +8,7 @@
# If you are interested in using this software in a commercial context,
# or in purchasing support, please contact the author.
RCS_ID = '$Id: ftp_server.py,v 1.8 1999/11/15 21:53:56 amos Exp $'
RCS_ID = '$Id: ftp_server.py,v 1.9 2000/01/14 02:35:56 amos Exp $'
# An extensible, configurable, asynchronous FTP server.
#
......@@ -336,6 +336,7 @@ class ftp_channel (asynchat.async_chat):
self.current_mode = t
self.respond ('200 Type set to %s.' % self.type_map[t])
def cmd_quit (self, line):
'terminate session'
self.respond ('221 Goodbye.')
......@@ -847,13 +848,16 @@ class xmit_channel (asynchat.async_chat):
self.channel = channel
self.client_addr = client_addr
asynchat.async_chat.__init__ (self)
# def __del__ (self):
# print 'xmit_channel.__del__()'
def readable (self):
return 0
def log(self, *args):
def log (*args):
pass
def readable (self):
return not self.connected
def writable (self):
return 1
......@@ -1054,13 +1058,7 @@ if os.name == 'posix':
self.log_info('FTP server shutting down. (received SIGINT)', 'warning')
# close everything down on SIGINT.
# of course this should be a cleaner shutdown.
sm = socket.socket_map
socket.socket_map = {}
for sock in sm.values():
try:
sock.close()
except:
pass
asyncore.close_all()
if __name__ == '__main__':
test (sys.argv[1])
......
lib/python/ZServer/medusa/http_server.py
View file @ 5e529dcf
......@@ -9,7 +9,7 @@
# interested in using this software in a commercial context, or in
# purchasing support, please contact the author.
RCS_ID = '$Id: http_server.py,v 1.13 1999/11/15 21:53:56 amos Exp $'
RCS_ID = '$Id: http_server.py,v 1.14 2000/01/14 02:35:56 amos Exp $'
# python modules
import os
......@@ -269,7 +269,7 @@ class http_request:
user_agent=self.get_header('user-agent')
if not user_agent: user_agent=''
referer=self.get_header('referer')
if not referer: referer=''
if not referer: referer=''
self.channel.server.logger.log (
self.channel.addr[0],
' - - [%s] "%s" %d %d "%s" "%s"\n' % (
......@@ -385,7 +385,7 @@ class http_channel (asynchat.async_chat):
def kill_zombies (self):
now = int (time.time())
for channel in asyncore.socket_map.keys():
for channel in asyncore.socket_map.values():
if channel.__class__ == self.__class__:
if (now - channel.creation_time) > channel.zombie_timeout:
channel.close()
......@@ -448,7 +448,7 @@ class http_channel (asynchat.async_chat):
# --------------------------------------------------
# crack the request header
# --------------------------------------------------
while lines and not lines[0]:
# as per the suggestion of http-1.1 section 4.1, (and
# Eric Parker <eparker@zyvex.com>), ignore a leading
......@@ -461,27 +461,18 @@ class http_channel (asynchat.async_chat):
return
request = lines[0]
try:
command, uri, version = crack_request (request)
except:
# deal with broken HTTP requests
try:
# maybe there were spaces in the URL
parts=string.split(request)
command, uri, version = crack_request(
'%s %s %s' % (parts[0], parts[1], parts[-1]))
except:
self.log_info('Bad HTTP request: %s' % request, 'error')
r = http_request (self, request,
None, None, None, join_headers(lines[1:]))
r.error(400)
return
command, uri, version = crack_request (request)
header = join_headers (lines[1:])
r = http_request (self, request, command, uri, version, header)
self.request_counter.increment()
self.server.total_requests.increment()
if command is None:
self.log_info('Bad HTTP request: %s' % request, 'error')
r.error(400)
return
# --------------------------------------------------
# handler selection and dispatch
# --------------------------------------------------
......@@ -612,7 +603,14 @@ class http_server (asyncore.dispatcher):
# accept. socketmodule.c:makesockaddr complains that the
# address family is unknown. We don't want the whole server
# to shut down because of this.
self.log_info('Server accept() threw an exception', 'warning')
self.log_info ('warning: server accept() threw an exception', 'warning')
return
except TypeError:
# unpack non-sequence. this can happen when a read event
# fires on a listening socket, but when we call accept()
# we get EWOULDBLOCK, so dispatcher.accept() returns None.
# Seen on FreeBSD3.
self.log_info ('warning: server accept() threw EWOULDBLOCK', 'warning')
return
self.channel_class (self, conn, addr)
......@@ -689,6 +687,8 @@ def crack_request (r):
else:
version = None
return string.lower (REQUEST.group (1)), REQUEST.group(2), version
else:
return None, None, None
class fifo:
def __init__ (self, list=None):
......
lib/python/ZServer/medusa/monitor.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
# Author: Sam Rushing <rushing@nightmare.com>
# Author: Sam Rushing <rushing@nightmare.com>
#
# python REPL channel.
#
RCS_ID = '$Id: monitor.py,v 1.5 1999/07/21 17:15:53 klm Exp $'
RCS_ID = '$Id: monitor.py,v 1.6 2000/01/14 02:35:56 amos Exp $'
import md5
import socket
import string
import sys
import time
import traceback
VERSION = string.split(RCS_ID)[2]
......@@ -23,330 +22,325 @@ from counter import counter
import producers
class monitor_channel (asynchat.async_chat):
try_linemode = 1
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = {}
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.push ("[Hint: try 'from __main__ import *']\r\n")
self.prompt()
self.number = server.total_sessions.as_long()
self.line_counter = counter()
self.multi_line = []
def handle_connect (self):
# send IAC DO LINEMODE
self.push ('\377\375\"')
def close (self):
self.server.closed_sessions.increment()
asynchat.async_chat.close(self)
def prompt (self):
self.push ('>>> ')
def collect_incoming_data (self, data):
self.data = self.data + data
if len(self.data) > 1024:
# denial of service.
self.push ('BCNU\r\n')
self.close_when_done()
def found_terminator (self):
line = self.clean_line (self.data)
self.data = ''
self.line_counter.increment()
# check for special case inputs...
if not line and not self.multi_line:
self.prompt()
return
if line in ['\004', 'exit']:
self.push ('BCNU\r\n')
self.close_when_done()
return
oldout = sys.stdout
olderr = sys.stderr
try:
p = output_producer(self, olderr)
sys.stdout = p
sys.stderr = p
try:
# this is, of course, a blocking operation.
# if you wanted to thread this, you would have
# to synchronize, etc... and treat the output
# like a pipe. Not Fun.
#
# try eval first. If that fails, try exec. If that fails,
# hurl.
try:
if self.multi_line:
# oh, this is horrible...
raise SyntaxError
co = compile (line, repr(self), 'eval')
result = eval (co, self.local_env)
method = 'eval'
if result is not None:
self.log_info(repr(result))
print repr(result)
self.local_env['_'] = result
except SyntaxError:
try:
if self.multi_line:
if line and line[0] in [' ','\t']:
self.multi_line.append (line)
self.push ('... ')
return
else:
self.multi_line.append (line)
line = string.join (self.multi_line, '\n')
co = compile (line, repr(self), 'exec')
self.multi_line = []
else:
co = compile (line, repr(self), 'exec')
except SyntaxError, why:
if why[0] == 'unexpected EOF while parsing':
self.push ('... ')
self.multi_line.append (line)
return
exec co in self.local_env
method = 'exec'
except:
method = 'exception'
self.multi_line = []
t, v, tb = sys.exc_info()
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info('%s %s %s' %(t, v, tbinfo), 'warning')
traceback.print_exception(t, v, tb)
del tb
finally:
sys.stdout = oldout
sys.stderr = olderr
self.log_info('%s:%s (%s)> %s' % (
self.number,
self.line_counter,
method,
repr(line))
)
self.push_with_producer (p)
self.prompt()
# for now, we ignore any telnet option stuff sent to
# us, and we process the backspace key ourselves.
# gee, it would be fun to write a full-blown line-editing
# environment, etc...
def clean_line (self, line):
chars = []
for ch in line:
oc = ord(ch)
if oc < 127:
if oc in [8,177]:
# backspace
chars = chars[:-1]
else:
chars.append (ch)
return string.join (chars, '')
try_linemode = 1
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = sys.modules['__main__'].__dict__.copy()
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.push ("[Hint: try 'from __main__ import *']\r\n")
self.prompt()
self.number = server.total_sessions.as_long()
self.line_counter = counter()
self.multi_line = []
def handle_connect (self):
# send IAC DO LINEMODE
self.push ('\377\375\"')
def close (self):
self.server.closed_sessions.increment()
asynchat.async_chat.close(self)
def prompt (self):
self.push ('>>> ')
def collect_incoming_data (self, data):
self.data = self.data + data
if len(self.data) > 1024:
# denial of service.
self.push ('BCNU\r\n')
self.close_when_done()
def found_terminator (self):
line = self.clean_line (self.data)
self.data = ''
self.line_counter.increment()
# check for special case inputs...
if not line and not self.multi_line:
self.prompt()
return
if line in ['\004', 'exit']:
self.push ('BCNU\r\n')
self.close_when_done()
return
oldout = sys.stdout
olderr = sys.stderr
try:
p = output_producer(self, olderr)
sys.stdout = p
sys.stderr = p
try:
# this is, of course, a blocking operation.
# if you wanted to thread this, you would have
# to synchronize, etc... and treat the output
# like a pipe. Not Fun.
#
# try eval first. If that fails, try exec. If that fails,
# hurl.
try:
if self.multi_line:
# oh, this is horrible...
raise SyntaxError
co = compile (line, repr(self), 'eval')
result = eval (co, self.local_env)
method = 'eval'
if result is not None:
print repr(result)
self.local_env['_'] = result
except SyntaxError:
try:
if self.multi_line:
if line and line[0] in [' ','\t']:
self.multi_line.append (line)
self.push ('... ')
return
else:
self.multi_line.append (line)
line = string.join (self.multi_line, '\n')
co = compile (line, repr(self), 'exec')
self.multi_line = []
else:
co = compile (line, repr(self), 'exec')
except SyntaxError, why:
if why[0] == 'unexpected EOF while parsing':
self.push ('... ')
self.multi_line.append (line)
return
exec co in self.local_env
method = 'exec'
except:
method = 'exception'
self.multi_line = []
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info('%s %s %s' %(t, v, tbinfo), 'warning')
finally:
sys.stdout = oldout
sys.stderr = olderr
self.log_info('%s:%s (%s)> %s' % (
self.number,
self.line_counter,
method,
repr(line))
)
self.push_with_producer (p)
self.prompt()
# for now, we ignore any telnet option stuff sent to
# us, and we process the backspace key ourselves.
# gee, it would be fun to write a full-blown line-editing
# environment, etc...
def clean_line (self, line):
chars = []
for ch in line:
oc = ord(ch)
if oc < 127:
if oc in [8,177]:
# backspace
chars = chars[:-1]
else:
chars.append (ch)
return string.join (chars, '')
class monitor_server (asyncore.dispatcher):
SERVER_IDENT = 'Monitor Server (V%s)' % VERSION
channel_class = monitor_channel
def __init__ (self, hostname='127.0.0.1', port=8023):
self.hostname = hostname
self.port = port
self.create_socket (socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind ((hostname, port))
self.log_info('%s started on port %d' % (self.SERVER_IDENT, port))
self.listen (5)
self.closed = 0
self.failed_auths = 0
self.total_sessions = counter()
self.closed_sessions = counter()
def writable (self):
return 0
def handle_accept (self):
conn, addr = self.accept()
self.log_info('Incoming monitor connection from %s:%d' % addr)
self.channel_class (self, conn, addr)
self.total_sessions.increment()
def status (self):
return producers.simple_producer (
'<h2>%s</h2>' % self.SERVER_IDENT
+ '<br><b>Total Sessions:</b> %s' % self.total_sessions
+ '<br><b>Current Sessions:</b> %d' % (
self.total_sessions.as_long()-self.closed_sessions.as_long()
)
)
SERVER_IDENT = 'Monitor Server (V%s)' % VERSION
channel_class = monitor_channel
def __init__ (self, hostname='127.0.0.1', port=8023):
self.hostname = hostname
self.port = port
self.create_socket (socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind ((hostname, port))
self.log_info('%s started on port %d' % (self.SERVER_IDENT, port))
self.listen (5)
self.closed = 0
self.failed_auths = 0
self.total_sessions = counter()
self.closed_sessions = counter()
def writable (self):
return 0
def handle_accept (self):
conn, addr = self.accept()
self.log_info('Incoming monitor connection from %s:%d' % addr)
self.channel_class (self, conn, addr)
self.total_sessions.increment()
def status (self):
return producers.simple_producer (
'<h2>%s</h2>' % self.SERVER_IDENT
+ '<br><b>Total Sessions:</b> %s' % self.total_sessions
+ '<br><b>Current Sessions:</b> %d' % (
self.total_sessions.as_long()-self.closed_sessions.as_long()
)
)
def hex_digest (s):
m = md5.md5()
m.update (s)
return string.joinfields (
map (lambda x: hex (ord (x))[2:], map (None, m.digest())),
'',
)
m = md5.md5()
m.update (s)
return string.joinfields (
map (lambda x: hex (ord (x))[2:], map (None, m.digest())),
'',
)
class secure_monitor_channel (monitor_channel):
authorized = 0
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = {}
# send timestamp string
self.timestamp = str(time.time())
self.count = 0
self.line_counter = counter()
self.number = int(server.total_sessions.as_long())
self.multi_line = []
self.push (self.timestamp + '\r\n')
def found_terminator (self):
if not self.authorized:
if hex_digest ('%s%s' % (self.timestamp, self.server.password)) != self.data:
self.log_info ('%s: failed authorization' % self, 'warning')
self.server.failed_auths = self.server.failed_auths + 1
self.close()
else:
self.authorized = 1
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.push ("[Hint: try 'from __main__ import *']\r\n")
self.prompt()
self.data = ''
else:
monitor_channel.found_terminator (self)
authorized = 0
def __init__ (self, server, sock, addr):
asynchat.async_chat.__init__ (self, sock)
self.server = server
self.addr = addr
self.set_terminator ('\r\n')
self.data = ''
# local bindings specific to this channel
self.local_env = {}
# send timestamp string
self.timestamp = str(time.time())
self.count = 0
self.line_counter = counter()
self.number = int(server.total_sessions.as_long())
self.multi_line = []
self.push (self.timestamp + '\r\n')
def found_terminator (self):
if not self.authorized:
if hex_digest ('%s%s' % (self.timestamp, self.server.password)) != self.data:
self.log_info ('%s: failed authorization' % self, 'warning')
self.server.failed_auths = self.server.failed_auths + 1
self.close()
else:
self.authorized = 1
self.push ('Python ' + sys.version + '\r\n')
self.push (sys.copyright+'\r\n')
self.push ('Welcome to %s\r\n' % self)
self.prompt()
self.data = ''
else:
monitor_channel.found_terminator (self)
class secure_encrypted_monitor_channel (secure_monitor_channel):
"Wrap send() and recv() with a stream cipher"
def __init__ (self, server, conn, addr):
key = server.password
self.outgoing = server.cipher.new (key)
self.incoming = server.cipher.new (key)
secure_monitor_channel.__init__ (self, server, conn, addr)
def send (self, data):
# send the encrypted data instead
ed = self.outgoing.encrypt (data)
return secure_monitor_channel.send (self, ed)
def recv (self, block_size):
data = secure_monitor_channel.recv (self, block_size)
if data:
dd = self.incoming.decrypt (data)
return dd
else:
return data
"Wrap send() and recv() with a stream cipher"
def __init__ (self, server, conn, addr):
key = server.password
self.outgoing = server.cipher.new (key)
self.incoming = server.cipher.new (key)
secure_monitor_channel.__init__ (self, server, conn, addr)
def send (self, data):
# send the encrypted data instead
ed = self.outgoing.encrypt (data)
return secure_monitor_channel.send (self, ed)
def recv (self, block_size):
data = secure_monitor_channel.recv (self, block_size)
if data:
dd = self.incoming.decrypt (data)
return dd
else:
return data
class secure_monitor_server (monitor_server):
channel_class = secure_monitor_channel
channel_class = secure_monitor_channel
def __init__ (self, password, hostname='', port=8023):
monitor_server.__init__ (self, hostname, port)
self.password = password
def __init__ (self, password, hostname='', port=8023):
monitor_server.__init__ (self, hostname, port)
self.password = password
def status (self):
p = monitor_server.status (self)
# kludge
p.data = p.data + ('<br><b>Failed Authorizations:</b> %d' % self.failed_auths)
return p
def status (self):
p = monitor_server.status (self)
# kludge
p.data = p.data + ('<br><b>Failed Authorizations:</b> %d' % self.failed_auths)
return p
# don't try to print from within any of the methods
# of this object. 8^)
class output_producer:
def __init__ (self, channel, real_stderr):
self.channel = channel
self.data = ''
# use _this_ for debug output
self.stderr = real_stderr
def check_data (self):
if len(self.data) > 1<<16:
# runaway output, close it.
self.channel.close()
def write (self, data):
lines = string.splitfields (data, '\n')
data = string.join (lines, '\r\n')
self.data = self.data + data
self.check_data()
def writeline (self, line):
self.data = self.data + line + '\r\n'
self.check_data()
def writelines (self, lines):
self.data = self.data + string.joinfields (
lines,
'\r\n'
) + '\r\n'
self.check_data()
def ready (self):
return (len (self.data) > 0)
def flush (self):
pass
def softspace (self, *args):
pass
def more (self):
if self.data:
result = self.data[:512]
self.data = self.data[512:]
return result
else:
return ''
def __init__ (self, channel, real_stderr):
self.channel = channel
self.data = ''
# use _this_ for debug output
self.stderr = real_stderr
def check_data (self):
if len(self.data) > 1<<16:
# runaway output, close it.
self.channel.close()
def write (self, data):
lines = string.splitfields (data, '\n')
data = string.join (lines, '\r\n')
self.data = self.data + data
self.check_data()
def writeline (self, line):
self.data = self.data + line + '\r\n'
self.check_data()
def writelines (self, lines):
self.data = self.data + string.joinfields (
lines,
'\r\n'
) + '\r\n'
self.check_data()
def ready (self):
return (len (self.data) > 0)
def flush (self):
pass
def softspace (self, *args):
pass
def more (self):
if self.data:
result = self.data[:512]
self.data = self.data[512:]
return result
else:
return ''
if __name__ == '__main__':
import string
import sys
if '-s' in sys.argv:
sys.argv.remove ('-s')
print 'Enter password: ',
password = raw_input()
else:
password = None
if '-e' in sys.argv:
sys.argv.remove ('-e')
encrypt = 1
else:
encrypt = 0
print sys.argv
if len(sys.argv) > 1:
port = string.atoi (sys.argv[1])
else:
port = 8023
if password is not None:
s = secure_monitor_server (password, '', port)
if encrypt:
s.channel_class = secure_encrypted_monitor_channel
import sapphire
s.cipher = sapphire
else:
s = monitor_server ('', port)
asyncore.loop()
import string
import sys
if '-s' in sys.argv:
sys.argv.remove ('-s')
print 'Enter password: ',
password = raw_input()
else:
password = None
if '-e' in sys.argv:
sys.argv.remove ('-e')
encrypt = 1
else:
encrypt = 0
print sys.argv
if len(sys.argv) > 1:
port = string.atoi (sys.argv[1])
else:
port = 8023
if password is not None:
s = secure_monitor_server (password, '', port)
if encrypt:
s.channel_class = secure_encrypted_monitor_channel
import sapphire
s.cipher = sapphire
else:
s = monitor_server ('', port)
asyncore.loop()
lib/python/ZServer/medusa/monitor_client.py
View file @ 5e529dcf
......@@ -60,7 +60,7 @@ class monitor_client (asynchat.async_chat):
def handle_close (self):
# close all the channels, which will make the standard main
# loop exit.
map (lambda x: x.close(), asyncore.socket_map.keys())
map (lambda x: x.close(), asyncore.socket_map.values())
def log (self, *ignore):
pass
......
lib/python/ZServer/medusa/select_trigger.py
View file @ 5e529dcf
# -*- Mode: Python; tab-width: 4 -*-
VERSION_STRING = "$Id: select_trigger.py,v 1.8 1999/10/29 15:11:58 brian Exp $"
VERSION_STRING = "$Id: select_trigger.py,v 1.9 2000/01/14 02:35:56 amos Exp $"
import asyncore
import asynchat
......@@ -9,275 +9,259 @@ import os
import socket
import string
import thread
if os.name == 'posix':
class trigger (asyncore.file_dispatcher):
"Wake up a call to select() running in the main thread"
# This is useful in a context where you are using Medusa's I/O
# subsystem to deliver data, but the data is generated by another
# thread. Normally, if Medusa is in the middle of a call to
# select(), new output data generated by another thread will have
# to sit until the call to select() either times out or returns.
# If the trigger is 'pulled' by another thread, it should immediately
# generate a READ event on the trigger object, which will force the
# select() invocation to return.
# A common use for this facility: letting Medusa manage I/O for a
# large number of connections; but routing each request through a
# thread chosen from a fixed-size thread pool. When a thread is
# acquired, a transaction is performed, but output data is
# accumulated into buffers that will be emptied more efficiently
# by Medusa. [picture a server that can process database queries
# rapidly, but doesn't want to tie up threads waiting to send data
# to low-bandwidth connections]
# The other major feature provided by this class is the ability to
# move work back into the main thread: if you call pull_trigger()
# with a thunk argument, when select() wakes up and receives the
# event it will call your thunk from within that thread. The main
# purpose of this is to remove the need to wrap thread locks around
# Medusa's data structures, which normally do not need them. [To see
# why this is true, imagine this scenario: A thread tries to push some
# new data onto a channel's outgoing data queue at the same time that
# the main thread is trying to remove some]
def __init__ (self):
r, w = os.pipe()
self.trigger = w
asyncore.file_dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
def __repr__ (self):
return '<select-trigger (pipe) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
# print 'PULL_TRIGGER: ', len(self.thunks)
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
os.write (self.trigger, 'x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info(
'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo),
'error')
self.thunks = []
finally:
self.lock.release()
class trigger (asyncore.file_dispatcher):
"Wake up a call to select() running in the main thread"
# This is useful in a context where you are using Medusa's I/O
# subsystem to deliver data, but the data is generated by another
# thread. Normally, if Medusa is in the middle of a call to
# select(), new output data generated by another thread will have
# to sit until the call to select() either times out or returns.
# If the trigger is 'pulled' by another thread, it should immediately
# generate a READ event on the trigger object, which will force the
# select() invocation to return.
# A common use for this facility: letting Medusa manage I/O for a
# large number of connections; but routing each request through a
# thread chosen from a fixed-size thread pool. When a thread is
# acquired, a transaction is performed, but output data is
# accumulated into buffers that will be emptied more efficiently
# by Medusa. [picture a server that can process database queries
# rapidly, but doesn't want to tie up threads waiting to send data
# to low-bandwidth connections]
# The other major feature provided by this class is the ability to
# move work back into the main thread: if you call pull_trigger()
# with a thunk argument, when select() wakes up and receives the
# event it will call your thunk from within that thread. The main
# purpose of this is to remove the need to wrap thread locks around
# Medusa's data structures, which normally do not need them. [To see
# why this is true, imagine this scenario: A thread tries to push some
# new data onto a channel's outgoing data queue at the same time that
# the main thread is trying to remove some]
def __init__ (self):
r, w = os.pipe()
self.trigger = w
asyncore.file_dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
def __repr__ (self):
return '<select-trigger (pipe) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
# print 'PULL_TRIGGER: ', len(self.thunks)
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
os.write (self.trigger, 'x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
print 'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo)
self.thunks = []
finally:
self.lock.release()
else:
class trigger (asyncore.dispatcher):
address = ('127.9.9.9', 19999)
def __init__ (self):
a = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
w = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
# set TCP_NODELAY to true to avoid buffering
w.setsockopt(socket.IPPROTO_TCP, 1, 1)
# tricky: get a pair of connected sockets
host='127.9.9.9'
port=19999
while 1:
try:
self.address=(host, port)
a.bind(self.address)
break
except:
if port <= 19950:
raise 'Bind Error', 'Cannot bind trigger!'
port=port - 1
a.listen (1)
w.setblocking (0)
try:
w.connect (self.address)
except:
pass
r, addr = a.accept()
a.close()
w.setblocking (1)
self.trigger = w
asyncore.dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
self._trigger_connected = 0
def __repr__ (self):
return '<select-trigger (loopback) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
self.trigger.send ('x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
self.log_info(
'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo),
'error')
self.thunks = []
finally:
self.lock.release()
# win32-safe version
class trigger (asyncore.dispatcher):
address = ('127.9.9.9', 19999)
def __init__ (self):
a = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
w = socket.socket (socket.AF_INET, socket.SOCK_STREAM)
# tricky: get a pair of connected sockets
a.bind (self.address)
a.listen (1)
w.setblocking (0)
try:
w.connect (self.address)
except:
pass
r, addr = a.accept()
a.close()
w.setblocking (1)
self.trigger = w
asyncore.dispatcher.__init__ (self, r)
self.lock = thread.allocate_lock()
self.thunks = []
self._trigger_connected = 0
def __repr__ (self):
return '<select-trigger (loopback) at %x>' % id(self)
def readable (self):
return 1
def writable (self):
return 0
def handle_connect (self):
pass
def pull_trigger (self, thunk=None):
if thunk:
try:
self.lock.acquire()
self.thunks.append (thunk)
finally:
self.lock.release()
self.trigger.send ('x')
def handle_read (self):
self.recv (8192)
try:
self.lock.acquire()
for thunk in self.thunks:
try:
thunk()
except:
(file, fun, line), t, v, tbinfo = asyncore.compact_traceback()
print 'exception in trigger thunk: (%s:%s %s)' % (t, v, tbinfo)
self.thunks = []
finally:
self.lock.release()
the_trigger = None
class trigger_file:
"A 'triggered' file object"
buffer_size = 4096
def __init__ (self, parent):
global the_trigger
if the_trigger is None:
the_trigger = trigger()
self.parent = parent
self.buffer = ''
def write (self, data):
self.buffer = self.buffer + data
if len(self.buffer) > self.buffer_size:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda d=d,p=self.parent: p.push (d)
)
def writeline (self, line):
self.write (line+'\r\n')
def writelines (self, lines):
self.write (
string.joinfields (
lines,
'\r\n'
) + '\r\n'
)
def flush (self):
if self.buffer:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda p=self.parent,d=d: p.push (d)
)
def softspace (self, *args):
pass
def close (self):
# in a derived class, you may want to call trigger_close() instead.
self.flush()
self.parent = None
def trigger_close (self):
d, self.buffer = self.buffer, ''
p, self.parent = self.parent, None
the_trigger.pull_trigger (
lambda p=p,d=d: (p.push(d), p.close_when_done())
)
"A 'triggered' file object"
buffer_size = 4096
def __init__ (self, parent):
global the_trigger
if the_trigger is None:
the_trigger = trigger()
self.parent = parent
self.buffer = ''
def write (self, data):
self.buffer = self.buffer + data
if len(self.buffer) > self.buffer_size:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda d=d,p=self.parent: p.push (d)
)
def writeline (self, line):
self.write (line+'\r\n')
def writelines (self, lines):
self.write (
string.joinfields (
lines,
'\r\n'
) + '\r\n'
)
def flush (self):
if self.buffer:
d, self.buffer = self.buffer, ''
the_trigger.pull_trigger (
lambda p=self.parent,d=d: p.push (d)
)
def softspace (self, *args):
pass
def close (self):
# in a derived class, you may want to call trigger_close() instead.
self.flush()
self.parent = None
def trigger_close (self):
d, self.buffer = self.buffer, ''
p, self.parent = self.parent, None
the_trigger.pull_trigger (
lambda p=p,d=d: (p.push(d), p.close_when_done())
)
if __name__ == '__main__':
import time
def thread_function (output_file, i, n):
print 'entering thread_function'
while n:
time.sleep (5)
output_file.write ('%2d.%2d %s\r\n' % (i, n, output_file))
output_file.flush()
n = n - 1
output_file.close()
print 'exiting thread_function'
class thread_parent (asynchat.async_chat):
def __init__ (self, conn, addr):
self.addr = addr
asynchat.async_chat.__init__ (self, conn)
self.set_terminator ('\r\n')
self.buffer = ''
self.count = 0
def collect_incoming_data (self, data):
self.buffer = self.buffer + data
def found_terminator (self):
data, self.buffer = self.buffer, ''
if not data:
asyncore.close_all()
print "done"
return
n = string.atoi (string.split (data)[0])
tf = trigger_file (self)
self.count = self.count + 1
thread.start_new_thread (thread_function, (tf, self.count, n))
class thread_server (asyncore.dispatcher):
def __init__ (self, family=socket.AF_INET, address=('', 9003)):
asyncore.dispatcher.__init__ (self)
self.create_socket (family, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind (address)
self.listen (5)
def handle_accept (self):
conn, addr = self.accept()
tp = thread_parent (conn, addr)
thread_server()
#asyncore.loop(1.0, use_poll=1)
try:
asyncore.loop ()
except:
asyncore.close_all()
import time
def thread_function (output_file, i, n):
print 'entering thread_function'
while n:
time.sleep (5)
output_file.write ('%2d.%2d %s\r\n' % (i, n, output_file))
output_file.flush()
n = n - 1
output_file.close()
print 'exiting thread_function'
class thread_parent (asynchat.async_chat):
def __init__ (self, conn, addr):
self.addr = addr
asynchat.async_chat.__init__ (self, conn)
self.set_terminator ('\r\n')
self.buffer = ''
self.count = 0
def collect_incoming_data (self, data):
self.buffer = self.buffer + data
def found_terminator (self):
data, self.buffer = self.buffer, ''
if not data:
asyncore.close_all()
print "done"
return
n = string.atoi (string.split (data)[0])
tf = trigger_file (self)
self.count = self.count + 1
thread.start_new_thread (thread_function, (tf, self.count, n))
class thread_server (asyncore.dispatcher):
def __init__ (self, family=socket.AF_INET, address=('', 9003)):
asyncore.dispatcher.__init__ (self)
self.create_socket (family, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind (address)
self.listen (5)
def handle_accept (self):
conn, addr = self.accept()
tp = thread_parent (conn, addr)
thread_server()
#asyncore.loop(1.0, use_poll=1)
try:
asyncore.loop ()
except:
asyncore.close_all()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7