Task3 pipeline

.clang-format

@@ -0,0 +1,7 @@
+BasedOnStyle: LLVM
+UseTab: Always
+TabWidth: 4
+IndentWidth: 4
+ColumnLimit: 100
+AllowShortIfStatementsOnASingleLine: true
+AllowShortLoopsOnASingleLine: true

.clang-tidy

@@ -0,0 +1,12 @@
+Checks: >
+  clang-diagnostic-*,
+  clang-analyzer-*,
+  bugprone-*,
+  -bugprone-easily-swappable-parameters,
+  performance-*,
+  readability-*,
+  -readability-magic-numbers,
+  -readability-function-cognitive-complexity,
+  -readability-identifier-length,
+  -readability-braces-around-statements
+HeaderFilterRegex: '^(src|include)/.*'

.github/workflows/ci.yml

@@ -0,0 +1,36 @@
+name: C CI
+
+on:
+  push:
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  build-and-test:
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Install dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y build-essential clang libcmocka-dev
+        sudo apt-get install -y build-essential clang clang-format clang-tidy libcmocka-dev bear
+
+    - name: Build
+      run: make CC=gcc
+
+    - name: Run tests
+      run: make test
+
+    - name: Check code formatting
+      run: |
+        find src/ -name '*.c' -o -name '*.h' | xargs clang-format --dry-run -Werror
+
+    - name: Generate compile_commands.json
+      run: bear -- make
+
+    - name: Static analysis
+      run: |
+        find src/ -name '*.c' | xargs clang-tidy --quiet

Makefile

@@ -0,0 +1,86 @@
+CC = gcc
+CFLAGS = -std=c11 -Wall -Wextra -O2 -D_POSIX_C_SOURCE=200809L \
+         -Iinclude -Ithird_party/stb -fopenmp -pthread
+CFLAGS_FAST = $(CFLAGS) -O3 -march=native
+CFLAGS_TEST = -std=c11 -Wall -Wextra -O0 -g -D_POSIX_C_SOURCE=200809L \
+              -Iinclude -Ithird_party/stb -fopenmp -pthread
+LDFLAGS = -lm -fopenmp -pthread
+TARGET = conv
+
+VPATH = src src/lib benchmarks
+
+SRCS = main.c conv.c cli.c filters.c imgio.c pipeline.c queue.c
+OBJS = $(SRCS:.c=.o)
+
+BENCH_DIR = benchmarks
+BENCH_BIN = conv_bench
+
+LIB_OBJS = conv.o cli.o filters.o imgio.o pipeline.o queue.o
+TEST_SRC  = tests/test_conv.c
+TEST_BIN  = test_runner
+
+all: $(TARGET)
+
+$(TARGET): $(OBJS)
+	$(CC) $(CFLAGS) $^ -o $@ $(LDFLAGS)
+
+%.o: %.c
+	$(CC) $(CFLAGS) -c $< -o $@
+
+bench_conv.o: src/lib/conv.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+bench_filters.o: src/lib/filters.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+bench_imgio.o: src/lib/imgio.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+bench_pipeline.o: src/lib/pipeline.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+bench_queue.o: src/lib/queue.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+$(BENCH_DIR)/bench_main.o: $(BENCH_DIR)/bench_main.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+$(BENCH_DIR)/benchmark.o: $(BENCH_DIR)/benchmark.c
+	$(CC) $(CFLAGS_FAST) -c $< -o $@
+
+$(BENCH_BIN): $(BENCH_DIR)/bench_main.o $(BENCH_DIR)/benchmark.o \
+              bench_conv.o bench_filters.o \
+              bench_imgio.o bench_pipeline.o bench_queue.o
+	$(CC) $(CFLAGS_FAST) $^ -o $@ $(LDFLAGS)
+
+bench: $(BENCH_BIN)
+	mkdir -p results
+	./$(BENCH_BIN) conv > results/bench.csv
+	@echo "Done: results/bench.csv"
+
+bench_pipeline: $(BENCH_BIN)
+	mkdir -p results
+	./$(BENCH_BIN) pipeline > results/bench_pipeline.csv
+	@echo "Done: results/bench_pipeline.csv"
+
+bench_all: bench bench_pipeline
+	python3 $(BENCH_DIR)/plot.py results/bench.csv results/bench_pipeline.csv
+	@echo "Plots saved to results/"
+
+test: $(TEST_BIN)
+	./$(TEST_BIN)
+
+$(TEST_BIN): $(LIB_OBJS) $(TEST_SRC)
+	$(CC) $(CFLAGS_TEST) $^ -o $@ -lm -lcmocka
+
+results:
+	mkdir -p results
+
+clean:
+	rm -f $(TARGET) $(BENCH_BIN) $(OBJS) \
+	      $(BENCH_DIR)/bench_main.o $(BENCH_DIR)/benchmark.o \
+	      bench_conv.o bench_filters.o bench_imgio.o \
+	      bench_pipeline.o bench_queue.o $(TEST_BIN)
+	rm -rf results *.o
+
+.PHONY: all test bench bench_pipeline bench_all clean results

