January 2011 - calendarserver-changes

[6835] CalendarServer/trunk/contrib/performance/httpclient.py
by source_changes＠macosforge.org 31 Jan '11

31 Jan '11

Revision: 6835 http://trac.macosforge.org/projects/calendarserver/changeset/6835 Author: exarkun(a)twistedmatrix.com Date: 2011-01-31 12:23:00 -0800 (Mon, 31 Jan 2011) Log Message: ----------- Add the MemoryConsumer used by the unit tests to ensure the right data is sent in the request body. Modified Paths: -------------- CalendarServer/trunk/contrib/performance/httpclient.py Modified: CalendarServer/trunk/contrib/performance/httpclient.py =================================================================== --- CalendarServer/trunk/contrib/performance/httpclient.py 2011-01-29 01:40:57 UTC (rev 6834) +++ CalendarServer/trunk/contrib/performance/httpclient.py 2011-01-31 20:23:00 UTC (rev 6835) @@ -21,7 +21,9 @@ from twisted.internet.protocol import Protocol from twisted.internet.defer import Deferred, succeed from twisted.web.iweb import IBodyProducer +from twisted.internet.interfaces import IConsumer + class _BufferReader(Protocol): def __init__(self, finished): self.finished = finished @@ -60,3 +62,21 @@ def startProducing(self, consumer): consumer.write(self._body) return succeed(None) + + + +class MemoryConsumer(object): + implements(IConsumer) + + def __init__(self): + self._buffer = [] + + + def write(self, bytes): + self._buffer.append(bytes) + + + def value(self): + if len(self._buffer) > 1: + self._buffer = ["".join(self._buffer)] + return "".join(self._buffer)

1 0

[6834] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6834 http://trac.macosforge.org/projects/calendarserver/changeset/6834 Author: glyph(a)apple.com Date: 2011-01-28 17:40:57 -0800 (Fri, 28 Jan 2011) Log Message: ----------- Clean up FIXMEs, doc bugs. Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:46 UTC (rev 6833) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:57 UTC (rev 6834) @@ -53,8 +53,9 @@ from twext.enterprise.ienterprise import AlreadyFinishedError, IAsyncTransaction -# FIXME: there should be no default, it should be discovered dynamically -# everywhere. Right now we're only using pgdb so we only support that. +# FIXME: there should be no default for DEFAULT_PARAM_STYLE, it should be +# discovered dynamically everywhere. Right now we're only using pgdb so we only +# support that. DEFAULT_PARAM_STYLE = 'pyformat' @@ -65,7 +66,7 @@ """ implements(IAsyncTransaction) - # FIXME: this should *really* be + # See DEFAULT_PARAM_STYLE FIXME above. paramstyle = DEFAULT_PARAM_STYLE def __init__(self, threadHolder, connection, cursor): @@ -178,9 +179,7 @@ implements(IAsyncTransaction) - # FIXME: this should be relayed from the connection factory of the thing - # creating the spooled transaction. - + # See DEFAULT_PARAM_STYLE FIXME above. paramstyle = DEFAULT_PARAM_STYLE def __init__(self): @@ -409,7 +408,6 @@ """ holder = self._createHolder() holder.start() - # FIXME: attach the holder to the txn so it can be aborted. txn = _ConnectingPsuedoTxn(self, holder) # take up a slot in the 'busy' list, sit there so we can be aborted. self.busy.append(txn) @@ -428,9 +426,6 @@ txn._baseTxn = baseTxn self.reclaim(txn) def maybeTryAgain(f): - # not all errors are errors in connect() - in particular there's the - # possibility of errors in submit(); TODO: test for shutdown while - # connecting. log.err(f) txn._retry = self.reactor.callLater(self.RETRY_TIMEOUT, resubmit) def resubmit(): @@ -443,7 +438,7 @@ def reclaim(self, txn): """ Shuck the L{PooledSqlTxn} wrapper off, and recycle the underlying - BaseSqlTxn into the free list. + L{BaseSqlTxn} into the free list. """ self.busy.remove(txn) baseTxn = txn._baseTxn @@ -548,7 +543,7 @@ class QueryComplete(Command): """ - A query issued with ExecSQL is complete. + A query issued with L{ExecSQL} is complete. """ arguments = [('queryID', String()), @@ -700,8 +695,7 @@ implements(IAsyncTransaction) - # FIXME: this needs to come from the other end of the wire. - + # See DEFAULT_PARAM_STYLE FIXME above. paramstyle = DEFAULT_PARAM_STYLE def __init__(self, client, transactionID): Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:46 UTC (rev 6833) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:57 UTC (rev 6834) @@ -32,6 +32,9 @@ class Child(object): + """ + An object with a L{Parent}, in its list of C{children}. + """ def __init__(self, parent): self.parent = parent self.parent.children.append(self) @@ -42,6 +45,9 @@ class Parent(object): + """ + An object with a list of L{Child}ren. + """ def __init__(self): self.children = [] @@ -115,7 +121,6 @@ class ConnectionFactory(Parent): - def __init__(self): Parent.__init__(self) self.idcounter = count(1) @@ -414,7 +419,6 @@ [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) - # FIXME: next, 'failed' case. def test_shutdownDuringAttemptFailed(self): @@ -436,6 +440,6 @@ [holder] = self.holders self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) - # FIXME: next, 'failed' case. +

