3

我正在编写代码以在旨在在 8 核 x86 系统上运行的 32 位 Linux 操作系统上使用 C/C++ 从二进制文件中读取无符号整数。该应用程序获取一个输入文件,其中一个接一个地包含小端格式的无符号整数。所以以字节为单位的输入文件大小是 4 的倍数。文件中可能有十亿个整数。读取和添加所有整数并以 64 位精度返回总和的最快方法是什么?

下面是我的实现。错误检查损坏数据不是这里的主要问题,在这种情况下输入文件被认为没有任何问题。

#include <iostream>
#include <fstream>
#include <pthread.h>
#include <string>
#include <string.h>


using namespace std;

string filepath;
unsigned int READBLOCKSIZE = 1024*1024;
unsigned long long nFileLength = 0;

unsigned long long accumulator = 0; // assuming 32 bit OS running on X86-64
unsigned int seekIndex[8] = {};
unsigned int threadBlockSize = 0; 
unsigned long long acc[8] = {};

pthread_t thread[8];
void* threadFunc(void* pThreadNum);

//time_t seconds1;
//time_t seconds2;

int main(int argc, char *argv[])
{
    if (argc < 2) 
    {
        cout << "Please enter a file path\n";
        return -1;
    }

    //seconds1 = time (NULL);
    //cout << "Start Time in seconds since January 1, 1970 -> " << seconds1 << "\n";

    string path(argv[1]);
    filepath = path;
    ifstream ifsReadFile(filepath.c_str(), ifstream::binary);  // Create FileStream for the file to be read
    if(0 == ifsReadFile.is_open()) 
    {
        cout << "Could not find/open input file\n";
        return -1;
    }

    ifsReadFile.seekg (0, ios::end);
    nFileLength = ifsReadFile.tellg();           // get file size
    ifsReadFile.seekg (0, ios::beg);



    if(nFileLength < 16*READBLOCKSIZE)
    {
        //cout << "Using One Thread\n"; //**
        char* readBuf = new char[READBLOCKSIZE];
        if(0 == readBuf) return -1;

        unsigned int startOffset = 0;   
        if(nFileLength >  READBLOCKSIZE)
        {
            while(startOffset + READBLOCKSIZE < nFileLength)
            {
                //ifsReadFile.flush();
                ifsReadFile.read(readBuf, READBLOCKSIZE);  // At this point ifsReadFile is open
                int* num = reinterpret_cast<int*>(readBuf);
                for(unsigned int i = 0 ; i < (READBLOCKSIZE/4) ; i++) 
                {
                    accumulator += *(num + i);  
                }
                startOffset += READBLOCKSIZE;
            }

        }

        if(nFileLength - (startOffset) > 0)
        {
            ifsReadFile.read(readBuf, nFileLength - (startOffset));  
            int* num = reinterpret_cast<int*>(readBuf);
            for(unsigned int i = 0 ; i < ((nFileLength - startOffset)/4) ; ++i) 
            {
                accumulator += *(num + i);  
            }
        }
        delete[] readBuf; readBuf = 0;
    }
    else
    {
        //cout << "Using 8 Threads\n"; //**
        unsigned int currthreadnum[8] = {0,1,2,3,4,5,6,7};
        if(nFileLength > 200000000) READBLOCKSIZE *= 16; // read larger blocks
        //cout << "Read Block Size -> " << READBLOCKSIZE << "\n";       

        if(nFileLength % 28)
        {
            threadBlockSize = (nFileLength / 28);
            threadBlockSize *= 4;
        }
        else
        {   
            threadBlockSize = (nFileLength / 7);
        }

        for(int i = 0; i < 8 ; ++i)
        {
            seekIndex[i] = i*threadBlockSize;
            //cout << seekIndex[i] << "\n";
        }
        pthread_create(&thread[0], NULL, threadFunc, (void*)(currthreadnum + 0));
        pthread_create(&thread[1], NULL, threadFunc, (void*)(currthreadnum + 1));
        pthread_create(&thread[2], NULL, threadFunc, (void*)(currthreadnum + 2));
        pthread_create(&thread[3], NULL, threadFunc, (void*)(currthreadnum + 3));
        pthread_create(&thread[4], NULL, threadFunc, (void*)(currthreadnum + 4));
        pthread_create(&thread[5], NULL, threadFunc, (void*)(currthreadnum + 5));
        pthread_create(&thread[6], NULL, threadFunc, (void*)(currthreadnum + 6));
        pthread_create(&thread[7], NULL, threadFunc, (void*)(currthreadnum + 7));

        pthread_join(thread[0], NULL);
        pthread_join(thread[1], NULL);
        pthread_join(thread[2], NULL);
        pthread_join(thread[3], NULL);
        pthread_join(thread[4], NULL);
        pthread_join(thread[5], NULL);
        pthread_join(thread[6], NULL);
        pthread_join(thread[7], NULL);

        for(int i = 0; i < 8; ++i)
        {
            accumulator += acc[i];
        }
    }

    //seconds2 = time (NULL);
    //cout << "End Time in seconds since January 1, 1970 -> " << seconds2 << "\n";
    //cout << "Total time to add " << nFileLength/4 << " integers -> " << seconds2 - seconds1 << " seconds\n";

    cout << accumulator << "\n";      
    return 0;
}

