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

지난 글에서 운영체제별 비동기IO 처리 메커니즘에 대해 알아봤습니다. 이제 본격적으로 클라이언트 연결을 수립하고 통신을 하는 Server 객체를 만들어 보도록 하겠습니다. 😎

⚠️ Warning
아래 코드는 오래된 코드일 수 있습니다. 이곳에서 최신 코드를 확인할 수 있습니다.

🌳 구조

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
.
├── include/
│   └── yami/
│       └── comm/
│           ├── server.hpp
│           └── socket.hpp
└── src/
    └── comm/
        ├── server.cpp
        ├── server_linux.cpp
        ├── server_windows.cpp
        ├── socket.cpp
        ├── socket_linux.cpp
        └── socket_windows.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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
// server.hpp

#ifndef __YAMI_COMM_SERVER_HPP__
#define __YAMI_COMM_SERVER_HPP__

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

#include <cstdint>
#include <unordered_map>

namespace Yami {
namespace Comm {
    
    class Server {
    public:
        Server(Utility::EventLoop&);
        ~Server();

        // 사용할 소켓의 형식 세팅
        Server& setDomain(YAMI_SOCKET_DOMAIN);
        Server& setType(YAMI_SOCKET_TYPE);
        Server& setPort(std::uint16_t);
        
        // 이벤트 발생시 실행할 콜백
        Yami::Utility::Callback<const std::string&> onError;
        Yami::Utility::Callback<Socket&> onAccept;
        Yami::Utility::Callback<Socket&, const std::size_t> onRecv;
        
        // 내부 소켓 초기화/해제
        bool init();
        void destroy();
        
        // 바인딩 된 주소로 listen
        bool listen();

    private:
        YAMI_SOCKET_DOMAIN _domain;
        YAMI_SOCKET_TYPE _type;
        std::uint16_t _port;

        Utility::EventLoop& _ev;
        Socket _listenSock;
        std::unordered_map<Socket_t, Socket> _acceptedSock;
        
        // 이벤트 핸들러
        void recvHandler(const Yami::Utility::Event&);
        void socketErrorHandler(Socket&);
    };

} // namespace Comm
} // namespace Yami
#endif //__YAMI_COMM_SERVER_HPP__

Server 객체의 멤버 변수에, 지난번에 만들었던 Socket과 EventLoop를 사용하는 모습입니다.

연결된 클라이언트들을 컨테이너에 담기 위해 _acceptedSock이라는 맵도 추가했습니다. 나중에 브로드캐스팅을 위해 연결된 소켓들을 한곳에 담는 컨테이너는 있어야 할 것 같았습니다.

💡 Info
현재 구현된 코드는 IPv4 주소 체계를 사용하는 TCP입니다. 나중에 UDP, RAW, 다른 주소 체계도 추가해 볼 생각입니다.

소스

 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
// server.cpp

#include "yami/comm/server.hpp"

namespace Yami {
namespace Comm {
    
    Server::Server(Utility::EventLoop& ev) : _ev(ev) {
        _ev.onEvent += std::bind(&Server::recvHandler, this, std::placeholders::_1);
    }
    
    Server::~Server() {
        _ev.stop();
        destroy();
    }

    Server& Server::setDomain(YAMI_SOCKET_DOMAIN domain) {
        _domain = domain;
        return *this;
    }

    Server& Server::setType(YAMI_SOCKET_TYPE type) {
        _type = type;
        return *this;
    }

    Server& Server::setPort(std::uint16_t port) {
        _port = port;
        return *this;
    }
    
    bool Server::init() {
        if(_listenSock.valid()) {
            return false;
        } else {
            bool ret = false;

            ret = _listenSock.init(_domain, _type);
            if(!ret) return false; 
            
            ret = _listenSock.setNonblocking();
            if(!ret) {
                _listenSock.destroy();
                return false;
            }
 
        }

        return true;
    }
    
} // namespace Comm
} // namespace Yami

Socket과 EventLoop를 크로스 플랫폼으로 구성했더니 초기화하는 부분은 공통으로 사용할 수 있었습니다.

리눅스

  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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
//server_linux.cpp

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

#include <cerrno>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#include <netinet/in.h>

namespace Yami {
namespace Comm {
    
    void Server::destroy() {
        if(_listenSock.valid()) {
            shutdown(_listenSock, SHUT_RDWR);
            _ev.unwatch(_listenSock);
            _listenSock.destroy();
        }
    }

