C程序: Parent read one file and child count the number of words.子程序有4个线程并且还使用了mapper和reducer

Question

我正在编写一个 C 程序，其中有一个 child 和一个 parent 。父级和子级使用共享内存共享数据。我正在做的是要求父进程将文件写入共享内存，子进程然后从共享内存中读取文件并输出一个列表，显示每个唯一单词的计数。

我要做的是在子程序中使用4个线程，并使用mapper和reducer来完成任务。

文本文件大约有 30000 行。如果我在文本文件中仅传递 20000 行，则我的程序运行正确，但现在正在运行整个文件。

如果有人可以看看我的程序并让我知道哪里出了问题，我将不胜感激。

以下是文本文件的链接:http://cis-linux1.temple.edu/~qzeng/cis5512-fall2016/papers/ANNA_KARENINA.txt

这是我尝试运行的代码:

#include<stdio.h>
#include<stdlib.h>
#include  <sys/types.h>
#include  <sys/ipc.h>
#include  <sys/shm.h>
#include  <time.h>
#include  <unistd.h> 
#include  <pthread.h>
#include  <string.h>
#include  <ctype.h>

#define NUM_THREADS     4

static key_t key = (key_t) 0;
static int size = 0;

struct thread_data
{
int thread_id;
char *msg;
char* wordary[10000][2];
int size;
};

struct thread_data thread_data_array[NUM_THREADS];

void *CountWords(void *threadarg)
{
int taskid, j, i=0, flag=0, index = 0, p, k,z, cnt, m;
char *msg_words, c, *word, buffer[8];
char* word_array[10000][2];
struct thread_data *my_data;
my_data = (struct thread_data *) threadarg;
taskid = my_data->thread_id;
msg_words = my_data->msg;
strcat(msg_words," ");
word = (char*) malloc(20);
word_array[0][0] = (char*) malloc(30);
word_array[0][1] = (char*) malloc(8);
FILE *out;

if(taskid==0)
out=fopen("out.txt","w"); 
//printf("%d\n", strlen(msg_words));
for(j=0; j < strlen(msg_words); j++)
{
    c = msg_words[j];
    c = tolower(c);

    if(c == '\n')
    {
        c = ' ';
    }

    if(!isspace(c))
    {
        word[i++] = c;
    }

    if(c == '\0')
    {
        break;
    }

    if(c == ' ')
    {
        flag = 0;

        for(k=0; k <= index; k++)
        {
        if(0 == strcmp(word_array[k][0],word))
        {
            flag = 1;
            cnt = atoi(word_array[k][1]);
            cnt++;
            sprintf(buffer, "%d", cnt);
            strcpy(word_array[k][1],buffer);
        }
    }

    if(flag == 0)
    {
        strcpy(word_array[index][0],word);
        strcpy(word_array[index][1],"1");
        index++;
        word_array[index][0]=(char*)malloc(30);
        word_array[index][1]=(char*)malloc(8);                  
    }

    for(p=0; p <= 20; p++)
    {
        word[p] = 0;
    }
    i = 0;

    //printf("%d",index);
    }
    //my_data->size = index;
}
printf("%d\n",index);
my_data->size = index;    
for(m = 0; m<index; m++)
{
//printf("%d",m);
    my_data->wordary[m][0] = (char*) malloc(30);
    my_data->wordary[m][1] = (char*) malloc(8);

    strcpy(my_data->wordary[m][0], word_array[m][0]);
    strcpy(my_data->wordary[m][1], word_array[m][1]);

    //printf("%s  %s\n", my_data->wordary[m][0], my_data->wordary[m][1]);
}



pthread_exit((void *)my_data);
}

