G
Giampaolo Rodola'
Hi,
I post this message here in the hope someone using asyncore could
review
this.
Since the thing I miss mostly in asyncore is a system for calling a
function after a certain amount of time, I spent the last 3 days
trying
to implement this with the hopes that this could be included in
asyncore
in the the future.
The logic consists in calling a certain function (the "scheduler") at
every loop to check if it is the proper time to call one or more
scheduled functions.
Such functions are scheduled by the new delayed_call class which is
very
similar to the DelayedCall class defined in /twisted/internet/base.py
I
drew on.
It provides a basic API which can be used for setting, resetting and
canceling the scheduled functions.
For better performance I used an heap queue structure. This way the
scheduler() only needs to check the scheduled functions due to expire
soonest.
diff file and modified asyncore.py can be found here:
http://bugs.python.org/issue1641
The following code sample implements an idle-timeout capability using
the attached modified asyncore library.
--- code snippet ---
import asyncore, asynchat, socket
class foo(asynchat.async_chat):
def __init__(self, conn=None):
asynchat.async_chat.__init__(self, conn)
self.set_terminator(None)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.connect(('127.0.0.1', 21))
self.scheduled_timeout = self.call_later(120,
self.handle_timeout)
def collect_incoming_data(self, data):
self.scheduled_timeout.reset()
# do something with the data...
def handle_timeout(self):
self.push("500 Connection timed out.\r\n")
self.close_when_done()
def close(self):
if not self.scheduled_timeout.cancelled:
self.scheduled_timeout.cancel()
asyncore.dispatcher.close(self)
foo()
asyncore.loop()
--- /code snippet ---
Today I played a little more with it and I tried to add bandwidth
throttling capabilities to the base asynchat.py.
The code could be surely improved but it's just an example to show
another useful feature which wouldn't be possible to implement without
having a "call_later" function under the hood:
--- code snippet ---
class throttled_async_chat(asynchat.async_chat):
# maximum number of bytes to transmit in a second (0 == no limit)
read_limit = 100 * 1024
write_limit = 100 * 1024
# smaller the buffers, the less bursty and smoother the throughput
ac_in_buffer_size = 2048
ac_out_buffer_size = 2048
def __init__(self, conn=None):
asynchat.async_chat.__init__(self, conn)
self.read_this_second = 0
self.written_this_second = 0
self.r_timenext = 0
self.w_timenext = 0
self.r_sleep = False
self.w_sleep = False
self.delayed_r = None
self.delayed_w = None
def readable(self):
return asynchat.async_chat.readable(self) and not self.r_sleep
def writable(self):
return asynchat.async_chat.writable(self) and not self.w_sleep
def recv(self, buffer_size):
chunk = asyncore.dispatcher.recv(self, buffer_size)
if self.read_limit:
self.read_this_second += len(chunk)
self.throttle_read()
return chunk
def send(self, data):
num_sent = asyncore.dispatcher.send(self, data)
if self.write_limit:
self.written_this_second += num_sent
self.throttle_write()
return num_sent
def throttle_read(self):
if self.read_this_second >= self.read_limit:
self.read_this_second = 0
now = time.time()
sleepfor = self.r_timenext - now
if sleepfor > 0:
# we've passed bandwidth limits
self.r_sleep = True
def unthrottle():
self.r_sleep = False
self.delayed_r = self.call_later((sleepfor * 2),
unthrottle)
self.r_timenext = now + 1
def throttle_write(self):
if self.written_this_second >= self.write_limit:
self.written_this_second = 0
now = time.time()
sleepfor = self.w_timenext - now
if sleepfor > 0:
# we've passed bandwidth limits
self.w_sleep = True
def unthrottle():
self.w_sleep = False
self.delayed_w = self.call_later((sleepfor * 2),
unthrottle)
self.w_timenext = now + 1
def close(self):
if self.delayed_r and not self.delayed_r.cancelled:
self.delayed_r.cancel()
if self.delayed_w and not self.delayed_w.cancelled:
self.delayed_w.cancel()
asyncore.dispatcher.close(self)
--- /code snippet ---
I don't know if there's a better way to implement this "call_later"
feature.
Maybe someone experienced with Twisted could provide a better
approach.
I would ask someone using asyncore to review this since, IMHO, it
would
fill a very big gap.
I'm an habitual asyncore user and I miss such thing very much.
Ciao,
-- Giampaolo
I post this message here in the hope someone using asyncore could
review
this.
Since the thing I miss mostly in asyncore is a system for calling a
function after a certain amount of time, I spent the last 3 days
trying
to implement this with the hopes that this could be included in
asyncore
in the the future.
The logic consists in calling a certain function (the "scheduler") at
every loop to check if it is the proper time to call one or more
scheduled functions.
Such functions are scheduled by the new delayed_call class which is
very
similar to the DelayedCall class defined in /twisted/internet/base.py
I
drew on.
It provides a basic API which can be used for setting, resetting and
canceling the scheduled functions.
For better performance I used an heap queue structure. This way the
scheduler() only needs to check the scheduled functions due to expire
soonest.
diff file and modified asyncore.py can be found here:
http://bugs.python.org/issue1641
The following code sample implements an idle-timeout capability using
the attached modified asyncore library.
--- code snippet ---
import asyncore, asynchat, socket
class foo(asynchat.async_chat):
def __init__(self, conn=None):
asynchat.async_chat.__init__(self, conn)
self.set_terminator(None)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.connect(('127.0.0.1', 21))
self.scheduled_timeout = self.call_later(120,
self.handle_timeout)
def collect_incoming_data(self, data):
self.scheduled_timeout.reset()
# do something with the data...
def handle_timeout(self):
self.push("500 Connection timed out.\r\n")
self.close_when_done()
def close(self):
if not self.scheduled_timeout.cancelled:
self.scheduled_timeout.cancel()
asyncore.dispatcher.close(self)
foo()
asyncore.loop()
--- /code snippet ---
Today I played a little more with it and I tried to add bandwidth
throttling capabilities to the base asynchat.py.
The code could be surely improved but it's just an example to show
another useful feature which wouldn't be possible to implement without
having a "call_later" function under the hood:
--- code snippet ---
class throttled_async_chat(asynchat.async_chat):
# maximum number of bytes to transmit in a second (0 == no limit)
read_limit = 100 * 1024
write_limit = 100 * 1024
# smaller the buffers, the less bursty and smoother the throughput
ac_in_buffer_size = 2048
ac_out_buffer_size = 2048
def __init__(self, conn=None):
asynchat.async_chat.__init__(self, conn)
self.read_this_second = 0
self.written_this_second = 0
self.r_timenext = 0
self.w_timenext = 0
self.r_sleep = False
self.w_sleep = False
self.delayed_r = None
self.delayed_w = None
def readable(self):
return asynchat.async_chat.readable(self) and not self.r_sleep
def writable(self):
return asynchat.async_chat.writable(self) and not self.w_sleep
def recv(self, buffer_size):
chunk = asyncore.dispatcher.recv(self, buffer_size)
if self.read_limit:
self.read_this_second += len(chunk)
self.throttle_read()
return chunk
def send(self, data):
num_sent = asyncore.dispatcher.send(self, data)
if self.write_limit:
self.written_this_second += num_sent
self.throttle_write()
return num_sent
def throttle_read(self):
if self.read_this_second >= self.read_limit:
self.read_this_second = 0
now = time.time()
sleepfor = self.r_timenext - now
if sleepfor > 0:
# we've passed bandwidth limits
self.r_sleep = True
def unthrottle():
self.r_sleep = False
self.delayed_r = self.call_later((sleepfor * 2),
unthrottle)
self.r_timenext = now + 1
def throttle_write(self):
if self.written_this_second >= self.write_limit:
self.written_this_second = 0
now = time.time()
sleepfor = self.w_timenext - now
if sleepfor > 0:
# we've passed bandwidth limits
self.w_sleep = True
def unthrottle():
self.w_sleep = False
self.delayed_w = self.call_later((sleepfor * 2),
unthrottle)
self.w_timenext = now + 1
def close(self):
if self.delayed_r and not self.delayed_r.cancelled:
self.delayed_r.cancel()
if self.delayed_w and not self.delayed_w.cancelled:
self.delayed_w.cancel()
asyncore.dispatcher.close(self)
--- /code snippet ---
I don't know if there's a better way to implement this "call_later"
feature.
Maybe someone experienced with Twisted could provide a better
approach.
I would ask someone using asyncore to review this since, IMHO, it
would
fill a very big gap.
I'm an habitual asyncore user and I miss such thing very much.
Ciao,
-- Giampaolo