単一プロデューサ・単一コンシューマのリングバッファ

def initial_state():    return {"head": 0, "tail": 0, "pushed": (), "popped": ()}def ring_producer(items, capacity):    def thread(shared):        tail = 0        for item in items:            yield            head = shared.get("head", 0)            if tail - head < capacity:                yield                shared["slot%d" % (tail % capacity)] = item                yield                tail += 1                shared["tail"] = tail                shared["pushed"] = shared.get("pushed", ()) + (item,)    return threaddef ring_consumer(attempts, capacity):    def thread(shared):        head = 0        for _ in range(attempts):            yield            tail = shared.get("tail", 0)            if head != tail:                yield                item = shared.get("slot%d" % (head % capacity))                yield                head += 1                shared["head"] = head                shared["popped"] = shared.get("popped", ()) + (item,)    return threadtwo = lambda shared: [ring_producer((7, 8), 2)(shared), ring_consumer(2, 2)(shared)]for prefix in ([0, 0, 0], [0, 0, 0, 1, 1, 1], [0, 0, 0, 0, 0, 0, 1, 1, 1, 1]):    print(prefix, run_prefix(two, prefix, initial_state())[0])

[0, 0, 0] {'head': 0, 'tail': 1, 'pushed': (7,), 'popped': (), 'slot0': 7}
[0, 0, 0, 1, 1, 1] {'head': 1, 'tail': 1, 'pushed': (7,), 'popped': (7,), 'slot0': 7}
[0, 0, 0, 0, 0, 0, 1, 1, 1, 1] {'head': 1, 'tail': 2, 'pushed': (7, 8), 'popped': (7,), 'slot0': 7, 'slot1': 8}

def violations(final):    pushed, popped = tuple(final.get("pushed", ())), tuple(final.get("popped", ()))    broken = []    if popped != pushed[: len(popped)]:        broken.append("order")    if len(popped) > len(pushed):        broken.append("count")    return sorted(broken)def verify(producers, attempts, capacity):    def make_threads(shared):        return [ring_producer(items, capacity)(shared) for items in producers] + [ring_consumer(attempts, capacity)(shared)]    outcomes = explore_all(make_threads, initial_state())    broken = {}    pushed = popped = 0    for key, count in outcomes.items():        final = dict(key)        pushed = max(pushed, len(final.get("pushed", ())))        popped = max(popped, len(final.get("popped", ())))        for name in violations(final):            broken[name] = broken.get(name, 0) + count    return {"schedules": sum(outcomes.values()), "broken": dict(sorted(broken.items())), "pushed": pushed, "popped": popped}print(violations({"pushed": (7, 8), "popped": (8,)}))print(verify([(7, 8)], 2, 2))print(verify([(7, 8)], 2, 1))

['order']
{'schedules': 101, 'broken': {}, 'pushed': 2, 'popped': 2}
{'schedules': 25, 'broken': {}, 'pushed': 2, 'popped': 2}

print(verify([(7,), (8,)], 2, 2))

{'schedules': 1072, 'broken': {'order': 458}, 'pushed': 2, 'popped': 1}

ticks = [int(round(float(c) * 100)) for c in load_bars()["close"].iloc[:4]]print(ticks)print("one producer: ", verify([ticks[:2]], 3, 2))print("two producers:", verify([ticks[:1], ticks[1:2]], 2, 2))

[9369, 9305, 9354, 9349]
one producer:  {'schedules': 291, 'broken': {}, 'pushed': 2, 'popped': 2}
two producers: {'schedules': 1072, 'broken': {'order': 458}, 'pushed': 2, 'popped': 1}