void  main()
{
 int    ShmID, index = 0;
 char* ShmPTR;
 pid_t  pid;
 int    status;
 clock_t begin, end;
 double time_spent;
 begin = clock();
 FILE *txtfile, *out_file;
 txtfile = fopen("test.txt", "r");

 fseek(txtfile, 0, SEEK_END); // seek to end of file
 size = ftell(txtfile); // get current file pointer
 fseek(txtfile, 0, SEEK_SET);
 //printf("size : %d", size);

 key = ftok(__FILE__,'x');
 ShmID = shmget(key, size, IPC_CREAT | 0666);
 if (ShmID < 0) {
      printf("*** shmget error (server) ***\n");
      exit(1);
 }
 printf("Server has received a shared memory\n");

 ShmPTR = (char *) shmat(ShmID, NULL, 0);
 if (ShmPTR == (char *)(-1)) {
      printf("*** shmat error (server) ***\n");
      exit(1);
 }
 printf("Server has attached the shared memory...\n");

 while(!feof(txtfile))
 {
    ShmPTR[index] = fgetc(txtfile);
    index++;    
 }

 //ShmPTR[index] = '\0';

 printf("Server is about to fork a child process...\n");
 pid = fork();
 if (pid < 0) 
 {
      printf("*** fork error (server) ***\n");
      exit(1);
 }
 else if (pid == 0) 
 {
    printf("   Client process started\n");
    //printf("%s",shm);
    pthread_t threads[NUM_THREADS];
    pthread_attr_t attr;
    int rc, t, shmsz1, shmsz2, shmsz3;
    char* split_ShmPTR[4]; 

    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    //printf("1111");
    //printf("%d\n",size);
    shmsz1 = (int)(size/4);
    shmsz2 = shmsz1*2;
    shmsz3 = shmsz1*3;

 // printf("%d  %d  %d\n", shmsz1, shmsz2, shmsz3);

    //printf("%c\n",ShmPTR[87]);

    while(ShmPTR[shmsz1] != ' ')
    { 
        shmsz1++;
    }

    //printf("%d  \n", shmsz1);
    //printf("%c1\n",ShmPTR[shmsz1]);
    split_ShmPTR[0] = (char*)malloc(shmsz1 + 1000);
    strncpy(split_ShmPTR[0],ShmPTR,shmsz1);


    while(ShmPTR[shmsz2] != ' ')
    { 
        shmsz2++;
    }

    split_ShmPTR[1] = (char*)malloc(shmsz2-shmsz1 + 1000);
    strncpy(split_ShmPTR[1],ShmPTR + shmsz1,shmsz2-shmsz1);


    while(ShmPTR[shmsz3] != ' ')
    { 
        shmsz3++;
    }

    split_ShmPTR[2] = (char*)malloc(shmsz3-shmsz2 + 1000);
    strncpy(split_ShmPTR[2],ShmPTR + shmsz2,shmsz3-shmsz2);


    split_ShmPTR[3] = (char*)malloc(size-shmsz3 + 10);
    strncpy(split_ShmPTR[3],ShmPTR + shmsz3,size-shmsz3);
//printf("%s\n",split_ShmPTR[3]); 

struct thread_data *my_words;
char* word_array_final[30000][2];
word_array_final[0][0] = (char*)malloc(30);
word_array_final[0][1] = (char*)malloc(8);
int q, r, flag1 = 0,  count, idx = 0, z;
char buff[8];

    for(t = 0; t<NUM_THREADS; t++)
    {
        thread_data_array[t].thread_id = t;
        thread_data_array[t].msg = split_ShmPTR[t];

        rc = pthread_create(&threads[t], NULL, CountWords, (void *) &thread_data_array[t]);
        if (rc)
        {
            printf("ERROR; return code from pthread_create() is %d\n", rc);
            exit(-1);
        }
        //pthread_join(threads[t],(void*)&my_words);
        //printf("%d  %s\n", my_words->thread_id, my_words->wordary[0][0]);
    }
    //pthread_exit(NULL);
    //printf("%s\n", thread_data_array[3].msg);
    pthread_attr_destroy(&attr);
    for(t = 0; t<NUM_THREADS; t++)
    {
        pthread_join(threads[t],(void*)&my_words);
        //printf("%d  %s\n", my_words->thread_id, my_words->wordary[1][0]);
        //printf("%d thread\n", t);
        //printf("%d",my_words->size); 
        if(t == 0)
        {
        //printf("%d  %s\n", my_words->thread_id, my_words->wordary[1][0]);
            for(q = 0; q < my_words->size; q++)
            {
                strcpy(word_array_final[idx][0], my_words->wordary[q][0]);
                strcpy(word_array_final[idx][1], my_words->wordary[q][1]);
                idx++;
                word_array_final[idx][0] = (char*)malloc(30);
            word_array_final[idx][1] = (char*)malloc(8);
            //printf("%s   %s\n", word_array_final[idx][0], word_array_final[idx][1]);
            }
        }
        else
        {
        //printf("%d  %s  %d\n", my_words->thread_id, my_words->wordary[1][0], my_words->size);
            for(q = 0; q<my_words->size; q++)
            {   
                flag1 = 0;
                for(r = 0; r<idx; r++)
                {
                    if(0 == (strcmp(word_array_final[r][0],my_words->wordary[q][0])))
                    {
                        flag1 = 1;
                    count = atoi(my_words->wordary[q][1]) + atoi(word_array_final[r][1]);
                    sprintf(buff, "%d", count);
                    strcpy(word_array_final[r][1],buff);
                    }
                    //printf("%s   %s1\n", word_array_final[idx][0], word_array_final[idx][1]); 
                }

                if(flag1 == 0)
            {
                strcpy(word_array_final[idx][0],my_words->wordary[q][0]);
                strcpy(word_array_final[idx][1],my_words->wordary[q][1]);
                idx++;
                word_array_final[idx][0]=(char*)malloc(30);
                word_array_final[idx][1]=(char*)malloc(8);                  
            }
            }
        }

    }

    out_file=fopen("output.txt","w");

    for(z=0; z<idx; z++)
{
    fprintf(out_file, "%s : %s\n", word_array_final[z][1], word_array_final[z][0]);
}

printf("done");
fclose(out_file);

    //pthread_exit(NULL);

    printf("   Client is about to exit\n");
    exit(0);
 }         
 wait(&status);
 printf("Server has detected the completion of its child...\n");
 shmdt((void *) ShmPTR);
 printf("Server has detached its shared memory...\n");
 shmctl(ShmID, IPC_RMID, NULL);
 printf("Server has removed its shared memory...\n");
 printf("Server exits...\n");
 end = clock();
 time_spent = (double)(end - begin) / CLOCKS_PER_SEC;
 printf("Time spent: %lf\n", time_spent);
 exit(0);
 }  

请帮助我，任何帮助都将不胜感激。

Answer 1

我找到了解决方案...这是一个愚蠢的错误...我的数组的大小较小...现在它已修复...并且该程序可以用作子程序 - 使用共享内存的父程序也具有映射器和 reducer 。