1 0

[6833] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/ test/test_adbapi2.py
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6833 http://trac.macosforge.org/projects/calendarserver/changeset/6833 Author: glyph(a)apple.com Date: 2011-01-28 17:40:46 -0800 (Fri, 28 Jan 2011) Log Message: ----------- Oh, okay, this already works. Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:36 UTC (rev 6832) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:46 UTC (rev 6833) @@ -256,9 +256,18 @@ self.pool._createHolder = self.makeAHolder self.clock = self.pool.reactor = Clock() self.pool.startService() - self.addCleanup(self.pool.stopService) + @inlineCallbacks + def tearDown(self): + """ + Make sure the service is stopped and the fake ThreadHolders are all + executing their queues so failed tests can exit cleanly. + """ + self.flushHolders() + yield self.pool.stopService() + + def flushHolders(self): """ Flush all pending C{submit}s since C{pauseHolders} was called. @@ -393,7 +402,7 @@ """ If L{ConnectionPool.stopService} is called while a connection attempt is outstanding, the resulting L{Deferred} won't be fired until the - connection attempt has succeeded. + connection attempt has finished; in this case, succeeded. """ self.pauseHolders() self.pool.connection() @@ -408,3 +417,25 @@ # FIXME: next, 'failed' case. + def test_shutdownDuringAttemptFailed(self): + """ + If L{ConnectionPool.stopService} is called while a connection attempt is + outstanding, the resulting L{Deferred} won't be fired until the + connection attempt has finished; in this case, failed. + """ + self.factory.defaultFail() + self.pauseHolders() + self.pool.connection() + stopd = [] + self.pool.stopService().addBoth(stopd.append) + self.assertEquals(stopd, []) + self.flushHolders() + errors = self.flushLoggedErrors(FakeConnectionError) + self.assertEquals(len(errors), 1) + self.assertEquals(stopd, [None]) + [holder] = self.holders + self.assertEquals(holder.started, True) + self.assertEquals(holder.stopped, True) + # FIXME: next, 'failed' case. + +

1 0

