멀티 스레드, 크리티컬 섹션 동기화

프로세서 (Processor)

컴퓨터 내부에서 수행하는 하드웨어의 Unit입니다.

CPU와 ALU(Arithmetic Logic Unit), 마이크로 프로세서와 같이 하드웨어 Unit을 의미합니다.

 

프로세스 (Process)

메모리 공간을 차지한 상태에서 프로세서에 의해 실행중인 프로그램입니다.

 

스레드 (Thread)

프로그램 실행 흐름의 단위입니다.

C++은 메인 스레드가 생성되면서 해당 메인 스레드는 main 함수를 동작합니다.

(Windows 응용 프로그램이라면 WinMain을 동작)

스레드는 환경에 따라 둘 이상의 스레드를 생성해서 동시에 진행할 수 있습니다.

(멀티 스레드)

 

순서

1. 프로그램 실행

2. 바이너리 코드가 메모리에 적재 

3. 프로세서에 의해 프로그램 실행 (프로세스)

4. 1개의 메인 스레드 생성 (프로그램 환경에 따라 여러 개의 스레드 생성)

5. 프로그램 동작

 

동기화 (Synchronization)

동기화라고하면 멀티 스레드를 같이 떠올리는게 좋을 것 같습니다.

CPU가 하나의 프로세스를 실행하고 있을 때, 멀티 스레딩도 하나의 프로세스 안에서 실행되는 것이므로

각각의 스레드는 서로 메모리를 공유할 수 있습니다. 이로 인해 예기치 않은 문제를 막는 것이 동기화입니다.

 

예를 들어 어떤 프로세스에 int a = 100 라는 값이 있습니다.

1번 스레드에서 ++ 연산을 한 번,  2번 스레드에서 -- 연산을 한 번 수행하는

2개의 스레드가 존재합니다. (멀티 스레딩)

 

1번 스레드가 실행되면 a의 값은 101이 됩니다.

그리고 운영체제의 요청으로 1번 스레드의 우선 순위가 2번 스레드로 밀려납니다.

1번 스레드에서 101이 된 a의 값을 레지스터에 저장해둡니다.

 

이후에 새로 실행되는 2번 스레드에서 레지스터에 저장된 a의 값 101을 가져옵니다.

그리고 2번 스레드에 예정된 -- 연산을 하면 a의 값은 101에서 100이 됩니다.

이것을 컨텍스트 스위칭이라고 합니다.

 

하지만 여러 개의 스레드는 하나의 프로세스 안에 존재해서

스레드마다 서로 메모리를 공유하는 것이 가능하다고 했습니다.

 

1번 스레드로 a의 값이 101이 되고, 레지스터에 덮어쓰기 전에

2번 스레드가 실행되면서 컨텍스트 스위칭이 발생할 수 있습니다.

 

그러면 레지스터에 저장되어 있는 a의 값은 여전히 기존 값인 100이므로

2번 스레드를 통해 a의 값은 100에서 99가 되어버립니다.

멀티 스레드에서 이러한 의도치 않은 동작을 동기화로 막아야합니다.

 

동기화는 커널 모드와 유저 모드가 존재합니다.

 

커널 모드는 운영체제의 커널단에서 지원하기 때문에 막강한 기능을 자랑하지만

유저 모드에 비해 속도가 느립니다. 뮤텍스, 세마포어, 이벤트가 있습니다.

 

반대로 유저 모드는 커널 모다 보다 기능이 약하지만 속도가 빠르다는 장점이 있습니다.

대표적으로 크리티컬 섹션이 있고 여기서 다루고자 하는 내용입니다.

 

유저 모드 동기화 : 크리티컬 섹션 (Critical section)

말 그대로 임계 영역을 보호할 수 있는 동기화 기법입니다.

CRITICAL_SECTION이라는 타입을 정의해서 사용합니다.

 

별도의 커널 오브젝트를 활용하지 않아서 속도가 빠르지만,

하나의 프로세스에 여러 개의 스레드 사이에서만 사용이 가능합니다.

 

현재 수행중인 스레드가 크리티컬 섹션을 벗어나기 전 까지,

다른 스레드가 해당 섹션에 접근 하려고하면 운영체제가 스케쥴링하지 않도록 막습니다.

 

스레드 생성, 크리티컬 섹션 C++ 소스 코드

#include <iostream>
 
// Thread 생성과 동기화를 사용할 수 있는 기능을 제공
#include <process.h>
#include <Windows.h>
 
using namespace std;
 
