Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include "io/tcp_session.h"
7 :
8 : #include <algorithm>
9 : #include <string>
10 :
11 : #include <boost/asio.hpp>
12 : #include <boost/asio/detail/socket_option.hpp>
13 : #include <boost/bind/bind.hpp>
14 : #include <boost/scoped_array.hpp>
15 : #include <boost/asio/detail/recycling_allocator.hpp>
16 :
17 : #include "base/logging.h"
18 : #include "base/address_util.h"
19 : #include "io/event_manager.h"
20 : #include "io/io_log.h"
21 : #include "io/io_utils.h"
22 : #include "io/tcp_message_write.h"
23 : #include "io/tcp_server.h"
24 : #include "base/address_util.h"
25 :
26 : using boost::asio::async_write;
27 : using boost::asio::buffer;
28 : using boost::asio::buffer_cast;
29 : using boost::asio::detail::socket_option::integer;
30 : using boost::asio::const_buffer;
31 : using boost::asio::mutable_buffer;
32 : using boost::asio::mutable_buffers_1;
33 : using boost::asio::null_buffers;
34 : using boost::asio::socket_base;
35 : using boost::bind;
36 : using boost::function;
37 : using boost::scoped_array;
38 : using boost::system::error_code;
39 : using std::min;
40 : using std::ostringstream;
41 : using std::string;
42 : using namespace boost::placeholders;
43 :
44 : using boost::asio::error::try_again;
45 : using boost::asio::error::would_block;
46 : using boost::asio::error::in_progress;
47 : using boost::asio::error::interrupted;
48 : using boost::asio::error::network_down;
49 : using boost::asio::error::network_reset;
50 : using boost::asio::error::network_unreachable;
51 : using boost::asio::error::no_buffer_space;
52 : using boost::asio::placeholders::error;
53 : using boost::asio::placeholders::bytes_transferred;
54 : using boost::asio::ip::tcp;
55 :
56 : int TcpSession::reader_task_id_ = -1;
57 :
58 : class TcpSession::Reader : public Task {
59 : public:
60 : typedef function<void(Buffer)> ReadHandler;
61 :
62 102428 : Reader(TcpSessionPtr session, ReadHandler read_fn, Buffer buffer)
63 409712 : : Task(session->reader_task_id(), session->GetSessionInstance()),
64 102428 : session_(session), read_fn_(read_fn), buffer_(buffer) {
65 102428 : }
66 102426 : virtual bool Run() {
67 102426 : if (session_->IsEstablished()) {
68 101189 : read_fn_(buffer_);
69 101188 : if (session_->IsReaderDeferred()) {
70 : // Update socket read block count.
71 1 : session_->stats_.read_block_start_time = UTCTimestampUsec();
72 1 : session_->stats_.read_blocked++;
73 1 : session_->server_->stats_.read_blocked++;
74 : } else {
75 101187 : session_->AsyncReadStart();
76 : }
77 : }
78 102428 : return true;
79 : }
80 0 : string Description() const { return "TcpSession::Reader"; }
81 :
82 : private:
83 : TcpSessionPtr session_;
84 : ReadHandler read_fn_;
85 : Buffer buffer_;
86 : };
87 :
88 29598 : TcpSession::TcpSession(
89 : TcpServer *server, Socket *socket, bool async_read_ready,
90 29598 : size_t buffer_send_size)
91 29599 : : server_(server),
92 29606 : socket_(socket),
93 29603 : read_on_connect_(async_read_ready),
94 29603 : established_(false),
95 29603 : closed_(false),
96 29600 : direction_(ACTIVE),
97 29590 : writer_(new TcpMessageWriter(this, buffer_send_size)),
98 88786 : name_("-") {
99 29578 : refcount_ = 0;
100 29607 : if (reader_task_id_ == -1) {
101 140 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
102 140 : reader_task_id_ = scheduler->GetTaskId("io::ReaderTask");
103 : }
104 29607 : if (server_) {
105 29591 : io_strand_.reset(new Strand(server->event_manager()->io_service()->get_executor()));
106 : }
107 29608 : defer_reader_ = false;
108 29609 : write_blocked_ = false;
109 29609 : tcp_close_in_progress_ = false;
110 29609 : }
111 :
112 29603 : TcpSession::~TcpSession() {
113 29603 : assert(!established_);
114 29603 : for (BufferQueue::iterator iter = buffer_queue_.begin();
115 36174 : iter != buffer_queue_.end(); ++iter) {
116 6573 : DeleteBuffer(*iter);
117 : }
118 29598 : buffer_queue_.clear();
119 29595 : }
120 :
121 1288920 : mutable_buffer TcpSession::AllocateBuffer(size_t buffer_size) {
122 1288920 : uint8_t *data = new uint8_t[buffer_size];
123 1288920 : mutable_buffer buffer = mutable_buffer(data, buffer_size);
124 1288920 : buffer_queue_.push_back(buffer);
125 1288920 : return buffer;
126 : }
127 :
128 1288881 : void TcpSession::DeleteBuffer(mutable_buffer buffer) {
129 1288881 : uint8_t *data = buffer_cast<uint8_t *>(buffer);
130 1288877 : delete[] data;
131 1288919 : }
132 :
133 1284619 : static int BufferCmp(const mutable_buffer &lhs, const const_buffer &rhs) {
134 1284619 : const uint8_t *lp = buffer_cast<uint8_t *>(lhs);
135 1284615 : const uint8_t *rp = buffer_cast<const uint8_t *>(rhs);
136 1284612 : if (lp < rp) {
137 2249 : return -1;
138 : }
139 1282363 : if (lp > rp) {
140 39 : return 1;
141 : }
142 1282324 : return 0;
143 : }
144 :
145 1270821 : void TcpSession::ReleaseBuffer(Buffer buffer) {
146 1270821 : std::scoped_lock lock(mutex_);
147 1270826 : ReleaseBufferLocked(buffer);
148 1270822 : }
149 :
150 1282339 : void TcpSession::ReleaseBufferLocked(Buffer buffer) {
151 1282339 : for (BufferQueue::iterator iter = buffer_queue_.begin();
152 1284627 : iter != buffer_queue_.end(); ++iter) {
153 1284623 : if (BufferCmp(*iter, buffer) == 0) {
154 1282324 : DeleteBuffer(*iter);
155 1282347 : buffer_queue_.erase(iter);
156 1282339 : return;
157 : }
158 : }
159 0 : assert(false);
160 : }
161 :
162 1300019 : void TcpSession::AsyncReadStartInternal(TcpSessionPtr session) {
163 : // Update socket read block time.
164 1300019 : if (stats_.read_block_start_time) {
165 1 : uint64_t blocked_usecs = UTCTimestampUsec() -
166 1 : stats_.read_block_start_time;
167 1 : stats_.read_block_start_time = 0;
168 1 : stats_.read_blocked_duration_usecs += blocked_usecs;
169 1 : server_->stats_.read_blocked_duration_usecs += blocked_usecs;
170 : }
171 :
172 1300019 : std::scoped_lock lock(mutex_);
173 1300019 : AsyncReadSome();
174 1300019 : }
175 :
176 1300010 : void TcpSession::AsyncReadStart() {
177 1300010 : if (io_strand_) {
178 1300008 : boost::asio::detail::recycling_allocator<void> allocator;
179 2600027 : io_strand_->post(bind(&TcpSession::AsyncReadStartInternal, this,
180 2600001 : TcpSessionPtr(this)), allocator);
181 : }
182 1300019 : }
183 :
184 2 : void TcpSession::SetDeferReader(bool defer_reader) {
185 2 : if (defer_reader_ != defer_reader) {
186 2 : defer_reader_ = defer_reader;
187 : // Call AsyncReadStart if reader was previously deferred
188 2 : if (!defer_reader_) {
189 1 : AsyncReadStart();
190 : }
191 : }
192 2 : }
193 :
194 269925 : void TcpSession::AsyncReadSome() {
195 269925 : if (IsEstablishedLocked()) {
196 537214 : socket()->async_read_some(null_buffers(),
197 537214 : bind(&TcpSession::AsyncReadHandler, TcpSessionPtr(this)));
198 : }
199 269925 : }
200 :
201 1021038 : void TcpSession::AsyncWrite(const uint8_t *data, std::size_t size) {
202 1021038 : async_write(*socket(), buffer(data, size),
203 2042076 : bind(&TcpSession::AsyncWriteHandler, TcpSessionPtr(this),
204 : error, bytes_transferred));
205 1021038 : }
206 :
207 123707 : TcpSession::Endpoint TcpSession::local_endpoint() const {
208 123707 : std::scoped_lock lock(mutex_);
209 123707 : if (!established_)
210 0 : return Endpoint();
211 :
212 123707 : error_code error;
213 123707 : Endpoint local = socket()->local_endpoint(error);
214 123707 : if (error) {
215 0 : return Endpoint();
216 : }
217 123707 : return local;
218 123707 : }
219 :
220 79339 : void TcpSession::set_observer(EventObserver observer) {
221 79339 : std::scoped_lock lock(obs_mutex_);
222 79342 : observer_ = observer;
223 79290 : }
224 :
225 27028 : void TcpSession::SetName() {
226 27028 : ostringstream out;
227 27028 : error_code error;
228 27028 : Endpoint local;
229 :
230 27028 : local = socket()->local_endpoint(error);
231 27028 : out << local.address().to_string() << ":" << local.port() << "::";
232 27028 : out << remote_.address().to_string() << ":" << remote_.port();
233 :
234 27028 : name_ = out.str();
235 :
236 27028 : out.str("");
237 27028 : std::string hostname = "";
238 27028 : if (local.address().is_v4()) {
239 27025 : hostname = ResolveCanonicalName(local.address().to_string());
240 : } else {
241 3 : hostname = ResolveCanonicalNameIPv6(local.address().to_string());
242 : }
243 27028 : out << hostname << ":" << remote_.address().to_string();
244 27028 : uve_key_str_ = out.str();
245 27028 : }
246 :
247 27028 : void TcpSession::SessionEstablished(Endpoint remote,
248 : Direction direction) {
249 27028 : established_ = true;
250 27028 : remote_ = remote;
251 27028 : remote_addr_str_ = remote.address().to_string();
252 27028 : direction_ = direction;
253 27028 : SetName();
254 27028 : }
255 :
256 13460 : void TcpSession::Accepted() {
257 13496 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_OUT,
258 : "Passive session Accept complete");
259 : {
260 13460 : std::scoped_lock obs_lock(obs_mutex_);
261 13460 : if (observer_) {
262 6622 : observer_(this, ACCEPT);
263 : }
264 13460 : }
265 :
266 13460 : if (read_on_connect_) {
267 251 : AsyncReadStart();
268 : }
269 13460 : }
270 :
271 13535 : bool TcpSession::Connected(Endpoint remote) {
272 13535 : assert(refcount_);
273 :
274 : {
275 13535 : std::scoped_lock lock(mutex_);
276 13535 : if (closed_) {
277 0 : return false;
278 : }
279 13535 : SessionEstablished(remote, TcpSession::ACTIVE);
280 13535 : }
281 13535 : SetSocketOptions();
282 :
283 13535 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_IN,
284 : "Active session connection complete");
285 :
286 : {
287 13535 : std::scoped_lock obs_lock(obs_mutex_);
288 13535 : if (observer_) {
289 13535 : observer_(this, CONNECT_COMPLETE);
290 : }
291 13535 : }
292 :
293 13535 : if (read_on_connect_) {
294 13535 : AsyncReadStart();
295 : }
296 13535 : return true;
297 : }
298 :
299 1372 : void TcpSession::ConnectFailed() {
300 1372 : std::scoped_lock obs_lock(obs_mutex_);
301 1372 : if (observer_) {
302 1274 : observer_(this, CONNECT_FAILED);
303 : }
304 1372 : }
305 :
306 : // Requires: lock must not be held
307 29624 : void TcpSession::CloseInternal(const error_code &ec,
308 : bool call_observer, bool notify_server) {
309 29624 : std::unique_lock<std::mutex> lock(mutex_);
310 :
311 29626 : if (socket() != NULL && !closed_) {
312 28501 : error_code error;
313 28501 : socket()->shutdown(tcp::socket::shutdown_both, error);
314 28509 : if (error) {
315 6021 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
316 : "Shutdown failed due to error: " << error.message());
317 : }
318 28509 : socket()->close(error);
319 : }
320 29629 : closed_ = true;
321 29629 : tcp_close_in_progress_ = false;
322 :
323 29628 : if (!established_) {
324 2601 : return;
325 : }
326 27027 : established_ = false;
327 :
328 : // copy the ec to close reason
329 27027 : close_reason_ = ec;
330 :
331 : // Take a reference through intrusive pointer to protect session from
332 : // possibly getting deleted from another thread.
333 27027 : TcpSessionPtr session = TcpSessionPtr(this);
334 27028 : lock.unlock();
335 :
336 27028 : if (call_observer) {
337 16373 : std::scoped_lock obs_lock(obs_mutex_);
338 16373 : if (observer_) {
339 11044 : observer_(this, CLOSE);
340 : }
341 16373 : }
342 :
343 27028 : if (notify_server) {
344 26995 : server_->OnSessionClose(this);
345 : }
346 29627 : }
347 :
348 1029876 : void TcpSession::TriggerAsyncReadHandler() {
349 1029876 : if (io_strand_) {
350 1029876 : boost::asio::detail::recycling_allocator<void> allocator;
351 2059752 : io_strand_->post(bind(&TcpSession::AsyncReadHandler,
352 2059752 : TcpSessionPtr(this)), allocator);
353 : }
354 1029876 : }
355 :
356 60808 : void TcpSession::Close() {
357 60808 : std::unique_lock<std::mutex> lock(mutex_);
358 :
359 : // Close can be called by application during cleanup. At this time
360 : // session may be already closed due to error and there may be write
361 : // data in the buffer, ignore if socket is closed.
362 60814 : if (closed_) {
363 39007 : return;
364 : }
365 :
366 21807 : if (server_ && writer_->IsWritePending()) {
367 8594 : tcp_close_in_progress_ = true;
368 8599 : return;
369 : }
370 13210 : lock.unlock();
371 :
372 13211 : error_code ec;
373 13211 : CloseInternal(ec, false);
374 60820 : }
375 :
376 : // virtual method overriden in derrived classes.
377 2 : void TcpSession::WriteReady(const error_code &error) {
378 2 : }
379 :
380 1828375 : void TcpSession::AsyncWriteHandler(TcpSessionPtr session,
381 : const error_code &error,
382 : std::size_t wrote) {
383 1828375 : std::unique_lock<std::mutex> lock(session->mutex_);
384 1828375 : if (session->IsSocketErrorHard(error)) {
385 54 : lock.unlock();
386 54 : TCP_SESSION_LOG_ERROR(session, TCP_DIR_OUT,
387 : "Write failed due to error: " << error.message());
388 54 : session->CloseInternal(error, true);
389 54 : return;
390 : }
391 :
392 : //
393 : // Ignore if connection is already closed.
394 : //
395 1828321 : if (session->IsClosedLocked()) return;
396 :
397 : // Update socket write bytes statistics.
398 1827675 : session->stats_.write_bytes += wrote;
399 1827675 : session->server_->stats_.write_bytes += wrote;
400 :
401 1827675 : bool send_ready = false;
402 1827675 : bool more_write = session->writer_->UpdateBufferQueue(wrote, &send_ready);
403 :
404 : // Subsequent write
405 1827675 : if (more_write) {
406 287397 : session->writer_->TriggerAsyncWrite();
407 1540278 : } else if (session->tcp_close_in_progress_) {
408 4809 : lock.unlock();
409 4809 : session->CloseInternal(error, true);
410 4809 : return;
411 : }
412 :
413 1822866 : lock.unlock();
414 1822866 : if (send_ready)
415 5 : session->WriteReady(error);
416 1822866 : return;
417 1828375 : }
418 :
419 1540995 : void TcpSession::AsyncWriteInternal(TcpSessionPtr session) {
420 :
421 1540995 : std::scoped_lock lock(session->mutex_);
422 :
423 : //
424 : // Ignore if connection is already closed.
425 : //
426 1540995 : if (session->IsClosedLocked()) return;
427 1540978 : session->writer_->TriggerAsyncWrite();
428 1540995 : }
429 :
430 1826236 : bool TcpSession::Send(const uint8_t *data, size_t size, size_t *sent) {
431 1826236 : bool ret = true;
432 1826236 : std::unique_lock<std::mutex> lock(mutex_);
433 :
434 : // Reset sent, if provided.
435 1826263 : if (sent) *sent = 0;
436 :
437 : //
438 : // If the session closed in the mean while, bail out
439 : // If session close is triggered, but close in progress, bail out
440 : //
441 1826263 : if (!IsEstablishedLocked()) return false;
442 :
443 1825619 : if (socket()->non_blocking()) {
444 1825544 : error_code error;
445 1825544 : int len = writer_->AsyncSend(data, size, &error);
446 1824975 : lock.unlock();
447 1825638 : if (len < 0) {
448 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
449 : "Write failed due to error: "
450 : << error.category().name() << " "
451 : << error.message());
452 0 : CloseInternal(error, true);
453 0 : return false;
454 : }
455 1825638 : if ((size_t) len != size)
456 15 : ret = false;
457 1825638 : if (sent) *sent = (len > 0) ? len : 0;
458 : }
459 1825638 : return ret;
460 1826234 : }
461 :
462 102428 : Task* TcpSession::CreateReaderTask(mutable_buffer buffer,
463 : size_t bytes_transferred) {
464 102428 : Buffer rdbuf(buffer_cast<const uint8_t *>(buffer), bytes_transferred);
465 102428 : Reader *task = new Reader(TcpSessionPtr(this),
466 102428 : bind(&TcpSession::OnRead, this, _1), rdbuf);
467 102428 : return (task);
468 : }
469 :
470 437702 : size_t TcpSession::ReadSome(mutable_buffer buffer, error_code *error) {
471 437702 : return socket()->read_some(mutable_buffers_1(buffer), *error);
472 : }
473 :
474 : // Tests with large data have shown large amounts of data being read in one
475 : // read_some() call, if available. Hence, allocate memory for all the bytes
476 : // available in the socket, but no less t han kDefaultBufferSize.
477 107397 : size_t TcpSession::GetReadBufferSize() const {
478 107397 : error_code error;
479 107397 : size_t size = socket_->available(error);
480 107397 : if (size < kDefaultBufferSize)
481 107042 : size = kDefaultBufferSize;
482 107397 : return size;
483 : }
484 :
485 1298434 : void TcpSession::AsyncReadHandler(TcpSessionPtr session) {
486 1298434 : std::unique_lock<std::mutex> lock(session->mutex_);
487 1298434 : if (session->closed_) {
488 9514 : return;
489 : }
490 :
491 : mutable_buffer buffer =
492 1288920 : session->AllocateBuffer(session->GetReadBufferSize());
493 :
494 1288920 : error_code error;
495 1288920 : size_t bytes_transferred = session->ReadSome(buffer, &error);
496 1288920 : if (session->IsSocketErrorHard(error)) {
497 11517 : session->ReleaseBufferLocked(buffer);
498 : // eof is returned when the peer closed the socket, no need to log error
499 11517 : if (error != boost::asio::error::eof) {
500 1451 : if (strcmp(error.category().name(), "asio.ssl") == 0 &&
501 0 : error.value() == SSL_SHORT_READ_ERROR) {
502 0 : TCP_SESSION_LOG_DEBUG(session, TCP_DIR_IN,
503 : "Read failed due to error "
504 : << error.category().name() << " "
505 : << error.value()
506 : << " : " << error.message());
507 : } else {
508 1451 : TCP_SESSION_LOG_ERROR(session, TCP_DIR_IN,
509 : "Read failed due to error "
510 : << error.category().name() << " "
511 : << error.value()
512 : << " : " << error.message());
513 : }
514 : }
515 11517 : lock.unlock();
516 11517 : session->CloseInternal(error, true);
517 11517 : return;
518 : }
519 :
520 : // Update read statistics.
521 1277403 : session->stats_.read_calls++;
522 1277403 : session->stats_.read_bytes += bytes_transferred;
523 1277403 : session->server_->stats_.read_calls++;
524 1277403 : session->server_->stats_.read_bytes += bytes_transferred;
525 :
526 1277403 : Task *task = session->CreateReaderTask(buffer, bytes_transferred);
527 : // Starting a new task for the session
528 1277403 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
529 1277403 : scheduler->Enqueue(task);
530 1298434 : }
531 :
532 4610 : int TcpSession::GetSessionInstance() const {
533 4610 : return Task::kTaskInstanceAny;
534 : }
535 :
536 :
537 20850 : int32_t TcpSession::local_port() const {
538 20850 : if (socket() == NULL) {
539 0 : return -1;
540 : }
541 20850 : error_code error;
542 20850 : Endpoint local = socket()->local_endpoint(error);
543 20848 : if (IsSocketErrorHard(error)) {
544 0 : return -1;
545 : }
546 20847 : return local.port();
547 : }
548 :
549 20847 : int32_t TcpSession::remote_port() const {
550 20847 : if (socket() == NULL) {
551 0 : return -1;
552 : }
553 20848 : error_code error;
554 20848 : Endpoint remote = socket()->remote_endpoint(error);
555 20849 : if (IsSocketErrorHard(error)) {
556 1242 : return -1;
557 : }
558 19607 : return remote.port();
559 : }
560 :
561 1405 : int TcpSession::SetMd5SocketOption(uint32_t peer_ip,
562 : const string &md5_password) {
563 1405 : return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip,
564 1405 : md5_password);
565 : }
566 :
567 0 : int TcpSession::ClearMd5SocketOption(uint32_t peer_ip) {
568 0 : return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip, "");
569 : }
570 :
571 4294 : int TcpSession::SetDscpSocketOption(uint8_t value) {
572 4294 : return server()->SetDscpSocketOption(socket()->native_handle(), value);
573 : }
574 :
575 968 : uint8_t TcpSession::GetDscpValue() const {
576 968 : return server_->GetDscpValue(socket()->native_handle());
577 : }
578 :
579 14660 : TcpMessageReader::TcpMessageReader(TcpSession *session,
580 14660 : ReceiveCallback callback)
581 14660 : : session_(session), callback_(callback), offset_(0), remain_(-1) {
582 14653 : }
583 :
584 14669 : TcpMessageReader::~TcpMessageReader() {
585 14669 : }
586 :
587 : // Returns a buffer allocation size that is larger than the message.
588 4 : int TcpMessageReader::AllocBufferSize(int length) {
589 4 : const int kMaxMessageSize = GetMaxMessageSize();
590 4 : if (length == -1) {
591 0 : return kMaxMessageSize;
592 : }
593 4 : int bufsize = 1 << 8;
594 6 : for (; bufsize < kMaxMessageSize && bufsize < length; bufsize <<= 1) {
595 : }
596 4 : return bufsize;
597 : }
598 :
599 4 : uint8_t *TcpMessageReader::BufferConcat(uint8_t *data, Buffer buffer,
600 : int msglength) {
601 4 : uint8_t *dst = data;
602 :
603 14 : while (!queue_.empty()) {
604 10 : Buffer head = queue_.front();
605 10 : const uint8_t *cp = TcpSession::BufferData(head) + offset_;
606 10 : int bytes = TcpSession::BufferSize(head) - offset_;
607 10 : assert((dst - data) + bytes < msglength);
608 10 : memcpy(dst, cp, bytes);
609 10 : dst += bytes;
610 10 : queue_.pop_front();
611 10 : session_->ReleaseBuffer(head);
612 10 : offset_ = 0;
613 10 : remain_ = -1;
614 : }
615 :
616 4 : int count = msglength - (dst - data);
617 4 : assert((dst - data) + count <= msglength);
618 4 : memcpy(dst, TcpSession::BufferData(buffer), count);
619 4 : offset_ = count;
620 :
621 4 : return data;
622 : }
623 :
624 6 : int TcpMessageReader::QueueByteLength() const {
625 6 : int total = 0;
626 6 : for (BufferQueue::const_iterator iter = queue_.begin();
627 18 : iter != queue_.end(); ++iter) {
628 12 : if (total == 0) {
629 6 : total = TcpSession::BufferSize(*iter) - offset_;
630 : } else {
631 6 : total += TcpSession::BufferSize(*iter);
632 : }
633 : }
634 6 : return total;
635 : }
636 :
637 4 : TcpMessageReader::Buffer TcpMessageReader::PullUp(
638 : uint8_t *data, Buffer buffer, size_t size) const {
639 4 : size_t offset = 0;
640 :
641 4 : for (BufferQueue::const_iterator iter = queue_.begin();
642 14 : iter != queue_.end(); ++iter) {
643 : const uint8_t *cp;
644 : int avail;
645 10 : if (offset == 0) {
646 4 : cp = TcpSession::BufferData(*iter) + offset_;
647 4 : avail = TcpSession::BufferSize(*iter) - offset_;
648 : } else {
649 6 : cp = TcpSession::BufferData(*iter);
650 6 : avail = TcpSession::BufferSize(*iter);
651 : }
652 10 : int remain = size - offset;
653 10 : avail = min(avail, remain);
654 10 : assert(offset + avail <= size);
655 10 : memcpy(data + offset, cp, avail);
656 10 : offset += avail;
657 : }
658 :
659 4 : int avail = TcpSession::BufferSize(buffer);
660 4 : int remain = size - offset;
661 4 : avail = min(avail, remain);
662 4 : assert(offset + avail <= size);
663 4 : memcpy(data + offset, TcpSession::BufferData(buffer), avail);
664 4 : offset += avail;
665 :
666 4 : if (offset < size) {
667 0 : return Buffer();
668 : }
669 4 : return Buffer(data, size);
670 : }
671 :
672 : // Read the socket stream and send messages to the peer object.
673 100595 : void TcpMessageReader::OnRead(Buffer buffer) {
674 100595 : const int kHeaderLenSize = GetHeaderLenSize();
675 100594 : size_t size = TcpSession::BufferSize(buffer);
676 100591 : TCP_SESSION_LOG_UT_DEBUG(session_, TCP_DIR_IN, "Read " << size << " bytes");
677 :
678 100594 : if (!queue_.empty()) {
679 10 : int msglength = MsgLength(queue_.front(), offset_);
680 10 : if (msglength < 0) {
681 6 : int queuelen = QueueByteLength();
682 6 : if (queuelen + static_cast<int>(size) < kHeaderLenSize) {
683 2 : queue_.push_back(buffer);
684 2 : return;
685 : }
686 4 : scoped_array<uint8_t> data(new uint8_t[kHeaderLenSize]);
687 4 : Buffer header = PullUp(data.get(), buffer, kHeaderLenSize);
688 4 : assert(TcpSession::BufferSize(header) == (size_t) kHeaderLenSize);
689 :
690 4 : msglength = MsgLength(header, 0);
691 4 : remain_ = msglength - queuelen;
692 4 : }
693 :
694 8 : assert(remain_ > 0);
695 8 : if (size < (size_t) remain_) {
696 4 : queue_.push_back(buffer);
697 4 : remain_ -= size;
698 4 : return;
699 : }
700 :
701 : // concat the buffers into a contiguous message.
702 4 : scoped_array<uint8_t> data(new uint8_t[AllocBufferSize(msglength)]);
703 4 : BufferConcat(data.get(), buffer, msglength);
704 4 : assert(remain_ == -1);
705 : // Receive the message
706 4 : bool success = callback_(data.get(), msglength);
707 4 : if (!success)
708 0 : return;
709 4 : }
710 :
711 100588 : int avail = size - offset_;
712 276961 : while (avail > 0) {
713 176403 : int msglength = MsgLength(buffer, offset_);
714 176399 : if (msglength < 0) {
715 2 : break;
716 : }
717 176397 : if (msglength > avail) {
718 2 : remain_ = msglength - avail;
719 2 : break;
720 : }
721 : // Receive the message
722 : bool success =
723 176395 : callback_(TcpSession::BufferData(buffer) + offset_, msglength);
724 176402 : offset_ += msglength;
725 176402 : avail -= msglength;
726 176402 : if (!success)
727 29 : return;
728 : }
729 :
730 100562 : if (avail > 0) {
731 4 : queue_.push_back(buffer);
732 : } else {
733 100558 : session_->ReleaseBuffer(buffer);
734 100559 : offset_ = 0;
735 100559 : assert(remain_ == -1);
736 : }
737 : }
738 :
739 : //
740 : // Check if a socker error is hard and fatal. Only then should we close the
741 : // socket. Soft errors like EINTR and EAGAIN should be ignored or properly
742 : // handled with retries
743 : //
744 3158987 : bool TcpSession::IsSocketErrorHard(const error_code &ec) {
745 3158987 : if (!ec)
746 3001425 : return false;
747 157561 : if (ec == try_again)
748 144750 : return false;
749 12813 : if (ec == would_block)
750 0 : return false;
751 12813 : if (ec == in_progress)
752 0 : return false;
753 12813 : if (ec == interrupted)
754 0 : return false;
755 12813 : if (ec == network_down)
756 0 : return false;
757 12813 : if (ec == network_reset)
758 0 : return false;
759 12813 : if (ec == network_unreachable)
760 0 : return false;
761 12813 : if (ec == no_buffer_space)
762 0 : return false;
763 :
764 12813 : return true;
765 : }
766 :
767 0 : error_code TcpSession::SetTcpNoDelay() {
768 0 : error_code ec;
769 0 : boost::asio::ip::tcp::no_delay no_delay_option(true);
770 0 : socket()->set_option(no_delay_option, ec);
771 0 : if (ec) {
772 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
773 : "tcp_no_delay set error: " << ec);
774 : }
775 0 : return ec;
776 : }
777 :
778 0 : error_code TcpSession::SetTcpSendBufSize(uint32_t size) {
779 0 : error_code ec;
780 0 : socket_base::send_buffer_size send_buffer_size_option(size);
781 0 : socket()->set_option(send_buffer_size_option, ec);
782 0 : if (ec) {
783 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
784 : "send_buffer_size set error: " << ec);
785 0 : return ec;
786 : }
787 :
788 0 : return ec;
789 : }
790 :
791 0 : error_code TcpSession::SetTcpRecvBufSize(uint32_t size) {
792 0 : error_code ec;
793 0 : socket_base::receive_buffer_size receive_buffer_size_option(size);
794 0 : socket()->set_option(receive_buffer_size_option, ec);
795 0 : if (ec) {
796 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_IN,
797 : "receive_buffer_size set error: " << ec);
798 0 : return ec;
799 : }
800 :
801 0 : return ec;
802 : }
803 :
804 6826 : error_code TcpSession::SetSocketKeepaliveOptions(int keepalive_time,
805 : int keepalive_intvl, int keepalive_probes, int tcp_user_timeout_val) {
806 6826 : error_code ec;
807 6826 : socket_base::keep_alive keep_alive_option(true);
808 6826 : socket()->set_option(keep_alive_option, ec);
809 6826 : if (ec) {
810 22 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
811 : "keep_alive set error: " << ec);
812 22 : return ec;
813 : }
814 : #ifdef TCP_KEEPIDLE
815 : typedef integer< IPPROTO_TCP, TCP_KEEPIDLE > keepalive_idle_time;
816 6804 : keepalive_idle_time keepalive_idle_time_option(keepalive_time);
817 6804 : socket()->set_option(keepalive_idle_time_option, ec);
818 6804 : if (ec) {
819 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
820 : "keepalive_idle_time: " << keepalive_time << " set error: " << ec);
821 0 : return ec;
822 : }
823 : #elif TCP_KEEPALIVE
824 : typedef integer< IPPROTO_TCP, TCP_KEEPALIVE > keepalive_idle_time;
825 : keepalive_idle_time keepalive_idle_time_option(keepalive_time);
826 : socket()->set_option(keepalive_idle_time_option, ec);
827 : if (ec) {
828 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
829 : "keepalive_idle_time: " << keepalive_time << " set error: " << ec);
830 : return ec;
831 : }
832 : #else
833 : #error No TCP keepalive option defined.
834 : #endif
835 : #ifdef TCP_KEEPINTVL
836 : typedef integer< IPPROTO_TCP, TCP_KEEPINTVL > keepalive_interval;
837 6804 : keepalive_interval keepalive_interval_option(keepalive_intvl);
838 6804 : socket()->set_option(keepalive_interval_option, ec);
839 6804 : if (ec) {
840 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
841 : "keepalive_interval: " << keepalive_intvl << " set error: " << ec);
842 0 : return ec;
843 : }
844 : #endif
845 : #ifdef TCP_KEEPCNT
846 : typedef integer< IPPROTO_TCP, TCP_KEEPCNT > keepalive_count;
847 6804 : keepalive_count keepalive_count_option(keepalive_probes);
848 6804 : socket()->set_option(keepalive_count_option, ec);
849 6804 : if (ec) {
850 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
851 : "keepalive_probes: " << keepalive_probes << " set error: " << ec);
852 0 : return ec;
853 : }
854 : #endif
855 : #ifdef TCP_USER_TIMEOUT
856 : typedef integer< IPPROTO_TCP, TCP_USER_TIMEOUT > tcp_user_timeout;
857 6804 : tcp_user_timeout tcp_user_timeout_option(tcp_user_timeout_val);
858 6804 : socket()->set_option(tcp_user_timeout_option, ec);
859 6804 : if (ec) {
860 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
861 : "tcp_user_timeout: " << tcp_user_timeout_val << " set error: "
862 : << ec);
863 0 : return ec;
864 : }
865 : #endif
866 :
867 6804 : return ec;
868 : }
869 :
870 42032 : error_code TcpSession::SetSocketOptions() {
871 42032 : error_code ec;
872 :
873 : //
874 : // Make socket write non-blocking
875 : //
876 42032 : socket()->non_blocking(true, ec);
877 42032 : if (ec) {
878 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_NA,
879 : "Cannot set socket non blocking: " << ec);
880 0 : return ec;
881 : }
882 :
883 42032 : char *buffer_size_str = getenv("TCP_SESSION_SOCKET_BUFFER_SIZE");
884 42026 : if (!buffer_size_str) return ec;
885 :
886 12 : uint32_t sz = static_cast<uint32_t>(strtoul(buffer_size_str, NULL, 0));
887 18 : if (sz) {
888 : //
889 : // Set socket send and receive buffer size
890 : //
891 : // Currently used only under test environments to trigger partial
892 : // sends more deterministically
893 : //
894 18 : socket_base::send_buffer_size send_buffer_size_option(sz);
895 18 : socket()->set_option(send_buffer_size_option, ec);
896 18 : if (ec) {
897 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
898 : "send_buffer_size set error: " << ec);
899 0 : return ec;
900 : }
901 :
902 18 : socket_base::receive_buffer_size receive_buffer_size_option(sz);
903 18 : socket()->set_option(receive_buffer_size_option, ec);
904 18 : if (ec) {
905 0 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_IN,
906 : "receive_buffer_size set error: " << ec);
907 0 : return ec;
908 : }
909 : }
910 :
911 18 : return ec;
912 : }
913 :
914 8 : void TcpSession::GetRxSocketStats(SocketIOStats *socket_stats) const {
915 8 : stats_.GetRxStats(socket_stats);
916 8 : }
917 :
918 6 : void TcpSession::GetTxSocketStats(SocketIOStats *socket_stats) const {
919 6 : stats_.GetTxStats(socket_stats);
920 6 : }
|
