mrrc 0.7.6

A Rust library for reading, writing, and manipulating MARC bibliographic records in ISO 2709 binary format
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225

"""
Benchmarks for ProducerConsumerPipeline - parallel infrastructure.

These benchmarks test the opt-in parallel API to achieve multi-threaded
performance gains without modifying the standard MARCReader iteration API.

The ProducerConsumerPipeline is the correct way to benchmark pymrrc multi-threaded
performance. It uses:
- Background producer thread reading file + scanning boundaries
- Rayon parallel parser pool for batch processing
- Bounded channel (1000 records) for backpressure
- Consumer thread iterating results

Expected performance: 2.0x speedup (2 threads), 3.74x (4 threads)
"""

import pytest
from concurrent.futures import ThreadPoolExecutor
from mrrc import ProducerConsumerPipeline


class TestProducerConsumerPipelineBasic:
    """Basic tests for ProducerConsumerPipeline API."""

    @pytest.mark.benchmark
    def test_pipeline_sequential_1x_10k(self, benchmark):
        """Baseline: Pipeline reading 1x 10k records."""
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def read_pipeline():
            pipeline = ProducerConsumerPipeline.from_file(fixture_path)
            count = 0
            for record in pipeline:
                count += 1
            return count

        result = benchmark(read_pipeline)
        assert result == 10000

    @pytest.mark.benchmark
    def test_pipeline_sequential_4x_10k(self, benchmark):
        """Baseline: Sequential pipeline reading of 4x 10k from disk."""
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def read_pipelines():
            total = 0
            for _ in range(4):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                for record in pipeline:
                    total += 1
            return total

        result = benchmark(read_pipelines)
        assert result == 40000

    @pytest.mark.benchmark
    def test_pipeline_parallel_2x_10k_threaded(self, benchmark):
        """Parallel: 2 threads, each with own ProducerConsumerPipeline.

        This is the CORRECT way to test multi-threaded performance:
        - Each thread gets its own pipeline (each spawns background producer)
        - Expected: 2.0x speedup vs sequential
        """
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def read_parallel():
            def read_pipeline(_):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                count = 0
                for record in pipeline:
                    count += 1
                return count

            with ThreadPoolExecutor(max_workers=2) as executor:
                results = list(executor.map(read_pipeline, range(2)))
            return sum(results)

        result = benchmark(read_parallel)
        assert result == 20000

    @pytest.mark.benchmark
    def test_pipeline_parallel_4x_10k_threaded(self, benchmark):
        """Parallel: 4 threads, each with own ProducerConsumerPipeline.

        This is the test that demonstrates parallel speedup with the pipeline.
        - 4 threads, each with independent pipeline
        - Each pipeline has its own background producer + rayon parser pool
        - Expected: 3.74x speedup vs sequential (from RESULTS.md)
        """
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def read_parallel():
            def read_pipeline(_):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                count = 0
                for record in pipeline:
                    count += 1
                return count

            with ThreadPoolExecutor(max_workers=4) as executor:
                results = list(executor.map(read_pipeline, range(4)))
            return sum(results)

        result = benchmark(read_parallel)
        assert result == 40000


class TestProducerConsumerPipelineWithExtraction:
    """Realistic workloads with field extraction."""

    @pytest.mark.benchmark
    def test_pipeline_sequential_extraction_4x_10k(self, benchmark):
        """Baseline: Sequential pipeline with field extraction (4x 10k)."""
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def read_with_extraction():
            total = 0
            for _ in range(4):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                for record in pipeline:
                    _ = record.title
                    _ = record.get_fields("100")
                    total += 1
            return total

        result = benchmark(read_with_extraction)
        assert result == 40000

    @pytest.mark.benchmark
    def test_pipeline_parallel_extraction_4x_10k_threaded(self, benchmark):
        """Parallel: 4 threads with field extraction (realistic workload).

        Pipeline + extraction with multi-threading:
        - Each thread reads + extracts via pipeline
        - Expected: 3.74x speedup
        """
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def read_with_extraction():
            def process_pipeline(_):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                count = 0
                for record in pipeline:
                    _ = record.title
                    _ = record.get_fields("100")
                    count += 1
                return count

            with ThreadPoolExecutor(max_workers=4) as executor:
                results = list(executor.map(process_pipeline, range(4)))
            return sum(results)

        result = benchmark(read_with_extraction)
        assert result == 40000


class TestProducerConsumerPipelineMultiFile:
    """Multi-file processing scenarios (typical use case)."""

    @pytest.mark.benchmark
    def test_process_4_files_sequential(self, benchmark):
        """Process 4 files sequentially using ProducerConsumerPipeline."""
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def process_sequential():
            total = 0
            for _ in range(4):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                for record in pipeline:
                    total += 1
            return total

        result = benchmark(process_sequential)
        assert result == 40000

    @pytest.mark.benchmark
    def test_process_4_files_parallel_4_threads(self, benchmark):
        """Process 4 files in parallel with 4 threads (optimal threading).

        This demonstrates the parallel pipeline infrastructure:
        - 4 threads, 4 files
        - Each thread: ProducerConsumerPipeline.from_file(filepath)
        - Each pipeline spawns its own background producer + parser pool
        - Expected: 3.74x speedup vs sequential
        """
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def process_parallel():
            def process_file(_):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                count = 0
                for record in pipeline:
                    count += 1
                return count

            with ThreadPoolExecutor(max_workers=4) as executor:
                results = list(executor.map(process_file, range(4)))
            return sum(results)

        result = benchmark(process_parallel)
        assert result == 40000

    @pytest.mark.benchmark
    def test_process_8_files_parallel_4_threads(self, benchmark):
        """Process 8 files with 4 threads (oversubscription test).

        Tests thread pool efficiency with more files than threads.
        Expected: ~2.0x speedup (limited by 4 threads, not 8 files)
        """
        fixture_path = "tests/data/fixtures/10k_records.mrc"

        def process_parallel():
            def process_file(_):
                pipeline = ProducerConsumerPipeline.from_file(fixture_path)
                count = 0
                for record in pipeline:
                    count += 1
                return count

            with ThreadPoolExecutor(max_workers=4) as executor:
                results = list(executor.map(process_file, range(8)))
            return sum(results)

        result = benchmark(process_parallel)
        assert result == 80000