zombie dal is here, should fix scheduler problems while retaining Row/Rows serialization

This commit is contained in:
mdipierro
2012-10-07 01:07:49 -05:00
parent 231a3e1278
commit 668658b7c9
2 changed files with 44 additions and 34 deletions
+1 -1
View File
@@ -1 +1 @@
Version 2.0.9 (2012-10-05 10:38:54) dev Version 2.0.9 (2012-10-07 01:07:45) dev
+43 -33
View File
@@ -177,9 +177,9 @@ CALLABLETYPES = (types.LambdaType, types.FunctionType,
types.MethodType, types.BuiltinMethodType) types.MethodType, types.BuiltinMethodType)
TABLE_ARGS = set( TABLE_ARGS = set(
('migrate','primarykey','fake_migrate','format', ('migrate','primarykey','fake_migrate','format','redefine',
'singular','plural','trigger_name','sequence_name', 'singular','plural','trigger_name','sequence_name',
'common_filter','polymodel','table_class','on_define')) 'common_filter','polymodel','table_class','on_define',))
SELECT_ARGS = set( SELECT_ARGS = set(
('orderby', 'groupby', 'limitby','required', 'cache', 'left', ('orderby', 'groupby', 'limitby','required', 'cache', 'left',
@@ -521,11 +521,11 @@ class ConnectionPool(object):
@staticmethod @staticmethod
def close_all_instances(action): def close_all_instances(action):
""" to close cleanly databases in a multithreaded environment """ """ to close cleanly databases in a multithreaded environment """
dbs = getattr(THREAD_LOCAL,'db_instances',{}).items() db_group = getattr(THREAD_LOCAL,'db_instances',{}).items()
for singleton_code, db in dbs: for db_uid, db_group in dbs:
if hasattr(db,'_adapter'): for db in db_group:
db._adapter.close(action) db._adapter.close(action)
del THREAD_LOCAL.db_instances[singleton_code] del THREAD_LOCAL.db_instances[db_uid]
if callable(action): if callable(action):
action(None) action(None)
return return
@@ -6585,23 +6585,31 @@ class DAL(object):
""" """
def __new__(cls, uri='sqlite://dummy.db', *args, **kwargs): def __new__(cls, uri='sqlite://dummy.db', *args, **kwargs):
if uri==None and not 'singleton_code' in kwargs:
# this deal with the special case of Dummy DAL for SQLFORM.factory
return super(DAL, cls).__new__(cls)
if not hasattr(THREAD_LOCAL,'db_instances'): if not hasattr(THREAD_LOCAL,'db_instances'):
THREAD_LOCAL.db_instances = {} THREAD_LOCAL.db_instances = {}
if 'singleton_code' in kwargs: if not hasattr(THREAD_LOCAL,'db_instances_zombie'):
singleton_code = kwargs['singleton_code'] THREAD_LOCAL.db_instances_zombie = {}
del kwargs['singleton_code'] if uri == '<zombie>':
singleton_code = hashlib.md5(repr(uri)).hexdigest() db_uid = kwargs['db_uid'] # a zombie must have a db_uid!
try: if db_uid in THREAD_LOCAL.db_instances:
db = THREAD_LOCAL.db_instances[singleton_code] db_group = THREAD_LOCAL.db_instances[db_uid]
if args or kwargs: db = db_group[-1]
raise RuntimeError, 'Cannot duplicate a Singleton' elif db_uid in THREAD_LOCAL.db_instances_zombie:
except KeyError: db = THREAD_LOCAL.db_instances_zombie[db_uid]
db = super(DAL, cls).__new__(cls) else:
THREAD_LOCAL.db_instances[singleton_code] = db db = super(DAL, cls).__new__(cls)
db._singleton_code = singleton_code THREAD_LOCAL.db_instances_zombie[db_uid] = db
else:
db_uid = kwargs.get('db_uid',hashlib.md5(repr(uri)).hexdigest())
if db_uid in THREAD_LOCAL.db_instances_zombie:
db = THREAD_LOCAL.db_instances_zombie[db_uid]
del THREAD_LOCAL.db_instances_zombie[db_uid]
else:
db = super(DAL, cls).__new__(cls)
db_group = THREAD_LOCAL.db_instances.get(db_uid,[])
db_group.append(db)
THREAD_LOCAL.db_instances[db_uid] = db_group
db._db_uid = db_uid
return db return db
@staticmethod @staticmethod
@@ -6657,7 +6665,7 @@ class DAL(object):
decode_credentials=False, driver_args=None, decode_credentials=False, driver_args=None,
adapter_args=None, attempts=5, auto_import=False, adapter_args=None, attempts=5, auto_import=False,
bigint_id=False,debug=False,lazy_tables=False, bigint_id=False,debug=False,lazy_tables=False,
singleton_code=None): db_uid=None):
""" """
Creates a new Database Abstraction Layer instance. Creates a new Database Abstraction Layer instance.
@@ -6685,7 +6693,7 @@ class DAL(object):
:attempts (defaults to 5). Number of times to attempt connecting :attempts (defaults to 5). Number of times to attempt connecting
""" """
if hasattr(self,'_adapter') or uri=='<lazy>': return if uri == '<zombie>' and db_uid is not None: return
if not decode_credentials: if not decode_credentials:
credential_decoder = lambda cred: cred credential_decoder = lambda cred: cred
@@ -7035,8 +7043,8 @@ def index():
if not isinstance(tablename,str): if not isinstance(tablename,str):
raise SyntaxError, "missing table name" raise SyntaxError, "missing table name"
elif hasattr(self,tablename) or tablename in self.tables: elif hasattr(self,tablename) or tablename in self.tables:
pass if not args.get('redefine',False):
# raise SyntaxError, 'table may be already defined: %s' % tablename raise SyntaxError, 'table already defined: %s' % tablename
elif tablename.startswith('_') or hasattr(self,tablename) or \ elif tablename.startswith('_') or hasattr(self,tablename) or \
REGEX_PYTHON_KEYWORDS.match(tablename): REGEX_PYTHON_KEYWORDS.match(tablename):
raise SyntaxError, 'invalid table name: %s' % tablename raise SyntaxError, 'invalid table name: %s' % tablename
@@ -7133,7 +7141,7 @@ def index():
if hasattr(self,'_uri'): if hasattr(self,'_uri'):
return '<DAL uri="%s">' % hide_password(str(self._uri)) return '<DAL uri="%s">' % hide_password(str(self._uri))
else: else:
return '<DAL singleton_code="%s">' % self._singleton_code return '<DAL db_uid="%s">' % self._db_uid
def smart_query(self,fields,text): def smart_query(self,fields,text):
return Set(self, smart_query(fields,text)) return Set(self, smart_query(fields,text))
@@ -7152,10 +7160,12 @@ def index():
self._adapter.rollback() self._adapter.rollback()
def close(self): def close(self):
adapter = self._adapter self._adapter.close()
if self._singleton_code in THREAD_LOCAL.db_instances: if self._db_uid in THREAD_LOCAL.db_instances:
del THREAD_LOCAL.db_instances[self._singleton_code] db_group = THREAD_LOCAL.db_instances[self._db_uid]
adapter.close() db_group.remove(self)
if not db_group:
del THREAD_LOCAL.db_instances[self._db_uid]
def executesql(self, query, placeholders=None, as_dict=False, def executesql(self, query, placeholders=None, as_dict=False,
fields=None, colnames=None): fields=None, colnames=None):
@@ -7284,11 +7294,11 @@ def index():
self[tablename].import_from_csv_file( self[tablename].import_from_csv_file(
ifile, id_map, null, unique, id_offset, *args, **kwargs) ifile, id_map, null, unique, id_offset, *args, **kwargs)
def DAL_unpickler(singleton_code): def DAL_unpickler(db_uid):
return DAL('<lazy>',singleton_code=singleton_code) return DAL('<zombie>',db_uid=db_uid)
def DAL_pickler(db): def DAL_pickler(db):
return DAL_unpickler, (db._singleton_code,) return DAL_unpickler, (db._db_uid,)
copy_reg.pickle(DAL, DAL_pickler, DAL_unpickler) copy_reg.pickle(DAL, DAL_pickler, DAL_unpickler)