README.md

@@ -0,0 +1,72 @@
+# Image Convolution
+
+Simple CLI tool for applying convolution filters to images.  
+Supports single-image processing and a dynamic pipeline for batch processing.
+
+## Build & Run
+
+```bash
+make                  # build
+make test             # run unit tests
+make bench            # benchmarks for parallel convolution → results/bench.csv
+make bench_pipeline   # benchmarks for pipeline convolution → results/bench_pipeline.csv
+make bench_all        # run all benchmarks & auto-generate plots
+```
+
+## Usage
+
+**Single image:**
+
+```bash
+./conv <input> <output> --filter <name> [--strategy <name>]
+```
+
+**Pipeline mode (batch):**
+
+```bash
+./conv --pipeline --filter <name> --workers <N> <in1> <out1> [<in2> <out2> ...]
+```
+
+**Filters:** `identity`, `blur`, `gaussian`, `sharpen`, `edges`, `emboss`  
+(add suffix `5/7/9` for larger kernels, e.g. `gaussian9`)
+
+**Strategies:** `sequential`, `rows`, `cols`, `pixels`, `tiles`
+
+**Formats:** BMP, PNG, JPEG
+
+## Pipeline Mode
+
+Processes multiple images concurrently using a 3-stage architecture:  
+`Reader → [bounded queue] → Convolver → [bounded queue] → Writer`
+
+Key features:
+
+- **Dynamic load balancing:** A dispatcher thread periodically checks queue fill levels and reassigns idle workers between `Reader`, `Convolver`, and `Writer` roles.
+- **Backpressure:** Fixed-capacity queues automatically block producers when consumers lag, preventing memory overflow.
+- **Worker count:** Minimum 3 threads required (auto-corrected if lower). Default: `4`.
+- **Inner parallelism:** Each convolver thread uses OpenMP. In benchmarks, inner threads are fixed to `2` to isolate pipeline scaling effects.
+
+## Example
+
+Single image:
+
+```bash
+./conv photo.jpg result.png --filter gaussian7 --strategy rows
+```
+
+Batch processing:
+
+```bash
+./conv --pipeline --filter edges5 --workers 6 \
+    a.png a_out.png b.jpg b_out.jpg c.bmp c_out.bmp
+```
+
+## Testing
+
+Tests verify convolution correctness using property-based checks (identity, composition, padding, boundary wrapping).  
+Additionally, test runs the pipeline on multiple images and **compares every output against the standard sequential mode** to guarantee bitwise identical results regardless of worker count or dynamic role switching.
+
+## Performance Analysis
+
+- **Task 2 (Single Image & Strategies):** See [benchmarks/parallel_conv_bench.md](benchmarks/parallel_conv_bench.md) for thread scaling and strategy comparison.
+- **Task 3 (Pipeline & Batch Processing):** See [benchmarks/pipeline_conv_bench.md](benchmarks/pipeline_conv_bench.md) for worker scaling, batch size analysis, and dynamic balancing results.

benchmarks/bench_main.c

