マルチスレッディングの最良の方法は、ファイル数の行を処理する

Question

私はいくつかのファイルを分割しています。ファイルのすべての行を（逐次的かつ独立して）処理したいと思います。

私はramのバッファにファイルの大きな塊を読み込むコードを書いていました。そして、マルチスレッドはバッファから行を読み込み、それらを処理するために競合します。擬似コードは以下の通りである：

do{ 
    do{  

    fread(buffer,500MB,1,file); 
    // creating threads 
    // let the threads compete to read from buffer and PROCESS independently 
    // end of threads 

    while(EOF not reached) 
    file = nextfile; 
while(there is another file to read) 

またはこのいずれか

void mt_ReadAndProcess(){ 
    lock(); 
    fread(buffer,50MB,1,file); 
    if(EOF reached) 
    file = nextfile; 
    unlock(); 
    process(); 
} 
main(){ 
    // create multi threads 
    // call mt_ReadAndProcess() with multi threads 
} 

処理（タイムリー）高価なプロセスです。

これを行うにはもっと良い方法はありますか？ファイルをより速く読み取るか、マルチスレッドで処理する方が良いでしょうか？

ありがとうすべて、

Ameer。

Answer 1

なぜ「バッファからの読み取りに競合する」スレッドが必要なのでしょうか？データは、の読書を行っているスレッドによって、のように簡単に分割することができます。バッファからデータを取得するという競合は、CPUとウォールクロックの両方の時間を浪費する可能性がありますが、何も得られません。

行単位で処理しているので、ファイルから行を読み込んで、ポインタによってバッファをワーカースレッドに渡します。

このような何か、あなたはPOSIX準拠のシステム上で実行していると仮定：

#include <unistd.h> 
#include <pthread.h> 

#define MAX_LINE_LEN 1024 
#define NUM_THREADS 8 

// linePipe holds pointers to lines sent to 
// worker threads 
static int linePipe[ 2 ]; 

// bufferPipe holds pointers to buffers returned 
// from worker threads and used to read data 
static int bufferPipe[ 2 ]; 

// thread function that actually does the work 
void *threadFunc(void *arg) 
{ 
    const char *linePtr; 

    for (;;) 
    { 
     // get a pointer to a line from the pipe 
     read(linePipe[ 1 ], &linePtr, sizeof(linePtr)); 

     // end loop on NULL linePtr value 
     if (!linePtr) 
     { 
      break; 
     } 

     // process line 

     // return the buffer 
     write(bufferPipe[ 0 ], &linePtr, sizeof(linePtr)); 
    } 

    return(NULL); 
} 

int main(int argc, char **argv) 
{ 
    pipe(linePipe); 
    pipe(bufferPipe); 

    // create buffers and load them into the buffer pipe for reading 
    for (int ii = 0; ii < (2 * NUM_THREADS); ii++) 
    { 
     char *buffer = malloc(MAX_LINE_LEN); 
     write(bufferPipe[ 0 ], &buffer, sizeof(buffer)); 
    } 

    pthread_t tids[ NUM_THREADS ]; 
    for (int ii = 0; ii < NUM_THREADS; ii++) 
    { 
     pthread_create(&(tids[ ii ]), NULL, thread_func, NULL); 
    } 

    FILE *fp = ... 

    for (;;) 
    { 
     char *linePtr; 

     // get the pointer to a buffer from the buffer pipe 
     read(bufferPipe[ 1 ], &linePtr, sizeof(linePtr)); 

     // read a line from the current file into the buffer 
     char *result = fgets(linePtr, MAX_LINE_LEN, fp); 

     if (result) 
     { 
      // send the line to the worker threads 
      write(linePipe, &linePtr, sizeof(linePtr)); 
     } 
     else 
     { 
      // either end loop, or open another file 
      fclose(fp); 
      fp = fopen(...); 
     } 
    } 

    // clean up and exit 

    // send NULL to cause worker threads to stop 
    char *nullPtr = NULL; 
    for (int ii = 0; ii < NUM_THREADS; ii++) 
    { 
     write(linePipe[ 0 ], &nullPtr, sizeof(nullPtr)); 
    } 

    // wait for worker threads to stop 
    for (int ii = 0; ii < NUM_THREADS; ii++) 
    { 
     pthread_join(tids[ ii ], NULL); 
    } 

    return(0); 
}