Cannot perform concurrent async queries on same database connection.

[LKay]

Few weeks ago this PR was merged #108 and was supposed to fix the issue with asycio.gather executing multiple queries asynchronously. Unfortunately it didn't fix the issue, so I'm recreating the bug report.

When multiple queries are wrapped in asyncio.gather and some also using sync with database.transaction() the exception is thrown:

...
File "/.../lib/python3.7/site-packages/asyncpg/connection.py", line 273, in execute
    return await self._protocol.query(query, timeout)
  File "asyncpg/protocol/protocol.pyx", line 301, in query
  File "asyncpg/protocol/protocol.pyx", line 659, in asyncpg.protocol.protocol.BaseProtocol._check_state
asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress

[lovelydinosaur]

We've got a test case for concurrent queries as part of that test case.

If there's still a remaining issue here, then the best place to start would be to try to work from that test case into a new case that demonstrates exactly how to reproduce the problem.

What's the absolute simplest possible example case you can reduce this down too that still presents the issue? Which database drivers does it present on (and are you able to replicate on others)?

[LKay]

The code is part of whole application, so can't really extract just runnable piece of code easily... I can just show some part that is failing, without much context:

await asyncio.gather(
  products_dao.create(
    Product(...)
  ),
  stock_dao.update(
    stock.copy(update=dict(
      ...
    ))
  )
)

Where mentioned dao's methods are somekind as follow:

    async def create(self, model: Model) -> Model:
        async with self.database.transaction():
            result = await self.database.fetch_one(
                self.table.insert()
                    .return_defaults()
                    .values(dict_to_insert(model.dict()))
            )

            return model.copy(
                update=record_to_dict(result)
            )

    async def update(self, model: Model) -> Model:
        async with self.database.transaction():
            updated_on = await self.database.execute(
                self.table.update()
                    .where(self.id_col == model.id()).values(model.dict())
                    .returning(self.table.c.updated_on)
            )

            return model.copy(
                update=dict(updated_on=updated_on)
            )

[lovelydinosaur]

The code is part of whole application, so can't really extract just runnable piece of code easily

Well okay, but nothings likely to progress very quickly if no-one's stepping up to isolate the issue.

If you can reduce this into an easily replicable example then I'll happily spend some time digging into it.

What I would say tho from looking at the example code given is: don't do that.

Just run the create and update operations sequentially. They won't happen quicker by being on seperate tasks, because they're occuring as part of the same database conenction and the database driver needs to perform them in sequence in any case. (Which is why #108 puts locking in place.)

[zevisert]

When multiple queries are wrapped in asyncio.gather and some also using async with database.transaction() the exception is thrown

Based on the similarity of what you're explaining is to issues like #452, #424, #155, #123, and maybe more that I'm forgetting about. The gist is that you cannot run concurrent transactions on the same connection, otherwise the database backends will error with a message like "another operation is in progress". So to solve this, we need databases to establish independent connections for each asyncio.Task (eg those created by asyncio.gather). That's what I have been working on getting correct in #546. I think that PR might solve this