1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
|
// 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.
//
#ifndef _LEVIN_CP_SERVER_H_
#define _LEVIN_CP_SERVER_H_
#include <winsock2.h>
#include <rpc.h>
#include <string>
#include <map>
#include <boost/shared_ptr.hpp>
#include "misc_log_ex.h"
//#include "threads_helper.h"
#include "syncobj.h"
#define ENABLE_PROFILING
#include "profile_tools.h"
#include "net_utils_base.h"
#include "pragma_comp_defs.h"
#define LEVIN_DEFAULT_DATA_BUFF_SIZE 2000
namespace epee
{
namespace net_utils
{
template<class TProtocol>
class cp_server_impl//: public abstract_handler
{
public:
cp_server_impl(/*abstract_handler* phandler = NULL*/);
virtual ~cp_server_impl();
bool init_server(int port_no);
bool deinit_server();
bool run_server(int threads_count = 0);
bool send_stop_signal();
bool is_stop_signal();
virtual bool on_net_idle(){return true;}
size_t get_active_connections_num();
typename TProtocol::config_type& get_config_object(){return m_config;}
private:
enum overlapped_operation_type
{
op_type_recv,
op_type_send,
op_type_stop
};
struct io_data_base
{
OVERLAPPED m_overlapped;
WSABUF DataBuf;
overlapped_operation_type m_op_type;
DWORD TotalBuffBytes;
volatile LONG m_is_in_use;
char Buffer[1];
};
PRAGMA_WARNING_PUSH
PRAGMA_WARNING_DISABLE_VS(4355)
template<class TProtocol>
struct connection: public net_utils::i_service_endpoint
{
connection(typename TProtocol::config_type& ref_config):m_sock(INVALID_SOCKET), m_tprotocol_handler(this, ref_config, context), m_psend_data(NULL), m_precv_data(NULL), m_asked_to_shutdown(0), m_connection_shutwoned(0)
{
}
//connection():m_sock(INVALID_SOCKET), m_tprotocol_handler(this, m_dummy_config, context), m_psend_data(NULL), m_precv_data(NULL), m_asked_to_shutdown(0), m_connection_shutwoned(0)
//{
//}
connection<TProtocol>& operator=(const connection<TProtocol>& obj)
{
return *this;
}
bool init_buffers()
{
m_psend_data = (io_data_base*)new char[sizeof(io_data_base) + LEVIN_DEFAULT_DATA_BUFF_SIZE-1];
m_psend_data->TotalBuffBytes = LEVIN_DEFAULT_DATA_BUFF_SIZE;
m_precv_data = (io_data_base*)new char[sizeof(io_data_base) + LEVIN_DEFAULT_DATA_BUFF_SIZE-1];
m_precv_data->TotalBuffBytes = LEVIN_DEFAULT_DATA_BUFF_SIZE;
return true;
}
bool query_shutdown()
{
if(!::InterlockedCompareExchange(&m_asked_to_shutdown, 1, 0))
{
m_psend_data->m_op_type = op_type_stop;
::PostQueuedCompletionStatus(m_completion_port, 0, (ULONG_PTR)this, &m_psend_data->m_overlapped);
}
return true;
}
//bool set_config(typename TProtocol::config_type& config)
//{
// this->~connection();
// new(this) connection<TProtocol>(config);
// return true;
//}
~connection()
{
if(m_psend_data)
delete m_psend_data;
if(m_precv_data)
delete m_precv_data;
}
virtual bool handle_send(const void* ptr, size_t cb)
{
PROFILE_FUNC("[handle_send]");
if(m_psend_data->TotalBuffBytes < cb)
resize_send_buff((DWORD)cb);
ZeroMemory(&m_psend_data->m_overlapped, sizeof(OVERLAPPED));
m_psend_data->DataBuf.len = (u_long)cb;//m_psend_data->TotalBuffBytes;
m_psend_data->DataBuf.buf = m_psend_data->Buffer;
memcpy(m_psend_data->DataBuf.buf, ptr, cb);
m_psend_data->m_op_type = op_type_send;
InterlockedExchange(&m_psend_data->m_is_in_use, 1);
DWORD bytes_sent = 0;
DWORD flags = 0;
int res = 0;
{
PROFILE_FUNC("[handle_send] ::WSASend");
res = ::WSASend(m_sock, &(m_psend_data->DataBuf), 1, &bytes_sent, flags, &(m_psend_data->m_overlapped), NULL);
}
if(res == SOCKET_ERROR )
{
int err = ::WSAGetLastError();
if(WSA_IO_PENDING == err )
return true;
}
LOG_ERROR("BIG FAIL: WSASend error code not correct, res=" << res << " last_err=" << err);
::InterlockedExchange(&m_psend_data->m_is_in_use, 0);
query_shutdown();
//closesocket(m_psend_data);
return false;
}else if(0 == res)
{
::InterlockedExchange(&m_psend_data->m_is_in_use, 0);
if(!bytes_sent || bytes_sent != cb)
{
int err = ::WSAGetLastError();
LOG_ERROR("BIG FAIL: WSASend immediatly complete? but bad results, res=" << res << " last_err=" << err);
query_shutdown();
return false;
}else
{
return true;
}
}
return true;
}
bool resize_send_buff(DWORD new_size)
{
if(m_psend_data->TotalBuffBytes >= new_size)
return true;
delete m_psend_data;
m_psend_data = (io_data_base*)new char[sizeof(io_data_base) + new_size-1];
m_psend_data->TotalBuffBytes = new_size;
LOG_PRINT("Connection buffer resized up to " << new_size, LOG_LEVEL_3);
return true;
}
SOCKET m_sock;
net_utils::connection_context_base context;
TProtocol m_tprotocol_handler;
typename TProtocol::config_type m_dummy_config;
io_data_base* m_precv_data;
io_data_base* m_psend_data;
HANDLE m_completion_port;
volatile LONG m_asked_to_shutdown;
volatile LONG m_connection_shutwoned;
};
PRAGMA_WARNING_POP
bool worker_thread_member();
static unsigned CALLBACK worker_thread(void* param);
bool add_new_connection(SOCKET new_sock, long ip_from, int port_from);
bool shutdown_connection(connection<TProtocol>* pconn);
typedef std::map<SOCKET, boost::shared_ptr<connection<TProtocol> > > connections_container;
SOCKET m_listen_socket;
HANDLE m_completion_port;
connections_container m_connections;
critical_section m_connections_lock;
int m_port;
volatile LONG m_stop;
//abstract_handler* m_phandler;
bool m_initialized;
volatile LONG m_worker_thread_counter;
typename TProtocol::config_type m_config;
};
}
}
#include "abstract_tcp_server_cp.inl"
#endif //_LEVIN_SERVER_H_
|
