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 102748 : Reader(TcpSessionPtr session, ReadHandler read_fn, Buffer buffer)
62 410992 : : Task(session->reader_task_id(), session->GetSessionInstance()),
63 102748 : session_(session), read_fn_(read_fn), buffer_(buffer) {
64 102748 : }
65 102744 : virtual bool Run() {
66 102744 : if (session_->IsEstablished()) {
67 101486 : read_fn_(buffer_);
68 101487 : 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 101486 : session_->AsyncReadStart();
75 : }
76 : }
77 102748 : 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 32501 : TcpSession::TcpSession(
88 : TcpServer *server, Socket *socket, bool async_read_ready,
89 32501 : size_t buffer_send_size)
90 32505 : : server_(server),
91 32510 : socket_(socket),
92 32509 : read_on_connect_(async_read_ready),
93 32509 : established_(false),
94 32509 : closed_(false),
95 32507 : direction_(ACTIVE),
96 32494 : writer_(new TcpMessageWriter(this, buffer_send_size)),
97 97510 : name_("-") {
98 32500 : refcount_ = 0;
99 32510 : if (reader_task_id_ == -1) {
100 135 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
101 135 : reader_task_id_ = scheduler->GetTaskId("io::ReaderTask");
102 : }
103 32511 : if (server_) {
104 32504 : io_strand_.reset(new Strand(server->event_manager()->io_service()->get_executor()));
105 : }
106 32511 : defer_reader_ = false;
107 32511 : write_blocked_ = false;
108 32511 : tcp_close_in_progress_ = false;
109 32511 : }
110 :
111 32505 : TcpSession::~TcpSession() {
112 32505 : assert(!established_);
113 32505 : for (BufferQueue::iterator iter = buffer_queue_.begin();
114 39215 : iter != buffer_queue_.end(); ++iter) {
115 6710 : DeleteBuffer(*iter);
116 : }
117 32505 : buffer_queue_.clear();
118 32506 : }
119 :
120 1320141 : mutable_buffer TcpSession::AllocateBuffer(size_t buffer_size) {
121 1320141 : uint8_t *data = new uint8_t[buffer_size];
122 1320141 : mutable_buffer buffer = mutable_buffer(data, buffer_size);
123 1320141 : buffer_queue_.push_back(buffer);
124 1320141 : return buffer;
125 : }
126 :
127 1320118 : void TcpSession::DeleteBuffer(mutable_buffer buffer) {
128 1320118 : uint8_t *data = buffer_cast<uint8_t *>(buffer);
129 1320116 : delete[] data;
130 1320141 : }
131 :
132 1315701 : static int BufferCmp(const mutable_buffer &lhs, const const_buffer &rhs) {
133 1315701 : const uint8_t *lp = buffer_cast<uint8_t *>(lhs);
134 1315703 : const uint8_t *rp = buffer_cast<const uint8_t *>(rhs);
135 1315701 : if (lp < rp) {
136 2233 : return -1;
137 : }
138 1313468 : if (lp > rp) {
139 52 : return 1;
140 : }
141 1313416 : return 0;
142 : }
143 :
144 1300733 : void TcpSession::ReleaseBuffer(Buffer buffer) {
145 1300733 : std::scoped_lock lock(mutex_);
146 1300744 : ReleaseBufferLocked(buffer);
147 1300739 : }
148 :
149 1313424 : void TcpSession::ReleaseBufferLocked(Buffer buffer) {
150 1313424 : for (BufferQueue::iterator iter = buffer_queue_.begin();
151 1315704 : iter != buffer_queue_.end(); ++iter) {
152 1315702 : if (BufferCmp(*iter, buffer) == 0) {
153 1313417 : DeleteBuffer(*iter);
154 1313431 : buffer_queue_.erase(iter);
155 1313421 : return;
156 : }
157 : }
158 0 : assert(false);
159 : }
160 :
161 1331058 : void TcpSession::AsyncReadStartInternal(TcpSessionPtr session) {
162 : // Update socket read block time.
163 1331058 : 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 1331058 : std::scoped_lock lock(mutex_);
172 1331058 : AsyncReadSome();
173 1331058 : }
174 :
175 1331040 : void TcpSession::AsyncReadStart() {
176 1331040 : if (io_strand_) {
177 1331036 : boost::asio::detail::recycling_allocator<void> allocator;
178 2662094 : io_strand_->post(bind(&TcpSession::AsyncReadStartInternal, this,
179 2662069 : TcpSessionPtr(this)), allocator);
180 : }
181 1331059 : }
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 273963 : void TcpSession::AsyncReadSome() {
194 273963 : if (IsEstablishedLocked()) {
195 545190 : socket()->async_read_some(null_buffers(),
196 545190 : bind(&TcpSession::AsyncReadHandler, TcpSessionPtr(this)));
197 : }
198 273963 : }
199 :
200 1029835 : void TcpSession::AsyncWrite(const uint8_t *data, std::size_t size) {
201 1029835 : async_write(*socket(), buffer(data, size),
202 2059670 : bind(&TcpSession::AsyncWriteHandler, TcpSessionPtr(this),
203 : error, bytes_transferred));
204 1029835 : }
205 :
206 118749 : TcpSession::Endpoint TcpSession::local_endpoint() const {
207 118749 : std::scoped_lock lock(mutex_);
208 118749 : if (!established_)
209 0 : return Endpoint();
210 :
211 118749 : error_code error;
212 118749 : Endpoint local = socket()->local_endpoint(error);
213 118749 : if (error) {
214 0 : return Endpoint();
215 : }
216 118749 : return local;
217 118749 : }
218 :
219 83993 : void TcpSession::set_observer(EventObserver observer) {
220 83993 : std::scoped_lock lock(obs_mutex_);
221 83999 : observer_ = observer;
222 83956 : }
223 :
224 29401 : void TcpSession::SetName() {
225 29401 : ostringstream out;
226 29401 : error_code error;
227 29401 : Endpoint local;
228 :
229 29401 : local = socket()->local_endpoint(error);
230 29401 : out << local.address().to_string() << ":" << local.port() << "::";
231 29401 : out << remote_.address().to_string() << ":" << remote_.port();
232 :
233 29401 : name_ = out.str();
234 :
235 29401 : out.str("");
236 29401 : std::string hostname = "";
237 29401 : if (local.address().is_v4()) {
238 29398 : hostname = ResolveCanonicalName(local.address().to_string());
239 : } else {
240 3 : hostname = ResolveCanonicalNameIPv6(local.address().to_string());
241 : }
242 29401 : out << hostname << ":" << remote_.address().to_string();
243 29401 : uve_key_str_ = out.str();
244 29401 : }
245 :
246 29401 : void TcpSession::SessionEstablished(Endpoint remote,
247 : Direction direction) {
248 29401 : established_ = true;
249 29401 : remote_ = remote;
250 29401 : remote_addr_str_ = remote.address().to_string();
251 29401 : direction_ = direction;
252 29401 : SetName();
253 29401 : }
254 :
255 14641 : void TcpSession::Accepted() {
256 14677 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_OUT,
257 : "Passive session Accept complete");
258 : {
259 14641 : std::scoped_lock obs_lock(obs_mutex_);
260 14641 : if (observer_) {
261 6630 : observer_(this, ACCEPT);
262 : }
263 14641 : }
264 :
265 14641 : if (read_on_connect_) {
266 251 : AsyncReadStart();
267 : }
268 14641 : }
269 :
270 14724 : bool TcpSession::Connected(Endpoint remote) {
271 14724 : assert(refcount_);
272 :
273 : {
274 14724 : std::scoped_lock lock(mutex_);
275 14724 : if (closed_) {
276 0 : return false;
277 : }
278 14724 : SessionEstablished(remote, TcpSession::ACTIVE);
279 14724 : }
280 14724 : SetSocketOptions();
281 :
282 14724 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_IN,
283 : "Active session connection complete");
284 :
285 : {
286 14724 : std::scoped_lock obs_lock(obs_mutex_);
287 14724 : if (observer_) {
288 14724 : observer_(this, CONNECT_COMPLETE);
289 : }
290 14724 : }
291 :
292 14724 : if (read_on_connect_) {
293 14724 : AsyncReadStart();
294 : }
295 14724 : return true;
296 : }
297 :
298 1901 : void TcpSession::ConnectFailed() {
299 1901 : std::scoped_lock obs_lock(obs_mutex_);
300 1901 : if (observer_) {
301 1804 : observer_(this, CONNECT_FAILED);
302 : }
303 1901 : }
304 :
305 : // Requires: lock must not be held
306 32527 : void TcpSession::CloseInternal(const error_code &ec,
307 : bool call_observer, bool notify_server) {
308 32527 : std::unique_lock<std::mutex> lock(mutex_);
309 :
310 32529 : if (socket() != NULL && !closed_) {
311 31407 : error_code error;
312 31407 : socket()->shutdown(tcp::socket::shutdown_both, error);
313 31412 : if (error) {
314 7571 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
315 : "Shutdown failed due to error: " << error.message());
316 : }
317 31411 : socket()->close(error);
318 : }
319 32531 : closed_ = true;
320 32531 : tcp_close_in_progress_ = false;
321 :
322 32531 : if (!established_) {
323 3130 : return;
324 : }
325 29401 : established_ = false;
326 :
327 : // copy the ec to close reason
328 29401 : close_reason_ = ec;
329 :
330 : // Take a reference through intrusive pointer to protect session from
331 : // possibly getting deleted from another thread.
332 29401 : TcpSessionPtr session = TcpSessionPtr(this);
333 29401 : lock.unlock();
334 :
335 29401 : if (call_observer) {
336 18330 : std::scoped_lock obs_lock(obs_mutex_);
337 18330 : if (observer_) {
338 12237 : observer_(this, CLOSE);
339 : }
340 18330 : }
341 :
342 29401 : if (notify_server) {
343 29364 : server_->OnSessionClose(this);
344 : }
345 32531 : }
346 :
347 1056908 : void TcpSession::TriggerAsyncReadHandler() {
348 1056908 : if (io_strand_) {
349 1056908 : boost::asio::detail::recycling_allocator<void> allocator;
350 2113816 : io_strand_->post(bind(&TcpSession::AsyncReadHandler,
351 2113816 : TcpSessionPtr(this)), allocator);
352 : }
353 1056908 : }
354 :
355 65214 : void TcpSession::Close() {
356 65214 : 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 65218 : if (closed_) {
362 41719 : return;
363 : }
364 :
365 23499 : if (server_ && writer_->IsWritePending()) {
366 9345 : tcp_close_in_progress_ = true;
367 9349 : return;
368 : }
369 14152 : lock.unlock();
370 :
371 14151 : error_code ec;
372 14151 : CloseInternal(ec, false);
373 65221 : }
374 :
375 : // virtual method overriden in derrived classes.
376 2 : void TcpSession::WriteReady(const error_code &error) {
377 2 : }
378 :
379 1860807 : void TcpSession::AsyncWriteHandler(TcpSessionPtr session,
380 : const error_code &error,
381 : std::size_t wrote) {
382 1860807 : std::unique_lock<std::mutex> lock(session->mutex_);
383 1860807 : if (session->IsSocketErrorHard(error)) {
384 51 : lock.unlock();
385 51 : TCP_SESSION_LOG_ERROR(session, TCP_DIR_OUT,
386 : "Write failed due to error: " << error.message());
387 51 : session->CloseInternal(error, true);
388 51 : return;
389 : }
390 :
391 : //
392 : // Ignore if connection is already closed.
393 : //
394 1860756 : if (session->IsClosedLocked()) return;
395 :
396 : // Update socket write bytes statistics.
397 1860123 : session->stats_.write_bytes += wrote;
398 1860123 : session->server_->stats_.write_bytes += wrote;
399 :
400 1860123 : bool send_ready = false;
401 1860123 : bool more_write = session->writer_->UpdateBufferQueue(wrote, &send_ready);
402 :
403 : // Subsequent write
404 1860123 : if (more_write) {
405 311040 : session->writer_->TriggerAsyncWrite();
406 1549083 : } else if (session->tcp_close_in_progress_) {
407 5606 : lock.unlock();
408 5606 : session->CloseInternal(error, true);
409 5606 : return;
410 : }
411 :
412 1854517 : lock.unlock();
413 1854517 : if (send_ready)
414 5 : session->WriteReady(error);
415 1854517 : return;
416 1860807 : }
417 :
418 1549779 : void TcpSession::AsyncWriteInternal(TcpSessionPtr session) {
419 :
420 1549779 : std::scoped_lock lock(session->mutex_);
421 :
422 : //
423 : // Ignore if connection is already closed.
424 : //
425 1549779 : if (session->IsClosedLocked()) return;
426 1549767 : session->writer_->TriggerAsyncWrite();
427 1549779 : }
428 :
429 1858729 : bool TcpSession::Send(const uint8_t *data, size_t size, size_t *sent) {
430 1858729 : bool ret = true;
431 1858729 : std::unique_lock<std::mutex> lock(mutex_);
432 :
433 : // Reset sent, if provided.
434 1858758 : 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 1858758 : if (!IsEstablishedLocked()) return false;
441 :
442 1858060 : if (socket()->non_blocking()) {
443 1857974 : error_code error;
444 1857974 : int len = writer_->AsyncSend(data, size, &error);
445 1857395 : lock.unlock();
446 1858054 : 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 1858054 : if ((size_t) len != size)
455 15 : ret = false;
456 1858054 : if (sent) *sent = (len > 0) ? len : 0;
457 : }
458 1858053 : return ret;
459 1858712 : }
460 :
461 102748 : Task* TcpSession::CreateReaderTask(mutable_buffer buffer,
462 : size_t bytes_transferred) {
463 102748 : Buffer rdbuf(buffer_cast<const uint8_t *>(buffer), bytes_transferred);
464 102748 : Reader *task = new Reader(TcpSessionPtr(this),
465 102748 : bind(&TcpSession::OnRead, this, _1), rdbuf);
466 102748 : return (task);
467 : }
468 :
469 445513 : size_t TcpSession::ReadSome(mutable_buffer buffer, error_code *error) {
470 445513 : 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 107870 : size_t TcpSession::GetReadBufferSize() const {
477 107870 : error_code error;
478 107870 : size_t size = socket_->available(error);
479 107870 : if (size < kDefaultBufferSize)
480 107536 : size = kDefaultBufferSize;
481 107870 : return size;
482 : }
483 :
484 1329450 : void TcpSession::AsyncReadHandler(TcpSessionPtr session) {
485 1329450 : std::unique_lock<std::mutex> lock(session->mutex_);
486 1329450 : if (session->closed_) {
487 9309 : return;
488 : }
489 :
490 : mutable_buffer buffer =
491 1320141 : session->AllocateBuffer(session->GetReadBufferSize());
492 :
493 1320141 : error_code error;
494 1320141 : size_t bytes_transferred = session->ReadSome(buffer, &error);
495 1320141 : if (session->IsSocketErrorHard(error)) {
496 12683 : session->ReleaseBufferLocked(buffer);
497 : // eof is returned when the peer closed the socket, no need to log error
498 12683 : if (error != boost::asio::error::eof) {
499 1454 : 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 1454 : 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 12683 : lock.unlock();
515 12683 : session->CloseInternal(error, true);
516 12683 : return;
517 : }
518 :
519 : // Update read statistics.
520 1307458 : session->stats_.read_calls++;
521 1307458 : session->stats_.read_bytes += bytes_transferred;
522 1307458 : session->server_->stats_.read_calls++;
523 1307458 : session->server_->stats_.read_bytes += bytes_transferred;
524 :
525 1307458 : Task *task = session->CreateReaderTask(buffer, bytes_transferred);
526 : // Starting a new task for the session
527 1307458 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
528 1307458 : scheduler->Enqueue(task);
529 1329450 : }
530 :
531 4667 : int TcpSession::GetSessionInstance() const {
532 4667 : return Task::kTaskInstanceAny;
533 : }
534 :
535 :
536 23413 : int32_t TcpSession::local_port() const {
537 23413 : if (socket() == NULL) {
538 0 : return -1;
539 : }
540 23413 : error_code error;
541 23413 : Endpoint local = socket()->local_endpoint(error);
542 23412 : if (IsSocketErrorHard(error)) {
543 0 : return -1;
544 : }
545 23411 : return local.port();
546 : }
547 :
548 23410 : int32_t TcpSession::remote_port() const {
549 23410 : if (socket() == NULL) {
550 0 : return -1;
551 : }
552 23410 : error_code error;
553 23410 : Endpoint remote = socket()->remote_endpoint(error);
554 23414 : if (IsSocketErrorHard(error)) {
555 1770 : return -1;
556 : }
557 21644 : return remote.port();
558 : }
559 :
560 1275 : int TcpSession::SetMd5SocketOption(uint32_t peer_ip,
561 : const string &md5_password) {
562 1275 : return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip,
563 1275 : 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 4153 : int TcpSession::SetDscpSocketOption(uint8_t value) {
571 4153 : 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 15999 : TcpMessageReader::TcpMessageReader(TcpSession *session,
579 15999 : ReceiveCallback callback)
580 15999 : : session_(session), callback_(callback), offset_(0), remain_(-1) {
581 16003 : }
582 :
583 16013 : TcpMessageReader::~TcpMessageReader() {
584 16013 : }
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 100886 : void TcpMessageReader::OnRead(Buffer buffer) {
673 100886 : const int kHeaderLenSize = GetHeaderLenSize();
674 100887 : size_t size = TcpSession::BufferSize(buffer);
675 100887 : TCP_SESSION_LOG_UT_DEBUG(session_, TCP_DIR_IN, "Read " << size << " bytes");
676 :
677 100887 : 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 100881 : int avail = size - offset_;
711 277181 : while (avail > 0) {
712 176336 : int msglength = MsgLength(buffer, offset_);
713 176335 : if (msglength < 0) {
714 2 : break;
715 : }
716 176333 : if (msglength > avail) {
717 2 : remain_ = msglength - avail;
718 2 : break;
719 : }
720 : // Receive the message
721 : bool success =
722 176331 : callback_(TcpSession::BufferData(buffer) + offset_, msglength);
723 176335 : offset_ += msglength;
724 176335 : avail -= msglength;
725 176335 : if (!success)
726 35 : return;
727 : }
728 :
729 100849 : if (avail > 0) {
730 4 : queue_.push_back(buffer);
731 : } else {
732 100845 : session_->ReleaseBuffer(buffer);
733 100846 : offset_ = 0;
734 100846 : 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 3227770 : bool TcpSession::IsSocketErrorHard(const error_code &ec) {
744 3227770 : if (!ec)
745 3065766 : return false;
746 162002 : if (ec == try_again)
747 147500 : return false;
748 14504 : if (ec == would_block)
749 0 : return false;
750 14504 : if (ec == in_progress)
751 0 : return false;
752 14504 : if (ec == interrupted)
753 0 : return false;
754 14504 : if (ec == network_down)
755 0 : return false;
756 14504 : if (ec == network_reset)
757 0 : return false;
758 14504 : if (ec == network_unreachable)
759 0 : return false;
760 14504 : if (ec == no_buffer_space)
761 0 : return false;
762 :
763 14504 : 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 6841 : error_code TcpSession::SetSocketKeepaliveOptions(int keepalive_time,
804 : int keepalive_intvl, int keepalive_probes, int tcp_user_timeout_val) {
805 6841 : error_code ec;
806 6841 : socket_base::keep_alive keep_alive_option(true);
807 6841 : socket()->set_option(keep_alive_option, ec);
808 6841 : 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 6819 : keepalive_idle_time keepalive_idle_time_option(keepalive_time);
816 6819 : socket()->set_option(keepalive_idle_time_option, ec);
817 6819 : 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 6819 : keepalive_interval keepalive_interval_option(keepalive_intvl);
837 6819 : socket()->set_option(keepalive_interval_option, ec);
838 6819 : 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 6819 : keepalive_count keepalive_count_option(keepalive_probes);
847 6819 : socket()->set_option(keepalive_count_option, ec);
848 6819 : 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 6819 : tcp_user_timeout tcp_user_timeout_option(tcp_user_timeout_val);
857 6819 : socket()->set_option(tcp_user_timeout_option, ec);
858 6819 : 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 6819 : return ec;
867 : }
868 :
869 46120 : error_code TcpSession::SetSocketOptions() {
870 46120 : error_code ec;
871 :
872 : //
873 : // Make socket write non-blocking
874 : //
875 46120 : socket()->non_blocking(true, ec);
876 46123 : 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 46123 : char *buffer_size_str = getenv("TCP_SESSION_SOCKET_BUFFER_SIZE");
883 46121 : 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 : }
|
