08 - 同步与并发¶

# Peterson算法 —— 两进程互斥的软件实现
# 共享变量
flag = [False, False]  # flag[i]=True 表示进程i想要进入临界区
turn = 0               # 指示谁有权进入临界区

def process_0():
    """进程0"""
    global turn
    flag[0] = True      # 表示进程0想进入临界区
    turn = 1            # 谦让：让进程1先来
    # 等待条件：进程1也想进入 且 轮到进程1
    while flag[1] and turn == 1:
        pass  # 忙等待
    # ===== 临界区 =====
    # 访问共享资源
    # ===== 退出区 =====
    flag[0] = False     # 进程0不再需要临界区

def process_1():
    """进程1"""
    global turn
    flag[1] = True      # 表示进程1想进入临界区
    turn = 0            # 谦让：让进程0先来
    while flag[0] and turn == 0:
        pass  # 忙等待
    # ===== 临界区 =====
    # 访问共享资源
    # ===== 退出区 =====
    flag[1] = False

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        lock.acquire()     # 加锁
        counter += 1       # 临界区
        lock.release()     # 解锁

# 推荐使用 with 语句（自动加锁/解锁）
def increment_safe():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

# 测试
t1 = threading.Thread(target=increment_safe)
t2 = threading.Thread(target=increment_safe)
t1.start(); t2.start()
t1.join(); t2.join()
print(f"Counter: {counter}")  # 200000  # Counter计数器：统计元素出现次数

import threading

# 二值信号量实现互斥
mutex = threading.Semaphore(1)  # 初始值为1
shared_data = 0

def worker(name):
    global shared_data
    for _ in range(10000):
        mutex.acquire()     # P操作 (wait)
        shared_data += 1    # 临界区
        mutex.release()     # V操作 (signal)

t1 = threading.Thread(target=worker, args=("T1",))
t2 = threading.Thread(target=worker, args=("T2",))
t1.start(); t2.start()
t1.join(); t2.join()
print(f"shared_data = {shared_data}")  # 20000

import threading
import time

# 信号量实现同步：保证A在B之前执行
sync_sem = threading.Semaphore(0)  # 初始值为0

def task_a():
    print("Task A: 完成数据准备")
    time.sleep(1)  # 模拟耗时操作
    print("Task A: 数据准备完成，通知Task B")
    sync_sem.release()  # V操作，信号量+1

def task_b():
    print("Task B: 等待Task A完成...")
    sync_sem.acquire()  # P操作，如果信号量=0则阻塞
    print("Task B: 收到通知，开始处理数据")

t_b = threading.Thread(target=task_b)
t_a = threading.Thread(target=task_a)
t_b.start(); t_a.start()
t_b.join(); t_a.join()

# 输出：
# Task B: 等待Task A完成...
# Task A: 完成数据准备
# Task A: 数据准备完成，通知Task B
# Task B: 收到通知，开始处理数据

import threading
import time

condition = threading.Condition()  # 内部包含一个Lock
items = []

def producer():
    for i in range(5):
        time.sleep(0.5)
        with condition:
            items.append(f"item-{i}")
            print(f"[生产者] 生产了 item-{i}，通知消费者")
            condition.notify()  # 唤醒一个等待的消费者

def consumer():
    while True:
        with condition:
            while not items:    # 使用while而不是if，防止虚假唤醒
                print("[消费者] 没有数据，等待中...")
                condition.wait()  # 释放锁并等待
            item = items.pop(0)
            print(f"[消费者] 消费了 {item}")
        if item == "item-4":
            break

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t2.start(); t1.start()
t1.join(); t2.join()

import threading
import time

class ReadWriteLock:
    """简单的读写锁实现（读者优先）"""
    def __init__(self):
        self.readers = 0
        self.mutex = threading.Lock()       # 保护readers计数
        self.write_lock = threading.Lock()  # 写互斥锁

    def acquire_read(self):
        with self.mutex:
            self.readers += 1
            if self.readers == 1:
                self.write_lock.acquire()  # 第一个读者阻止写者

    def release_read(self):
        with self.mutex:
            self.readers -= 1
            if self.readers == 0:
                self.write_lock.release()  # 最后一个读者释放写锁

    def acquire_write(self):
        self.write_lock.acquire()

    def release_write(self):
        self.write_lock.release()