[6832] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6832 http://trac.macosforge.org/projects/calendarserver/changeset/6832 Author: glyph(a)apple.com Date: 2011-01-28 17:40:36 -0800 (Fri, 28 Jan 2011) Log Message: ----------- Small cleanups; test that lists are emptied out, empty them out as we go along rather than all at once, get rid of no-longer-used connectionFactory attribute on BaseSqlTxn. Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:26 UTC (rev 6831) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:36 UTC (rev 6832) @@ -68,7 +68,7 @@ # FIXME: this should *really* be paramstyle = DEFAULT_PARAM_STYLE - def __init__(self, connectionFactory, threadHolder, connection, cursor): + def __init__(self, threadHolder, connection, cursor): self._completed = True self._cursor = cursor self._connection = connection @@ -354,20 +354,26 @@ @inlineCallbacks def stopService(self): """ - Forcibly abort any outstanding transactions. + Forcibly abort any outstanding transactions, and release all resources + (notably, threads). """ + # Phase 1: All of the busy transactions must be aborted first. As each + # one is aborted, it will remove itself from the list. while self.busy: busy = self.busy[0] try: yield busy.abort() except: log.err() - # all transactions should now be in the free list, since 'abort()' will - # have put them there. - for free in self.free: + # Phase 2: All transactions should now be in the free list, since + # 'abort()' will have put them there. Shut down all the associated + # ThreadHolders. + while self.free: + # (Stopping a BaseSqlTxn doesn't automatically recycle it / remove + # it the way aborting a PooledSqlTxn does, so we need to .pop() + # here.) + free = self.free.pop() yield free.stop() - self.busy = [] - self.free = [] def _createHolder(self): @@ -415,7 +421,6 @@ return (connection, cursor) def finishInit((connection, cursor)): baseTxn = BaseSqlTxn( - connectionFactory=self.connectionFactory, threadHolder=holder, connection=connection, cursor=cursor Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:26 UTC (rev 6831) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:36 UTC (rev 6832) @@ -337,6 +337,8 @@ self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, False) yield self.pool.stopService() + self.assertEquals(self.pool.busy, []) + self.assertEquals(self.pool.free, []) self.assertEquals(len(self.holders), 1) self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True)

1 0

