April 2014 - tor-commits - lists.torproject.org

[bridgedb/develop] Add unit tests for bridgedb.Storage
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit 766cc675c52488e658a0884f0e0b5c8b044731b2 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Mar 21 02:48:00 2014 +0000 Add unit tests for bridgedb.Storage --- lib/bridgedb/test/test_Storage.py | 56 +++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) diff --git a/lib/bridgedb/test/test_Storage.py b/lib/bridgedb/test/test_Storage.py new file mode 100644 index 0000000..8b0affd --- /dev/null +++ b/lib/bridgedb/test/test_Storage.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python +"""Unittests for the :mod:`bridgedb.Storage` module.""" + +from twisted.python import log +from twisted.trial import unittest +import bridgedb.Storage as Storage +from twisted.internet import reactor +from twisted.internet.threads import deferToThread +import os +import threading +from time import sleep + +class DatabaseTest(unittest.TestCase): + def setUp(self): + self.dbfname = 'test-bridgedb.sqlite' + Storage.setDBFilename(self.dbfname) + + def tearDown(self): + if os.path.isfile(self.dbfname): + os.unlink(self.dbfname) + Storage.clearGlobalDB() + + def _runAndDie(self, timeout, func): + with func(): + sleep(timeout) + + def _cb_assertTrue(self, result): + self.assertTrue(result) + + def _cb_assertFalse(self, result): + self.assertFalse(result) + + def _eb_Failure(self, failure): + self.fail(failure) + + def test_getDB_FalseWhenLocked(self): + Storage._LOCK = threading.Lock() + Storage._LOCK.acquire() + self.assertFalse(Storage._LOCK.acquire(False)) + + def test_getDB_AcquireLock(self): + Storage.initializeDBLock() + with Storage.getDB() as db: + self.assertIsInstance(db, Storage.Database) + self.assertTrue(Storage.dbIsLocked()) + self.assertEqual(db, Storage._OPENED_DB) + + def test_getDB_ConcurrencyLock(self): + timeout = 1 + d1 = deferToThread(self._runAndDie, timeout, Storage.getDB) + d1.addCallback(self._cb_assertFalse) + d1.addErrback(self._eb_Failure) + d2 = deferToThread(Storage.getDB, False) + d2.addCallback(self._cb_assertFalse) + d2.addErrback(self._eb_Failure) + d2.addCallback(self._cb_assertTrue, Storage.getDB(False))

1 0

[bridgedb/develop] Add tests for SIGHUP and SIGUSR1 signals
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit 9c4baf91858e51dc0dabb8475c4582260201ad72 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Feb 7 03:15:03 2014 +0000 Add tests for SIGHUP and SIGUSR1 signals --- lib/bridgedb/test/test_bridgedb.py | 32 +++++++++++++++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/lib/bridgedb/test/test_bridgedb.py b/lib/bridgedb/test/test_bridgedb.py index 5dc890d..1f4470b 100644 --- a/lib/bridgedb/test/test_bridgedb.py +++ b/lib/bridgedb/test/test_bridgedb.py @@ -89,8 +89,38 @@ class BridgeDBCliTest(unittest.TestCase): self.doCopyFile(eindesc, eidesc, 'duplicated cached-extrainfo(.new)') print("Running `bridgedb' to test server startups...") - bridgedbProc = Popen([bridgedbScript, '-r', runDir]) + # Sorry Windows users + devnull = open('/dev/null', 'w') + bridgedbProc = Popen([bridgedbScript, '-r', runDir], stdout=devnull) time.sleep(30) + assignments = pjoin(runDir, 'assignments.log') + self.assertTrue(os.path.isfile(assignments)) + os.unlink(assignments) + bridgedbProc.send_signal(signal.SIGHUP) + time.sleep(5) + try: + self.assertTrue(os.path.isfile(assignments)) + except self.failureException as e: + bridgedbProc.send_signal(signal.SIGKILL) + bridgedbProcCode = bridgedbProc.wait() + print("`bridgedb' exited with status code %d" % int(bridgedbProcCode)) + raise e + bridgedbProc.send_signal(signal.SIGUSR1) + time.sleep(5) + buckets = [['email', False], ['https', False], ['unallocated', False]] + for rundirfile in os.listdir(runDir): + for bucket in buckets: + if rundirfile.startswith(bucket[0]): + bucket[1] = True + break + for bucket in buckets: + try: + self.assertTrue(bucket[1], "%s bucket was not dumped!" % bucket[0]) + except self.failureException as e: + bridgedbProc.send_signal(signal.SIGKILL) + bridgedbProcCode = bridgedbProc.wait() + print("`bridgedb' exited with status code %d" % int(bridgedbProcCode)) + raise e bridgedbProc.send_signal(signal.SIGINT) bridgedbProcCode = bridgedbProc.wait() print("`bridgedb' exited with status code %d" % int(bridgedbProcCode))

1 0

