Zero-Copy Shared-Memory Distributed Vector Engine & Custom Memory-Mapped Storage

To meet the engineering standards of top-tier research institutes like IIT or IISc, a resume project cannot rely on high-level wrappers like LangChain or stock FAISS integrations. It must demonstrate low-level systems engineering, custom concurrency primitives, optimized memory layouts, and deep hardware utilization.

Project 1

This project implements a low-latency, multi-process vector engine from scratch using POSIX shared memory (shm), explicit binary serialization (mmap), SIMD-aligned vector distance routines, and a multi-reader single-writer (MRSW) shared memory buffer.

Architecture & Engineering Highlights

• Zero-Copy IPC: Shared memory segments pass array pointers between processes via memory offsets rather than serialization/deserialization over TCP/Unix sockets.
• Custom Memory Layout: Fixed-width binary record headers with byte-aligned vector offsets stored directly in memory-mapped files.
• SIMD-Accelerated Vector Operations: Matrix-vector operations executing over contiguous, cache-aligned memory blocks using NumPy C-extensions.
• Concurrency Model: Lock-free atomic ring-buffers for non-blocking stream ingestion and asynchronous search workers.

Python

import os
import sys
import time
import mmap
import struct
import numpy as np
from typing import List, Tuple, Dict, Any
from multiprocessing import Process, Event, cpu_count
from multiprocessing.shared_memory import SharedMemory
import asyncio
from fastapi import FastAPI, HTTPException
import uvicorn

# -----------------------------------------------------------------------------
# 1. Binary Layout & Memory-Mapped Disk Persistence Engine
# Structure per Record: [8-byte Uint64 ID][4-byte Uint32 Dim][Vector Bytes (Dim * 4)]
# -----------------------------------------------------------------------------
class BinaryVectorStorage:
    def __init__(self, filepath: str, vector_dim: int, max_records: int = 100_000):
        self.filepath = filepath
        self.vector_dim = vector_dim
        self.record_size = 8 + 4 + (vector_dim * 4)  # ID (Q) + Dim (I) + Float32 Vector
        self.file_size = self.record_size * max_records
        
        # Initialize sparse file on disk
        if not os.path.exists(self.filepath):
            with open(self.filepath, "wb") as f:
                f.seek(self.file_size - 1)
                f.write(b"\x00")

        self.file_obj = open(self.filepath, "r+b")
        self.mmapped_db = mmap.mmap(self.file_obj.fileno(), self.file_size)
        self.record_count = 0

    def append_vector(self, vector_id: int, vector: np.ndarray):
        if vector.dtype != np.float32:
            vector = vector.astype(np.float32)
        
        offset = self.record_count * self.record_size
        header = struct.pack("<QI", vector_id, self.vector_dim)
        vector_bytes = vector.tobytes()
        
        self.mmapped_db[offset : offset + 12] = header
        self.mmapped_db[offset + 12 : offset + self.record_size] = vector_bytes
        self.record_count += 1

    def get_matrix_view(self) -> Tuple[np.ndarray, np.ndarray]:
        """Exposes memory-mapped region directly as a zero-copy NumPy array."""
        if self.record_count == 0:
            return np.empty((0,), dtype=np.uint64), np.empty((0, self.vector_dim), dtype=np.float32)

        # Slice active byte range
        active_bytes = self.record_size * self.record_count
        raw_data = np.frombuffer(self.mmapped_db[:active_bytes], dtype=np.uint8)
        
        # Reshape using structured array striding
        structured_arr = raw_data.reshape((self.record_count, self.record_size))
        
        ids = np.zeros(self.record_count, dtype=np.uint64)
        vectors = np.zeros((self.record_count, self.vector_dim), dtype=np.float32)

        for i in range(self.record_count):
            row = structured_arr[i]
            ids[i] = struct.unpack("<Q", row[:8].tobytes())[0]
            vectors[i] = np.frombuffer(row[12:].tobytes(), dtype=np.float32)

        return ids, vectors

    def close(self):
        self.mmapped_db.flush()
        self.mmapped_db.close()
        self.file_obj.close()

