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

// 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

#ifndef __NET_BASE_SELECTOR_H__
#define __NET_BASE_SELECTOR_H__

#include <sys/epoll.h>
#include <sys/types.h>
#include <list>
#include <deque>
#include <set>
#include <map>

#include <whisperlib/common/base/types.h>
#include WHISPER_HASH_MAP_HEADER

#include <whisperlib/common/base/callback.h>
#include <whisperlib/common/sync/mutex.h>
#include <whisperlib/common/sync/thread.h>

// This, it seems, is not defined in our standard includes
#ifndef EPOLLRDHUP
#define EPOLLRDHUP 0x2000
#endif

// Just a helper function

namespace net {

class Selectable;

//
// Magic arround an epoll -
//
// //////////////////
//
// NOTE:  IT IS THREAD-UNSAFE - except where noted
//
//
class Selector {
 public:
  Selector();
  ~Selector();

  // Register an I/O object for read/write/error event callbacks. By default:
  //  - read callback is enabled.
  //  - write callback is disabled.
  //  - error callback is enabled.
  // returns:
  //   success status.
  bool Register(Selectable* s);

  // Unregister a previously registered I/O object.
  //  s - the selectable object to be unregistered
  void Unregister(Selectable* s);

  // Enable/disable a certain event callback for the given selectable
  // -- Call this only from the select loop
  void EnableWriteCallback(Selectable* s, bool enable) {
    UpdateDesire(s, enable, kWantWrite);
  }
  void EnableReadCallback(Selectable* s, bool enable) {
    UpdateDesire(s, enable, kWantRead);
  }

  // Runs the main select loop
  void Loop();

  // Makes an exit from the select loop
  void MakeLoopExit();
  // Returns true if the selector is no longer in the loop (though
  // registered callbacks can still execute !!)
  bool IsExiting() const {
    return should_end_;
  }

  // Returns true if this call was made from the select server thread
  bool IsInSelectThread() const;

  //
  // Runs this closure in the select loop
  // - THIS IS SAFE TO CALL FROM ANOTHER THREAD -
  // NOTE: It is legal to add closures while the selector is shutting down..
  //       (i.e. IsExiting() == true)
  //
 private:
  template <typename T> static void GeneralAsynchronousDelete(T* ob) {
    delete ob;
  }
 public:
  bool RunInSelectLoop(Closure* callback);
  template <typename T> bool DeleteInSelectLoop(T* ob) {
    return RunInSelectLoop(
        NewCallback(&Selector::GeneralAsynchronousDelete<T>, ob));
  }

  // Functions for running in the select loop the given Closure after
  // a specified time interval

  // -- THESE ARE NOT THREAD SAFE --

  // Registers the given callback to be run after timeout_in_ms miliseconds.
  // Returns the ticks in ms when the callback will be run
  int64 RegisterAlarm(Closure* callback, int64 timeout_in_ms);
  // Cancels a previously registered alarm. Returns if the callback was found
  // as registered.
  bool UnregisterAlarm(Closure* callback);
  // Changes the timeout for the given callback to timeout_in_ms from *this*
  // moment.
  bool ReregisterAlarm(Closure* callback, int64 timeout_in_ms);

  // The current moment when the select loop was broken:
  int64 now() const { return now_; }

  // Desires of selectables
  static const int32 kWantRead  = 1;
  static const int32 kWantWrite = 2;
  static const int32 kWantError = 4;
 private:
  //  add a file descriptor in the epoll, and link to the given data pointer.
  bool EpollCtlAdd(int fd, void* user_data, int event);
  //  remove a file descriptor from the epoll
  bool EpollCtlDel(int fd);

  // helper that turns on/off fd desires in the assoiciated RegistrationData
  void UpdateDesire(Selectable* s, bool enable, int32 desire);

  // Cleans and closes the entire list of selectable objects
  void CleanAndCloseAll();

  // This runs all the functions from to_run_ (if any)
  void RunAllClosures();

  // This writes a byte in the internal pipe in order to make the
  // select loop wake up
  void SendWakeSignal();

  pthread_t tid_;       // selector's internal thread id
  bool should_end_;     // a make loop exit was called
  int epfd_;            // epoll file descriptor

  typedef set<Selectable*> SelectableSet;
  typedef list<Selectable*> SelectableList;
  // typedef multimap<int64, Closure*> AlarmsMap;
  typedef set< pair<int64, Closure*> > AlarmsSet;
  typedef hash_map<Closure*, int64> ReverseAlarmsMap;

  SelectableSet registered_;   // the set of registered I/O objects
  AlarmsSet alarms_;           // the registered alarms..
  ReverseAlarmsMap reverse_alarms_;
  // this maps from alarm to wake up time
  // -> allows us to change the deadline for
  // an alarm

  synch::Mutex mutex_;         // mutex that guards the access to
  // internal control stuff
  int signal_pipe_[2];         // these two file descriptors are for waking the
  // selector when a function needs to be executed
  // in the select loop
  deque<Closure*> to_run_;     // functions registered to be run in the select
  // loop

  int64           now_;        // the time of the last cycle exit
  // we wake up from the loop every 100 ms by default
  static const int32 kStandardWakeUpTimeMs = 100;

  DISALLOW_EVIL_CONSTRUCTORS(Selector);
};

class SelectorThread {
 public:
  SelectorThread()
      : thread_(NewCallback(this, &SelectorThread::Execution)) {
  }
  ~SelectorThread() {
    selector_.RunInSelectLoop(NewCallback(&selector_,
                                          &net::Selector::MakeLoopExit));
    thread_.Join();
  }
  void Start() {
    CHECK(thread_.SetJoinable());
    CHECK(thread_.Start());
  }

  const Selector* selector() const {
    return &selector_;
  }
  Selector* mutable_selector() {
    return &selector_;
  }

 private:
  void Execution() {
    selector_.Loop();
  }
  thread::Thread thread_;
  Selector selector_;

  DISALLOW_EVIL_CONSTRUCTORS(SelectorThread);
};
}
#endif  // __NET_BASE_SELECTOR__