manager.hpp

Go to the documentation of this file.

/***************************************************************************
* Copyright (C) 2007 Eddie Carle [eddie@erctech.org]                       *
*                                                                          *
* This file is part of fastcgi++.                                          *
*                                                                          *
* fastcgi++ is free software: you can redistribute it and/or modify it     *
* under the terms of the GNU Lesser General Public License as  published   *
* by the Free Software Foundation, either version 3 of the License, or (at *
* your option) any later version.                                          *
*                                                                          *
* fastcgi++ 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 Lesser General Public     *
* License for more details.                                                *
*                                                                          *
* You should have received a copy of the GNU Lesser General Public License *
* along with fastcgi++.  If not, see <http://www.gnu.org/licenses/>.       *
****************************************************************************/


#ifndef MANAGER_HPP
#define MANAGER_HPP

#include <map>
#include <string>
#include <queue>
#include <algorithm>
#include <cstring>

#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/thread/shared_mutex.hpp>

#include <signal.h>
#include <pthread.h>

#include <fastcgi++/protocol.hpp>
#include <fastcgi++/transceiver.hpp>

namespace Fastcgipp
{

   class ManagerPar
   {
   public:

      ManagerPar(int fd, const boost::function<void(Protocol::FullId, Message)>& sendMessage_, bool doSetupSignals);

      ~ManagerPar() { instance=0; }


      void stop();


      void terminate();
      

      void setupSignals();

      size_t getMessagesSize() const { return messages.size(); }

   protected:
      Transceiver transceiver;


      class Tasks: public std::queue<Protocol::FullId>, public boost::mutex {};

      Tasks tasks;

      std::queue<Message> messages;


      void localHandler(Protocol::FullId id);

      bool asleep;
      boost::mutex sleepMutex;

      pthread_t threadId;


      bool stopBool;
      boost::mutex stopMutex;

      bool terminateBool;
      boost::mutex terminateMutex;

   private:
      static void signalHandler(int signum);
      static ManagerPar* instance;
   };
   

   template<class T> class Manager: public ManagerPar
   {
   public:

      Manager(int fd=0, bool doSetupSignals=true): ManagerPar(fd, boost::bind(&Manager::push, boost::ref(*this), _1, _2), doSetupSignals) {}


      void handler();


      void push(Protocol::FullId id, Message message);

      size_t getRequestsSize() const { return requests.size(); }
   private:

      class Requests: public std::map<Protocol::FullId, boost::shared_ptr<T> >, public boost::shared_mutex {};

      Requests requests;
   };
}

template<class T> void Fastcgipp::Manager<T>::push(Protocol::FullId id, Message message)
{
   using namespace std;
   using namespace Protocol;
   using namespace boost;

   if(id.fcgiId)
   {
      shared_lock<shared_mutex> reqReadLock(requests);
      typename Requests::iterator it(requests.find(id));
      if(it!=requests.end())
      {
         lock_guard<mutex> mesLock(it->second->messages);
         it->second->messages.push(message);
         lock_guard<mutex> tasksLock(tasks);
         tasks.push(id);
      }
      else if(!message.type)
      {
         Header& header=*(Header*)message.data.get();
         if(header.getType()==BEGIN_REQUEST)
         {
            BeginRequest& body=*(BeginRequest*)(message.data.get()+sizeof(Header));

            reqReadLock.unlock();
            unique_lock<shared_mutex> reqWriteLock(requests);

            boost::shared_ptr<T>& request = requests[id];
            request.reset(new T);
            request->set(id, transceiver, body.getRole(), !body.getKeepConn(), boost::bind(&Manager::push, boost::ref(*this), id, _1));
         }
         else
            return;
      }
   }
   else
   {
      messages.push(message);
      tasks.push(id);
   }

   lock_guard<mutex> sleepLock(sleepMutex);
   if(asleep)
      transceiver.wake();
}

template<class T> void Fastcgipp::Manager<T>::handler()
{
   using namespace std;
   using namespace boost;

   threadId=pthread_self();

   while(1)
   {{
      {
         lock_guard<mutex> stopLock(stopMutex);
         if(stopBool)
         {
            stopBool=false;
            return;
         }
      }
      
      bool sleep=transceiver.handler();

      {
         lock_guard<mutex> terminateLock(terminateMutex);
         if(terminateBool)
         {
            shared_lock<shared_mutex> requestsLock(requests);
            if(requests.empty() && sleep)
            {
               terminateBool=false;
               return;
            }
         }
      }

      unique_lock<mutex> tasksLock(tasks);
      unique_lock<mutex> sleepLock(sleepMutex);

      if(tasks.empty())
      {
         tasksLock.unlock();

         asleep=true;
         sleepLock.unlock();

         if(sleep) transceiver.sleep();

         sleepLock.lock();
         asleep=false;
         sleepLock.unlock();

         continue;
      }

      sleepLock.unlock();

      Protocol::FullId id=tasks.front();
      tasks.pop();
      tasksLock.unlock();

      if(id.fcgiId==0)
         localHandler(id);
      else
      {
         shared_lock<shared_mutex> reqReadLock(requests);
         typename map<Protocol::FullId, boost::shared_ptr<T> >::iterator it(requests.find(id));
         if(it!=requests.end() && it->second->handler())
         {
            reqReadLock.unlock();
            unique_lock<shared_mutex> reqWriteLock(requests);

            requests.erase(it);
         }
      }
   }}
}

#endif