l7vsd/include/session_thread_control.h

   1 /*!
   2  *    @file    session_thread_control.h
   3  *    @brief    session used 2 threads. thread pooling unit is 2 thread control.
   4  *
   5  * L7VSD: Linux Virtual Server for Layer7 Load Balancing
   6  * Copyright (C) 2009  NTT COMWARE Corporation.
   7  *
   8  * This program is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2.1 of the License, or (at your option) any later version.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Lesser General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Lesser General Public
  19  * License along with this library; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
  21  * 02110-1301 USA
  22  *
  23  **********************************************************************/
  24
  25 #ifndef    SESSION_THREAD_CONTROL_H
  26 #define    SESSION_THREAD_CONTROL_H
  27 #include <boost/thread.hpp>
  28 #include <boost/thread/condition.hpp>
  29 #include <boost/bind.hpp>
  30 #include <boost/noncopyable.hpp>
  31 #include <boost/shared_ptr.hpp>
  32 #include "tcp_session.h"
  33
  34 #include "wrlock.h"
  35 #include "parameter.h"
  36 #include "error_code.h"
  37
  38 #define PARAM_SCHED_PRIORITY    "task_scheduler_priority"
  39 #define    NUM_OF_CORE_USES    "num_of_core_uses"
  40
  41 namespace l7vs
  42 {
  43
  44 class    tcp_session;
  45
  46 //!
  47 //!    @brief    session thread pooling utility class.
  48 //! @class    session_thread_control is session thread pool utility.
  49 class    session_thread_control : private boost::noncopyable
  50 {
  51 public:
  52         typedef    boost::shared_ptr< boost::thread >    thread_ptr;        //! shared_ptr thread typedef
  53         typedef    boost::shared_ptr< tcp_session >    session_ptr;    //! shared_ptr session typedef
  54         typedef    boost::thread::id                    thread_id_type;    //! thread id typedef
  55         enum    state_tag {   //! @enum state_tag upthread and down thread state enum
  56                 WAIT    = 0,    //! thread pooling mode
  57                 RUNNING,        //! thread running mode
  58                 EXIT            //! thread exit mode
  59         };
  60 protected:
  61         thread_ptr            upthread;                    //! upstream thread
  62         wr_mutex            upthread_state_mutex;        //! mutex for upstream thread status
  63         state_tag            upthread_state;                //! upstream thread state
  64         boost::mutex        upthread_condition_mutex;    //! upthread condition use mutex
  65         boost::condition    upthread_condition;            //! upthread condition
  66         boost::mutex        upthread_running_mutex;
  67         boost::mutex        upthread_joining_mutex;
  68         boost::condition    upthread_joining_condition;
  69         thread_ptr            downthread;                    //! downstream thread
  70         wr_mutex            downthread_state_mutex;        //! mutex for downstream thread status
  71         state_tag            downthread_state;            //! downstream thread state
  72         boost::mutex        downthread_condition_mutex;    //! downstream condition use mutex
  73         boost::condition    downthread_condition;        //! downstream condition
  74         boost::mutex        downthread_running_mutex;
  75         boost::mutex        downthread_joining_mutex;
  76         boost::condition    downthread_joining_condition;
  77         session_ptr            session;                    //! session class shared pointer
  78         void                upstream_run();                //! upstream thread bind function
  79         void                downstream_run();            //! downstream thread bind function
  80
  81         cpu_set_t            vsnic_cpumask;
  82         cpu_set_t            rsnic_cpumask;
  83
  84         int                    sched_algorithm;
  85         int                    sched_priority;
  86         int                    num_of_core_uses;
  87 public:
  88         //! constructor.
  89         //! @param session_ptr    session class shared ptr
  90         session_thread_control(tcp_session *ptr, cpu_set_t in_upcpu, cpu_set_t in_downcpu, int    in_sched_algorithm) :
  91                 upthread_state(WAIT),
  92                 downthread_state(WAIT),
  93                 vsnic_cpumask(in_upcpu),
  94                 rsnic_cpumask(in_downcpu),
  95                 sched_algorithm(in_sched_algorithm) {
  96                 int            int_val;
  97                 l7vs::error_code    err;
  98                 Parameter        param;
  99                 //get scheduler priority parameter
 100                 int_val    = param.get_int(l7vs::PARAM_COMP_L7VSD, PARAM_SCHED_PRIORITY, err);
 101                 if (!err)
 102                         if ((int_val > sched_get_priority_min(in_sched_algorithm)) || (int_val < sched_get_priority_max(in_sched_algorithm)))
 103                                 sched_priority = int_val;
 104                         else
 105                                 sched_priority = sched_get_priority_min(in_sched_algorithm);
 106                 else
 107                         sched_priority = 10;
 108                 //get number of use cpu_cores
 109                 int_val = param.get_int(l7vs::PARAM_COMP_VIRTUALSERVICE, NUM_OF_CORE_USES, err);
 110                 if (!err)
 111                         num_of_core_uses = int_val;
 112                 else
 113                         num_of_core_uses = 0;
 114
 115                 session.reset(ptr);
 116
 117         }
 118         //! destructor
 119         ~session_thread_control() {
 120         }
 121
 122         //! create up down thread
 123         void start_thread();
 124
 125         //! session shared ptr getter
 126         //! @return session shared ptr
 127         session_ptr        get_session() {
 128                 return session;
 129         }
 130         //! upstream thread start function
 131         void            startupstream();
 132         //! downstream thread start function
 133         void            startdownstream();
 134         //! upstream thread stop function
 135         void            stopupstream();
 136         //! downstream thread stop function
 137         void            stopdownstream();
 138         //! all thread destroy function.
 139         void            join();
 140         //! upstream-thread id getter
 141         //! @return thread_id_type    upstream thread id
 142         thread_id_type    get_upthread_id() {
 143                 return upthread->get_id();
 144         }
 145         //! downstream-thread id getter
 146         //! @return thread_id_type    downstream thread id
 147         thread_id_type    get_downthread_id() {
 148                 return downthread->get_id();
 149         }
 150         //! upthread mutex reference getter
 151         //! @return reference of boost::mutex
 152         boost::mutex    &get_upthread_mutex();
 153         //! downthread mutex reference getter
 154         //! @return reference of boost::mutex
 155         boost::mutex    &get_downthread_mutex();
 156
 157         void            session_stop() {
 158                 session->set_virtual_service_message(tcp_session::SESSION_END, boost::asio::ip::tcp::endpoint());
 159         }
 160         void            session_sorry_mode_change(int sorry_flag) {
 161                 if (INT_MAX == sorry_flag)
 162                         session->set_virtual_service_message(tcp_session::SORRY_STATE_DISABLE, boost::asio::ip::tcp::endpoint());
 163                 else if (0 != sorry_flag)
 164                         session->set_virtual_service_message(tcp_session::SORRY_STATE_ENABLE, boost::asio::ip::tcp::endpoint());
 165         }
 166         void            session_sorry_enable() {
 167                 session->set_virtual_service_message(tcp_session::SORRY_STATE_ENABLE, boost::asio::ip::tcp::endpoint());
 168         }
 169         void            session_sorry_disable() {
 170                 session->set_virtual_service_message(tcp_session::SORRY_STATE_DISABLE, boost::asio::ip::tcp::endpoint());
 171         }
 172         void            session_pause_on() {
 173                 session->set_virtual_service_message(tcp_session::SESSION_PAUSE_ON, boost::asio::ip::tcp::endpoint());
 174         }
 175         void            session_pause_off() {
 176                 session->set_virtual_service_message(tcp_session::SESSION_PAUSE_OFF, boost::asio::ip::tcp::endpoint());
 177         }
 178
 179         void session_accesslog_output_mode_on() {
 180                 session->set_virtual_service_message(tcp_session::ACCESS_LOG_ON, boost::asio::ip::tcp::endpoint());
 181         }
 182         void session_accesslog_output_mode_off() {
 183                 session->set_virtual_service_message(tcp_session::ACCESS_LOG_OFF, boost::asio::ip::tcp::endpoint());
 184         }
 185         void            session_access_log_output_mode_change(int accesslog_flag) {
 186                 if (accesslog_flag == true)
 187                         session->set_virtual_service_message(tcp_session::ACCESS_LOG_ON, boost::asio::ip::tcp::endpoint());
 188                 else if (accesslog_flag == false)
 189                         session->set_virtual_service_message(tcp_session::ACCESS_LOG_OFF, boost::asio::ip::tcp::endpoint());
 190         }
 191
 192         void            session_realserver_remove(const boost::asio::ip::tcp::endpoint &in) {
 193                 session->set_virtual_service_message(tcp_session::REALSERVER_REMOVE, in);
 194         }
 195
 196
 197 };
 198
 199 }//    namespace l7vs
 200 #endif    //SESSION_THREAD_CONTROL_H