TCP client library using Winsock WSAEventSelect in C++ - Take 2

Question

This is my second take on creating a tcp client C++ class. The class declaration is general but the implementation is Windows only at this stage and uses the Microsoft Winsock WSAEventSelect model.

The first attempt is here:

TCP client library using Winsock WSAEventSelect in C++

for some background.

The changes are mostly removing the static socket and event handle variables which meant a caller could only use one instance of the class in a program. Also the lifecycle of the thread used to handle socket events was flawed so that has been addressed.

I may have missed some test cases so would appreciate any comments whatsoever.

A concern I have is that events are raised and the library calls into user defined callback functions. These callback functions will be called from a different thread to the thread used to call the library functions so that could be a concern. not sure yet how to fix that issue.

Also there is no logging in the library itself - not really sure how to fix that one either. Possibly take an output stream for logging in the constructor.

Anyway, any comments at all would be very much appreciated.

tcpclient.hpp header:

/*
Description: Asynchronous socket client interface
Author: Angus Comber
*/
#ifndef TCPCLIENT_HPP_
#define TCPCLIENT_HPP_
#include <thread> // comms thread
#include <functional> // callback functions
namespace itl {
 enum SOCKET_ERROR_CODE { ERR_FATAL, ERR_INFORMATIONAL };
 // forward declare pimpl
 struct socketimpl;
 class tcpclient {
 public:
 /// construct with async connection, data received and exception callbacks 
 /// and optional data buffersize
 tcpclient(std::function<void()> connectcb, 
 std::function<void(const unsigned char*, unsigned)> receivecb, 
 std::function<void(const int, const char*)> exceptioncb, 
 const int buffersize = 4096);
 ~tcpclient();
 //! connect to a host
 bool connect(const char* host, unsigned short port);
 //! send byte stream to remote endpoint
 unsigned send(const unsigned char* data, unsigned length);
 //! send string remote endpoint
 unsigned send(const char* data);
 //! close connection
 int close();
 //! check if connected
 bool is_connected() const;
 tcpclient(const tcpclient&) = delete;
 tcpclient& operator=(const tcpclient&) = delete;
 private:
 int closesocket();
 int shutdown();
 void exception_event(const int error, const char* message);
 //! async connected event
 std::function<void()> connectfunc_;
 //! async data received event
 std::function<void(const unsigned char*, unsigned)> receivefunc_;
 //! async exception event
 std::function<void(const int, const char*)> exceptionfunc_;
 std::thread reader_thread;
 void comms_channel_handler();
 void stop_comms_channel();
 void start_comms_channel();
 bool shutting_down_;
 bool connected_;
 const int buffersize_;
 socketimpl *pimpl_; // Handle object - pimpl
 };
} // namespace itl
#endif // TCPCLIENT_HPP_

tcpclient implementation:

