Transposing a CSV file (with '|' instead of ',')

!What is this?

People these days love to throw GPUs at every problem they can.
Sometimes it works well, other times it's a waste of power. 
This program was written to show that you don't need a big high end GPU to parse a CSV file quickly, 
in reponse to https://github.com/antonmks/nvParse

Can probably be optimized more. Gotta try running it on a SSD sometime.
Either way, much more competitive than the strawman CPU implementations on the above project page.

LICENSE

Copyright (c) 2015 Nicolas Guillemot

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

main.cpp

#define NOMINMAX
#include <windows.h>

#define _SCL_SECURE_NO_WARNINGS 1
#include <cassert>
#include <vector>
#include <algorithm>
#include <memory>

// Use this to print back the table.
// Can be used to diff with the input to check for bugs.
#define ECHO_INPUT 0

#if ECHO_INPUT
    // for _setmode...
    #include <io.h>
    #include <fcntl.h>
#endif

#define INPUT_FILENAME "lineitem.tbl"
// #define INPUT_FILENAME "lineitem_small.tbl"

// eyeballed from the file...
static const int kNumFields = 16;
static const int kMaxFieldLengths[kNumFields] = { 10,10,5,1,2,10,4,4,1,1,10,10,10,17,7,50 };
static const int kMaxLineLength = 256;

struct ParseCSVArgs
{
    LONGLONG start;
    LONGLONG size;
    const char* data;
    std::vector<std::unique_ptr<char[]>>* out_columns;
    size_t* out_numRows;
};

DWORD WINAPI ParseCSV(LPVOID threadParam)
{
    auto args = (const ParseCSVArgs*)(threadParam);

    // Count number of lines
    size_t lineCount = 0;
    for (LONGLONG i = args->start; i < args->start + args->size; i++)
    {
        if (args->data[i] == '\n')
        {
            lineCount++;
        }
    }

    *args->out_numRows = lineCount;
    for (size_t col = 0; col < args->out_columns->size(); col++)
    {
        (*args->out_columns)[col].reset(new char[kMaxFieldLengths[col] * lineCount]);
    }

    char lineBuffer[kMaxLineLength];
    int lineBufferIndex = 0;
    int columnIndex = 0;
    std::vector<size_t> columnSizes(kNumFields);
    LONGLONG end = args->start + args->size;
    for (LONGLONG i = args->start; i < end; i++)
    {
        char curr = args->data[i];

        if (curr == '\n')
        {
            lineBufferIndex = 0;
            columnIndex = 0;
            continue;
        }

        if (curr == '|')
        {
            assert(lineBufferIndex <= kMaxFieldLengths[columnIndex]);

            size_t oldSize = columnSizes[columnIndex];
            columnSizes[columnIndex] += kMaxFieldLengths[columnIndex];
            assert(columnSizes[columnIndex] <= size_t(kMaxFieldLengths[columnIndex] * lineCount));
            
            char* column = (*args->out_columns)[columnIndex].get();
            std::copy(lineBuffer, lineBuffer + lineBufferIndex, column + oldSize);
            std::fill_n(column + oldSize + lineBufferIndex, kMaxFieldLengths[columnIndex] - lineBufferIndex, 0);

            lineBufferIndex = 0;
            columnIndex++;
            continue;
        }

        assert(lineBufferIndex < kMaxLineLength);
        lineBuffer[lineBufferIndex] = curr;
        lineBufferIndex++;
    }
    
    return 0;
}

