transactions and concurrency

[dlax]

It seems that transaction code does not work well with concurrent tasks. For instance, with the following test:

def test_concurrency(conn):
    conn.autocommit = True
            
    def fn(value):
        with conn.transaction():
            cur = conn.execute("select %s", (value,))
        return cur
    
    values = range(2)                                                               
    with concurrent.futures.ThreadPoolExecutor() as e:
        cursors = e.map(fn, values)
    assert sum(cur.fetchone()[0] for cur in cursors) == sum(values)

I get:

tests/test_transaction.py::test_concurrency FAILED                               [100%]

======================================= FAILURES =======================================
___________________________________ test_concurrency ___________________________________

conn = <psycopg.Connection [INTRANS] (host=localhost database=psycopg_test) at 0x7f7366336d00>

    def test_concurrency(conn):
        conn.autocommit = True
    
        def fn(value):
            with conn.transaction():
                cur = conn.execute("select %s", (value,))
            return cur
    
        values = range(2)
        with concurrent.futures.ThreadPoolExecutor() as e:
            cursors = e.map(fn, values)
>       assert sum(cur.fetchone()[0] for cur in cursors) == sum(values)

tests/test_transaction.py:665: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
tests/test_transaction.py:665: in <genexpr>
    assert sum(cur.fetchone()[0] for cur in cursors) == sum(values)
/usr/lib/python3.9/concurrent/futures/_base.py:600: in result_iterator
    yield fs.pop().result()
/usr/lib/python3.9/concurrent/futures/_base.py:433: in result
    return self.__get_result()
/usr/lib/python3.9/concurrent/futures/_base.py:389: in __get_result
    raise self._exception
/usr/lib/python3.9/concurrent/futures/thread.py:52: in run
    result = self.fn(*self.args, **self.kwargs)
tests/test_transaction.py:659: in fn
    cur = conn.execute("select %s", (value,))
/usr/lib/python3.9/contextlib.py:124: in __exit__
    next(self.gen)
psycopg/psycopg/connection.py:844: in transaction
    yield tx
psycopg/psycopg/transaction.py:215: in __exit__
    return self._conn.wait(self._exit_gen(exc_type, exc_val, exc_tb))
psycopg/psycopg/connection.py:867: in wait
    return waiting.wait(gen, self.pgconn.socket, timeout=timeout)
psycopg/psycopg/waiting.py:228: in wait_epoll
    s = next(gen)
psycopg/psycopg/transaction.py:122: in _exit_gen
    yield from self._commit_gen()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <psycopg.Transaction (active) [INTRANS] (host=localhost database=psycopg_test) at 0x7f7366353100>

    def _commit_gen(self) -> PQGen[PGresult]:
>       assert self._conn._savepoints[-1] == self._savepoint_name
E       AssertionError

psycopg/psycopg/transaction.py:138: AssertionError

See dlax@73dfb40 (branch transaction-concurrency in my fork) implementing the tests.

[dvarrazzo]

Uhm... I see, and I am not sure we can do anything about it, apart from throwing a better error message, and blame the user instead of dying of assert 😄

Nested transactions are, by definition, nested. Having two actors racing to create them and exiting them in an order different than the opposite order in which they were entered wasn't considered. We considered that order an invariant, as the assert enforces.

What I would do is probably:

• make the test reliable so that it fails every time (now it fails randomly)
• convert the assert into a ProgrammingError, change the test to expect that
• check if we can protect changing the state of the transaction by a lock, maybe the connection lock, to make sure to manipulate the transaction stack atomically

@asqui if you have any input we'd be happy to hear. Have you ever had to deal with concurrent situations like this in your psycopg2 wrapper?

[asqui]

Documentation

So I went looking at the docs to remind myself what the concurrency guarantees of psycopg3 are but I could not find any mention of threadsafety... is this a gap in the docs or am I just not looking in the right place?

The psycopg2 docs are here for reference, and the TL;DR is "connections are threadsafe, cursors are not". Maybe this distinction becomes a bit more fuzzy given that you can execute SQL directly on a connection in psycopg3 whereas in psycopg2 you cannot.

Concepts

Documentation aside, the code sample above doesn't make much sense to me. The code is saying:

1. Use connection A to execute a query within a transaction.
2. Do 1 concurrently on multiple threads

Given that the transaction is a connection-level concept, how do you expect this to work? The idea just doesn't make sense to me.

This is similar to any other connection-level concept, like session settings. Take time zone for an example:

1. Using connection A, set the session timezone to TZ1, and execute a query (that returns a TIMESTAMPTZ)
2. Using connection A, set the session timezone to TZ2, and execute a query (that returns a TIMESTAMPTZ)
3. Do 1 and 2 concurrently on different threads

What timezone will the output of 1 and 2 be in? It's undefined because it depends on the ordering of events around altering the connection-level concept of time zone the two separate threads.

Implementation

Concepts aside, I agree that the implementation should be robust, and give helpful error messages.
I've not looked at the current state of the code just yet, but from memory, I believe the initial implementation:

1. Had the internal state manipulation protected by a lock
2. Had an elaborate error message for this scenario describing that out-of-order context exit had occurred.

I believe there was also unit test coverage for this scenario (which you can reproduce without multiple threads by just calling __enter__ and __exit__ directly, for two separate transactions, with the exits in the wrong order), and I thought we also converted the assertion to a ProgrammerError on the basis that assertions should be reserved for things that "cannot happen unless there is a bug in the implementation" (though maybe that was for a different error scenario other than this specific one)

TL;DR

Sorry, that's a lot of text with no clear answers...
My main question is @dlax : What did you expect this code to do? I don't think there is any way that it could work correctly unless you put a lock in your fn() to serialise the transactions on concurrent threads. What am I missing?