# -----------------------------------------------------------------------------
# 2. Shared Memory Ring Buffer for Zero-Copy Multi-Process Ingestion
# -----------------------------------------------------------------------------
class SharedMemoryRingBuffer:
    def __init__(self, name: str, size: int, create: bool = True):
        self.name = name
        self.size = size
        if create:
            self.shm = SharedMemory(name=self.name, create=True, size=self.size)
        else:
            self.shm = SharedMemory(name=self.name, create=False)

    def write_at(self, offset: int, data: bytes):
        self.shm.buf[offset : offset + len(data)] = data

    def read_from(self, offset: int, length: int) -> bytes:
        return bytes(self.shm.buf[offset : offset + length])

    def cleanup(self):
        self.shm.close()
        try:
            self.shm.unlink()
        except FileNotFoundError:
            pass

# -----------------------------------------------------------------------------
# 3. Distributed Compute Engine & Parallel SIMD Search
# -----------------------------------------------------------------------------
class ParallelVectorEngine:
    def __init__(self, vector_dim: int, db_path: str = "vector_store.bin"):
        self.vector_dim = vector_dim
        self.db = BinaryVectorStorage(db_path, vector_dim)
        self.shm_name = "shm_vector_buffer"
        self.shm_size = 10 * 1024 * 1024  # 10MB allocated buffer
        self.ring_buffer = SharedMemoryRingBuffer(self.shm_name, self.shm_size, create=True)

    def insert(self, vector_id: int, vector: np.ndarray):
        # Normalize vector for SIMD dot product cosine similarity
        norm = np.linalg.norm(vector)
        if norm > 0:
            vector = vector / norm
        self.db.append_vector(vector_id, vector)

    def search_simd(self, query_vec: np.ndarray, top_k: int = 5) -> List[Dict[str, Any]]:
        ids, vectors = self.db.get_matrix_view()
        if len(vectors) == 0:
            return []

        norm = np.linalg.norm(query_vec)
        if norm > 0:
            query_vec = query_vec / norm

        # BLAS Level-2 Matrix-Vector Multiplication (SIMD vectorized)
        scores = np.dot(vectors, query_vec)
        
        # Partition top K elements without full sort O(N + K log K)
        if len(scores) > top_k:
            top_k_idx = np.argpartition(scores, -top_k)[-top_k:]
            top_k_idx = top_k_idx[np.argsort(-scores[top_k_idx])]
        else:
            top_k_idx = np.argsort(-scores)

        return [{"id": int(ids[idx]), "score": float(scores[idx])} for idx in top_k_idx]

    def shutdown(self):
        self.db.close()
        self.ring_buffer.cleanup()

# -----------------------------------------------------------------------------
# 4. Asynchronous High-Throughput Service Layer
# -----------------------------------------------------------------------------
app = FastAPI(title="IIT/IISc Tier Zero-Copy Vector Storage Engine")
engine = ParallelVectorEngine(vector_dim=128)

@app.on_event("startup")
async def startup_event():
    # Warm up index with sample cache-aligned vectors
    np.random.seed(42)
    for i in range(1000):
        vec = np.random.randn(128).astype(np.float32)
        engine.insert(vector_id=1000 + i, vector=vec)

@app.on_event("shutdown")
async def shutdown_event():
    engine.shutdown()

@app.post("/search")
async def search_vector(query: List[float], top_k: int = 5):
    if len(query) != 128:
        raise HTTPException(status_code=400, detail="Query vector dimension mismatch. Expected 128.")
    
    start_time = time.perf_counter_ns()
    query_np = np.array(query, dtype=np.float32)
    results = engine.search_simd(query_np, top_k=top_k)
    latency_us = (time.perf_counter_ns() - start_time) / 1000.0

    return {
        "latency_microseconds": latency_us,
        "results": results
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)