<commit_before>#include "cpp_redis/network/windows/io_service.hpp"
#include "cpp_redis/redis_error.hpp"

namespace cpp_redis {

namespace network {

const std::shared_ptr<io_service>&
io_service::get_instance(void) {
  static std::shared_ptr<io_service> instance = std::shared_ptr<io_service>{new io_service};
  return instance;
}

io_service::io_service(size_t max_worker_threads)
: m_should_stop(false) {
  //Determine the size of the thread pool dynamically.
  //2 * number of processors in the system is our rule here.
  SYSTEM_INFO info;
  ::GetSystemInfo(&info);
  m_worker_thread_pool_size = (info.dwNumberOfProcessors * 2);

  if (m_worker_thread_pool_size > max_worker_threads)
    m_worker_thread_pool_size = max_worker_threads;

  WSADATA wsaData;
  int nRet = 0;
  if ((nRet = WSAStartup(0x202, &wsaData)) != 0) //Start winsock before any other socket calls.
    throw cpp_redis::redis_error("Could not init cpp_redis::io_service, WSAStartup() failure");

  //Create completion port.  Pass 0 for parameter 4 to allow as many threads as there are processors in the system
  m_completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
  ;
  if (INVALID_HANDLE_VALUE == m_completion_port)
    throw cpp_redis::redis_error("Could not init cpp_redis::io_service, CreateIoCompletionPort() failure");

  //Now startup worker thread pool which will service our async io requests
  for (unsigned int i = 0; i < m_worker_thread_pool_size; i++) {
    m_worker_threads.push_back(std::thread(&io_service::process_io, this));
  }
}

io_service::~io_service(void) {
  shutdown();
}

void
io_service::shutdown() {
  m_should_stop = true;

  //Iterate all of our sockets and shutdown any IO worker threads by posting a issuing a special
  //message to the thread to tell them to wake up and shut down.
  io_context_info* pInfo = NULL;

  auto sock_it = m_sockets.begin();
  while (sock_it != m_sockets.end()) {
    auto& info = sock_it->second;
    //Post for each of our worker threads.
    int workers = m_worker_threads.size();
    for (int i = 0; i < workers; i++)
      PostQueuedCompletionStatus(m_completion_port, 0, NULL, NULL); //Use NULL for the completion key to wake them up.
    sock_it++;
  }

  // Wait for the threads to finish
  for (auto& t : m_worker_threads)
    t.join();

  //close the completion port otherwise the worker threads will all be waiting on GetQueuedCompletionStatus()
  if (m_completion_port) {
    CloseHandle(m_completion_port);
    m_completion_port = NULL;
  }
}

//! add or remove a given socket from the io service
//! untrack should never be called from inside a callback
void
io_service::track(SOCKET sock, const disconnection_handler_t& handler) {
  std::lock_guard<std::recursive_mutex> lock(m_socket_mutex);

  //Add the socket to our map and return the allocated struct
  auto& info = m_sockets[sock];

  info.hsock                 = sock;
  info.disconnection_handler = handler;

  //Associate the socket with our io completion port.
  if (NULL == CreateIoCompletionPort((HANDLE) sock, m_completion_port, (DWORD_PTR) &m_sockets[sock], 0)) {
    throw cpp_redis::redis_error("Track() failed to create CreateIoCompletionPort");
  }
}

void
io_service::untrack(SOCKET sock) {
  std::lock_guard<std::recursive_mutex> lock(m_socket_mutex);
  m_sockets.erase(sock);
}

bool
io_service::async_read(SOCKET sock, std::vector<char>& buffer, std::size_t read_size, const read_callback_t& callback) {
  std::lock_guard<std::recursive_mutex> lock(m_socket_mutex);

  auto sock_it = m_sockets.find(sock);
  if (sock_it == m_sockets.end())
    return false;

  //Resize the buffer to the correct size if we need it to be to hold the incoming data.
  if (buffer.size() < read_size)
    buffer.resize(read_size);

  auto& sockinfo         = sock_it->second;
  sockinfo.read_buffer   = &buffer;
  sockinfo.read_callback = callback;

  DWORD dwRecvNumBytes = 0;
  DWORD dwFlags        = 0;
  WSABUF buffRecv;

  buffRecv.buf = sockinfo.read_buffer->data();
  buffRecv.len = read_size;

  //We need a new overlapped struct for EACH overlapped operation.
  //we reuse them over and over.
  io_context_info* p_io_info = sockinfo.get_pool_io_context();

  int nRet = WSARecv(sock, &buffRecv, 1, &dwRecvNumBytes, &dwFlags, &(p_io_info->overlapped), NULL);

  if (nRet == SOCKET_ERROR && (ERROR_IO_PENDING != WSAGetLastError())) {
    //Fire the disconnect handler
    sockinfo.disconnection_handler(*this);
    return false;
  }

  return true;
}<commit_msg>fix windows io_service: set the io_state to IO_OP_READ in async_read<commit_after>#include "cpp_redis/network/windows/io_service.hpp"
#include "cpp_redis/redis_error.hpp"

namespace cpp_redis {

namespace network {

const std::shared_ptr<io_service>&
io_service::get_instance(void) {
  static std::shared_ptr<io_service> instance = std::shared_ptr<io_service>{new io_service};
  return instance;
}

io_service::io_service(size_t max_worker_threads)
: m_should_stop(false) {
  //Determine the size of the thread pool dynamically.
  //2 * number of processors in the system is our rule here.
  SYSTEM_INFO info;
  ::GetSystemInfo(&info);
  m_worker_thread_pool_size = (info.dwNumberOfProcessors * 2);

  if (m_worker_thread_pool_size > max_worker_threads)
    m_worker_thread_pool_size = max_worker_threads;

  WSADATA wsaData;
  int nRet = 0;
  if ((nRet = WSAStartup(0x202, &wsaData)) != 0) //Start winsock before any other socket calls.
    throw cpp_redis::redis_error("Could not init cpp_redis::io_service, WSAStartup() failure");

  //Create completion port.  Pass 0 for parameter 4 to allow as many threads as there are processors in the system
  m_completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
  ;
  if (INVALID_HANDLE_VALUE == m_completion_port)
    throw cpp_redis::redis_error("Could not init cpp_redis::io_service, CreateIoCompletionPort() failure");

  //Now startup worker thread pool which will service our async io requests
  for (unsigned int i = 0; i < m_worker_thread_pool_size; i++) {
    m_worker_threads.push_back(std::thread(&io_service::process_io, this));
  }
}

io_service::~io_service(void) {
  shutdown();
}

void
io_service::shutdown() {
  m_should_stop = true;

  //Iterate all of our sockets and shutdown any IO worker threads by posting a issuing a special
  //message to the thread to tell them to wake up and shut down.
  io_context_info* pInfo = NULL;

  auto sock_it = m_sockets.begin();
  while (sock_it != m_sockets.end()) {
    auto& info = sock_it->second;
    //Post for each of our worker threads.
    int workers = m_worker_threads.size();
    for (int i = 0; i < workers; i++)
      PostQueuedCompletionStatus(m_completion_port, 0, NULL, NULL); //Use NULL for the completion key to wake them up.
    sock_it++;
  }

  // Wait for the threads to finish
  for (auto& t : m_worker_threads)
    t.join();

  //close the completion port otherwise the worker threads will all be waiting on GetQueuedCompletionStatus()
  if (m_completion_port) {
    CloseHandle(m_completion_port);
    m_completion_port = NULL;
  }
}

//! add or remove a given socket from the io service
//! untrack should never be called from inside a callback
void
io_service::track(SOCKET sock, const disconnection_handler_t& handler) {
  std::lock_guard<std::recursive_mutex> lock(m_socket_mutex);

  //Add the socket to our map and return the allocated struct
  auto& info = m_sockets[sock];

  info.hsock                 = sock;
  info.disconnection_handler = handler;

  //Associate the socket with our io completion port.
  if (NULL == CreateIoCompletionPort((HANDLE) sock, m_completion_port, (DWORD_PTR) &m_sockets[sock], 0)) {
    throw cpp_redis::redis_error("Track() failed to create CreateIoCompletionPort");
  }
}

void
io_service::untrack(SOCKET sock) {
  std::lock_guard<std::recursive_mutex> lock(m_socket_mutex);
  m_sockets.erase(sock);
}

bool
io_service::async_read(SOCKET sock, std::vector<char>& buffer, std::size_t read_size, const read_callback_t& callback) {
  std::lock_guard<std::recursive_mutex> lock(m_socket_mutex);

  auto sock_it = m_sockets.find(sock);
  if (sock_it == m_sockets.end())
    return false;

  //Resize the buffer to the correct size if we need it to be to hold the incoming data.
  if (buffer.size() < read_size)
    buffer.resize(read_size);

  auto& sockinfo         = sock_it->second;
  sockinfo.read_buffer   = &buffer;
  sockinfo.read_callback = callback;

  DWORD dwRecvNumBytes = 0;
  DWORD dwFlags        = 0;
  WSABUF buffRecv;

  buffRecv.buf = sockinfo.read_buffer->data();
  buffRecv.len = read_size;

  //We need a new overlapped struct for EACH overlapped operation.
  //we reuse them over and over.
  io_context_info* p_io_info = sockinfo.get_pool_io_context();
  p_io_info->eOperation = IO_OP_READ;

  int nRet = WSARecv(sock, &buffRecv, 1, &dwRecvNumBytes, &dwFlags, &(p_io_info->overlapped), NULL);

  if (nRet == SOCKET_ERROR && (ERROR_IO_PENDING != WSAGetLastError())) {
    //Fire the disconnect handler
    sockinfo.disconnection_handler(*this);
    return false;
  }

  return true;
}
{'fraction_non_alphanumeric': 0.10379746835443038, 'fraction_numerical': 0.004182718767198679, 'mean_word_length': 4.408333333333333, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5093143', 'n_tokens_mistral': 3100, 'n_tokens_neox': 2984, 'n_words': 847}