소켓 프로그래밍 - (24) 뮤텍스(Mutex)를 이용한 멀티 스레드 통신

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

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

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

1. 뮤텍스(Mutex) :: http://www.crocus.co.kr/486

2. 조건 변수(Condition value) :: http://www.crocus.co.kr/526

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

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

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

코드 설명 

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

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

mutex_server.cpp

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

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

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

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

이렇게 해야 자식 스레드에서 어셉트를 하는동안 기다릴 수 있게 된다. 

만약 자식 스레드가 먼저 들어오게 된다면, wait를 통해 기다리게 한다.

다음으로 connectFD가 생기면, getFD라는 전역변수로 넘겨주고, 시그널을 해주게되면, 자식 스레드중 한개가 깨어나서

getFD의 값을 받아내어 통신을 하게된다. 이때 받아낸 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

< mutex_server.cpp >

//mutex_server.cpp
 
#include "tcp.h"
  
// 뮤텍스 생성 및 컨디션 생성
pthread_mutex_t tcpmutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t tcpcond = PTHREAD_COND_INITIALIZER;
 
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)
    {
    // 조건 변수를 이용한 뮤텍스 락을 건다.
    pthread_mutex_lock(&tcpmutex);
     
    // 메인 스레드에서 FD를 받지 못하면 wait
    while(getFD == -1)
        pthread_cond_wait(&tcpcond,&tcpmutex);
 
    connectFD = getFD;
    threadPool.push(connectFD);
    getFD = -1;
    
    printf("큐 크기 :: %lu\n",threadPool.size());
    ssize_t receivedBytes;
 
    // read할 값이 있다면 계속 읽어들인다.
    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); 
        
    pthread_mutex_unlock(&tcpmutex);
      
    }
 
    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");
 
 
    // 스레드 생성(큐 사이즈 만큼)    
    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);
 
    // 메인 스레드 역시 자식 스레드에 방해를 받지 않기위해 락을 건다.
    pthread_mutex_lock(&tcpmutex);
 
        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--;
            pthread_mutex_unlock(&tcpmutex);
        }
        
        else
        {
            // FD를 전역 변수를 이용하여 보낸다.(스레드 인자 사용이 불가능)
            getFD = connectFD;
 
            // signal을 이용하여 자식 스레드 중 1개를 깨운다.
            pthread_cond_signal(&tcpcond);
            pthread_mutex_unlock(&tcpmutex);
         }
 
 
        
        }
        
    }
    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 >>