미완성 코드다. g++ 로 컴파일 하면, 메시지큐를 통해서 worker thread에서 main thread로 분석된 Term이 전달되는걸 확인할 수 있다. 메시지큐를 통한 데이터 통신의 기본적인 구현은 끝났다고 볼 수 있다. 코드를 좀더 깔끔하게 하고, 몇 군데 예외처리를 해주어야 한다.

#include <pthread.h> 
#include <unistd.h> 
#include <stdlib.h> 
#include <stdio.h> 
#include <sys/types.h> 
#include <sys/ipc.h> 
#include <sys/msg.h> 
#include <sys/stat.h> 
#include <regex.h> 
 
#include <fcntl.h> 
#include <string.h> 
 
#define ThreadNum 4 
 
pthread_mutex_t mutex_lock; 
pthread_cond_t  sync_cond; 
 
pthread_mutex_t gmutex = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t  gcond  = PTHREAD_COND_INITIALIZER; 
 
const char seps[] = " -.,()\";:{}'+@/<>[]|!?#\n"; // Token 
 
// 주고 받을 데이터 
struct Data 
{ 
    int queuenum; 
    int msgtype;  
    int tid; 
    char msg[256]; 
    int size; 
    key_t key_id; 
}; 
 
// worker 쓰레드에 넘겨줄 정보 
struct DocInfo 
{ 
    int tnum;       // thread id 
    int start;      // 문서시작 위치 
    int end;        // 문서 끝 위치 
    char fname[80]; // 작업 파일명 
    char regex[24]; // 정규표현 문자열  
}; 
 
/* 
 * worker 쓰레드 함수 
 */ 
void *Tfunction(void *data) 
{ 
    struct Data lData; 
    struct DocInfo *lDocInfo;  
    FILE *fp; 
    char line[256]; 
    int rtv; 
    int bsize; 
    int readn = 0; 
  regex_t preg; 
  char *tr; 
 
    lData = *(struct Data *)(data); 
    lDocInfo = (struct DocInfo *)lData.msg;  
 
    bsize = lDocInfo->end - lDocInfo->start; 
    printf("DEBUG Thread %d %d\n", lDocInfo->tnum, bsize); 
    if((fp = fopen(lDocInfo->fname, "r")) == NULL) 
    { 
        perror("Fopen Error"); 
        exit(0); 
    } 
 
    lseek(fileno(fp), lDocInfo->start, SEEK_SET);     
  rtv = regcomp(&preg, lDocInfo->regex, REG_EXTENDED|REG_NOSUB); 
 
    pthread_mutex_lock(&mutex_lock); 
    pthread_cond_signal(&sync_cond); 
    pthread_mutex_unlock(&mutex_lock); 
 
    if (rtv != 0) 
    { 
        regerror(rtv, &preg, lDocInfo->regex, REG_EXTENDED|REG_NOSUB); 
        exit(0); 
    } 
    for (readn = 0; readn < bsize; ) 
    { 
        if(fgets(line, 256, fp) == NULL) 
            break; 
        tr = strtok(line, seps); 
        if (tr != NULL) 
        { 
            if (regexec(&preg, tr, 0, NULL, 0) == 0) 
            { 
                lData.queuenum = 1; 
                lData.msgtype = 1 << 1; 
                lData.size = strlen(tr); 
                sprintf(lData.msg, "%s", tr); 
 
                if(msgsnd(lData.key_id, (void *)&lData, sizeof(lData), IPC_NOWAIT) <0) 
                { 
                    perror("msg snd error"); 
                    exit(0); 
                } 
            } 
        } 
        readn += strlen(line); 
        sleep(1); 
    } 
    // 종료 메시지를 보낸다. 
} 
 
 
int main(int argc, char **argv) 
{ 
    struct Data lData; 
    struct DocInfo lDocInfo;  
 
    pthread_t p_thread[ThreadNum]; 
 
    int fd; 
    char *fname; 
    int fsize = 0; 
    int blocksize = 0; 
    struct stat fileinfo; 
    char *regex; 
 
    // Message queue 
    key_t key_id; 
    int msgtype;     
 
    fname = argv[1]; 
    regex =  argv[2]; 
 
    // 메시지큐 생성 
    key_id = msgget((key_t)8888, IPC_CREAT|0666); 
    if (key_id == -1) 
    { 
        perror("msgget error : "); 
        exit(0); 
    } 
 
    // Open File 
    fd = open(fname, O_RDONLY);     
    if (fd < 0) 
    { 
        perror("Open file error"); 
        return 1; 
    } 
    if(fstat(fd, &fileinfo)< 0)  
    { 
        return 1; 
    } 
    fsize = fileinfo.st_size; 
    printf("T Size is %d\n", fsize); 
 
    blocksize =  fsize / ThreadNum;  
 
    for (int i = 0; i < ThreadNum; i++) 
    { 
        lDocInfo.start = (i*blocksize); 
        lDocInfo.end = lDocInfo.start+blocksize; 
        sprintf(lDocInfo.fname, "%s", fname); 
        sprintf(lDocInfo.regex, "%s", regex); 
        lDocInfo.tnum = i; 
 
        if (i == (ThreadNum -1)) 
            lDocInfo.end += fsize%ThreadNum; 
        lData.queuenum = 1;  
        lData.msgtype = 1 << 2;  
        lData.size = sizeof(lDocInfo);  
        lData.key_id = key_id; 
        lData.tid = lDocInfo.tnum;  
        memcpy((void *)lData.msg, (void *)&lDocInfo, sizeof(lDocInfo)); 
 
        pthread_mutex_lock(&mutex_lock); 
        pthread_create(&p_thread[i], NULL, Tfunction, (void *)&lData); 
        pthread_cond_wait(&sync_cond, &mutex_lock); 
        pthread_mutex_unlock(&mutex_lock); 
    } 
 
    while(1) 
    { 
        if(msgrcv(key_id, (void *)&lData, sizeof(lData), (1 >> 1), 0)    == -1) 
        { 
            perror("msgrcv error: "); 
        } 
        printf("Get Term %d %s\n", lData.tid, lData.msg); 
    } 
} 
 