[bridgedb/develop] Use the new idiom throughout code
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit 63f046df60767bae1a2ec31e67346c4ebb6769f8 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Mar 21 03:10:43 2014 +0000 Use the new idiom throughout code --- lib/bridgedb/Bridges.py | 66 +++++++-------- lib/bridgedb/Bucket.py | 104 +++++++++++------------ lib/bridgedb/Dist.py | 115 ++++++++++++------------- lib/bridgedb/Stability.py | 204 ++++++++++++++++++++++----------------------- lib/bridgedb/Tests.py | 4 +- 5 files changed, 247 insertions(+), 246 deletions(-) diff --git a/lib/bridgedb/Bridges.py b/lib/bridgedb/Bridges.py index 2fb784a..bfe1316 100644 --- a/lib/bridgedb/Bridges.py +++ b/lib/bridgedb/Bridges.py @@ -308,38 +308,38 @@ class Bridge(object): A bridge is 'familiar' if 1/8 of all active bridges have appeared more recently than it, or if it has been around for a Weighted Time of 8 days. """ - db = bridgedb.Storage.getDB() - return db.getBridgeHistory(self.fingerprint).familiar + with bridgedb.Storage.getDB() as db: + return db.getBridgeHistory(self.fingerprint).familiar @property def wfu(self): """Weighted Fractional Uptime""" - db = bridgedb.Storage.getDB() - return db.getBridgeHistory(self.fingerprint).weightedFractionalUptime + with bridgedb.Storage.getDB() as db: + return db.getBridgeHistory(self.fingerprint).weightedFractionalUptime @property def weightedTime(self): """Weighted Time""" - db = bridgedb.Storage.getDB() - return db.getBridgeHistory(self.fingerprint).weightedTime + with bridgedb.Storage.getDB() as db: + return db.getBridgeHistory(self.fingerprint).weightedTime @property def wmtbac(self): """Weighted Mean Time Between Address Change""" - db = bridgedb.Storage.getDB() - return db.getBridgeHistory(self.fingerprint).wmtbac + with bridgedb.Storage.getDB() as db: + return db.getBridgeHistory(self.fingerprint).wmtbac @property def tosa(self): """the Time On Same Address (TOSA)""" - db = bridgedb.Storage.getDB() - return db.getBridgeHistory(self.fingerprint).tosa + with bridgedb.Storage.getDB() as db: + return db.getBridgeHistory(self.fingerprint).tosa @property def weightedUptime(self): """Weighted Uptime""" - db = bridgedb.Storage.getDB() - return db.getBridgeHistory(self.fingerprint).weightedUptime + with bridgedb.Storage.getDB() as db: + return db.getBridgeHistory(self.fingerprint).weightedUptime def getDescriptorDigests(desc): """Return the SHA-1 hash hexdigests of all descriptor descs @@ -1122,19 +1122,19 @@ class UnallocatedHolder(BridgeHolder): self.fingerprints = [] def dumpAssignments(self, f, description=""): - db = bridgedb.Storage.getDB() - allBridges = db.getAllBridges() - for bridge in allBridges: - if bridge.hex_key not in self.fingerprints: - continue - dist = bridge.distributor - desc = [ description ] - if dist.startswith(bridgedb.Bucket.PSEUDO_DISTRI_PREFIX): - dist = dist.replace(bridgedb.Bucket.PSEUDO_DISTRI_PREFIX, "") - desc.append("bucket=%s" % dist) - elif dist != "unallocated": - continue - f.write("%s %s\n" % (bridge.hex_key, " ".join(desc).strip())) + with bridgedb.Storage.getDB() as db: + allBridges = db.getAllBridges() + for bridge in allBridges: + if bridge.hex_key not in self.fingerprints: + continue + dist = bridge.distributor + desc = [ description ] + if dist.startswith(bridgedb.Bucket.PSEUDO_DISTRI_PREFIX): + dist = dist.replace(bridgedb.Bucket.PSEUDO_DISTRI_PREFIX, "") + desc.append("bucket=%s" % dist) + elif dist != "unallocated": + continue + f.write("%s %s\n" % (bridge.hex_key, " ".join(desc).strip())) class BridgeSplitter(BridgeHolder): """A BridgeHolder that splits incoming bridges up based on an hmac, @@ -1186,7 +1186,6 @@ class BridgeSplitter(BridgeHolder): def insert(self, bridge): assert self.rings - db = bridgedb.Storage.getDB() for s in self.statsHolders: s.insert(bridge) @@ -1205,16 +1204,17 @@ class BridgeSplitter(BridgeHolder): validRings = self.rings + self.pseudoRings - ringname = db.insertBridgeAndGetRing(bridge, ringname, time.time(), + with bridgedb.Storage.getDB() as db: + ringname = db.insertBridgeAndGetRing(bridge, ringname, time.time(), validRings) - db.commit() + db.commit() - # Pseudo distributors are always held in the "unallocated" ring - if ringname in self.pseudoRings: - ringname = "unallocated" + # Pseudo distributors are always held in the "unallocated" ring + if ringname in self.pseudoRings: + ringname = "unallocated" - ring = self.ringsByName.get(ringname) - ring.insert(bridge) + ring = self.ringsByName.get(ringname) + ring.insert(bridge) def dumpAssignments(self, f, description=""): for name,ring in self.ringsByName.iteritems(): diff --git a/lib/bridgedb/Bucket.py b/lib/bridgedb/Bucket.py index 0038146..416c1ea 100644 --- a/lib/bridgedb/Bucket.py +++ b/lib/bridgedb/Bucket.py @@ -109,22 +109,18 @@ class BucketManager: self.unallocatedList = [] self.unallocated_available = False self.distributor_prefix = PSEUDO_DISTRI_PREFIX - self.db = bridgedb.Storage.Database(self.cfg.DB_FILE+".sqlite", - self.cfg.DB_FILE) - - def __del__(self): - self.db.close() def addToUnallocatedList(self, hex_key): """Add a bridge by hex_key into the unallocated pool """ - try: - self.db.updateDistributorForHexKey("unallocated", hex_key) - except: - self.db.rollback() - raise - else: - self.db.commit() + with bridgedb.Storage.getDB() as db: + try: + db.updateDistributorForHexKey("unallocated", hex_key) + except: + db.rollback() + raise + else: + db.commit() self.unallocatedList.append(hex_key) self.unallocated_available = True @@ -144,17 +140,18 @@ class BucketManager: # Mark pseudo-allocators in the database as such allocator_name = bucket.name #print "KEY: %d NAME: %s" % (hex_key, allocator_name) - try: - self.db.updateDistributorForHexKey(allocator_name, hex_key) - except: - self.db.rollback() - # Ok, this seems useless, but for consistancy's sake, we'll - # re-assign the bridge from this missed db update attempt to the - # unallocated list. Remember? We pop()'d it before. - self.addToUnallocatedList(hex_key) - raise - else: - self.db.commit() + with bridgedb.Storage.getDB() as db: + try: + db.updateDistributorForHexKey(allocator_name, hex_key) + except: + db.rollback() + # Ok, this seems useless, but for consistancy's sake, we'll + # re-assign the bridge from this missed db update attempt to the + # unallocated list. Remember? We pop()'d it before. + self.addToUnallocatedList(hex_key) + raise + else: + db.commit() bucket.allocated += 1 if len(self.unallocatedList) < 1: self.unallocated_available = False @@ -171,7 +168,8 @@ class BucketManager: self.bucketList.append(d) # Loop through all bridges and sort out distributors - allBridges = self.db.getAllBridges() + with bridgedb.Storage.getDB() as db: + allBridges = db.getAllBridges() for bridge in allBridges: if bridge.distributor == "unallocated": self.addToUnallocatedList(bridge.hex_key) @@ -215,38 +213,40 @@ class BucketManager: logging.debug("Dumping bridge assignments to file: %r" % filename) # get the bridge histories and sort by Time On Same Address bridgeHistories = [] - for b in bridges: - bh = self.db.getBridgeHistory(b.hex_key) - if bh: bridgeHistories.append(bh) - bridgeHistories.sort(lambda x,y: cmp(x.weightedFractionalUptime, - y.weightedFractionalUptime)) - - # for a bridge, get the list of countries it might not work in - blocklist = dict() - if getattr(self.cfg, "COUNTRY_BLOCK_FILE", None) is not None: - f = open(self.cfg.COUNTRY_BLOCK_FILE, 'r') - for ID,address,portlist,countries in bridgedb.Bridges.parseCountryBlockFile(f): - blocklist[toHex(ID)] = countries - f.close() - - try: - f = open(filename, 'w') - for bh in bridgeHistories: - days = bh.tosa / long(60*60*24) - line = "%s:%s\t(%d %s)" % \ - (bh.ip, bh.port, days, _("""days at this address""")) - if str(bh.fingerprint) in blocklist.keys(): - line = line + "\t%s: (%s)" % (_("""(Might be blocked)"""), - ",".join(blocklist[bh.fingerprint]),) - f.write(line + '\n') - f.close() - except IOError: - print "I/O error: %s" % filename + with bridgedb.Storage.getDB() as db: + for b in bridges: + bh = db.getBridgeHistory(b.hex_key) + if bh: bridgeHistories.append(bh) + bridgeHistories.sort(lambda x,y: cmp(x.weightedFractionalUptime, + y.weightedFractionalUptime)) + + # for a bridge, get the list of countries it might not work in + blocklist = dict() + if getattr(self.cfg, "COUNTRY_BLOCK_FILE", None) is not None: + f = open(self.cfg.COUNTRY_BLOCK_FILE, 'r') + for ID,address,portlist,countries in bridgedb.Bridges.parseCountryBlockFile(f): + blocklist[toHex(ID)] = countries + f.close() + + try: + f = open(filename, 'w') + for bh in bridgeHistories: + days = bh.tosa / long(60*60*24) + line = "%s:%s\t(%d %s)" % \ + (bh.ip, bh.port, days, _("""days at this address""")) + if str(bh.fingerprint) in blocklist.keys(): + line = line + "\t%s: (%s)" % (_("""(Might be blocked)"""), + ",".join(blocklist[bh.fingerprint]),) + f.write(line + '\n') + f.close() + except IOError: + print "I/O error: %s" % filename def dumpBridges(self): """Dump all known file distributors to files, sort by distributor """ - allBridges = self.db.getAllBridges() + with bridgedb.Storage.getDB() as db: + allBridges = db.getAllBridges() bridgeDict = {} # Sort returned bridges by distributor for bridge in allBridges: diff --git a/lib/bridgedb/Dist.py b/lib/bridgedb/Dist.py index dddc84b..3bfb2d8 100644 --- a/lib/bridgedb/Dist.py +++ b/lib/bridgedb/Dist.py @@ -474,58 +474,59 @@ class EmailBasedDistributor(Distributor): if emailaddress is None: return [] #XXXX raise an exception. - db = bridgedb.Storage.getDB() - wasWarned = db.getWarnedEmail(emailaddress) - lastSaw = db.getEmailTime(emailaddress) - - logging.info("Attempting to return for %d bridges for %s..." - % (N, Util.logSafely(emailaddress))) - - if lastSaw is not None and lastSaw + MAX_EMAIL_RATE >= now: - logging.info("Client %s sent duplicate request within %d seconds." - % (Util.logSafely(emailaddress), MAX_EMAIL_RATE)) - if wasWarned: - logging.info( - "Client was already warned about duplicate requests.") - raise IgnoreEmail("Client was warned", - Util.logSafely(emailaddress)) + with bridgedb.Storage.getDB() as db: + wasWarned = db.getWarnedEmail(emailaddress) + lastSaw = db.getEmailTime(emailaddress) + + logging.info("Attempting to return for %d bridges for %s..." + % (N, Util.logSafely(emailaddress))) + + if lastSaw is not None and lastSaw + MAX_EMAIL_RATE >= now: + logging.info("Client %s sent duplicate request within %d seconds." + % (Util.logSafely(emailaddress), MAX_EMAIL_RATE)) + if wasWarned: + logging.info( + "Client was already warned about duplicate requests.") + raise IgnoreEmail("Client was warned", + Util.logSafely(emailaddress)) + else: + logging.info("Sending duplicate request warning to %s..." + % Util.logSafely(emailaddress)) + db.setWarnedEmail(emailaddress, True, now) + db.commit() + + raise TooSoonEmail("Too many emails; wait till later", emailaddress) + + # warning period is over + elif wasWarned: + db.setWarnedEmail(emailaddress, False) + + pos = self.emailHmac("<%s>%s" % (epoch, emailaddress)) + + ring = None + ruleset = frozenset(bridgeFilterRules) + if ruleset in self.splitter.filterRings.keys(): + logging.debug("Cache hit %s" % ruleset) + _, ring = self.splitter.filterRings[ruleset] else: - logging.info("Sending duplicate request warning to %s..." - % Util.logSafely(emailaddress)) - db.setWarnedEmail(emailaddress, True, now) - db.commit() - - raise TooSoonEmail("Too many emails; wait till later", emailaddress) - - # warning period is over - elif wasWarned: - db.setWarnedEmail(emailaddress, False) - - pos = self.emailHmac("<%s>%s" % (epoch, emailaddress)) - - ring = None - ruleset = frozenset(bridgeFilterRules) - if ruleset in self.splitter.filterRings.keys(): - logging.debug("Cache hit %s" % ruleset) - _, ring = self.splitter.filterRings[ruleset] - else: - # cache miss, add new ring - logging.debug("Cache miss %s" % ruleset) + # cache miss, add new ring + logging.debug("Cache miss %s" % ruleset) - # add new ring - key1 = getHMAC(self.splitter.key, "Order-Bridges-In-Ring") - ring = bridgedb.Bridges.BridgeRing(key1, self.answerParameters) - # debug log: cache miss - self.splitter.addRing(ring, ruleset, - filterBridgesByRules(bridgeFilterRules), - populate_from=self.splitter.bridges) + # add new ring + key1 = getHMAC(self.splitter.key, + "Order-Bridges-In-Ring") + ring = bridgedb.Bridges.BridgeRing(key1, self.answerParameters) + # debug log: cache miss + self.splitter.addRing(ring, ruleset, + filterBridgesByRules(bridgeFilterRules), + populate_from=self.splitter.bridges) - numBridgesToReturn = getNumBridgesPerAnswer(ring, - max_bridges_per_answer=N) - result = ring.getBridges(pos, numBridgesToReturn) + numBridgesToReturn = getNumBridgesPerAnswer(ring, + max_bridges_per_answer=N) + result = ring.getBridges(pos, numBridgesToReturn) - db.setEmailTime(emailaddress, now) - db.commit() + db.setEmailTime(emailaddress, now) + db.commit() return result @@ -533,15 +534,15 @@ class EmailBasedDistributor(Distributor): return len(self.splitter) def cleanDatabase(self): - db = bridgedb.Storage.getDB() - try: - db.cleanEmailedBridges(time.time()-MAX_EMAIL_RATE) - db.cleanWarnedEmails(time.time()-MAX_EMAIL_RATE) - except: - db.rollback() - raise - else: - db.commit() + with bridgedb.Storage.getDB() as db: + try: + db.cleanEmailedBridges(time.time()-MAX_EMAIL_RATE) + db.cleanWarnedEmails(time.time()-MAX_EMAIL_RATE) + except: + db.rollback() + raise + else: + db.commit() def dumpAssignments(self, f, description=""): self.splitter.dumpAssignments(f, description) diff --git a/lib/bridgedb/Stability.py b/lib/bridgedb/Stability.py index 5cd7e8e..4c3777d 100644 --- a/lib/bridgedb/Stability.py +++ b/lib/bridgedb/Stability.py @@ -111,14 +111,14 @@ class BridgeHistory(object): # return True if self.weightedTime is greater than the weightedTime # of the > bottom 1/8 all bridges, sorted by weightedTime - db = bridgedb.Storage.getDB() - allWeightedTimes = [ bh.weightedTime for bh in db.getAllBridgeHistory()] - numBridges = len(allWeightedTimes) - logging.debug("Got %d weightedTimes", numBridges) - allWeightedTimes.sort() - if self.weightedTime >= allWeightedTimes[numBridges/8]: - return True - return False + with bridgedb.Storage.getDB() as db: + allWeightedTimes = [ bh.weightedTime for bh in db.getAllBridgeHistory()] + numBridges = len(allWeightedTimes) + logging.debug("Got %d weightedTimes", numBridges) + allWeightedTimes.sort() + if self.weightedTime >= allWeightedTimes[numBridges/8]: + return True + return False @property def wmtbac(self): @@ -134,104 +134,104 @@ class BridgeHistory(object): return totalRunlength / totalWeights def addOrUpdateBridgeHistory(bridge, timestamp): - db = bridgedb.Storage.getDB() - bhe = db.getBridgeHistory(bridge.fingerprint) - if not bhe: - # This is the first status, assume 60 minutes. - secondsSinceLastStatusPublication = long(60*60) - lastSeenWithDifferentAddressAndPort = timestamp * long(1000) - lastSeenWithThisAddressAndPort = timestamp * long(1000) - - bhe = BridgeHistory( - bridge.fingerprint, bridge.ip, bridge.orport, - 0,#weightedUptime - 0,#weightedTime - 0,#weightedRunLength - 0,# totalRunWeights - lastSeenWithDifferentAddressAndPort, # first timestamnp - lastSeenWithThisAddressAndPort, - 0,#lastDiscountedHistoryValues, - 0,#lastUpdatedWeightedTime - ) - # first time we have seen this descriptor - db.updateIntoBridgeHistory(bhe) - # Calculate the seconds since the last parsed status. If this is - # the first status or we haven't seen a status for more than 60 - # minutes, assume 60 minutes. - statusPublicationMillis = long(timestamp * 1000) - if (statusPublicationMillis - bhe.lastSeenWithThisAddressAndPort) > 60*60*1000: - secondsSinceLastStatusPublication = long(60*60) - logging.debug("Capping secondsSinceLastStatusPublication to 1 hour") - # otherwise, roll with it - else: - secondsSinceLastStatusPublication = \ - (statusPublicationMillis - bhe.lastSeenWithThisAddressAndPort)/1000 - if secondsSinceLastStatusPublication <= 0 and bhe.weightedTime > 0: - # old descriptor, bail - logging.warn("Received old descriptor for bridge %s with timestamp %d", - bhe.fingerprint, statusPublicationMillis/1000) - return bhe + with bridgedb.Storage.getDB() as db: + bhe = db.getBridgeHistory(bridge.fingerprint) + if not bhe: + # This is the first status, assume 60 minutes. + secondsSinceLastStatusPublication = long(60*60) + lastSeenWithDifferentAddressAndPort = timestamp * long(1000) + lastSeenWithThisAddressAndPort = timestamp * long(1000) - # iterate over all known bridges and apply weighting factor - discountAndPruneBridgeHistories(statusPublicationMillis) + bhe = BridgeHistory( + bridge.fingerprint, bridge.ip, bridge.orport, + 0,#weightedUptime + 0,#weightedTime + 0,#weightedRunLength + 0,# totalRunWeights + lastSeenWithDifferentAddressAndPort, # first timestamnp + lastSeenWithThisAddressAndPort, + 0,#lastDiscountedHistoryValues, + 0,#lastUpdatedWeightedTime + ) + # first time we have seen this descriptor + db.updateIntoBridgeHistory(bhe) + # Calculate the seconds since the last parsed status. If this is + # the first status or we haven't seen a status for more than 60 + # minutes, assume 60 minutes. + statusPublicationMillis = long(timestamp * 1000) + if (statusPublicationMillis - bhe.lastSeenWithThisAddressAndPort) > 60*60*1000: + secondsSinceLastStatusPublication = long(60*60) + logging.debug("Capping secondsSinceLastStatusPublication to 1 hour") + # otherwise, roll with it + else: + secondsSinceLastStatusPublication = \ + (statusPublicationMillis - bhe.lastSeenWithThisAddressAndPort)/1000 + if secondsSinceLastStatusPublication <= 0 and bhe.weightedTime > 0: + # old descriptor, bail + logging.warn("Received old descriptor for bridge %s with timestamp %d", + bhe.fingerprint, statusPublicationMillis/1000) + return bhe - # Update the weighted times of bridges - updateWeightedTime(statusPublicationMillis) - - # For Running Bridges only: - # compare the stored history against the descriptor and see if the - # bridge has changed its address or port - bhe = db.getBridgeHistory(bridge.fingerprint) - - if not bridge.running: - logging.info("%s is not running" % bridge.fingerprint) - return bhe - - # Parse the descriptor and see if the address or port changed - # If so, store the weighted run time - if bridge.orport != bhe.port or bridge.ip != bhe.ip: - bhe.totalRunWeights += 1.0; - bhe.weightedRunLength += bhe.tosa - bhe.lastSeenWithDifferentAddressAndPort =\ - bhe.lastSeenWithThisAddressAndPort - - # Regardless of whether the bridge is new, kept or changed - # its address and port, raise its WFU times and note its - # current address and port, and that we saw it using them. - bhe.weightedUptime += secondsSinceLastStatusPublication - bhe.lastSeenWithThisAddressAndPort = statusPublicationMillis - bhe.ip = str(bridge.ip) - bhe.port = bridge.orport - return db.updateIntoBridgeHistory(bhe) + # iterate over all known bridges and apply weighting factor + discountAndPruneBridgeHistories(statusPublicationMillis) + + # Update the weighted times of bridges + updateWeightedTime(statusPublicationMillis) + + # For Running Bridges only: + # compare the stored history against the descriptor and see if the + # bridge has changed its address or port + bhe = db.getBridgeHistory(bridge.fingerprint) + + if not bridge.running: + logging.info("%s is not running" % bridge.fingerprint) + return bhe + + # Parse the descriptor and see if the address or port changed + # If so, store the weighted run time + if bridge.orport != bhe.port or bridge.ip != bhe.ip: + bhe.totalRunWeights += 1.0; + bhe.weightedRunLength += bhe.tosa + bhe.lastSeenWithDifferentAddressAndPort =\ + bhe.lastSeenWithThisAddressAndPort + + # Regardless of whether the bridge is new, kept or changed + # its address and port, raise its WFU times and note its + # current address and port, and that we saw it using them. + bhe.weightedUptime += secondsSinceLastStatusPublication + bhe.lastSeenWithThisAddressAndPort = statusPublicationMillis + bhe.ip = str(bridge.ip) + bhe.port = bridge.orport + return db.updateIntoBridgeHistory(bhe) def discountAndPruneBridgeHistories(discountUntilMillis): - db = bridgedb.Storage.getDB() - bhToRemove = [] - bhToUpdate = [] - - for bh in db.getAllBridgeHistory(): - # discount previous values by factor of 0.95 every 12 hours - bh.discountWeightedFractionalUptimeAndWeightedTime(discountUntilMillis) - # give the thing at least 24 hours before pruning it - if bh.weightedFractionalUptime < 1 and bh.weightedTime > 60*60*24: - logging.debug("Removing bridge from history: %s" % bh.fingerprint) - bhToRemove.append(bh.fingerprint) - else: - bhToUpdate.append(bh) - - for k in bhToUpdate: db.updateIntoBridgeHistory(k) - for k in bhToRemove: db.delBridgeHistory(k) + with bridgedb.Storage.getDB() as db: + bhToRemove = [] + bhToUpdate = [] + + for bh in db.getAllBridgeHistory(): + # discount previous values by factor of 0.95 every 12 hours + bh.discountWeightedFractionalUptimeAndWeightedTime(discountUntilMillis) + # give the thing at least 24 hours before pruning it + if bh.weightedFractionalUptime < 1 and bh.weightedTime > 60*60*24: + logging.debug("Removing bridge from history: %s" % bh.fingerprint) + bhToRemove.append(bh.fingerprint) + else: + bhToUpdate.append(bh) + + for k in bhToUpdate: db.updateIntoBridgeHistory(k) + for k in bhToRemove: db.delBridgeHistory(k) def updateWeightedTime(statusPublicationMillis): bhToUpdate = [] - db = bridgedb.Storage.getDB() - for bh in db.getBridgesLastUpdatedBefore(statusPublicationMillis): - interval = (statusPublicationMillis - bh.lastUpdatedWeightedTime)/1000 - if interval > 0: - bh.weightedTime += min(3600,interval) # cap to 1hr - bh.lastUpdatedWeightedTime = statusPublicationMillis - #db.updateIntoBridgeHistory(bh) - bhToUpdate.append(bh) - - for bh in bhToUpdate: - db.updateIntoBridgeHistory(bh) + with bridgedb.Storage.getDB() as db: + for bh in db.getBridgesLastUpdatedBefore(statusPublicationMillis): + interval = (statusPublicationMillis - bh.lastUpdatedWeightedTime)/1000 + if interval > 0: + bh.weightedTime += min(3600,interval) # cap to 1hr + bh.lastUpdatedWeightedTime = statusPublicationMillis + #db.updateIntoBridgeHistory(bh) + bhToUpdate.append(bh) + + for bh in bhToUpdate: + db.updateIntoBridgeHistory(bh) diff --git a/lib/bridgedb/Tests.py b/lib/bridgedb/Tests.py index 378f1ae..72dfe5e 100644 --- a/lib/bridgedb/Tests.py +++ b/lib/bridgedb/Tests.py @@ -206,7 +206,7 @@ class EmailBridgeDistTests(unittest.TestCase): def setUp(self): self.fd, self.fname = tempfile.mkstemp() self.db = bridgedb.Storage.Database(self.fname) - bridgedb.Storage.setGlobalDB(self.db) + bridgedb.Storage.setDB(self.db) self.cur = self.db._conn.cursor() def tearDown(self): @@ -672,7 +672,7 @@ class BridgeStabilityTests(unittest.TestCase): def setUp(self): self.fd, self.fname = tempfile.mkstemp() self.db = bridgedb.Storage.Database(self.fname) - bridgedb.Storage.setGlobalDB(self.db) + bridgedb.Storage.setDB(self.db) self.cur = self.db._conn.cursor() def tearDown(self):

1 0

[bridgedb/develop] Reparse descriptors in background thread on signal
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit bd652aa560cfaf15b9b37979aba73e9f8f548b4e Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Feb 7 01:16:34 2014 +0000 Reparse descriptors in background thread on signal Create temporary distributors and fill them with bridges then overwrite the old distributors' rings with the rings from the temporary distributors. Factor out distributor creation code. Bug fix 5232 --- CHANGELOG | 7 ++ lib/bridgedb/Main.py | 203 +++++++++++++++++++++++++++++++------------------- 2 files changed, 134 insertions(+), 76 deletions(-) diff --git a/CHANGELOG b/CHANGELOG index 092ce8c..0498e4f 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -1,3 +1,10 @@ +Changes in version 0.1.7 - + * FIXES #5232 Perform long running and blocking transactions + in background threads. Primarily this moves bridge + descriptor reparsing into another thread and + significantly increased the availability of bridgedb, + as a result. + Changes in version 0.1.6 - 2014-03-26 BridgeDB 0.1.6 includes fixes for the following bugs: * FIXES #11196 BridgeDB should use leekspin diff --git a/lib/bridgedb/Main.py b/lib/bridgedb/Main.py index 7fc4c6c..6c099b8 100644 --- a/lib/bridgedb/Main.py +++ b/lib/bridgedb/Main.py @@ -92,6 +92,7 @@ def load(state, splitter, clear=False): logging.info("Loading bridges...") + bridges = {} status = {} addresses = {} timestamps = {} @@ -118,8 +119,6 @@ def load(state, splitter, clear=False): logging.debug("Closing network status file") f.close() - db = bridgedb.Storage.getDB() - for fname in state.BRIDGE_FILES: logging.info("Opening bridge-server-descriptor file: '%s'" % fname) f = open(fname, 'r') @@ -191,6 +190,27 @@ def load(state, splitter, clear=False): logging.debug("Closing blocking-countries document") f.close() + def updateBridgeHistory(bridges, timestamps): + if not hasattr(state, 'config'): + logging.info("updateBridgeHistory(): Config file not set "\ + "in State file.") + return + if state.COLLECT_TIMESTAMPS: + logging.debug("Beginning bridge stability calculations") + for bridge in bridges.values(): + if bridge.getID() in timestamps.keys(): + ts = timestamps[bridge.getID()][:] + ts.sort() + for timestamp in ts: + logging.debug( + "Updating BridgeHistory timestamps for %s: %s" + % (bridge.fingerprint, timestamp)) + bridgedb.Stability.addOrUpdateBridgeHistory( + bridge, timestamp) + logging.debug("Stability calculations complete") + + reactor.callInThread(updateBridgeHistory, bridges, timestamps) + bridges = None state.save() return @@ -327,7 +347,7 @@ def _reloadFn(*args): def _handleSIGHUP(*args): """Called when we receive a SIGHUP; invokes _reloadFn.""" - reactor.callLater(0, _reloadFn, *args) + reactor.callInThread(_reloadFn) def _handleSIGUSR1(*args): """Handler for SIGUSR1. Calls :func:`~bridgedb.runner.doDumpBridges`.""" @@ -340,7 +360,7 @@ def _handleSIGUSR1(*args): cfg = loadConfig(state.CONFIG_FILE, state.config) logging.info("Dumping bridge assignments to files...") - reactor.callLater(0, runner.doDumpBridges, cfg) + reactor.callInThread(runner.doDumpBridges, cfg) class ProxyCategory: @@ -351,6 +371,75 @@ class ProxyCategory: def replaceProxyList(self, ipset): self.ipset = ipset +def replaceBridgeRings(current, replacement): + """Replace the current thing with the new one""" + current.splitter = replacement.splitter + +def createBridgeRings(cfg, proxyList, key): + """Create the bridge distributors defined by the config file + + :type cfg: :class:`Conf` + :param cfg: The current configuration, including any in-memory + settings (i.e. settings whose values were not obtained from the + config file, but were set via a function somewhere) + :type proxyList: :class:`ProxyCategory` + :param proxyList: The container for the IP addresses of any currently + known open proxies. + :param bytes key: Splitter master key + :rtype: tuple + :returns: A BridgeSplitter splitter, an IPBasedDistributor or None, + and an EmailBasedDistributor or None. + """ + + # Create a BridgeSplitter to assign the bridges to the different + # distributors. + splitter = Bridges.BridgeSplitter(crypto.getHMAC(key, "Splitter-Key")) + logging.debug("Created splitter: %r" % splitter) + + # Create ring parameters. + ringParams = Bridges.BridgeRingParameters(needPorts=cfg.FORCE_PORTS, + needFlags=cfg.FORCE_FLAGS) + + emailDistributor = ipDistributor = None + # As appropriate, create an IP-based distributor. + if cfg.HTTPS_DIST and cfg.HTTPS_SHARE: + logging.debug("Setting up HTTPS Distributor...") + categories = [] + if proxyList.ipset: + logging.debug("Adding proxyList to HTTPS Distributor categories.") + categories.append(proxyList) + logging.debug("HTTPS Distributor categories: '%s'" % categories) + + ipDistributor = Dist.IPBasedDistributor( + Dist.uniformMap, + cfg.N_IP_CLUSTERS, + crypto.getHMAC(key, "HTTPS-IP-Dist-Key"), + categories, + answerParameters=ringParams) + splitter.addRing(ipDistributor, "https", cfg.HTTPS_SHARE) + + # As appropriate, create an email-based distributor. + if cfg.EMAIL_DIST and cfg.EMAIL_SHARE: + logging.debug("Setting up Email Distributor...") + emailDistributor = Dist.EmailBasedDistributor( + crypto.getHMAC(key, "Email-Dist-Key"), + cfg.EMAIL_DOMAIN_MAP.copy(), + cfg.EMAIL_DOMAIN_RULES.copy(), + answerParameters=ringParams) + splitter.addRing(emailDistributor, "email", cfg.EMAIL_SHARE) + + # As appropriate, tell the splitter to leave some bridges unallocated. + if cfg.RESERVED_SHARE: + splitter.addRing(Bridges.UnallocatedHolder(), + "unallocated", + cfg.RESERVED_SHARE) + + # Add pseudo distributors to splitter + for pseudoRing in cfg.FILE_BUCKETS.keys(): + splitter.addPseudoRing(pseudoRing) + + return splitter, emailDistributor, ipDistributor + def startup(options): """Parse bridges, @@ -402,73 +491,18 @@ def startup(options): # Load the master key, or create a new one. key = crypto.getKey(config.MASTER_KEY_FILE) - # Initialize our DB file. - db = bridgedb.Storage.Database(config.DB_FILE + ".sqlite", config.DB_FILE) - # TODO: move setGlobalDB to bridgedb.persistent.State class - bridgedb.Storage.setGlobalDB(db) - # Get a proxy list. proxyList = ProxyCategory() proxyList.replaceProxyList(loadProxyList(config)) - # Create a BridgeSplitter to assign the bridges to the different - # distributors. - splitter = Bridges.BridgeSplitter(crypto.getHMAC(key, "Splitter-Key")) - logging.debug("Created splitter: %r" % splitter) - - # Create ring parameters. - ringParams = Bridges.BridgeRingParameters(needPorts=config.FORCE_PORTS, - needFlags=config.FORCE_FLAGS) - emailDistributor = ipDistributor = None - # As appropriate, create an IP-based distributor. - if config.HTTPS_DIST and config.HTTPS_SHARE: - logging.debug("Setting up HTTPS Distributor...") - categories = [] - if proxyList.ipset: - logging.debug("Adding proxyList to HTTPS Distributor categories.") - categories.append(proxyList) - logging.debug("HTTPS Distributor categories: '%s'" % categories) - - ipDistributor = Dist.IPBasedDistributor( - Dist.uniformMap, - config.N_IP_CLUSTERS, - crypto.getHMAC(key, "HTTPS-IP-Dist-Key"), - categories, - answerParameters=ringParams) - splitter.addRing(ipDistributor, "https", config.HTTPS_SHARE) - #webSchedule = Time.IntervalSchedule("day", 2) - webSchedule = Time.NoSchedule() - - # As appropriate, create an email-based distributor. - if config.EMAIL_DIST and config.EMAIL_SHARE: - logging.debug("Setting up Email Distributor...") - emailDistributor = Dist.EmailBasedDistributor( - crypto.getHMAC(key, "Email-Dist-Key"), - config.EMAIL_DOMAIN_MAP.copy(), - config.EMAIL_DOMAIN_RULES.copy(), - answerParameters=ringParams) - splitter.addRing(emailDistributor, "email", config.EMAIL_SHARE) - #emailSchedule = Time.IntervalSchedule("day", 1) - emailSchedule = Time.NoSchedule() - - # As appropriate, tell the splitter to leave some bridges unallocated. - if config.RESERVED_SHARE: - splitter.addRing(Bridges.UnallocatedHolder(), - "unallocated", - config.RESERVED_SHARE) - - # Add pseudo distributors to splitter - for pseudoRing in config.FILE_BUCKETS.keys(): - splitter.addPseudoRing(pseudoRing) - # Save our state state.proxyList = proxyList state.key = key state.save() - def reload(*args): + def reload(inThread=True): """Reload settings, proxy lists, and bridges. State should be saved before calling this method, and will be saved @@ -514,6 +548,14 @@ def startup(options): logging.debug("Saving state again before reparsing descriptors...") state.save() logging.info("Reparsing bridge descriptors...") + + (splitter, + emailDistributorTmp, + ipDistributorTmp) = createBridgeRings(cfg, proxyList, key) + + # Initialize our DB file. + db = bridgedb.Storage.Database(cfg.DB_FILE + ".sqlite", cfg.DB_FILE) + bridgedb.Storage.setGlobalDB(db) load(state, splitter, clear=False) state = persistent.load() @@ -521,32 +563,32 @@ def startup(options): logging.debug("Replacing the list of open proxies...") state.proxyList.replaceProxyList(loadProxyList(cfg)) - if emailDistributor is not None: - emailDistributor.prepopulateRings() # create default rings + if emailDistributorTmp is not None: + emailDistributorTmp.prepopulateRings() # create default rings logging.info("Bridges allotted for %s distribution: %d" - % (emailDistributor.name, - len(emailDistributor.splitter))) + % (emailDistributorTmp.name, + len(emailDistributorTmp.splitter))) else: logging.warn("No email distributor created!") - if ipDistributor is not None: - ipDistributor.prepopulateRings() # create default rings + if ipDistributorTmp is not None: + ipDistributorTmp.prepopulateRings() # create default rings logging.info("Bridges allotted for %s distribution: %d" - % (ipDistributor.name, - len(ipDistributor.splitter))) + % (ipDistributorTmp.name, + len(ipDistributorTmp.splitter))) logging.info("\tNum bridges:\tFilter set:") nSubrings = 0 - ipSubrings = ipDistributor.splitter.filterRings + ipSubrings = ipDistributorTmp.splitter.filterRings for (ringname, (filterFn, subring)) in ipSubrings.items(): nSubrings += 1 filterSet = ' '.join( - ipDistributor.splitter.extractFilterNames(ringname)) + ipDistributorTmp.splitter.extractFilterNames(ringname)) logging.info("\t%2d bridges\t%s" % (len(subring), filterSet)) logging.info("Total subrings for %s: %d" - % (ipDistributor.name, nSubrings)) + % (ipDistributorTmp.name, nSubrings)) else: logging.warn("No HTTP(S) distributor created!") @@ -563,21 +605,32 @@ def startup(options): except IOError: logging.info("I/O error while writing assignments to: '%s'" % state.ASSIGNMENTS_FILE) - state.save() + if inThread: + reactor.callFromThread(replaceBridgeRings, ipDistributor, ipDistributorTmp) + reactor.callFromThread(replaceBridgeRings, emailDistributor, emailDistributorTmp) + else: + # We're still starting up. Return these distributors so + # they are configured in the outer-namespace + return emailDistributorTmp, ipDistributorTmp + global _reloadFn _reloadFn = reload signal.signal(signal.SIGHUP, _handleSIGHUP) signal.signal(signal.SIGUSR1, _handleSIGUSR1) - # And actually load it to start parsing. - reload() + # And actually load it to start parsing. Get back our distributors. + emailDistributor, ipDistributor = reload(False) # Configure all servers: if config.HTTPS_DIST and config.HTTPS_SHARE: + #webSchedule = Time.IntervalSchedule("day", 2) + webSchedule = Time.NoSchedule() HTTPServer.addWebServer(config, ipDistributor, webSchedule) if config.EMAIL_DIST and config.EMAIL_SHARE: + #emailSchedule = Time.IntervalSchedule("day", 1) + emailSchedule = Time.NoSchedule() EmailServer.addSMTPServer(config, emailDistributor, emailSchedule) # Actually run the servers. @@ -587,8 +640,6 @@ def startup(options): except KeyboardInterrupt: logging.fatal("Received keyboard interrupt. Shutting down...") finally: - logging.info("Closing databases...") - db.close() if config.PIDFILE: os.unlink(config.PIDFILE) logging.info("Exiting...")

