PHP Generators -- The 4 GB CSV That Crashed Production

<?php

/**
 * A generator factory — calling this returns a Generator object instantly.
 * No code inside the function body runs yet.
 */
function countUpTo(int $limit): Generator
{
    echo "[generator] Starting execution\n"; // runs on first next()/current()

    for ($number = 1; $number <= $limit; $number++) {
        echo "[generator] About to yield $number\n";

        yield $number; // <-- execution PAUSES here; $number is the yielded value

        echo "[generator] Resumed after yield $number\n"; // runs on the NEXT next() call
    }

    echo "[generator] Generator exhausted\n"; // runs after the loop ends
}

// --- Calling the function does NOT execute any code yet ---
$generator = countUpTo(3);
echo "Generator object created. No output above this line yet.\n\n";

// --- Manually driving the generator ---
$generator->current(); // initialises + runs to first yield
echo "First value: " . $generator->current() . "\n\n";

$generator->next();    // resumes from yield 1, runs to yield 2
echo "Second value: " . $generator->current() . "\n\n";

$generator->next();    // resumes from yield 2, runs to yield 3
echo "Third value: " . $generator->current() . "\n\n";

$generator->next();    // resumes from yield 3, loop ends, generator exhausts
echo "Valid? " . ($generator->valid() ? 'yes' : 'no') . "\n";

<?php

/**
 * Yields one parsed CSV row at a time — never loads the whole file.
 * Type: Generator<int, array<string, string>, void, void>
 */
function readCsvLazily(string $filePath, bool $hasHeaderRow = true): Generator
{
    $fileHandle = fopen($filePath, 'r');
    if ($fileHandle === false) {
        throw new RuntimeException("Cannot open file: $filePath");
    }

    try {
        $headers = [];

        if ($hasHeaderRow) {
            // Read the first line as column headers
            $headers = fgetcsv($fileHandle);
            if ($headers === false) {
                return; // empty file — generator yields nothing
            }
        }

        $rowIndex = 0;
        while (($rawFields = fgetcsv($fileHandle)) !== false) {
            // Combine headers with row values so callers get named keys
            $namedRow = $hasHeaderRow
                ? array_combine($headers, $rawFields)
                : $rawFields;

            yield $rowIndex => $namedRow; // key => value yield

            $rowIndex++;
        }
    } finally {
        // The finally block ALWAYS runs — even if the caller breaks early
        fclose($fileHandle);
        echo "[cleanup] File handle closed.\n";
    }
}

/**
 * Filtering generator — wraps another generator, yields only matching rows.
 * This is the 'pipe' pattern: generator -> generator -> consumer.
 */
function filterRows(Generator $source, string $column, string $matchValue): Generator
{
    foreach ($source as $rowIndex => $row) {
        if (isset($row[$column]) && $row[$column] === $matchValue) {
            yield $rowIndex => $row; // pass through the original key
        }
    }
}

// --- Demo: create a small sample CSV in memory to make this runnable ---
$tempFile = tempnam(sys_get_temp_dir(), 'csv_demo_');
file_put_contents($tempFile,
    "id,name,department,salary\n" .
    "1,Alice,Engineering,95000\n" .
    "2,Bob,Marketing,72000\n" .
    "3,Carol,Engineering,102000\n" .
    "4,Dave,Marketing,68000\n" .
    "5,Eve,Engineering,88000\n"
);

$memoryBefore = memory_get_usage();

// Build a lazy pipeline — nothing reads the file yet
$allRows        = readCsvLazily($tempFile);
$engineeringRows = filterRows($allRows, 'department', 'Engineering');

echo "Memory after pipeline setup (nothing read yet): "
    . number_format(memory_get_usage() - $memoryBefore) . " bytes\n\n";

// NOW we pull values — file is read one line at a time
echo "Engineering department employees:\n";
foreach ($engineeringRows as $index => $employee) {
    echo "  Row $index: {$employee['name']} — \${$employee['salary']}\n";
}

unlink($tempFile); // remove the temp file

<?php

/**
 * A running-total accumulator implemented as a coroutine.
 *
 * - Yields a status string each time a number is received.
 * - Returns the final total when closed via throw().
 *
 * Caller sends numbers; generator responds with a receipt.
 */
