<commit_before>/* Rapicorn
 * Copyright (C) 2006-2007 <NAME>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * A copy of the GNU Lesser General Public License should ship along
 * with this library; if not, see http://www.gnu.org/copyleft/.
 */
#include "loop.hh"
#include <sys/poll.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <glib.h> // FIXME: GTimeVal
#include <algorithm>
#include <cstring>

namespace Rapicorn {

/* --- assert poll constants --- */
RAPICORN_STATIC_ASSERT (PollFD::IN     == POLLIN);
RAPICORN_STATIC_ASSERT (PollFD::PRI    == POLLPRI);
RAPICORN_STATIC_ASSERT (PollFD::OUT    == POLLOUT);
RAPICORN_STATIC_ASSERT (PollFD::RDNORM == POLLRDNORM);
RAPICORN_STATIC_ASSERT (PollFD::RDBAND == POLLRDBAND);
RAPICORN_STATIC_ASSERT (PollFD::WRNORM == POLLWRNORM);
RAPICORN_STATIC_ASSERT (PollFD::WRBAND == POLLWRBAND);
RAPICORN_STATIC_ASSERT (PollFD::ERR    == POLLERR);
RAPICORN_STATIC_ASSERT (PollFD::HUP    == POLLHUP);
RAPICORN_STATIC_ASSERT (PollFD::NVAL   == POLLNVAL);

/* --- assert pollfd struct --- */
RAPICORN_STATIC_ASSERT (sizeof   (PollFD)               == sizeof   (struct pollfd));
RAPICORN_STATIC_ASSERT (offsetof (PollFD, fd)           == offsetof (struct pollfd, fd));
RAPICORN_STATIC_ASSERT (sizeof (((PollFD*) 0)->fd)      == sizeof (((struct pollfd*) 0)->fd));
RAPICORN_STATIC_ASSERT (offsetof (PollFD, events)       == offsetof (struct pollfd, events));
RAPICORN_STATIC_ASSERT (sizeof (((PollFD*) 0)->events)  == sizeof (((struct pollfd*) 0)->events));
RAPICORN_STATIC_ASSERT (offsetof (PollFD, revents)      == offsetof (struct pollfd, revents));
RAPICORN_STATIC_ASSERT (sizeof (((PollFD*) 0)->revents) == sizeof (((struct pollfd*) 0)->revents));


enum {
  UNCHECKED             = 0,
  PREPARED,
  NEEDS_DISPATCH,
  _DUMMY = UINT_MAX
};

/* --- EventLoop --- */
EventLoop::EventLoop ()
{}

EventLoop::~EventLoop ()
{}

uint64
EventLoop::get_current_time_usecs ()
{
  GTimeVal current_time;
  g_get_current_time (&current_time);
  return current_time.tv_sec * 1000000ULL + current_time.tv_usec;
}

void
EventLoop::remove (uint   id)
{
  if (!try_remove (id))
    g_warning ("%s: failed to remove loop source: %u", G_STRFUNC, id);
}

class EventLoopImpl;
static vector<EventLoopImpl*> rapicorn_main_loops;
static int                    rapicorn_wakeups[2] = { -1, -1 }; // [0] = readable; [1] = writable

/* --- EventLoopImpl --- */
class EventLoopImpl : public virtual EventLoop {
  typedef std::list<Source*>        SourceList;
  typedef std::map<int, SourceList> SourceListMap;
  Mutex          m_mutex;
public:                                 // EventLoop::iterate_loops
  int            m_max_priority;        // EventLoop::iterate_loops
  bool           m_must_dispatch;       // EventLoop::iterate_loops
private:
  uint           m_counter;             // FIXME: need to handle wrapping at some point
  SourceListMap  m_sources;
  vector<PollFD> m_pfds;
  enum {
    WILL_CHECK,
    WILL_DISPATCH
  };
  uint8          m_istate;
  inline Source*
  find_first()
  {
    for (SourceListMap::iterator it = m_sources.begin(); it != m_sources.end(); it++)
      {
        SourceList slist = (*it).second;
        for (SourceList::iterator lit = slist.begin(); lit != slist.end(); lit++)
          return *lit;
      }
    return NULL;
  }
  inline Source*
  find_source (uint id)
  {
    for (SourceListMap::iterator it = m_sources.begin(); it != m_sources.end(); it++)
      {
        SourceList slist = (*it).second;
        for (SourceList::iterator lit = slist.begin(); lit != slist.end(); lit++)
          if (id == (*lit)->m_id)
            return *lit;
      }
    return NULL;
  }
  void
  real_remove_Lm (Source *source)
  {
    assert (source->m_main_loop == this);
    SourceList &slist = m_sources[source->m_priority];
    slist.remove (source);
    if (slist.empty())
      m_sources.erase (m_sources.find (source->m_priority));
    bool needs_destroy = source->m_main_loop == this;
    source->m_main_loop = NULL;
    source->m_loop_state = UNCHECKED;
    m_mutex.unlock();
    if (needs_destroy)
      source->destroy();
    unref (source);
    m_mutex.lock();
  }
  void
  remove_loop_M ()
  {
    vector<EventLoopImpl*>::iterator it = find (rapicorn_main_loops.begin(), rapicorn_main_loops.end(), this);
    if (it != rapicorn_main_loops.end())
      rapicorn_main_loops.erase (it);
  }
public:
  EventLoopImpl () :
    m_max_priority (INT_MAX),
    m_must_dispatch (0),
    m_counter (1),
    m_istate (WILL_CHECK)
  {
    AutoLocker locker (m_mutex);
    return_if_fail (find (rapicorn_main_loops.begin(), rapicorn_main_loops.end(), this) == rapicorn_main_loops.end());
    assert (rapicorn_thread_entered());
    rapicorn_main_loops.push_back (this);
  }
  ~EventLoopImpl()
  {
    kill_sources();
    AutoLocker locker (m_mutex);
    remove_loop_M();
  }
  virtual bool prepare_sources  (int                  &max_priority,
                                 vector<PollFD>       &pfds,
                                 int64                &timeout_usecs);
  virtual bool check_sources    (const int             max_priority,
                                 const vector<PollFD> &pfds);
  virtual void dispatch_sources (const int             max_priority);
  void
  wakeup_L (void)
  {
    if (rapicorn_wakeups[1] >= 0)
      {
        char w = 'w';
        int err;
        do
          err = write (rapicorn_wakeups[1], &w, 1);
        while (err < 0 && (errno == EINTR || errno == EAGAIN));
      }
  }
  virtual void
  wakeup (void)
  {
    AutoLocker locker (m_mutex);
    wakeup_L();
  }
  virtual uint
  add_source (Source   *source,
              int       priority)
  {
    AutoLocker locker (m_mutex);
    return_val_if_fail (source->m_main_loop == NULL, 0);
    ref_sink (source);
    source->m_main_loop = this;
    source->m_id = m_counter++;
    if (!source->m_id)
      error ("EventLoop::m_counter overflow, please report");
    source->m_loop_state = UNCHECKED;
    source->m_priority = priority;
    std::list<Source*> &slist = m_sources[priority];
    slist.push_back (source);
    wakeup_L();
    return source->m_id;
  }<commit_msg>loop.cc: added FIXME<commit_after>/* Rapicorn
 * Copyright (C) 2006-2007 <NAME>ik
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * A copy of the GNU Lesser General Public License should ship along
 * with this library; if not, see http://www.gnu.org/copyleft/.
 */
#include "loop.hh"
#include <sys/poll.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>
#include <glib.h> // FIXME: GTimeVal
#include <algorithm>
#include <cstring>

namespace Rapicorn {

/* --- assert poll constants --- */
RAPICORN_STATIC_ASSERT (PollFD::IN     == POLLIN);
RAPICORN_STATIC_ASSERT (PollFD::PRI    == POLLPRI);
RAPICORN_STATIC_ASSERT (PollFD::OUT    == POLLOUT);
RAPICORN_STATIC_ASSERT (PollFD::RDNORM == POLLRDNORM);
RAPICORN_STATIC_ASSERT (PollFD::RDBAND == POLLRDBAND);
RAPICORN_STATIC_ASSERT (PollFD::WRNORM == POLLWRNORM);
RAPICORN_STATIC_ASSERT (PollFD::WRBAND == POLLWRBAND);
RAPICORN_STATIC_ASSERT (PollFD::ERR    == POLLERR);
RAPICORN_STATIC_ASSERT (PollFD::HUP    == POLLHUP);
RAPICORN_STATIC_ASSERT (PollFD::NVAL   == POLLNVAL);

/* --- assert pollfd struct --- */
RAPICORN_STATIC_ASSERT (sizeof   (PollFD)               == sizeof   (struct pollfd));
RAPICORN_STATIC_ASSERT (offsetof (PollFD, fd)           == offsetof (struct pollfd, fd));
RAPICORN_STATIC_ASSERT (sizeof (((PollFD*) 0)->fd)      == sizeof (((struct pollfd*) 0)->fd));
RAPICORN_STATIC_ASSERT (offsetof (PollFD, events)       == offsetof (struct pollfd, events));
RAPICORN_STATIC_ASSERT (sizeof (((PollFD*) 0)->events)  == sizeof (((struct pollfd*) 0)->events));
RAPICORN_STATIC_ASSERT (offsetof (PollFD, revents)      == offsetof (struct pollfd, revents));
RAPICORN_STATIC_ASSERT (sizeof (((PollFD*) 0)->revents) == sizeof (((struct pollfd*) 0)->revents));


enum {
  UNCHECKED             = 0,
  PREPARED,
  NEEDS_DISPATCH,
  _DUMMY = UINT_MAX
};

/* --- EventLoop --- */
EventLoop::EventLoop ()
{}

EventLoop::~EventLoop ()
{}

uint64
EventLoop::get_current_time_usecs ()
{
  GTimeVal current_time;
  g_get_current_time (&current_time);
  return current_time.tv_sec * 1000000ULL + current_time.tv_usec;
}

void
EventLoop::remove (uint   id)
{
  if (!try_remove (id))
    g_warning ("%s: failed to remove loop source: %u", G_STRFUNC, id);
}

class EventLoopImpl;
static vector<EventLoopImpl*> rapicorn_main_loops;
static int                    rapicorn_wakeups[2] = { -1, -1 }; // [0] = readable; [1] = writable

/* --- EventLoopImpl --- */
class EventLoopImpl : public virtual EventLoop {
  typedef std::list<Source*>        SourceList;
  typedef std::map<int, SourceList> SourceListMap;
  Mutex          m_mutex;
public:                                 // EventLoop::iterate_loops
  int            m_max_priority;        // EventLoop::iterate_loops
  bool           m_must_dispatch;       // EventLoop::iterate_loops
private:
  uint           m_counter;             // FIXME: need to handle wrapping at some point
  SourceListMap  m_sources;
  vector<PollFD> m_pfds;
  enum {
    WILL_CHECK,
    WILL_DISPATCH
  };
  uint8          m_istate;
  inline Source*
  find_first()
  {
    for (SourceListMap::iterator it = m_sources.begin(); it != m_sources.end(); it++)
      {
        SourceList slist = (*it).second;
        for (SourceList::iterator lit = slist.begin(); lit != slist.end(); lit++)
          return *lit;
      }
    return NULL;
  }
  inline Source*
  find_source (uint id)
  {
    for (SourceListMap::iterator it = m_sources.begin(); it != m_sources.end(); it++)
      {
        SourceList slist = (*it).second;
        for (SourceList::iterator lit = slist.begin(); lit != slist.end(); lit++)
          if (id == (*lit)->m_id)
            return *lit;
      }
    return NULL;
  }
  void
  real_remove_Lm (Source *source)
  {
    assert (source->m_main_loop == this);
    SourceList &slist = m_sources[source->m_priority];
    slist.remove (source);
    if (slist.empty())
      m_sources.erase (m_sources.find (source->m_priority));
    bool needs_destroy = source->m_main_loop == this;
    source->m_main_loop = NULL;
    source->m_loop_state = UNCHECKED;
    m_mutex.unlock();
    if (needs_destroy)
      source->destroy();
    unref (source);
    m_mutex.lock();
  }
  void
  remove_loop_M ()
  {
    vector<EventLoopImpl*>::iterator it = find (rapicorn_main_loops.begin(), rapicorn_main_loops.end(), this);
    if (it != rapicorn_main_loops.end())
      rapicorn_main_loops.erase (it);
  }
public:
  EventLoopImpl () :
    m_max_priority (INT_MAX),
    m_must_dispatch (0),
    m_counter (1),
    m_istate (WILL_CHECK)
  {
    AutoLocker locker (m_mutex);
    return_if_fail (find (rapicorn_main_loops.begin(), rapicorn_main_loops.end(), this) == rapicorn_main_loops.end());
    assert (rapicorn_thread_entered());
    rapicorn_main_loops.push_back (this);
  }
  ~EventLoopImpl()
  {
    kill_sources();
    AutoLocker locker (m_mutex);
    remove_loop_M();
  }
  virtual bool prepare_sources  (int                  &max_priority,
                                 vector<PollFD>       &pfds,
                                 int64                &timeout_usecs);
  virtual bool check_sources    (const int             max_priority,
                                 const vector<PollFD> &pfds);
  virtual void dispatch_sources (const int             max_priority);
  void
  wakeup_L (void)
  {
    if (rapicorn_wakeups[1] >= 0)
      {
        char w = 'w';
        int err;
        do
          err = write (rapicorn_wakeups[1], &w, 1);
        while (err < 0 && (errno == EINTR || errno == EAGAIN));
      }
  }
  virtual void
  wakeup (void)
  {
    AutoLocker locker (m_mutex);
    wakeup_L();
  }
  virtual uint
  add_source (Source   *source,
              int       priority)
  {
    AutoLocker locker (m_mutex);
    return_val_if_fail (source->m_main_loop == NULL, 0);
    ref_sink (source);
    source->m_main_loop = this;
    source->m_id = m_counter++;
    if (!source->m_id) // FIXME: need simple counter/id allocation scheme
      error ("EventLoop::m_counter overflow, please report");
    source->m_loop_state = UNCHECKED;
    source->m_priority = priority;
    std::list<Source*> &slist = m_sources[priority];
    slist.push_back (source);
    wakeup_L();
    return source->m_id;
  }
{'fraction_non_alphanumeric': 0.12723987351302515, 'fraction_numerical': 0.006324348742659238, 'mean_word_length': 2.9532738095238096, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5844998', 'n_tokens_mistral': 4613, 'n_tokens_neox': 4435, 'n_words': 1072}