<commit_before>#include "IOLoop.h"

#include "Logger.h"
#include "IOHandle.h"

#include <iostream>

#include <unistd.h>


namespace XXX {



/* Called on the IO thread when a ioreq attempt has completed */
static void __on_connect(uv_connect_t* __req, int status )
{
  std::unique_ptr<uv_connect_t> connect_req(__req);

  // IOLoop has set itself as the uv_loop data member
  IOLoop * ioloop = static_cast<IOLoop* >(__req->handle->loop->data);

  // get the user object
  std::unique_ptr<io_request> ioreq  (( io_request*) connect_req->data );
  Logger * __logptr = ioreq->logptr;

  if (status < 0)
  {
    // TODO: question for libuv: is there a way to get the errno ?
    _WARN_( "connect error, " <<  uv_strerror(status) );

    ioloop->add_active_handle(nullptr, status, ioreq.get());

    delete ioreq->tcp_connect;
  }
  else
  {

    IOHandle* ioh = new IOHandle( (uv_stream_t *) ioreq->tcp_connect,
                                  ioloop );
    ioloop->add_active_handle(ioh, 0, ioreq.get());
  }

}

static void __on_tcp_connect(uv_stream_t* server, int status)
{
  // IOLoop has set itself as the uv_loop data member
  IOLoop * myIOLoop = static_cast<IOLoop* >(server->loop->data);

  if (status < 0)
  {
    // TODO: logging
    fprintf(stderr, "New connection error %s\n", uv_strerror(status));
    return;
  }

  uv_tcp_t *client = new uv_tcp_t();
  uv_tcp_init(myIOLoop->uv_loop(), client);

  if (uv_accept(server, (uv_stream_t *) client) == 0)
  {
    IOHandle* ioh = new IOHandle( (uv_stream_t *) client, myIOLoop);

    int fd = client->io_watcher.fd;
    std::cout << "accept: type=" << client->type
              << ", fd=" << fd << "\n";

    if (client->type == UV_TCP)
    {
      std::cout << "got tcp fd "<<  fd<< "\n";
    }

    // register the stream before beginning read operations
    myIOLoop->add_passive_handle( ioh );

  //   // NOTE: registration of read event moved into the handle

  //   // // new client is accepted, identified via its stream handle ... need to put
  //   // // in place a lot more session tracking here.
  //   // uv_read_start((uv_stream_t *) client, alloc_buffer, io_on_read);
  }
  else
  {
    uv_close((uv_handle_t *) client, NULL);
  }

}


static void __io_async_cb(uv_async_t* handle)
{
  IOLoop* p = static_cast<IOLoop*>( handle->data );
  p->on_async();
}

static void __io_on_timer(uv_timer_t* handle)
{
  IOLoop* p = static_cast<IOLoop*>( handle->data );
  p->on_timer();
}

IOLoop::IOLoop(Logger * logptr,
               AsyncCallback timer_cb,
               AsyncCallback async_cb)
  : __logptr( logptr),
    m_uv_loop( new uv_loop_t() ),
    m_timer( new uv_timer_t() ),
    m_continue_loop(true),
    m_timer_cb ( timer_cb ),
    m_async_cb ( async_cb )
{
  uv_loop_init(m_uv_loop);
  m_uv_loop->data = this;

  // set up the async handler
  uv_async_init(m_uv_loop, &m_async, __io_async_cb);
  m_async.data = this;

  // set up timer

  uv_timer_init(m_uv_loop, m_timer.get());
  m_timer->data = this;
  uv_timer_start(m_timer.get(), __io_on_timer, 10, 10);

  // TODO: I prefer to not have to do this.  Need to review what is the correct
  // policy for handling sigpipe using libuv.
  //  signal(SIGPIPE, SIG_IGN);
}

void IOLoop::stop()
{
  m_continue_loop = false;
  async_send();
  m_thread.join();
}

IOLoop::~IOLoop()
{
  // TODO: in here, should I at least try to give the sockets once last attempt to gracefull close?

  if (m_continue_loop) stop();

  for (auto & item : m_handles) delete item;

  m_handles.clear();

  // uv_loop kept as raw pointer, because need to pass to several uv functions
  uv_loop_close(m_uv_loop);<commit_msg>define interval as macro<commit_after>#include "IOLoop.h"

#include "Logger.h"
#include "IOHandle.h"

#include <iostream>

#include <unistd.h>

#define SYSTEM_HEARTBEAT_MS 5000

namespace XXX {



/* Called on the IO thread when a ioreq attempt has completed */
static void __on_connect(uv_connect_t* __req, int status )
{
  std::unique_ptr<uv_connect_t> connect_req(__req);

  // IOLoop has set itself as the uv_loop data member
  IOLoop * ioloop = static_cast<IOLoop* >(__req->handle->loop->data);

  // get the user object
  std::unique_ptr<io_request> ioreq  (( io_request*) connect_req->data );
  Logger * __logptr = ioreq->logptr;

  if (status < 0)
  {
    // TODO: question for libuv: is there a way to get the errno ?
    _WARN_( "connect error, " <<  uv_strerror(status) );

    ioloop->add_active_handle(nullptr, status, ioreq.get());

    delete ioreq->tcp_connect;
  }
  else
  {

    IOHandle* ioh = new IOHandle( (uv_stream_t *) ioreq->tcp_connect,
                                  ioloop );
    ioloop->add_active_handle(ioh, 0, ioreq.get());
  }

}

static void __on_tcp_connect(uv_stream_t* server, int status)
{
  // IOLoop has set itself as the uv_loop data member
  IOLoop * myIOLoop = static_cast<IOLoop* >(server->loop->data);

  if (status < 0)
  {
    // TODO: logging
    fprintf(stderr, "New connection error %s\n", uv_strerror(status));
    return;
  }

  uv_tcp_t *client = new uv_tcp_t();
  uv_tcp_init(myIOLoop->uv_loop(), client);

  if (uv_accept(server, (uv_stream_t *) client) == 0)
  {
    IOHandle* ioh = new IOHandle( (uv_stream_t *) client, myIOLoop);

    int fd = client->io_watcher.fd;
    std::cout << "accept: type=" << client->type
              << ", fd=" << fd << "\n";

    if (client->type == UV_TCP)
    {
      std::cout << "got tcp fd "<<  fd<< "\n";
    }

    // register the stream before beginning read operations
    myIOLoop->add_passive_handle( ioh );

  //   // NOTE: registration of read event moved into the handle

  //   // // new client is accepted, identified via its stream handle ... need to put
  //   // // in place a lot more session tracking here.
  //   // uv_read_start((uv_stream_t *) client, alloc_buffer, io_on_read);
  }
  else
  {
    uv_close((uv_handle_t *) client, NULL);
  }

}


static void __io_async_cb(uv_async_t* handle)
{
  IOLoop* p = static_cast<IOLoop*>( handle->data );
  p->on_async();
}

static void __io_on_timer(uv_timer_t* handle)
{
  IOLoop* p = static_cast<IOLoop*>( handle->data );
  p->on_timer();
}

IOLoop::IOLoop(Logger * logptr,
               AsyncCallback timer_cb,
               AsyncCallback async_cb)
  : __logptr( logptr),
    m_uv_loop( new uv_loop_t() ),
    m_timer( new uv_timer_t() ),
    m_continue_loop(true),
    m_timer_cb ( timer_cb ),
    m_async_cb ( async_cb )
{
  uv_loop_init(m_uv_loop);
  m_uv_loop->data = this;

  // set up the async handler
  uv_async_init(m_uv_loop, &m_async, __io_async_cb);
  m_async.data = this;

  // set up timer

  uv_timer_init(m_uv_loop, m_timer.get());
  m_timer->data = this;
  uv_timer_start(m_timer.get(), __io_on_timer, SYSTEM_HEARTBEAT_MS, SYSTEM_HEARTBEAT_MS);

  // TODO: I prefer to not have to do this.  Need to review what is the correct
  // policy for handling sigpipe using libuv.
  //  signal(SIGPIPE, SIG_IGN);
}

void IOLoop::stop()
{
  m_continue_loop = false;
  async_send();
  m_thread.join();
}

IOLoop::~IOLoop()
{
  // TODO: in here, should I at least try to give the sockets once last attempt to gracefull close?

  if (m_continue_loop) stop();

  for (auto & item : m_handles) delete item;

  m_handles.clear();

  // uv_loop kept as raw pointer, because need to pass to several uv functions
  uv_loop_close(m_uv_loop);
{'fraction_non_alphanumeric': 0.13059497816593887, 'fraction_numerical': 0.002183406113537118, 'mean_word_length': 3.185608223872073, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4678334', 'n_tokens_mistral': 2773, 'n_tokens_neox': 2658, 'n_words': 681}