// Critical Section을 원하는 시점에 처리하기 위한 클래스
class CSync
{
public:
    CSync(CRITICAL_SECTION* pCrt)
    {
        m_pCrt = pCrt;
 
        // 객체 생성시 그 시점부터 Critical Section에 진입하게 된다.
        // 후발 스레드가 진입하려고 하는 경우 진입 대기 상태로 만든다.
        // 이후, 선점 스레드(EnterCriticalSection)가 크리티컬 영역에서 벗어났으면
        // 후발 스레드가 실행하게 된다.
        EnterCriticalSection(m_pCrt);
    }
 
    ~CSync()
    {
        // 진입이 끝난 임계 영역을 해제한다.
        // (CRITICAL_SECTION을 처음부터 갱신)
        LeaveCriticalSection(m_pCrt);
    }
 
private:
    CRITICAL_SECTION*    m_pCrt;
};
 
CRITICAL_SECTION    g_Crt;
 
unsigned int __stdcall Thread1(void* pArg)
{
    int* pNumber = (int*)pArg;
 
    // EnterCriticalSection 호출
    CSync sync(&g_Crt);
 
    for (int i = 0; i < 10000; ++i)
    {
        ++(*pNumber);
    }
 
    // 스레드 종료시 LeaveCriticalSection 호출
    return 0;
}
 
unsigned int __stdcall Thread2(void* pArg)
{
    int* pNumber = (int*)pArg;
 
    CSync sync(&g_Crt);
 
    for (int i = 0; i < 10000; ++i)
    {
        --(*pNumber);
    }
 
    return 0;
}
 
int main()
{
    // 크리티컬 섹션 초기화
    InitializeCriticalSection(&g_Crt);
 
    int iNumber = 100;
 
    // 2개의 스레드를 생성한다.
    // _beginthreadex 함수 내부적으로 CreateThread 함수를 호출한다.
    // 기존의 _beginthread 함수와 차이점은 별도의 메모리 블록을 할당해서 
    // 동기화에서 더욱 안전한 동작을 보장을 해준다고 한다.
    // 스레드를 생성하면 프로세스 핸들 테이블에 핸들이 등록되며 Usage Count가 증가하게 된다.
    // 1) void *security : 스레드의 보안관련 정보를 전달한다(없으면 null)
    // 2) unsigned stack_size : 스택 사이즈 전달(기본 크기 1MB로 설정하려면 0)
    // 3) unsigned (* start_address)(void*) : 쓰레드 시작함수 주소 전달
    // 4) void *arglist : 함수 호출 시 전달하고 싶은 인자 전달
    // 5) unsigned initflag : 스레드 생성 이후의 행동을 결정. 0을 전달하면 생성과 동시에 실행
    // 6) unsigned *thraddr : 쓰레드 ID의 저장을 위한 변수의 주소값
    HANDLE hThread1 = (HANDLE)_beginthreadex(nullptr, 0, Thread1, &iNumber, 0, 0);
    HANDLE hThread2 = (HANDLE)_beginthreadex(nullptr, 0, Thread2, &iNumber, 0, 0);
 
    // 특정 스레드의 시그널이 발생하기 전 까지 다른 스레드(여기서는 메인 스레드)의 
    // 실행 흐름을 대기시킨다. 스레드 실행이 종료되면 넌 시그널 -> 시그널 상태로 바뀐다.
    // (시그널 상태는 핸들값을 통해서 판단하게 된다)
    WaitForSingleObject(hThread1, INFINITE);
    WaitForSingleObject(hThread2, INFINITE);
 
    cout << iNumber << endl;
 
    // 사용이 끝난 핸들을 닫아서 CIRITICAL_SECTION의 연결 고리를 끊는다.
     CloseHandle(hThread1);
     CloseHandle(hThread2);
 
    DeleteCriticalSection(&g_Crt);
 
    return 0;
}

 

멀티 스레딩에서 동기화를 하지 않으면 스레드의 인자로 넘겨준 iNumber의 값이 랜덤으로 바뀝니다.

그러나 크리티컬 섹션 동기화를 통해 iNumber의 값이 일정하게 100을 출력하게 됩니다.

 

멀티 스레딩 : 서버 C++ 소스 코드

#include <WinSock2.h>
#include <iostream>
#include <process.h>
#include <unordered_map>
 
using namespace std;
 
#pragma comment(lib, "ws2_32.lib")
 
#define PORT             7890
#define PACKET_LENGTH    1024
 