1 0

[bridgedb/develop] Return a database handle via generator
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit 2a99f95e6bd7357c6adbd2e5f4f09b24d169f770 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Mar 21 02:48:46 2014 +0000 Return a database handle via generator Define a new idiom for retrieving an instance of bridgedb.Storage.Database Use: with bridgedb.Storage.getDB() as db: db.<do stuff with the database> --- lib/bridgedb/Storage.py | 156 ++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 149 insertions(+), 7 deletions(-) diff --git a/lib/bridgedb/Storage.py b/lib/bridgedb/Storage.py index fdeea23..d9d8b29 100644 --- a/lib/bridgedb/Storage.py +++ b/lib/bridgedb/Storage.py @@ -9,10 +9,15 @@ import binascii import sqlite3 import time import hashlib +from contextlib import contextmanager +from contextlib import GeneratorContextManager +from functools import wraps from ipaddr import IPAddress, IPv6Address, IPv4Address +import sys import bridgedb.Stability as Stability from bridgedb.Stability import BridgeHistory +import threading toHex = binascii.b2a_hex fromHex = binascii.a2b_hex @@ -201,13 +206,14 @@ class BridgeData: self.first_seen = first_seen self.last_seen = last_seen -class Database: +class Database(object): def __init__(self, sqlite_fname, db_fname=None): if db_fname is None: self._conn = openDatabase(sqlite_fname) else: self._conn = openOrConvertDatabase(sqlite_fname, db_fname) self._cur = self._conn.cursor() + self.sqlite_fname = sqlite_fname def commit(self): self._conn.commit() @@ -216,6 +222,7 @@ class Database: self._conn.rollback() def close(self): + #print "Closing DB" self._cur.close() self._conn.close() @@ -498,12 +505,147 @@ def openOrConvertDatabase(sqlite_file, db_file): conn.commit() return conn -_THE_DB = None +class DBGeneratorContextManager(GeneratorContextManager): + """Overload __exit__() so we can call the generator many times""" + def __exit__(self, type, value, traceback): + """Significantly based on contextlib.py""" + if type is None: + try: + self.gen.next() + except StopIteration: + return + return + else: + if value is None: + # Need to force instantiation so we can reliably + # tell if we get the same exception back + value = type() + try: + self.gen.throw(type, value, traceback) + raise RuntimeError("generator didn't stop after throw()") + except StopIteration, exc: + # Suppress the exception *unless* it's the same exception that + # was passed to throw(). This prevents a StopIteration + # raised inside the "with" statement from being suppressed + return exc is not value + except: + # only re-raise if it's *not* the exception that was + # passed to throw(), because __exit__() must not raise + # an exception unless __exit__() itself failed. But throw() + # has to raise the exception to signal propagation, so this + # fixes the impedance mismatch between the throw() protocol + # and the __exit__() protocol. + # + if sys.exc_info()[1] is not value: + raise +def contextmanager(func): + @wraps(func) + def helper(*args, **kwds): + return DBGeneratorContextManager(func(*args, **kwds)) + return helper + +_DB_FNAME = None +_LOCK = None +_LOCKED = False +_OPENED_DB = None +_TOC = None + +def clearGlobalDB(): + """Start from scratch""" + global _DB_FNAME + global _LOCK + global _LOCKED + global _OPENED_DB + global _TOC + + _DB_FNAME = None + _LOCK = None + _LOCKED = False + _OPENED_DB = None + _TOC = None + +def checkAndConvertDB(sqlite_file, db_file): + openOrConvertDatabase(sqlite_file, db_file).close() + +def setDBFilename(sqlite_fname): + global _DB_FNAME + _DB_FNAME = sqlite_fname + +@contextmanager +def getDB(block=True): + """Generator: Return a usable database handler + + Always return a :class:`bridgedb.Storage.Database` that is + usable within the current thread. If a connection already exists + and it was created by the current thread, then return the + associated :class:`bridgedb.Storage.Database` instance. Otherwise, + create a new instance, blocking until the existing connection + is closed, if applicable. + + Note: This is a blocking call, be careful about deadlocks! + + :rtype: :class:`bridgedb.Storage.Database` + :returns: An instance of :class:`bridgedb.Storage.Database` used to + query the database + """ + global _DB_FNAME + global _LOCK + global _LOCKED + global _OPENED_DB + global _TOC + + refcount = 0 + + if not _LOCK: + _LOCK = threading.RLock() + this_thread = threading.current_thread().ident -def setGlobalDB(db): - global _THE_DB - _THE_DB = db + #print "_DB_FNAME: %s, _LOCK: %s, _LOCKED: %s, _OPENED_DB: %s, _TOC: %s" % \ + # (_DB_FNAME, _LOCK, _LOCKED, _OPENED_DB, _TOC) + + try: + if _TOC == this_thread: + refcount += 1 + #print "yielding db from earlier" + logging.debug("Refcount: %d" % refcount) + yield _OPENED_DB + logging.debug("Checking lock status") + logging.debug("lock _LOCKED: %s" % _LOCKED) + locked = _LOCK.acquire(False) + logging.debug("lock aquired: %s" % locked) + if locked: _LOCK.release() + if _LOCK.acquire(block): + db = Database(_DB_FNAME) + assert not _LOCKED + _LOCKED = True + _OPENED_DB = db + _TOC = this_thread + refcount += 1 + #print "yielding db of type: %s" % type(db) + #print "_DB_FNAME: %s, _LOCK: %s, _LOCKED: %s, _OPENED_DB: %s, _TOC: %s" % \ + # (_DB_FNAME, _LOCK, _LOCKED, _OPENED_DB, _TOC) + logging.debug("yielding db") + yield db + else: + #print "yielding False" + yield False + finally: + logging.debug("Refcount is %d in finally" % refcount) + if refcount == 1: + #assert _LOCKED + _LOCK.release() + assert _LOCK.acquire() + _LOCK.release() + _LOCKED = False + _OPENED_DB = None + _TOC = None + db.close() + logging.debug("Refcount resulted in closed") + else: + refcount -= 1 -def getDB(): - return _THE_DB +def closeDB(): + next(getDB(), None) +def dbIsLocked(): + return _LOCKED

