Skip to content

test_threading

Thread-safety tests for shared stream objects.

frameCPP streams are stateful, and the bindings release the GIL around I/O, so stream access must be serialized in the bindings themselves. These tests drive shared readers/writers from many threads at once; on unserialized builds they corrupt the stream ("stream is no longer in good state") or crash outright.

TestSharedReader

test_concurrent_reads_independent_readers 

test_concurrent_reads_independent_readers(test_gwf_file)

Concurrent reads of the same file via separate handles.

Source code in gwframe/tests/test_threading.py 
60
61
62
63
64
65
66
67
68
69
def test_concurrent_reads_independent_readers(self, test_gwf_file):
    """Concurrent reads of the same file via separate handles."""
    reference = gwframe.read(test_gwf_file, STRAIN_CHANNEL).array

    def worker(_tid):
        for _ in range(4):
            ts = gwframe.read(test_gwf_file, STRAIN_CHANNEL)
            np.testing.assert_array_equal(ts.array, reference)

    assert run_threads(worker) == []

test_concurrent_reads_shared_reader 

test_concurrent_reads_shared_reader(test_gwf_file)

Many threads reading through one FrameReader must not corrupt it.

Source code in gwframe/tests/test_threading.py 
48
49
50
51
52
53
54
55
56
57
58
def test_concurrent_reads_shared_reader(self, test_gwf_file):
    """Many threads reading through one FrameReader must not corrupt it."""
    reader = gwframe.FrameReader(test_gwf_file)
    reference = reader.read(STRAIN_CHANNEL).array

    def worker(_tid):
        for _ in range(N_ITER):
            ts = reader.read(STRAIN_CHANNEL)
            np.testing.assert_array_equal(ts.array, reference)

    assert run_threads(worker) == []

TestSharedWriter

test_concurrent_writes_independent_files 

test_concurrent_writes_independent_files(tmp_path)

Concurrent writes to distinct files via separate writers.

Source code in gwframe/tests/test_threading.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def test_concurrent_writes_independent_files(self, tmp_path):
    """Concurrent writes to distinct files via separate writers."""
    sample_rate = 1024

    def worker(tid):
        for i in range(4):
            path = tmp_path / f"out_{tid}_{i}.gwf"
            data = np.arange(sample_rate, dtype=np.float64) * (tid + 1)
            gwframe.write(path, {f"X1:TEST_{tid}": data}, 1000000000.0, sample_rate)
            back = gwframe.read(path, f"X1:TEST_{tid}")
            np.testing.assert_array_equal(back.array, data)

    assert run_threads(worker) == []

test_concurrent_writes_shared_writer 

test_concurrent_writes_shared_writer(tmp_path)

Many threads writing through one FrameWriter must not corrupt it.

All threads write the same channel name: frames whose channel sets differ produce a broken TOC even single-threaded (separate, preexisting bug), which would mask the threading result here.

Source code in gwframe/tests/test_threading.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def test_concurrent_writes_shared_writer(self, tmp_path):
    """Many threads writing through one FrameWriter must not corrupt it.

    All threads write the same channel name: frames whose channel sets
    differ produce a broken TOC even single-threaded (separate,
    preexisting bug), which would mask the threading result here.
    """
    path = tmp_path / "threaded.gwf"
    sample_rate = 1024
    data = {
        tid: np.full(sample_rate, float(tid), dtype=np.float64)
        for tid in range(N_THREADS)
    }
    with gwframe.FrameWriter(path) as writer:

        def worker(tid):
            for i in range(N_ITER):
                # unique start time per frame, no cross-thread coordination
                frame_index = tid * N_ITER + i
                writer.write(
                    data[tid],
                    start=1000000000.0 + frame_index,
                    sample_rate=sample_rate,
                    name="X1:TEST",
                )

        failures = run_threads(worker)
    assert failures == []

    # every frame written must read back intact, with the data of the
    # thread its start time encodes
    frames = list(gwframe.read_frames(path))
    assert len(frames) == N_THREADS * N_ITER
    for frame in frames:
        ts = frame["X1:TEST"]
        tid = int(ts.start - 1000000000.0) // N_ITER
        np.testing.assert_array_equal(ts.array, data[tid])

run_threads 

run_threads(worker)

Run worker(tid) in N_THREADS threads, starting simultaneously.

Returns the list of exceptions raised by workers.

Source code in gwframe/tests/test_threading.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def run_threads(worker):
    """Run worker(tid) in N_THREADS threads, starting simultaneously.

    Returns the list of exceptions raised by workers.
    """
    barrier = threading.Barrier(N_THREADS)
    failures = []

    def wrap(tid):
        try:
            barrier.wait()
            worker(tid)
        except Exception as exc:  # noqa: BLE001
            failures.append(exc)

    threads = [threading.Thread(target=wrap, args=(tid,)) for tid in range(N_THREADS)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    return failures