4 공유메모리 구현

공유메모리는 커널에 메모리 공간을 할당함으로써, 시스템 전역적으로 자원을 사용할 수 있도록 지원하는 IPC 설비중 하나다. 이러한 특성을 이용 쓰레드간 메시지 교환은 공유메모리상에 환형큐를 만드는 것으로 구현가능할 수 있을 것이다.

이 환형큐에는 여러개의 쓰레드가 접근을 하게 될 것이므로, 쓰레드간 동기화가 필수 적일건데, mutex가 아닌 레코드 잠금으로도 쓰레드간 접근제어를 할 수 있다. 레코드 잠금을 통해서 쓰레드동기화를 제어하는 방법은 공유메모리와 파일잠금을 이용한 프로세스간 데이터 공유에서 이미 다룬바가 있다.

쓸만한 예제라고 생각은 하지만 이번 구현에 그대로 적용하기에는 성격이 다르다. 이번 구현은 여러개의 생산자와 하나의 소비자가 존재하는 형식이기 때문이다. 뭐 그렇다고 크게 문제될건 없다. 아래와 같이 생산/소비자 방식을 약간 수정하는 것으로 해결책을 삼았다.

쓰레드간 동기화는 그대로 파일의 레코드 잠금을 이용할 것이다. 이 경우 다수의 생산자를 제어해야 하는데, 처음 8 byte를 생산자간 접근제어를 위한 잠금으로 사용할 것이다. 처음 레코드에 대한 생산자만이 레코드에 큐(빨간색)에 쓸권한을 가지게 된다.

데이터 쓰기 권한을 얻은 쓰레드는 공유메모리에 데이터를 쓰고, 해당 공유메모리에 대응되는 레코드에 잠금을 풀게된다. 그러면 소비자는 잠금을 얻고, 잠금에 대응되는 공유메모리를 찾아가서 정보를 읽어오면 된다. 예제로 제시한 문서를 이해하고 있다면, 위의 방식으로 수정하는건 그리 어렵지 않을 것이니, 굳이 코드를 만들진 않도록 하겠다.