Commit: 9a42cba8f6a3ef11e1315fb54cbd822c2cae9d00 - ultramonkey-l7-v3 (git)

UltraMonkey-L7 V3(multi-thread implementation)

Commit MetaInfo

• Révision: 9a42cba8f6a3ef11e1315fb54cbd822c2cae9d00 (tree)
• l'heure: 2012-07-30 15:55:41
• Auteur: Hiroaki Nakano <nakano.hiroaki@nttc...>
• Commiter: Hiroaki Nakano

Message de Log

ticket 29142 リアルサーバ切り離し時クラッシュ対処

SSL+sessionless+リアルサーバ切り離しで、まれに以下の現象が出る。
- l7vsdがクラッシュ（コアダンプ）する
- l7vsdプロセスのCPU使用率が100%に張り付く

原因は、非同期処理の待ち合わせを行うactive_sessions.do_all()を
l7vsadm -d実行時の通信スレッド停止・再開でも流用していたため。
そのために、非同期で待ち合わせていたスレッドが、l7vsadm -dで
再開させられていた。

対処として、active_sessions.do_all()を廃止し、l7vsadm -d時は
各スレッドに終了処理に入るようメッセージを飛ばす方式とした。
また、新規接続処理に、l7vsadmコマンドの待ち合わせ処理を追加。

Change Summary

• modified: l7vsd/include/tcp_session.h
• modified: l7vsd/include/virtualservice.h
• modified: l7vsd/src/tcp_session.cpp
• modified: l7vsd/src/virtualservice_base.cpp
• modified: l7vsd/src/virtualservice_tcp.cpp

Modification

--- a/l7vsd/include/tcp_session.h
+++ b/l7vsd/include/tcp_session.h
@@ -129,7 +129,7 @@ public:
         void down_thread_run();
         //! realserver remove
         //! @param[in] target endpoint
-        void realserver_remove(boost::asio::ip::tcp::endpoint &);
+        void realserver_remove(const boost::asio::ip::tcp::endpoint &);
 protected:
         typedef data_buff_base<boost::asio::ip::tcp>            tcp_data;
         typedef boost::asio::ip::tcp::endpoint endpoint;

--- a/l7vsd/include/virtualservice.h
+++ b/l7vsd/include/virtualservice.h
@@ -268,6 +268,10 @@ protected:
         bool downqos_alert_flag;        //! downstream QoS alert flag
         bool sessionpool_alert_flag;    //! sessionpool alert flag
 