int main()
{
    BOOL ok;

    LARGE_INTEGER startTime, endTime;
    LARGE_INTEGER frequency;
    ok = QueryPerformanceFrequency(&frequency);
    assert(ok);

    ok = QueryPerformanceCounter(&startTime);
    assert(ok);

    SYSTEM_INFO sysInfo;
    GetSystemInfo(&sysInfo);
    int numProcessors = (int)sysInfo.dwNumberOfProcessors;
    assert(numProcessors >= 1);
    numProcessors = 8;

    HANDLE file = CreateFileA(INPUT_FILENAME, GENERIC_READ, FILE_SHARE_READ, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL | FILE_FLAG_SEQUENTIAL_SCAN, NULL);
    assert(file != INVALID_HANDLE_VALUE);

    LARGE_INTEGER fileSize;
    ok = GetFileSizeEx(file, &fileSize);
    assert(ok);

    HANDLE fileMapping = CreateFileMapping(file, NULL, PAGE_READONLY, 0, 0, NULL);
    assert(fileMapping != INVALID_HANDLE_VALUE);

    LPVOID fileMapView = MapViewOfFile(fileMapping, FILE_MAP_READ, 0, 0, 0);
    auto fileMapViewChar = (const char*)fileMapView;
    assert(fileMapView != NULL);

    // gotta make sure no two cores are sharing the same row in the file...
    LONGLONG flooredBytesPerCore = fileSize.QuadPart / numProcessors;
    std::vector<LONGLONG> startsPerCore(numProcessors);
    std::vector<LONGLONG> sizesPerCore(numProcessors);
    {
        startsPerCore[0] = 0;
        sizesPerCore[0] = fileSize.QuadPart - flooredBytesPerCore * (numProcessors - 1);
        while (fileMapViewChar[startsPerCore[0] + sizesPerCore[0] - 1] != '\n')
        {
            sizesPerCore[0]++;
        }

        LONGLONG offset = startsPerCore[0] + sizesPerCore[0];
        for (size_t i = 1; i < sizesPerCore.size(); i++)
        {
            startsPerCore[i] = offset;
            sizesPerCore[i] = flooredBytesPerCore;
            while (startsPerCore[i] + sizesPerCore[i] <= fileSize.QuadPart
               &&  fileMapViewChar[startsPerCore[i] + sizesPerCore[i] - 1] != '\n')
            {
                sizesPerCore[i]++;
            }
            startsPerCore[i] = std::min(startsPerCore[i], fileSize.QuadPart);
            sizesPerCore[i] = std::min(startsPerCore[i] + sizesPerCore[i], fileSize.QuadPart) - startsPerCore[i];
            offset = startsPerCore[i] + sizesPerCore[i];
        }
    }

    std::vector<std::vector<std::unique_ptr<char[]>>> out_columns_list(numProcessors);
    for (size_t i = 0; i < out_columns_list.size(); i++)
    {
        out_columns_list[i].resize(kNumFields);
    }

    std::vector<size_t> out_numRows(numProcessors);
    std::vector<ParseCSVArgs> findNewlinesArgs(numProcessors);
    for (int i = 0; i < numProcessors; i++)
    {
        findNewlinesArgs[i].start = startsPerCore[i];
        findNewlinesArgs[i].size = sizesPerCore[i];
        findNewlinesArgs[i].data = (const char*) fileMapView;
        findNewlinesArgs[i].out_columns = &out_columns_list[i];
        findNewlinesArgs[i].out_numRows = &out_numRows[i];
    }

    std::vector<HANDLE> threads;
    for (int i = numProcessors; i > 1; i--)
    {
        HANDLE thread = CreateThread(NULL, 0, ParseCSV, &findNewlinesArgs[i - 1], 0, NULL);
        assert(thread != INVALID_HANDLE_VALUE);
        threads.push_back(thread);
    }
    ParseCSV(&findNewlinesArgs[0]);

    for (HANDLE thread : threads)
    {
        DWORD result = WaitForSingleObject(thread, INFINITE);
        assert(result == WAIT_OBJECT_0);
    }

    size_t totalNumRows = 0;
    for (size_t numRows : out_numRows)
    {
        totalNumRows += numRows;
    }

    std::vector<std::unique_ptr<char[]>> joinedColumns(kNumFields);
    for (size_t col = 0; col < joinedColumns.size(); col++)
    {
        joinedColumns[col].reset(new char[totalNumRows * kMaxFieldLengths[col]]);
    }

    for (int c = 0; c < kNumFields; c++)
    {
        size_t row = 0;
        for (int i = 0; i < numProcessors; i++)
        {
            std::copy(out_columns_list[i][c].get(),
                      out_columns_list[i][c].get() + out_numRows[i] * kMaxFieldLengths[c],
                      joinedColumns[c].get() + row * kMaxFieldLengths[c]);
            row += out_numRows[i];
        }
    }

#if ECHO_INPUT
    _setmode(_fileno(stdout), _O_BINARY);
    for (size_t row = 0; row < totalNumRows; row++)
    {
        for (size_t col = 0; col < kNumFields; col++)
        {
            const char* entry = joinedColumns[col].get() + row * kMaxFieldLengths[col];
            size_t entryLen = strnlen(entry, kMaxFieldLengths[col]);
            fwrite(entry, entryLen, 1, stdout);
            fprintf(stdout, "|");
        }
        printf("\n"); // to match the line endings in the input data
        fflush(stdout);
    }
#endif

    ok = QueryPerformanceCounter(&endTime);
    assert(ok);

    LARGE_INTEGER elapsedMicroseconds;
    elapsedMicroseconds.QuadPart = endTime.QuadPart - startTime.QuadPart;
    elapsedMicroseconds.QuadPart *= 1000;
    elapsedMicroseconds.QuadPart /= frequency.QuadPart;

#if !ECHO_INPUT
    printf("Time: %lldms\n", elapsedMicroseconds.QuadPart);
#endif
}

Output of 10 runs

These runs are using the "lineitem.tbl" file. It is a 750MB text file of 6001215 lines.

Time: 1251ms
Time: 1251ms
Time: 1169ms
Time: 1181ms
Time: 1217ms
Time: 1148ms
Time: 1165ms
Time: 1198ms
Time: 1143ms
Time: 1161ms

Specs

Windows says

Processor: Intel(R) Core(TM) i7 CPU 975 @ 3.33GHz
Installed Memory (RAM): 6.00 GB
System type: 64-bit Operating System, x64-based processor