C++ 소켓 크로스 플랫폼 개발 - 2

지난 글에서 리눅스와 윈도우에서 동시에 같이 사용할 수 있는 소켓 클래스를 만들어 봤습니다. 이번 글에서는 소켓에 데이터가 들어오는 것을 감지하고 처리할 수 있는 EventLoop 객체를 만들어 보겠습니다. 👍

EventLoop 객체에 사용되는 비동기 IO 처리 역시 운영체제별로 다르므로 각 운영체제에 맞게 구현해야 하며 사용되는 메커니즘은 아래와 같습니다.

🤖 I/O 이벤트 처리 방식

구분	Linux	Windows
모델	epoll	IOCP
방식	I/O 준비가 되면 알림	비동기 작업이 완료되면 알림
의미	읽기/쓰기를 진행할 수 있음	제출한 읽기/쓰기 작업이 끝남
처리	직접 `recv()`/`send()` 등을 호출하여 처리 시작	완료된 결과를 확인한 후 후속 작업 처리
API	`epoll_create1()`, `epoll_ctl()`, `epoll_wait()`	`CreateIoCompletionPort()`, `GetQueuedCompletionStatus()`

간단하게 리눅스에서는 알림이 오면 소켓 버퍼에서 직접 데이터를 가져오면 되고 윈도우에서는 미리 “데이터를 이 버퍼에 받아주세요.” 하고 예약을 걸어두는 것입니다.

🌳 구성

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
.
├── include/
│   └── yami/
│       └── utility/
│           └── event_loop.hpp      # 인터페이스
└── src/
    └── utility/
        ├── event_loop.cpp          # 공통 구현부
        ├── event_loop_linux.cpp    # 리눅스 구현부
        └── event_loop_windows.cpp  # 윈도우 구현부

소켓과 비슷한 방법으로 나누었습니다.

🎯 구현

⚠️ Warning
아래 코드는 최신 코드가 아닐 수 있습니다. 이곳에서 최신의 전체 코드를 확인할 수 있습니다.

헤더

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
// event_loop.hpp

#ifndef __YAMI_UTILITY_EVENTLOOP_HPP__
#define __YAMI_UTILITY_EVENTLOOP_HPP__

#if defined(_WIN32)
#include <windows.h>
typedef HANDLE Handle_t;
typedef ULONG_PTR UserData;
#define YAMI_EVENT_LOOP_ERROR NULL
#define YAMI_EVENT_LOOP_INVALID_HANDLE INVALID_HANDLE_VALUE

#elif defined(__linux__)
#include <sys/epoll.h>
typedef int Handle_t;
typedef void* UserData;
#define YAMI_EVENT_LOOP_ERROR (-1)
#define YAMI_EVENT_LOOP_INVALID_HANDLE (-1)

#endif

#include "yami/utility/callback.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>

namespace Yami {
namespace Utility {
    
    enum class EventType {
        Unknown = 0,
        Read,
        Write,
        Error
    };

    struct Event {
        EventType type;
        Handle_t handle;
        void* userData;
        std::size_t bytesTransferred;
    };

    class EventLoop {
    public:
        EventLoop();
        ~EventLoop();
        
        // prevent copy
        EventLoop(const EventLoop&) = delete;
        EventLoop& operator=(const EventLoop&) = delete;

        // allow move
        EventLoop(EventLoop&&);
        EventLoop& operator=(EventLoop&&);

        bool valid();
        
        bool init();
        void destroy();
        
        bool watch(Handle_t);
        bool unwatch(Handle_t);
        
        bool run();
        void stop();
        bool isRunning();
        
        Utility::Callback<const Event&> onEvent;
        Utility::Callback<> onError;

    private:
        Handle_t _handle;
        Handle_t _signal;
        
        std::mutex _mutex;
        std::condition_variable _cv;
        std::thread _worker;
        bool _running;

        int worker();
    };
}
}