function runningTotalAccumulator(): Generator
{
    $total     = 0;
    $itemCount = 0;

    // The generator starts here when current()/send(null) is called
    echo "[accumulator] Ready to receive numbers.\n";

    while (true) {
        // yield does two jobs at once:
        //   1. Sends a status string OUT to the caller
        //   2. Receives the next number sent IN via send()
        $incomingNumber = yield "Current total: $total (items: $itemCount)";

        if ($incomingNumber === null) {
            // send(null) is what foreach internally does on the first next() —
            // we treat it as a no-op to handle gracefully
            continue;
        }

        $total     += $incomingNumber;
        $itemCount++;
        echo "[accumulator] Added $incomingNumber → running total = $total\n";
    }
}

$accumulator = runningTotalAccumulator();

// Initialise the generator — runs to first yield, prints the 'Ready' line
$initialStatus = $accumulator->current();
echo "Status: $initialStatus\n\n";

// Send values in — each send() resumes the generator and returns the NEW yielded value
$numbers = [150, 75, 200, 25];
foreach ($numbers as $number) {
    $status = $accumulator->send($number); // sends number IN, gets status string OUT
    echo "Caller received: $status\n";
}

echo "\n";

// --- Demonstrating throw() for clean cancellation ---
echo "Cancelling accumulator early via throw()...\n";
try {
    $accumulator->throw(new RuntimeException('Cancelled by caller'));
} catch (RuntimeException $exception) {
    // throw() re-throws the exception in the caller's scope if the
    // generator doesn't catch it internally
    echo "Caught in caller: " . $exception->getMessage() . "\n";
}

// --- Demonstrating getReturn() ---

/**
 * A generator that streams progress updates and returns a summary array.
 */
function processOrderBatch(array $orderIds): Generator
{
    $processedCount = 0;
    $failedIds      = [];

    foreach ($orderIds as $orderId) {
        // Simulate work — in production this might be a DB write
        $success = ($orderId % 7 !== 0); // pretend multiples of 7 fail

        if ($success) {
            $processedCount++;
            yield "Processed order #$orderId"; // stream progress to caller
        } else {
            $failedIds[] = $orderId;
            yield "FAILED order #$orderId";
        }
    }

    // return is only accessible via getReturn() after generator exhausts
    return [
        'processed' => $processedCount,
        'failed'    => $failedIds,
        'total'     => count($orderIds),
    ];
}

$orderIds     = [101, 102, 103, 105, 112, 114, 119];
$orderBatch   = processOrderBatch($orderIds);

echo "\n--- Order Batch Processing ---\n";
foreach ($orderBatch as $progressMessage) {
    echo $progressMessage . "\n";
}

// Generator is now exhausted — getReturn() is safe to call
$summary = $orderBatch->getReturn();
echo "\nSummary: {$summary['processed']} processed, "
    . count($summary['failed']) . " failed "
    . "(IDs: " . implode(', ', $summary['failed']) . ")\n";

<?php

/**
 * Recursively walks a nested directory tree, yielding file paths.
 * Uses yield from for recursive delegation — no re-yielding loop needed.
 */
function walkDirectoryTree(string $directoryPath): Generator
{
    $entries = new DirectoryIterator($directoryPath);

    foreach ($entries as $entry) {
        if ($entry->isDot()) {
            continue; // skip '.' and '..'
        }

        if ($entry->isDir()) {
            // Delegate to a recursive call — send()/throw() are proxied through.
            // The return value of the inner generator is captured in $subDirCount.
            $subDirCount = yield from walkDirectoryTree($entry->getPathname());

            echo "[delegation] Subdirectory '{$entry->getFilename()}' contained $subDirCount file(s)\n";
        } else {
            yield $entry->getPathname(); // yield the file path to the top-level caller
        }
    }

    // This return value flows back to the parent generator via yield from
    return iterator_count(walkDirectoryTree($directoryPath)); // count files in this dir only
}

/**
 * Simpler example showing yield from with arrays and return capture.
 * This makes the delegation mechanics easy to observe without filesystem access.
 */
function innerSequence(string $label): Generator
{
    echo "[inner:$label] starting\n";
    yield "$label-A";
    yield "$label-B";
    echo "[inner:$label] done\n";
    return "metadata-from-$label"; // captured by yield from in the outer generator
}

function outerSequence(): Generator
{
    yield "outer-start";

    // yield from array — delegation works with plain arrays too
    yield from ["array-item-1", "array-item-2"];

    // yield from generator — return value is captured
    $returnedMetadata = yield from innerSequence('alpha');
    echo "[outer] inner returned: $returnedMetadata\n";

    $returnedMetadata = yield from innerSequence('beta');
    echo "[outer] inner returned: $returnedMetadata\n";

    yield "outer-end";

    return "outer-summary";
}

echo "=== Delegated sequence output ===\n";
foreach (outerSequence() as $index => $value) {
    echo "  [$index] $value\n";
}

