[C코드] :: MULTIPROCESS QUEUE code (멀티 프로세스를 이용한 큐 코드)

헤더파일 : MultiProcessWithSemaphore.h

/*******************************************************************************
 * MultiProcessWithSemaphore.h
 *
 *
 *
 * IDENTIFICATION & REVISION
 *        $Id$
 *
 * NOTES
 *
 *
 ******************************************************************************/

/**
 * @file MultiProcessWithSemaphore.h
 * @brief 세마포어를 이용한 멀티 프로세스 큐
 *        헤더파일
 */


#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/shm.h>
#include <semaphore.h>
#include <fcntl.h>
#include <time.h>


#define KEY_ID          9300
#define SIZE            8192
#define PROCESS_COUNT   10
#define DATA_MAX_COUNT  5

#define STL_SUCCESS     0
#define STL_FAILURE     1


#define STL_TRY_THROW( aExpression, aLabel )    \
    do                                          \
    {                                           \
        if( !(aExpression) )                    \
        {                                       \
            goto aLabel;                        \
        }                                       \
    } while( 0 )


#define STL_CATCH( aLabel )                     \
    goto STL_FINISH_LABEL;                      \
    aLabel:


#define STL_FINISH                              \
    STL_FINISH_LABEL:


typedef struct qNode
{
    int mData;

    struct qNode * mPrevNode;
    struct qNode * mNextNode;
} qNode;


typedef struct qQueue
{
    qNode * mFrontNode;
    qNode * mRearNode;
} qQueue;


typedef struct  qStruct
{
    qQueue  mQueue;
    void  * mShmAddr;
    int     mCount;
    sem_t   mSemaphore;
} qStruct;


void * qInitialize( qStruct * aStruct );


void * qEnqueue( qStruct * aStruct,
                 int       aProcessId );


void * qDequeue( qStruct * aStruct,
                 int       aProcessId );

소스파일 : MultiProcessWithSemaphore.c

/*******************************************************************************
 * MultiProcessWithSemaphore.c
 *
 *
 *
 * IDENTIFICATION & REVISION
 *        $Id$
 *
 * NOTES
 *
 *
 ******************************************************************************/

/**
 * @file MultiProcessWithSemaphore.c
 * @brief 세마포어를 이용한 멀티 프로세스 큐
 */


#include "MultiProcessWithSemaphore.h"


void * gShmAddrStart;


int main()
{
    pid_t  sProcessId[PROCESS_COUNT];

    void * sShmAddr   = NULL;


    int sShmId        = 0;
    int sIsDetachShm  = 0;
    int sIsRemoveShm  = 0;
    int sRepeatCount  = 0;
    int sStatus       = 0;

    int sStructSize   = 0;
    int sNodeSize     = 0;

    qStruct * sStruct = NULL;

    
    printf("+-----------------------------------+\n");
    printf("| [SYSTEM] :: program is Started... |\n");
    printf("+-----------------------------------+\n");


    /*
     * Initialize
     */


    /* 공유 메모리 생성 */
    sShmId = shmget( KEY_ID,
                     SIZE,
                     IPC_CREAT | 0666 );

    /* 공유 메모리 생성 실패 확인 */
    STL_TRY_THROW( sShmId != -1,
                   RAMP_ERROR_01 );

    printf("공유 메모리를 생성하였습니다.\n");


    /* 공유 메모리 attach */
    gShmAddrStart = shmat( sShmId,
                           (void *)0,
                           0 );

    /* attach 실패 확인 */
    STL_TRY_THROW( gShmAddrStart != (void *)-1,
                   RAMP_ERROR_02 );

    printf("공유 메모리를 attach 하였습니다.\n");

    
    /*
     * attach한 shared memory 주소로 구조체를 캐스팅
     *
     *  < SHARED MEMORY >
     *  +---------+-------------------------------------------------------
     *  | sStruct |
     *  +---------+-------------------------------------------------------
     *  ^
     *  gShmAddrStart
     *  
     */
    
    sStruct = (qStruct *)gShmAddrStart;

    /* 구조체 sStruct 초기화 */
    qInitialize( sStruct );

    /* 각 구조체들의 크기 */
    sStructSize = sizeof( qStruct );
    sNodeSize   = sizeof( qNode );

    /* ( shared memory 주소값 ) + ( 구조체 크기 )값을 구조체 멤버에 저장 */
    sStruct->mShmAddr = gShmAddrStart + sStructSize;

    
    /*
     * DO
     */


    for( sRepeatCount = 0; sRepeatCount < PROCESS_COUNT; sRepeatCount++ )
    {
        sProcessId[sRepeatCount] = fork();


        if( sProcessId[sRepeatCount] > 0 )
        {
            /* 부모 프로세스 */
            /* do nothing */
        }
        else if( sProcessId[sRepeatCount] == 0 )
        {
            /* 자식 프로세스 */
            /* 자식 프로세스 절반은 enqueue, 절반은 dequeue */
            if( sRepeatCount % 2 == 0 )
            {
                /* Enqueue */
                qEnqueue( sStruct, sRepeatCount );
            }
            else
            {
                /* Dequeue */
                qDequeue( sStruct, sRepeatCount );
            }
            
            exit( STL_SUCCESS );
        }
        else
        {
            /* fork 실패 */
            STL_TRY_THROW( STL_SUCCESS,
                           RAMP_ERROR_04 );
        }
    }



    /*
     * Finalize
     */


    /* 부모 프로세스의 wait 호출 */
    for( sRepeatCount = 0; sRepeatCount < PROCESS_COUNT; sRepeatCount++ )
    {
        sProcessId[sRepeatCount] = wait( &sStatus );
    }
    
    
    /* 세마포어 종료 */
    sem_destroy( &sStruct->mSemaphore );

    printf("세마포어를 종료하였습니다.\n");
    
    /* 공유 메모리 detach */
    sIsDetachShm = shmdt( gShmAddrStart );

    /* detach 실패 확인 */
    STL_TRY_THROW( sIsDetachShm != -1,
                   RAMP_ERROR_03 );
    
    printf("공유 메모리를 detach 하였습니다.\n");
    


    printf("+-----------------------------------+\n");
    printf("| [SYSTEM] :: program is Exited...  |\n");
    printf("+-----------------------------------+\n");
    

    
    return STL_SUCCESS;

    
    STL_CATCH( RAMP_ERROR_01 )
    {
        printf("[ERROR-01] :: cannot get shared memory which key '%d'.\n", KEY_ID );
    }

    STL_CATCH( RAMP_ERROR_02 )
    {
        printf("[ERROR-02] :: cannot attach shared memory which key '%d'.\n", KEY_ID );
    }

    STL_CATCH( RAMP_ERROR_03 )
    {
        printf("[ERROR-03] :: cannot detach shared memory which key '%d'.\n", KEY_ID );
    }

    STL_CATCH( RAMP_ERROR_04 )
    {
        printf("[ERROR-04] :: cannot create child process.\n");
    }

    
    STL_FINISH;
    

    return STL_FAILURE;
}