[6831] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6831 http://trac.macosforge.org/projects/calendarserver/changeset/6831 Author: glyph(a)apple.com Date: 2011-01-28 17:40:26 -0800 (Fri, 28 Jan 2011) Log Message: ----------- test for stopService at another point, make sure we don't lose track of the connection attempt in progress Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:16 UTC (rev 6830) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:26 UTC (rev 6831) @@ -293,7 +293,8 @@ _retry = None - def __init__(self, holder): + def __init__(self, pool, holder): + self._pool = pool self._holder = holder def abort(self): @@ -301,8 +302,12 @@ # raise NotImplementedError() if self._retry is not None: self._retry.cancel() - # deferred, should be returned - self._holder.stop() + d = self._holder.stop() + def removeme(ignored): + if self in self._pool.busy: + self._pool.busy.remove(self) + d.addCallback(removeme) + return d @@ -351,7 +356,8 @@ """ Forcibly abort any outstanding transactions. """ - for busy in self.busy[:]: + while self.busy: + busy = self.busy[0] try: yield busy.abort() except: @@ -398,16 +404,12 @@ holder = self._createHolder() holder.start() # FIXME: attach the holder to the txn so it can be aborted. - txn = _ConnectingPsuedoTxn(holder) + txn = _ConnectingPsuedoTxn(self, holder) # take up a slot in the 'busy' list, sit there so we can be aborted. self.busy.append(txn) def initCursor(): # support threadlevel=1; we can't necessarily cursor() in a # different thread than we do transactions in. - - # TODO: Re-try connect when it fails. Specify a timeout. That - # should happen in this layer because we need to be able to stop - # the reconnect attempt if it's hanging. connection = self.connectionFactory() cursor = connection.cursor() return (connection, cursor) Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:16 UTC (rev 6830) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:26 UTC (rev 6831) @@ -27,6 +27,7 @@ from twisted.internet.defer import execute from twisted.internet.task import Clock +from twisted.internet.defer import Deferred from twext.enterprise.adbapi2 import ConnectionPool @@ -188,9 +189,11 @@ execution is easier to control. """ - def __init__(self): + def __init__(self, test): self.started = False self.stopped = False + self.test = test + self.queue = [] def start(self): @@ -204,17 +207,38 @@ """ Mark this L{FakeThreadHolder} as stopped. """ - self.stopped = True + def stopped(nothing): + self.stopped = True + return self.submit(lambda : None).addCallback(stopped) def submit(self, work): """ - Call the function. + Call the function (or queue it) """ - return execute(work) + if self.test.paused: + d = Deferred() + self.queue.append((d, work)) + return d + else: + return execute(work) + def flush(self): + """ + Fire all deferreds previously returned from submit. + """ + self.queue, queue = [], self.queue + for (d, work) in queue: + try: + result = work() + except: + d.errback() + else: + d.callback(result) + + class ConnectionPoolTests(TestCase): """ Tests for L{ConnectionPool}. @@ -225,6 +249,7 @@ Create a L{ConnectionPool} attached to a C{ConnectionFactory}. Start the L{ConnectionPool}. """ + self.paused = False self.holders = [] self.factory = ConnectionFactory() self.pool = ConnectionPool(self.factory.connect, maxConnections=2) @@ -234,11 +259,28 @@ self.addCleanup(self.pool.stopService) + def flushHolders(self): + """ + Flush all pending C{submit}s since C{pauseHolders} was called. + """ + self.paused = False + for holder in self.holders: + holder.flush() + + + def pauseHolders(self): + """ + Pause all L{FakeThreadHolder}s, causing C{submit} to return an unfired + L{Deferred}. + """ + self.paused = True + + def makeAHolder(self): """ Make a ThreadHolder-alike. """ - fth = FakeThreadHolder() + fth = FakeThreadHolder(self) self.holders.append(fth) return fth @@ -344,3 +386,23 @@ self.assertEquals(holder.started, True) self.assertEquals(holder.stopped, True) + + def test_shutdownDuringAttemptSuccess(self): + """ + If L{ConnectionPool.stopService} is called while a connection attempt is + outstanding, the resulting L{Deferred} won't be fired until the + connection attempt has succeeded. + """ + self.pauseHolders() + self.pool.connection() + stopd = [] + self.pool.stopService().addBoth(stopd.append) + self.assertEquals(stopd, []) + self.flushHolders() + self.assertEquals(stopd, [None]) + [holder] = self.holders + self.assertEquals(holder.started, True) + self.assertEquals(holder.stopped, True) + # FIXME: next, 'failed' case. + +

1 0

[6830] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6830 http://trac.macosforge.org/projects/calendarserver/changeset/6830 Author: glyph(a)apple.com Date: 2011-01-28 17:40:16 -0800 (Fri, 28 Jan 2011) Log Message: ----------- shut down while waiting for retry Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:05 UTC (rev 6829) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:16 UTC (rev 6830) @@ -291,14 +291,18 @@ class _ConnectingPsuedoTxn(object): - def __init__(self): - pass + _retry = None + def __init__(self, holder): + self._holder = holder def abort(self): # not implemented yet, but let's fail rather than break the test # raise NotImplementedError() - pass + if self._retry is not None: + self._retry.cancel() + # deferred, should be returned + self._holder.stop() @@ -394,7 +398,7 @@ holder = self._createHolder() holder.start() # FIXME: attach the holder to the txn so it can be aborted. - txn = _ConnectingPsuedoTxn() + txn = _ConnectingPsuedoTxn(holder) # take up a slot in the 'busy' list, sit there so we can be aborted. self.busy.append(txn) def initCursor(): @@ -421,7 +425,7 @@ # possibility of errors in submit(); TODO: test for shutdown while # connecting. log.err(f) - self.reactor.callLater(self.RETRY_TIMEOUT, resubmit) + txn._retry = self.reactor.callLater(self.RETRY_TIMEOUT, resubmit) def resubmit(): d = holder.submit(initCursor) d.addCallbacks(finishInit, maybeTryAgain) Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:05 UTC (rev 6829) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:40:16 UTC (rev 6830) @@ -325,3 +325,22 @@ self.assertEquals(happened, [[[1, "alpha"]]]) + def test_shutdownDuringRetry(self): + """ + If a L{ConnectionPool} is attempting to shut down while it's in the + process of re-trying a connection attempt that received an error, the + connection attempt should be cancelled and the shutdown should complete + as normal. + """ + self.factory.defaultFail() + self.pool.connection() + errors = self.flushLoggedErrors(FakeConnectionError) + self.assertEquals(len(errors), 1) + stopd = [] + self.pool.stopService().addBoth(stopd.append) + self.assertEquals([None], stopd) + self.assertEquals(self.clock.calls, []) + [holder] = self.holders + self.assertEquals(holder.started, True) + self.assertEquals(holder.stopped, True) +