// --- Real-world: merging multiple paginated API result generators ---

function simulateApiPage(string $resource, int $page): Generator
{
    // Simulates fetching a page of records from a REST API
    $itemsPerPage = 3;
    $start        = ($page - 1) * $itemsPerPage + 1;
    $end          = $start + $itemsPerPage - 1;

    for ($id = $start; $id <= $end; $id++) {
        yield ['resource' => $resource, 'id' => $id, 'page' => $page];
    }

    return $itemsPerPage; // how many items this page had
}

function fetchAllPages(string $resource, int $totalPages): Generator
{
    $totalFetched = 0;
    for ($page = 1; $page <= $totalPages; $page++) {
        $pageCount     = yield from simulateApiPage($resource, $page);
        $totalFetched += $pageCount;
        echo "[paginator] Fetched page $page ($pageCount items, $totalFetched total so far)\n";
    }
    return $totalFetched;
}

echo "\n=== Paginated API fetch ===\n";
$apiGenerator = fetchAllPages('products', 2);
foreach ($apiGenerator as $record) {
    echo "  Product ID {$record['id']} (page {$record['page']})\n";
}
echo "Total fetched: " . $apiGenerator->getReturn() . "\n";

<?php

/**
 * Simulates an external API that returns paginated results.
 * In production, replace with Guzzle or curl calls.
 */
class ExternalApi
{
    public function fetchPage(string $endpoint, int $page): array
    {
        // Simulate network latency
        usleep(20000); // 20ms

        if ($page > 5) {
            return ['items' => [], 'next_page' => null];
        }

        $items = [];
        for ($i = 1; $i <= 20; $i++) {
            $items[] = [
                'id'   => ($page - 1) * 20 + $i,
                'name' => "Item " . (($page - 1) * 20 + $i),
            ];
        }

        return [
            'items'     => $items,
            'next_page' => $page + 1,
            'rate_limited' => false,
        ];
    }
}

/**
 * Generator that fetches all pages and yields individual records.
 * Accepts send() to inject control signals (e.g., 'throttle').
 */
function fetchAllRecords(string $endpoint): Generator
{
    $api      = new ExternalApi();
    $page     = 1;
    $total    = 0;

    while (true) {
        $response = $api->fetchPage($endpoint, $page);

        if (empty($response['items'])) {
            break; // no more data
        }

        if ($response['rate_limited']) {
            // Caller can send a signal to retry after delay
            $signal = yield []; // yield empty to get control from caller
            if ($signal === 'throttle') {
                echo "[generator] Throttling...\n";
                sleep(1); // backoff
                continue; // retry same page
            }
        }

        foreach ($response['items'] as $item) {
            yield $item;
            $total++;
        }

        $page = $response['next_page'];
        if ($page === null) {
            break;
        }
    }

    return $total;
}

$generator = fetchAllRecords('/api/products');

echo "=== Paginated API Client ===\n";
foreach ($generator as $record) {
    echo "  Processed: {$record['id']} - {$record['name']}\n";
    // Simulate a scenario where we need to throttle after batch
    if ($record['id'] % 60 === 0) {
        // In real code, you'd call $generator->send('throttle') after checking API limits
    }
}
echo "Total records: " . $generator->getReturn() . "\n";
echo "Peak memory: " . memory_get_peak_usage(true) / 1024 . " KB\n";

// io.thecodeforge
function parseUserCsv(string $path): \Generator {
    $handle = fopen($path, 'r');
    $headers = fgetcsv($handle);
    while ($row = fgetcsv($handle)) {
        $record = array_combine($headers, $row);
        yield $record['user_id'] => [
            'email' => $record['email'],
            'role' => $record['role'],
            'created_at' => $record['created_at'],
        ];
    }
    fclose($handle);
}

foreach (parseUserCsv('/var/data/import.csv') as $id => $user) {
    echo "Processed user $id ({$user['email']})" . PHP_EOL;
}

// io.thecodeforge
function streamLargeQuery(PDO $pdo, string $sql): \Generator {
    $stmt = $pdo->prepare($sql);
    $stmt->execute();
    try {
        while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
            yield $row['id'] => $row;
        }
    } finally {
        $stmt->closeCursor();
        echo "Cursor closed early or normally." . PHP_EOL;
    }
}

$gen = streamLargeQuery($pdo, 'SELECT * FROM logs WHERE processed = 0 LIMIT 100000');
foreach ($gen as $id => $row) {
    if (time() % 2 === 0) {
        break; // Simulate early exit
    }
}
// Outputs: Cursor closed early or normally.