소켓 프로그래밍 - (25) 세마포어(Semaphore)를 이용한 멀티 스레드 통신

- 본 내용은 Linux (Ubuntu 14.04 lts)를 기반으로 제작되었습니다. -

소캣 프로그래밍은 TCP/IP 기반으로 하였습니다.

코딩에 이용되는 중요한 것들

1. 세마포어(Semaphore) :: http://www.crocus.co.kr/487

3. 스레드 개념 및 응용 :: http://www.crocus.co.kr/484

4. TCP 통신 :: http://www.crocus.co.kr/463

5. cpp 파일 컴파일 방법 :: http://www.crocus.co.kr/527

코드 설명 

클라이언트에서 두 수를 보내면, 그 두 수를 서버에서 합하여 클라이언트에 정답을 보내는 코드이다.

스레드를 생성하여 서버와 클라이언트간 통신이 되도록 하고, 모두다 concurrent하게 돌도록 하는것이 목적이다. 

semaphore_server.cpp

이 코드는 바인딩, 리스닝은 그대로 하고 어셉트를 메인 스레드에서 하여 어셉트가 되면 스레드에서 통신을 하도록 하는 코드이다.

우선 스레드를 미리 생성해준다.  미리 생성하는 이유는 , 클라이언트가 들어올 때 마다 스레드를 만들어주고 제거해준다면, 

많은 오버헤드가 발생 할 수 있기에, 최대 수용인원으로 제한하는 미리 생성된 스레드를 이용한다.

그다음으로 어셉트를 하기전에 메인 스레드에서 lock을걸어 자식 스레드가 돌지 않도록 해준다.

위의 과정은 뮤텍스에서는 존재하지만, 세마포어에서는 존재하지 않는다.

왜냐하면, 세마포어를 초기화 하는 단계에서 이미 0으로 초기화를 했기에, 자식 스레드가 먼저 돌 일이 없기 때문이다.

그리고 나서 connectFD가 생기면(클라이언트가 접속을 하면), getFD라는 전역변수로 넘겨주고, 

post(V)를 해주게되면, 자식 스레드중 한개가 깨어나서 getFD의 값을 받아내어 통신을 하게된다. 

그리고 나서 스레드는 wait(P)를 통해 세마포어를 이용한다.

이때 받아낸 FD는 큐에 저장시킴으로써 향후 이용할 수 있도록 한다.

client.cpp

두 수를 보내고 받는 용도이다. 

4. TCP 통신 :: http://www.crocus.co.kr/463 를 참조하면 될 것 같다.

 

< tcp.h >

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

//tcp.h   #include <iostream> #include <pthread.h> #include <queue> #include <semaphore.h> #include <stdio.h> #include <string.h> #include <stdlib.h> #include <netinet/in.h> #include <sys/socket.h> #include <arpa/inet.h> #include <unistd.h>   #define PORT 20162 #define BUFFER_SIZE 100 #define LISTEN_QUEUE_SIZE 5   using namespace std;   //                                                       This source code Copyright belongs to Crocus //                                                        If you want to see more? click here >>

Crocus

< semaphore_server.cpp >

//semaphore_server.cpp
 
#include "tcp.h"
  
// 세마포어 생성
sem_t consume;
 
int clientNum = 0;
int getFD = -1;
int id = 0;
 
// 스레드 풀 생성
queue<int> threadPool;
 
