提交: 6ea7bd6f07f1a87e64a426aa9fdc2a585a6bb7c3 - ultramonkey-l7-v3 (git)

UltraMonkey-L7 V3(multi-thread implementation)

Commit MetaInfo

• 修訂: 6ea7bd6f07f1a87e64a426aa9fdc2a585a6bb7c3 (tree)
• 時間: 2012-07-31 11:40:39
• 作者: Hiroaki Nakano <nakano.hiroaki@nttc...>
• Commiter: Hiroaki Nakano

Log Message

Merge branch 'ssl_fix_by_CW' into v3.0.4-devel

Change Summary

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

差異

--- 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;
@@ -536,6 +536,7 @@ protected:
         virtual void down_thread_client_handle_async_write_some(const TCP_PROCESS_TYPE_TAG);
         virtual void down_thread_sorryserver_async_read_some_handler(const boost::system::error_code &error_code, std::size_t len);
         virtual void down_thread_sorryserver_handle_async_read_some(const TCP_PROCESS_TYPE_TAG);
+	virtual void up_thread_client_ssl_socket_clear_socket_handler();
 
         //! down thread receive from realserver and raise module event of handle_realserver_recv
         //! @param[in]        process_type is process type

--- a/l7vsd/include/tcp_ssl_socket.h
+++ b/l7vsd/include/tcp_ssl_socket.h
@@ -131,7 +131,7 @@ public:
 
         virtual std::size_t     read_some(const boost::asio::mutable_buffers_1 &buffers, boost::system::error_code &error_code) {
                 boost::mutex::scoped_lock       lock(ssl_mutex);
-                if (write_con > 0) {
+                if (write_con > 0 || handshake_con > 0) {
                         error_code = boost::asio::error::try_again;
                         return 0;
                 }
@@ -150,7 +150,7 @@ public:
         virtual size_t   write_some(const boost::asio::const_buffers_1 &buffer, boost::system::error_code &error_code) {
                 boost::mutex::scoped_lock       lock(ssl_mutex);
 
-                if (read_con > 0) {
+                if (read_con > 0 || handshake_con > 0) {
                         error_code = boost::asio::error::try_again;
                         return 0;
                 }

--- 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))) {
@@ -895,7 +897,19 @@ void tcp_session::up_thread_run()
 
         if (ssl_flag) {
                 client_ssl_socket.wait_async_event_all_end();
-                client_ssl_socket.clear_socket();
+                upthread_status = UPTHREAD_LOCK;
+                parent_dispatcher.post(boost::bind(&tcp_session::up_thread_client_ssl_socket_clear_socket_handler,this));
+                boost::mutex::scoped_lock lock(upthread_status_mutex);
+                while (unlikely(upthread_status == UPTHREAD_LOCK)) {
+                        to_time(LOCKTIMEOUT, xt);
+                        upthread_status_cond.timed_wait(lock, xt);
+                        tcp_thread_message *msg = up_thread_message_que.pop();
+                        if (msg) {      // message is alive.
+                                up_thread_next_call_function.second(LOCAL_PROC);
+                                delete msg;
+                                msg = NULL;
+                        }
+                }       // lockmode while loop end.
         }
 
         upthread_status = UPTHREAD_SLEEP;
@@ -1191,7 +1205,7 @@ void tcp_session::up_thread_client_accept_fail_event(const TCP_PROCESS_TYPE_TAG
 
                 boost::format   fmt("Thread ID[%d] tcp_ssl_socket::handshake[%s]");
                 fmt % boost::this_thread::get_id() % handshake_error_code.message();
-                Logger::putLogInfo(LOG_CAT_L7VSD_SESSION, 71, fmt.str(), __FILE__, __LINE__);
+                Logger::putLogError(LOG_CAT_L7VSD_SESSION, 71, fmt.str(), __FILE__, __LINE__);
 
         }
         up_thread_next_call_function = up_thread_function_array[func_tag];
@@ -1199,22 +1213,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();
 }
 
 
@@ -3874,7 +3906,7 @@ void tcp_session::down_thread_realserver_handle_async_read_some(const tcp_sessio
                                 realserver_socket->async_read_some(boost::asio::buffer(down_thread_data_dest_side.get_data()), handler);
                         else
                                 func_tag = DOWN_FUNC_REALSERVER_DISCONNECT;
-                        realserver_socket_mutex.lock();
+                        realserver_socket_mutex.unlock();
                 } else { //recv error
                         func_tag = DOWN_FUNC_REALSERVER_DISCONNECT;
                         boost::format   fmt("Thread ID[%d] down_thread_realserver_handle_async_read_some recv error:%s");
@@ -4017,6 +4049,30 @@ void tcp_session::down_thread_sorryserver_async_read_some_handler(const boost::s
         downthread_status_cond.notify_one();
 }
 
+void tcp_session::up_thread_client_ssl_socket_clear_socket_handler()
+{
+        if (unlikely(LOG_LV_DEBUG == Logger::getLogLevel(LOG_CAT_L7VSD_SESSION))) {
+                boost::format formatter("Thread ID[%d] FUNC IN up_thread_client_ssl_socket_clear_socket_handler");
+                formatter % boost::this_thread::get_id();
+                Logger::putLogDebug(LOG_CAT_L7VSD_SESSION, 999, formatter.str(), __FILE__, __LINE__);
+        }
+
+        client_ssl_socket.clear_socket();
+
+        tcp_thread_message *mes = new tcp_thread_message();
+        mes->message = up_que_function_map[UP_FUNC_PAUSE_OFF_EVENT];
+#ifdef  DEBUG
+        mes->func_tag_name = func_tag_to_string(UP_FUNC_PAUSE_OFF_EVENT);
+        {
+                boost::format   fmt("Thread ID[%d] up_queue.push : %s");
+                fmt % boost::this_thread::get_id() % func_tag_to_string(UP_FUNC_PAUSE_OFF_EVENT);
+                Logger::putLogInfo(LOG_CAT_L7VSD_SESSION, 999, fmt.str(), __FILE__, __LINE__);
+        }
+#endif
+        while (!up_thread_message_que.push(mes)) {}
+        upthread_status_cond.notify_one();
+}
+
 void tcp_session::down_thread_sorryserver_handle_async_read_some(tcp_session::TCP_PROCESS_TYPE_TAG)
 {
 

--- 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( "