    bool Server::listen() {
        if(!_listenSock.valid()) {
            return false;
        }

        if(_domain == YAMI_SOCKET_DOMAIN::IPv4) {
            sockaddr_in address{};
            address.sin_family = static_cast<int>(_domain);
            address.sin_addr.s_addr = INADDR_ANY;
            address.sin_port = htons(_port);
            
            if(_type == YAMI_SOCKET_TYPE::TCP) {
                // prevent "address aleady in use"
                int opt = 1;
                if(::setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) == YAMI_COMM_ERROR) {
                    return false;
                }
                
                if(::bind(_listenSock, (struct sockaddr*)&address, sizeof(address)) == YAMI_COMM_SOCKET_ERROR) {
                    destroy();
                    return false;
                }
                
                if(::listen(_listenSock, SOMAXCONN) == YAMI_COMM_SOCKET_ERROR) {
                    destroy();
                    return false;
                }
                
                _ev.watch(_listenSock);
            }
        }

        return true;
    }
    
    void Server::recvHandler(const Yami::Utility::Event& event) {
        if(event.type != Yami::Utility::EventType::Read) {
            return;
        }
        
        const Handle_t handle = event.handle;
        if(handle == _listenSock) {
            // accept
            sockaddr_in addr{};
            socklen_t addrLen = sizeof(addr);

            while(true) {
                Socket_t acceptedSock = accept(
                    _listenSock,
                    reinterpret_cast<struct sockaddr*>(&addr),
                    &addrLen
                );
                
                if(acceptedSock != YAMI_COMM_INVALID_SOCKET) {
                    _ev.watch(acceptedSock);
                    Socket socket(acceptedSock);

                    auto it = _acceptedSock.insert({acceptedSock, std::move(socket)});
                    if(it.second) {
                        Socket& sock = it.first->second;
                        sock.setNonblocking();
                        sock.onError += std::bind(&Server::socketErrorHandler, this, std::placeholders::_1);
                        onAccept(it.first->second);
                    } else {
                        // TODO: 소켓 중복 처리
                    }
                    
                    continue;
                }

                // EPOLLET 분기 처리
                if(errno == EINTR) {
                    continue;
                } else if(errno == EAGAIN || errno == EWOULDBLOCK) {
                    break;
                } else {
                    // error
                    destroy();
                    break;
                }
            }
        } else {
            // recv
            Socket& sock = _acceptedSock[handle];
            while(true) {
                ssize_t bytes = read(handle, sock.buffer().data(), sock.buffer().remain());

                if(bytes > 0) {
                    // receive
                    sock.buffer().increase(bytes);
                    onRecv(sock, bytes);
                }
                else if(bytes == 0) {
                    // buffer space lack
                    if(sock.buffer().remain() == 0) {
                        sock.buffer().resize(sock.buffer().capacity()+1);
                        continue;
                    }

                    // client closed
                    socketErrorHandler(sock);
                    break;
                } else {
                    if(errno == EAGAIN || errno == EWOULDBLOCK) {
                        break;
                    } else if(errno == EINTR) {
                        continue;
                    } else {
                        // error
                        socketErrorHandler(sock);
                        break;
                    }
                }
            }

            
        }
    }
    
    void Server::socketErrorHandler(Socket& sock) {
        shutdown(sock, SHUT_RDWR);
        _ev.unwatch(sock);
        close(sock);
        _acceptedSock.erase(sock);
    }

} // namespace Comm
} // namespace Yami

윈도우

  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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
// server_windows.cpp

#include "yami/comm/server.hpp"
#include "yami/comm/socket.hpp"
#include "yami/utility/event_loop.hpp"
#include <cstring>

namespace Yami {
namespace Comm {
    BOOL post_accept(Socket&);
    BOOL post_recv(Socket&);
    
    struct AcceptContext {
        OVERLAPPED ov;
        Yami::Utility::EventType type;
        std::size_t length;
        Socket acceptSocket;
        char buffer[2 * (sizeof(SOCKADDR_STORAGE) + 16)];
    };

    struct ReceiveContext {
        OVERLAPPED ov;
        Yami::Utility::EventType type;
        std::size_t length;
        WSABUF wsabuf;
    };

    void Server::destroy() {
        if(_listenSock.valid()) {
            shutdown(_listenSock, SD_BOTH);
            _listenSock.destroy();
        }
    }