#endif //__YAMI_UTILITY_EVENTLOOP_HPP__

epoll 또는 IOCP를 다루기 위한 객체를 Handle_t로 데이터 타입을 맞췄고 이벤트 감시를 위한 worker()와 스레드를 생성할 때 순서를 지키기 위한 mutex와 condition_variable도 추가된 모습입니다.

_signal은 객체가 소멸할 때 worker()의 종료를 알리는 변수(eventfd)로 리눅스에서만 쓰이고 나중에 .cpp로 옮겨야 할 것 같습니다. 😵

소스

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
// event_loop.cpp

#include "yami/utility/event_loop.hpp"

namespace Yami {
namespace Utility {

    EventLoop::EventLoop() 
        : _handle(YAMI_EVENT_LOOP_INVALID_HANDLE)
        , _running(false) {}

    EventLoop::~EventLoop() {
        destroy();
    }
    
    bool EventLoop::valid() {
        return _handle != YAMI_EVENT_LOOP_INVALID_HANDLE && _signal != YAMI_EVENT_LOOP_INVALID_HANDLE;
    }
    
    bool EventLoop::run() {
        if(!(valid() && !isRunning())) return false;

        _running = false;
        _worker = std::thread(&EventLoop::worker, this);
        std::unique_lock<std::mutex> lock(_mutex);
        _cv.wait(lock, [this]() { return _running; });
        
        return true;
    }


    bool EventLoop::isRunning() {
        std::lock_guard<std::mutex> lock(_mutex);
        return _running;
    }
    
}
}

공통 구현 부분은 멤버 변수들의 초기화 및 상태 값 반환하는 메서드들이 있습니다.

리눅스

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
#include "yami/utility/event_loop.hpp"
#include <cerrno>
#include <fcntl.h>
#include <mutex>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <unistd.h>

#define EPOLL_MAX_EVENT 1024

namespace Yami {
namespace Utility {
    
    bool EventLoop::init() {
        // aleady initialized
        if(valid()) return false;

        _handle = epoll_create1(0);
        if(_handle == YAMI_EVENT_LOOP_INVALID_HANDLE) return false;

        _signal = eventfd(0, 0);
        if(_signal == YAMI_EVENT_LOOP_INVALID_HANDLE) {
            close(_handle);
            _handle = YAMI_EVENT_LOOP_INVALID_HANDLE;
            return false;
        }

        watch(_signal);
        
        return valid();
    }
    
    void EventLoop::destroy() {
        unwatch(_signal);
        close(_signal);
        _signal = YAMI_EVENT_LOOP_INVALID_HANDLE;

        close(_handle);
        _handle = YAMI_EVENT_LOOP_INVALID_HANDLE;
    }

    bool EventLoop::watch(Handle_t fd) {
        int flags = fcntl(fd, F_GETFL);
        if(flags == YAMI_EVENT_LOOP_ERROR) {
            return false;
        }
        
        if(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == YAMI_EVENT_LOOP_ERROR) {
            return false;
        }

        epoll_event ev{};
        
        // TODO: 이벤트가 달라질 수 있다면 매개변수로 받아야 함
        ev.events = EPOLLIN | EPOLLET;
        ev.data.fd = fd;

        return epoll_ctl(_handle, EPOLL_CTL_ADD, fd, &ev) != YAMI_EVENT_LOOP_ERROR;
    }
    
    bool EventLoop::unwatch(Handle_t fd) {
        return epoll_ctl(_handle, EPOLL_CTL_DEL, fd, nullptr) != YAMI_EVENT_LOOP_ERROR;        
    }
    
    void EventLoop::stop() {
        uint64_t u = 1;
        write(_signal, &u, sizeof(u));
        
        if(_worker.joinable()) {
            _worker.join();
        }
    }

