Author: tmckay
Date: 2011-01-23 21:37:05 +0000 (Sun, 23 Jan 2011)
New Revision: 4483
Modified:
branches/scale_testing/wooly/python/wooly/wsgiserver/__init__.py
Log:
Queue throughput stuff
Modified: branches/scale_testing/wooly/python/wooly/wsgiserver/__init__.py
===================================================================
--- branches/scale_testing/wooly/python/wooly/wsgiserver/__init__.py 2011-01-23 19:36:05
UTC (rev 4482)
+++ branches/scale_testing/wooly/python/wooly/wsgiserver/__init__.py 2011-01-23 21:37:05
UTC (rev 4483)
@@ -1354,11 +1354,24 @@
self.ready = False
self.server = server
threading.Thread.__init__(self)
-
+
def run(self):
+ # put profile results in a subdirectory
+ ds = time.strftime("%m%d%y_%H%M%S",time.localtime())
+ p = os.path.join("./", \
+ "prof_"+
time.strftime("%m%d%y_%H%M%S",time.localtime()))
+
+ if not os.path.exists(p):
+ try:
+ os.makedirs(p)
+ except:
+ pass
+ p = os.path.join(p, self.getName())
+
def profWrapper():
try:
+ counter = 0
self.ready = True
while True:
conn = self.server.requests.get()
@@ -1367,27 +1380,62 @@
self.conn = conn
try:
+ #cProfile.runctx('conn.communicate()', globals(),
locals(), p+"_"+str(counter))
conn.communicate()
finally:
conn.close()
self.conn = None
+ counter += 1
except (KeyboardInterrupt, SystemExit), exc:
self.server.interrupt = exc
- # put profile results in a subdirectory
+ #cProfile.runctx('profWrapper()', globals(), locals(), p)
+ profWrapper()
+
+class TimerObject(object):
+ def __init__(self,name):
+ super(TimerObject, self).__init__()
+ self.tuples = []
+ self.total = 0
+ self.samples = 0
+ self.size_total = 0
+ self.name = name
+
ds = time.strftime("%m%d%y_%H%M%S",time.localtime())
- p = os.path.join("./", \
+ self.prof_path = os.path.join("./", \
"prof_"+
time.strftime("%m%d%y_%H%M%S",time.localtime()))
- if not os.path.exists(p):
+ if not os.path.exists(self.prof_path):
try:
- os.makedirs(p)
+ os.makedirs(self.prof_path)
except:
pass
+ self.fname = os.path.join(self.prof_path, name)
- p = os.path.join(p, self.getName())
- cProfile.runctx('profWrapper()', globals(), locals(), p)
+ def record(self, before, after, size):
+ duration = after - before
+ self.tuples.append((before, after, duration, size))
+ self.total += duration
+ self.size_total += size
+ self.samples += 1
+ def dump(self):
+ # write the number of samples
+ # write the average
+ # loop and write all the tuples
+ f = open(self.fname, "w+")
+ f.write("Timer: " + self.name+"\n")
+ f.write("Total number of ops: " + str(self.samples) + "\n")
+ f.write("Total duration: " + str(self.total) + "\n")
+ if self.total == 0:
+ return
+ f.write("Average time: " + str(self.total / self.samples) +
"\n")
+ f.write("Average size: " + str(self.size_total / self.samples) +
"\n")
+ f.write("\n")
+ for s in self.tuples:
+ f.write(time.ctime(s[0])+", ")
+ f.write(time.ctime(s[1])+", "+str(s[2])+"
"+str(s[3])+"\n")
+
class ThreadPool(object):
"""A Request Queue for the CherryPyWSGIServer which pools threads.
@@ -1401,8 +1449,27 @@
self.max = max
self._threads = []
self._queue = Queue.Queue()
- self.get = self._queue.get
-
+ #self.get = self._queue.get
+
+ self.put_timer = TimerObject("enqueue")
+ self.get_timer = TimerObject("dequeue")
+ self.last_put_call = time.time()
+ self.put_call_timer = TimerObject("put_delta")
+
+ def get(self):
+ #before = time.time()
+ #size = self._queue.qsize()
+ while True:
+ try:
+ res = self._queue.get()
+ break
+ except:
+ pass
+ after = time.time()
+ if res != None:
+ self.get_timer.record(res[1], after, res[2])
+ return res[0]
+
def start(self):
"""Start the pool of threads."""
for i in range(self.min):
@@ -1420,9 +1487,19 @@
idle = property(_get_idle, doc=_get_idle.__doc__)
def put(self, obj):
- self._queue.put(obj)
+ before = time.time()
+ size = self._queue.qsize()
+ self._queue.put((obj,before,size))
+
+ self.put_call_timer.record(self.last_put_call, before, size)
+ self.last_put_call = before
+ #after = time.time()
+ #self.put_timer.record(before, after, size)
if obj is _SHUTDOWNREQUEST:
return
+
+ def qsize(self):
+ return self._queue.qsize()
def grow(self, amount):
"""Spawn new worker threads (not above
self.max)."""
@@ -1484,8 +1561,12 @@
KeyboardInterrupt), exc1:
pass
+ self.get_timer.dump()
+ self.put_timer.dump()
+ self.put_call_timer.dump()
+
try:
import fcntl
except ImportError:
@@ -1629,77 +1710,85 @@
For UNIX sockets, supply the filename as a string.""")
+
+
def start(self):
- """Run the server forever."""
- # We don't have to trap KeyboardInterrupt or SystemExit here,
- # because cherrpy.server already does so, calling self.stop() for us.
- # If you're using this server with another framework, you should
- # trap those exceptions in whatever code block calls start().
- self._interrupt = None
-
- # Select the appropriate socket
- if isinstance(self.bind_addr, basestring):
- # AF_UNIX socket
+ def startWrapper():
+ """Run the server forever."""
+ # We don't have to trap KeyboardInterrupt or SystemExit here,
+ # because cherrpy.server already does so, calling self.stop() for us.
+ # If you're using this server with another framework, you should
+ # trap those exceptions in whatever code block calls start().
+ self._interrupt = None
- # So we can reuse the socket...
- try: os.unlink(self.bind_addr)
- except: pass
+ # Select the appropriate socket
+ if isinstance(self.bind_addr, basestring):
+ # AF_UNIX socket
- # So everyone can access the socket...
- try: os.chmod(self.bind_addr, 0777)
- except: pass
+ # So we can reuse the socket...
+ try: os.unlink(self.bind_addr)
+ except: pass
+
+ # So everyone can access the socket...
+ try: os.chmod(self.bind_addr, 0777)
+ except: pass
- info = [(socket.AF_UNIX, socket.SOCK_STREAM, 0, "",
self.bind_addr)]
- else:
- # AF_INET or AF_INET6 socket
- # Get the correct address family for our host (allows IPv6 addresses)
- host, port = self.bind_addr
- try:
- info = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
- socket.SOCK_STREAM, 0, socket.AI_PASSIVE)
- except socket.gaierror:
- # Probably a DNS issue. Assume IPv4.
- info = [(socket.AF_INET, socket.SOCK_STREAM, 0, "",
self.bind_addr)]
+ info = [(socket.AF_UNIX, socket.SOCK_STREAM, 0, "",
self.bind_addr)]
+ else:
+ # AF_INET or AF_INET6 socket
+ # Get the correct address family for our host (allows IPv6 addresses)
+ host, port = self.bind_addr
+ try:
+ info = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
+ socket.SOCK_STREAM, 0, socket.AI_PASSIVE)
+ except socket.gaierror:
+ # Probably a DNS issue. Assume IPv4.
+ info = [(socket.AF_INET, socket.SOCK_STREAM, 0, "",
self.bind_addr)]
+
+ self.socket = None
+ msg = "No socket could be created"
+ for res in info:
+ af, socktype, proto, canonname, sa = res
+ try:
+ self.bind(af, socktype, proto)
+ except socket.error, msg:
+ if self.socket:
+ self.socket.close()
+ self.socket = None
+ continue
+ break
+ if not self.socket:
+ raise socket.error(msg)
+
+ # Timeout so KeyboardInterrupt can be caught on Win32
+ self.socket.settimeout(1)
+ self.socket.listen(self.request_queue_size)
- self.socket = None
- msg = "No socket could be created"
- for res in info:
- af, socktype, proto, canonname, sa = res
- try:
- self.bind(af, socktype, proto)
- except socket.error, msg:
- if self.socket:
- self.socket.close()
- self.socket = None
- continue
- break
- if not self.socket:
- raise socket.error(msg)
+ # Create worker threads
+ self.requests.start()
- # Timeout so KeyboardInterrupt can be caught on Win32
- self.socket.settimeout(1)
- self.socket.listen(self.request_queue_size)
-
- # Create worker threads
- self.requests.start()
-
- self.ready = True
- while self.ready:
- self.tick()
- if self.interrupt:
- while self.interrupt is True:
- # Wait for self.stop() to complete. See _set_interrupt.
- time.sleep(0.1)
+ self.ready = True
+ while self.ready:
+ self.tick()
if self.interrupt:
- raise self.interrupt
+ while self.interrupt is True:
+ # Wait for self.stop() to complete. See _set_interrupt.
+ time.sleep(0.1)
+ if self.interrupt:
+ raise self.interrupt
- idle = self.requests.idle
+ idle = self.requests.idle
- if idle < 16:
- self.requests.grow(16)
- elif idle > 32:
- self.requests.shrink(2)
+ if idle < 16:
+ self.requests.grow(16)
+ elif idle > 32:
+ self.requests.shrink(2)
+
+ startWrapper()
+# p = os.path.join(self.prof_path, "Cherry Main")
+# cProfile.runctx('startWrapper()', globals(), locals(), p)
+
def bind(self, family, type, proto=0):
"""Create (or recreate) the actual socket
object."""
self.socket = socket.socket(family, type, proto)
@@ -1793,6 +1882,7 @@
conn = self.ConnectionClass(s, self.wsgi_app, environ, makefile)
self.requests.put(conn)
+
except socket.timeout:
# The only reason for the timeout in start() is so we can
# notice keyboard interrupts on Win32, which don't interrupt