1 0

[bridgedb/develop] Add useful output during startup integration test
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit a3d59f0b2b9870ee8ee184bfb6617844891f8b57 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Wed Mar 26 05:26:07 2014 +0000 Add useful output during startup integration test --- lib/bridgedb/test/test_bridgedb.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/lib/bridgedb/test/test_bridgedb.py b/lib/bridgedb/test/test_bridgedb.py index 1f4470b..e73e2bc 100644 --- a/lib/bridgedb/test/test_bridgedb.py +++ b/lib/bridgedb/test/test_bridgedb.py @@ -92,10 +92,12 @@ class BridgeDBCliTest(unittest.TestCase): # Sorry Windows users devnull = open('/dev/null', 'w') bridgedbProc = Popen([bridgedbScript, '-r', runDir], stdout=devnull) + print("Waiting 30 seconds while bridgedb loads...") time.sleep(30) assignments = pjoin(runDir, 'assignments.log') self.assertTrue(os.path.isfile(assignments)) os.unlink(assignments) + print("Sending SIGHUP, checking for assignments.log ...") bridgedbProc.send_signal(signal.SIGHUP) time.sleep(5) try: @@ -105,6 +107,7 @@ class BridgeDBCliTest(unittest.TestCase): bridgedbProcCode = bridgedbProc.wait() print("`bridgedb' exited with status code %d" % int(bridgedbProcCode)) raise e + print("Sending SIGUSR1, checking for bucket files...") bridgedbProc.send_signal(signal.SIGUSR1) time.sleep(5) buckets = [['email', False], ['https', False], ['unallocated', False]] @@ -121,6 +124,7 @@ class BridgeDBCliTest(unittest.TestCase): bridgedbProcCode = bridgedbProc.wait() print("`bridgedb' exited with status code %d" % int(bridgedbProcCode)) raise e + print("Done. Killing processes.") bridgedbProc.send_signal(signal.SIGINT) bridgedbProcCode = bridgedbProc.wait() print("`bridgedb' exited with status code %d" % int(bridgedbProcCode))

