Network.cpp

#if !defined( EMSCRIPTEN )
 
#include "Network.h"
#include "Connection.h"
 
#include "../Platform/Threads.h"
#include "../Platform/Timer.h"
 
#include <algorithm>
#include <condition_variable>
#include <vector>
#include <chrono>
 
#if !defined( _WIN32 )
  #include <sys/select.h>
#endif
 
using namespace std::chrono;
 
namespace Cogs::Network {
  Thread*                        gThread = nullptr;
  bool                           gExit = false;
  std::mutex                     gMutex;
  std::vector< ConnectionBase* > gNewConnections;
  std::vector< ConnectionBase* > gConnections;
 
  std::condition_variable        gAddCV;
  std::condition_variable        gRemoveCV;
 
  void  ThreadFn();
}
 
void Cogs::Network::addConnection(ConnectionBase* connection) {
  std::unique_lock lock(gMutex, std::defer_lock);
  bool do_lock = !gThread || (std::this_thread::get_id() != gThread->get_id());
  if(do_lock) lock.lock();
 
  if ((std::find(gNewConnections.begin(), gNewConnections.end(), connection) == gNewConnections.end()) &&
      (std::find(gConnections.begin(), gConnections.end(), connection) == gConnections.end())) {
    gNewConnections.push_back(connection);
 
    if (!gThread) {
      gThread = new Thread(&ThreadFn);
      Threads::setName(*gThread, "Network");
    }
    else if(gThread && (std::this_thread::get_id() != gThread->get_id())){
      if(do_lock) lock.unlock();
      gAddCV.notify_all();
    }
  }
}
 
void Cogs::Network::removeConnection(ConnectionBase* connection) {
  std::unique_lock lock(gMutex, std::defer_lock);
  bool do_lock = !gThread || (std::this_thread::get_id() != gThread->get_id());
  if(do_lock) lock.lock();
  auto              ne = gNewConnections.end();
  auto              ni = std::find(gNewConnections.begin(), ne, connection);
 
  if (ni != ne) {
    gNewConnections.erase(ni);
  }
  else {
    auto e = gConnections.end();
    auto i = std::find(gConnections.begin(), e, connection);
 
    if (i != e) {
      if(do_lock) lock.unlock();
      connection->remove();
      if(do_lock) lock.lock();
 
      if (gThread && (std::this_thread::get_id() != gThread->get_id())) {
        for (;;) {
          e = gConnections.end();
          if (std::find(gConnections.begin(), e, connection) == e) {
            break;
          }
          gRemoveCV.wait(lock);
        }
      }
    }
  }
}
 
void Cogs::Network::closedown() {
  if (gThread) {
    gExit = true;
    gAddCV.notify_all();
    gThread->join();
 
    delete gThread;
    gThread = nullptr;
    gExit = false;
  }
}
 
void Cogs::Network::ThreadFn() {
  uint32_t          current = 0;
  std::unique_lock  lock(gMutex);
 
  while (!gExit) {
    gConnections.insert(gConnections.end(), gNewConnections.begin(), gNewConnections.end());
    gNewConnections.clear();
    size_t  connectionCount = gConnections.size();
    lock.unlock();
    gRemoveCV.notify_all();
    lock.lock();
 
    if (connectionCount) {
      fd_set  socketHandles;
      Socket  highest = 0;
      timeval timeout;
      int     connectedCount = 0;
 
      for (auto connection : gConnections) {
        connection->processOutgoing();
      }
 
      FD_ZERO(&socketHandles);
 
      if (connectionCount < FD_SETSIZE) {
        for (auto connection : gConnections) {
          if (connection->isConnected()) {
            FD_SET(connection->getSocket(), &socketHandles);
            highest = std::max(highest, connection->getSocket());
          }
        }
      }
      else {
        if (current >= connectionCount) {
          current = 0;
        }
        for (uint32_t i = 0; i < FD_SETSIZE; ++i) {
          ConnectionBase* connection = gConnections[current];
 
          if (connection->isConnected()) {
            FD_SET(connection->getSocket(), &socketHandles);
            highest = std::max(highest, connection->getSocket());
          }
          current = ++current % connectionCount;
        }
      }
 
      timeout.tv_sec = 0;
      timeout.tv_usec = 50000;
 
      lock.unlock();
      if (select(static_cast<int>(highest + 1), &socketHandles, nullptr, nullptr, &timeout) > 0) {
        std::lock_guard select_lock(gMutex);
 
        for (auto connection : gConnections) {
          if (FD_ISSET(connection->getSocket(), &socketHandles)) {
            connection->processIncoming();
          }
        }
      }
      lock.lock();
 
      for (auto i = gConnections.begin(); i != gConnections.end(); ) {
        ConnectionBase* connection = *i;
 
        if (!connection->isConnected()) {
          if (connection->shouldDelete()) {
            delete connection;
            i = gConnections.erase(i);
          }
          else if (connection->isAutoConnectEnabled()) {
            if (connection->getReconnectTime() < Timer::currentTimeMilliseconds()) {
              connection->reconnect();
            }
            ++i;
          }
          else {
            i = gConnections.erase(i);
          }
        }
        else if (connection->shouldRemove()) {
          connection->disconnect();
          if (connection->shouldDelete()) {
            delete connection;
          }
          i = gConnections.erase(i);
        }
        else {
          ++connectedCount;
          ++i;
        }
      }
      if (!connectedCount) {
        gAddCV.wait_for(lock, 1ms);
      }
    }
    else {
      if(!gExit) {
        gAddCV.wait(lock);
      }
    }
  }
  gNewConnections.clear();
  gConnections.clear();
}
 
#endif