병렬 프로세스를 통한 대량 처리 Sample 프로그램

이 프로그램은 대량의 파일을 병렬 프로세스를 통하여 빠른 처리를 하기 위한 Sample 프로그램입니다. 

 

header file include

/*
*  대량의 파일에 대해 병렬로 처리하기 위한 프로그램의 예제.
*  이 프로그램은 특정 디렉토리 아래의 /---/dbio 디렉토리에 있는 
*   tar 파일을 병렬초 처리하는 프로그램입니다.
*/
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <sys/types.h>
#include <unistd.h>
#include <dirent.h>
#include <sys/stat.h>
#include <signal.h>

 

처리할 데이터는 파일 단위이므로 path와 파일명에 대한 자료구조 정의 

#define LEN_PATH        512
#define LEN_FILENAME    128

/*
* 파일의 위치와 파일 이름을 저장하는 구조체
*/
typedef struct 
{
    char path[LEN_PATH];            /* 파일의 위치*/
    char filename[LEN_FILENAME];    /* 파일명 */
} file_info_t;

file_info_t *tar_file_list;         /* 파일정보를 저장할 Pointer 변수 */

 

전역 변수와 함수 선언

int tar_file_count = 0;         /* 실제 파일 정보의 갯수 */
int tar_file_max_count = 20000; /* 파일의 정보를 저장할 공간 크기 (최초 20000개)*/
int process_count = 0;          /* 병렬로 실행할 갯수 */
int prc_file_start;             /* 하나의 프로세스가 처리할 파일의 위치(index) */
int prc_file_count;             /* 하나의 프로세스가 처리할 갯수 */
char source_home[1024];         /* 파일을 검색할 시작 위치 */

int search_file_list(const char *job);  /* 파일 정보를 읽어들이는 함수 */
int loading(void);                      /* 파일을 처리하는 logic */

 

도움말 함수

int help(char *argv[])
{
    fprintf(stderr, "USAGE:\n\t%s -load [process갯수]\n", argv[0]);
    fprintf(stderr, "\t-load : loading을 하라는 option option\n");
    fprintf(stderr, "\t전체 dbio tar를 loading하는 프로그램입니다.\n");
}

 

main 함수 정의

int main(int argc, char *argv[])
{
    DIR           *dir_ptr = NULL;
    struct dirent *file    = NULL;
    struct stat   buf;
    char   filename[1024];
    int    ret;
    int    idx;

    /* argv[]를 통하여 입력받을 수도 있도록 수정할 수 있음 */
    snprintf(source_home, 1024, "%s/src/shmd", getenv("HOME"));

    signal(SIGCHLD, SIG_IGN);

    if(argc == 1) {
        help(argv);
        return 1;
    }

    if(strcmp(argv[1], "-load") != 0) {
        help(argv);
        return 1;
    }

    if(argc == 3) {
        process_count = atoi(argv[2]);
    }

    if(process_count <= 0) {
        process_count = 100;  /* default 100개 프로세스로 처리 */
    }

    tar_file_list = (file_info_t *)malloc(sizeof(file_info_t) * tar_file_max_count);

    if((dir_ptr = opendir(source_home)) == NULL) {
        fprintf(stderr, "%s DIRECTORY INFO READ ERROR.\n", source_home);
        return 1;
    }

    while((file = readdir(dir_ptr)) != NULL) {
        if(strcmp(file->d_name, ".") == 0 || strcmp(file->d_name, "..") == 0 ) {
             continue;
        }
        sprintf(filename, "%s/%s", source_home, file->d_name);
        if(stat(filename, &buf) == -1) {
            continue;
        }
        if(S_ISDIR(buf.st_mode)) {
            if(strlen(file->d_name) != 3) {
                continue;
            }
            search_file_list(file->d_name);
        }
    }
    closedir(dir_ptr);

    prc_file_count = tar_file_count / process_count;  /* 하나의 프로세스가 처리할 갯수 정의 */
    prc_file_start = 0;

    for(idx = 0; idx < process_count; idx++) {
        if((ret = fork()) == 0) { // child process
            loading();
            exit(0);
        } else if(ret < 0) {
            exit(0);
        }
        prc_file_start += prc_file_count;
    }

    /* 1/n 하고 나머지 부분은 스스로 처리 */
    loading();
    return 0;
} 

 

처리할 파일 목록을 검색하는 로직

int search_file_list(const char *job)
{
    DIR *dir_ptr = NULL;
    struct dirent *file = NULL;
    struct stat   buf;
    char   filename[1024];
    char   command[1024];
    char   home[1024];
    char   *ext;

    sprintf(home, "%s/%s/dbio", source_home, job);

    if((dir_ptr = opendir(home)) == NULL) {
        return 0;
    }

    while((file = readdir(dir_ptr)) != NULL) {
        sprintf(filename, "%s/%s", home, file->d_name);
        if(stat(filename, &buf) == -1) {
            continue;
        }
        if(S_ISREG(buf.st_mode)) {
            if((ext = strrchr(file->d_name, '.')) == NULL) {
                continue;
            }
            /* tar 파일이 아니면 skip */
            if(strcmp(ext, ".tar") != 0) {
                continue;
            }
            /* 할당한 최대 저장 공간보다 파일의 갯수가 많으면 메모리 재할당  */
            if(tar_file_count >= tar_file_max_count) {
                tar_file_max_count += 10000;  /* 10000개씩 메모리 더 할당*/
                tar_file_list = (file_info_t *)realloc(tar_file_list, sizeof(file_info_t) * tar_file_max_count);
            }
            strncpy(tar_file_list[tar_file_count].path,     home,           LEN_PATH);
            strncpy(tar_file_list[tar_file_count].filename, file->d_name,   LEN_FILENAME);
            tar_file_count++;
        }
    }

    closedir(dir_ptr);
    return 0;
} 

 

하나의 프로세스가 실제로 처리하는 로직

int loading(void)
{
    char   filename[1024];
    char   command[1024];
    int    to_pos = prc_file_start + prc_file_count;

    signal(SIGCHLD, SIG_IGN);

    for(;prc_file_start < to_pos && prc_file_start < tar_file_count; prc_file_start++) {
        /* 파일이 있는 디렉토리로 이동 */
        chdir(tar_file_list[prc_file_start].path);

        /* 파일처리 명령어 실행: 명령어는 업무에 따라서 설정 */
        sprintf(command, "--파일 명령어-- %s", tar_file_list[prc_file_start].filename);
        printf("command: %s\n", command);
        system(command);
    }

    return 0;
}