# 示例使用
rw_lock = ReadWriteLock()
shared_data = {"value": 0}

def reader(name):
    for _ in range(3):
        rw_lock.acquire_read()
        print(f"[读者{name}] 读取: {shared_data['value']}")
        time.sleep(0.1)
        rw_lock.release_read()

def writer(name):
    for i in range(3):
        rw_lock.acquire_write()
        shared_data['value'] += 1
        print(f"[写者{name}] 写入: {shared_data['value']}")
        time.sleep(0.2)
        rw_lock.release_write()

import threading
import queue
import time
import random

buffer = queue.Queue(maxsize=5)

def producer(name):
    for i in range(10):
        item = f"{name}-item-{i}"
        buffer.put(item)
        print(f"[生产者{name}] 生产: {item}, 缓冲区大小: {buffer.qsize()}")
        time.sleep(random.uniform(0.1, 0.5))

def consumer(name):
    while True:
        try:  # try/except捕获异常
            item = buffer.get(timeout=3)
            print(f"[消费者{name}] 消费: {item}, 缓冲区大小: {buffer.qsize()}")
            buffer.task_done()
            time.sleep(random.uniform(0.2, 0.6))
        except queue.Empty:
            print(f"[消费者{name}] 超时退出")
            break

# 启动
threads = []
for i in range(2):
    t = threading.Thread(target=producer, args=(f"P{i}",))
    threads.append(t)
for i in range(3):
    t = threading.Thread(target=consumer, args=(f"C{i}",))
    threads.append(t)
for t in threads:
    t.start()
for t in threads:
    t.join()

import threading
import time
import random

BUFFER_SIZE = 5
buffer = [None] * BUFFER_SIZE
in_idx = 0   # 生产者放入位置
out_idx = 0  # 消费者取出位置

mutex = threading.Semaphore(1)         # 互斥访问缓冲区
empty = threading.Semaphore(BUFFER_SIZE)  # 空槽位数量
full = threading.Semaphore(0)          # 满槽位数量

def producer(name):
    global in_idx
    for i in range(8):
        item = f"{name}-{i}"
        empty.acquire()    # P(empty)：等待空槽位
        mutex.acquire()    # P(mutex)：互斥访问
        buffer[in_idx] = item
        print(f"[生产者{name}] 放入 {item} 到位置 {in_idx}")
        in_idx = (in_idx + 1) % BUFFER_SIZE
        mutex.release()    # V(mutex)
        full.release()     # V(full)：增加一个满槽位
        time.sleep(random.uniform(0.1, 0.3))

def consumer(name):
    global out_idx
    for _ in range(8):
        full.acquire()     # P(full)：等待满槽位
        mutex.acquire()    # P(mutex)：互斥访问
        item = buffer[out_idx]
        print(f"[消费者{name}] 取出 {item} 从位置 {out_idx}")
        out_idx = (out_idx + 1) % BUFFER_SIZE
        mutex.release()    # V(mutex)
        empty.release()    # V(empty)：增加一个空槽位
        time.sleep(random.uniform(0.2, 0.4))

# 注意: P操作顺序不能颠倒！先P(empty/full)再P(mutex)，否则可能死锁

import threading
import time
import random

# === 读者优先版本 ===
class ReaderWriterProblem:
    def __init__(self):
        self.data = "初始数据"
        self.reader_count = 0
        self.mutex = threading.Lock()          # 保护reader_count
        self.write_lock = threading.Lock()     # 写互斥（读写互斥、写写互斥）

    def reader(self, name):
        for _ in range(3):
            # 读者进入
            with self.mutex:
                self.reader_count += 1
                if self.reader_count == 1:
                    self.write_lock.acquire()  # 第一个读者锁定写者

            # 读取数据（多个读者可以并行）
            print(f"  [读者{name}] 正在读取: '{self.data}' (当前读者数: {self.reader_count})")
            time.sleep(random.uniform(0.1, 0.3))

            # 读者退出
            with self.mutex:
                self.reader_count -= 1
                if self.reader_count == 0:
                    self.write_lock.release()  # 最后一个读者释放写锁

            time.sleep(random.uniform(0.1, 0.2))

    def writer(self, name):
        for i in range(2):
            self.write_lock.acquire()     # 写者独占

            self.data = f"{name}的第{i+1}次写入"
            print(f"[写者{name}] >>> 写入: '{self.data}'")
            time.sleep(random.uniform(0.2, 0.4))

            self.write_lock.release()     # 写者释放
            time.sleep(random.uniform(0.1, 0.3))