/*
Description: Implementation of asynchronous socket client 
using winsock event model
Author: Angus Comber
*/
#include "tcpclient.hpp"
#ifdef WIN32
#ifndef WIN32_LEAN_AND_MEAN
#define WIN32_LEAN_AND_MEAN
#endif
#include <winsock2.h> // Windows sockets v2
#include <ws2tcpip.h> // getaddrinfo
#pragma comment(lib,"ws2_32.lib") //winsock2 lib
#include <string>
namespace itl {
// hide implementation details from header
struct socketimpl
{
public:
 SOCKET socket_ = INVALID_SOCKET;
 WSAEVENT comms_event;
};
tcpclient::tcpclient(std::function<void()> connectcb, 
 std::function<void(const unsigned char*, unsigned)> receivecb, 
 std::function<void(const int, const char*)> exceptioncb, 
 const int buffersize)
 : connectfunc_(connectcb), receivefunc_(receivecb), exceptionfunc_(exceptioncb), 
 buffersize_(buffersize), shutting_down_(false), connected_(false),
 pimpl_(new socketimpl) {
 WSADATA w = { 0 };
 int error = WSAStartup(0x0202, &w);
 if (error || w.wVersion != 0x0202)
 { // there was an error
 throw "Could not initialise Winsock2";
 }
}
// creates socket and runs thread for processing socket comms
void tcpclient::start_comms_channel() {
 stop_comms_channel();
 pimpl_->socket_ = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); // Create socket
 shutting_down_ = false;
 // launch listener thread to handle any received data
 reader_thread = std::thread(&tcpclient::comms_channel_handler, this);
 // yield main thread to give reader_thread time to get going
 std::this_thread::yield();
}
tcpclient::~tcpclient() {
 close();
 WSACleanup();
 delete pimpl_;
}
bool tcpclient::connect(const char* host, unsigned short port) {
 if (connected_) {
 exception_event(ERR_INFORMATIONAL, "attempt to connect failed - socket client already connected, disconnect first");
 return false; // already connected, client must disconnect first
 }
 addrinfo* result = NULL;
 addrinfo hints = {};
 sockaddr_in target = {};
 hints.ai_family = AF_UNSPEC;
 hints.ai_socktype = SOCK_STREAM;
 hints.ai_protocol = IPPROTO_TCP;
 // inet_pton() returns 1 on success. It returns -1 if there was an error (errno is set), 
 // or 0 if the input isn't a valid IP address.
 int ret = inet_pton(AF_INET, host, &(target.sin_addr));
 if (ret != 1) {
 // ok so we assume not a proper ip address, so try getaddrinfo - might be a hostname/domain name
 ret = getaddrinfo(host, NULL, &hints, &result);
 if (ret != 0) {
 std::string s = "getaddrinfo failed with error: " + std::to_string(ret) + '\n';
 exception_event(ERR_INFORMATIONAL, s.c_str());
 return false;
 }
 memcpy(&target, result->ai_addr, sizeof(sockaddr_in));
 }
 target.sin_family = AF_INET; // IPv4
 target.sin_port = htons(port);
 // stop comms thread created from a (possible) previous connection
 stop_comms_channel();
 // start reading thread now that we know connection arguments are valid
 start_comms_channel();
 ret = ::connect(pimpl_->socket_, reinterpret_cast<sockaddr *>(&target), sizeof(sockaddr));
 if (ret != 0) {
 ret = WSAGetLastError();
 if (ret == WSAEWOULDBLOCK) {
 // normal asynchronous connection
 ret = 0; 
 }
 else {
 char* s = NULL;
 FormatMessageA(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
 NULL, ret,
 MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
 s, 0, NULL);
 exception_event(ERR_INFORMATIONAL, s);
 }
 }
 return ret == 0;
}
bool tcpclient::is_connected() const {
 return connected_;
}
unsigned tcpclient::send(const unsigned char* data, unsigned length) {
 return ::send(pimpl_->socket_, reinterpret_cast<const char*>(data), length, 0);
}
unsigned tcpclient::send(const char* data) {
 return ::send(pimpl_->socket_, data, strlen(data), 0);
}
int tcpclient::close() {
 stop_comms_channel();
 int ret = 0;
 if (pimpl_->socket_ != INVALID_SOCKET) {
 shutdown();
 ret = closesocket();
 }
 return ret;
}
int tcpclient::closesocket() {
 int ret = INVALID_SOCKET;
 if (pimpl_->socket_ != INVALID_SOCKET) {
 ret = ::closesocket(pimpl_->socket_);
 pimpl_->socket_ = INVALID_SOCKET;
 }
 return ret;
}
int tcpclient::shutdown() {
 int ret = INVALID_SOCKET;
 if (pimpl_->socket_ != INVALID_SOCKET) {
 // SD_SEND says to server, we have no more data to send
 // server may respond with response data
 ret = ::shutdown(pimpl_->socket_, SD_SEND);
 }
 return ret;
}
void tcpclient::exception_event(const int error, const char* message) {
 if (exceptionfunc_) {
 exceptionfunc_(error, message);
 }
 // cleanup if get an exception
 stop_comms_channel();
}
void tcpclient::stop_comms_channel() {
 // these 2 make thread loop stop - so thread runs out
 shutting_down_ = true;
 if (pimpl_->comms_event != WSA_INVALID_EVENT) {
 WSASetEvent(pimpl_->comms_event);
 }
 // you cannot join reader thread to itself - so check
 // wait for thread to stop
 if (reader_thread.joinable() && std::this_thread::get_id() != reader_thread.get_id()) {
 reader_thread.join(); // prevents crash - due to terminate being called on running thread still 'alive'
 }
}
void tcpclient::comms_channel_handler() {
 // Create an event object to be used with this socket
 pimpl_->comms_event = WSACreateEvent();
 if (pimpl_->comms_event == WSA_INVALID_EVENT)
 {
 exception_event(ERR_FATAL, "Error creating winsock WSACreateEvent object");
 }
 // setup winsock event handling 
 int ret = WSAEventSelect(pimpl_->socket_, pimpl_->comms_event, FD_READ | FD_CONNECT | FD_CLOSE);
 if (ret != 0) {
 int lasterror = WSAGetLastError();
 std::string s = "Winsock communication error - unable to create event object winsock error " + std::to_string(lasterror);
 exception_event(ERR_FATAL, s.c_str()); 
 return;
 }
 // Handle async network events
 WSANETWORKEVENTS events;
 char* buffer = new char[buffersize_]();
 while (!shutting_down_) {
 // Wait for a socket event
 DWORD dwRet = WSAWaitForMultipleEvents(1, &pimpl_->comms_event, FALSE, WSA_INFINITE, FALSE);
 if (dwRet == WSA_WAIT_TIMEOUT)
 {
 // will never get to here as using WSA_INFINITE
 break;
 }
 // Type of event that occurred on socket_
 int nRet = WSAEnumNetworkEvents(pimpl_->socket_, pimpl_->comms_event, &events);
 if (nRet == SOCKET_ERROR)
 {
 exception_event(ERR_FATAL, "Winsock WaitForCommsEvent socket error");
 break;
 }
 // Handle events
 // Connect event
 if (events.lNetworkEvents & FD_CONNECT)
 {
 if (events.iErrorCode[FD_CONNECT_BIT] != 0)
 {
 // if asynchronous connect failure, we get informed here
 exception_event(ERR_FATAL, "Winsock events error code for FD_CONNECT_BIT");
 break;
 }
 connected_ = true;
 if (connectfunc_) {
 connectfunc_();
 }
 }
 // Read event
 if (events.lNetworkEvents & FD_READ)
 {
 // No need to keep reading, will get a new event for any data not read
 int bytes = recv(pimpl_->socket_, buffer, buffersize_, 0);
 if (bytes == SOCKET_ERROR)
 {
 exception_event(ERR_FATAL, "Winsock socket error reading received data");
 break;
 }
 if (bytes > 0)
 {
 if (receivefunc_) {
 receivefunc_(reinterpret_cast<const unsigned char*>(buffer), bytes);
 }
 }
 }
 // Close event
 if (events.lNetworkEvents & FD_CLOSE)
 {
 connected_ = false;
 exception_event(ERR_INFORMATIONAL, "Socket closed");
 break;
 }
 }
 pimpl_->comms_event = WSA_INVALID_EVENT;
 delete [] buffer;
}
} // namespace itl
#elif
#error Only Windows platform supported
#endif