    int EventLoop::worker() {
        bool terminated = false;
        std::vector<epoll_event> events(EPOLL_MAX_EVENT);
        int i, eventCount = -1;
        Event event{};

        {
            std::lock_guard<std::mutex> lock(_mutex);
            _running = true;
        }
        _cv.notify_one();
        while(!terminated) {
            eventCount = epoll_wait(_handle, events.data(), EPOLL_MAX_EVENT, -1);
            if(eventCount < 0) {
                if(errno != EINTR) {
                    onError();
                    terminated = true;
                    break;
                }
            }

            for(i=0; i<eventCount; i++) {
                epoll_event& ev = events[i];
                
                if(ev.events & EPOLLIN) {
                    if(ev.data.fd == _signal) {
                        terminated = true;
                        break;
                    }
                    
                    // TODO: 병목 발생 시 Executor 추가하기 
                    event.type = EventType::Read;
                }                
                
                if(ev.events & EPOLLOUT) {
                    event.type = EventType::Write;
                }
                event.handle = ev.data.fd;
                
                onEvent(event);
            }
        }

        std::lock_guard<std::mutex> lock(_mutex);
        _running = false;
        return 0;
    }

} // namespace Utility
} // namespace Yami

epoll_create1()을 이용하여 epoll 객체를 생성하고 epoll_ctl()로 생성된 epoll 객체에 핸들(소켓)을 등록(watch)합니다.

💡 Info
epoll 이벤트 플래그
EPOLLIN - 읽을 수 있음
해당 파일 디스크립터에서 읽기 작업을 진행할 수 있다는 알림입니다. TCP 소켓에서는 주로 다음 상황에 해당합니다.
수신 데이터가 도착함
상대가 연결을 정상 종료해 read()/recv()가 0을 반환할 수 있음
리스닝 소켓에 새 연결이 대기 중임
EPOLLOUT - 쓸 수 있음
해당 파일 디스크립터에 쓰기 작업을 진행할 수 있다는 알림입니다. TCP 소켓에서는 송신 버퍼에 공간이 생겨 데이터를 보낼 수 있을 때 주로 발생합니다.
주의할 점은 대부분의 연결된 소켓은 대체로 쓸 수 있는 상태라는 것입니다. 따라서 처음부터 EPOLLOUT을 계속 등록하면 이벤트가 반복적으로 발생해 바쁜 루프를 만들 수 있습니다.
EPOLLET - 엣지 트리거
EPOLLET은 엣지 트리거(Edge-Triggered, ET) 방식으로 이벤트를 받겠다는 뜻입니다. 준비 상태가 계속 유지되는 동안 매번 알리는 것이 아니라, 준비 상태가 바뀌는 시점에 주로 알립니다.

해당 프로젝트에선 엣지 트리거 모드로 구현 되었습니다. errno == EAGAIN 또는 error == EWOULDBLOCK로 반환될 때까지 읽어야 합니다.

윈도우

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
// event_loop_windows.cpp

#include "yami/comm/socket.hpp"
#include "yami/utility/event_loop.hpp"

namespace Yami {
namespace Utility {

    constexpr static ULONG_PTR kStopKey = 1;
    struct OverlappedEx {
        OVERLAPPED ov;
        EventType type;
        std::size_t length;
    };

    EventLoop::EventLoop(EventLoop&& other) 
        : _handle(std::exchange(other._handle, YAMI_EVENT_LOOP_INVALID_HANDLE))
        , _running(std::exchange(other._running, false)) {}

    EventLoop& EventLoop::operator=(EventLoop&& other) {
        destroy();

        _handle = std::exchange(other._handle, YAMI_EVENT_LOOP_INVALID_HANDLE);
        _running = std::exchange(other._running, false);

        return *this;
    }
    
    bool EventLoop::init() {
        if(valid()) return false;

        _handle = CreateIoCompletionPort(
            YAMI_EVENT_LOOP_INVALID_HANDLE,
            nullptr,
            0,
            0
        );

        return valid();
    }

