queued_worker_pool.h

Go to the documentation of this file.

/*
 * Copyright 2011 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http:///www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


#ifndef PAGESPEED_KERNEL_THREAD_QUEUED_WORKER_POOL_H_
#define PAGESPEED_KERNEL_THREAD_QUEUED_WORKER_POOL_H_

#include <cstddef>  
#include <deque>
#include <set>
#include <vector>

#include "pagespeed/kernel/base/abstract_mutex.h"
#include "pagespeed/kernel/base/basictypes.h"
#include "pagespeed/kernel/base/function.h"
#include "pagespeed/kernel/base/scoped_ptr.h"
#include "pagespeed/kernel/base/string.h"
#include "pagespeed/kernel/base/string_util.h"
#include "pagespeed/kernel/base/thread_annotations.h"
#include "pagespeed/kernel/base/thread_system.h"
#include "pagespeed/kernel/thread/sequence.h"

namespace net_instaweb {

class QueuedWorker;
class Waveform;

class QueuedWorkerPool {
 public:
  static const int kNoLoadShedding = -1;

  QueuedWorkerPool(int max_workers, StringPiece thread_name_base,
                   ThreadSystem* thread_system);
  ~QueuedWorkerPool();

  class Sequence : public net_instaweb::Sequence {
   public:
    class AddFunction : public Function {
     public:
      AddFunction(net_instaweb::Sequence* sequence, Function* callback)
          : sequence_(sequence), callback_(callback) { }
      virtual ~AddFunction();

     protected:
      virtual void Run() {
        sequence_->Add(callback_);
      }
      virtual void Cancel() {
        sequence_->Add(MakeFunction(callback_, &Function::CallCancel));
      }

     private:
      net_instaweb::Sequence* sequence_;
      Function* callback_;
      
    };

    void Add(Function* function) LOCKS_EXCLUDED(sequence_mutex_);

    void set_queue_size_stat(Waveform* x) { queue_size_ = x; }

    void set_max_queue_size(size_t x) { max_queue_size_ = x; }

    void CancelPendingFunctions() LOCKS_EXCLUDED(sequence_mutex_);

   private:
    Sequence(ThreadSystem* thread_system, QueuedWorkerPool* pool);

    ~Sequence();

    void Reset();

    void WaitForShutDown() LOCKS_EXCLUDED(sequence_mutex_);

    bool InitiateShutDown() LOCKS_EXCLUDED(sequence_mutex_);

    Function* NextFunction() LOCKS_EXCLUDED(sequence_mutex_);

    bool IsBusy() EXCLUSIVE_LOCKS_REQUIRED(sequence_mutex_);

    int CancelTasksOnWorkQueue() EXCLUSIVE_LOCKS_REQUIRED(sequence_mutex_);

    void Cancel() LOCKS_EXCLUDED(sequence_mutex_);

    friend class QueuedWorkerPool;
    std::deque<Function*> work_queue_ GUARDED_BY(sequence_mutex_);
    scoped_ptr<ThreadSystem::CondvarCapableMutex> sequence_mutex_;
    QueuedWorkerPool* pool_;
    bool shutdown_ GUARDED_BY(sequence_mutex_);
    bool active_ GUARDED_BY(sequence_mutex_);
    scoped_ptr<ThreadSystem::Condvar> termination_condvar_
        GUARDED_BY(sequence_mutex_);
    Waveform* queue_size_;
    size_t max_queue_size_;

    
  };

  typedef std::set<Sequence*> SequenceSet;

  Sequence* NewSequence(); 

  void FreeSequence(Sequence* sequence);

  void ShutDown();

  void InitiateShutDown();

  void WaitForShutDownComplete();

  static bool AreBusy(const SequenceSet& sequences);

  void SetLoadSheddingThreshold(int x);

  void set_queue_size_stat(Waveform* x) { queue_size_ = x; }

 private:
  friend class Sequence;
  void Run(Sequence* sequence, QueuedWorker* worker);
  void QueueSequence(Sequence* sequence);
  Sequence* AssignWorkerToNextSequence(QueuedWorker* worker);
  void SequenceNoLongerActive(Sequence* sequence);

  ThreadSystem* thread_system_;
  scoped_ptr<AbstractMutex> mutex_;

  std::set<QueuedWorker*> active_workers_;
  std::vector<QueuedWorker*> available_workers_;

  std::vector<Sequence*> all_sequences_;
  std::deque<Sequence*> queued_sequences_;
  std::vector<Sequence*> free_sequences_;

  GoogleString thread_name_base_;

  size_t max_workers_;
  bool shutdown_;

  Waveform* queue_size_;
  int load_shedding_threshold_;

  
};

}  

#endif