Commit 2fe24944 authored by Jim Fulton's avatar Jim Fulton

Added a database removeVersionPool method to remove a version pool.

parent b301836e
......@@ -13,8 +13,8 @@
##############################################################################
"""Database objects
$Id: DB.py,v 1.50 2003/04/23 20:05:51 jeremy Exp $"""
__version__='$Revision: 1.50 $'[11:-2]
$Id: DB.py,v 1.51 2003/06/24 21:29:54 jim Exp $"""
__version__='$Revision: 1.51 $'[11:-2]
import cPickle, cStringIO, sys, POSException, UndoLogCompatible
from Connection import Connection
......@@ -136,12 +136,23 @@ class DB(UndoLogCompatible.UndoLogCompatible):
am.closedConnection(connection)
version=connection._version
pools,pooll=self._pools
try:
pool, allocated, pool_lock = pools[version]
except KeyError:
# No such version. We must have deleted the pool.
# Just let the connection go.
# We need to break circular refs to make it really go:
connection.__dict__.clear()
return
pool.append(connection)
if len(pool)==1:
# Pool now usable again, unlock it.
pool_lock.release()
finally: self._r()
finally:
self._r()
def _connectionMap(self, f):
self._a()
......@@ -384,7 +395,7 @@ class DB(UndoLogCompatible.UndoLogCompatible):
return c
pools,pooll=self._pools
pools, pooll = self._pools
# pools is a mapping object:
#
......@@ -473,6 +484,19 @@ class DB(UndoLogCompatible.UndoLogCompatible):
finally: self._r()
def removeVersionPool(self, version):
pools, pooll = self._pools
info = pools.get(version)
if info:
del pools[version]
pool, allocated, pool_lock = info
pooll.remove((pool, allocated))
try: pool_lock.release()
except: pass
del pool[:]
del allocated[:]
6
def connectionDebugInfo(self):
r=[]
pools,pooll=self._pools
......
......@@ -11,22 +11,27 @@
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
import os
import time
import unittest
import ZODB
import ZODB.MappingStorage
import ZODB.FileStorage
from ZODB.tests.MinPO import MinPO
class DBTests(unittest.TestCase):
def setUp(self):
store = ZODB.MappingStorage.MappingStorage()
self.__path = os.path.abspath('test.fs')
store = ZODB.FileStorage.FileStorage(self.__path)
self.db = ZODB.DB(store)
def tearDown(self):
self.db.close()
for s in ('', '.index', '.lock', '.tmp'):
if os.path.exists(self.__path+s):
os.remove(self.__path+s)
def dowork(self, version=''):
c = self.db.open(version)
......@@ -37,7 +42,6 @@ class DBTests(unittest.TestCase):
o.value = MinPO(i)
get_transaction().commit()
o = o.value
print r.items()
c.close()
# make sure the basic methods are callable
......@@ -49,5 +53,78 @@ class DBTests(unittest.TestCase):
self.db.setVersionCacheDeactivateAfter(12) # deprecated
self.db.setVersionCacheSize(15)
def test_removeVersionPool(self):
# Test that we can remove a version pool
# This is white box because we check some internal data structures
self.dowork()
self.dowork('v2')
c1 = self.db.open('v1')
c1.close() # return to pool
c12 = self.db.open('v1')
c12.close() # return to pool
self.assert_(c1 is c12) # should be same
pools, pooll = self.db._pools
self.assertEqual(len(pools), 3)
self.assertEqual(len(pooll), 3)
self.db.removeVersionPool('v1')
self.assertEqual(len(pools), 2)
self.assertEqual(len(pooll), 2)
c12 = self.db.open('v1')
c12.close() # return to pool
self.assert_(c1 is not c12) # should be different
self.assertEqual(len(pools), 3)
self.assertEqual(len(pooll), 3)
def _test_for_leak(self):
self.dowork()
self.dowork('v2')
while 1:
c1 = self.db.open('v1')
self.db.removeVersionPool('v1')
c1.close() # return to pool
def test_removeVersionPool_while_connection_open(self):
# Test that we can remove a version pool
# This is white box because we check some internal data structures
self.dowork()
self.dowork('v2')
c1 = self.db.open('v1')
c1.close() # return to pool
c12 = self.db.open('v1')
self.assert_(c1 is c12) # should be same
pools, pooll = self.db._pools
self.assertEqual(len(pools), 3)
self.assertEqual(len(pooll), 3)
self.db.removeVersionPool('v1')
self.assertEqual(len(pools), 2)
self.assertEqual(len(pooll), 2)
c12.close() # should leave pools alone
self.assertEqual(len(pools), 2)
self.assertEqual(len(pooll), 2)
c12 = self.db.open('v1')
c12.close() # return to pool
self.assert_(c1 is not c12) # should be different
self.assertEqual(len(pools), 3)
self.assertEqual(len(pooll), 3)
def test_suite():
return unittest.makeSuite(DBTests)
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