    void EventLoop::destroy() {
        CloseHandle(_handle);
        _handle = YAMI_EVENT_LOOP_INVALID_HANDLE;
    }
    
    bool EventLoop::watch(Handle_t handle) {
        Handle_t result = CreateIoCompletionPort(
            handle, 
            _handle, 
            reinterpret_cast<ULONG_PTR>(handle), // 완료(콜백시점)에서 사용할 User data
            0);

        return result != YAMI_EVENT_LOOP_INVALID_HANDLE;
    }
    
    bool EventLoop::unwatch(Handle_t handle) { 
        // Windows 해당 함수가 없음
        return true;
    }

    void EventLoop::stop() {
        PostQueuedCompletionStatus(
            _handle,
            0,
            kStopKey,
            nullptr
        );

        if(_worker.joinable()) {
            _worker.join();
        }
    }

    int EventLoop::worker() {
        bool terminated = false;
        DWORD bytes = 0;
        ULONG_PTR key = 0;
        OVERLAPPED* overlapped = nullptr;
        BOOL ok = FALSE;
        Event ev{};

        {
            std::lock_guard<std::mutex> lock(_mutex);
            _running = true;
        }
        _cv.notify_one();
        while(!terminated) {
            ok = GetQueuedCompletionStatus(
                _handle, 
                &bytes, 
                &key,
                &overlapped,
                INFINITE
            );

            if(key == kStopKey && overlapped == nullptr) {
                terminated = true;
                break;
            }

            if(overlapped == nullptr) {
                onError();
                break;
            }
            
            if(ok) {
                OverlappedEx* ovEx = reinterpret_cast<OverlappedEx*>(overlapped);
                ovEx->length = bytes;
                
                ev.type = ovEx->type;
                ev.handle = reinterpret_cast<Handle_t>(key);
                ev.userData = reinterpret_cast<void*>(overlapped);

                onEvent(ev);
            } else {
                onError();
            }
        }

        std::lock_guard<std::mutex> lock(_mutex);
        _running = false;
        return 0;
    }
} // namespace Utility
} // namespace Yami

CreateIoCompletionPort()를 이용하여 IOCP 객체를 만들고 소켓을 등록합니다. worker()에서는 GetQueuedCompletionStatus()을 이용하여 작업이 완료된 내용을 꺼내서 처리합니다.

💡 Info
윈도우에서 살펴봐야 할 내용은 OVERLAPPED객체 입니다. 예약을 걸 때 반드시 전달해야 하는 객체로, 나중에 완료 통지를 받을 때 같이 받아볼 수 있습니다. UserData 느낌입니다.
이 객체는 그대로 사용하는 것이 아니라 확장해서 사용하는데, 완료 통지를 받을 때 필요한 데이터를 예약할 때 미리 세팅해 놓고 예약을 걸어둡니다. 완료 통지가 오면 이벤트 타입이나 들어온 데이터의 크기, 연결된 소켓 등 여러 가지 정보를 받아볼 수 있습니다.

📝 마무리

운영체제별 비동기IO 처리 메커니즘을 공부해 볼 수 있었던 계기가 됐습니다.

OVERLAPPED를 확장해서 사용하는 방식이 처음에는 조금 이해가 되지 않았는데 reinterpret_cast를 통해 사용자가 유연하게 구조를 만들 수 있다는 점을 새롭게 알게 되었습니다. 마치 클래스의 상속과 비슷하게 사용할 수 있어서 좋았습니다.

다음 글은 이제 Socket과 EventLoop를 가지고 실제로 통신할 수 있는 Server를 만들어 보도록 하겠습니다. 😎

🤖 I/O 이벤트 처리 방식#

🌳 구성#

🎯 구현#

헤더#

소스#

리눅스#

윈도우#

📝 마무리#

🤖 I/O 이벤트 처리 방식

🌳 구성

🎯 구현

헤더

소스

리눅스

윈도우

📝 마무리