* added busy map so we don't select()

  events that are currently ongoing
* add disconnect log
* change busy map logic
* removed shadowing of a variable in
  remove lambda
* re-wrote the recv handling on
  server again, should be good
  now
* added -wall in build for all platforms

Author: lpcvoid

CMakeLists.txt

@@ -2,6 +2,13 @@ cmake_minimum_required(VERSION 3.15)
 project(netlib)
 set(CMAKE_CXX_STANDARD 20)
 
+# we want the maximum amount of compiler bickering
+if(MSVC)
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /W4")
+else()
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall")
+endif()
+
 find_package(Threads)
 
 set(NETLIB_SRC

src/server.hpp

@@ -7,6 +7,7 @@
 #include <cassert>
 #include <functional>
 #include <iostream>
+#include <map>
 #include <mutex>
 #include <optional>
 #include <thread>
@@ -43,7 +44,8 @@ namespace netlib {
         callback_error_t _cb_on_error{};
         std::thread _accept_thread;
         std::thread _processor_thread;
-        netlib::thread_pool _thread_pool;
+        netlib::thread_pool _thread_pool = netlib::thread_pool::create<1,1>();
+        std::map<socket_t, std::atomic<bool>> _busy_map;
 
         inline void processing_func() {
           while (_server_active) {
@@ -54,9 +56,14 @@ namespace netlib {
             {
               std::lock_guard<std::mutex> lock(_mutex);
               local_clients = _clients;
-              for (auto& client : local_clients) {
+              for (auto &client : local_clients) {
                 socket_t fd = client.socket.get_raw().value();
-                if (highest_fd < fd){
+                assert(_busy_map.contains(fd));
+                if (_busy_map[fd]) {
+                  // this fd is currently being handled by a task in threadpool
+                  continue;
+                }
+                if (highest_fd < fd) {
                   highest_fd = fd;
                 }
                 FD_SET(fd, &fdset);
@@ -69,20 +76,31 @@ namespace netlib {
             int32_t select_res = ::select(highest_fd + 1, &fdset, nullptr, nullptr, &tv);
             if (select_res > 0) {
               std::vector<client_endpoint> client_refs(select_res);
-              uint32_t index = 0;
+              int32_t index = 0;
               {
+                std::lock_guard<std::mutex> lock(_mutex);
                 for (auto& client : local_clients) {
                   socket_t fd = client.socket.get_raw().value();
                   if (FD_ISSET(fd, &fdset)){
                     client_refs[index++] = client;
+                    assert(_busy_map.contains(fd));
+                    _busy_map[fd] = true;
                   }
                 }
               }
               assert(index == select_res);
               //add callback tasks to threadpool for processing
               for (auto& client_to_recv : client_refs) {
                 _thread_pool.add_task([&](client_endpoint ce){
-                  this->handle_client(ce);
+                  socket_t id = ce.socket.get_raw().value();
+                  std::error_condition error = this->handle_client(ce);
+                  if ((error) && (_cb_on_error)) {
+                      _cb_on_error(ce, error);
+                  }
+                  std::lock_guard<std::mutex> lock(_mutex);
+                  if (_busy_map.contains(id)) {
+                    _busy_map[id] = false;
+                  }
                 }, client_to_recv);
               }
 
@@ -100,7 +118,7 @@ namespace netlib {
           while (_server_active) {
             std::this_thread::sleep_for(std::chrono::milliseconds(10));
             new_endpoint.addr_len = sizeof(addrinfo);
-            int32_t status = ::accept(_listener_sock->get_raw().value(), &new_endpoint.addr, &new_endpoint.addr_len);
+            socket_t status = ::accept(_listener_sock->get_raw().value(), &new_endpoint.addr, &new_endpoint.addr_len);
             if (status > 0) {
               new_endpoint.socket.set_raw(status);
               new_endpoint.socket.set_nonblocking(true);
@@ -120,86 +138,82 @@ namespace netlib {
                   if (_cb_on_error) {
                     _cb_on_error(new_endpoint, std::errc::connection_aborted);
                   }
-                  new_endpoint.socket.close();
+                  remove_client(new_endpoint);
                   std::cout << "server kicked client accept" << std::endl;
                   continue;
                 }
               }
               if (new_endpoint.socket.is_valid()) {
                 std::lock_guard<std::mutex> lock(_mutex);
+                std::cout << "server added new client, id " << status << std::endl;
                 _clients.push_back(new_endpoint);
+                _busy_map[status] = false;
               }
             }
           }
         }
 
         inline std::error_condition handle_client(client_endpoint endpoint) {
           std::vector<uint8_t> total_buffer;
-          std::array<uint8_t, 2048> buffer{};
+          std::array<uint8_t, 1024> buffer{};
           ssize_t recv_res = 0;
-          ssize_t recv_res_cycle = 0;
-          while ((recv_res_cycle = ::recv(endpoint.socket.get_raw().value(), buffer.data(), buffer.size(), 0)) > 0) {
-            total_buffer.insert(total_buffer.end(), buffer.begin(), buffer.begin() + recv_res_cycle);
-            recv_res += recv_res_cycle;
-          }
-          if (recv_res == 0){
-            std::cout << "server recv_res == 0" << std::endl;
-            if (_cb_on_error) {
-              _cb_on_error(endpoint, std::errc::connection_aborted);
-            }
-            remove_client(endpoint.socket.get_raw().value());
-            endpoint.socket.close();
-            std::cout << "server kicked client recv 0" << std::endl;
-            return std::errc::connection_aborted;
-          } else if (recv_res < 0) {
-            //error
-            std::cout << "server recv_res == -1" << std::endl;
-            std::error_condition recv_error = socket_get_last_error();
-            //we do not want to spam callback with wouldblock messages
-            //for portability we shall check both EAGAIN and EWOULDBLOCK
-            if ((recv_error != std::errc::resource_unavailable_try_again) &&
-                  (recv_error != std::errc::operation_would_block)) {
-              if (_cb_on_error) {
-                _cb_on_error(endpoint, recv_error);
-              }
-            }
-            return recv_error;
-          } else {
-            std::cout << "server recv_res > 0: " << recv_res << " " << total_buffer.size() << std::endl;
-            //we got data
-            if (_cb_on_recv) {
-              netlib::server_response response = _cb_on_recv(endpoint, total_buffer);
-              if (!response.answer.empty()) {
-                ssize_t send_result = ::send(endpoint.socket.get_raw().value(),
-                                             response.answer.data(),
-                                             response.answer.size(),
-                                             0);
-                if ((send_result != response.answer.size()) && (_cb_on_error)) {
-                  _cb_on_error(endpoint, socket_get_last_error());
-                }
-              }
-              if (response.terminate) {
-                if (_cb_on_error) {
-                  _cb_on_error(endpoint, std::errc::connection_aborted);
+          while (true) {
+            recv_res = ::recv(endpoint.socket.get_raw().value(),
+                                    buffer.data(), buffer.size(), 0);
+            if (recv_res > 0) {
+              total_buffer.insert(total_buffer.end(), buffer.begin(),
+                                  buffer.begin() + recv_res);
+            } else if (recv_res == 0) {
+              std::cout << "server recv_res == 0" << std::endl;
+              remove_client(endpoint);
+              std::cout << "server kicked client recv 0" << std::endl;
+              return std::errc::connection_aborted;
+            } else if (recv_res < 0) {
+              // error
+              std::error_condition recv_error = socket_get_last_error();
+              if ((recv_error == std::errc::resource_unavailable_try_again) ||
+                  (recv_error == std::errc::operation_would_block)) {
+                // no more data, return what we got
+                if (_cb_on_recv) {
+                  netlib::server_response response =
+                      _cb_on_recv(endpoint, total_buffer);
+                  if (!response.answer.empty()) {
+                    ssize_t send_result = ::send(
+                        endpoint.socket.get_raw().value(),
+                        response.answer.data(), response.answer.size(), 0);
+                    if (send_result != response.answer.size()) {
+                      return socket_get_last_error();
+                    }
+                  }
+                  if (response.terminate) {
+                    remove_client(endpoint);
+                    std::cout << "server kicked client because wanted"
+                              << std::endl;
+                    return std::errc::connection_aborted;
+                  }
                 }
-                remove_client(endpoint.socket.get_raw().value());
-                endpoint.socket.close();
-                std::cout << "server kicked client because wanted" << std::endl;
+                return {};
+              } else {
+                std::cout << "server recv_res == -1" << std::endl;
+                return recv_error;
               }
-
             }
-            return {};
           }
-
         }
 
-        bool remove_client(socket_t socket_id) {
+        bool remove_client(client_endpoint& ce) {
           //the remove_if-> erase idiom is perhaps my most hated part about std containers
           std::lock_guard<std::mutex> lock(_mutex);
-          return _clients.erase(
-              std::remove_if(_clients.begin(), _clients.end(), [&](const client_endpoint& ce) {
-                return ce.socket.get_raw() == socket_id;
-              }),_clients.end()) != _clients.end();
+          std::size_t client_count = _clients.size();
+          _busy_map.erase(ce.socket.get_raw().value());
+          std::cout << "remove_client, client count at start " << _clients.size() << std::endl;
+          std::erase_if(_clients, [&](const client_endpoint& single_endpoint){
+            return (ce.socket.get_raw().value() == single_endpoint.socket.get_raw().value());
+          });
+          std::cout << "removed client, id " << ce.socket.get_raw().value() << std::endl;
+          ce.socket.close();
+          assert((client_count - _clients.size()) == 1);
+          return true;
         }
 
       public:
@@ -233,9 +247,7 @@ namespace netlib {
               close_and_free();
               continue;
             }
-            _listener_sock->set_nonblocking(true);
-            //_listener_sock->set_reuseaddr(true);
-
+            _listener_sock->set_nonblocking(true); //we want to be able to join
             int32_t res = ::bind(_listener_sock->get_raw().value(), res_addrinfo->ai_addr, res_addrinfo->ai_addrlen);
             if (res < 0) {
               close_and_free();

src/thread_pool.hpp

@@ -42,7 +42,7 @@ namespace netlib {
     }
     void create_threads(std::size_t thread_count) {
       _thread_pool.reserve(thread_count + _thread_pool.size());
-      for (int i = 0; i < thread_count; ++i) {
+      for (uint32_t i = 0; i < thread_count; ++i) {
         add_thread();
       }
     }