1 0

[6829] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/ adbapi2.py
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6829 http://trac.macosforge.org/projects/calendarserver/changeset/6829 Author: glyph(a)apple.com Date: 2011-01-28 17:40:05 -0800 (Fri, 28 Jan 2011) Log Message: ----------- retry (too aggressively) Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:39:55 UTC (rev 6828) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:40:05 UTC (rev 6829) @@ -407,7 +407,6 @@ connection = self.connectionFactory() cursor = connection.cursor() return (connection, cursor) - d = holder.submit(initCursor) def finishInit((connection, cursor)): baseTxn = BaseSqlTxn( connectionFactory=self.connectionFactory, @@ -417,7 +416,16 @@ ) txn._baseTxn = baseTxn self.reclaim(txn) - d.addCallbacks(finishInit, log.err) + def maybeTryAgain(f): + # not all errors are errors in connect() - in particular there's the + # possibility of errors in submit(); TODO: test for shutdown while + # connecting. + log.err(f) + self.reactor.callLater(self.RETRY_TIMEOUT, resubmit) + def resubmit(): + d = holder.submit(initCursor) + d.addCallbacks(finishInit, maybeTryAgain) + resubmit() return txn

1 0

[6828] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/ test/test_adbapi2.py
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6828 http://trac.macosforge.org/projects/calendarserver/changeset/6828 Author: glyph(a)apple.com Date: 2011-01-28 17:39:55 -0800 (Fri, 28 Jan 2011) Log Message: ----------- correct the test Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:39:45 UTC (rev 6827) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:39:55 UTC (rev 6828) @@ -318,13 +318,10 @@ happened = [] d.addBoth(happened.append) self.assertEquals(happened, []) - self.clock.advance(self.pool.RETRY_TIMEOUT) + self.clock.advance(self.pool.RETRY_TIMEOUT + 0.01) checkOneFailure() self.assertEquals(happened, []) - self.clock.advance(self.pool.RETRY_TIMEOUT) - checkOneFailure() - self.assertEquals(happened, []) - self.clock.advance(self.pool.RETRY_TIMEOUT) - self.assertEquals(happened, [[1, "alpha"]]) + self.clock.advance(self.pool.RETRY_TIMEOUT + 0.01) + self.assertEquals(happened, [[[1, "alpha"]]])

1 0

