补全：给临界区计时

👁️ 3 人浏览 💬 0 人评论 ❤️ 添加收藏

本节的东西：

model(tasks)   时间模型：每条线程 (算, 改) 两段耗时；粗锁全串行 = Σ(算+改)，细锁算并行改串行 = max(算) + Σ改；交回 (粗, 细)
timed(fn, workers, work)   开 workers 条线程各跑 fn(work)，交回耗时
coarse(work)   sleep(work) 关在锁里再改 total；fine(work)   sleep 在锁外，只把改关在锁里

补全 held_ratio(fn, workers, work)：用 n02 的办法量每条线程持锁的时长（在 fn 里包一层计时的锁），交回「持锁总时长 ÷ 总耗时」的比值是否大于 1（粗锁：持锁总和 ≈ 总耗时；细锁：远小于）。为简单，fn 接收一个 held 列表，自己把每次持锁时长 append 进去。

提交你的答案

请登录后提交答案。

去登录

← 补全：一把大锁拆成每键一把 C 里的锁 →

更多题目

代码编辑器

语言： Ctrl + Enter 运行

import threading
import time

class Counter:
    def __init__(self):
        self.n = 0

def inc(self):
        tmp = self.n
        time.sleep(0.001)
        self.n = tmp + 1

class SafeCounter(Counter):
    def __init__(self):
        super().__init__()
        self.lock = threading.Lock()

def inc(self):
        with self.lock:
            tmp = self.n
            time.sleep(0.001)
            self.n = tmp + 1

def hammer(counter, workers=2, times=30):
    def job():
        for _ in range(times):
            counter.inc()
    ts = [threading.Thread(target=job) for _ in range(workers)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    return counter.n

def probe(lock):
    """这把锁现在能不能立刻拿到？能就拿了再放回去，交回 True；拿不到交回 False。"""
    if lock.acquire(blocking=False):
        lock.release()
        return True
    return False

class Guard:
    """手写的 with 替身：进就 acquire，出就 release——不管是正常出还是异常出。"""
    def __init__(self, lock):
        self.lock = lock

def __enter__(self):
        self.lock.acquire()
        return self

def __exit__(self, exc_type, exc, tb):
        self.lock.release()
        return False

class Box:
    """有界队列：满了 put 等、空了 get 等；等的时候把锁交出去（wait_for 会放锁、醒来再拿回）。"""
    def __init__(self, cap):
        self.cap = cap
        self.items = []
        self.cond = threading.Condition()

def put(self, x):
        with self.cond:
            self.cond.wait_for(lambda: len(self.items) < self.cap)
            self.items.append(x)
            self.cond.notify_all()

def get(self):
        with self.cond:
            self.cond.wait_for(lambda: len(self.items) > 0)
            x = self.items.pop(0)
            self.cond.notify_all()
            return x

def run_pc(cap, n):
    """一个生产者放 0..n-1，一个消费者取 n 个；交回消费者拿到的顺序。"""
    box = Box(cap)
    got = []

def producer():
        for i in range(n):
            box.put(i)

def consumer():
        for _ in range(n):
            got.append(box.get())
    p = threading.Thread(target=producer, daemon=True)
    c = threading.Thread(target=consumer, daemon=True)
    p.start()
    c.start()
    p.join(timeout=2)      # 写坏了的 get 会让生产者永远等——别把判题机拖死
    c.join(timeout=2)
    return got

class Gate:
    """最多 n 个同时在里面；顺便记下峰值（峰值计数器自己用一把锁保护）。"""
    def __init__(self, n):
        self.sem = threading.Semaphore(n)
        self.lock = threading.Lock()
        self.inside = 0
        self.peak = 0

def __enter__(self):
        self.sem.acquire()
        with self.lock:
            self.inside += 1
            self.peak = max(self.peak, self.inside)

def __exit__(self, *a):
        with self.lock:
            self.inside -= 1
        self.sem.release()
        return False

def run_gate(n, workers, hold=0.03):
    gate = Gate(n)

def job():
        with gate:
            time.sleep(hold)
    ts = [threading.Thread(target=job) for _ in range(workers)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    return gate.peak

def model(tasks):
    """tasks: 每条线程 (算, 改) 两段耗时。粗锁：全串行 = Σ(算+改)；细锁：算并行、改串行 = max(算) + Σ改。交回 (粗, 细)。"""
    coarse = sum(a + b for a, b in tasks)
    fine = max(a for a, b in tasks) + sum(b for a, b in tasks)
    return coarse, fine

def timed(fn, workers, work=0.03):
    """开 workers 条线程各跑一次 fn(work)，交回耗时。"""
    ts = [threading.Thread(target=fn, args=(work,)) for _ in range(workers)]
    t0 = time.perf_counter()
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    return time.perf_counter() - t0

LOCK = threading.Lock()
total = [0]

def coarse(work):
    with LOCK:
        time.sleep(work)          # 耗时的「算」也关在锁里
        total[0] += 1

def fine(work):
    time.sleep(work)              # 「算」在锁外
    with LOCK:
        total[0] += 1

TABLE = {
    "python3": ("threading.Lock()", "lock.acquire()", "lock.release()", "with lock:"),
    "c": ("pthread_mutex_t m", "pthread_mutex_lock(&m)", "pthread_mutex_unlock(&m)", "（没有：只能手动配对）"),
    "cpp": ("std::mutex m", "m.lock()", "m.unlock()", "std::lock_guard<std::mutex> g(m)"),
}

def count_pairs(src):
    """C 源码里 lock 和 unlock 各出现几次。"""
    return src.count("pthread_mutex_lock("), src.count("pthread_mutex_unlock(")

def find_leak(src):
    """按行扫 C 源码：拿了锁之后、放锁之前遇到 return，就是一条泄漏；交回泄漏所在的行号列表（从 1 起）。"""
    held = False
    leaks = []
    for i, line in enumerate(src.split("\n"), 1):
        s = line.strip()
        if "pthread_mutex_lock(" in s:
            held = True
        elif "pthread_mutex_unlock(" in s:
            held = False
        elif "return" in s and held:
            leaks.append(i)
    return leaks

def guard_span(src):
    """C++ 源码：lock_guard 声明所在行到它所在花括号块结束的那一行（从 1 起），交回 (起, 止)。"""
    lines = src.split("\n")
    start = None
    depth = 0
    for i, line in enumerate(lines, 1):
        if "lock_guard" in line and start is None:
            start = i
            depth = 0
        if start is not None:
            depth += line.count("{") - line.count("}")
            if depth < 0:
                return start, i
    return start, len(lines)

def coarse_t(work, held):
    t0 = time.perf_counter()
    with LOCK:
        time.sleep(work)
        total[0] += 1
    held.append(time.perf_counter() - t0)

def fine_t(work, held):
    time.sleep(work)
    t0 = time.perf_counter()
    with LOCK:
        total[0] += 1
    held.append(time.perf_counter() - t0)

def held_ratio(fn, workers, work=0.03):
    # TODO: 开 workers 条线程各跑 fn(work, held)，量总耗时；return sum(held) / 总耗时
    return 0.0

rc = held_ratio(coarse_t, 3)
rf = held_ratio(fine_t, 3)
print(str(rc > 0.9) + "/" + str(rf < 0.5) + "/" + str(rc > rf))

本次输入：

输出：

👩‍🏫

AI

💬 题目评论

全部评论