Win32 socket이 Thread-safe할까?
지난번에는 닷넷의 System.Net.Sockets.Socket 타입에 대한 thread-safe 이야기를 했었는데요.
System.Net.Sockets.Socket이 Thread-safe할까?
; https://www.sysnet.pe.kr/2/0/1469
그럼, 이번에는 닷넷의 하위로 내려가서 윈도우 운영체제의 socket에 대한 thread-safe 문제를 살펴보겠습니다. 검색을 좀 해보면 이에 대해서 많은 질문/답변이 있는 것을 확인할 수 있는데, 그만큼 의견도 다양합니다. ^^ 우선, thread-safe하지 않다는 몇몇 글을 볼까요?
Is Winsock thread-safe?
; http://tangentsoft.net/wskfaq/intermediate.html#threadsafety
C socket API is thread safe?
; http://stackoverflow.com/questions/2354417/c-socket-api-is-thread-safe
첫 번째 글에서는 send/receive에 대한 동시 호출은 안전하지만 send/send는 그렇지 않다고 합니다. 두 번째 글의 덧글에는 "Sending data via a socket is not a atomic transaction - any non-atomic transaction will require a lock/synchronisation. This is independent of the platform."라고 해서 역시 thread-safe하지 않다고 합니다.
반면에 다음의 글에서는 의견이 다릅니다.
Are parallel calls to send/recv on the same socket valid?
; http://stackoverflow.com/questions/1981372/are-parallel-calls-to-send-recv-on-the-same-socket-valid
위의 덧글에는 다음과 같은 설명을 포함합니다.
1) POSIX defines send/recv as atomic operations
2) The socket descriptor belongs to the process, not to a particular thread. Hence, it is possible to send/receive to/from the same socket in different threads, the OS will handle the synchronization.
애석하게도 이건 그들의 의견일 뿐 명백하게 문서화된 내용이 없다는 것이 문제입니다. 저도 MSDN 문서에서 socket 관련한 내용을 뒤져보았지만 마이크로소프트는 이에 대해 thread safe/not-safe에 대한 어떠한 명시도 하지 않고 있습니다.
따라서, 이 글의 결론은 마이크로소프트에 의해 공식적으로 확인된 것은 아니고 제 개인적인 의견을 담고 있다는 것을 미리 ^^ 밝혀두는 바입니다.
이거저거 조사해 보면 일단 제 의견은 Win32에서 제공되는 socket.send가 thread-safe 하다는 쪽에 무게를 두고 있습니다. 왜냐고요? ^^
우선,
지난번 글에 쓴 것처럼 Microsoft는 Win32 Socket에 대한 thread-safe는 명시하지 않았지만 .NET Framework의 Socket에 대한 thread-safe은 명시를 했습니다. 이게 어떤 의미를 갖냐면... .NET도 결국 내부적으로는 Win32 Socket의 send를 호출하기 때문에 간접적인 증거로 작용할 수 있습니다. 실제로 System.Net.Sockets.Socket.Send 메소드를 .NET Reflector로 보면 다음과 같이 어떠한 내부적인 잠금 없이 곧바로 Win32 send를 호출하는 것을 확인할 수 있습니다.
public unsafe int Send(byte[] buffer, int offset, int size, SocketFlags socketFlags, out SocketError errorCode)
{
// ...[생략]...
if (buffer.Length == 0)
{
num = UnsafeNclNativeMethods.OSSOCK.send(this.m_Handle.DangerousGetHandle(), null, 0, socketFlags);
}
else
{
fixed (byte* numRef = buffer)
{
num = UnsafeNclNativeMethods.OSSOCK.send(this.m_Handle.DangerousGetHandle(), numRef + offset, size, socketFlags);
}
}
// ...[생략]...
return num;
}
[DllImport("ws2_32.dll", SetLastError=true)]
internal static extern unsafe int send([In] IntPtr socketHandle, [In] byte* pinnedBuffer, [In] int len, [In] SocketFlags socketFlags);
따라서, Win32 Socket 역시 thread-safe 하다는 결론이 나옵니다.
또 다른 간접적인 증거가 하나 있다면 WSASend에 설명된 MSDN의 문서입니다.
WSASend function
; https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-wsasend
If you are using I/O completion ports, be aware that the order of calls made to WSASend is also the order in which the buffers are populated. WSASend should not be called on the same socket simultaneously from different threads, because it can result in an unpredictable buffer order.
위의 글에서는 WSASend가 "I/O completion ports"를 사용하는 상황에서는 버퍼 관리 때문에 thread-safe하지 않다고 나옵니다. 만약, 소켓 자체가 thread-safe하지 않았으면 애당초 이런 글이 나왔을리 없으므로, 조심스럽게 소켓이 thread-safe하지 않을까 하는 결론이 나옵니다.
비록 문서상으로 명확하게 밝혀진 것은 아니지만, 이에 대해
지난번 글에서 했던 것과 동일한 테스트를 통해 검증해 보는 것은 어떨까요? ^^
우선, C/C++ 클라이언트 프로그램을 단일 스레드 예제로 닷넷 소켓 서버와 호환되게 맞춰서 만들어 보았습니다.
#include "stdafx.h"
#include <Windows.h>
#include <WinSock2.h>
#include "CommonPacket.h"
#pragma comment(lib, "Ws2_32.lib")
#include <string>
using namespace std;
bool MustSendBuffer(SOCKET socket, byte *dataBuf, int mustSend);
int _tmain(int argc, _TCHAR* argv[])
{
::Sleep(2000);
string body = "";
string chunk = "";
for (int i = 0; i < 10; i ++)
{
chunk += to_string(i);
}
int loopCount = 10000;
for (int i = 0; i < loopCount; i ++)
{
body += chunk;
}
/*
UTF-8 CPP
; http://sourceforge.net/projects/utfcpp/
*/
// 문자열을 utf-8 인코딩 시키고
vector<unsigned char> dataBuf;
utf8::utf16to8(body.begin(), body.end(), back_inserter(dataBuf));
WSADATA wsaData;
int result = WSAStartup(MAKEWORD(2, 2), &wsaData);
if (result != NO_ERROR)
{
wprintf(L"WSAStartup function failed with error: %d\n", result);
return 1;
}
SOCKET socket = INVALID_SOCKET;
do
{
sockaddr_in target;
target.sin_family = AF_INET;
// target.sin_addr.s_addr = inet_addr("192.168.0.70");
target.sin_addr.s_addr = inet_addr("127.0.0.1");
target.sin_port = htons(11200);
socket = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (socket == INVALID_SOCKET)
{
wprintf(L"socket function failed with error: %ld\n", WSAGetLastError());
break;
}
result = ::connect(socket, (SOCKADDR *)&target, sizeof(target));
if (result == SOCKET_ERROR)
{
wprintf(L"connect function failed with error: %ld\n", WSAGetLastError());
break;
}
int instanceId = 0;
::send(socket, (const char *)&instanceId, sizeof(int), 0);
for (int i = 0; i < loopCount; i++)
{
CommonPacket packet(i);
packet.AddData(dataBuf);
BYTE *dataBuf = packet.GetBuffer();
int bufferSize = packet.GetBufferSize();
MustSendBuffer(socket, dataBuf, bufferSize);
}
} while (false);
if (socket != INVALID_SOCKET)
{
result = ::closesocket(socket);
if (result == SOCKET_ERROR)
{
wprintf(L"closesocket function failed with error: %ld\n", WSAGetLastError());
}
socket = INVALID_SOCKET;
printf("TCP Client socket: Closed\n");
}
WSACleanup();
return 0;
}
bool MustSendBuffer(SOCKET socket, byte *dataBuf, int mustSend)
{
int pos = 0;
while (true)
{
int sentLength = ::send(socket, (const char *)dataBuf + pos, mustSend, 0);
if (sentLength == 0)
{
return false;
}
if (sentLength == -1)
{
printf("Socket Failed: %d", ::WSAGetLastError());
return false;
}
mustSend -= sentLength;
pos += sentLength;
if (mustSend == 0)
{
return true;
}
}
}
물론 단일 스레드 예제이므로 잘 동작합니다. ^^ 그다음, 이것을 다중 스레드로 버전으로 바꿔 보았습니다.
#include "stdafx.h"
#include <Windows.h>
#include <WinSock2.h>
#include "..\SocketClientST\CommonPacket.h"
#pragma comment(lib, "Ws2_32.lib")
#include <string>
#include <thread>
#include <ppl.h>
using namespace std;
typedef struct tagThreadParam
{
int Sent;
Concurrency::critical_section *Sync;
SOCKET ClientSocket;
vector<CommonPacket *> *Packets;
} ThreadParam;
bool MustSendBuffer(SOCKET socket, byte *dataBuf, int mustSend);
void sendBufferThread(ThreadParam *threadParam)
{
while (true)
{
CommonPacket *packet = nullptr;
threadParam->Sync->lock();
{
if (threadParam->Packets->size() != 0)
{
packet = threadParam->Packets->back();
threadParam->Packets->pop_back();
threadParam->Sent ++;
}
}
threadParam->Sync->unlock();
if (packet == nullptr)
{
break;
}
BYTE *dataBuf = packet->GetBuffer();
int bufferSize = packet->GetBufferSize();
MustSendBuffer(threadParam->ClientSocket, dataBuf, bufferSize);
}
}
int _tmain(int argc, _TCHAR* argv[])
{
::Sleep(2000);
string body = "";
string chunk = "";
for (int i = 0; i < 10; i ++)
{
chunk += to_string(i);
}
int loopCount = 10000;
for (int i = 0; i < loopCount; i ++)
{
body += chunk;
}
vector<unsigned char> dataBuf;
utf8::utf16to8(body.begin(), body.end(), back_inserter(dataBuf));
WSADATA wsaData;
int result = WSAStartup(MAKEWORD(2, 2), &wsaData);
if (result != NO_ERROR)
{
wprintf(L"WSAStartup function failed with error: %d\n", result);
return 1;
}
SOCKET socket = INVALID_SOCKET;
do
{
sockaddr_in target;
target.sin_family = AF_INET;
// target.sin_addr.s_addr = inet_addr("192.168.0.70");
target.sin_addr.s_addr = inet_addr("127.0.0.1");
target.sin_port = htons(11200);
socket = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (socket == INVALID_SOCKET)
{
wprintf(L"socket function failed with error: %ld\n", WSAGetLastError());
break;
}
result = ::connect(socket, (SOCKADDR *)&target, sizeof(target));
if (result == SOCKET_ERROR)
{
wprintf(L"connect function failed with error: %ld\n", WSAGetLastError());
break;
}
int instanceId = 0;
::send(socket, (const char *)&instanceId, sizeof(int), 0);
vector<CommonPacket *> packets;
for (int i = 0; i < loopCount; i ++)
{
CommonPacket *packet = new CommonPacket(i);
packet->AddData(dataBuf);
packets.push_back(packet);
}
vector<std::thread *> threads;
ThreadParam param;
Concurrency::critical_section sync;
param.Sync = &sync;
param.Packets = &packets;
param.ClientSocket = socket;
// 20개의 스레드를 만들어서,
// vector에 담아둔 CommonPacket 내용을 socket send API를 통해 서버로 전송
for (int i = 0; i < 20; i ++)
{
thread *aThread = new thread(sendBufferThread, ¶m);
threads.push_back(aThread);
}
for (size_t i = 0; i < threads.size(); i ++)
{
threads[i]->join();
delete threads[i];
}
threads.clear();
for (size_t i = 0; i < packets.size(); i ++)
{
delete packets[i];
}
packets.clear();
} while (false);
if (socket != INVALID_SOCKET)
{
result = ::closesocket(socket);
if (result == SOCKET_ERROR)
{
wprintf(L"closesocket function failed with error: %ld\n", WSAGetLastError());
}
socket = INVALID_SOCKET;
printf("TCP Client socket: Closed\n");
}
WSACleanup();
return 0;
}
오~~~ 훌륭합니다. ^^ C/C++ 표준의 발전으로 threads가 포함되어 지난번에 작성했던 C# 예제를 거의 1:1 매핑 식으로 C/C++로 포팅하는 작업이 자연스럽게 이뤄집니다.
결과를 실행해 보면 닷넷의 Socket 때와 마찬가지로 서버 측에서의 데이터 검증 작업이 성공하는 것을 확인할 수 있습니다. (물론, 지난번 글에도 언급했지만, 이건 실험값에 불과하다는 점을 간과해서는 안됩니다.)
이 글에서 사용된 C/C++ 예제 역시 첨부해 두었습니다. (참고로, 서버 예제는
지난번 글의 코드와 완전히 동일합니다.) 이번에도 역시 테스트 코드의 조건에 의문 사항이나 개선이 필요하면 덧글 부탁드립니다
[이 글에 대해서 여러분들과 의견을 공유하고 싶습니다. 틀리거나 미흡한 부분 또는 의문 사항이 있으시면 언제든 댓글 남겨주십시오.]