typedef struct _tagUserInfo
{
    SOCKET hSocket;
    HANDLE hThread;
}UserInfo, *PUserInfo;
 
unordered_map<SOCKET, PUserInfo>    g_mapUserInfo;
 
CRITICAL_SECTION    g_Crt;
 
class CSync
{
    CRITICAL_SECTION*    m_pCrt;
 
public:
    CSync(CRITICAL_SECTION* pCrt)    :
        m_pCrt(pCrt)
    {
        EnterCriticalSection(m_pCrt);
    }
 
    ~CSync()
    {
        LeaveCriticalSection(m_pCrt);
    }
};
 
unsigned int __stdcall RecvThread(void* pArg)
{
    PUserInfo pInfo = (PUserInfo)pArg;
 
    while (true)
    {
        char Packet[PACKET_LENGTH] = {};
 
        int iLength = recv(pInfo->hSocket, Packet, PACKET_LENGTH, 0);
 
        // 패킷 데이터를 받지 못했을 때
        if (iLength == 0 || iLength == -1)
            break;
    
        cout << pInfo->hSocket << " Recv Message from Client : " << Packet << endl;
 
        // EnterCriticalSection
        CSync sync(&g_Crt);
 
        unordered_map<SOCKET, PUserInfo>::iterator iter = g_mapUserInfo.begin();
        unordered_map<SOCKET, PUserInfo>::iterator iterEnd = g_mapUserInfo.end();
 
        for (; iter != iterEnd; ++iter)
        {
            if (iter->second == pInfo)
                continue;
 
            send(iter->second->hSocket, Packet, iLength, 0);
        }
        // LeaveCriticalSection
    }
    
    // EnterCriticalSection
    CSync    sync(&g_Crt);
 
    cout << pInfo->hSocket << " Client Disconnected" << endl;
 
    g_mapUserInfo.erase(pInfo->hSocket);
 
    // LeaveCriticalSection
    return unsigned int();
}
 
int main()
{
    // Main Thread에서 접속 승인 역할만 해준다.
    WSADATA wsaData;
    WSAStartup(MAKEWORD(2, 2), &wsaData);
 
    SOCKET hListen = socket(PF_INET, SOCK_STREAM, 0);
 
    SOCKADDR_IN tAddr = {};
    tAddr.sin_family = AF_INET;
 
    // 메모리 저장은 빅 엔디안 방식으로 변환
    tAddr.sin_port = htons(PORT);
    tAddr.sin_addr.s_addr = htonl(INADDR_ANY);
 
    bind(hListen, (SOCKADDR*)&tAddr, sizeof(tAddr));
 
    listen(hListen, SOMAXCONN);
 
    // 크리티컬 섹션 초기화
    InitializeCriticalSection(&g_Crt);
 
    // 여러 개의 클라에서 연결이 요청 된 소켓을 차례대로 accept 한다.
    while (true)
    {
        SOCKADDR_IN tAddr = {};
        int iSize = sizeof(tAddr);
        
        // 연결 되었을 때 accept 소켓을 생성하고 이 소켓의 정보를 UserInfo에 담는다.
        // 또한, 각 소켓들로부터 받을 Recv Tread의 정보도 UserInfo에 담는다.
        SOCKET hSocket = accept(hListen, (SOCKADDR*)&tAddr, &iSize);
 
        if (hSocket == SOCKET_ERROR)
        {
            closesocket(hListen);
            WSACleanup();
            return 0;
        }
 
        // 소켓이 연결되었다는 알림을 서버에서 출력한다.
        cout << hSocket << " Client Connected" << endl;
 
        PUserInfo pInfo = new UserInfo;
        pInfo->hSocket = hSocket;
 
        // 클라 소켓으로부터 받게 될 패킷을 별도의 스레드에서 처리하기 위한 스레드
        HANDLE hThread = (HANDLE)_beginthreadex(nullptr, 0, RecvThread, (void*)pInfo, 0, 0);
 
        // 연결이 종료되었을 때 스레딩이 넌 시그널 상태라면 이를 대기시키는 것과
        // 핸들을 닫기 위해 UserInfo에 받아놓는다.
        pInfo->hThread = hThread;
 
        // 클라에게 접속이 성공되었다는 패킷을 전송한다.
        // 이를 하지 않으면 클라에서 recv 처리 할 때 까지 블로킹 상태에 빠지게 됨.
        char Packet[PACKET_LENGTH] = {};
 
        strcpy_s(Packet, "접속 성공");
 
        send(hSocket, Packet, strlen(Packet), 0);
 
        // EnterCriticalSection
        CSync sync(&g_Crt);
 
        g_mapUserInfo.insert(make_pair(hSocket, pInfo));
 
        // LeaveCriticalSection
    }
    
    DeleteCriticalSection(&g_Crt);
 
    unordered_map<SOCKET, PUserInfo>::iterator iter = g_mapUserInfo.begin();
    unordered_map<SOCKET, PUserInfo>::iterator iterEnd = g_mapUserInfo.end();
 
    for (; iter != iterEnd; ++iter)
    {
        closesocket(iter->second->hSocket);
 
        // 넌 시그널 상태에서 처리하고 있는 스레드가 있을 경우 INIFITE로 대기시킨다.
        // (스레드의 실행이 완료되고 시그널 상태가 될 때 까지 기다림)
        WaitForSingleObject(iter->second->hThread, INFINITE);
 
        CloseHandle(iter->second->hThread);
 
        delete iter->second;
    }
 
    WSACleanup();
 
    return 0;
}

 

