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 106345 : Reader(TcpSessionPtr session, ReadHandler read_fn, Buffer buffer)
62 425380 : : Task(session->reader_task_id(), session->GetSessionInstance()),
63 106345 : session_(session), read_fn_(read_fn), buffer_(buffer) {
64 106345 : }
65 106342 : virtual bool Run() {
66 106342 : if (session_->IsEstablished()) {
67 104957 : read_fn_(buffer_);
68 104955 : 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 104954 : session_->AsyncReadStart();
75 : }
76 : }
77 106341 : 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 32089 : TcpSession::TcpSession(
88 : TcpServer *server, Socket *socket, bool async_read_ready,
89 32089 : size_t buffer_send_size)
90 32089 : : server_(server),
91 32095 : socket_(socket),
92 32094 : read_on_connect_(async_read_ready),
93 32094 : established_(false),
94 32094 : closed_(false),
95 32093 : direction_(ACTIVE),
96 32075 : writer_(new TcpMessageWriter(this, buffer_send_size)),
97 96255 : name_("-") {
98 32071 : refcount_ = 0;
99 32095 : if (reader_task_id_ == -1) {
100 132 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
101 130 : reader_task_id_ = scheduler->GetTaskId("io::ReaderTask");
102 : }
103 32094 : if (server_) {
104 32086 : io_strand_.reset(new Strand(server->event_manager()->io_service()->get_executor()));
105 : }
106 32094 : defer_reader_ = false;
107 32095 : write_blocked_ = false;
108 32095 : tcp_close_in_progress_ = false;
109 32094 : }
110 :
111 32081 : TcpSession::~TcpSession() {
112 32081 : assert(!established_);
113 32081 : for (BufferQueue::iterator iter = buffer_queue_.begin();
114 38792 : iter != buffer_queue_.end(); ++iter) {
115 6712 : DeleteBuffer(*iter);
116 : }
117 32078 : buffer_queue_.clear();
118 32077 : }
119 :
120 1334538 : mutable_buffer TcpSession::AllocateBuffer(size_t buffer_size) {
121 1334538 : uint8_t *data = new uint8_t[buffer_size];
122 1334538 : mutable_buffer buffer = mutable_buffer(data, buffer_size);
123 1334538 : buffer_queue_.push_back(buffer);
124 1334538 : return buffer;
125 : }
126 :
127 1334488 : void TcpSession::DeleteBuffer(mutable_buffer buffer) {
128 1334488 : uint8_t *data = buffer_cast<uint8_t *>(buffer);
129 1334486 : delete[] data;
130 1334538 : }
131 :
132 1330074 : static int BufferCmp(const mutable_buffer &lhs, const const_buffer &rhs) {
133 1330074 : const uint8_t *lp = buffer_cast<uint8_t *>(lhs);
134 1330073 : const uint8_t *rp = buffer_cast<const uint8_t *>(rhs);
135 1330074 : if (lp < rp) {
136 2227 : return -1;
137 : }
138 1327847 : if (lp > rp) {
139 59 : return 1;
140 : }
141 1327788 : return 0;
142 : }
143 :
144 1315337 : void TcpSession::ReleaseBuffer(Buffer buffer) {
145 1315337 : std::scoped_lock lock(mutex_);
146 1315352 : ReleaseBufferLocked(buffer);
147 1315336 : }
148 :
149 1327817 : void TcpSession::ReleaseBufferLocked(Buffer buffer) {
150 1327817 : for (BufferQueue::iterator iter = buffer_queue_.begin();
151 1330088 : iter != buffer_queue_.end(); ++iter) {
152 1330079 : if (BufferCmp(*iter, buffer) == 0) {
153 1327787 : DeleteBuffer(*iter);
154 1327825 : buffer_queue_.erase(iter);
155 1327802 : return;
156 : }
157 : }
158 0 : assert(false);
159 : }
160 :
161 1346003 : void TcpSession::AsyncReadStartInternal(TcpSessionPtr session) {
162 : // Update socket read block time.
163 1346003 : 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 1346003 : std::scoped_lock lock(mutex_);
172 1346003 : AsyncReadSome();
173 1346003 : }
174 :
175 1345968 : void TcpSession::AsyncReadStart() {
176 1345968 : if (io_strand_) {
177 1345972 : boost::asio::detail::recycling_allocator<void> allocator;
178 2691975 : io_strand_->post(bind(&TcpSession::AsyncReadStartInternal, this,
179 2691936 : TcpSessionPtr(this)), allocator);
180 : }
181 1345996 : }
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 278972 : void TcpSession::AsyncReadSome() {
194 278972 : if (IsEstablishedLocked()) {
195 554998 : socket()->async_read_some(null_buffers(),
196 554998 : bind(&TcpSession::AsyncReadHandler, TcpSessionPtr(this)));
197 : }
198 278972 : }
199 :
200 1044476 : void TcpSession::AsyncWrite(const uint8_t *data, std::size_t size) {
201 1044476 : async_write(*socket(), buffer(data, size),
202 2088952 : bind(&TcpSession::AsyncWriteHandler, TcpSessionPtr(this),
203 : error, bytes_transferred));
204 1044476 : }
205 :
206 250402 : TcpSession::Endpoint TcpSession::local_endpoint() const {
207 250402 : std::scoped_lock lock(mutex_);
208 250402 : if (!established_)
209 0 : return Endpoint();
210 :
211 250402 : error_code error;
212 250402 : Endpoint local = socket()->local_endpoint(error);
213 250402 : if (error) {
214 0 : return Endpoint();
215 : }
216 250402 : return local;
217 250402 : }
218 :
219 83200 : void TcpSession::set_observer(EventObserver observer) {
220 83200 : std::scoped_lock lock(obs_mutex_);
221 83211 : observer_ = observer;
222 83157 : }
223 :
224 29288 : void TcpSession::SetName() {
225 29288 : ostringstream out;
226 29288 : error_code error;
227 29288 : Endpoint local;
228 :
229 29288 : local = socket()->local_endpoint(error);
230 29288 : out << local.address().to_string() << ":" << local.port() << "::";
231 29288 : out << remote_.address().to_string() << ":" << remote_.port();
232 :
233 29288 : name_ = out.str();
234 :
235 29288 : out.str("");
236 29288 : std::string hostname = "";
237 29288 : if (local.address().is_v4()) {
238 29285 : hostname = ResolveCanonicalName(local.address().to_string());
239 : } else {
240 3 : hostname = ResolveCanonicalNameIPv6(local.address().to_string());
241 : }
242 29288 : out << hostname << ":" << remote_.address().to_string();
243 29288 : uve_key_str_ = out.str();
244 29288 : }
245 :
246 29288 : void TcpSession::SessionEstablished(Endpoint remote,
247 : Direction direction) {
248 29288 : established_ = true;
249 29288 : remote_ = remote;
250 29288 : remote_addr_str_ = remote.address().to_string();
251 29288 : direction_ = direction;
252 29288 : SetName();
253 29288 : }
254 :
255 14577 : void TcpSession::Accepted() {
256 14613 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_OUT,
257 : "Passive session Accept complete");
258 : {
259 14577 : std::scoped_lock obs_lock(obs_mutex_);
260 14577 : if (observer_) {
261 6652 : observer_(this, ACCEPT);
262 : }
263 14577 : }
264 :
265 14577 : if (read_on_connect_) {
266 251 : AsyncReadStart();
267 : }
268 14577 : }
269 :
270 14665 : bool TcpSession::Connected(Endpoint remote) {
271 14665 : assert(refcount_);
272 :
273 : {
274 14665 : std::scoped_lock lock(mutex_);
275 14665 : if (closed_) {
276 0 : return false;
277 : }
278 14665 : SessionEstablished(remote, TcpSession::ACTIVE);
279 14665 : }
280 14665 : SetSocketOptions();
281 :
282 14665 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_IN,
283 : "Active session connection complete");
284 :
285 : {
286 14665 : std::scoped_lock obs_lock(obs_mutex_);
287 14665 : if (observer_) {
288 14665 : observer_(this, CONNECT_COMPLETE);
289 : }
290 14665 : }
291 :
292 14665 : if (read_on_connect_) {
293 14665 : AsyncReadStart();
294 : }
295 14665 : return true;
296 : }
297 :
298 1598 : void TcpSession::ConnectFailed() {
299 1598 : std::scoped_lock obs_lock(obs_mutex_);
300 1598 : if (observer_) {
301 1499 : observer_(this, CONNECT_FAILED);
302 : }
303 1598 : }
304 :
305 : // Requires: lock must not be held
306 32117 : void TcpSession::CloseInternal(const error_code &ec,
307 : bool call_observer, bool notify_server) {
308 32117 : std::unique_lock<std::mutex> lock(mutex_);
309 :
310 32123 : if (socket() != NULL && !closed_) {
311 30987 : error_code error;
312 30987 : socket()->shutdown(tcp::socket::shutdown_both, error);
313 30996 : if (error) {
314 7099 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
315 : "Shutdown failed due to error: " << error.message());
316 : }
317 30995 : socket()->close(error);
318 : }
319 32123 : closed_ = true;
320 32123 : tcp_close_in_progress_ = false;
321 :
322 32126 : if (!established_) {
323 2838 : return;
324 : }
325 29288 : established_ = false;
326 :
327 : // copy the ec to close reason
328 29288 : close_reason_ = ec;
329 :
330 : // Take a reference through intrusive pointer to protect session from
331 : // possibly getting deleted from another thread.
332 29288 : TcpSessionPtr session = TcpSessionPtr(this);
333 29288 : lock.unlock();
334 :
335 29287 : if (call_observer) {
336 17804 : std::scoped_lock obs_lock(obs_mutex_);
337 17804 : if (observer_) {
338 12024 : observer_(this, CLOSE);
339 : }
340 17804 : }
341 :
342 29287 : if (notify_server) {
343 29242 : server_->OnSessionClose(this);
344 : }
345 32124 : }
346 :
347 1066781 : void TcpSession::TriggerAsyncReadHandler() {
348 1066781 : if (io_strand_) {
349 1066781 : boost::asio::detail::recycling_allocator<void> allocator;
350 2133562 : io_strand_->post(bind(&TcpSession::AsyncReadHandler,
351 2133562 : TcpSessionPtr(this)), allocator);
352 : }
353 1066781 : }
354 :
355 65198 : void TcpSession::Close() {
356 65198 : 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 65204 : if (closed_) {
362 41765 : return;
363 : }
364 :
365 23439 : if (server_ && writer_->IsWritePending()) {
366 9185 : tcp_close_in_progress_ = true;
367 9189 : return;
368 : }
369 14249 : lock.unlock();
370 :
371 14249 : error_code ec;
372 14249 : CloseInternal(ec, false);
373 65212 : }
374 :
375 : // virtual method overriden in derrived classes.
376 2 : void TcpSession::WriteReady(const error_code &error) {
377 2 : }
378 :
379 1884546 : void TcpSession::AsyncWriteHandler(TcpSessionPtr session,
380 : const error_code &error,
381 : std::size_t wrote) {
382 1884546 : std::unique_lock<std::mutex> lock(session->mutex_);
383 1884546 : if (session->IsSocketErrorHard(error)) {
384 61 : lock.unlock();
385 61 : TCP_SESSION_LOG_ERROR(session, TCP_DIR_OUT,
386 : "Write failed due to error: " << error.message());
387 61 : session->CloseInternal(error, true);
388 61 : return;
389 : }
390 :
391 : //
392 : // Ignore if connection is already closed.
393 : //
394 1884485 : if (session->IsClosedLocked()) return;
395 :
396 : // Update socket write bytes statistics.
397 1883937 : session->stats_.write_bytes += wrote;
398 1883937 : session->server_->stats_.write_bytes += wrote;
399 :
400 1883937 : bool send_ready = false;
401 1883937 : bool more_write = session->writer_->UpdateBufferQueue(wrote, &send_ready);
402 :
403 : // Subsequent write
404 1883937 : if (more_write) {
405 323633 : session->writer_->TriggerAsyncWrite();
406 1560304 : } else if (session->tcp_close_in_progress_) {
407 5292 : lock.unlock();
408 5292 : session->CloseInternal(error, true);
409 5292 : return;
410 : }
411 :
412 1878645 : lock.unlock();
413 1878645 : if (send_ready)
414 5 : session->WriteReady(error);
415 1878645 : return;
416 1884546 : }
417 :
418 1560945 : void TcpSession::AsyncWriteInternal(TcpSessionPtr session) {
419 :
420 1560945 : std::scoped_lock lock(session->mutex_);
421 :
422 : //
423 : // Ignore if connection is already closed.
424 : //
425 1560945 : if (session->IsClosedLocked()) return;
426 1560913 : session->writer_->TriggerAsyncWrite();
427 1560945 : }
428 :
429 1882586 : bool TcpSession::Send(const uint8_t *data, size_t size, size_t *sent) {
430 1882586 : bool ret = true;
431 1882586 : std::unique_lock<std::mutex> lock(mutex_);
432 :
433 : // Reset sent, if provided.
434 1882618 : 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 1882618 : if (!IsEstablishedLocked()) return false;
441 :
442 1881896 : if (socket()->non_blocking()) {
443 1881830 : error_code error;
444 1881830 : int len = writer_->AsyncSend(data, size, &error);
445 1881120 : lock.unlock();
446 1881911 : 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 1881911 : if ((size_t) len != size)
455 15 : ret = false;
456 1881911 : if (sent) *sent = (len > 0) ? len : 0;
457 : }
458 1881909 : return ret;
459 1882569 : }
460 :
461 106345 : Task* TcpSession::CreateReaderTask(mutable_buffer buffer,
462 : size_t bytes_transferred) {
463 106345 : Buffer rdbuf(buffer_cast<const uint8_t *>(buffer), bytes_transferred);
464 106345 : Reader *task = new Reader(TcpSessionPtr(this),
465 106345 : bind(&TcpSession::OnRead, this, _1), rdbuf);
466 106345 : return (task);
467 : }
468 :
469 450074 : size_t TcpSession::ReadSome(mutable_buffer buffer, error_code *error) {
470 450074 : 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 112251 : size_t TcpSession::GetReadBufferSize() const {
477 112251 : error_code error;
478 112251 : size_t size = socket_->available(error);
479 112251 : if (size < kDefaultBufferSize)
480 111939 : size = kDefaultBufferSize;
481 112251 : return size;
482 : }
483 :
484 1344227 : void TcpSession::AsyncReadHandler(TcpSessionPtr session) {
485 1344227 : std::unique_lock<std::mutex> lock(session->mutex_);
486 1344227 : if (session->closed_) {
487 9689 : return;
488 : }
489 :
490 : mutable_buffer buffer =
491 1334538 : session->AllocateBuffer(session->GetReadBufferSize());
492 :
493 1334538 : error_code error;
494 1334538 : size_t bytes_transferred = session->ReadSome(buffer, &error);
495 1334538 : if (session->IsSocketErrorHard(error)) {
496 12467 : session->ReleaseBufferLocked(buffer);
497 : // eof is returned when the peer closed the socket, no need to log error
498 12467 : if (error != boost::asio::error::eof) {
499 1451 : 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 1451 : 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 12467 : lock.unlock();
515 12467 : session->CloseInternal(error, true);
516 12467 : return;
517 : }
518 :
519 : // Update read statistics.
520 1322071 : session->stats_.read_calls++;
521 1322071 : session->stats_.read_bytes += bytes_transferred;
522 1322071 : session->server_->stats_.read_calls++;
523 1322071 : session->server_->stats_.read_bytes += bytes_transferred;
524 :
525 1322071 : Task *task = session->CreateReaderTask(buffer, bytes_transferred);
526 : // Starting a new task for the session
527 1322071 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
528 1322071 : scheduler->Enqueue(task);
529 1344227 : }
530 :
531 4616 : int TcpSession::GetSessionInstance() const {
532 4616 : return Task::kTaskInstanceAny;
533 : }
534 :
535 :
536 21132 : int32_t TcpSession::local_port() const {
537 21132 : if (socket() == NULL) {
538 0 : return -1;
539 : }
540 21132 : error_code error;
541 21132 : Endpoint local = socket()->local_endpoint(error);
542 21133 : if (IsSocketErrorHard(error)) {
543 0 : return -1;
544 : }
545 21132 : return local.port();
546 : }
547 :
548 21132 : int32_t TcpSession::remote_port() const {
549 21132 : if (socket() == NULL) {
550 0 : return -1;
551 : }
552 21132 : error_code error;
553 21132 : Endpoint remote = socket()->remote_endpoint(error);
554 21134 : if (IsSocketErrorHard(error)) {
555 1470 : return -1;
556 : }
557 19664 : return remote.port();
558 : }
559 :
560 1754 : int TcpSession::SetMd5SocketOption(uint32_t peer_ip,
561 : const string &md5_password) {
562 1754 : return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip,
563 1754 : 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 4586 : int TcpSession::SetDscpSocketOption(uint8_t value) {
571 4586 : 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 16855 : TcpMessageReader::TcpMessageReader(TcpSession *session,
579 16855 : ReceiveCallback callback)
580 16855 : : session_(session), callback_(callback), offset_(0), remain_(-1) {
581 16848 : }
582 :
583 16863 : TcpMessageReader::~TcpMessageReader() {
584 16863 : }
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 104374 : void TcpMessageReader::OnRead(Buffer buffer) {
673 104374 : const int kHeaderLenSize = GetHeaderLenSize();
674 104373 : size_t size = TcpSession::BufferSize(buffer);
675 104371 : TCP_SESSION_LOG_UT_DEBUG(session_, TCP_DIR_IN, "Read " << size << " bytes");
676 :
677 104375 : 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 104368 : int avail = size - offset_;
711 284034 : while (avail > 0) {
712 179706 : int msglength = MsgLength(buffer, offset_);
713 179703 : if (msglength < 0) {
714 2 : break;
715 : }
716 179701 : if (msglength > avail) {
717 2 : remain_ = msglength - avail;
718 2 : break;
719 : }
720 : // Receive the message
721 : bool success =
722 179699 : callback_(TcpSession::BufferData(buffer) + offset_, msglength);
723 179705 : offset_ += msglength;
724 179705 : avail -= msglength;
725 179705 : if (!success)
726 39 : return;
727 : }
728 :
729 104332 : if (avail > 0) {
730 4 : queue_.push_back(buffer);
731 : } else {
732 104328 : session_->ReleaseBuffer(buffer);
733 104329 : offset_ = 0;
734 104329 : 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 3261341 : bool TcpSession::IsSocketErrorHard(const error_code &ec) {
744 3261341 : if (!ec)
745 3098720 : return false;
746 162619 : if (ec == try_again)
747 148621 : return false;
748 13998 : if (ec == would_block)
749 0 : return false;
750 13998 : if (ec == in_progress)
751 0 : return false;
752 13998 : if (ec == interrupted)
753 0 : return false;
754 13998 : if (ec == network_down)
755 0 : return false;
756 13998 : if (ec == network_reset)
757 0 : return false;
758 13998 : if (ec == network_unreachable)
759 0 : return false;
760 13998 : if (ec == no_buffer_space)
761 0 : return false;
762 :
763 13998 : 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 45646 : error_code TcpSession::SetSocketOptions() {
870 45646 : error_code ec;
871 :
872 : //
873 : // Make socket write non-blocking
874 : //
875 45646 : socket()->non_blocking(true, ec);
876 45647 : 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 45646 : char *buffer_size_str = getenv("TCP_SESSION_SOCKET_BUFFER_SIZE");
883 45647 : if (!buffer_size_str) return ec;
884 :
885 28 : uint32_t sz = static_cast<uint32_t>(strtoul(buffer_size_str, NULL, 0));
886 30 : 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 30 : socket_base::send_buffer_size send_buffer_size_option(sz);
894 30 : socket()->set_option(send_buffer_size_option, ec);
895 30 : 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 30 : socket_base::receive_buffer_size receive_buffer_size_option(sz);
902 30 : socket()->set_option(receive_buffer_size_option, ec);
903 30 : 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 30 : 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 : }
|