void * qInitialize( qStruct * aStruct )
{
    aStruct->mQueue.mFrontNode = NULL;
    aStruct->mQueue.mRearNode  = NULL;
    
    aStruct->mShmAddr = NULL;
    aStruct->mCount   = 0;

    /* 세마포어 초기화 */
    sem_init( &aStruct->mSemaphore,
              1,
              1 );
}


void * qEnqueue( qStruct * aStruct,
                 int       aProcessId )
{
    int     sRepeatCount = 0;
    int     sNodeSize    = 0;

    qNode * sNewNode     = NULL;

    sNodeSize = sizeof( qNode );

    /* rand() */
    srand( aProcessId );

    while( sRepeatCount < DATA_MAX_COUNT )
    {
        /* semaphore lock */
        sem_wait( &aStruct->mSemaphore );

        if( aStruct->mCount >= PROCESS_COUNT )
        {
            /* printf("queue is full!\n"); */
        }
        
        else
        {
            /*
             *  < SHARED MEMORY >
             *  +---------+----------+--------------------------------------------
             *  | sStruct | sNewNode |                   
             *  +---------+----------+--------------------------------------------
             *            ^
             *            mShmAddr
             *  
             */
            
            sNewNode = (qNode *)aStruct->mShmAddr;
            aStruct->mShmAddr += sNodeSize;


            /* 두 자리 임의의 수를 sNewNode의 mData에 저장 */
            sNewNode->mData = rand()%90 + 10;
            sNewNode->mNextNode = NULL;
        
            if( aStruct->mCount == 0 )
            {
                aStruct->mQueue.mFrontNode = sNewNode;
                sNewNode->mPrevNode = NULL;
            }
            else
            {
                aStruct->mQueue.mRearNode->mNextNode = sNewNode;
                sNewNode->mPrevNode = aStruct->mQueue.mRearNode;
            }

            aStruct->mQueue.mRearNode = sNewNode;
            aStruct->mCount++;

            printf("[#%3d][ENQUEUE_DATA] : %d  QUEUE STATUS [ %2d/%d ]\n",
                   aProcessId,
                   aStruct->mQueue.mRearNode->mData,
                   aStruct->mCount,
                   PROCESS_COUNT );

            sRepeatCount++;
        }

        /* semaphore unlock */
        sem_post( &aStruct->mSemaphore );
        usleep(50);
    }
}


void * qDequeue( qStruct * aStruct,
                 int       aProcessId )
{
    int    sRepeatCount = 0;
    int    sOutputData  = 0;
    int    sStructSize  = 0;
    int    sNodeSize    = 0;
    
    void * sShmAddr     = NULL;

    sStructSize = sizeof( qStruct );
    sNodeSize   = sizeof( qNode );
    

    while( sRepeatCount < DATA_MAX_COUNT )
    {
        sem_wait( &aStruct->mSemaphore );

        if( aStruct->mCount == 0 )
        {
            /* printf("queue is empty!\n"); */
        }

        else
        {
            sOutputData = aStruct->mQueue.mFrontNode->mData;

            if( aStruct->mCount == 1 )
            {
                aStruct->mQueue.mFrontNode = NULL;
                aStruct->mQueue.mRearNode  = NULL;
                aStruct->mCount            = 0;
            }

            else
            {
                aStruct->mQueue.mFrontNode = aStruct->mQueue.mFrontNode->mNextNode;
                aStruct->mQueue.mFrontNode->mPrevNode = NULL;
                aStruct->mCount--;
            }

            printf("[#%3d][DEQUEUE_DATA] : %d  QUEUE STATUS [ %2d/%d ]\n",
                   aProcessId,
                   sOutputData,
                   aStruct->mCount,
                   PROCESS_COUNT );

            sRepeatCount++;
        }

        sem_post( &aStruct->mSemaphore );
        usleep(50);
    }
}

공부/코딩연습 등의 이유로 얼마든지 퍼가셔도 좋습니다.

하지만 라인마다 의미를 파악하지 않고 무작정 복사 붙여넣기는

아무것도 남지 않습니다.

또한 댓글로 궁금하신 라인 등 얼마든지 물어보시면 

바로바로 대답해드리겠습니다 :)

공부를 게을리하지 맙시다!