root/trunk/whisperlib/net/base/selector.cc

// Copyright (c) 2009, Whispersoft s.r.l.
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
// * Neither the name of Whispersoft s.r.l. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// Authors: Cosmin Tudorache & Catalin Popescu

#include "net/base/selector.h"

#include <pthread.h>
#include <unistd.h>
#include <fcntl.h>

#include "common/base/log.h"
#include "common/base/errno.h"
#include "common/base/timer.h"

#include "net/base/selectable.h"

//////////////////////////////////////////////////////////////////////

DEFINE_bool(selector_high_alarm_precission,
            false,
            "Loose some CPU time and gain that extra milisecond precission "
            "for selector alarms..");

DEFINE_int32(debug_check_long_callbacks_ms,
             500,
             "If greater than zero, we check (in debug mode only !) "
             "that processing functions, callbacks and alarm functions take "
             "less then this amount of time, in miliseconds");

//////////////////////////////////////////////////////////////////////

namespace net {

Selector::Selector()
  : tid_(0),
    should_end_(false),
    epfd_(INVALID_FD_VALUE),
    now_(timer::TicksMsec()) {
  epfd_ = epoll_create(10);
  CHECK_GE(epfd_, 0) << "epoll_create() failed: "
                     << GetLastSystemErrorDescription();
  CHECK(!pipe(signal_pipe_)) <<  "pipe() failed:  "
                             << GetLastSystemErrorDescription();
  for ( int i = 0; i < NUMBEROF(signal_pipe_); ++i ) {
    const int flags = fcntl(signal_pipe_[i], F_GETFL, 0);
    CHECK_GE(flags, 0) << " fcntl fail: " << GetLastSystemErrorDescription();
    CHECK_GE(fcntl(signal_pipe_[i], F_SETFL, flags | O_NONBLOCK), 0)
      << "fcntl fail: " << GetLastSystemErrorDescription();
  }
  CHECK(EpollCtlAdd(signal_pipe_[0], NULL, EPOLLIN | EPOLLRDHUP | EPOLLHUP));
}

Selector::~Selector() {
  CHECK_EQ(tid_, 0);
  CHECK(registered_.empty());
}

void Selector::Loop() {
  CHECK_EQ(tid_, 0) << "Loop already started -- bad !";
  should_end_ = false;
  tid_ = pthread_self();
  LOG_INFO << "Starting selector loop";
  while ( !should_end_ ) {
    struct epoll_event event = { 0, };
    int32 to_sleep = kStandardWakeUpTimeMs;
    if ( FLAGS_selector_high_alarm_precission ) {
      now_ = timer::TicksMsec();
    }
    if ( !alarms_.empty() ) {
      if ( alarms_.begin()->first < now_ + to_sleep ) {
        to_sleep = alarms_.begin()->first - now_;
      }
    }
    const int num_events = epoll_wait(epfd_, &event, 1, to_sleep);
    if ( num_events == -1 && errno != EINTR ) {
      LOG_ERROR << "epoll_wait() error: " << GetLastSystemErrorDescription();
      break;
    }
    now_ = timer::TicksMsec();
    if ( num_events > 0 ) {
      DCHECK_EQ(num_events, 1);
      Selectable* const s = reinterpret_cast<Selectable *>(event.data.ptr);
#ifdef _DEBUG
      const int64 processing_begin =
          FLAGS_debug_check_long_callbacks_ms > 0 ? timer::TicksMsec() : 0;
#endif
      if ( s != NULL ) {
        // During HandleXEvent the obj may be closed loosing so track of
        // it's fd value.
        bool keep_processing = true;
        if ( event.events & (EPOLLERR | EPOLLHUP | EPOLLRDHUP) ) {
          keep_processing = s->HandleErrorEvent(event.events) &&
                            s->GetFd() != INVALID_FD_VALUE;
        }
        if ( keep_processing && (event.events & (EPOLLIN | EPOLLPRI)) ) {
          keep_processing = s->HandleReadEvent(event.events) &&
                            s->GetFd() != INVALID_FD_VALUE;
        }
        if ( keep_processing && (event.events & EPOLLOUT) ) {
          s->HandleWriteEvent(event.events);
        }
      }  // else was probably a wake signal..
#ifdef _DEBUG
      if ( FLAGS_debug_check_long_callbacks_ms > 0 ) {
        const int64 processing_end = timer::TicksMsec();
        if ( processing_end - processing_begin >
             FLAGS_debug_check_long_callbacks_ms ) {
          LOG_ERROR << " ====>> Unexpectedly long event processing: "
                    << " event: " << event.events
                    << " time spent:  "
                    << processing_end - processing_begin;
        }
      }
#endif
    }  // else, was probably a timeout
    // Runs some closures..
    bool are_some_closures = false;
    if ( FLAGS_selector_high_alarm_precission ) {
      now_ = timer::TicksMsec();
    }
    do {
      mutex_.Lock();
      are_some_closures = !to_run_.empty();
      mutex_.Unlock();
      RunAllClosures();
    } while ( are_some_closures );
    // Run the alarms
    if ( FLAGS_selector_high_alarm_precission ) {
      now_ = timer::TicksMsec();
    }
    while ( !alarms_.empty() ) {
      if ( alarms_.begin()->first <= now_ ) {
        Closure* closure = alarms_.begin()->second;
        const bool erased = reverse_alarms_.erase(closure);
        alarms_.erase(alarms_.begin());
#ifdef _DEBUG
      const int64 processing_begin =
          FLAGS_debug_check_long_callbacks_ms > 0 ? timer::TicksMsec() : 0;
#endif
      if ( erased ) {
        closure->Run();
      } else {
        LOG_ERROR << " Strange alarm mismatching: " << closure;
      }
#ifdef _DEBUG
      if ( FLAGS_debug_check_long_callbacks_ms > 0 ) {
        const int64 processing_end = timer::TicksMsec();
        if ( processing_end - processing_begin >
             FLAGS_debug_check_long_callbacks_ms ) {
          LOG_ERROR << " ====>> Unexpectedly long alarm processing: "
                    << " callback: " << closure
                    << " time spent:  "
                    << processing_end - processing_begin;
        }
      }
#endif
      } else {
        break;
      }
    }
  }

  LOG_INFO << "Closing all the active connections in the selector...";
  CleanAndCloseAll();

  // Run all the remaining closures...
  bool are_some_closures = false;
  int32 are_some_closures_count = 0;
  do {
    LOG_INFO << "Running the selector closures on shutdown, iteration "
             << are_some_closures_count++;
    RunAllClosures();
    mutex_.Lock();
    are_some_closures = !to_run_.empty();
    mutex_.Unlock();
  } while ( are_some_closures );

  // cleanup epoll
  ::close(epfd_);
  epfd_ = INVALID_FD_VALUE;
  tid_ = 0;

  LOG_INFO << "Selector loop terminated";
}

void Selector::MakeLoopExit() {
  CHECK(IsInSelectThread());
  should_end_ = true;
}

bool Selector::IsInSelectThread() const {
  return tid_ == pthread_self();
}

bool Selector::RunInSelectLoop(Closure* callback) {
  #ifdef _DEBUG
  callback->set_selector_registered(true);
  #endif
  mutex_.Lock();
  to_run_.push_back(callback);
  mutex_.Unlock();
  if ( !IsInSelectThread() ) {
    SendWakeSignal();
  }
  return true;
}
int64 Selector::RegisterAlarm(Closure* callback, int64 timeout_in_ms) {
  CHECK(callback != NULL);
  CHECK(reverse_alarms_.find(callback) == reverse_alarms_.end())
    << "Same alarm added multiple times -- BAD";
  // const int64 now = timer::TicksMsec();
  const int64 wake_up_time = now_ + timeout_in_ms;
  CHECK(timeout_in_ms < 0 || timeout_in_ms <= wake_up_time)
    << "Overflow, timeout_in_ms: " << timeout_in_ms << " is too big";
  alarms_.insert(make_pair(wake_up_time, callback));
  reverse_alarms_[callback] = wake_up_time;

  // We do not need to wake .. we are in the select loop :)
  return wake_up_time;
}
bool Selector::UnregisterAlarm(Closure* callback) {
  ReverseAlarmsMap::iterator it = reverse_alarms_.find(callback);
  if ( it == reverse_alarms_.end() ) {
    return false;
  }
  alarms_.erase(make_pair(it->second, callback));
  reverse_alarms_.erase(it);
  return true;
}

bool Selector::ReregisterAlarm(Closure* callback, int64 timeout_in_ms) {
  CHECK(callback != NULL);
  ReverseAlarmsMap::iterator it = reverse_alarms_.find(callback);
  if ( it != reverse_alarms_.end() ) {
    alarms_.erase(make_pair(it->second, callback));
    const int64 wake_up_time = now_ + timeout_in_ms;
    alarms_.insert(make_pair(wake_up_time, callback));
    it->second = wake_up_time;
    return true;
  } else {
    RegisterAlarm(callback, timeout_in_ms);
    return false;
  }
}

//////////////////////////////////////////////////////////////////////

void Selector::CleanAndCloseAll() {
  // It is some discussion, whether to do some CHECK if connections are
  // left at this point or to close them or to just skip it. The ideea is
  // that we preffer forcing a close on them for the server case and also
  // client connections when we end a program.
  if ( !registered_.empty() ) {
    DLOG_DEBUG << "Select loop ended with " << registered_.size()
              << " registered connections.";
    // We just close the fd and care about nothing ..
    while ( !registered_.empty() ) {
      (*registered_.begin())->Close();
    }
  }
}

void Selector::RunAllClosures() {
  char buffer[1024];
  int cb;
  while ( (cb = ::read(signal_pipe_[0], buffer, sizeof(buffer))) > 0 ) {
    DLOG(10) << " Cleaned some " << cb << " bytes from signal pipe.";
  }
  mutex_.Lock();
  list<Closure*> to_run(to_run_.size());
  copy(to_run_.begin(), to_run_.end(), to_run.begin());
  to_run_.clear();
  mutex_.Unlock();
  while ( !to_run.empty() ) {
#ifdef _DEBUG
    const int64 processing_begin =
        FLAGS_debug_check_long_callbacks_ms > 0 ? timer::TicksMsec() : 0;
    to_run.front()->set_selector_registered(false);
#endif
    to_run.front()->Run();
#ifdef _DEBUG
    if ( FLAGS_debug_check_long_callbacks_ms > 0 ) {
      const int64 processing_end = timer::TicksMsec();
      if ( processing_end - processing_begin >
           FLAGS_debug_check_long_callbacks_ms ) {
        LOG_ERROR << " ====>> Unexpectedly long callback processing: "
                  << " callback: " << to_run.front()
                  << " time spent:  "
                  << processing_end - processing_begin;
      }
    }
#endif
    to_run.pop_front();
  }
}

void Selector::SendWakeSignal() {
  int8 byte = 0;
  if ( sizeof(byte) != ::write(signal_pipe_[1], &byte, sizeof(byte)) ) {
    LOG_ERROR << "Error writing a wake-up byte in selector signal pipe";
  }
}

//////////////////////////////////////////////////////////////////////

bool Selector::Register(Selectable* s) {
  DCHECK(IsInSelectThread() || tid_ == 0);
  CHECK(s->selector() == this);
  const int fd = s->GetFd();

  SelectableSet::const_iterator it = registered_.find(s);
  if ( it != registered_.end() ) {
    // selectable obj already registered
    CHECK_EQ((*it)->GetFd(), fd);
    return true;
  }
  // Insert in the local set of registered objs
  registered_.insert(s);

  // Insert in epoll
  if ( fd != INVALID_FD_VALUE ) {
    if ( !EpollCtlAdd(fd, s, s->GetDesiredEvents()) ) {
      return false;
    }
  }
  return true;
}

void Selector::Unregister(Selectable* s) {
  DCHECK(IsInSelectThread() || tid_ == 0);
  CHECK(s->selector() == this);
  const int fd = s->GetFd();

  const SelectableSet::iterator it = registered_.find(s);
  DCHECK(it != registered_.end()) << " Cannot unregister : " << fd;

  // remove from epoll and internal lists
  if ( fd != INVALID_FD_VALUE ) {
    EpollCtlDel(fd);
  }
  registered_.erase(it);
}

//////////////////////////////////////////////////////////////////////

bool Selector::EpollCtlAdd(int fd, void* user_data, int events) {
  epoll_event event = { events, };
  event.data.ptr = user_data;

  DLOG_DEBUG << "  Adding to epoll: " << fd;
  if ( epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &event) < 0 ) {
    LOG_ERROR  << "System error on epoll_ctl: "
               << GetLastSystemErrorDescription()
               << " for events: " << hex << events << dec;
    return false;
  }
  return true;
}

bool Selector::EpollCtlDel(int fd) {
  epoll_event event = { 0, };
  DLOG_DEBUG << "  Removing from epoll: " << fd;
  if ( epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, &event) < 0 ) {
    LOG_ERROR  << "System error on epoll_ctl: "
               << GetLastSystemErrorDescription()
               << " for fd: " << fd;
    return false;
  }
  return true;
}

void Selector::UpdateDesire(Selectable* s, bool enable, int32 desire) {
  // DCHECK(registered_.find(s) != registered_.end()) << fd;
  if ( ((s->desire_ & desire) && enable) ||
       ((~s->desire_ & desire) && !enable) ) {
    return;  // already set ..
  }
  if ( enable ) {
    s->desire_ |= desire;
  } else {
    s->desire_ &= ~desire;
  }
  const int fd = s->GetFd();
  DCHECK_NE(fd, INVALID_FD_VALUE);
  epoll_event event = { s->GetDesiredEvents(), };
  event.data.ptr = s;
  if ( epoll_ctl(epfd_, EPOLL_CTL_MOD, fd, &event) ) {
    LOG_ERROR  << "Error in epoll_ctl: "  << GetLastSystemErrorDescription();
  }
}

//////////////////////////////////////////////////////////////////////
}