1 0

[bridgedb/develop] Rewrite database lock acquisition function
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit bd05227c5eefc528b72d57d53ac3f139fc2f0d5e Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Mar 21 04:54:48 2014 +0000 Rewrite database lock acquisition function Restrict access to the database handle when multiple threads request the ability to query. thanks to yawning --- lib/bridgedb/Storage.py | 114 ++++++++++++++++++++++++----------------------- 1 file changed, 58 insertions(+), 56 deletions(-) diff --git a/lib/bridgedb/Storage.py b/lib/bridgedb/Storage.py index d9d8b29..72c0af2 100644 --- a/lib/bridgedb/Storage.py +++ b/lib/bridgedb/Storage.py @@ -506,9 +506,20 @@ def openOrConvertDatabase(sqlite_file, db_file): return conn class DBGeneratorContextManager(GeneratorContextManager): - """Overload __exit__() so we can call the generator many times""" + """Helper for @contextmanager decorator. + + Overload __exit__() so we can call the generator many times + """ def __exit__(self, type, value, traceback): - """Significantly based on contextlib.py""" + """Handle exiting a with statement block + + Progress generator or throw exception + + Significantly based on contextlib.py + + :throws: `RuntimeError` if the generator doesn't stop after + exception is thrown + """ if type is None: try: self.gen.next() @@ -538,7 +549,12 @@ class DBGeneratorContextManager(GeneratorContextManager): # if sys.exc_info()[1] is not value: raise + def contextmanager(func): + """Decorator to for :func:`Storage.getDB()` + + Define getDB() for use by with statement content manager + """ @wraps(func) def helper(*args, **kwds): return DBGeneratorContextManager(func(*args, **kwds)) @@ -546,23 +562,36 @@ def contextmanager(func): _DB_FNAME = None _LOCK = None -_LOCKED = False +_LOCKED = 0 _OPENED_DB = None -_TOC = None +_REFCOUNT = 0 def clearGlobalDB(): - """Start from scratch""" + """Start from scratch + + This is currently only used in unit tests. + """ global _DB_FNAME global _LOCK global _LOCKED global _OPENED_DB - global _TOC _DB_FNAME = None _LOCK = None - _LOCKED = False + _LOCKED = 0 _OPENED_DB = None - _TOC = None + _REFCOUNT = 0 + +def initializeDBLock(): + """Create the lock + + This must be called before the first database query + """ + global _LOCK + + if not _LOCK: + _LOCK = threading.RLock() + assert _LOCK def checkAndConvertDB(sqlite_file, db_file): openOrConvertDatabase(sqlite_file, db_file).close() @@ -582,7 +611,8 @@ def getDB(block=True): create a new instance, blocking until the existing connection is closed, if applicable. - Note: This is a blocking call, be careful about deadlocks! + Note: This is a blocking call (by default), be careful about + deadlocks! :rtype: :class:`bridgedb.Storage.Database` :returns: An instance of :class:`bridgedb.Storage.Database` used to @@ -592,60 +622,32 @@ def getDB(block=True): global _LOCK global _LOCKED global _OPENED_DB - global _TOC + global _REFCOUNT - refcount = 0 - - if not _LOCK: - _LOCK = threading.RLock() - this_thread = threading.current_thread().ident + assert _LOCK + try: + own_lock = _LOCK.acquire(block) + if own_lock: + _LOCKED += 1 - #print "_DB_FNAME: %s, _LOCK: %s, _LOCKED: %s, _OPENED_DB: %s, _TOC: %s" % \ - # (_DB_FNAME, _LOCK, _LOCKED, _OPENED_DB, _TOC) + if not _OPENED_DB: + assert _REFCOUNT == 0 + _OPENED_DB = Database(_DB_FNAME) - try: - if _TOC == this_thread: - refcount += 1 - #print "yielding db from earlier" - logging.debug("Refcount: %d" % refcount) + _REFCOUNT += 1 yield _OPENED_DB - logging.debug("Checking lock status") - logging.debug("lock _LOCKED: %s" % _LOCKED) - locked = _LOCK.acquire(False) - logging.debug("lock aquired: %s" % locked) - if locked: _LOCK.release() - if _LOCK.acquire(block): - db = Database(_DB_FNAME) - assert not _LOCKED - _LOCKED = True - _OPENED_DB = db - _TOC = this_thread - refcount += 1 - #print "yielding db of type: %s" % type(db) - #print "_DB_FNAME: %s, _LOCK: %s, _LOCKED: %s, _OPENED_DB: %s, _TOC: %s" % \ - # (_DB_FNAME, _LOCK, _LOCKED, _OPENED_DB, _TOC) - logging.debug("yielding db") - yield db else: - #print "yielding False" yield False finally: - logging.debug("Refcount is %d in finally" % refcount) - if refcount == 1: - #assert _LOCKED - _LOCK.release() - assert _LOCK.acquire() + assert own_lock + try: + _REFCOUNT -= 1 + if _REFCOUNT == 0: + _OPENED_DB.close() + _OPENED_DB = None + finally: + _LOCKED -= 1 _LOCK.release() - _LOCKED = False - _OPENED_DB = None - _TOC = None - db.close() - logging.debug("Refcount resulted in closed") - else: - refcount -= 1 - -def closeDB(): - next(getDB(), None) def dbIsLocked(): - return _LOCKED + return _LOCKED != 0