# 测试
rw = ReaderWriterProblem()
threads = []
for i in range(3):
    threads.append(threading.Thread(target=rw.reader, args=(f"R{i}",)))
for i in range(2):
    threads.append(threading.Thread(target=rw.writer, args=(f"W{i}",)))

for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"最终数据: '{rw.data}'")

import threading
import time
import random

NUM_PHILOSOPHERS = 5
chopsticks = [threading.Lock() for _ in range(NUM_PHILOSOPHERS)]

def philosopher_hierarchy(id):
    """资源层次法：总是先拿编号小的筷子"""
    left = id
    right = (id + 1) % NUM_PHILOSOPHERS

    # 总是先获取编号较小的筷子
    first = min(left, right)
    second = max(left, right)

    for meal in range(3):
        # 思考
        print(f"哲学家{id}: 思考中...")
        time.sleep(random.uniform(0.1, 0.3))

        # 拿筷子（按编号顺序）
        chopsticks[first].acquire()
        chopsticks[second].acquire()

        # 吃饭
        print(f"哲学家{id}: 正在吃第{meal+1}顿饭 🍜")
        time.sleep(random.uniform(0.1, 0.2))

        # 放筷子
        chopsticks[second].release()
        chopsticks[first].release()

    print(f"哲学家{id}: 已吃饱")

# 启动
threads = [threading.Thread(target=philosopher_hierarchy, args=(i,))
           for i in range(NUM_PHILOSOPHERS)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("所有哲学家都吃饱了！")

import threading
import time
import random

NUM_PHILOSOPHERS = 5
chopsticks = [threading.Lock() for _ in range(NUM_PHILOSOPHERS)]
# 最多允许4个哲学家同时尝试拿筷子（防止5个都拿了左筷子导致死锁）
waiter = threading.Semaphore(NUM_PHILOSOPHERS - 1)

def philosopher_waiter(id):
    """服务生解法：限制同时进餐人数"""
    left = id
    right = (id + 1) % NUM_PHILOSOPHERS

    for meal in range(3):
        print(f"哲学家{id}: 思考中...")
        time.sleep(random.uniform(0.1, 0.3))

        waiter.acquire()          # 向服务生请求许可
        chopsticks[left].acquire()
        chopsticks[right].acquire()

        print(f"哲学家{id}: 正在吃第{meal+1}顿饭 🍜")
        time.sleep(random.uniform(0.1, 0.2))

        chopsticks[right].release()
        chopsticks[left].release()
        waiter.release()          # 归还许可

    print(f"哲学家{id}: 已吃饱")

threads = [threading.Thread(target=philosopher_waiter, args=(i,))
           for i in range(NUM_PHILOSOPHERS)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("所有哲学家都吃饱了！")

# CAS的伪代码模拟
import threading  # 线程池/多线程：并发执行任务

class AtomicInteger:
    """使用CAS思想实现的原子整数（Python模拟）"""
    def __init__(self, value=0):
        self._value = value
        self._lock = threading.Lock()  # Python层面模拟原子性

    def compare_and_swap(self, expected, new_value):
        """CAS操作"""
        with self._lock:  # 实际CPU中这是一条原子指令
            if self._value == expected:
                self._value = new_value
                return True
            return False

    def get(self):
        return self._value

    def increment_and_get(self):
        """无锁自增"""
        while True:
            current = self._value
            if self.compare_and_swap(current, current + 1):
                return current + 1
            # CAS失败，重试（自旋）

# 测试
atom = AtomicInteger(0)
def worker():
    for _ in range(10000):
        atom.increment_and_get()

threads = [threading.Thread(target=worker) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"最终值: {atom.get()}")  # 40000

balance = 1000
def withdraw(amount):
    global balance
    if balance >= amount:
        balance -= amount
        return True
    return False