@@ -0,0 +1,120 @@
+#include "benchmark.h"
+#include "filters.h"
+#include <math.h>
+#include <omp.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+static void run_conv_bench(void) {
+	const size_t IMG_SIZE = 512;
+	const char *FILTER_NAME = "gaussian5";
+	const char *strat_names[] = {"sequential", "per_pixel", "by_rows", "by_cols", "tiles"};
+	const ParallelStrategy strats[] = {STRATEGY_SEQUENTIAL, STRATEGY_PER_PIXEL, STRATEGY_BY_ROWS,
+									   STRATEGY_BY_COLS, STRATEGY_TILES};
+	const int threads[] = {1, 2, 4, 8};
+
+	fprintf(stderr, "bench conv: %zux%zu  filter=%s\n", IMG_SIZE, IMG_SIZE, FILTER_NAME);
+
+	struct Filter f = {0};
+	for (size_t k = 0; k < filter_count; ++k) {
+		if (strcmp(filter_list[k].name, FILTER_NAME) == 0) {
+			f = filter_list[k].create();
+			break;
+		}
+	}
+
+	Image *src = create_synthetic_image(IMG_SIZE, IMG_SIZE, 3);
+	Image *dst = create_synthetic_image(IMG_SIZE, IMG_SIZE, 3);
+	if (!src || !dst) {
+		fprintf(stderr, "Out of memory\n");
+		free_image(src);
+		free_image(dst);
+		return;
+	}
+
+	bench_print_header();
+
+	BenchStats baseline = bench_run_conv(src, dst, &f, STRATEGY_SEQUENTIAL, 50);
+	bench_print_csv("sequential", 1, IMG_SIZE, IMG_SIZE, FILTER_NAME, &baseline, 0.0);
+
+	for (size_t st = 1; st < 5; ++st) {
+		for (size_t ti = 0; ti < 4; ++ti) {
+			int t = threads[ti];
+			omp_set_num_threads(t);
+			BenchStats stats = bench_run_conv(src, dst, &f, strats[st], 50);
+			bench_print_csv(strat_names[st], t, IMG_SIZE, IMG_SIZE, FILTER_NAME, &stats,
+							baseline.mean);
+		}
+	}
+
+	free_image(src);
+	free_image(dst);
+	fprintf(stderr, "bench conv: Done\n");
+}
+
+static void run_pipeline_bench(void) {
+	const size_t n_images_list[] = {10, 20};
+	const size_t IMG_W = 512;
+	const size_t IMG_H = 512;
+	const char *FILTER_NAME = "gaussian5";
+	const size_t n_workers_list[] = {3, 4, 6, 10};
+	const char *strat_names[] = {"sequential", "per_pixel", "by_rows", "by_cols", "tiles"};
+	const ParallelStrategy strats[] = {STRATEGY_SEQUENTIAL, STRATEGY_PER_PIXEL, STRATEGY_BY_ROWS,
+									   STRATEGY_BY_COLS, STRATEGY_TILES};
+	const size_t N_STRATS = sizeof(strats) / sizeof(strats[0]);
+	const size_t N_WORKERS = sizeof(n_workers_list) / sizeof(n_workers_list[0]);
+	const size_t N_BATCHES = sizeof(n_images_list) / sizeof(n_images_list[0]);
+	const int REPEAT = 10;
+	const int FIXED_INNER_THREADS = 2;
+
+	omp_set_num_threads(FIXED_INNER_THREADS);
+
+	struct Filter f = {0};
+	for (size_t k = 0; k < filter_count; ++k) {
+		if (strcmp(filter_list[k].name, FILTER_NAME) == 0) {
+			f = filter_list[k].create();
+			break;
+		}
+	}
+
+	bench_print_pipeline_header();
+
+	for (size_t ni = 0; ni < N_BATCHES; ni++) {
+		size_t n_images = n_images_list[ni];
+		fprintf(stderr, "bench pipeline: n_images=%zu  %zux%zu  filter=%s  inner_threads=%d\n",
+				n_images, IMG_W, IMG_H, FILTER_NAME, FIXED_INNER_THREADS);
+
+		BenchStats baseline = bench_run_pipeline(n_images, IMG_W, IMG_H, &f, STRATEGY_SEQUENTIAL, 1,
+												 "sequential", REPEAT);
+		bench_print_pipeline_csv("sequential", 1, n_images, IMG_W, IMG_H, FILTER_NAME, "sequential",
+								 &baseline, 0.0);
+
+		for (size_t wi = 0; wi < N_WORKERS; wi++) {
+			size_t nw = n_workers_list[wi];
+			for (size_t si = 0; si < N_STRATS; si++) {
+				BenchStats stats = bench_run_pipeline(n_images, IMG_W, IMG_H, &f, strats[si], nw,
+													  "pipeline", REPEAT);
+				bench_print_pipeline_csv("pipeline", nw, n_images, IMG_W, IMG_H, FILTER_NAME,
+										 strat_names[si], &stats, baseline.mean);
+			}
+		}
+	}
+
+	fprintf(stderr, "bench pipeline: Done\n");
+	omp_set_num_threads(omp_get_max_threads());
+}
+
+int main(int argc, char *argv[]) {
+	int do_conv = 1;
+	int do_pipeline = 1;
+	if (argc == 2) {
+		do_conv = (strcmp(argv[1], "conv") == 0);
+		do_pipeline = (strcmp(argv[1], "pipeline") == 0);
+	}
+
+	if (do_conv) run_conv_bench();
+	if (do_pipeline) run_pipeline_bench();
+
+	return 0;
+}