void *threadfunc(void *argumentPointer)
{
    int tid = id++; // 스레드마다 id 부여
    int connectFD = -1;
 
    ssize_t receivedBytes;
    char readBuff[BUFFER_SIZE];
    char sendBuff[BUFFER_SIZE];
 
    while(1)
    {
    // 스레드를 이용하기 위해서는 consume 세마포어에 값이 1이상 있어야 한다.
    // consume 세마포어를 1 줄인다.(P)
        sem_wait(&consume);
    
    connectFD = getFD;
    threadPool.push(connectFD);
    getFD = -1;
    
    printf("큐 크기 :: %lu\n",threadPool.size());
    ssize_t receivedBytes;
 
    int n[2];
    int sum = 0;
    int cnt = 0;
 
    // 두 수를 받기위해 cnt < 2라는 조건을 이용한다.
        while(cnt < 2)
        {           
        receivedBytes = read(connectFD, readBuff, BUFFER_SIZE);
                printf("%lu bytes read\n", receivedBytes);
                readBuff[receivedBytes] = '\0';
                fflush(stdout);
    
        // 받은 값은 문자열 형식이니 이것을 숫자로 변환해준다.
        n[cnt] = atoi(readBuff);
        printf("id :: %d // input num :: %d\n",tid,n[cnt]);
 
        sum += n[cnt];     
        cnt++;
 
        // cnt가 2가되면 합계를 내어 클라이언트로 전송해준다.
        if(cnt == 2)
               {    
                   printf("sum :: %d\n",sum);
                    sprintf(sendBuff,"%d",sum);
                      write(connectFD, sendBuff, strlen(sendBuff));        
        }
        }
                
        // 계산이 끝나고 스레드 탈출 전 계산을 해준다.
    threadPool.pop();
    clientNum--;
        close(connectFD); 
        
    }
 
    return NULL;
}
 
 
int main() 
{
    int i;
    pthread_t threadID[LISTEN_QUEUE_SIZE];
   
    struct sockaddr_in listenSocket;
 
    memset(&listenSocket, 0, sizeof(listenSocket));
 
    listenSocket.sin_family = AF_INET;
    listenSocket.sin_addr.s_addr = htonl(INADDR_ANY);
    listenSocket.sin_port = htons(PORT);
 
    int listenFD = socket(AF_INET, SOCK_STREAM, 0);
    int connectFD;
 
 
    // 바인딩
    if (bind(listenFD, (struct sockaddr *) &listenSocket, sizeof(listenSocket)) == -1) {
        printf("Can not bind.\n");
        return -1;
    }
 
    // 리스닝
    if (listen(listenFD, LISTEN_QUEUE_SIZE) == -1) {
        printf("Listen fail.\n");
        return -1;
    }
 
    printf("Waiting for clients...\n");
 
    // 처음에는 스레드가 움직이면 안되니 0으로 초기화한다.
    sem_init(&consume, 0, 0);
 
    // 스레드 생성(큐 사이즈 만큼)    
    for(i = 0 ; i < LISTEN_QUEUE_SIZE; i ++)
    pthread_create(&threadID[i], NULL, threadfunc, (void *)&i);
 
    
 
    ssize_t receivedBytes;
    char readBuff[BUFFER_SIZE];
    char sendBuff[BUFFER_SIZE];
 
    while (1) 
    {
        struct sockaddr_in connectSocket, peerSocket;
 
        socklen_t connectSocketLength = sizeof(connectSocket);
    
    /*
     뮤텍스는 이부분에서 락을 걸고 언락을 해주지만,
     세마포어 입장에서는 이 부분에서 P,V가 없어도 
         자식 스레드와 충돌이 일어나지 않는다고 생각하기에 이용하지 않았다.
     세마포어 초기화 자체에서 0으로 했기에 자식 스레드가 먼저 돌 일이없다.
    */
    // 어셉트를 메인 스레드에서 진행한다. 
        while((connectFD = accept(listenFD, (struct sockaddr*)&connectSocket, (socklen_t *)&connectSocketLength)) >= 0)
        {
        clientNum++;
        
        printf("현재 수용 인원 :: %d / %d \n", clientNum, LISTEN_QUEUE_SIZE);
        if(clientNum > LISTEN_QUEUE_SIZE)
        { 
            // Send Error(Full)
            sprintf(sendBuff,"%s","현재 서버 수용 인원이 꽉 차 접속할 수 없습니다. 잠시후 다시 시도하세요.\n");
                   write(connectFD, sendBuff, strlen(sendBuff));
            clientNum--;
            
        }
        
        else
        {
            // FD를 전역 변수를 이용하여 보낸다.(스레드 인자 사용이 불가능)
            getFD = connectFD;
 
            // consume 세마포어를 1개 늘려준다.(V)
            sem_post(&consume);
         }
 
 
        
        }
        
    }
    close(listenFD);
 
    return 0;
}
 
 
//                                                       This source code Copyright belongs to Crocus
//                                                        If you want to see more? click here >>

< client.cpp >

//client.cpp
 
# include "tcp.h"
 
int main(int argc, char** argv)
{
    if (argc != 2) 
    {
        printf("Usage: %s IPv4-address\n", argv[0]);
        return -1;
    }
 
    struct sockaddr_in connectSocket;
 
    memset(&connectSocket, 0, sizeof(connectSocket));
 
    connectSocket.sin_family = AF_INET;
    inet_aton(argv[1], (struct in_addr*) &connectSocket.sin_addr.s_addr);
    connectSocket.sin_port = htons(PORT);
 
    int connectFD = socket(AF_INET, SOCK_STREAM, 0);
 
    if (connect(connectFD, (struct sockaddr*) &connectSocket, sizeof(connectSocket)) == -1) 
    {
        printf("Can not connect.\n");
        return -1;
    }
 
    else 
    {
        int readBytes, writtenBytes;
        char sendBuffer[BUFFER_SIZE];
        char receiveBuffer[BUFFER_SIZE];
 
            //서버에 문자열을 보낸 뒤 서버가 보낸 echo를 받아 출력.
            printf("수 1 :: ");
 
            fgets(sendBuffer,BUFFER_SIZE,stdin);
 
            write(connectFD, sendBuffer, strlen(sendBuffer));
            
 
            printf("수 2 :: ");
 
            fgets(sendBuffer,BUFFER_SIZE,stdin);
 
            write(connectFD, sendBuffer, strlen(sendBuffer));
 
            readBytes = read(connectFD, receiveBuffer, BUFFER_SIZE);
            printf("%d bytes read\n", readBytes);
            receiveBuffer[readBytes] = '\0';
        
        if(readBytes != 100)
            printf("서버에서 계산한 두 수의 합 :: %s\n",receiveBuffer);  
          
        fputs(receiveBuffer, stdout);
            fflush(stdout);
 
        
        
        
    }
 
    close(connectFD);
    return 0;
 
    
}    
 
 
//                                                       This source code Copyright belongs to Crocus
//                                                        If you want to see more? click here >>