    bool Server::listen() {
        if(!_listenSock.valid()) {
            return false;
        }

        if(_domain == YAMI_SOCKET_DOMAIN::IPv4) {
            sockaddr_in addr{};
            addr.sin_family = AF_INET;
            addr.sin_addr.s_addr = htonl(INADDR_ANY);
            addr.sin_port = htons(_port);
            
            if(_type == YAMI_SOCKET_TYPE::TCP) {
                // prevent "address aleady in use"
                int opt = 1;
                if(::setsockopt(_listenSock, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<const char*>(&opt), sizeof(opt)) == YAMI_COMM_SOCKET_ERROR) {
                    return false;
                }

                if(::bind(_listenSock, reinterpret_cast<const sockaddr*>(&addr), sizeof(addr)) == YAMI_COMM_SOCKET_ERROR) {
                    destroy();
                    return false;
                }
                
                if(::listen(_listenSock, SOMAXCONN) == YAMI_COMM_SOCKET_ERROR) {
                    destroy();
                    return false;
                }
                
                _ev.watch(reinterpret_cast<Handle_t>(static_cast<Socket_t>(_listenSock)));

                BOOL ok = post_accept(_listenSock);
                if(!ok) {
                    const int err = WSAGetLastError();
                    if(err != ERROR_IO_PENDING) {
                        return false;
                    }
                }
            }
        }

        return true;
    }
    
    void Server::recvHandler(const Yami::Utility::Event &ev) {
        // IOCP는 userData(ov)에 명시됨.
        if(ev.type == Yami::Utility::EventType::Read) {
            Socket_t sock = reinterpret_cast<Socket_t>(ev.handle);
            if(sock == _listenSock) {
                // Accept
                AcceptContext* ctx = reinterpret_cast<AcceptContext*>(ev.userData);

                if(setsockopt(
                    ctx->acceptSocket,
                    SOL_SOCKET,
                    SO_UPDATE_ACCEPT_CONTEXT,
                    reinterpret_cast<const char*>(&sock),
                    sizeof(sock)
                ) != YAMI_COMM_SOCKET_ERROR) {
                    Socket acceptSock(std::move(ctx->acceptSocket)); 

                    auto it = _acceptedSock.emplace(acceptSock, std::move(acceptSock));
                    if(it.second) {
                        acceptSock.onError += std::bind(&Server::socketErrorHandler, this, std::placeholders::_1);               
                        _ev.watch(reinterpret_cast<Handle_t>(it.first->first));
                        post_recv(it.first->second);
                        onAccept(it.first->second);
                    } else {
                        // TODO: 중복 처리
                    }
                }
                
                delete ctx;
                post_accept(_listenSock);
            } else {
                // Receive
                Socket& socket = _acceptedSock[sock];
                ReceiveContext* ctx = reinterpret_cast<ReceiveContext*>(ev.userData);
                DS::RingBuffer& buffer = socket.buffer();

                if(ctx->length > 0) {
                    buffer.increase(ctx->length);
                    post_recv(socket);
                    onRecv(socket, ctx->length); 
                } else if(ctx->length == 0) {
                    if(buffer.remain() == 0) {
                        buffer.resize(buffer.capacity() + 1);
                        post_recv(socket);
                    } else {
                        socketErrorHandler(socket);
                    }
                } else {
                    const int err = WSAGetLastError();
                    if(err == WSAEWOULDBLOCK) {
                        // 재시도...
                    } else {
                        socketErrorHandler(socket);
                    }
                }
                
                delete ctx;
            }
        }
    }
    
    void Server::socketErrorHandler(Socket& sock) {
        shutdown(sock, SD_BOTH);
        _ev.unwatch(reinterpret_cast<Handle_t>(static_cast<Socket_t>(sock)));
        _acceptedSock.erase(sock);
    }
    