main.cpp testing/exercising source:

#include "tcpclient.hpp"
#include <stdio.h>
#include <string.h>
#include <chrono>
#include <iostream>
static void printhex(const unsigned char* data, const unsigned len)
{
 for (size_t i = 0; i < len; ++i) {
 printf("%02x ", data[i]);
 if ((i + 1) % 16 == 0)
 putchar('\n');
 }
 putchar('\n');
}
class http_client
{
public:
 http_client() : client(NULL) {
 client = new itl::tcpclient([this]() { connect_handler(); }, 
 [this](const unsigned char* data, unsigned length) { data_handler(data, length); }, 
 [this](const int errorcode, const char* error_message) { exception_handler(errorcode, error_message); });
 printf("http_client ctor called\n");
 }
 ~http_client() { delete client; printf("http_client dtor called\n"); }
 void close() { client->close(); printf("http_client close called\n"); }
 void connect(const char* host, const unsigned short port) {
 printf("http_client connect called\n");
 host_ = host;
 port_ = port;
 client->connect(host, port);
 }
 void do_connect() {
 printf("do_connect called\n");
 std::string url = "GET / HTTP/1.0\r\nHost: " + host_ + "\r\nConnection: keep-alive\r\nUser-Agent: test socket program/1.0\r\nAccept-Encoding: gzip/\r\nAccept-Charset: ISO-8859-1,UTF-8;q=0.7,*;q=0.7\r\nCache-Control: no - cache/\r\nAccept-Language: de, en; q = 0.7, en - us; q = 0.3/\r\n\r\n";
 client->send(url.c_str());
 }
 void data_handler(const unsigned char* data, unsigned length) {
 printf("data_handler called %u bytes\n", length);
 printhex(data, length);
 }
 void connect_handler() {
 printf("connect_handler called\n");
 do_connect();
 }
 void exception_handler(const int errorcode, const char* error_message) {
 printf("Error %d, %s\n", errorcode, error_message);
 }
private:
 std::string host_;
 unsigned port_;
 itl::tcpclient* client;
 http_client(const http_client&) = delete;
 http_client& operator=(const http_client&) = delete;
};
int main() {
 http_client client; 
 // bogus host
 client.connect("codereview.stockexchange.com", 80);
 // stop program finishing before we have downloaded all web content
 for (int i = 0; i < 5; ++i) {
 putchar('.');
 std::this_thread::sleep_for(std::chrono::seconds(1));
 }
 // this host does connect - but immediate rejection
 client.connect("codereview.stickexchange.com", 80);
 // stop program finishing before we have downloaded all web content
 for (int i = 0; i < 5; ++i) {
 putchar('.');
 std::this_thread::sleep_for(std::chrono::seconds(1));
 }
 // now try successful case - these hosts set Connection Keep-alive meaning they don't immediately
 // close the connection and so we close at end of test
 http_client client2; // test for concurrent clients
 http_client client3;
 client.connect("codereview.stackexchange.com", 80);
 client2.connect("codereview.stackexchange.com", 80);
 client3.connect("codereview.stackexchange.com", 80);
 // stop program finishing before we have downloaded all web content
 for (int i = 0; i < 5; ++i) {
 putchar('.');
 std::this_thread::sleep_for(std::chrono::seconds(1));
 }
 client.close();
 client2.close();
 client3.close();
}