+        bool adm_cmd_wait_flag;         //! wait for l7vsadm done
+        boost::mutex adm_cmd_wait_flag_mutex;
+        boost::condition adm_cmd_wait_flag_cond;
+ 
         void         load_parameter(l7vs::error_code &);
 
         virtual void handle_replication_interrupt(

--- a/l7vsd/src/tcp_session.cpp
+++ b/l7vsd/src/tcp_session.cpp
@@ -595,7 +595,9 @@ void tcp_session::set_virtual_service_message(const TCP_VIRTUAL_SERVICE_MESSAGE_
                         fmt % boost::this_thread::get_id();
                         Logger::putLogDebug(LOG_CAT_L7VSD_SESSION, 999, fmt.str(), __FILE__, __LINE__);
                 }
-                break;
+                //break;
+                realserver_remove(endpoint_); 
+                return;
         case SORRY_STATE_ENABLE:
                 //----Debug log----------------------------------------------------------------------
                 if (unlikely(LOG_LV_DEBUG == Logger::getLogLevel(LOG_CAT_L7VSD_SESSION))) {
@@ -1199,22 +1201,40 @@ void tcp_session::up_thread_client_accept_fail_event(const TCP_PROCESS_TYPE_TAG
 
 //! real server remove
 //! @param[in]        target endpoint
-void tcp_session::realserver_remove(endpoint &target_endpoint)
+void tcp_session::realserver_remove(const endpoint &target_endpoint)
 {
+
+        if (target_endpoint != realserver_endpoint && target_endpoint != connecting_endpoint) return;
+
         tcp_thread_message *up_msg = new tcp_thread_message;
-        up_thread_function_pair up_func = up_thread_function_array[UP_FUNC_REALSERVER_CHECK];
+        up_thread_function_pair up_func = up_thread_function_array[UP_FUNC_EXIT];
         up_msg->message = up_func.second;
         up_msg->endpoint_info = target_endpoint;
 #ifdef  DEBUG
-        up_msg->func_tag_name = func_tag_to_string(UP_FUNC_REALSERVER_CHECK);
+        up_msg->func_tag_name = func_tag_to_string(UP_FUNC_EXIT);
         {
                 boost::format   fmt("Thread ID[%d] up_queue.push : %s");
-                fmt % boost::this_thread::get_id() % func_tag_to_string(UP_FUNC_REALSERVER_CHECK);
+                fmt % boost::this_thread::get_id() % func_tag_to_string(UP_FUNC_EXIT);
                 Logger::putLogInfo(LOG_CAT_L7VSD_SESSION, 999, fmt.str(), __FILE__, __LINE__);
         }
 #endif
         while (!up_thread_message_que.push(up_msg)) {}
         upthread_status_cond.notify_one();
+
+        tcp_thread_message *down_msg = new tcp_thread_message;
+        down_thread_function_pair down_func = down_thread_function_array[DOWN_FUNC_EXIT];
+        down_msg->message = down_func.second;
+        down_msg->endpoint_info = target_endpoint;
+#ifdef  DEBUG
+        down_msg->func_tag_name = func_tag_to_string(DOWN_FUNC_EXIT);
+        {
+                boost::format   fmt("Thread ID[%d] down_queue.push : %s");
+                fmt % boost::this_thread::get_id() % func_tag_to_string(DOWN_FUNC_EXIT);
+                Logger::putLogInfo(LOG_CAT_L7VSD_SESSION, 999, fmt.str(), __FILE__, __LINE__);
+        }
+#endif
+        while (!down_thread_message_que.push(down_msg)) {}
+        downthread_status_cond.notify_one();
 }
 
 

--- a/l7vsd/src/virtualservice_base.cpp
+++ b/l7vsd/src/virtualservice_base.cpp
@@ -69,6 +69,7 @@ l7vs::virtualservice_base::virtualservice_base(const l7vs::l7vsd &invsd,
         upqos_alert_flag = false;
         downqos_alert_flag = false;
         sessionpool_alert_flag = false;
+        adm_cmd_wait_flag = false;
 
         rs_list.clear();
         protomod = NULL;

--- a/l7vsd/src/virtualservice_tcp.cpp
+++ b/l7vsd/src/virtualservice_tcp.cpp
@@ -352,6 +352,13 @@ void l7vs::virtualservice_tcp::handle_accept(const l7vs::session_thread_control
                 return;
         }
 
+        {
+		boost::mutex::scoped_lock       lock(adm_cmd_wait_flag_mutex);
+                if (unlikely(adm_cmd_wait_flag)){
+			adm_cmd_wait_flag_cond.wait(lock);
+		}
+        }
+
         session_thread_control *stc_ptr_noconst = const_cast<session_thread_control *>(stc_ptr);
 
         if (unlikely(err == boost::asio::error::operation_aborted)) {   // nomal exit case
@@ -1290,9 +1297,6 @@ void l7vs::virtualservice_tcp::add_realserver(const l7vs::virtualservice_element
                 }
         }
 
-        //pause active sessions
-        active_sessions.do_all(boost::bind(&session_thread_control::session_pause_on, _1));
-
         //add realserver
         for (std::vector<realserver_element>::iterator itr = in_element.realserver_vector.begin();
              itr != in_element.realserver_vector.end();
@@ -1305,9 +1309,6 @@ void l7vs::virtualservice_tcp::add_realserver(const l7vs::virtualservice_element
                 rs_list.push_back(rs);
         }
 
-        //run active sessions
-        active_sessions.do_all(boost::bind(&session_thread_control::session_pause_off, _1));
-
         if (unlikely(LOG_LV_DEBUG == Logger::getLogLevel(LOG_CAT_L7VSD_VIRTUALSERVICE))) {
                 boost::format formatter("out_function: void virtualservice_tcp::add_realserver( "
                                         "const l7vs::virtualservice_element& in,"
@@ -1384,8 +1385,11 @@ void l7vs::virtualservice_tcp::edit_realserver(const l7vs::virtualservice_elemen
                 }
         }
 
-        //pause active sessions
-        active_sessions.do_all(boost::bind(&session_thread_control::session_pause_on, _1));
+        //lock adm_cmd_wait_flag on
+        adm_cmd_wait_flag_mutex.lock();
+        adm_cmd_wait_flag = true;
+        adm_cmd_wait_flag_cond.notify_one();
+        adm_cmd_wait_flag_mutex.unlock();
 
         //edit realserver
         for (std::vector<realserver_element>::iterator itr = in_element.realserver_vector.begin();
@@ -1406,8 +1410,11 @@ void l7vs::virtualservice_tcp::edit_realserver(const l7vs::virtualservice_elemen
                 }
         }
 
-        //run active sessions
-        active_sessions.do_all(boost::bind(&session_thread_control::session_pause_off, _1));
+        //lock adm_cmd_wait_flag off
+        adm_cmd_wait_flag_mutex.lock();
+        adm_cmd_wait_flag = false;
+        adm_cmd_wait_flag_cond.notify_one();
+        adm_cmd_wait_flag_mutex.unlock();
 
         err.setter(false, "");
         if (unlikely(LOG_LV_DEBUG == Logger::getLogLevel(LOG_CAT_L7VSD_VIRTUALSERVICE))) {
@@ -1486,8 +1493,11 @@ void l7vs::virtualservice_tcp::del_realserver(const l7vs::virtualservice_element
                 }
         }
 
-        //pause active sessions
-        active_sessions.do_all(boost::bind(&session_thread_control::session_pause_on, _1));
+        //lock adm_cmd_wait_flag on
+        adm_cmd_wait_flag_mutex.lock();
+        adm_cmd_wait_flag = true;
+        adm_cmd_wait_flag_cond.notify_one();
+        adm_cmd_wait_flag_mutex.unlock();
 
         //del realserver
         for (std::vector<realserver_element>::iterator itr = in_element.realserver_vector.begin();
@@ -1503,8 +1513,11 @@ void l7vs::virtualservice_tcp::del_realserver(const l7vs::virtualservice_element
                 }
         }
 
-        //run active sessions
-        active_sessions.do_all(boost::bind(&session_thread_control::session_pause_off, _1));
+        //lock adm_cmd_wait_flag off
+        adm_cmd_wait_flag_mutex.lock();
+        adm_cmd_wait_flag = false;
+        adm_cmd_wait_flag_cond.notify_one();
+        adm_cmd_wait_flag_mutex.unlock();
 
         if (unlikely(LOG_LV_DEBUG == Logger::getLogLevel(LOG_CAT_L7VSD_VIRTUALSERVICE))) {
                 boost::format formatter("out_function: void virtualservice_tcp::del_realserver( "