scheduler.h

Go to the documentation of this file.

// Copyright 2011 Google Inc.

#ifndef PAGESPEED_KERNEL_THREAD_SCHEDULER_H_
#define PAGESPEED_KERNEL_THREAD_SCHEDULER_H_

#include <set>

#include "pagespeed/kernel/base/basictypes.h"
#include "pagespeed/kernel/base/condvar.h"
#include "pagespeed/kernel/base/function.h"
#include "pagespeed/kernel/base/scoped_ptr.h"
#include "pagespeed/kernel/base/thread_annotations.h"
#include "pagespeed/kernel/base/thread_system.h"
#include "pagespeed/kernel/base/timer.h"
#include "pagespeed/kernel/thread/queued_worker_pool.h"

#define SCHEDULER_CANCEL_OUTSTANDING_ALARMS_ON_DESTRUCTION 0

namespace net_instaweb {

class Scheduler {
 public:
  class Alarm;

  class Sequence;

  struct CompareAlarms {
    bool operator()(const Alarm* a, const Alarm* b) const;
  };

  Scheduler(ThreadSystem* thread_system, Timer* timer);
  virtual ~Scheduler();

  ThreadSystem::CondvarCapableMutex* mutex() LOCK_RETURNED(mutex_) {
    return mutex_.get();
  }

  void DCheckLocked() EXCLUSIVE_LOCKS_REQUIRED(mutex()) {
    mutex_->DCheckLocked();
  }


  void BlockingTimedWaitMs(int64 timeout_ms) EXCLUSIVE_LOCKS_REQUIRED(mutex()) {
    BlockingTimedWaitUs(timeout_ms * Timer::kMsUs);
  }
  void BlockingTimedWaitUs(int64 timeout_us) EXCLUSIVE_LOCKS_REQUIRED(mutex());

  void TimedWaitMs(int64 timeout_ms, Function* callback)
      EXCLUSIVE_LOCKS_REQUIRED(mutex());

  void Signal() EXCLUSIVE_LOCKS_REQUIRED(mutex());


  Alarm* AddAlarmAtUs(int64 wakeup_time_us, Function* callback)
      LOCKS_EXCLUDED(mutex());

  Alarm* AddAlarmAtUsMutexHeld(int64 wakeup_time_us, Function* callback)
      EXCLUSIVE_LOCKS_REQUIRED(mutex());

  bool CancelAlarm(Alarm* alarm) EXCLUSIVE_LOCKS_REQUIRED(mutex());

  bool ProcessAlarmsOrWaitUs(int64 timeout_us)
      EXCLUSIVE_LOCKS_REQUIRED(mutex());

  Timer* timer() { return timer_; }

  ThreadSystem* thread_system() { return thread_system_; }

  void Wakeup() { condvar_->Broadcast(); }

  virtual void RegisterWorker(QueuedWorkerPool::Sequence* w);
  virtual void UnregisterWorker(QueuedWorkerPool::Sequence* w);

  int64 RunAlarms(bool* ran_alarms) EXCLUSIVE_LOCKS_REQUIRED(mutex());

  Sequence* NewSequence();

 protected:
  virtual void AwaitWakeupUntilUs(int64 wakeup_time_us);

  bool running_waiting_alarms() const { return running_waiting_alarms_; }

 private:
  class CondVarTimeout;
  class CondVarCallbackTimeout;
  friend class SchedulerTest;

  typedef std::set<Alarm*, CompareAlarms> AlarmSet;

  void InsertAlarmAtUsMutexHeld(int64 wakeup_time_us,
                                bool broadcast_on_wakeup_change,
                                Alarm* alarm);
  void CancelWaiting(Alarm* alarm);
  bool NoPendingAlarms();

  ThreadSystem* thread_system_;
  Timer* timer_;
  scoped_ptr<ThreadSystem::CondvarCapableMutex> mutex_;
  scoped_ptr<ThreadSystem::Condvar> condvar_;
  uint32 index_; 
  AlarmSet outstanding_alarms_; 
  int64 signal_count_; 
  AlarmSet waiting_alarms_; 
  bool running_waiting_alarms_; 
  
};

class SchedulerBlockingFunction : public Function {
 public:
  explicit SchedulerBlockingFunction(Scheduler* scheduler);
  virtual ~SchedulerBlockingFunction();
  virtual void Run();
  virtual void Cancel();
  bool Block();
 private:
  Scheduler* scheduler_;
  bool success_;
  bool done_; 
  
};

}  

#endif