[6827] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/ adbapi2.py
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6827 http://trac.macosforge.org/projects/calendarserver/changeset/6827 Author: glyph(a)apple.com Date: 2011-01-28 17:39:45 -0800 (Fri, 28 Jan 2011) Log Message: ----------- starting to move logic around Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:39:35 UTC (rev 6826) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:39:45 UTC (rev 6827) @@ -68,28 +68,13 @@ # FIXME: this should *really* be paramstyle = DEFAULT_PARAM_STYLE - def __init__(self, connectionFactory, threadHolder): - self._completed = False - self._cursor = None + def __init__(self, connectionFactory, threadHolder, connection, cursor): + self._completed = True + self._cursor = cursor + self._connection = connection self._holder = threadHolder - self._holder.start() - def initCursor(): - # support threadlevel=1; we can't necessarily cursor() in a - # different thread than we do transactions in. - # TODO: Re-try connect when it fails. Specify a timeout. That - # should happen in this layer because we need to be able to stop - # the reconnect attempt if it's hanging. - self._connection = connectionFactory() - self._cursor = self._connection.cursor() - - # Note: no locking necessary here; since this gets submitted first, all - # subsequent submitted work-units will be in line behind it and the - # cursor will already have been initialized. - self._holder.submit(initCursor).addErrback(log.err) - - def _reallyExecSQL(self, sql, args=None, raiseOnZeroRowCount=None): if args is None: args = [] @@ -304,6 +289,19 @@ +class _ConnectingPsuedoTxn(object): + + def __init__(self): + pass + + + def abort(self): + # not implemented yet, but let's fail rather than break the test + # raise NotImplementedError() + pass + + + class ConnectionPool(Service, object): """ This is a central service that has a threadpool and executes SQL statements @@ -379,27 +377,58 @@ tracking = self.busy if self.free: basetxn = self.free.pop(0) - elif len(self.busy) < self.maxConnections: - basetxn = BaseSqlTxn( - connectionFactory=self.connectionFactory, - threadHolder=self._createHolder() - ) else: basetxn = SpooledTxn() tracking = self.waiting txn = PooledSqlTxn(self, basetxn) tracking.append(txn) + if tracking is self.waiting and len(self.busy) < self.maxConnections: + self._startOneMore() return txn + def _startOneMore(self): + """ + Start one more BaseSqlTxn. + """ + holder = self._createHolder() + holder.start() + # FIXME: attach the holder to the txn so it can be aborted. + txn = _ConnectingPsuedoTxn() + # take up a slot in the 'busy' list, sit there so we can be aborted. + self.busy.append(txn) + def initCursor(): + # support threadlevel=1; we can't necessarily cursor() in a + # different thread than we do transactions in. + + # TODO: Re-try connect when it fails. Specify a timeout. That + # should happen in this layer because we need to be able to stop + # the reconnect attempt if it's hanging. + connection = self.connectionFactory() + cursor = connection.cursor() + return (connection, cursor) + d = holder.submit(initCursor) + def finishInit((connection, cursor)): + baseTxn = BaseSqlTxn( + connectionFactory=self.connectionFactory, + threadHolder=holder, + connection=connection, + cursor=cursor + ) + txn._baseTxn = baseTxn + self.reclaim(txn) + d.addCallbacks(finishInit, log.err) + return txn + + def reclaim(self, txn): """ Shuck the L{PooledSqlTxn} wrapper off, and recycle the underlying BaseSqlTxn into the free list. """ + self.busy.remove(txn) baseTxn = txn._baseTxn baseTxn.reset() - self.busy.remove(txn) if self.waiting: waiting = self.waiting.pop(0) waiting._baseTxn._unspool(baseTxn)

1 0

[6826] CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise
by source_changes＠macosforge.org 29 Jan '11

29 Jan '11

