manager.hpp

Go to the documentation of this file.

/*******************************************************************************
* Copyright (C) 2016 Eddie Carle [eddie@isatec.ca]                             *
*                                                                              *
* 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 FASTCGIPP_MANAGER_HPP
#define FASTCGIPP_MANAGER_HPP

#include <map>
#include <list>
#include <algorithm>
#include <thread>
#include <mutex>
#include <shared_mutex>
#include <memory>
#include <functional>

#include "fastcgi++/protocol.hpp"
#include "fastcgi++/transceiver.hpp"
#include "fastcgi++/request.hpp"

namespace Fastcgipp
{

    class Manager_base
    {
    public:

        Manager_base(unsigned threads);

        ~Manager_base();


        void terminate();


        void stop();


        void start();

        void join();


        static void setupSignals();


        bool listen()
        {
            return m_transceiver.listen();
        }


        bool listen(
                const char* name,
                uint32_t permissions = 0xffffffffUL,
                const char* owner = nullptr,
                const char* group = nullptr)
        {
            return m_transceiver.listen(name, permissions, owner, group);
        }


        bool listen(
                const char* interface,
                const char* service)
        {
            return m_transceiver.listen(interface, service);
        }

        void push(Protocol::RequestId id, Message&& message);

    protected:
        virtual std::unique_ptr<Request_base> makeRequest(
                const Protocol::RequestId& id,
                const Protocol::Role& role,
                bool kill) =0;

        Transceiver m_transceiver;

    private:
        std::queue<Protocol::RequestId> m_tasks;

        std::mutex m_tasksMutex;

        Protocol::Requests<std::unique_ptr<Request_base>> m_requests;

        std::shared_timed_mutex m_requestsMutex;

        std::queue<std::pair<Message, Socket>> m_messages;

        std::mutex m_messagesMutex;

        void handler();


        inline void localHandler();

        bool m_terminate;

        bool m_stop;

        std::mutex m_startStopMutex;

        std::vector<std::thread> m_threads;

        std::condition_variable m_wake;

        static void signalHandler(int signum);

        static Manager_base* instance;

#if FASTCGIPP_LOG_LEVEL > 3
        std::atomic_ullong m_requestCount;

        size_t m_maxRequests;

        std::atomic_ullong m_managementRecordCount;

        std::atomic_ullong m_badSocketMessageCount;

        std::atomic_ullong m_badSocketKillCount;

        std::atomic_ullong m_messageCount;

        unsigned m_activeThreads;

        unsigned m_maxActiveThreads;
#endif
    };


    template<class RequestT> class Manager: public Manager_base
    {
    public:

        Manager(unsigned threads = std::thread::hardware_concurrency()):
            Manager_base(threads)
        {}

    private:
        std::unique_ptr<Request_base> makeRequest(
                const Protocol::RequestId& id,
                const Protocol::Role& role,
                bool kill)
        {
            using namespace std::placeholders;

            std::unique_ptr<Request_base> request(new RequestT);
            static_cast<RequestT&>(*request).configure(
                    id,
                    role,
                    kill,
                    std::bind(&Transceiver::send, &m_transceiver, _1, _2, _3),
                    std::bind(&Manager_base::push, this, id, _1));
            return request;
        }

    };
}

#endif