12:03 p.m.
Author: justi9
Date: 2008-12-17 13:03:47 -0500 (Wed, 17 Dec 2008)
New Revision: 3016
Modified:
mgmt/trunk/mint/bin/mint-bench
mgmt/trunk/mint/python/mint/__init__.py
mgmt/trunk/mint/python/mint/tools.py
mgmt/trunk/mint/python/mint/update.py
Log:
Consolidate commit and rollback behaviors under methods.
Pass the update thread into update objects.
Isolate the update invocation from the run loop, so that switching to
a single-threaded mode for profiling is possible.
Shorten the mint-bench profile calibration step.
Add more counters to the update thread, for tracking deferred and
discarded updates.
Modified: mgmt/trunk/mint/bin/mint-bench
===================================================================
--- mgmt/trunk/mint/bin/mint-bench 2008-12-17 18:00:07 UTC (rev 3015)
+++ mgmt/trunk/mint/bin/mint-bench 2008-12-17 18:03:47 UTC (rev 3016)
@@ -20,8 +20,8 @@
biases = list()
- for i in range(5):
- bias = prof.calibrate(100000)
+ for i in range(4):
+ bias = prof.calibrate(20000)
biases.append(bias)
print i, bias
Modified: mgmt/trunk/mint/python/mint/__init__.py
===================================================================
--- mgmt/trunk/mint/python/mint/__init__.py 2008-12-17 18:00:07 UTC (rev 3015)
+++ mgmt/trunk/mint/python/mint/__init__.py 2008-12-17 18:03:47 UTC (rev 3016)
@@ -384,6 +384,8 @@
self.qmfSession = qmf.console.Session \
(self, manageConnections=True, rcvObjects=self.updateObjects)
+ self.updateThread.init()
+
def start(self):
self.updateThread.start()
Modified: mgmt/trunk/mint/python/mint/tools.py
===================================================================
--- mgmt/trunk/mint/python/mint/tools.py 2008-12-17 18:00:07 UTC (rev 3015)
+++ mgmt/trunk/mint/python/mint/tools.py 2008-12-17 18:03:47 UTC (rev 3016)
@@ -67,9 +67,10 @@
if "debug" in opts:
level = "debug"
+ file = sys.stderr
+ else:
+ file = opts.get("log-file", sys.stderr)
- file = opts.get("log-file", sys.stderr)
-
enable_logging("mint", level, file)
data = opts.get("data",
"postgresql://cumin@localhost/cumin")
@@ -157,22 +158,33 @@
enq_last = 0
deq_last = 0
- print "enqs", "\t", "deqs", "\t",
"depth"
+ head = "%10s %10s %10s %10s %10s" % \
+ ("enqs", "deqs", "depth",
"discard", "defer")
+ row = "%10i %10i %10i %10i %10i"
+ print head
+
while True:
sleep(1)
- enq = model.updateThread.enqueueCount
- deq = model.updateThread.dequeueCount
+ ut = model.updateThread
- enq_rate = enq - enq_last
- deq_rate = deq - deq_last
+ enq = ut.enqueueCount
+ deq = ut.dequeueCount
- print enq_rate, "\t", deq_rate, "\t", enq - deq
+ print row % (enq - enq_last,
+ deq - deq_last,
+ enq - deq,
+ ut.discardCount,
+ ut.deferCount)
enq_last = enq
deq_last = deq
finally:
+ #from threading import enumerate
+ #for item in enumerate():
+ # print item
+
for broker in added:
model.delBroker(broker)
Modified: mgmt/trunk/mint/python/mint/update.py
===================================================================
--- mgmt/trunk/mint/python/mint/update.py 2008-12-17 18:00:07 UTC (rev 3015)
+++ mgmt/trunk/mint/python/mint/update.py 2008-12-17 18:03:47 UTC (rev 3016)
@@ -25,8 +25,15 @@
self.enqueueCount = 0
self.dequeueCount = 0
+ self.discardCount = 0
+ self.deferCount = 0
self.commitThreshold = 100
+ self.conn = None
+
+ def init(self):
+ self.conn = self.model.dbConn.getConnection()
+
def enqueue(self, update):
try:
self.updates.put(update)
@@ -36,12 +43,12 @@
log.exception("Queue is full")
if self.updates.qsize() > 1000:
- # This is an attempt to yield
+ # This is an attempt to yield from the enqueueing thread (this
+ # method's caller) to the update thread
+
sleep(0.1)
def run(self):
- conn = self.model.dbConn.getConnection()
-
while True:
try:
update = self.updates.get(True, 1)
@@ -59,35 +66,37 @@
log.debug("Queue is empty")
continue
- try:
- update.process(conn)
+ self.process_update(update)
+
+ def process_update(self, update):
+ try:
+ update.process(self)
- if self.dequeueCount % self.commitThreshold == 0 \
- or self.updates.qsize() == 0:
- # commit only every "commitThreshold" updates, or whenever
- # the update queue is empty
+ if self.dequeueCount % self.commitThreshold == 0 \
+ or self.updates.qsize() == 0:
+ # commit only every "commitThreshold" updates, or whenever
+ # the update queue is empty
- if profile:
- start = clock()
+ self.commit()
+ except:
+ log.exception("Update failed")
- conn.commit()
+ self.rollback()
- for broker in self.model.mintBrokersByQmfBroker.values():
- broker.objectDatabaseIds.commit()
+ def cursor(self):
+ return self.conn.cursor()
- profile.commitTime += clock() - start
- else:
- conn.commit()
+ def commit(self):
+ self.conn.commit()
- for broker in self.model.mintBrokersByQmfBroker.values():
- broker.objectDatabaseIds.commit()
- except:
- conn.rollback()
+ for broker in self.model.mintBrokersByQmfBroker.values():
+ broker.objectDatabaseIds.commit()
- for broker in self.model.mintBrokersByQmfBroker.values():
- broker.objectDatabaseIds.rollback()
+ def rollback(self):
+ self.conn.rollback()
- log.exception("Update failed")
+ for broker in self.model.mintBrokersByQmfBroker.values():
+ broker.objectDatabaseIds.rollback()
class ReferenceException(Exception):
def __init__(self, sought):
@@ -120,7 +129,7 @@
except KeyError:
raise ReferenceException(name)
- def process(self, conn):
+ def process(self, thread):
pass
def processAttributes(self, attrs, cls):
@@ -171,11 +180,13 @@
results[name] = t
class PropertyUpdate(ModelUpdate):
- def process(self, conn):
+ def process(self, thread):
try:
cls = self.getClass()
except ReferenceException, e:
log.info("Referenced class %r not found", e.sought)
+
+ thread.discardCount += 1
return
oid = self.object.getObjectId()
@@ -191,6 +202,7 @@
except KeyError:
self.broker.orphans[oid] = list((self,))
+ thread.deferCount += 1
return
timestamps = self.object.getTimestamps()
@@ -208,7 +220,7 @@
attrs["qmfClassKey"] = str(self.object.getClassKey())
attrs["qmfBrokerId"] = str(self.broker.qmfBroker.getBrokerId())
- cursor = conn.cursor()
+ cursor = thread.cursor()
# Cases:
#
@@ -271,7 +283,7 @@
len(orphans))
for orphan in orphans:
- self.model.updateThread.enqueue(orphan)
+ thread.enqueue(orphan)
except KeyError:
pass
@@ -295,7 +307,7 @@
self.broker.databaseId = id
class StatisticUpdate(ModelUpdate):
- def process(self, conn):
+ def process(self, thread):
try:
cls = self.getClass()
except ReferenceException, e:
@@ -308,7 +320,7 @@
id = self.broker.objectDatabaseIds.get(oid)
if id is None:
- # Just drop it; we'll get more stats later
+ thread.discardCount += 1
return
timestamps = self.object.getTimestamps()
@@ -318,18 +330,20 @@
if t < tnow - timedelta(seconds=30):
log.debug("Update is %i seconds old; skipping it", (tnow -t).seconds)
+
+ thread.discardCount += 1
return
try:
attrs = self.processAttributes(self.object.getStatistics(), statsCls)
except ReferenceException:
- # Drop it
+ thread.discardCount += 1
return
attrs["qmfUpdateTime"] = t > tnow and tnow or t
attrs["%s_id" % cls.sqlmeta.table] = id
- cursor = conn.cursor()
+ cursor = thread.cursor()
op = SqlInsert(statsCls, attrs)
op.execute(cursor, attrs)
@@ -342,7 +356,7 @@
self.seq = seq
- def process(self, conn):
+ def process(self, thread):
self.model.lock()
try:
@@ -355,19 +369,19 @@
def __init__(self, model):
super(DBExpireUpdate, self).__init__(model, None, None)
- def process(self, conn):
- cursor = conn.cursor()
+ def process(self, thread):
+ cursor = thread.cursor()
attrs = self.model.dbExpireThread.attrs
total = 0
- conn.commit()
+ thread.commit()
for op in self.model.dbExpireThread.ops:
log.debug("Running expire op %s", op)
count = op.execute(cursor, attrs)
- conn.commit()
+ thread.commit()
log.debug("%i records expired", count)