Revision: 6826 http://trac.macosforge.org/projects/calendarserver/changeset/6826 Author: glyph(a)apple.com Date: 2011-01-28 17:39:35 -0800 (Fri, 28 Jan 2011) Log Message: ----------- improve the test a little more Modified Paths: -------------- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:39:23 UTC (rev 6825) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/adbapi2.py 2011-01-29 01:39:35 UTC (rev 6826) @@ -1,4 +1,4 @@ -# -*- test-case-name: twext.enterprise.test. -*- +# -*- test-case-name: twext.enterprise.test.test_adbapi2 -*- ## # Copyright (c) 2010 Apple Inc. All rights reserved. # @@ -69,10 +69,6 @@ paramstyle = DEFAULT_PARAM_STYLE def __init__(self, connectionFactory, threadHolder): - """ - @param connectionFactory: A 0-argument callable which returns a DB-API - 2.0 connection. - """ self._completed = False self._cursor = None self._holder = threadHolder @@ -129,6 +125,8 @@ if not self._completed: self._completed = True def reallyCommit(): + if self._cursor is None: + return self._connection.commit() result = self._holder.submit(reallyCommit) return result @@ -140,6 +138,8 @@ if not self._completed: self._completed = True def reallyAbort(): + if self._cursor is None: + return self._connection.rollback() result = self._holder.submit(reallyAbort) return result @@ -174,7 +174,11 @@ self._stopped = True holder = self._holder self._holder = None - holder.submit(self._connection.close) + def _reallyClose(): + if self._cursor is None: + return + self._connection.close() + holder.submit(_reallyClose) return holder.stop() @@ -313,9 +317,18 @@ than this many concurrent connections to the database. @type maxConnections: C{int} + + @ivar reactor: The reactor used for scheduling threads as well as retries + for failed connect() attempts. + + @type reactor: L{IReactorTime} and L{IReactorThreads} provider. """ + reactor = _reactor + RETRY_TIMEOUT = 10.0 + + def __init__(self, connectionFactory, maxConnections=10): super(ConnectionPool, self).__init__() self.free = [] @@ -353,7 +366,7 @@ """ Create a L{ThreadHolder}. (Test hook.) """ - return ThreadHolder(_reactor) + return ThreadHolder(self.reactor) def connection(self, label="<unlabeled>"): @@ -363,7 +376,7 @@ @return: an L{IAsyncTransaction} """ - overload = False + tracking = self.busy if self.free: basetxn = self.free.pop(0) elif len(self.busy) < self.maxConnections: @@ -373,12 +386,9 @@ ) else: basetxn = SpooledTxn() - overload = True + tracking = self.waiting txn = PooledSqlTxn(self, basetxn) - if overload: - self.waiting.append(txn) - else: - self.busy.append(txn) + tracking.append(txn) return txn Modified: CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py =================================================================== --- CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:39:23 UTC (rev 6825) +++ CalendarServer/branches/users/glyph/db-reconnect/twext/enterprise/test/test_adbapi2.py 2011-01-29 01:39:35 UTC (rev 6826) @@ -25,6 +25,8 @@ from twisted.internet.defer import inlineCallbacks from twisted.internet.defer import execute +from twisted.internet.task import Clock + from twext.enterprise.adbapi2 import ConnectionPool @@ -227,6 +229,7 @@ self.factory = ConnectionFactory() self.pool = ConnectionPool(self.factory.connect, maxConnections=2) self.pool._createHolder = self.makeAHolder + self.clock = self.pool.reactor = Clock() self.pool.startService() self.addCleanup(self.pool.stopService) @@ -303,13 +306,25 @@ exception, the L{ConnectionPool} will log the exception and delay execution of a new connection's SQL methods until an attempt succeeds. """ - self.factory.defaultFail() + self.factory.willFail() + self.factory.willFail() + self.factory.willConnect() c = self.pool.connection() - errors = self.flushLoggedErrors(FakeConnectionError) - self.assertEquals(len(errors), 1) + def checkOneFailure(): + errors = self.flushLoggedErrors(FakeConnectionError) + self.assertEquals(len(errors), 1) + checkOneFailure() d = c.execSQL("alpha") happened = [] d.addBoth(happened.append) self.assertEquals(happened, []) + self.clock.advance(self.pool.RETRY_TIMEOUT) + checkOneFailure() + self.assertEquals(happened, []) + self.clock.advance(self.pool.RETRY_TIMEOUT) + checkOneFailure() + self.assertEquals(happened, []) + self.clock.advance(self.pool.RETRY_TIMEOUT) + self.assertEquals(happened, [[1, "alpha"]])

1 0