如何处理异步操作中的竞态条件

import asyncio
import threading

class AsyncLock:
    def __init__(self):
        self._lock = threading.Lock()
        self._waiters = []

    async def acquire(self):
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        with self._lock:
            if not self._waiters:
                self._waiters.append(future)
                return
            self._waiters.append(future)
        await future

    def release(self):
        with self._lock:
            if self._waiters:
                future = self._waiters.pop(0)
                future.set_result(None)

lock = AsyncLock()

async def critical_section(task_id):
    await lock.acquire()
    try:
        print(f"Task {task_id}: Entered critical section")
        await asyncio.sleep(1) # 模拟耗时操作
        print(f"Task {task_id}: Exiting critical section")
    finally:
        lock.release()

async def main():
    tasks = [asyncio.create_task(critical_section(i)) for i in range(3)]
    await asyncio.gather(*tasks)

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

import asyncio
import atomic

counter = atomic.AtomicCounter(0)

async def increment_counter(task_id):
    for _ in range(1000):
        counter.inc()
        #await asyncio.sleep(0) # 模拟并发，去掉注释可能更容易观察到竞态条件（如果没用原子操作）
    print(f"Task {task_id}: Counter incremented")

async def main():
    tasks = [asyncio.create_task(increment_counter(i)) for i in range(5)]
    await asyncio.gather(*tasks)
    print(f"Final counter value: {counter.value}")

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

import asyncio
import queue

message_queue = queue.Queue()

async def producer(task_id):
    for i in range(10):
        message = f"Message from task {task_id}: {i}"
        message_queue.put(message)
        print(f"Task {task_id}: Produced message {message}")
        await asyncio.sleep(0.1)

async def consumer():
    while True:
        message = message_queue.get()
        print(f"Consumer: Received message {message}")
        message_queue.task_done() # Important! Indicate that a formerly enqueued task is complete
        await asyncio.sleep(0.2)

async def main():
    tasks = [asyncio.create_task(producer(i)) for i in range(3)]
    consumer_task = asyncio.create_task(consumer())
    await asyncio.gather(*tasks)
    await message_queue.join() # Wait until all items in the queue have been gotten and processed
    consumer_task.cancel()

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

async def transfer_funds(from_account, to_account, amount):
    try:
        # Start transaction (hypothetical)
        await start_transaction()

        # Debit from_account
        await debit_account(from_account, amount)

        # Credit to_account
        await credit_account(to_account, amount)

        # Commit transaction (hypothetical)
        await commit_transaction()
        print(f"Successfully transferred {amount} from {from_account} to {to_account}")

    except Exception as e:
        # Rollback transaction (hypothetical)
        await rollback_transaction()
        print(f"Transaction failed: {e}")

# (Simplified example - in real-world scenarios, this would typically be implemented with a database)
import asyncio

class DataStore:
    def __init__(self, initial_value):
        self.value = initial_value
        self.version = 0

    async def update(self, updater):
        original_value = self.value
        original_version = self.version
        new_value = updater(original_value)

        # Simulate checking for updates (in a real system, this would involve checking a version number in the database)
        await asyncio.sleep(0.1) # Simulate latency

        if self.version != original_version:
            print("Conflict detected! Retrying...")
            return False # Indicate failure to update

        self.value = new_value
        self.version += 1
        print(f"Updated value to {self.value}, version {self.version}")
        return True # Indicate successful update


data_store = DataStore(100)

async def task(task_id):
    success = False
    while not success:
        success = await data_store.update(lambda x: x + 1)
    print(f"Task {task_id} completed")


async def main():
    tasks = [asyncio.create_task(task(i)) for i in range(3)]
    await asyncio.gather(*tasks)

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

from dataclasses import dataclass

@dataclass(frozen=True)
class ImmutableData:
    value: int

async def modify_data(data, task_id):
    new_data = ImmutableData(data.value + 1)
    print(f"Task {task_id}: Modified data to {new_data}")
    return new_data

async def main():
    data = ImmutableData(10)
    tasks = [asyncio.create_task(modify_data(data, i)) for i in range(3)]
    results = await asyncio.gather(*tasks)

    # Note: Each task creates a *new* ImmutableData instance. The original 'data' remains unchanged.
    print(f"Original data: {data}")
    for i, result in enumerate(results):
        print(f"Task {i} result: {result}")

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