Hey there! Let's tackle your problem step by step. The core issue here is that your polling-based reader thread is hogging resources and blocking the writer—using a useless asyncio.sleep(1) is just a band-aid. Instead, we can use asyncio's Condition variable to implement efficient, non-blocking state change notifications, which is exactly what you need.

Why Your Current Setup Fails #

Your reader thread's infinite loop keeps checking the state over and over, which wastes CPU cycles and can block the writer due to GIL contention (since Python threads are limited by the Global Interpreter Lock). The asyncio.sleep(1) lets the event loop breathe temporarily, but it's not a real solution—it introduces unnecessary delay and doesn't guarantee you'll catch state changes immediately.

The Right Approach: Asyncio Condition Variables #

Asyncio's Condition is designed for scenarios where one or more coroutines need to wait until a specific condition is met (like your state changing). Here's how to refactor your Datastruct class to use it properly:

Step 1: Refactor the Datastruct Class #

We'll add a Condition instance to handle state change notifications, and modify the update/wait logic to use it:

import asyncio

class Datastruct:
    def __init__(self):
        self.elements = {}
        # Create a condition variable to manage state change notifications
        self._state_condition = asyncio.Condition()

    async def update_state(self, key, value):
        """Update the state and notify all waiting coroutines."""
        # Acquire the condition lock to ensure atomic state modification
        async with self._state_condition:
            self.elements[key] = value
            # Notify all waiting coroutines that the state has changed
            self._state_condition.notify_all()

    async def get_state(self, key):
        """Safely get the current state (with lock to avoid race conditions)."""
        async with self._state_condition:
            return self.elements.get(key)

    async def wait_for_state_change(self, key, initial_state):
        """Wait until the state changes from the initial value, then return the new state."""
        async with self._state_condition:
            # Keep waiting until the state differs from initial
            while self.elements.get(key) == initial_state:
                # Release the lock and wait for a notification
                await self._state_condition.wait()
            # Return the new state once it's changed
            return self.elements.get(key)

Step 2: Rewrite Your Reader Coroutine #

Instead of polling in an infinite loop, the reader will now wait efficiently for state change notifications:

async def reader_coroutine(datastruct, target_key):
    # Get the initial state first (safely with the condition lock)
    initial_state = await datastruct.get_state(target_key)
    print(f"Initial state: {initial_state}")

    while True:
        # Wait until the state changes from the current initial state
        new_state = await datastruct.wait_for_state_change(target_key, initial_state)
        # Execute your action when the state changes
        print(f"State changed! New state: {new_state}")
        # Update the initial state to watch for the next change
        initial_state = new_state

Step 3: Rewrite Your Writer Coroutine (or Bridge Threads) #

If your writer is already an async coroutine (e.g., handling async socket I/O), this is straightforward:

async def writer_coroutine(datastruct, target_key):
    # Simulate receiving new messages from a socket (replace with your actual logic)
    while True:
        new_msg = await receive_socket_message()  # Async socket read
        if new_msg:
            await datastruct.update_state(target_key, new_msg)

If your writer is a synchronous thread (e.g., using blocking socket calls), you can bridge it to the asyncio event loop with asyncio.run_coroutine_threadsafe:

import threading

def sync_writer_thread(datastruct, target_key, loop):
    while True:
        # Blocking socket read (replace with your actual sync logic)
        new_msg = blocking_socket_receive()
        if new_msg:
            # Schedule the async update on the event loop
            asyncio.run_coroutine_threadsafe(
                datastruct.update_state(target_key, new_msg),
                loop
            )

# To start the thread:
loop = asyncio.get_event_loop()
threading.Thread(
    target=sync_writer_thread,
    args=(datastruct, "my_key", loop),
    daemon=True
).start()

Why This Works #

• No More Polling: The reader only wakes up when the state actually changes, so it doesn't waste CPU or block the writer.
• Atomic Operations: The async with self._state_condition blocks ensure that state modifications and checks are atomic, eliminating race conditions.
• Efficient Notifications: notify_all() wakes up all waiting coroutines immediately when the state changes, so you don't miss updates or introduce delays.

Putting It All Together #

To run everything in the asyncio event loop:

async def main():
    datastruct = Datastruct()
    # Start reader and writer coroutines
    await asyncio.gather(
        reader_coroutine(datastruct, "my_key"),
        writer_coroutine(datastruct, "my_key")
    )

if __name__ == "__main__":
    asyncio.run(main())

内容的提问来源于stack exchange，提问作者deadlock