1 0

[bridgedb/develop] Bump leekspin to v0.1.3
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit 6ce6bcc999882535169aaaa68797d31f3be2c0d7 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Thu Mar 27 03:10:05 2014 +0000 Bump leekspin to v0.1.3 --- .travis.yml | 2 +- CHANGELOG | 3 +++ setup.py | 2 +- 3 files changed, 5 insertions(+), 2 deletions(-) diff --git a/.travis.yml b/.travis.yml index 5553a81..7aef0ca 100644 --- a/.travis.yml +++ b/.travis.yml @@ -28,7 +28,7 @@ python: install: - pip install -r requirements.txt - - pip install --no-use-wheel leekspin==0.1.1 coverage==3.7 coveralls==0.3 sure==1.2.2 + - pip install --no-use-wheel leekspin==0.1.3 coverage==3.7 coveralls==0.3 sure==1.2.2 - make install script: diff --git a/CHANGELOG b/CHANGELOG index 0498e4f..4f36267 100644 --- a/CHANGELOG +++ b/CHANGELOG @@ -5,6 +5,9 @@ Changes in version 0.1.7 - significantly increased the availability of bridgedb, as a result. +And includes the following general changes: + * BUMPS leekspin version to 0.1.3 + Changes in version 0.1.6 - 2014-03-26 BridgeDB 0.1.6 includes fixes for the following bugs: * FIXES #11196 BridgeDB should use leekspin diff --git a/setup.py b/setup.py index d859f87..d77bc1d 100644 --- a/setup.py +++ b/setup.py @@ -277,7 +277,7 @@ setuptools.setup( scripts=['scripts/bridgedb'], extras_require={'test': ["sure==1.2.2", "coverage==3.7", - "leekspin==0.1.1"]}, + "leekspin==0.1.3"]}, zip_safe=False, cmdclass=get_cmdclass(), include_package_data=True,

