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