멀티 스레딩 : 클라이언트 C++ 소스 코드

#define _WINSOCK_DEPRECATED_NO_WARNINGS
 
#include <WinSock2.h>
#include <iostream>
#include <process.h>
 
using namespace std;
 
#pragma comment(lib, "ws2_32.lib")
 
#define PORT             7890
#define PACKET_LENGTH    1024
#define SERVER_IP        "127.0.0.1"
 
bool    g_bLoop = true;
 
unsigned int __stdcall RecvThread(void* pArg)
{
    SOCKET hSocket = (SOCKET)pArg;
 
    while (g_bLoop)
    {
        char Packet[PACKET_LENGTH] = {};
 
        recv(hSocket, Packet, PACKET_LENGTH, 0);
 
        cout << Packet << endl;
    }
 
    return unsigned int();
}
 
int main()
{
    WSADATA wsaData;
    WSAStartup(MAKEWORD(2, 2), &wsaData);
 
    // 세번째 인자 0은 IPPROTO_HOPOPTS의 값이다.
    // 이 값을 쓰면 자동으로 TCP 또는 UDP address family가 자동으로 잡힌다.
    SOCKET hSocket = socket(PF_INET, SOCK_STREAM, 0);
 
    SOCKADDR_IN tAddr = {};
    tAddr.sin_family = AF_INET;
    tAddr.sin_port = htons(PORT);
    tAddr.sin_addr.s_addr = inet_addr(SERVER_IP);
 
    // 서버에서 바인딩 된 소켓에 연결 요청
    connect(hSocket, (SOCKADDR*)&tAddr, sizeof(tAddr));
 
    char Packet[PACKET_LENGTH] = {};
 
    // 서버와 연결이 되었을 경우 패킷을 받는다.
    recv(hSocket, Packet, PACKET_LENGTH, 0);
 
    // 서버로 부터 패킷 데이터 메시지를 출력한다.
    cout << "Server Messag : " << Packet << endl;
 
    // 자신이 보낸 패킷 데이터 메시지를 recv해줄 스레드를 생성한다.
    // (서버에 여러 개의 클라이언트를 연결해서 자신이 출력한 메시지를
    // 다른 클라이언트에서도 패킷 메시지를 보여지게 하기 위함)
    HANDLE hThread = (HANDLE)_beginthreadex(nullptr, 0, RecvThread, (void*)hSocket, 0, 0);
 
    while (true)
    {
        cout << "Input : ";
        memset(Packet, 0, sizeof(Packet));
 
        cin >> Packet;
 
        // 입력 메시지로 q 또는 Q를 입력했을 경우 연결을 종료한다.
        if (strcmp(Packet, "q") == 0 || strcmp(Packet, "Q") == 0)
        {
            g_bLoop = false;
            break;
        }
        
        // 그게 아니라면 입력된 패킷 데이터 메시지를 전송한다.
        send(hSocket, Packet, strlen(Packet), 0);
    }
 
    closesocket(hSocket);
 
    // 다른 클라 소켓으로부터 받은 패킷 데이터 메시지의 실행 흐름을 제어한다.
    // (스레드 실행 흐름 제어), 넌 시그널 상태라면 INFINITE로 대기시킴.
    WaitForSingleObject(hThread, INFINITE);
 
    CloseHandle(hThread);
 
    WSACleanup();
 
    return 0;
}

 

실행 결과

서버 / 클라 연결 후 패킷 입출력

클라이언트 연결 종료 후