    BOOL post_accept(Socket& listenSock) {
        static LPFN_ACCEPTEX acceptEx = nullptr;
        
        if(!acceptEx) {
            GUID guid = WSAID_ACCEPTEX;
            DWORD returned = 0;

            if(WSAIoctl(
                listenSock,
                SIO_GET_EXTENSION_FUNCTION_POINTER,
                &guid, sizeof(guid),
                &acceptEx, sizeof(acceptEx),
                &returned,
                nullptr, nullptr
            ) == YAMI_COMM_SOCKET_ERROR) {
                return FALSE;
            }
        }
        
        AcceptContext* ctx = new AcceptContext;
        std::memset(&ctx->ov, 0, sizeof(OVERLAPPED));
        ctx->type = Yami::Utility::EventType::Read;
        ctx->acceptSocket.init(listenSock.getDomain(), listenSock.getType());
        ctx->acceptSocket.setNonblocking();
        
        if(!ctx->acceptSocket.valid()) {
            return FALSE;
        }

        DWORD bytes = 0;
        return acceptEx(
            listenSock,
            ctx->acceptSocket,
            ctx->buffer,
            0,
            sizeof(SOCKADDR_STORAGE) + 16,
            sizeof(SOCKADDR_STORAGE) + 16,
            &bytes,
            &ctx->ov
        );
    }
    
    BOOL post_recv(Socket& acceptSock) {
        ReceiveContext* ctx = new ReceiveContext;
        std::memset(&ctx->ov, 0, sizeof(OVERLAPPED));
        ctx->type = Utility::EventType::Read;
        ctx->wsabuf.buf = acceptSock.buffer().data();
        ctx->wsabuf.len = acceptSock.buffer().remain();
        
        DWORD flags = 0;
        DWORD bytesReceived = 0;

        int result = WSARecv(
            acceptSock,
            &ctx->wsabuf,
            1,
            &bytesReceived,
            &flags,
            &ctx->ov,
            nullptr
        );
        
        if(result == YAMI_COMM_SOCKET_ERROR) {
            const int err = WSAGetLastError();
            if(err != WSA_IO_PENDING) {
                acceptSock.onError(acceptSock);
                return false;
            }
        }

        return TRUE;
    }

} // namespace Comm
} // namespace Yami

에코 테스트

 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
// test_socket.cpp

TEST_CASE(SERVER_ECHO_TEST) {
    bool destroy = false;

    // EventLoop 준비
    Yami::Utility::EventLoop ev;
    ev.init();
    // Server 준비
    Yami::Comm::Server server(ev);
    // 사용할 소켓 세팅
    server
        .setDomain(Yami::Comm::YAMI_SOCKET_DOMAIN::IPv4)
        .setType(Yami::Comm::YAMI_SOCKET_TYPE::TCP)
        .setPort(1234)
        .init();
    
    // 이벤트 핸들러 세팅
    server.onError += [](const std::string& msg) {
        std::cout << msg << "\n";
    };
    
    server.onAccept += [](Yami::Comm::Socket& sock) {
        std::string tmp = "hello, world!\n";
        sock.send(tmp.data(), tmp.size());
    };
    
    server.onRecv += [](Yami::Comm::Socket& sock, const std::size_t size) {
        std::cout << "onRecv socket: " << sock << "\n";
        std::size_t s = sock.buffer().used();
        std::string tmp(sock.buffer().read(s), s);
        std::cout << tmp;
        
        sock.send(tmp.data(), s);
    };

    // 구동
    ASSERT_TRUE(server.listen());
    ASSERT_TRUE(ev.run());

    // 종료를 막기 위한 입력대기 (엔터 입력 종료)
    getchar();

    // 종료
    ev.stop();
    server.destroy();
}

echo-test.wepb

에코 테스트가 잘 되는 모습입니다! 👍

마무리

우선 EventLoop를 직접 사용해 보니 막상 이벤트가 발생해서 데이터를 처리하는 방법 자체가 달라 운영체제에 종속되어 프로그래밍할 수밖에 없었습니다. recvHandler()에 사용된 epoll이나 IOCP 관련 내용들을 모두 EventLoop에 넣을 수 있으면 좋을 것 같습니다. 아마 생각을 조금 더 해봐야 할 것 같습니다.

아무튼 Server 객체 자체는 리눅스에서든, 윈도우에서든 모두 빌드가 되고 작동합니다!

다음 타겟은 정해지지 않았지만 클라이언트도 만들어볼 수 있겠고, HTTP 서버도 만들어볼 수 있을 것 같습니다. 타겟을 생각해 보고 다음 포스팅에서 볼 수 있도록 하겠습니다.

🌳 구조#

🎯 구현#

헤더#

소스#

리눅스#

윈도우#

에코 테스트#

마무리#

🌳 구조

🎯 구현

헤더

소스

리눅스

윈도우

에코 테스트

마무리