src/RequestHandler.h

Go to the documentation of this file.

/*
 *   MetriVis - Metrics Visualization Application
 *
 *
 *   License notice:
 *
 *   This program is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   This program 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 General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 */


#ifndef _METRIVIS_REQUEST_HANDLER_H
#define _METRIVIS_REQUEST_HANDLER_H

#include <string>
#include <queue>
#include <stack>
#include <boost/thread/xtime.hpp>
#include <boost/bind.hpp>

#include "BaseClass.h"
#include "RequestDescription.h"
#include "lib/threadpool/threadpool.hpp"


namespace metrivis {


class RequestHandlerBase {

public:
  RequestHandlerBase();


  virtual ~RequestHandlerBase();

  virtual void PushRequest(const RequestDescription& request_description);


  virtual void Terminate();

//   virtual void
};


template <class T, int ID = -1>
class RequestHandler : public RequestHandlerBase, public BaseClass {

public:

  RequestHandler(int number_of_threads)
    : thread_pool_(number_of_threads) {
    InitObject("RequestHandler", ID);
  }

  virtual ~RequestHandler() {
  }


  virtual void PushRequest(const RequestDescription& request_description) {
    Log(LOG_DEBUG, StrPrint("PushRequest for: %s",
        request_description.file_name().c_str()));

    thread_pool_.PushRequest(request_description);
  }


  virtual void Terminate() {
    thread_pool_.Terminate();
    Log(LOG_DEBUG, "RequestHandler terminated.");
  }


private:
  int number_of_threads_;   

  template<class S>
  class ThreadPool {

    public:

      ThreadPool(int number_of_threads) {
        pool_.size_controller().resize(number_of_threads);
        for(int i = 0; i < number_of_threads; i++) {
          dispatchers_.push(new S(i));
        }
        // Just created the pool.
        post_mortem_ = false;
      }

      ~ThreadPool() {
        // Check if pool is uninitialized cleanly.
        if( !post_mortem_) {
          printf("ThreadPool::~ThreadPool() was not uninit cleanly. \n");
        }

        // Remove all dispatcher objects.
        while( !dispatchers_.empty()) {
          S* tmp_ref = dispatchers_.top();
          dispatchers_.pop();
          delete tmp_ref;
        }
      }


      void PushRequest(const RequestDescription& request_description) {
        request_desriptions_.push(request_description);

        // Kick it on.
        ScheduleRequests();
      }


      void ScheduleRequestsManually() {
        ScheduleRequests();
      }


      void Terminate() {
        // std::queue doesn't know any clear() function. So pop them all off;)
        while( !request_desriptions_.empty()) {
          request_desriptions_.pop();
        }
        // Remove all requests from the schedule.
        pool_.clear();
        // Now wait until all running jobs finished.
        pool_.wait();

        // This pool has been successfully shut down.
        post_mortem_ = true;
      }


    protected:

      const RequestDescription& PopRequest() const {
        return request_desriptions_.pop();
      }


    private:

      void ScheduleRequests() {
        pool_.schedule(boost::bind(&ThreadPool<S>::ProcessRequest, this));
      }


      void ProcessRequest() {

        S* dispatcher;
        // Now get a free dispatcher.
        {
          boost::mutex::scoped_lock lock(dispatcher_mutex_);
          // Make sure that there are dispatchers defined.
          assert(!dispatchers_.empty());
          // Get the next dispatcher.
          dispatcher = dispatchers_.top();
          dispatchers_.pop();
        }

        printf("[ThreadPool::ProcessRequest() - processing request]\n");

        // Assign request to worker.
        dispatcher->set_request_description(request_desriptions_.front());
        request_desriptions_.pop();

        dispatcher->Init();

        // We do not directly spawn another thread in the dispatcher but execute
        // it from the current one.
        dispatcher->RunBlocking();

        dispatcher->Uninit();

//         boost::xtime xt;
//         boost::xtime_get(&xt, boost::TIME_UTC);
//         xt.sec += 4;
//         boost::thread::sleep(xt);

        // Return the dispatcher after usage.
        {
          boost::mutex::scoped_lock lock(dispatcher_mutex_);
          dispatchers_.push(dispatcher);
        }
      }


    private:

      boost::threadpool::pool pool_;  
      std::queue<RequestDescription> request_desriptions_; 

      boost::mutex dispatcher_mutex_; 
      std::stack<S*> dispatchers_;    

      bool post_mortem_;              

  }; // class ThreadPool


  ThreadPool<T> thread_pool_;         

}; // class RequestHandler



} // namespace metrivis

#endif // _METRIVIS_REQUEST_HANDLER_H

---