Answer 1

stop_comms_channel

In connect, you have the code:

// stop comms thread created from a (possible) previous connection
stop_comms_channel();
// start reading thread now that we know connection arguments are valid
start_comms_channel();

In turn, start_comms_channel does:

void tcpclient::start_comms_channel() {
 stop_comms_channel();

One of the calls to stop_comms_channel is unnecessary (probably the one in connect?).

connected_

When a network event causes the socket to be closed, connected_ is set back to false. This doesn't happen however if the socket is closed by the client calling the close method. So you can't always reuse the same client by calling close, followed by connect. Close should probably reset the flag.

exception_event

I can't help but wonder if your exception_event method should be doing the cleanup, before calling exceptionfunc_.

void tcpclient::exception_event(const int error, const char* message) {
 // cleanup if get an exception
 stop_comms_channel();
 if (exceptionfunc_) {
 exceptionfunc_(error, message);
 }
}

Does this make sense? Maybe, it depends what you're expecting to happen in the exception callback. I can see a potential usage where the client might want to attempt to reconnect in an the event of an exception. With your current handler the socket would be immediately shutdown when the exception handler completed. Other work would need to be done to support this kind of callback however, since there's a chance that connect would be being called on the worker thread (so you couldn't shutdown / restart the thread).

Logging

Logging strategies are often viewed more at an application level than at an individual class level. So, for example you might have a global logger that exposed log methods. If you don't want to go down that route, then since you're already providing callbacks for events, you could allow clients to register a logging callback as well. They can always register nullptr if they aren't interested in logging.

Redundant code

if (dwRet == WSA_WAIT_TIMEOUT)
{
 // will never get to here as using WSA_INFINITE
 break;
}

If it can't happen, why are you checking it?

Callback definitions

Consider typedefing the callback methods to cleanup the method definitions and make them more explicit:

typedef std::function<void()> ConnectCallback;

etc.

Thread safety

You're not protecting any of the member variables in the tcpclient class. This is probably going to work out ok in most situations, since the overlap is small relative to the more expensive network operations. However there are some race conditions that aren't handled. So, for example in connect, the first thing it does is check if the client is already connected:

if (connected_) {
 exception_event(ERR_INFORMATIONAL, "attempt to connect failed - socket client already connected, disconnect first");
 return false; // already connected, client must disconnect first
}

The connected flag isn't set in the connect thread, it is set in the worker thread, after it has been started. This means that it's possible to call connect twice in succession without this guard triggering or an error being raised:

client.connect("codereview.stockexchange.com", 80);
client.connect("codereview.stockexchange.com", 80);

 http_client ctor called
 http_client connect called
 http_client connect called
 connect_handler called
 do_connect called
 .connect_handler called
 do_connect called

This is probably wrong...