mirror of
https://github.com/monero-project/monero.git
synced 2024-12-25 17:49:28 -05:00
Remove levin_client_async.*
This commit is contained in:
Jeffrey
parent
0da8bdc2e0
commit
4b3d9de65b
@ -1,585 +0,0 @@
|
||||
// Copyright (c) 2006-2013, Andrey N. Sabelnikov, www.sabelnikov.net
|
||||
// All rights reserved.
|
||||
//
|
||||
// Redistribution and use in source and binary forms, with or without
|
||||
// modification, are permitted provided that the following conditions are met:
|
||||
// * Redistributions of source code must retain the above copyright
|
||||
// notice, this list of conditions and the following disclaimer.
|
||||
// * Redistributions in binary form must reproduce the above copyright
|
||||
// notice, this list of conditions and the following disclaimer in the
|
||||
// documentation and/or other materials provided with the distribution.
|
||||
// * Neither the name of the Andrey N. Sabelnikov nor the
|
||||
// names of its contributors may be used to endorse or promote products
|
||||
// derived from this software without specific prior written permission.
|
||||
//
|
||||
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
|
||||
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
|
||||
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
|
||||
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER BE LIABLE FOR ANY
|
||||
// DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
|
||||
// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
|
||||
// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
|
||||
// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
|
||||
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
//
|
||||
|
||||
|
||||
#pragma once
|
||||
|
||||
#include ""
|
||||
#include "net_helper.h"
|
||||
#include "levin_base.h"
|
||||
|
||||
#undef MONERO_DEFAULT_LOG_CATEGORY
|
||||
#define MONERO_DEFAULT_LOG_CATEGORY "net"
|
||||
|
||||
|
||||
namespace epee
|
||||
{
|
||||
namespace levin
|
||||
{
|
||||
|
||||
/************************************************************************
|
||||
* levin_client_async - probably it is not really fast implementation,
|
||||
* each handler thread could make up to 30 ms latency.
|
||||
* But, handling events in reader thread will cause dead locks in
|
||||
* case of recursive call (call invoke() to the same connection
|
||||
* on reader thread on remote invoke() handler)
|
||||
***********************************************************************/
|
||||
|
||||
|
||||
class levin_client_async
|
||||
{
|
||||
levin_commands_handler* m_pcommands_handler;
|
||||
void (*commands_handler_destroy)(levin_commands_handler*);
|
||||
volatile uint32_t m_is_stop;
|
||||
volatile uint32_t m_threads_count;
|
||||
::critical_section m_send_lock;
|
||||
|
||||
std::string m_local_invoke_buff;
|
||||
::critical_section m_local_invoke_buff_lock;
|
||||
volatile int m_invoke_res;
|
||||
|
||||
volatile uint32_t m_invoke_data_ready;
|
||||
volatile uint32_t m_invoke_is_active;
|
||||
|
||||
boost::mutex m_invoke_event;
|
||||
boost::condition_variable m_invoke_cond;
|
||||
size_t m_timeout;
|
||||
|
||||
::critical_section m_recieved_packets_lock;
|
||||
struct packet_entry
|
||||
{
|
||||
bucket_head m_hd;
|
||||
std::string m_body;
|
||||
uint32_t m_connection_index;
|
||||
};
|
||||
std::list<packet_entry> m_recieved_packets;
|
||||
/*
|
||||
m_current_connection_index needed when some connection was broken and reconnected - in this
|
||||
case we could have some received packets in que, which shoud not be handled
|
||||
*/
|
||||
volatile uint32_t m_current_connection_index;
|
||||
::critical_section m_invoke_lock;
|
||||
::critical_section m_reciev_packet_lock;
|
||||
::critical_section m_connection_lock;
|
||||
net_utils::blocked_mode_client m_transport;
|
||||
public:
|
||||
levin_client_async():m_pcommands_handler(NULL), commands_handler_destroy(NULL), m_is_stop(0), m_threads_count(0), m_invoke_data_ready(0), m_invoke_is_active(0)
|
||||
{}
|
||||
levin_client_async(const levin_client_async& /*v*/):m_pcommands_handler(NULL), commands_handler_destroy(NULL), m_is_stop(0), m_threads_count(0), m_invoke_data_ready(0), m_invoke_is_active(0)
|
||||
{}
|
||||
~levin_client_async()
|
||||
{
|
||||
boost::interprocess::ipcdetail::atomic_write32(&m_is_stop, 1);
|
||||
disconnect();
|
||||
|
||||
|
||||
while(boost::interprocess::ipcdetail::atomic_read32(&m_threads_count))
|
||||
::Sleep(100);
|
||||
|
||||
set_handler(NULL);
|
||||
}
|
||||
|
||||
void set_handler(levin_commands_handler* phandler, void (*destroy)(levin_commands_handler*) = NULL)
|
||||
{
|
||||
if (commands_handler_destroy && m_pcommands_handler)
|
||||
(*commands_handler_destroy)(m_pcommands_handler);
|
||||
m_pcommands_handler = phandler;
|
||||
m_pcommands_handler_destroy = destroy;
|
||||
}
|
||||
|
||||
bool connect(uint32_t ip, uint32_t port, uint32_t timeout)
|
||||
{
|
||||
loop_call_guard();
|
||||
critical_region cr(m_connection_lock);
|
||||
|
||||
m_timeout = timeout;
|
||||
bool res = false;
|
||||
CRITICAL_REGION_BEGIN(m_reciev_packet_lock);
|
||||
CRITICAL_REGION_BEGIN(m_send_lock);
|
||||
res = levin_client_impl::connect(ip, port, timeout);
|
||||
boost::interprocess::ipcdetail::atomic_inc32(&m_current_connection_index);
|
||||
CRITICAL_REGION_END();
|
||||
CRITICAL_REGION_END();
|
||||
if(res && !boost::interprocess::ipcdetail::atomic_read32(&m_threads_count) )
|
||||
{
|
||||
//boost::interprocess::ipcdetail::atomic_write32(&m_is_stop, 0);//m_is_stop = false;
|
||||
boost::thread( boost::bind(&levin_duplex_client::reciever_thread, this) );
|
||||
boost::thread( boost::bind(&levin_duplex_client::handler_thread, this) );
|
||||
boost::thread( boost::bind(&levin_duplex_client::handler_thread, this) );
|
||||
}
|
||||
|
||||
return res;
|
||||
}
|
||||
bool is_connected()
|
||||
{
|
||||
loop_call_guard();
|
||||
critical_region cr(m_cs);
|
||||
return levin_client_impl::is_connected();
|
||||
}
|
||||
|
||||
inline
|
||||
bool check_connection()
|
||||
{
|
||||
loop_call_guard();
|
||||
critical_region cr(m_cs);
|
||||
|
||||
if(!is_connected())
|
||||
{
|
||||
if( !reconnect() )
|
||||
{
|
||||
LOG_ERROR("Reconnect Failed. Failed to invoke() because not connected!");
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
//------------------------------------------------------------------------------
|
||||
inline
|
||||
bool recv_n(SOCKET s, char* pbuff, size_t cb)
|
||||
{
|
||||
while(cb)
|
||||
{
|
||||
int res = ::recv(m_socket, pbuff, (int)cb, 0);
|
||||
|
||||
if(SOCKET_ERROR == res)
|
||||
{
|
||||
if(!m_connected)
|
||||
return false;
|
||||
|
||||
int err = ::WSAGetLastError();
|
||||
LOG_ERROR("Failed to recv(), err = " << err << " \"" << socket_errors::get_socket_error_text(err) <<"\"");
|
||||
disconnect();
|
||||
//reconnect();
|
||||
return false;
|
||||
}else if(res == 0)
|
||||
{
|
||||
disconnect();
|
||||
//reconnect();
|
||||
return false;
|
||||
}
|
||||
LOG_PRINT_L4("[" << m_socket <<"] RECV " << res);
|
||||
cb -= res;
|
||||
pbuff += res;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
//------------------------------------------------------------------------------
|
||||
inline
|
||||
bool recv_n(SOCKET s, std::string& buff)
|
||||
{
|
||||
size_t cb_remain = buff.size();
|
||||
char* m_current_ptr = (char*)buff.data();
|
||||
return recv_n(s, m_current_ptr, cb_remain);
|
||||
}
|
||||
|
||||
bool disconnect()
|
||||
{
|
||||
//boost::interprocess::ipcdetail::atomic_write32(&m_is_stop, 1);//m_is_stop = true;
|
||||
loop_call_guard();
|
||||
critical_region cr(m_cs);
|
||||
levin_client_impl::disconnect();
|
||||
|
||||
CRITICAL_REGION_BEGIN(m_local_invoke_buff_lock);
|
||||
m_local_invoke_buff.clear();
|
||||
m_invoke_res = LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
CRITICAL_REGION_END();
|
||||
boost::interprocess::ipcdetail::atomic_write32(&m_invoke_data_ready, 1); //m_invoke_data_ready = true;
|
||||
m_invoke_cond.notify_all();
|
||||
return true;
|
||||
}
|
||||
|
||||
void loop_call_guard()
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
void on_leave_invoke()
|
||||
{
|
||||
boost::interprocess::ipcdetail::atomic_write32(&m_invoke_is_active, 0);
|
||||
}
|
||||
|
||||
int invoke(const GUID& target, int command, const std::string& in_buff, std::string& buff_out)
|
||||
{
|
||||
|
||||
critical_region cr_invoke(m_invoke_lock);
|
||||
|
||||
boost::interprocess::ipcdetail::atomic_write32(&m_invoke_is_active, 1);
|
||||
boost::interprocess::ipcdetail::atomic_write32(&m_invoke_data_ready, 0);
|
||||
misc_utils::destr_ptr hdlr = misc_utils::add_exit_scope_handler(boost::bind(&levin_duplex_client::on_leave_invoke, this));
|
||||
|
||||
loop_call_guard();
|
||||
|
||||
if(!check_connection())
|
||||
return LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
|
||||
|
||||
bucket_head head = {0};
|
||||
head.m_signature = LEVIN_SIGNATURE;
|
||||
head.m_cb = in_buff.size();
|
||||
head.m_have_to_return_data = true;
|
||||
head.m_id = target;
|
||||
#ifdef TRACE_LEVIN_PACKETS_BY_GUIDS
|
||||
::UuidCreate(&head.m_id);
|
||||
#endif
|
||||
head.m_command = command;
|
||||
head.m_protocol_version = LEVIN_PROTOCOL_VER_1;
|
||||
head.m_flags = LEVIN_PACKET_REQUEST;
|
||||
LOG_PRINT("[" << m_socket <<"] Sending invoke data", LOG_LEVEL_4);
|
||||
|
||||
CRITICAL_REGION_BEGIN(m_send_lock);
|
||||
LOG_PRINT_L4("[" << m_socket <<"] SEND " << sizeof(head));
|
||||
int res = ::send(m_socket, (const char*)&head, sizeof(head), 0);
|
||||
if(SOCKET_ERROR == res)
|
||||
{
|
||||
int err = ::WSAGetLastError();
|
||||
LOG_ERROR("Failed to send(), err = " << err << " \"" << socket_errors::get_socket_error_text(err) <<"\"");
|
||||
disconnect();
|
||||
return LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
}
|
||||
LOG_PRINT_L4("[" << m_socket <<"] SEND " << (int)in_buff.size());
|
||||
res = ::send(m_socket, in_buff.data(), (int)in_buff.size(), 0);
|
||||
if(SOCKET_ERROR == res)
|
||||
{
|
||||
int err = ::WSAGetLastError();
|
||||
LOG_ERROR("Failed to send(), err = " << err << " \"" << socket_errors::get_socket_error_text(err) <<"\"");
|
||||
disconnect();
|
||||
return LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
}
|
||||
CRITICAL_REGION_END();
|
||||
LOG_PRINT_L4("LEVIN_PACKET_SENT. [len=" << head.m_cb << ", flags=" << head.m_flags << ", is_cmd=" << head.m_have_to_return_data <<", cmd_id = " << head.m_command << ", pr_v=" << head.m_protocol_version << ", uid=" << string_tools::get_str_from_guid_a(head.m_id) << "]");
|
||||
|
||||
//hard coded timeout in 10 minutes for maximum invoke period. if it happens, it could mean only some real troubles.
|
||||
boost::system_time timeout = boost::get_system_time()+ boost::posix_time::milliseconds(100);
|
||||
size_t timeout_count = 0;
|
||||
boost::unique_lock<boost::mutex> lock(m_invoke_event);
|
||||
|
||||
while(!boost::interprocess::ipcdetail::atomic_read32(&m_invoke_data_ready))
|
||||
{
|
||||
if(!m_invoke_cond.timed_wait(lock, timeout))
|
||||
{
|
||||
if(timeout_count < 10)
|
||||
{
|
||||
//workaround to avoid freezing at timed_wait called after notify_all.
|
||||
timeout = boost::get_system_time()+ boost::posix_time::milliseconds(100);
|
||||
++timeout_count;
|
||||
continue;
|
||||
}else if(timeout_count == 10)
|
||||
{
|
||||
//workaround to avoid freezing at timed_wait called after notify_all.
|
||||
timeout = boost::get_system_time()+ boost::posix_time::minutes(10);
|
||||
++timeout_count;
|
||||
continue;
|
||||
}else
|
||||
{
|
||||
LOG_PRINT("[" << m_socket <<"] Timeout on waiting invoke result. ", LOG_LEVEL_0);
|
||||
//disconnect();
|
||||
return LEVIN_ERROR_CONNECTION_TIMEDOUT;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
CRITICAL_REGION_BEGIN(m_local_invoke_buff_lock);
|
||||
buff_out.swap(m_local_invoke_buff);
|
||||
m_local_invoke_buff.clear();
|
||||
CRITICAL_REGION_END();
|
||||
return m_invoke_res;
|
||||
}
|
||||
|
||||
int notify(const GUID& target, int command, const std::string& in_buff)
|
||||
{
|
||||
if(!check_connection())
|
||||
return LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
|
||||
bucket_head head = {0};
|
||||
head.m_signature = LEVIN_SIGNATURE;
|
||||
head.m_cb = in_buff.size();
|
||||
head.m_have_to_return_data = false;
|
||||
head.m_id = target;
|
||||
#ifdef TRACE_LEVIN_PACKETS_BY_GUIDS
|
||||
::UuidCreate(&head.m_id);
|
||||
#endif
|
||||
head.m_command = command;
|
||||
head.m_protocol_version = LEVIN_PROTOCOL_VER_1;
|
||||
head.m_flags = LEVIN_PACKET_REQUEST;
|
||||
CRITICAL_REGION_BEGIN(m_send_lock);
|
||||
LOG_PRINT_L4("[" << m_socket <<"] SEND " << sizeof(head));
|
||||
int res = ::send(m_socket, (const char*)&head, sizeof(head), 0);
|
||||
if(SOCKET_ERROR == res)
|
||||
{
|
||||
int err = ::WSAGetLastError();
|
||||
LOG_ERROR("Failed to send(), err = " << err << " \"" << socket_errors::get_socket_error_text(err) <<"\"");
|
||||
disconnect();
|
||||
return LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
}
|
||||
LOG_PRINT_L4("[" << m_socket <<"] SEND " << (int)in_buff.size());
|
||||
res = ::send(m_socket, in_buff.data(), (int)in_buff.size(), 0);
|
||||
if(SOCKET_ERROR == res)
|
||||
{
|
||||
int err = ::WSAGetLastError();
|
||||
LOG_ERROR("Failed to send(), err = " << err << " \"" << socket_errors::get_socket_error_text(err) <<"\"");
|
||||
disconnect();
|
||||
return LEVIN_ERROR_CONNECTION_DESTROYED;
|
||||
}
|
||||
CRITICAL_REGION_END();
|
||||
LOG_PRINT_L4("LEVIN_PACKET_SENT. [len=" << head.m_cb << ", flags=" << head.m_flags << ", is_cmd=" << head.m_have_to_return_data <<", cmd_id = " << head.m_command << ", pr_v=" << head.m_protocol_version << ", uid=" << string_tools::get_str_from_guid_a(head.m_id) << "]");
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
|
||||
private:
|
||||
bool have_some_data(SOCKET sock, int interval = 1)
|
||||
{
|
||||
fd_set fds;
|
||||
FD_ZERO(&fds);
|
||||
FD_SET(sock, &fds);
|
||||
|
||||
fd_set fdse;
|
||||
FD_ZERO(&fdse);
|
||||
FD_SET(sock, &fdse);
|
||||
|
||||
|
||||
timeval tv;
|
||||
tv.tv_sec = interval;
|
||||
tv.tv_usec = 0;
|
||||
|
||||
int sel_res = select(0, &fds, 0, &fdse, &tv);
|
||||
if(0 == sel_res)
|
||||
return false;
|
||||
else if(sel_res == SOCKET_ERROR)
|
||||
{
|
||||
if(m_is_stop)
|
||||
return false;
|
||||
int err_code = ::WSAGetLastError();
|
||||
LOG_ERROR("Filed to call select, err code = " << err_code);
|
||||
disconnect();
|
||||
}else
|
||||
{
|
||||
if(fds.fd_array[0])
|
||||
{//some read operations was performed
|
||||
return true;
|
||||
}else if(fdse.fd_array[0])
|
||||
{//some error was at the socket
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
bool reciev_and_process_incoming_data()
|
||||
{
|
||||
bucket_head head = {0};
|
||||
uint32_t conn_index = 0;
|
||||
bool is_request = false;
|
||||
std::string local_buff;
|
||||
CRITICAL_REGION_BEGIN(m_reciev_packet_lock);//to protect from socket reconnect between head and body
|
||||
|
||||
if(!recv_n(m_socket, (char*)&head, sizeof(head)))
|
||||
{
|
||||
if(m_is_stop)
|
||||
return false;
|
||||
LOG_ERROR("Failed to recv_n");
|
||||
return false;
|
||||
}
|
||||
|
||||
conn_index = boost::interprocess::ipcdetail::atomic_read32(&m_current_connection_index);
|
||||
|
||||
if(head.m_signature!=LEVIN_SIGNATURE)
|
||||
{
|
||||
LOG_ERROR("Signature mismatch in response");
|
||||
return false;
|
||||
}
|
||||
|
||||
is_request = (head.m_protocol_version == LEVIN_PROTOCOL_VER_1 && head.m_flags&LEVIN_PACKET_REQUEST);
|
||||
|
||||
|
||||
local_buff.resize((size_t)head.m_cb);
|
||||
if(!recv_n(m_socket, local_buff))
|
||||
{
|
||||
if(m_is_stop)
|
||||
return false;
|
||||
LOG_ERROR("Filed to reciev");
|
||||
return false;
|
||||
}
|
||||
CRITICAL_REGION_END();
|
||||
|
||||
LOG_PRINT_L4("LEVIN_PACKET_RECEIVED. [len=" << head.m_cb << ", flags=" << head.m_flags << ", is_cmd=" << head.m_have_to_return_data <<", cmd_id = " << head.m_command << ", pr_v=" << head.m_protocol_version << ", uid=" << string_tools::get_str_from_guid_a(head.m_id) << "]");
|
||||
|
||||
if(is_request)
|
||||
{
|
||||
CRITICAL_REGION_BEGIN(m_recieved_packets_lock);
|
||||
m_recieved_packets.resize(m_recieved_packets.size() + 1);
|
||||
m_recieved_packets.back().m_hd = head;
|
||||
m_recieved_packets.back().m_body.swap(local_buff);
|
||||
m_recieved_packets.back().m_connection_index = conn_index;
|
||||
CRITICAL_REGION_END();
|
||||
/*
|
||||
|
||||
*/
|
||||
}else
|
||||
{//this is some response
|
||||
|
||||
CRITICAL_REGION_BEGIN(m_local_invoke_buff_lock);
|
||||
m_local_invoke_buff.swap(local_buff);
|
||||
m_invoke_res = head.m_return_code;
|
||||
CRITICAL_REGION_END();
|
||||
boost::interprocess::ipcdetail::atomic_write32(&m_invoke_data_ready, 1); //m_invoke_data_ready = true;
|
||||
m_invoke_cond.notify_all();
|
||||
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool reciever_thread()
|
||||
{
|
||||
LOG_PRINT_L3("[" << m_socket <<"] Socket reciever thread started.[m_threads_count=" << m_threads_count << "]");
|
||||
log_space::log_singletone::set_thread_log_prefix("RECIEVER_WORKER");
|
||||
boost::interprocess::ipcdetail::atomic_inc32(&m_threads_count);
|
||||
|
||||
while(!m_is_stop)
|
||||
{
|
||||
if(!m_connected)
|
||||
{
|
||||
Sleep(100);
|
||||
continue;
|
||||
}
|
||||
|
||||
if(have_some_data(m_socket, 1))
|
||||
{
|
||||
if(!reciev_and_process_incoming_data())
|
||||
{
|
||||
if(m_is_stop)
|
||||
{
|
||||
break;//boost::interprocess::ipcdetail::atomic_dec32(&m_threads_count);
|
||||
//return true;
|
||||
}
|
||||
LOG_ERROR("Failed to reciev_and_process_incoming_data. shutting down");
|
||||
//boost::interprocess::ipcdetail::atomic_dec32(&m_threads_count);
|
||||
//disconnect_no_wait();
|
||||
//break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
boost::interprocess::ipcdetail::atomic_dec32(&m_threads_count);
|
||||
LOG_PRINT_L3("[" << m_socket <<"] Socket reciever thread stopped.[m_threads_count=" << m_threads_count << "]");
|
||||
return true;
|
||||
}
|
||||
|
||||
bool process_recieved_packet(bucket_head& head, const std::string& local_buff, uint32_t conn_index)
|
||||
{
|
||||
|
||||
net_utils::connection_context_base conn_context;
|
||||
conn_context.m_remote_address = m_address;
|
||||
if(head.m_have_to_return_data)
|
||||
{
|
||||
std::string return_buff;
|
||||
if(m_pcommands_handler)
|
||||
head.m_return_code = m_pcommands_handler->invoke(head.m_id, head.m_command, local_buff, return_buff, conn_context);
|
||||
else
|
||||
head.m_return_code = LEVIN_ERROR_CONNECTION_HANDLER_NOT_DEFINED;
|
||||
|
||||
|
||||
|
||||
head.m_cb = return_buff.size();
|
||||
head.m_have_to_return_data = false;
|
||||
head.m_protocol_version = LEVIN_PROTOCOL_VER_1;
|
||||
head.m_flags = LEVIN_PACKET_RESPONSE;
|
||||
|
||||
std::string send_buff((const char*)&head, sizeof(head));
|
||||
send_buff += return_buff;
|
||||
CRITICAL_REGION_BEGIN(m_send_lock);
|
||||
if(conn_index != boost::interprocess::ipcdetail::atomic_read32(&m_current_connection_index))
|
||||
{//there was reconnect, send response back is not allowed
|
||||
return true;
|
||||
}
|
||||
int res = ::send(m_socket, (const char*)send_buff.data(), send_buff.size(), 0);
|
||||
if(res == SOCKET_ERROR)
|
||||
{
|
||||
int err_code = ::WSAGetLastError();
|
||||
LOG_ERROR("Failed to send, err = " << err_code);
|
||||
return false;
|
||||
}
|
||||
CRITICAL_REGION_END();
|
||||
LOG_PRINT_L4("LEVIN_PACKET_SENT. [len=" << head.m_cb << ", flags=" << head.m_flags << ", is_cmd=" << head.m_have_to_return_data <<", cmd_id = " << head.m_command << ", pr_v=" << head.m_protocol_version << ", uid=" << string_tools::get_str_from_guid_a(head.m_id) << "]");
|
||||
|
||||
}
|
||||
else
|
||||
{
|
||||
if(m_pcommands_handler)
|
||||
m_pcommands_handler->notify(head.m_id, head.m_command, local_buff, conn_context);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool handler_thread()
|
||||
{
|
||||
LOG_PRINT_L3("[" << m_socket <<"] Socket handler thread started.[m_threads_count=" << m_threads_count << "]");
|
||||
log_space::log_singletone::set_thread_log_prefix("HANDLER_WORKER");
|
||||
boost::interprocess::ipcdetail::atomic_inc32(&m_threads_count);
|
||||
|
||||
while(!m_is_stop)
|
||||
{
|
||||
bool have_some_work = false;
|
||||
std::string local_buff;
|
||||
bucket_head bh = {0};
|
||||
uint32_t conn_index = 0;
|
||||
|
||||
CRITICAL_REGION_BEGIN(m_recieved_packets_lock);
|
||||
if(m_recieved_packets.size())
|
||||
{
|
||||
bh = m_recieved_packets.begin()->m_hd;
|
||||
conn_index = m_recieved_packets.begin()->m_connection_index;
|
||||
local_buff.swap(m_recieved_packets.begin()->m_body);
|
||||
have_some_work = true;
|
||||
m_recieved_packets.pop_front();
|
||||
}
|
||||
CRITICAL_REGION_END();
|
||||
|
||||
if(have_some_work)
|
||||
{
|
||||
process_recieved_packet(bh, local_buff, conn_index);
|
||||
}else
|
||||
{
|
||||
//Idle when no work
|
||||
Sleep(30);
|
||||
}
|
||||
}
|
||||
|
||||
boost::interprocess::ipcdetail::atomic_dec32(&m_threads_count);
|
||||
LOG_PRINT_L3("[" << m_socket <<"] Socket handler thread stopped.[m_threads_count=" << m_threads_count << "]");
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user