Postgres: Cannot perform operation: another operation is in progress #22
Comments
|
@kellen did you find a solution for this? |
|
@AntonOfTheWoods no, I added redis as the backend instead. |
pwoolvett
added a commit
to pwoolvett/broadcaster
that referenced
this issue
May 6, 2022
This fix adds a lock (asyncio.Event based) to avoid `asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress` fixes encode#22
Open
pwoolvett
added a commit
to pwoolvett/broadcaster
that referenced
this issue
May 13, 2022
This fix adds an asyncio.Lock to avoid `asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress` fixes encode#22
pwoolvett
added a commit
to pwoolvett/broadcaster
that referenced
this issue
May 13, 2022
This fix adds an asyncio.Lock to avoid `asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress` fixes encode#22
|
@AntonOfTheWoods @app.websocket("/viewers", name="viewers_ws")
async def events_ws(websocket: WebSocket):
await websocket.accept()
await run_until_first_complete(
(viewers_ws_receiver, {"websocket": websocket}),
(viewers_ws_sender, {"websocket": websocket}),
)
my_lock = threading.Lock()
async def viewers_ws_receiver(websocket: WebSocket):
await asyncio.sleep(0.01) # <-- pass error
async for message in websocket.iter_text():
await broadcast.publish(channel="viewers", message=message)
async def viewers_ws_sender(websocket: WebSocket):
async with broadcast.subscribe(channel="viewers") as subscriber:
async for event in subscriber:
counter = 0
with my_lock:
# do something with event.message
counter = ...
await websocket.send_json({"viewers": counter}) |
The problem with your solution is that it's unreliable. If, for some reason, the connection is slow, you'll end up with this error anyway. |
|
It does not reproduce for me using the code above. |
just realised I'm now I'm getting the error while publishing an event. I'll try to come up with a minimal example here's an example: import asyncio
import threading
import pytest
from broadcaster import Broadcast
from app.core.config import settings
@pytest.mark.asyncio
async def test_broadcaster():
# broadcast = Broadcast(str(settings.REDIS_URL)) # -> this works
broadcast = Broadcast(str(settings.SQLALCHEMY_DATABASE_URI).replace("+psycopg", "")) # -> this doesn't work
await broadcast.connect()
# Run multiple receivers and senders in parallel
await asyncio.gather(
*[viewers_ws_receiver(broadcast) for _ in range(2)], # 5 parallel receivers
*[viewers_ws_sender(broadcast) for _ in range(2)], # 5 parallel senders
)
my_lock = threading.Lock()
async def viewers_ws_sender(broadcast: Broadcast):
await asyncio.sleep(0.01)
while True:
await broadcast.publish(channel="viewers", message="message")
await asyncio.sleep(1)
async def viewers_ws_receiver(broadcast: Broadcast):
async with broadcast.subscribe(channel="viewers") as subscriber:
async for event in subscriber:
print(event) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Using broadcaster w/ fastapi and seeing an exception when using broadcaster via websockets.
Relevant parts:
Full trace:
Doing this via code that looks something like this, where I think the lock is just slowing things down and exposing the contention.
Update: Refactored down to a single websocket and not using a lock for anything and saw this exception again. 🤷
The text was updated successfully, but these errors were encountered: