Meilleure façon de Multithreading pour traiter les lignes du nombre de fichiers

Question

J'ai quelques fichiers séparés, je veux traiter chaque ligne des fichiers (séquentiellement et indépendamment), et je veux que ce soit rapide.

J'ai donc écrit un code pour lire un gros morceau d'un fichier dans un tampon sur RAM, puis les multi-threads vont rivaliser pour lire les lignes du tampon et les traiter. le pseudo-code est la suivante:

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) 

Ou celui-ci:

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 
} 

Le processus est un processus coûteux (en temps opportun).

Y a-t-il une meilleure façon de faire cela? une meilleure façon de lire le fichier plus rapidement ou de le traiter avec plusieurs threads?

Merci à tous,

Ameer.

Answer 1

Pourquoi souhaitez-vous avoir des threads "concurrencer à lire à partir du tampon"? Les données peuvent être facilement partitionnées car elles sont lues par le thread faisant la lecture. Contester pour obtenir des données à partir d'un tampon ne gagne rien tout en gaspillant probablement à la fois le temps processeur et l'horloge murale.

Puisque vous traitez ligne par ligne, lisez simplement les lignes du fichier et transmettez les tampons par le pointeur aux threads de travail.

En supposant que vous utilisez sur un système compatible POSIX, quelque chose comme ceci:

#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); 
}