1 0

[bridgedb/develop] Only update distributor's hashring if it's still active
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit a79d154de5db6293321f6c68835e8f4ff25032d8 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Fri Mar 28 00:15:16 2014 +0000 Only update distributor's hashring if it's still active TODO: shutdown the distributor in this situation --- lib/bridgedb/Main.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/lib/bridgedb/Main.py b/lib/bridgedb/Main.py index a28b75f..1f7e52d 100644 --- a/lib/bridgedb/Main.py +++ b/lib/bridgedb/Main.py @@ -609,8 +609,10 @@ def startup(options): state.save() if inThread: - reactor.callFromThread(replaceBridgeRings, ipDistributor, ipDistributorTmp) - reactor.callFromThread(replaceBridgeRings, emailDistributor, emailDistributorTmp) + if ipDistributorTmp: + reactor.callFromThread(replaceBridgeRings, ipDistributor, ipDistributorTmp) + if emailDistributorTmp: + reactor.callFromThread(replaceBridgeRings, emailDistributor, emailDistributorTmp) else: # We're still starting up. Return these distributors so # they are configured in the outer-namespace

1 0

[bridgedb/develop] Add TODO mentioning that we should be able to disable distributors
by isis＠torproject.org 02 Apr '14

02 Apr '14

commit 8dfcd5d7f77950828d3c95f93417baf7ea70c6b7 Author: Matthew Finkel <Matthew.Finkel(a)gmail.com> Date: Sun Mar 30 13:36:49 2014 +0000 Add TODO mentioning that we should be able to disable distributors If we reparse our config and it tells us that a distributor, which is currently running, should be disabled then we should be able to shut it down. Also wrap long lines. --- lib/bridgedb/Main.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/lib/bridgedb/Main.py b/lib/bridgedb/Main.py index 1f7e52d..8a1b650 100644 --- a/lib/bridgedb/Main.py +++ b/lib/bridgedb/Main.py @@ -609,10 +609,14 @@ def startup(options): state.save() if inThread: + # XXX shutdown the distributors if they were previously running + # and should now be disabled if ipDistributorTmp: - reactor.callFromThread(replaceBridgeRings, ipDistributor, ipDistributorTmp) + reactor.callFromThread(replaceBridgeRings, + ipDistributor, ipDistributorTmp) if emailDistributorTmp: - reactor.callFromThread(replaceBridgeRings, emailDistributor, emailDistributorTmp) + reactor.callFromThread(replaceBridgeRings, + emailDistributor, emailDistributorTmp) else: # We're still starting up. Return these distributors so # they are configured in the outer-namespace

1 0