Build a usable pthread worker pool that counts real file bytes, lines, and words. A four-slot queue coordinates the work; one owned result slot per input keeps the final report deterministic.
Implement a scanner whose output is correct before adding a worker pool.
Publish a job only when a ring-buffer slot is available.
Connect the queue and scanner with a small worker loop that cannot scramble the report.
Distinguish “no work yet” from “no work will ever arrive,” then finish the lifetime of shared objects safely.
Check partial startup, per-file failure, malformed CLI input, and deterministic output with bounded automated tests.
Mina volunteers at a school robotics club. At the end of a workshop she receives several small note files and wants a reliable inventory: how many bytes, logical lines, and words are in each? Some notes are empty, one has no final newline, and a copied filename may be wrong. She needs a useful report even when one file cannot be opened.
She begins with one precise scanner and checks it against tiny synthetic notes. Next she numbers the file arguments and feeds their indices through a queue with four positions. When every position is occupied, her producer waits for space. Several workers can take different jobs, but each worker opens its own file and writes only that job’s result. This makes the ownership rule visible before concurrency becomes complicated.
The last job being submitted is not the same as the last job being finished. Mina closes the queue to announce that no more jobs will arrive, lets consumers drain the remaining jobs, and joins all workers. Only then does she print the result slots in original input order. A missing file becomes a clearly marked failed slot, not a misleading row of zeros.
Finally, she tests short and long inputs, repeated schedules, and carefully injected failures. The project demonstrates a correct coordination pattern; these small examples make no claim that more threads are faster. All code, fixtures, and explanations are original Leon teaching material, independent of supplied assignments.
A finite pool and bounded job list make resource use understandable. Zero jobs is a normal lifecycle case.
Check: Try 1, 16, 0, 17, 2x, no worker argument, and 129 file arguments. Invalid CLI input exits 2 with no report.
The producer must wait when full instead of growing storage. Consumers must sleep when no work can yet be taken.
Check: Inspect both predicate while loops and test fewer jobs, more jobs, and 37 jobs across repeated runs.
Concurrent completion may change order. Separate result slots and a later reporting phase preserve the input order.
Check: Compare complete stdout for 1, 2, 4, and 16 workers against an independent sequential oracle.
Empty files, final newlines, and buffer boundaries need definitions before they need code.
Check: Empty gives 0/0/0. The evening fixture gives 26 bytes, 2 lines, and 5 words despite having only one LF.
A teaching run should have explicit storage and work bounds. Partial counts are not a complete result.
Check: Test the exact byte limit, one byte above it, missing paths, directories, and a FIFO without a writer.
Joining a worker that waits for nonexistent future work can hang forever. Uncreated thread handles are not valid join targets.
Check: Temporary test shims force failure before worker 1 and after two workers. Both runs must exit 1 without output or a timeout.
Validate all arguments, initialize storage, create the whole pool, publish each job index once, then close the queue. The same main thread later joins and reports.
A ring stores at most four indices. count, head, closed, and entries are protected by one mutex. not_empty and not_full wake threads that must then recheck their predicates.
Each worker repeatedly takes one index, releases the queue lock, opens its own regular file, and writes that index’s FileStats. Files may finish in any order.
After every successful join, main reads slots from zero through J−1. A failed slot prints its error category, while successful slots print complete counts.
INITIALIZE → CREATE ALL WORKERS → PRODUCE INDICES → CLOSE → DRAIN → JOIN → REPORT. Consumers may scan while main produces. OPEN+EMPTY means wait; OPEN+FULL means producer waits; CLOSED+NONEMPTY means keep consuming; CLOSED+EMPTY means exit. If creation fails, no indices have been produced: CLOSE → JOIN CREATED WORKERS → DESTROY → EXIT 1.
| Resource | Owner | Release |
|---|---|---|
| Queue fields and synchronization objects | Main owns their lifetime. All threads borrow the queue; every live state access holds its mutex. | Main destroys both conditions and the mutex only after all created workers join. |
| argv paths and WorkerContext | Main lends stable, read-only paths and a fully initialized context to every worker. | The context remains in main’s stack frame through the final join; argv survives until process exit. |
| One FileStats slot per job | The worker that pops index i exclusively writes results[i]. Main reads only after joining all workers. | Storage is a bounded main-stack array; it needs no free and stays alive throughout worker access. |
| File descriptor and 4096-byte buffer | Only the scanning worker uses its descriptor, file offset, buffer, and word state. | scan_file attempts close on every path after a successful open; its local buffer ends with the call. |
| Joinable thread handles | Main stores only successfully created handles in the prefix counted by started. | Join that prefix on both successful startup and partial startup failure; never join an uncreated handle. |
Implement a scanner whose output is correct before adding a worker pool.
Concurrency cannot repair an ambiguous definition. A line is an LF-terminated segment or the final nonempty suffix. A word begins when a non-separator follows either a separator or the start of the stream. These are two small state machines, and their state must survive a short read or a buffer boundary.
Read stats.h first: a result carries counts, a status, and an optional system error. Only SCAN_OK permits complete counts to be reported.
Trace empty, a, a\n, and a b through the counters. Keep in_word and last outside the read loop. Test received < 0 before its size_t conversion.
At EOF, add one logical line only for a nonempty stream whose final byte is not LF. On read failure, preserve the error and suppress partial counts.
Teaching extract from the complete project. Build the full files below to run it.
static void count_stream(int descriptor, FileStats *result) {
unsigned char buffer[4096];
bool in_word = false;
unsigned char last = '\n';
for (;;) {
ssize_t received = read(descriptor, buffer, sizeof buffer);
if (received < 0) {
if (errno == EINTR) {
continue;
}
result->status = SCAN_READ;
result->system_error = errno;
return;
}
if (received == 0) {
break;
}
size_t length = (size_t)received;
if (length > FILE_BYTE_LIMIT - result->bytes) {
result->status = SCAN_TOO_LARGE;
return;
}
result->bytes += length;
for (size_t i = 0; i < length; i++) {
unsigned char byte = buffer[i];
if (byte == '\n') {
result->lines++;
}
bool space = separator(byte);
if (!space && !in_word) {
result->words++;
}
in_word = !space;
last = byte;
}
}
if (result->bytes != 0 && last != '\n') {
result->lines++;
}
}
This is an extract of the final scanner, not a standalone program. separator and the types come from the full project. result->bytes tracks completed chunks; the subtraction guard prevents counting beyond the 8 MiB contract. The read buffer is reused, while in_word and last survive its reuse.
Every byte is a separator, so no word starts. There is one LF, followed by vertical-tab/form-feed bytes; that nonempty suffix adds the second logical line.
Publish a job only when a ring-buffer slot is available.
The queue separates storage capacity from total job count. head is the oldest occupied position; count is how many positions are occupied. The next insertion is (head + count) modulo capacity. Holding one mutex makes a check followed by an update one protected transition. Waiting must release that mutex so a consumer can create space.
Draw four boxes and trace push 0, push 1, pop, push 2. Verify the invariant 0 <= count <= 4 before and after each operation.
Read queue_push beside queue_pop. Producers wait on not_full; consumers wait on not_empty. Both use while with the closed flag.
Signal after the protected count change, then unlock. Keep file reading entirely outside the queue lock.
Teaching extract from the complete project. Build the full files below to run it.
bool queue_push(WorkQueue *queue, size_t job) {
thread_check(pthread_mutex_lock(&queue->mutex), "mutex lock");
while (queue->count == QUEUE_CAPACITY && !queue->closed) {
thread_check(pthread_cond_wait(&queue->not_full, &queue->mutex),
"wait not_full");
}
if (queue->closed) {
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
return false;
}
size_t tail = (queue->head + queue->count) % QUEUE_CAPACITY;
queue->entries[tail] = job;
queue->count++;
thread_check(pthread_cond_signal(&queue->not_empty), "signal not_empty");
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
return true;
}
The complete function is extracted from queue.c. A condition wakeup is not a reserved free slot; the while loop rechecks the shared state. A closed queue rejects a new job even if there is room. queue_pop removes one index under the same mutex and signals not_full.
Other workers could not pop, and main could not push, until the whole file finished. The lock protects brief queue transitions, not the independent scanning work.
Connect the queue and scanner with a small worker loop that cannot scramble the report.
A shared result array does not automatically imply a race. Each job index enters the queue once, and one locked pop removes it once. Consequently exactly one worker writes each element. Workers share a read-only context but not a writable counter or file offset. Main waits until all writers finish before observing the array.
Follow job i from argv[i + 2] to queue_push(i), queue_pop(&job), and results[job]. Job numbers printed to users start at one.
Keep queue, context, paths, and results alive through every join. Do not pass the address of a changing loop-local job variable to a worker.
Let scan_file record an error in the same slot on failure. The consumer loop must still take later jobs.
Teaching extract from the complete project. Build the full files below to run it.
static void *consume(void *argument) {
WorkerContext *context = argument;
size_t job;
while (queue_pop(context->queue, &job)) {
/* Exactly one pop receives each index. No other worker writes this
result, and main waits for all joins before reading any result. */
scan_file(context->paths[job], &context->results[job]);
}
return NULL;
}
This final-program extract uses the WorkerContext declared in main.c. queue_pop returns after releasing the queue mutex, so scans overlap without holding that lock. No worker prints. Distinct FileStats objects have distinct writers, and main’s later joins supply the necessary visibility before reading them.
It goes into results[7]. Main prints that slot eighth, after joining every worker. Completion order never selects output order.
Distinguish “no work yet” from “no work will ever arrive,” then finish the lifetime of shared objects safely.
An empty queue alone cannot tell a consumer whether to wait or leave. The closed flag adds that information. Closing does not erase pending indices. After close, consumers keep popping while count is positive, then return when it reaches zero. A broadcast wakes all sleepers because every idle worker must discover termination, even when the file list was empty.
Read the pop predicate as “wait while empty AND still open.” A closed, nonempty queue is still consumable.
After the producer loop, call queue_close exactly as a phase transition. Trace the zero-job case: all workers must eventually see closed and leave.
Join every created worker, then destroy synchronization objects and print. Never destroy a mutex or condition that a worker may still access.
Teaching extract from the complete project. Build the full files below to run it.
void queue_close(WorkQueue *queue) {
thread_check(pthread_mutex_lock(&queue->mutex), "mutex lock");
queue->closed = true;
/* All empty-queue consumers must learn that no future job can arrive. */
thread_check(pthread_cond_broadcast(&queue->not_empty), "broadcast not_empty");
thread_check(pthread_cond_broadcast(&queue->not_full), "broadcast not_full");
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
}
This queue.c extract changes closed while holding the same mutex used by both waits. not_empty broadcasts end empty-queue waits; not_full also wakes any producer observing closure. In the current program only main closes, after production, so its normal producer is never blocked during that close. queue_pop drains rather than discards.
The other workers may remain asleep forever because no later job arrival will signal them. Broadcasting lets each reacquire the mutex, observe closed-and-empty, and return.
Check partial startup, per-file failure, malformed CLI input, and deterministic output with bounded automated tests.
A thread handle is valid only after successful creation. Starting production before the full pool exists would complicate startup failure because some jobs might already be running. This design deliberately creates the pool first. A startup failure can then close an empty queue, join the successful prefix of handles, and return without a misleading partial report.
Trace started = 0 and started = 2 in the creation-failure branch. Only those already created workers are joined; no job indices have been published.
Use make test. It compiles in a temporary copy, calculates expected counts independently, checks negative cases, and repeats 37-job runs with different worker counts.
Read the temporary fault-injection shims for pthread_create and read. They force rare failures safely. Optionally run python3 test.py --tsan where the Clang runtime supports it.
Teaching extract from the complete project. Build the full files below to run it.
for (; started < workers; started++) {
error = pthread_create(&threads[started], NULL, consume, &context);
if (error != 0) {
fprintf(stderr, "thread creation failed after %zu workers "
"(pthread error %d)\n", started, error);
/* No jobs are published until the whole pool exists. */
queue_close(&queue);
join_workers(threads, started);
queue_destroy(&queue);
return EXIT_FAILURE;
}
}
This is the actual pool-creation loop from main.c; started, workers, context, queue, threads, and error are declared just above it. Runtime mutex/condition/join errors use a separate documented fail-fast policy. Do not describe that policy as graceful recovery. The full program and its error checks are the runnable unit.
No. They can expose defects in the executions they observe. The stronger reasoning still needs unique job delivery, one writer per slot, protected queue predicates, correct object lifetimes, and joins before reading results.
Use Clang with C11 support, make and Python 3 on macOS or Linux. On Windows use a Linux environment such as WSL. The supplied Makefiles use Clang. Open a terminal in the download directory. Unzip the archive, then enter its folder. make builds the executable; make test checks its behavior.
unzip comp2017-thread-pipeline.zip
cd comp2017-thread-pipeline
make
make test
./thread_pipeline 3 fixtures/morning.txt fixtures/evening.txt fixtures/empty.txt
job=1 bytes=37 lines=2 words=6 job=2 bytes=26 lines=2 words=5 job=3 bytes=0 lines=0 words=0 summary files=3 ok=3 failed=0
Exit status: 0.
The final unterminated evening line is counted and the empty file succeeds. stdout lists file arguments in their original order. Exit status is 0; stderr is empty. Run make once first; compilation output is not included in this stdout panel.
./thread_pipeline 1 fixtures/morning.txt fixtures/evening.txt fixtures/blank.txt fixtures/empty.txt fixtures/spaces.txt
job=1 bytes=37 lines=2 words=6 job=2 bytes=26 lines=2 words=5 job=3 bytes=2 lines=2 words=0 job=4 bytes=0 lines=0 words=0 job=5 bytes=6 lines=2 words=0 summary files=5 ok=5 failed=0
Exit status: 0.
Five jobs pass through a queue that stores only four indices at once, even with one consumer. Blank lines and whitespace-only content remain meaningful cases. Exit status is 0; stderr is empty. The sample establishes correctness, not throughput.
./thread_pipeline 16
summary files=0 ok=0 failed=0
Exit status: 0.
An empty job list is valid. Main closes the queue and broadcasts; each consumer observes empty-and-closed and returns. Every worker is joined. Exit status is 0 with no stderr.
./thread_pipeline 2 fixtures/morning.txt fixtures/missing.txt fixtures/evening.txt
job=1 bytes=37 lines=2 words=6 job=2 error=open job=3 bytes=26 lines=2 words=5 summary files=3 ok=2 failed=1
Exit status: 1.
The second job reports error=open. The third job still succeeds and prints third. Exit status is 1, and stderr names job 2, the path, and the OS error; the OS wording is platform-dependent. Only exact stdout is shown here.
./thread_pipeline 0 fixtures/morning.txt
Exit status: 2. No standard output. Read the exit status and any diagnostic on stderr.
No report is printed. Exit status is 2 and stderr prints the accepted usage and bounds. Validation happens before any worker is created. Empty stdout is intentional.
Consumers are still waiting because closure was not broadcast or closed is missing from the wait predicate.
Run ./thread_pipeline 16 with a timeout. Read queue_pop and queue_close together; check that closed changes under the mutex and all waiting consumers are notified.
Wait while count is zero AND the queue is open; broadcast on close, then join. Never try to repair the hang with a sleep.
Workers print on completion, or main reads before joining all workers.
Search main.c for printf and verify that the per-job printing loop is reached only after join_workers. Repeat the 37-job input with 1 and 16 workers.
Workers write their own indexed slot only. Main joins every worker and then traverses the result array in argument order.
The EOF correction is missing or is applied even after a final LF.
Inspect the fixture bytes and count their LF bytes. The evening file has one LF and a nonempty suffix; blank.txt has two LFs and no suffix.
Add one only when bytes != 0 and last != LF. Do not infer a logical line from the fact that a buffer was read.
in_word was reinitialized for each chunk instead of each file.
Use the generated boundary.bin case and trace the final byte of the first read and the first byte of the next. Neither is a separator.
Keep in_word outside the read loop. Only a separator changes it to false.
Main joined waiting workers before closing the queue, or joined an invalid uncreated handle.
Run the injected failure cases in test.py. Track started and confirm that no indices have been published yet.
Close the empty queue first, join exactly the successful prefix, destroy it, and return 1. Do not proceed to the producer loop.
Reporting ignores ScanStatus and treats initial counters as a finished scan.
Compare empty.txt with a nonexistent path. The counts may both start at zero, but their status values must differ.
Print numeric counts only for SCAN_OK. Preserve an error row at the failed job index and return a nonzero process status.
One summary line: files=0 ok=0 failed=0; exit 0 within the subprocess timeout.
Tests the close broadcast and the difference between empty and closed.
Each input has exactly one result in input order; all supplied files succeed.
Covers fewer and more jobs than queue slots without assuming which thread runs first.
Exact hard-coded totals are checked, including empty 0/0/0 and evening 26/2/5.
Separates the stated logical-line and word definitions from terminal appearance.
The independent byte-regex oracle matches the C counts; the word is not split at the chunk boundary.
Checks stream state retention and avoidance of C-string assumptions.
0, 17, signs, spaces, suffix text, missing arguments, huge numbers, and 129 jobs exit 2; 128 jobs and 0002 workers are accepted.
Verifies validation before threads start and bounded numeric accumulation.
The second row is error=open, the third still reports valid counts, summary failed=1, and process exit is 1.
A per-file failure must not cancel unrelated work or hide its position.
Both report error=not-regular without hanging; files are created only in the test’s temporary directory.
Checks type validation and nonblocking FIFO opening before the rejection.
The exact limit succeeds; the larger file reports too-large without numeric partial counts.
Checks the inclusive limit and the subtraction guard.
Workers 1, 2, 4, and 16 all match an independent oracle and the one-worker stdout, byte for byte.
Exercises varied schedules and queue reuse; it is bounded evidence, not proof of all schedules or a speed benchmark.
New pushes fail; existing indices drain in FIFO order; the final pop returns false; repeated close remains safe.
Directly distinguishes close-and-drain from discard-on-close.
Failure before any worker or after two workers exits 1 cleanly; a read EIO reports error=read; one EINTR retries successfully.
Tests rare negative paths with temporary build shims rather than exhausting system resources.
Main initializes the shared context before creating workers. Queue transitions are synchronized by one mutex. Each job index is produced once and removed once, giving its slot exactly one writer. Distinct array elements are separate objects. Main never reads any slot during the worker phase; successful joins synchronize with completion before reporting. This is an ownership-and-phase argument, not an assumption about timing. It would stop being valid if workers updated a shared total or if main polled partially written results.
The four-slot queue limits waiting indices and applies backpressure. At most 16 files are actively scanned, and up to 128 result slots are kept for ordered output. Each scanner reuses one 4096-byte buffer, so file length does not increase its memory. Runtime thread stacks also consume memory and are not included in that 4096-byte figure. Counting takes O(B + J) work for B bytes and J jobs, plus scheduling and waiting. Rejecting an oversized file may read one extra buffer to discover the limit was exceeded.
Queue initialization unwinds initialized components. Partial thread creation has orderly close/join/destroy cleanup. Per-file open, metadata, type, read, size, and close errors remain visible while unrelated jobs continue. Unexpected runtime synchronization errors instead terminate the process immediately after a diagnostic. That policy does not promise a complete report or graceful cleanup. No thread cancellation, signal shutdown, or file-read deadline is implemented. A slow regular-file filesystem or repeated EINTR can still delay a run.
Logical lines include a final nonempty suffix, so this line count differs from counting LF alone. Exactly six ASCII whitespace bytes split words; multibyte characters and zero bytes are ordinary non-separators. No Unicode segmentation or locale-sensitive classification is claimed. Input files must stay unchanged for repeatable results. The supplied synthetic regular files are the intended learning inputs; accepting user paths does not make this a secure filesystem sandbox.
A fixed pool avoids creating one thread per file, but startup, locks, storage bandwidth, and file-cache effects still cost time. These tiny runs deliberately print no timings. Repeated deterministic output and the optional ThreadSanitizer run check observed behavior, not every possible schedule. Tests use independent byte-count and regex expectations, bounded own fixtures, temporary builds, and subprocess timeouts. A genuine performance extension must compare equivalent workloads and report repeated measurements.
These are the complete files in the download. The numbers suggest a reading order. The file list links directly to each source block.
The result contract and bounded file size
Keep these file names and folders when building. Long lines scroll within the code panel.
/* Leon | Original teaching project. */
#ifndef STATS_H
#define STATS_H
#include <stddef.h>
#define FILE_BYTE_LIMIT (8u * 1024u * 1024u)
typedef enum {
SCAN_OK,
SCAN_OPEN,
SCAN_METADATA,
SCAN_NOT_REGULAR,
SCAN_READ,
SCAN_TOO_LARGE,
SCAN_CLOSE
} ScanStatus;
typedef struct {
size_t bytes;
size_t lines;
size_t words;
ScanStatus status;
int system_error;
} FileStats;
/* The caller exclusively owns *result until this call returns.
Failure counters are partial and must not be reported as valid counts. */
void scan_file(const char *path, FileStats *result);
const char *scan_status_name(ScanStatus status);
#endif
Read FILE_BYTE_LIMIT as 8 MiB, then separate the three size_t counters from ScanStatus.
system_error stores errno only for operations that actually supply one; logical rejections need no invented system error.
The scan_file contract gives the caller one exclusively owned output object. Failed counters are partial and must be hidden.
One worker’s independent regular-file scanner
Keep these file names and folders when building. Long lines scroll within the code panel.
/* Leon | Original teaching project. */
#include "stats.h"
#include <errno.h>
#include <fcntl.h>
#include <stdbool.h>
#include <sys/stat.h>
#include <unistd.h>
/* Fixed ASCII separators make results independent of process locale. */
static bool separator(unsigned char byte) {
return byte == ' ' || byte == '\t' || byte == '\n' || byte == '\r' ||
byte == '\v' || byte == '\f';
}
static void count_stream(int descriptor, FileStats *result) {
unsigned char buffer[4096];
bool in_word = false;
unsigned char last = '\n';
for (;;) {
ssize_t received = read(descriptor, buffer, sizeof buffer);
if (received < 0) {
if (errno == EINTR) {
continue;
}
result->status = SCAN_READ;
result->system_error = errno;
return;
}
if (received == 0) {
break;
}
size_t length = (size_t)received;
if (length > FILE_BYTE_LIMIT - result->bytes) {
result->status = SCAN_TOO_LARGE;
return;
}
result->bytes += length;
for (size_t i = 0; i < length; i++) {
unsigned char byte = buffer[i];
if (byte == '\n') {
result->lines++;
}
bool space = separator(byte);
if (!space && !in_word) {
result->words++;
}
in_word = !space;
last = byte;
}
}
if (result->bytes != 0 && last != '\n') {
result->lines++;
}
}
void scan_file(const char *path, FileStats *result) {
*result = (FileStats){0};
/* Nonblocking open avoids waiting forever for a FIFO writer. Only a
regular file is accepted after opening; each worker owns its fd. */
int descriptor = open(path, O_RDONLY | O_NONBLOCK);
if (descriptor < 0) {
result->status = SCAN_OPEN;
result->system_error = errno;
return;
}
struct stat info;
if (fstat(descriptor, &info) != 0) {
result->status = SCAN_METADATA;
result->system_error = errno;
} else if (!S_ISREG(info.st_mode)) {
result->status = SCAN_NOT_REGULAR;
} else {
count_stream(descriptor, result);
}
/* Preserve the first error; do not retry close after an uncertain result. */
if (close(descriptor) != 0 && result->status == SCAN_OK) {
result->status = SCAN_CLOSE;
result->system_error = errno;
}
}
const char *scan_status_name(ScanStatus status) {
switch (status) {
case SCAN_OK: return "ok";
case SCAN_OPEN: return "open";
case SCAN_METADATA: return "metadata";
case SCAN_NOT_REGULAR: return "not-regular";
case SCAN_READ: return "read";
case SCAN_TOO_LARGE: return "too-large";
case SCAN_CLOSE: return "close";
}
return "unknown";
}
separator defines exactly six whitespace bytes. count_stream maintains word state across every read.
read errors are checked before converting the signed result. EINTR retries; EOF triggers the final-line correction.
scan_file opens, checks fstat, scans only regular files, and attempts close on all post-open paths. The first failure wins.
The shared state and ownership boundary
Keep these file names and folders when building. Long lines scroll within the code panel.
/* Leon | Original teaching project. */
#ifndef QUEUE_H
#define QUEUE_H
#include <pthread.h>
#include <stdbool.h>
#include <stddef.h>
#define QUEUE_CAPACITY 4u
/* Queue entries are job indices, not owning pointers. Protect all fields
after initialization with mutex. Destroy only after every worker joins. */
typedef struct {
size_t entries[QUEUE_CAPACITY];
size_t head;
size_t count;
bool closed;
pthread_mutex_t mutex;
pthread_cond_t not_empty;
pthread_cond_t not_full;
} WorkQueue;
/* Runtime synchronization failures terminate the process; see README. */
void thread_check(int error, const char *operation);
int queue_init(WorkQueue *queue);
bool queue_push(WorkQueue *queue, size_t job);
bool queue_pop(WorkQueue *queue, size_t *job);
void queue_close(WorkQueue *queue);
void queue_destroy(WorkQueue *queue);
#endif
entries holds four size_t indices, not file descriptors or heap objects.
head, count, and closed describe the queue state; one mutex protects them together.
The two condition variables name different predicates. The API makes close separate from destruction.
Locked ring transitions and close-and-drain behavior
Keep these file names and folders when building. Long lines scroll within the code panel.
/* Leon | Original teaching project. */
#include "queue.h"
#include <stdio.h>
#include <stdlib.h>
void thread_check(int error, const char *operation) {
if (error != 0) {
fprintf(stderr, "fatal: %s failed (pthread error %d)\n", operation, error);
/* Continuing could access unprotected data or strand a waiter.
This teaching policy exits the process, without orderly cleanup. */
exit(EXIT_FAILURE);
}
}
int queue_init(WorkQueue *queue) {
queue->head = 0;
queue->count = 0;
queue->closed = false;
int error = pthread_mutex_init(&queue->mutex, NULL);
if (error != 0) {
return error;
}
error = pthread_cond_init(&queue->not_empty, NULL);
if (error != 0) {
thread_check(pthread_mutex_destroy(&queue->mutex), "mutex destroy");
return error;
}
error = pthread_cond_init(&queue->not_full, NULL);
if (error != 0) {
thread_check(pthread_cond_destroy(&queue->not_empty), "condition destroy");
thread_check(pthread_mutex_destroy(&queue->mutex), "mutex destroy");
}
return error;
}
bool queue_push(WorkQueue *queue, size_t job) {
thread_check(pthread_mutex_lock(&queue->mutex), "mutex lock");
while (queue->count == QUEUE_CAPACITY && !queue->closed) {
thread_check(pthread_cond_wait(&queue->not_full, &queue->mutex),
"wait not_full");
}
if (queue->closed) {
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
return false;
}
size_t tail = (queue->head + queue->count) % QUEUE_CAPACITY;
queue->entries[tail] = job;
queue->count++;
thread_check(pthread_cond_signal(&queue->not_empty), "signal not_empty");
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
return true;
}
bool queue_pop(WorkQueue *queue, size_t *job) {
thread_check(pthread_mutex_lock(&queue->mutex), "mutex lock");
while (queue->count == 0 && !queue->closed) {
thread_check(pthread_cond_wait(&queue->not_empty, &queue->mutex),
"wait not_empty");
}
if (queue->count == 0) {
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
return false;
}
*job = queue->entries[queue->head];
queue->head = (queue->head + 1) % QUEUE_CAPACITY;
queue->count--;
thread_check(pthread_cond_signal(&queue->not_full), "signal not_full");
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
return true;
}
void queue_close(WorkQueue *queue) {
thread_check(pthread_mutex_lock(&queue->mutex), "mutex lock");
queue->closed = true;
/* All empty-queue consumers must learn that no future job can arrive. */
thread_check(pthread_cond_broadcast(&queue->not_empty), "broadcast not_empty");
thread_check(pthread_cond_broadcast(&queue->not_full), "broadcast not_full");
thread_check(pthread_mutex_unlock(&queue->mutex), "mutex unlock");
}
void queue_destroy(WorkQueue *queue) {
thread_check(pthread_cond_destroy(&queue->not_full), "condition destroy");
thread_check(pthread_cond_destroy(&queue->not_empty), "condition destroy");
thread_check(pthread_mutex_destroy(&queue->mutex), "mutex destroy");
}
queue_init initializes components in order and unwinds earlier components if a later initialization fails.
Compare push and pop: both lock, wait in while, modify a protected state, notify the opposite side, and unlock.
queue_close broadcasts without discarding entries. queue_destroy is allowed only after all users have finished; runtime synchronization errors fail fast.
CLI, lifetime management, worker loop, and ordered output
Keep these file names and folders when building. Long lines scroll within the code panel.
/* Leon | Original teaching project. */
#include "queue.h"
#include "stats.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#define MAX_WORKERS 16u
#define MAX_JOBS 128u
typedef struct {
WorkQueue *queue;
char **paths;
FileStats *results;
} WorkerContext;
static bool parse_workers(const char *text, size_t *workers) {
size_t value = 0;
if (*text == '\0') {
return false;
}
for (const unsigned char *p = (const unsigned char *)text; *p != '\0'; p++) {
if (*p < '0' || *p > '9') {
return false;
}
value = value * 10 + (size_t)(*p - '0');
if (value > MAX_WORKERS) {
return false;
}
}
if (value == 0) {
return false;
}
*workers = value;
return true;
}
static void *consume(void *argument) {
WorkerContext *context = argument;
size_t job;
while (queue_pop(context->queue, &job)) {
/* Exactly one pop receives each index. No other worker writes this
result, and main waits for all joins before reading any result. */
scan_file(context->paths[job], &context->results[job]);
}
return NULL;
}
static void join_workers(pthread_t *threads, size_t count) {
for (size_t i = 0; i < count; i++) {
thread_check(pthread_join(threads[i], NULL), "thread join");
}
}
static int print_results(char **paths, FileStats *results, size_t count) {
size_t failures = 0;
for (size_t i = 0; i < count; i++) {
FileStats *result = &results[i];
if (result->status == SCAN_OK) {
printf("job=%zu bytes=%zu lines=%zu words=%zu\n",
i + 1, result->bytes, result->lines, result->words);
} else {
const char *kind = scan_status_name(result->status);
printf("job=%zu error=%s\n", i + 1, kind);
fprintf(stderr, "job %zu (%s): %s", i + 1, paths[i], kind);
if (result->system_error != 0) {
fprintf(stderr, ": %s", strerror(result->system_error));
}
fputc('\n', stderr);
failures++;
}
}
printf("summary files=%zu ok=%zu failed=%zu\n",
count, count - failures, failures);
if (fflush(stdout) != 0 || ferror(stdout)) {
fputs("output failed\n", stderr);
return EXIT_FAILURE;
}
return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
}
int main(int argc, char **argv) {
size_t workers;
if (argc < 2 || !parse_workers(argv[1], &workers) ||
(size_t)(argc - 2) > MAX_JOBS) {
fputs("usage: thread_pipeline WORKERS [FILE ...]\n"
"WORKERS: 1..16; FILE: 0..128 regular files, each <= 8388608 bytes\n",
stderr);
return 2;
}
size_t jobs = (size_t)(argc - 2);
WorkQueue queue;
int error = queue_init(&queue);
if (error != 0) {
fprintf(stderr, "queue initialization failed (pthread error %d)\n", error);
return EXIT_FAILURE;
}
FileStats results[MAX_JOBS] = {{0}};
pthread_t threads[MAX_WORKERS];
WorkerContext context = {&queue, argv + 2, results};
size_t started = 0;
for (; started < workers; started++) {
error = pthread_create(&threads[started], NULL, consume, &context);
if (error != 0) {
fprintf(stderr, "thread creation failed after %zu workers "
"(pthread error %d)\n", started, error);
/* No jobs are published until the whole pool exists. */
queue_close(&queue);
join_workers(threads, started);
queue_destroy(&queue);
return EXIT_FAILURE;
}
}
for (size_t job = 0; job < jobs; job++) {
if (!queue_push(&queue, job)) {
fputs("internal error: queue closed during production\n", stderr);
queue_close(&queue);
join_workers(threads, started);
queue_destroy(&queue);
return EXIT_FAILURE;
}
}
queue_close(&queue);
join_workers(threads, started);
queue_destroy(&queue);
return print_results(argv + 2, results, jobs);
}
parse_workers accepts decimal digits only and rejects as soon as the value exceeds 16. The bounded intermediate value avoids arithmetic overflow.
Main initializes one context and creates every worker before publishing indices. The started prefix is the only valid join list on failure.
consume writes the slot matching its popped index. Main closes, joins, destroys, then print_results reports in input order and returns failure if any file failed.
Strict build, isolated tests, and local cleanup
Keep these file names and folders when building. Long lines scroll within the code panel.
# Leon | Original teaching project.
CC = clang
CPPFLAGS = -D_POSIX_C_SOURCE=200809L
CFLAGS = -std=c11 -Wall -Wextra -Werror -pedantic -O2 -pthread
LDFLAGS = -pthread
.PHONY: all test clean
all: thread_pipeline
thread_pipeline: main.o queue.o stats.o
$(CC) $(LDFLAGS) -o $@ main.o queue.o stats.o
main.o: main.c queue.h stats.h
queue.o: queue.c queue.h
stats.o: stats.c stats.h
test:
python3 test.py
clean:
rm -f thread_pipeline main.o queue.o stats.o
The default compiler is clang with C11, POSIX feature selection, pthread support, and warnings treated as errors.
Header dependencies rebuild the affected translation units. -pthread is used during compilation and linking.
make test delegates to python3 test.py, which builds in a temporary copy. make clean removes only this project’s ordinary local build products.
Independent oracle, boundary tests, stress, and controlled faults
Keep these file names and folders when building. Long lines scroll within the code panel.
#!/usr/bin/env python3
# Leon | Original teaching project.
"""Build only in a temporary copy; use independent byte oracles and timeouts."""
import argparse
import os
from pathlib import Path
import re
import shutil
import subprocess
import tempfile
SOURCE = Path(__file__).resolve().parent
FLAGS = ["-D_POSIX_C_SOURCE=200809L", "-std=c11", "-Wall", "-Wextra",
"-Werror", "-pedantic", "-O2", "-pthread"]
LIMIT = 8 * 1024 * 1024
def command(args, cwd, timeout=20):
return subprocess.run(args, cwd=cwd, text=True, capture_output=True,
timeout=timeout, check=False)
def successful(args, cwd, timeout=30):
result = command(args, cwd, timeout)
assert result.returncode == 0, (args, result.stdout, result.stderr)
return result
def expected_line(index, data):
# A regular expression oracle, independent of the C in_word state machine.
lines = data.count(b"\n") + int(bool(data) and not data.endswith(b"\n"))
words = len(re.findall(rb"[^ \t\r\n\v\f]+", data))
return f"job={index} bytes={len(data)} lines={lines} words={words}\n"
def expect_files(root, files, workers, binary="thread_pipeline"):
result = successful([f"./{binary}", str(workers), *files], root)
expected = "".join(expected_line(i, (root / path).read_bytes())
for i, path in enumerate(files, 1))
expected += f"summary files={len(files)} ok={len(files)} failed=0\n"
assert result.stdout == expected, (result.stdout, expected)
assert result.stderr == "", result.stderr
return result.stdout
def variant(root, name, source, define, shim, flags):
# Only a temporary translation unit is instrumented; shipped C is unchanged.
(root / f"{name}_shim.c").write_text(shim, encoding="utf-8")
successful(["clang", *flags, f"-D{define}", "-c", source,
"-o", f"{name}.o"], root)
successful(["clang", *flags, "-c", f"{name}_shim.c", "-o",
f"{name}_shim.o"], root)
objects = [f"{name}.o" if item == source else item.replace(".c", ".o")
for item in ("main.c", "queue.c", "stats.c")]
successful(["clang", *flags, *objects, f"{name}_shim.o", "-o", name], root)
def run_tests(root, flags):
successful(["make", "all", "CFLAGS=" + " ".join(flags),
"LDFLAGS=" + " ".join(flags)], root)
supplied = [f"fixtures/{name}.txt" for name in
("morning", "evening", "blank", "empty", "spaces")]
expect_files(root, [], 16)
expect_files(root, supplied[:1], 1)
expect_files(root, supplied[:2], 16)
expect_files(root, supplied, 4)
print("PASS empty queue, empty file, and fewer/more jobs than capacity")
# Hard-coded expected counts catch an error shared by both implementations.
exact = successful(["./thread_pipeline", "3", *supplied], root)
assert exact.stdout == (
"job=1 bytes=37 lines=2 words=6\n"
"job=2 bytes=26 lines=2 words=5\n"
"job=3 bytes=2 lines=2 words=0\n"
"job=4 bytes=0 lines=0 words=0\n"
"job=5 bytes=6 lines=2 words=0\n"
"summary files=5 ok=5 failed=0\n")
print("PASS exact fixture counts including final unterminated line")
binary_bytes = b"x" * 4095 + b"YZ\x00\xff \r\n\tfinal"
(root / "boundary.bin").write_bytes(binary_bytes)
expect_files(root, ["boundary.bin"], 2)
print("PASS word crossing read-buffer boundary and non-ASCII bytes")
for args in ([], ["0"], ["17"], ["-1"], ["+2"], ["2x"], [""],
[" 2"], ["99999999999999999999999999"],
["1", *([supplied[0]] * 129)]):
result = command(["./thread_pipeline", *args], root)
assert result.returncode == 2 and result.stdout == "", result
assert result.stderr.startswith("usage: thread_pipeline WORKERS"), result
expect_files(root, supplied[:1], "0002")
expect_files(root, [supplied[0]] * 128, 16)
print("PASS malformed CLI, accepted leading zeros, and 128-job boundary")
result = command(["./thread_pipeline", "4", supplied[0], "missing.txt",
supplied[1]], root)
assert result.returncode == 1, result
assert result.stdout == (expected_line(1, (root / supplied[0]).read_bytes()) +
"job=2 error=open\n" +
expected_line(3, (root / supplied[1]).read_bytes()) +
"summary files=3 ok=2 failed=1\n"), result.stdout
assert "job 2 (missing.txt): open:" in result.stderr, result.stderr
print("PASS missing file preserves other results and returns failure")
os.mkfifo(root / "pipe")
result = command(["./thread_pipeline", "2", "fixtures", "pipe"], root)
assert result.returncode == 1, result
assert result.stdout == ("job=1 error=not-regular\n"
"job=2 error=not-regular\n"
"summary files=2 ok=0 failed=2\n"), result.stdout
print("PASS directory and FIFO rejected without waiting for a writer")
(root / "limit.txt").write_bytes(b"z" * LIMIT)
expect_files(root, ["limit.txt"], 1)
with (root / "limit.txt").open("ab") as handle:
handle.write(b"z")
result = command(["./thread_pipeline", "1", "limit.txt"], root)
assert result.returncode == 1, result
assert result.stdout == ("job=1 error=too-large\n"
"summary files=1 ok=0 failed=1\n"), result.stdout
print("PASS exact byte limit and oversized input without partial statistics")
stress = []
for i in range(37):
path = f"stress-{i:02}.txt"
data = (b"alpha\tbeta\n\n" * (i * 71)) + (b"tail" if i % 2 else b"")
(root / path).write_bytes(data)
stress.append(path)
reference = expect_files(root, stress, 1)
for repeat in range(12):
for workers in (1, 2, 4, 16):
output = expect_files(root, stress, workers)
assert output == reference, (repeat, workers)
print("PASS 48 repeated schedules: 37 jobs, workers 1/2/4/16, stable output")
probe = r'''/* Leon | Original teaching project. */
#include "queue.h"
#include <assert.h>
int main(void) {
WorkQueue queue;
assert(queue_init(&queue) == 0);
for (size_t i = 0; i < QUEUE_CAPACITY; i++) {
assert(queue_push(&queue, i + 10));
}
queue_close(&queue);
assert(!queue_push(&queue, 99));
for (size_t i = 0; i < QUEUE_CAPACITY; i++) {
size_t job;
assert(queue_pop(&queue, &job));
assert(job == i + 10);
}
size_t job;
assert(!queue_pop(&queue, &job));
queue_close(&queue);
queue_destroy(&queue);
return 0;
}
'''
(root / "queue_probe.c").write_text(probe, encoding="utf-8")
successful(["clang", *flags, "queue_probe.c", "queue.o", "-o", "queue_probe"], root)
successful(["./queue_probe"], root)
print("PASS queue close drains FIFO entries and rejects new work")
for after in (0, 2):
shim = r'''/* Leon | Original teaching project. */
#include <errno.h>
#include <pthread.h>
static unsigned int calls;
int fixture_create(pthread_t *thread, const pthread_attr_t *attributes,
void *(*start)(void *), void *argument) {
if (calls++ == FAIL_AFTER) {
return EAGAIN;
}
return pthread_create(thread, attributes, start, argument);
}
'''.replace("FAIL_AFTER", str(after))
name = f"create_failure_{after}"
variant(root, name, "main.c", "pthread_create=fixture_create", shim, flags)
result = command([f"./{name}", "4", *supplied], root)
assert result.returncode == 1 and result.stdout == "", result
assert f"thread creation failed after {after} workers" in result.stderr, result
print("PASS injected creation failure before first/after two workers; no hang")
for mode in ("error", "interrupt"):
body = "errno = EIO; return -1;" if mode == "error" else (
"static int first = 1; if (first) { first = 0; errno = EINTR; return -1; }"
" return read(descriptor, buffer, count);")
shim = r'''/* Leon | Original teaching project. */
#include <errno.h>
#include <stddef.h>
#include <unistd.h>
ssize_t fixture_read(int descriptor, void *buffer, size_t count) {
(void)descriptor; (void)buffer; (void)count;
BODY
}
'''.replace("BODY", body)
name = f"read_{mode}"
variant(root, name, "stats.c", "read=fixture_read", shim, flags)
if mode == "error":
result = command([f"./{name}", "1", supplied[0]], root)
assert result.returncode == 1, result
assert result.stdout == ("job=1 error=read\n"
"summary files=1 ok=0 failed=1\n"), result.stdout
assert "read:" in result.stderr, result.stderr
else:
expect_files(root, supplied[:1], 1, name)
print("PASS injected read error and interrupted-read retry")
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--tsan", action="store_true",
help="require a working clang ThreadSanitizer runtime")
args = parser.parse_args()
flags = FLAGS + (["-O1", "-g", "-fsanitize=thread"] if args.tsan else [])
with tempfile.TemporaryDirectory(prefix="guoliang-thread-test-") as directory:
root = Path(directory) / "project"
shutil.copytree(SOURCE, root, ignore=shutil.ignore_patterns(
"*.o", "thread_pipeline", "__pycache__"))
run_tests(root, flags)
print("All pipeline tests passed" + (" (ThreadSanitizer)." if args.tsan else "."))
if __name__ == "__main__":
main()
TemporaryDirectory plus copytree isolates every build and synthetic generated input. subprocess uses argument lists and explicit timeouts.
A byte-oriented regular expression counts words independently of the C state machine. Hard-coded fixture output checks the oracle too.
Read the 48-run loop, close-and-drain queue probe, and temporary pthread_create/read shims. --tsan requires a working sanitizer and does not silently skip failures.
Synthetic note with a final LF
Keep these file names and folders when building. Long lines scroll within the code panel.
Robotics club
Build small test often
The first line is Robotics club.
The second line is Build small test often and ends with LF.
Expect 37 bytes, 2 logical lines, and 6 words; the final LF adds no empty line.
Synthetic note without a final LF
Keep these file names and folders when building. Long lines scroll within the code panel.
Check all wheels
Then restA tab separates Check from all.
The second line Then rest reaches EOF without LF.
Expect 26 bytes, 2 logical lines, and 5 words; EOF supplies the second line count.
Two newline bytes and no words
Keep these file names and folders when building. Long lines scroll within the code panel.
Invisible-byte view (hexadecimal): 0a 0a
The file is exactly LF followed by LF.
Each LF finishes a logical line even when that line has no letters.
Expect 2 bytes, 2 lines, and 0 words; do not add a third line after EOF.
A zero-byte fixture
Keep these file names and folders when building. Long lines scroll within the code panel.
Invisible-byte view (hexadecimal): empty file; zero bytes
This file deliberately has no content and no author header because any bytes would destroy the test.
The scanner sees EOF on its first read and never enters the byte loop.
Expect 0 bytes, 0 lines, and 0 words, while the file itself is still a successful job.
All six separators in a fixed order
Keep these file names and folders when building. Long lines scroll within the code panel.
Invisible-byte view (hexadecimal): 20 09 0d 0a 0b 0c
The bytes are space, tab, CR, LF, vertical tab, and form feed.
The suffix after LF contains bytes, although a terminal may not display them visibly.
Expect 6 bytes, 2 logical lines, and 0 words. This tests the exact separator definition, not locale behavior.
Complete English command-line guide
Keep these file names and folders when building. Long lines scroll within the code panel.
# Project 3 · A bounded file-statistics pipeline
Leon | Original teaching project. [中文说明](README.zh.md)
Mina helps a school robotics club archive its daily notes. She receives a list of small text files and wants a count of their bytes, lines, and words. Some notes are empty. One ends without a newline. A missing file must remain visible in the report while the other notes are still counted. She wants to try several workers, but the report must always follow her original file list.
Her first useful tool counts one file precisely. She then places numbered jobs into a four-slot queue. Workers take those numbers and open their own files. The queue applies backpressure: when all four slots are occupied, Mina's producer waits instead of allocating another slot. A worker's completion time never determines its output position. Main prints the numbered results only after all workers have finished.
This is an original guided engineering project, not a supplied assignment solution. C11, POSIX threads, Clang, Make, and Python 3 are required. The source targets macOS and Linux; no network or third-party Python packages are used.
## Build and run
From this project directory:
```sh
make
./thread_pipeline 3 fixtures/morning.txt fixtures/evening.txt fixtures/empty.txt
```
Program stdout (the compiler commands printed by `make` are separate):
```text
job=1 bytes=37 lines=2 words=6
job=2 bytes=26 lines=2 words=5
job=3 bytes=0 lines=0 words=0
summary files=3 ok=3 failed=0
```
An empty file list is valid. Workers wake when main closes the empty queue:
```sh
./thread_pipeline 16
```
```text
summary files=0 ok=0 failed=0
```
A missing file does not discard completed work:
```sh
./thread_pipeline 2 fixtures/morning.txt fixtures/missing.txt fixtures/evening.txt
```
```text
job=1 bytes=37 lines=2 words=6
job=2 error=open
job=3 bytes=26 lines=2 words=5
summary files=3 ok=2 failed=1
```
The first two runs exit with 0. The missing-file run exits with 1 and also writes a diagnostic to stderr. Its operating-system error wording may vary. Job 2 means the second file argument, not worker 2. Files may repeat; each argument is a separate job.
```sh
./thread_pipeline 0 fixtures/morning.txt
```
This prints usage to stderr, prints no stdout, and exits with 2. WORKERS must contain only decimal digits and represent 1 through 16; leading zeros are accepted. At most 128 file arguments are accepted. There is no option parser: every argument after WORKERS is a file path.
## Decide what is being counted
* Bytes count every byte read, including whitespace and zero bytes. This is not a Unicode character count.
* Lines count LF bytes (`\n`) plus one when a nonempty file ends with a different byte. Thus empty is 0 lines, `a` is 1, `a\n` is 1, and `\n\n` is 2. A final newline does not invent another empty line. This differs from tools that count only newline characters.
* Words are maximal runs separated by exactly space, tab, LF, CR, vertical tab, or form feed. The definition is independent of locale. Zero bytes and non-ASCII bytes are ordinary non-separators, not string terminators or decoded letters.
* Each regular file may contain at most 8,388,608 bytes (8 MiB). A larger file produces `error=too-large`; its partial counters are not presented as a complete answer. `in_word` survives each 4096-byte read, so a word crossing a buffer boundary stays one word.
The synthetic fixtures are 37-byte morning notes, 26-byte evening notes without a final LF, two blank lines, a zero-byte file, and six whitespace bytes. The whitespace fixture has two logical lines because bytes follow its LF. These definitions are the contract against which tests compare the program.
## Read the implementation in five stages
1. **Specify a file result.** Read `stats.h`, then `separator` and `count_stream` in `stats.c`. Track `bytes`, `lines`, `words`, `in_word`, and `last` on `a b\n`. The loop counts word beginnings. EOF finishes the final logical line exactly once. `read` returns a signed count; check for an error before converting it to `size_t`.
2. **Build the bounded queue.** Read `queue.h` and `queue_push`. `head` identifies the next occupied slot. The next insertion is `(head + count) % 4`. The invariant is `0 <= count <= 4`. A push waits while full and open. A pop waits while empty and open. Every check and every mutation uses the same mutex. `pthread_cond_wait` releases that mutex while waiting and reacquires it before returning. A notification asks a waiter to check again; it does not reserve a slot. Use `while`, including after a spurious wakeup.
3. **Give workers exclusive results.** Read `WorkerContext` and `consume` in `main.c`. Main gives each input one index exactly once. A pop removes that index while holding the queue lock, so only one worker receives it. The worker owns `results[index]` during scanning and opens/closes its own descriptor. File reads happen outside the queue lock. This lets other workers take jobs while a scan is running.
4. **Close, drain, and join.** Read `queue_close` and the final half of `main`. Closing forbids future pushes but does not remove queued work. A pop returns false only when the queue is empty and closed. Broadcasting wakes every idle consumer, including all sixteen consumers for zero jobs. Main joins each successfully created worker before printing or destroying shared objects.
5. **Validate ordinary and failed runs.** Read `parse_workers`, startup cleanup, `print_results`, and `test.py`. No jobs are published until every requested worker exists. If thread creation fails, close the still-empty queue, join only the workers actually created, destroy the queue, and return failure. A file error occupies its own result slot while other jobs continue. Stable output is tested against independent expected values, never against worker timing.
## Ownership and execution phases
```text
main: initialize queue + context + result storage
|
+--> create all workers --> publish indices --> close queue
| | |
workers: +--> wait/pop -------+--> scan --> pop --+--> drain --> return
|
main: join every created worker --> destroy queue --> print slots 0..J-1
creation failure: close empty queue --> join created workers --> destroy --> exit 1
```
The queue, context, and result array live in main's stack frame until after all joins. `argv` paths live throughout the process and are borrowed without mutation. Queue entries own no file or allocation; they are indices. A worker owns its local read buffer and opened file descriptor. `scan_file` attempts to close that descriptor on every post-open path. Main owns all joinable thread handles and joins only successful creations.
Initialization before `pthread_create` makes the initialized context available to workers. The mutex orders queue reads and writes and safely hands an index from producer to consumer. Distinct array elements are distinct objects: workers never write the same `FileStats`. Main does not read any result until successful `pthread_join` calls have synchronized it with every worker's completion. These ownership and phase rules remove the need for a result mutex. Merely promising that main will “probably run later” would not be sufficient.
## Errors and deliberate limits
`scan_file` reports open, metadata, non-regular-file, read, size-limit, and close errors. The first failure is preserved. Interrupted reads (`EINTR`) retry; ordinary short reads are processed and followed by another read. Nonblocking open avoids waiting for a FIFO writer before rejecting non-regular inputs with `fstat`. This is intended for the supplied regular fixtures and other stable regular files, not arbitrary devices, hostile paths, live-changing files, or a secure filesystem sandbox.
Queue initialization returns a checked error and unwinds initialized components. Thread creation has orderly cleanup. Unexpected runtime mutex/condition/join/destroy errors instead use a documented fail-fast policy: print an error and terminate the process, without promising graceful cleanup or a complete report. These pthread functions return their error code directly. The program does not use `errno` to interpret them. An unsuccessful `close` is reported without retrying an uncertain descriptor state.
There is no cancellation, signal-driven shutdown, or per-file wall-clock deadline. A slow filesystem may still delay a regular-file read, and repeated signals may repeatedly interrupt it. Tests have subprocess timeouts; the program itself does not guarantee a time bound. Input files must remain unchanged for reproducible counts. Native Windows is outside this POSIX target. Checked stdout failures return failure; a broken pipe can also terminate the process through the usual SIGPIPE behavior.
Counting costs O(B + J), where B is the bytes scanned and J is the job count. Queue operations use constant storage and constant work apart from waiting. The queue has four indices; result storage has a fixed upper bound of 128 entries; there are at most 16 workers, each with one 4096-byte read buffer plus its runtime thread stack. Storage does not grow with file length. A byte-limit check can read one extra buffer before rejecting an oversized file. Tiny fixtures provide correctness evidence, not evidence of a speedup. Storage contention and thread startup can make extra workers slower.
## Test and investigate
```sh
make test
python3 test.py --tsan
make clean
```
`make test` builds and runs in a temporary copy, even if the source directory already contains a local build. It never changes source fixtures. Every subprocess has a timeout. The optional second command requires a working Clang ThreadSanitizer compiler and runtime; an unavailable or failing sanitizer run is an error, not a silent pass.
The test suite checks exact fixture output, no jobs, empty files, one and sixteen workers, below/above queue capacity, maximum job count, malformed worker counts, missing files, directory/FIFO rejection, buffer-boundary words, binary bytes, and both sides of the file-size limit. A separate queue probe checks FIFO order, close-and-drain, and push-after-close rejection. Forty-eight repeated runs scan 37 bounded synthetic files with 1, 2, 4, and 16 workers. All compare with an independent byte/regular-expression oracle and one-worker output.
Temporary test-only C shims inject thread creation failure before the first worker and after two workers, a read failure, and one interrupted read. They exercise cleanup without exhausting system resources. These compile-time substitutions are not switches in the shipped program. Stress tests and ThreadSanitizer can expose bugs; neither proves that every possible schedule is correct. The invariant and happens-before argument remain necessary.
If a zero-job run hangs, inspect the close broadcast and the `count == 0 && !closed` predicate. If a final unterminated line disappears, inspect the EOF correction. If output changes order, look for worker-side printing or a missing join. If words split at a 4096-byte boundary, check whether `in_word` was incorrectly reset inside the read loop. If a missing file appears as zero counts, inspect whether the status is tested before printing counters.
## File map and next experiments
`main.c` owns CLI validation, pool startup, publication, joins, and output. `queue.h` declares the bounded state; `queue.c` implements its locked transitions. `stats.h` defines result/error data; `stats.c` scans one regular file. `Makefile` supplies strict build/test/clean targets. `test.py` provides the independent oracle, stress checks, and fault injection. Both README files explain the same project in different languages. The five files in `fixtures/` are synthetic inputs with intentionally different boundaries.
For a first extension, make queue capacity configurable with a bounded range, then test capacities 1, 4, and 8 against unchanged result output. For a second, add measured performance experiments on fixed, larger local fixtures. Record total bytes, compiler options, worker count, elapsed time, repeated runs, and filesystem-cache conditions; compute throughput as bytes divided by seconds. Accept a speed claim only when measurements support it. Keep timing outside deterministic result output.
## Primary references
The design explanations are original. These official POSIX pages support the API contracts:
* [Condition waits and predicate rechecking](https://pubs.opengroup.org/onlinepubs/9799919799/functions/pthread_cond_clockwait.html)
* [pthread_join](https://pubs.opengroup.org/onlinepubs/9799919799.2024edition/functions/pthread_join.html)
* [Memory synchronization, section 4.15.2](https://pubs.opengroup.org/onlinepubs/9799919799/basedefs/V1_chap04.html)
* [pthread_create, Issue 6 API contract](https://pubs.opengroup.org/onlinepubs/000095399/functions/pthread_create.html)
* [read](https://pubs.opengroup.org/onlinepubs/9799919799/functions/read.html)
Start with the story and exact build/run commands, then check the counting definitions against each fixture.
Follow the five reading stages and the phase diagram; connect every resource to its owner and end of lifetime.
Read the failure policy, complexity, test commands, and bounded extensions before changing the design.
Equivalent Chinese command-line guide
Keep these file names and folders when building. Long lines scroll within the code panel.
# 项目 3 · 有界文件统计流水线
Leon | 原创教学项目。[English guide](README.md)
米娜帮学校机器人社整理每天的笔记。她收到一组小文本文件，需要统计字节数、行数和单词数。有些文件是空的，有些文件末尾没有换行。某个文件不存在时，报告也必须说明问题，并保留其他文件的统计。她想尝试多个工作线程，但报告的顺序必须与输入文件列表一致。
她先做出能准确统计单个文件的工具，然后把编号任务放入只有四个位置的队列。工作线程取出编号，各自打开文件。队列满时，主线程等待空位，不再分配新位置。这就是背压。任务先完成不代表先输出。主线程等所有工作线程结束后，按编号打印结果。
这是原创的引导式工程项目，不是课程作业答案。需要 C11、POSIX 线程、Clang、Make 和 Python 3。源码面向 macOS 与 Linux，不需要网络，也不需要第三方 Python 库。
## 编译与运行
在本项目目录中执行：
```sh
make
./thread_pipeline 3 fixtures/morning.txt fixtures/evening.txt fixtures/empty.txt
```
程序的标准输出如下。`make` 显示的编译命令不属于程序输出。
```text
job=1 bytes=37 lines=2 words=6
job=2 bytes=26 lines=2 words=5
job=3 bytes=0 lines=0 words=0
summary files=3 ok=3 failed=0
```
文件列表可以为空。主线程关闭空队列后，等待的线程会醒来并退出。
```sh
./thread_pipeline 16
```
```text
summary files=0 ok=0 failed=0
```
某个文件不存在，不会丢弃其他结果。
```sh
./thread_pipeline 2 fixtures/morning.txt fixtures/missing.txt fixtures/evening.txt
```
```text
job=1 bytes=37 lines=2 words=6
job=2 error=open
job=3 bytes=26 lines=2 words=5
summary files=3 ok=2 failed=1
```
前两次运行的退出码为 0。缺少文件的运行退出码为 1，还会向标准错误写入诊断。系统错误文字因平台而异。`job=2` 表示第二个文件参数，不是第二个工作线程。相同路径可以重复出现，每个参数都算独立任务。
```sh
./thread_pipeline 0 fixtures/morning.txt
```
该命令只向标准错误打印用法，不产生标准输出，退出码为 2。WORKERS 只能包含十进制数字，数值必须在 1 到 16 之间，可以有前导零。最多接受 128 个文件参数。这里不解析选项，WORKERS 后的每个参数都当作文件路径。
## 先确定统计定义
* 字节数计算读到的每个字节，包括空白和零字节。它不是 Unicode 字符数。
* 行数等于 LF 字节（`\n`）的数量；若文件非空且最后一个字节不是 LF，再加一。空文件有 0 行，`a` 有 1 行，`a\n` 有 1 行，`\n\n` 有 2 行。末尾的换行不会凭空增加一行。这与只计算换行符数量的工具不同。
* 单词是由非分隔字节组成的最长连续段。分隔字节仅为空格、制表符、LF、CR、垂直制表符和换页符，定义不受地区设置影响。零字节和非 ASCII 字节都视为普通非分隔字节；不会当作字符串结尾，也不会解码成文字。
* 每个普通文件最多 8,388,608 字节，即 8 MiB。超过上限时输出 `error=too-large`，不把部分计数冒充完整统计。`in_word` 状态在每次 4096 字节的读取之间保留，因此跨缓冲区的一个单词仍只计算一次。
合成样例包括 37 字节的早间笔记、26 字节且末尾没有 LF 的晚间笔记、两行空行、零字节文件，以及六个空白字节。空白样例的 LF 后还有字节，所以按本项目定义有两行。测试必须以这些约定为准。
## 分五步阅读实现
1. **定义单个文件的结果。** 先读 `stats.h`，再读 `stats.c` 中的 `separator` 和 `count_stream`。用 `a b\n` 手动追踪 `bytes`、`lines`、`words`、`in_word` 和 `last`。循环统计单词的起点。到达文件末尾后，最多补计最后一行。`read` 的返回类型可以表示负数，必须先检查错误，再转换成 `size_t`。
2. **建立有界队列。** 阅读 `queue.h` 和 `queue_push`。`head` 指向下一个要取出的元素，插入位置为 `(head + count) % 4`。不变量为 `0 <= count <= 4`。队列满且未关闭时，生产者等待；队列空且未关闭时，消费者等待。检查和修改共用同一把互斥锁。`pthread_cond_wait` 在等待时释放锁，返回前重新获得锁。通知只表示需要重新检查条件，并不预留位置，因此必须用 `while`，也要允许虚假唤醒。
3. **让每个结果只有一个写入者。** 阅读 `main.c` 中的 `WorkerContext` 和 `consume`。主线程给每个输入分配一个编号，并且只入队一次。取出编号时持有队列锁，因此只有一个工作线程能得到它。扫描期间，该线程独占 `results[index]`，并拥有自己打开的文件描述符。文件读取发生在队列锁之外，让其他线程能够继续取任务。
4. **关闭、排空、等待结束。** 阅读 `queue_close` 和 `main` 后半段。关闭只禁止新任务，不删除队列中的旧任务。队列既空又关闭时，`queue_pop` 才返回 false。广播使所有空闲消费者醒来，包括零任务时的十六个线程。主线程等待每个成功创建的线程结束后，才销毁共享对象并打印。
5. **验证成功和失败路径。** 阅读 `parse_workers`、启动清理、`print_results` 和 `test.py`。全部工作线程创建成功后才发布任务。若创建失败，关闭仍为空的队列，只等待已经成功创建的线程，然后销毁队列并返回失败。文件错误存入该文件自己的结果槽，其他任务继续。测试比较独立算出的正确结果，不依赖线程的速度或执行先后。
## 所有权和执行阶段
```text
主线程：初始化队列、上下文、结果数组
|
+--> 创建全部工作线程 --> 发布编号 --> 关闭队列
| | |
工作线程： +--> 等待/取任务 -+--> 扫描 --> 取任务 --> 排空 --> 返回
|
主线程：等待每个已创建线程结束 --> 销毁队列 --> 按 0..J-1 打印结果
创建失败：关闭空队列 --> 等待已创建线程 --> 销毁 --> 退出码 1
```
队列、上下文和结果数组保存在主线程的栈帧中，所有 join 完成后才结束其生命周期。`argv` 中的路径在进程运行期间保持有效，线程只借用，不修改。队列元素只是编号，不拥有文件或动态分配。每个工作线程拥有自己的读取缓冲区和文件描述符。成功打开后，`scan_file` 的每条路径都会尝试关闭描述符。主线程拥有可等待的线程句柄，只 join 成功创建的线程。
`pthread_create` 之前的初始化使工作线程能够访问正确的上下文。互斥锁为队列的读写建立先后关系，也把编号安全地交给消费者。数组的不同元素是不同对象，工作线程不会同时写同一个 `FileStats`。主线程在所有 `pthread_join` 成功后才读取结果，这些同步操作让工作线程完成的写入对主线程可见。所有权和阶段限制共同省去了结果锁。仅仅期待主线程“大概会晚一点运行”不能保证正确。
## 错误与有意保留的限制
`scan_file` 区分打开、元数据查询、非普通文件、读取、超出大小和关闭错误。出现多个问题时保留第一个。读取因 `EINTR` 中断会重试；短读取的有效字节会被处理，然后继续读取。非阻塞打开避免因 FIFO 没有写入方而无限等待，随后用 `fstat` 拒绝非普通文件。本项目用于所附普通样例和其他稳定的普通文件，不是任意设备、恶意路径、持续变化文件的处理器，也不是安全的文件系统沙箱。
队列初始化会检查错误并撤销已初始化的部件。线程创建失败有正常的回收路径。运行过程中若互斥锁、条件变量、join 或销毁操作意外失败，则采用明确的快速失败策略：打印错误并结束整个进程，不承诺完整报告或有序清理。pthread 函数直接返回错误码，不能用 `errno` 解读。关闭失败会报告，不会在描述符状态不确定时盲目重试。
本项目没有线程取消、信号驱动退出或单文件时间限制。慢文件系统仍可能拖延普通文件读取；连续信号也可能不断中断读取。测试给子进程设置超时，但程序本身不保证运行时间上限。要得到可重复的计数，输入文件在扫描期间不能改变。原生 Windows 不属于本 POSIX 目标。能检测到的标准输出错误会返回失败；管道读端关闭时，也可能按默认 SIGPIPE 行为直接结束进程。
计数工作量为 O(B + J)，B 为扫描字节数，J 为任务数。队列操作除等待外只需常数工作和固定存储。队列只放四个编号，结果数组最多 128 项，工作线程最多 16 个，每个线程有一个 4096 字节读取缓冲区以及系统管理的线程栈。内存不会随文件长度增长。判断文件超限时可能额外读取一个缓冲区。小样例只能验证正确性，不能证明多线程更快；存储竞争和线程启动开销反而可能让它更慢。
## 测试和定位问题
```sh
make test
python3 test.py --tsan
make clean
```
`make test` 始终在临时副本中编译和运行，即使源码目录已有本地构建产物也一样。不会修改样例。每个子进程都有超时。可选的第二条命令要求 Clang 的 ThreadSanitizer 编译器支持和运行环境可用；不可用或检测失败都会报错，不会假装通过。
测试覆盖精确样例输出、零任务、空文件、单线程和十六线程、任务数少于与多于队列容量、任务数上限、非法线程数、缺失文件、目录与 FIFO、跨缓冲区单词、二进制字节，以及文件大小上限的两侧。独立的队列探针检查先进先出、关闭后排空和关闭后拒绝入队。另有 48 次重复运行，使用 37 个有界合成文件以及 1、2、4、16 个工作线程，分别与独立字节/正则表达式算法和单线程输出比较。
测试还在临时目录编译专用 C 替身，注入第一个线程创建前失败、成功创建两个线程后失败、读取失败，以及一次中断读取。这能检验错误路径而不耗尽系统资源。编译期替换只存在于测试副本，不是正式程序的运行开关。压力测试和 ThreadSanitizer 能发现问题，但都不能证明所有调度正确；仍需要不变量与同步关系的论证。
零任务运行卡住时，检查关闭广播和 `count == 0 && !closed` 条件。末尾无换行的行漏算时，检查 EOF 后补计。输出顺序变化时，检查是否在工作线程中打印，或是否漏了 join。4096 字节边界把单词拆开时，检查 `in_word` 是否错误地放到读取循环内部初始化。缺失文件被打印成零计数时，检查打印计数前是否判断了错误状态。
## 文件导览与后续实验
`main.c` 管理命令行、线程启动、任务发布、join 和报告。`queue.h` 声明有界状态，`queue.c` 实现受锁保护的状态变化。`stats.h` 定义结果与错误，`stats.c` 扫描单个普通文件。`Makefile` 提供严格编译、测试和清理目标。`test.py` 包含独立算法、重复运行和故障注入。两个 README 用不同语言介绍同一个项目。`fixtures/` 的五个文件都是有意设计边界情况的合成输入。
第一个扩展可以把队列容量改成有界参数，并验证容量为 1、4、8 时结果不变。第二个扩展可以对固定、较大的本地样例做性能实验：记录总字节数、编译选项、线程数、耗时、重复次数和文件缓存条件，用字节数除以秒数得到吞吐量。只有测量支持时才能声称提速。时间信息应放在确定性统计输出之外。
## 官方资料
教学解释均为原创，下列官方 POSIX 页面用于核对 API 约定：
* [条件等待与条件重新检查](https://pubs.opengroup.org/onlinepubs/9799919799/functions/pthread_cond_clockwait.html)
* [pthread_join](https://pubs.opengroup.org/onlinepubs/9799919799.2024edition/functions/pthread_join.html)
* [第 4.15.2 节：内存同步](https://pubs.opengroup.org/onlinepubs/9799919799/basedefs/V1_chap04.html)
* [pthread_create：Issue 6 的 API 约定](https://pubs.opengroup.org/onlinepubs/000095399/functions/pthread_create.html)
* [read](https://pubs.opengroup.org/onlinepubs/9799919799/functions/read.html)
Use the same commands and expected stdout as in the English guide; program output remains English.
Review the Chinese explanations of predicates, ownership, queue closure, and join-based visibility.
Keep the two language guides consistent when changing limits, counting semantics, or error behavior.
Add a validated capacity argument from 1 through 8, keeping fixed maximum storage and the same mutex/condition rules. Compare capacity with worker count: they control waiting jobs and active consumers separately.
Capacities 1, 4, and 8 must produce identical output for zero jobs, 37 jobs, and missing files. Invalid capacities must fail before threads start. Explain each updated invariant.
Generate bounded larger local fixtures, measure elapsed time outside the result format, and compare 1, 2, 4, and 16 workers with identical bytes. Record compiler settings, repetitions, file sizes, and cache conditions. Throughput is total successful bytes divided by elapsed seconds.
Counts remain byte-for-byte identical. Report repeated measurements and their variability; make a speedup claim only if evidence supports it. Keep fixtures bounded and generated inside a temporary test directory.