Environment: SQLAlchemy 0.9.7, SQLite back-end, autoflush=True, mostly
using scoped session, multi-platform Python 2.7
I'm working on a multi-threaded app, but with fairly low concurrency (user
actions and scheduled tasks are on different threads, for example). I think
I have a fair understanding of SQLite's locking model, the implications of
having autoflush=True, etc. but I still occasionally get the
"OperationError database is locked".
The challenge with this error is that it is raised on a thread that could
be doing everything right while some other thread is keeping a transaction
open for too long. Plus, unless another thread tries to write, the
"misbehaving" thread goes mostly unnoticed.
So I thought I'd try to write a "long transaction detector" that would
start a timer when when a flush happens, and cancel it if a commit or
rollback happens. If the timer expires (after a few seconds), a warning is
logged with the stack trace of the flush. The idea is that I can see
information about the misbehaving thread rather than the innocent victim,
and I'll see problems even without actual contention.
While I'm still experimenting to see if this strategy will work, it looks
like I'm getting some false positives. It seems like sometimes I get an
after_flush event even when there are no modifications to the database.
This is based on having the engine logging turned on so I can see the SQL
being emitted. I'll see something like:
BEGIN (implicit)
SELECT
SELECT
...
after_flush event
SELECT
SELECT
...
My timer expires
Code is below -- the class is passed as the class_ parameter to
sessionmaker.
Am I doing something dumb? Is this a reasonable strategy for my goals? Any
ideas on the false positive?
Thanks,
Eric
class LongTransactionDetector(sqlalchemy.orm.Session):
TIMEOUT = 3
def __init__(self, *args, **kwargs):
super(LongTransactionDetector, self).__init__(*args, **kwargs)
self._timer = None
self._stack = None
self._flush_time = None
event.listen(self, 'after_flush', self._after_flush)
event.listen(self, 'after_commit', self._after_commit)
event.listen(self, 'after_rollback', self._after_rollback)
event.listen(self, 'after_soft_rollback', self._after_soft_rollback)
def close(self):
# Calling close on a session automatically emits a ROLLBACK, but we
# don't seem to get an event for it, so we'll just hook it here.
super(LongTransactionDetector, self).close()
self._stop_timing()
def _after_flush(self, session, flush_context):
if any([session.new, session.dirty, session.deleted]):
self._begin_timing()
def _after_commit(self, session):
self._stop_timing()
def _after_rollback(self, session):
self._stop_timing()
def _after_soft_rollback(self, session, previous_transaction):
self._stop_timing()
def _begin_timing(self):
if self._timer is None:
logger.debug('set transaction timer')
self._flush_time = datetime.datetime.now()
self._stack = traceback.extract_stack()
self._thread = threading.current_thread()
self._timer = threading.Timer(self.TIMEOUT, self._on_timer)
self._timer.start()
def _stop_timing(self):
if self._timer:
logger.debug('clear transaction timer')
self._timer.cancel()
self._timer = None
self._stack = None
def _on_timer(self):
trace = ''.join(traceback.format_list(self._stack))
logger.warning('long transaction detected on {!r} {}\n{}\n{}'
.format(self._thread, self._flush_time, trace))
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to sqlalchemy+unsubscr...@googlegroups.com.
To post to this group, send email to sqlalchemy@googlegroups.com.
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