[dvarrazzo]

@asqui I think there is a single test for out-of-order execution, but I think it has a typo (same transaction names) which didn't make it trigger this assert. Calling them s1, s2 it will. I don't quite understand the comment on the test...

    # Will not (always) catch out-of-order exits
    with conn.transaction(force_rollback=True):
        tx1 = conn.transaction("s1")
        tx2 = conn.transaction("s1")
        tx1.__enter__()
        tx2.__enter__()
        tx1.__exit__(None, None, None)
        tx2.__exit__(None, None, None)

Looking at the git history there were other tests for out-of-order executions but it seems I dropped them in some refactoring.

The concurrency I'd like to guarantee is threadsafe connections. The issue can be triggered using different cursors, it's not related to cursors.

I have changed the test to make it fail reliably: see 153d53d.

I assume @dlax just stumbled on an assert error experimenting with pipeline mode and didn't have any specific expectation :)

It seems, indeed, a case of execution out-of-order, which is something I understand you wanted to guard from since the beginning. I think we can just improve the way we do it with the change proposed in the same commit above, raising ProgrammingError if out-of-order execution is detected.

In case there is an error inside the block, in this case, I am making the "bad transaction nesting" error override the inner error (mostly because it would be an error either on commit and on rollback and to be aggressive in forcing the user to not make it happen).

[dlax]

I'm not really expecting this to work. I was experimenting with another thing without thinking about the meaning of that code too much. Then (as a programmer) an assertion error usually means a bug to me, so I just reported that here. Anyways, I agree that a `ProgrammingError` would be appropriate here.

[dvarrazzo]

Looking at the git history there were other tests for out-of-order executions but it seems I dropped them in some refactoring.

@asqui I found why I did that. Originally you could create an un-initiated transaction. Probably conn.transaction() would have created it, and just using with conn.transaction() would have entered it.

I didn't see an use case for unentered transactions, and saw different modes of failure, so I made conn.transaction() a context manager and to return a transaction already entered.

"so it's impossible to get out-of-order exiting", I must have thought 😄

I'll check what else I have killed which would be better to reinstate. However now it's clearer what was the original intention.

Thank you very much!

[dvarrazzo]

@dvarrazzo: ... check if we can protect changing the state of the transaction by a lock, maybe the connection lock, to make sure to manipulate the transaction stack atomically

@asqui: ... Had the internal state manipulation protected by a lock

yes, the lock is still there. State change is indeed protected by the connection lock and the error only happens because of the order in which enter/exit are executed, not because of a race in critical sections.

... Had an elaborate error message for this scenario describing that out-of-order context exit had occurred.

Found it too: it was

    _out_of_order_err = ProgrammingError(
        "Out-of-order Transaction context exits. Are you "
        "calling __exit__() manually and getting it wrong?"
    )

which is not a condition that can happen anymore unless people use private interfaces (i.e. instantiating Transaction objects manually, which is not part of the api).

So, ProgrammingError will be :)

[dvarrazzo]

Fix ready. will raise a ProgrammingError with a message like:

transactions not correctly nested: <psycopg.AsyncTransaction ...> would roll back in the wrong order compared to the savepoint '_pg3_2'

transactions not correctly nested: <psycopg.Transaction '_pg3_2' ...> would commit in the wrong order compared to the top-level transaction

[asqui]

Hah, as usual, @dvarrazzo has already fixed the problem before I've even had a chance to look at it properly :-)

Some more thoughts/observations:

• "not a condition that can happen anymore unless people use private interfaces" are you counting __exit__ as a private interface that should never be called by a user? I think it is true that the gap between Transaction.__init__ and __enter__ is "inaccessible" to users, but I still think __exit__ is separate and it's not unreasonable/impossible for a user to be calling __exit__ manually. e.g. we do this in test helpers where we need to manage start and end of the transaction separately, not using a with block.
• Related to this question: Would the unit test for this not be a lot simpler written as direct calls to __enter__ and __exit__ (perhaps with a comment stating that this simulates what can happen in a multi-threaded environment)? I know that's a little bit less "realistic" but I think the simplicity is worth it. (For comparison, here is how the test I'm suggesting looks like: 73c8c9f#diff-2079d68e37348367b27f2b657376ca000235adf0cd4739a4213c8c18bb9c1869L618)
• That # Will not (always) catch out-of-order exits block you cited is in a test titled test_named_savepoints_with_repeated_names_works() so is demonstrating that if you manually specify savepoint names, and reuse the same name twice, your nested transactions (savepoints) will work correctly, but detection of out-of-order exits will not. (this is a contra-test to the other tests regarding out-of-order exits that were since removed)

[dvarrazzo]

Hi @asqui :)

I still think exit is separate and it's not unreasonable/impossible for a user to be calling exit manually. e.g. we do this in test helpers where we need to manage start and end of the transaction separately, not using a with block.

Yes, I understand. Your type of test can be probably still organised using blocks, but I see how making sure to throw a reasonable exception makes sense.

And yes, the test would be simpler. What happened is that we started thinking it was a concurrency issue and only later we reasoned that it was "normal" out of order execution. I think it's still good as a demonstration of a scenario in which the issue can happen and it might reveal other types of regressions, such as deadlocks.

if you manually specify savepoint names, and reuse the same name twice, your nested transactions (savepoints) will work correctly, but detection of out-of-order exits will not.

Ah, I see, thank you! But this sounds like an issue to solve. That test there makes it look like it's a wanted behaviour. I would have rather make the test fail (because it doesn't raise) and xfail it.

I think I will:

• add an ooo test of the type I dropped to make sure manual exits are accounted for
• leave the thread/task based there, because I like a good event dance :p
• make ooo detection work with similarly named savepoints (by storing and checking the position in the stack at push/pop).

How does it sound?