void* threadFunc(void* pThreadNum)
{
    unsigned int threadNum = *reinterpret_cast<int*>(pThreadNum);
    char* localReadBuf = new char[READBLOCKSIZE];
    unsigned int startOffset = seekIndex[threadNum];
    ifstream ifs(filepath.c_str(), ifstream::binary);  // Create FileStream for the file to be read
    if(0 == ifs.is_open()) 
    {
        cout << "Could not find/open input file\n";
        return 0;
    }   
    ifs.seekg (startOffset, ios::beg); // Seek to the correct offset for this thread
    acc[threadNum] = 0;
    unsigned int endOffset = startOffset + threadBlockSize;
    if(endOffset > nFileLength) endOffset = nFileLength; // for last thread
    //cout << threadNum << "-" << startOffset << "-" << endOffset << "\n"; 
    if((endOffset - startOffset) >  READBLOCKSIZE)
    {
        while(startOffset + READBLOCKSIZE < endOffset)
        {
            ifs.read(localReadBuf, READBLOCKSIZE);  // At this point ifs is open
            int* num = reinterpret_cast<int*>(localReadBuf);
            for(unsigned int i = 0 ; i < (READBLOCKSIZE/4) ; i++) 
            {
                acc[threadNum] += *(num + i);   
            }
            startOffset += READBLOCKSIZE;
        }   
    }

    if(endOffset - startOffset > 0)
    {
        ifs.read(localReadBuf, endOffset - startOffset);
        int* num = reinterpret_cast<int*>(localReadBuf);
        for(unsigned int i = 0 ; i < ((endOffset - startOffset)/4) ; ++i) 
        {
            acc[threadNum] += *(num + i);   
        }
    }

    //cout << "Thread " << threadNum + 1 << " subsum = " << acc[threadNum] << "\n"; //**
    delete[] localReadBuf; localReadBuf = 0;
    return 0;
}

我编写了一个小 C# 程序来生成输入二进制文件以进行测试。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;

namespace BinaryNumWriter
{
    class Program
    {
        static UInt64 total = 0;
        static void Main(string[] args)
        {
            BinaryWriter bw = new BinaryWriter(File.Open("test.txt", FileMode.Create));
            Random rn = new Random();
            for (UInt32 i = 1; i <= 500000000; ++i)
            {
                UInt32 num = (UInt32)rn.Next(0, 0xffff);
                bw.Write(num);
                total += num;
            }
            bw.Flush();
            bw.Close();
        }
    }
}

在具有 2 GB RAM 和 Ubuntu 9.10 32 位的 Core i5 机器 @ 3.33 Ghz(它的四核,但我目前得到的)上运行该程序具有以下性能数字

100 个整数 ~ 0 秒(否则我真的不得不吸) 100000 个整数 < 0 秒 100000000 个整数 ~ 7 秒 500000000 个整数 ~ 29 秒(1.86 GB 输入文件)

我不确定硬盘是 5400RPM 还是 7200RPM。我尝试了不同的缓冲区大小进行读取,发现一次读取 16 MB 的大输入文件是个不错的选择。

有没有更好的方法可以更快地从文件中读取以提高整体性能?有没有更聪明的方法可以更快地添加大型整数数组并重复折叠?我编写代码的方式是否有任何重大障碍/我是否在做一些明显错误的事情,这会花费大量时间?

我可以做些什么来加快读取和添加数据的过程?

谢谢。

钦梅

4

2 回答 2

3

如果您想快速读取(或写入)大量数据,并且不想对这些数据进行太多处理,则需要避免缓冲区之间数据的额外副本。这意味着您要避免 fstream 或 FILE 抽象(因为它们引入了需要复制的额外缓冲区),并避免在内核和用户缓冲区之间复制内容的读/写类型调用。

相反,在 linux 上,您想使用 mmap(2)。在 64 位操作系统上,只需将整个文件 mmap 到内存中,用于madvise(MADV_SEQUENTIAL)告诉内核您将主要按顺序访问它,然后就可以了。对于 32 位操作系统,您需要以块的形式进行映射,每次都取消映射前一个块。与您当前的结构非常相似,每个线程一次映射一个固定大小的块应该可以正常工作。

于 2012-06-08T19:22:54.673 回答
3

以您的方式从多个线程访问机械硬盘需要一些头部运动(读取速度慢)。你几乎肯定会受到 IO 限制(1.86GB 文件为 65MBps)。尝试通过以下方式改变策略:

  • 启动 8 个线程 - 我们称它们为消费者
  • 8 个线程将等待数据可用
  • 在主线程中开始读取文件的块(比如 256KB),因此成为消费者的提供者
  • 主线程到达 EOF 并向工作人员发出没有更多可用数据的信号
  • 主线程等待 8 个工人加入。

你需要相当多的同步才能让它完美地工作,我认为它会通过顺序文件访问完全最大化你的 HDD / 文件系统 IO 能力。YMMV 处理小文件,可以以闪电般的速度从缓存中缓存和提供服务。

您可以尝试的另一件事是仅启动 7 个线程,为主线程和系统的其余部分保留一个空闲 CPU。

..或获得SSD :)

编辑:

为简单起见,看看您可以多快地读取文件(丢弃缓冲区)而无需处理,单线程。加上 epsilon 是您完成此操作的速度的理论限制。

于 2012-06-08T18:25:48.247 回答