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 11908 : Reader(TcpSessionPtr session, ReadHandler read_fn, Buffer buffer)
62 47632 : : Task(session->reader_task_id(), session->GetSessionInstance()),
63 11908 : session_(session), read_fn_(read_fn), buffer_(buffer) {
64 11908 : }
65 11908 : virtual bool Run() {
66 11908 : if (session_->IsEstablished()) {
67 11868 : read_fn_(buffer_);
68 11868 : 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 11867 : session_->AsyncReadStart();
75 : }
76 : }
77 11908 : 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 3586 : TcpSession::TcpSession(
88 : TcpServer *server, Socket *socket, bool async_read_ready,
89 3586 : size_t buffer_send_size)
90 3584 : : server_(server),
91 3586 : socket_(socket),
92 3586 : read_on_connect_(async_read_ready),
93 3586 : established_(false),
94 3586 : closed_(false),
95 3584 : direction_(ACTIVE),
96 3575 : writer_(new TcpMessageWriter(this, buffer_send_size)),
97 10746 : name_("-") {
98 3573 : refcount_ = 0;
99 3587 : if (reader_task_id_ == -1) {
100 60 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
101 60 : reader_task_id_ = scheduler->GetTaskId("io::ReaderTask");
102 : }
103 3587 : if (server_) {
104 3579 : io_strand_.reset(new Strand(server->event_manager()->io_service()->get_executor()));
105 : }
106 3587 : defer_reader_ = false;
107 3587 : write_blocked_ = false;
108 3587 : tcp_close_in_progress_ = false;
109 3587 : }
110 :
111 3587 : TcpSession::~TcpSession() {
112 3587 : assert(!established_);
113 3587 : for (BufferQueue::iterator iter = buffer_queue_.begin();
114 8242 : iter != buffer_queue_.end(); ++iter) {
115 4655 : DeleteBuffer(*iter);
116 : }
117 3587 : buffer_queue_.clear();
118 3587 : }
119 :
120 173324 : mutable_buffer TcpSession::AllocateBuffer(size_t buffer_size) {
121 173324 : uint8_t *data = new uint8_t[buffer_size];
122 173324 : mutable_buffer buffer = mutable_buffer(data, buffer_size);
123 173324 : buffer_queue_.push_back(buffer);
124 173324 : return buffer;
125 : }
126 :
127 173318 : void TcpSession::DeleteBuffer(mutable_buffer buffer) {
128 173318 : uint8_t *data = buffer_cast<uint8_t *>(buffer);
129 173318 : delete[] data;
130 173324 : }
131 :
132 170940 : static int BufferCmp(const mutable_buffer &lhs, const const_buffer &rhs) {
133 170940 : const uint8_t *lp = buffer_cast<uint8_t *>(lhs);
134 170940 : const uint8_t *rp = buffer_cast<const uint8_t *>(rhs);
135 170939 : if (lp < rp) {
136 1941 : return -1;
137 : }
138 168998 : if (lp > rp) {
139 332 : return 1;
140 : }
141 168666 : return 0;
142 : }
143 :
144 167496 : void TcpSession::ReleaseBuffer(Buffer buffer) {
145 167496 : std::scoped_lock lock(mutex_);
146 167497 : ReleaseBufferLocked(buffer);
147 167495 : }
148 :
149 168669 : void TcpSession::ReleaseBufferLocked(Buffer buffer) {
150 168669 : for (BufferQueue::iterator iter = buffer_queue_.begin();
151 170942 : iter != buffer_queue_.end(); ++iter) {
152 170942 : if (BufferCmp(*iter, buffer) == 0) {
153 168665 : DeleteBuffer(*iter);
154 168669 : buffer_queue_.erase(iter);
155 168667 : return;
156 : }
157 : }
158 0 : assert(false);
159 : }
160 :
161 175019 : void TcpSession::AsyncReadStartInternal(TcpSessionPtr session) {
162 : // Update socket read block time.
163 175019 : 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 175019 : std::scoped_lock lock(mutex_);
172 175019 : AsyncReadSome();
173 175019 : }
174 :
175 175022 : void TcpSession::AsyncReadStart() {
176 175022 : if (io_strand_) {
177 175022 : boost::asio::detail::recycling_allocator<void> allocator;
178 350043 : io_strand_->post(bind(&TcpSession::AsyncReadStartInternal, this,
179 350041 : TcpSessionPtr(this)), allocator);
180 : }
181 175022 : }
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 26166 : void TcpSession::AsyncReadSome() {
194 26166 : if (IsEstablishedLocked()) {
195 51520 : socket()->async_read_some(null_buffers(),
196 51520 : bind(&TcpSession::AsyncReadHandler, TcpSessionPtr(this)));
197 : }
198 26166 : }
199 :
200 133921 : void TcpSession::AsyncWrite(const uint8_t *data, std::size_t size) {
201 133921 : async_write(*socket(), buffer(data, size),
202 267842 : bind(&TcpSession::AsyncWriteHandler, TcpSessionPtr(this),
203 : error, bytes_transferred));
204 133921 : }
205 :
206 167283 : TcpSession::Endpoint TcpSession::local_endpoint() const {
207 167283 : std::scoped_lock lock(mutex_);
208 167283 : if (!established_)
209 0 : return Endpoint();
210 :
211 167283 : error_code error;
212 167283 : Endpoint local = socket()->local_endpoint(error);
213 167283 : if (error) {
214 0 : return Endpoint();
215 : }
216 167283 : return local;
217 167283 : }
218 :
219 7939 : void TcpSession::set_observer(EventObserver observer) {
220 7939 : std::scoped_lock lock(obs_mutex_);
221 7939 : observer_ = observer;
222 7933 : }
223 :
224 3332 : void TcpSession::SetName() {
225 3332 : ostringstream out;
226 3332 : error_code error;
227 3332 : Endpoint local;
228 :
229 3332 : local = socket()->local_endpoint(error);
230 3332 : out << local.address().to_string() << ":" << local.port() << "::";
231 3332 : out << remote_.address().to_string() << ":" << remote_.port();
232 :
233 3332 : name_ = out.str();
234 :
235 3332 : out.str("");
236 3332 : std::string hostname = "";
237 3332 : if (local.address().is_v4()) {
238 3329 : hostname = ResolveCanonicalName(local.address().to_string());
239 : } else {
240 3 : hostname = ResolveCanonicalNameIPv6(local.address().to_string());
241 : }
242 3332 : out << hostname << ":" << remote_.address().to_string();
243 3332 : uve_key_str_ = out.str();
244 3332 : }
245 :
246 3332 : void TcpSession::SessionEstablished(Endpoint remote,
247 : Direction direction) {
248 3332 : established_ = true;
249 3332 : remote_ = remote;
250 3332 : remote_addr_str_ = remote.address().to_string();
251 3332 : direction_ = direction;
252 3332 : SetName();
253 3332 : }
254 :
255 1605 : void TcpSession::Accepted() {
256 1605 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_OUT,
257 : "Passive session Accept complete");
258 : {
259 1605 : std::scoped_lock obs_lock(obs_mutex_);
260 1605 : if (observer_) {
261 301 : observer_(this, ACCEPT);
262 : }
263 1605 : }
264 :
265 1605 : if (read_on_connect_) {
266 214 : AsyncReadStart();
267 : }
268 1605 : }
269 :
270 1690 : bool TcpSession::Connected(Endpoint remote) {
271 1690 : assert(refcount_);
272 :
273 : {
274 1690 : std::scoped_lock lock(mutex_);
275 1690 : if (closed_) {
276 0 : return false;
277 : }
278 1690 : SessionEstablished(remote, TcpSession::ACTIVE);
279 1690 : }
280 1690 : SetSocketOptions();
281 :
282 1690 : TCP_SESSION_LOG_DEBUG(this, TCP_DIR_IN,
283 : "Active session connection complete");
284 :
285 : {
286 1690 : std::scoped_lock obs_lock(obs_mutex_);
287 1690 : if (observer_) {
288 1690 : observer_(this, CONNECT_COMPLETE);
289 : }
290 1690 : }
291 :
292 1690 : if (read_on_connect_) {
293 1690 : AsyncReadStart();
294 : }
295 1690 : return true;
296 : }
297 :
298 141 : void TcpSession::ConnectFailed() {
299 141 : std::scoped_lock obs_lock(obs_mutex_);
300 141 : if (observer_) {
301 139 : observer_(this, CONNECT_FAILED);
302 : }
303 141 : }
304 :
305 : // Requires: lock must not be held
306 3591 : void TcpSession::CloseInternal(const error_code &ec,
307 : bool call_observer, bool notify_server) {
308 3591 : std::unique_lock<std::mutex> lock(mutex_);
309 :
310 3591 : if (socket() != NULL && !closed_) {
311 3584 : error_code error;
312 3584 : socket()->shutdown(tcp::socket::shutdown_both, error);
313 3583 : if (error) {
314 804 : TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
315 : "Shutdown failed due to error: " << error.message());
316 : }
317 3583 : socket()->close(error);
318 : }
319 3591 : closed_ = true;
320 3591 : tcp_close_in_progress_ = false;
321 :
322 3591 : if (!established_) {
323 259 : return;
324 : }
325 3332 : established_ = false;
326 :
327 : // copy the ec to close reason
328 3332 : close_reason_ = ec;
329 :
330 : // Take a reference through intrusive pointer to protect session from
331 : // possibly getting deleted from another thread.
332 3332 : TcpSessionPtr session = TcpSessionPtr(this);
333 3332 : lock.unlock();
334 :
335 3332 : if (call_observer) {
336 2197 : std::scoped_lock obs_lock(obs_mutex_);
337 2197 : if (observer_) {
338 955 : observer_(this, CLOSE);
339 : }
340 2197 : }
341 :
342 3332 : if (notify_server) {
343 3295 : server_->OnSessionClose(this);
344 : }
345 3590 : }
346 :
347 148836 : void TcpSession::TriggerAsyncReadHandler() {
348 148836 : if (io_strand_) {
349 148836 : boost::asio::detail::recycling_allocator<void> allocator;
350 297672 : io_strand_->post(bind(&TcpSession::AsyncReadHandler,
351 297672 : TcpSessionPtr(this)), allocator);
352 : }
353 148836 : }
354 :
355 7577 : void TcpSession::Close() {
356 7577 : 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 7576 : if (closed_) {
362 4131 : return;
363 : }
364 :
365 3445 : if (server_ && writer_->IsWritePending()) {
366 2091 : tcp_close_in_progress_ = true;
367 2095 : return;
368 : }
369 1353 : lock.unlock();
370 :
371 1353 : error_code ec;
372 1353 : CloseInternal(ec, false);
373 7579 : }
374 :
375 : // virtual method overriden in derrived classes.
376 2 : void TcpSession::WriteReady(const error_code &error) {
377 2 : }
378 :
379 260915 : void TcpSession::AsyncWriteHandler(TcpSessionPtr session,
380 : const error_code &error,
381 : std::size_t wrote) {
382 260915 : std::unique_lock<std::mutex> lock(session->mutex_);
383 260915 : if (session->IsSocketErrorHard(error)) {
384 1 : lock.unlock();
385 1 : TCP_SESSION_LOG_ERROR(session, TCP_DIR_OUT,
386 : "Write failed due to error: " << error.message());
387 1 : session->CloseInternal(error, true);
388 1 : return;
389 : }
390 :
391 : //
392 : // Ignore if connection is already closed.
393 : //
394 260914 : if (session->IsClosedLocked()) return;
395 :
396 : // Update socket write bytes statistics.
397 260912 : session->stats_.write_bytes += wrote;
398 260912 : session->server_->stats_.write_bytes += wrote;
399 :
400 260912 : bool send_ready = false;
401 260912 : bool more_write = session->writer_->UpdateBufferQueue(wrote, &send_ready);
402 :
403 : // Subsequent write
404 260912 : if (more_write) {
405 41173 : session->writer_->TriggerAsyncWrite();
406 219739 : } else if (session->tcp_close_in_progress_) {
407 1026 : lock.unlock();
408 1026 : session->CloseInternal(error, true);
409 1026 : return;
410 : }
411 :
412 259886 : lock.unlock();
413 259886 : if (send_ready)
414 4 : session->WriteReady(error);
415 259886 : return;
416 260915 : }
417 :
418 219744 : void TcpSession::AsyncWriteInternal(TcpSessionPtr session) {
419 :
420 219744 : std::scoped_lock lock(session->mutex_);
421 :
422 : //
423 : // Ignore if connection is already closed.
424 : //
425 219744 : if (session->IsClosedLocked()) return;
426 219742 : session->writer_->TriggerAsyncWrite();
427 219744 : }
428 :
429 258769 : bool TcpSession::Send(const uint8_t *data, size_t size, size_t *sent) {
430 258769 : bool ret = true;
431 258769 : std::unique_lock<std::mutex> lock(mutex_);
432 :
433 : // Reset sent, if provided.
434 258824 : 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 258824 : if (!IsEstablishedLocked()) return false;
441 :
442 258669 : if (socket()->non_blocking()) {
443 258647 : error_code error;
444 258647 : int len = writer_->AsyncSend(data, size, &error);
445 258526 : lock.unlock();
446 258673 : 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 258673 : if ((size_t) len != size)
455 14 : ret = false;
456 258673 : if (sent) *sent = (len > 0) ? len : 0;
457 : }
458 258677 : return ret;
459 258822 : }
460 :
461 11908 : Task* TcpSession::CreateReaderTask(mutable_buffer buffer,
462 : size_t bytes_transferred) {
463 11908 : Buffer rdbuf(buffer_cast<const uint8_t *>(buffer), bytes_transferred);
464 11908 : Reader *task = new Reader(TcpSessionPtr(this),
465 11908 : bind(&TcpSession::OnRead, this, _1), rdbuf);
466 11908 : return (task);
467 : }
468 :
469 39130 : size_t TcpSession::ReadSome(mutable_buffer buffer, error_code *error) {
470 39130 : 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 12769 : size_t TcpSession::GetReadBufferSize() const {
477 12769 : error_code error;
478 12769 : size_t size = socket_->available(error);
479 12769 : if (size < kDefaultBufferSize)
480 12466 : size = kDefaultBufferSize;
481 12769 : return size;
482 : }
483 :
484 174546 : void TcpSession::AsyncReadHandler(TcpSessionPtr session) {
485 174546 : std::unique_lock<std::mutex> lock(session->mutex_);
486 174546 : if (session->closed_) {
487 1222 : return;
488 : }
489 :
490 : mutable_buffer buffer =
491 173324 : session->AllocateBuffer(session->GetReadBufferSize());
492 :
493 173324 : error_code error;
494 173324 : size_t bytes_transferred = session->ReadSome(buffer, &error);
495 173324 : if (session->IsSocketErrorHard(error)) {
496 1172 : session->ReleaseBufferLocked(buffer);
497 : // eof is returned when the peer closed the socket, no need to log error
498 1172 : if (error != boost::asio::error::eof) {
499 131 : 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 131 : 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 1172 : lock.unlock();
515 1172 : session->CloseInternal(error, true);
516 1172 : return;
517 : }
518 :
519 : // Update read statistics.
520 172152 : session->stats_.read_calls++;
521 172152 : session->stats_.read_bytes += bytes_transferred;
522 172152 : session->server_->stats_.read_calls++;
523 172152 : session->server_->stats_.read_bytes += bytes_transferred;
524 :
525 172152 : Task *task = session->CreateReaderTask(buffer, bytes_transferred);
526 : // Starting a new task for the session
527 172152 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
528 172152 : scheduler->Enqueue(task);
529 174546 : }
530 :
531 4631 : int TcpSession::GetSessionInstance() const {
532 4631 : return Task::kTaskInstanceAny;
533 : }
534 :
535 :
536 1013 : int32_t TcpSession::local_port() const {
537 1013 : if (socket() == NULL) {
538 0 : return -1;
539 : }
540 1013 : error_code error;
541 1013 : Endpoint local = socket()->local_endpoint(error);
542 1012 : if (IsSocketErrorHard(error)) {
543 0 : return -1;
544 : }
545 1012 : return local.port();
546 : }
547 :
548 1012 : int32_t TcpSession::remote_port() const {
549 1012 : if (socket() == NULL) {
550 0 : return -1;
551 : }
552 1012 : error_code error;
553 1012 : Endpoint remote = socket()->remote_endpoint(error);
554 1013 : if (IsSocketErrorHard(error)) {
555 179 : return -1;
556 : }
557 834 : return remote.port();
558 : }
559 :
560 0 : int TcpSession::SetMd5SocketOption(uint32_t peer_ip,
561 : const string &md5_password) {
562 0 : return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip,
563 0 : 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 737 : int TcpSession::SetDscpSocketOption(uint8_t value) {
571 737 : return server()->SetDscpSocketOption(socket()->native_handle(), value);
572 : }
573 :
574 0 : uint8_t TcpSession::GetDscpValue() const {
575 0 : return server_->GetDscpValue(socket()->native_handle());
576 : }
577 :
578 2306 : TcpMessageReader::TcpMessageReader(TcpSession *session,
579 2306 : ReceiveCallback callback)
580 2306 : : session_(session), callback_(callback), offset_(0), remain_(-1) {
581 2297 : }
582 :
583 2307 : TcpMessageReader::~TcpMessageReader() {
584 2307 : }
585 :
586 : // Returns a buffer allocation size that is larger than the message.
587 2 : int TcpMessageReader::AllocBufferSize(int length) {
588 2 : const int kMaxMessageSize = GetMaxMessageSize();
589 2 : if (length == -1) {
590 0 : return kMaxMessageSize;
591 : }
592 2 : int bufsize = 1 << 8;
593 3 : for (; bufsize < kMaxMessageSize && bufsize < length; bufsize <<= 1) {
594 : }
595 2 : return bufsize;
596 : }
597 :
598 2 : uint8_t *TcpMessageReader::BufferConcat(uint8_t *data, Buffer buffer,
599 : int msglength) {
600 2 : uint8_t *dst = data;
601 :
602 7 : while (!queue_.empty()) {
603 5 : Buffer head = queue_.front();
604 5 : const uint8_t *cp = TcpSession::BufferData(head) + offset_;
605 5 : int bytes = TcpSession::BufferSize(head) - offset_;
606 5 : assert((dst - data) + bytes < msglength);
607 5 : memcpy(dst, cp, bytes);
608 5 : dst += bytes;
609 5 : queue_.pop_front();
610 5 : session_->ReleaseBuffer(head);
611 5 : offset_ = 0;
612 5 : remain_ = -1;
613 : }
614 :
615 2 : int count = msglength - (dst - data);
616 2 : assert((dst - data) + count <= msglength);
617 2 : memcpy(dst, TcpSession::BufferData(buffer), count);
618 2 : offset_ = count;
619 :
620 2 : return data;
621 : }
622 :
623 3 : int TcpMessageReader::QueueByteLength() const {
624 3 : int total = 0;
625 3 : for (BufferQueue::const_iterator iter = queue_.begin();
626 9 : iter != queue_.end(); ++iter) {
627 6 : if (total == 0) {
628 3 : total = TcpSession::BufferSize(*iter) - offset_;
629 : } else {
630 3 : total += TcpSession::BufferSize(*iter);
631 : }
632 : }
633 3 : return total;
634 : }
635 :
636 2 : TcpMessageReader::Buffer TcpMessageReader::PullUp(
637 : uint8_t *data, Buffer buffer, size_t size) const {
638 2 : size_t offset = 0;
639 :
640 2 : for (BufferQueue::const_iterator iter = queue_.begin();
641 7 : iter != queue_.end(); ++iter) {
642 : const uint8_t *cp;
643 : int avail;
644 5 : if (offset == 0) {
645 2 : cp = TcpSession::BufferData(*iter) + offset_;
646 2 : avail = TcpSession::BufferSize(*iter) - offset_;
647 : } else {
648 3 : cp = TcpSession::BufferData(*iter);
649 3 : avail = TcpSession::BufferSize(*iter);
650 : }
651 5 : int remain = size - offset;
652 5 : avail = min(avail, remain);
653 5 : assert(offset + avail <= size);
654 5 : memcpy(data + offset, cp, avail);
655 5 : offset += avail;
656 : }
657 :
658 2 : int avail = TcpSession::BufferSize(buffer);
659 2 : int remain = size - offset;
660 2 : avail = min(avail, remain);
661 2 : assert(offset + avail <= size);
662 2 : memcpy(data + offset, TcpSession::BufferData(buffer), avail);
663 2 : offset += avail;
664 :
665 2 : if (offset < size) {
666 0 : return Buffer();
667 : }
668 2 : return Buffer(data, size);
669 : }
670 :
671 : // Read the socket stream and send messages to the peer object.
672 11283 : void TcpMessageReader::OnRead(Buffer buffer) {
673 11283 : const int kHeaderLenSize = GetHeaderLenSize();
674 11283 : size_t size = TcpSession::BufferSize(buffer);
675 11282 : TCP_SESSION_LOG_UT_DEBUG(session_, TCP_DIR_IN, "Read " << size << " bytes");
676 :
677 11282 : if (!queue_.empty()) {
678 5 : int msglength = MsgLength(queue_.front(), offset_);
679 5 : if (msglength < 0) {
680 3 : int queuelen = QueueByteLength();
681 3 : if (queuelen + static_cast<int>(size) < kHeaderLenSize) {
682 1 : queue_.push_back(buffer);
683 1 : return;
684 : }
685 2 : scoped_array<uint8_t> data(new uint8_t[kHeaderLenSize]);
686 2 : Buffer header = PullUp(data.get(), buffer, kHeaderLenSize);
687 2 : assert(TcpSession::BufferSize(header) == (size_t) kHeaderLenSize);
688 :
689 2 : msglength = MsgLength(header, 0);
690 2 : remain_ = msglength - queuelen;
691 2 : }
692 :
693 4 : assert(remain_ > 0);
694 4 : if (size < (size_t) remain_) {
695 2 : queue_.push_back(buffer);
696 2 : remain_ -= size;
697 2 : return;
698 : }
699 :
700 : // concat the buffers into a contiguous message.
701 2 : scoped_array<uint8_t> data(new uint8_t[AllocBufferSize(msglength)]);
702 2 : BufferConcat(data.get(), buffer, msglength);
703 2 : assert(remain_ == -1);
704 : // Receive the message
705 2 : bool success = callback_(data.get(), msglength);
706 2 : if (!success)
707 0 : return;
708 2 : }
709 :
710 11280 : int avail = size - offset_;
711 32305 : while (avail > 0) {
712 21030 : int msglength = MsgLength(buffer, offset_);
713 21026 : if (msglength < 0) {
714 1 : break;
715 : }
716 21025 : if (msglength > avail) {
717 1 : remain_ = msglength - avail;
718 1 : break;
719 : }
720 : // Receive the message
721 : bool success =
722 21024 : callback_(TcpSession::BufferData(buffer) + offset_, msglength);
723 21027 : offset_ += msglength;
724 21027 : avail -= msglength;
725 21027 : if (!success)
726 2 : return;
727 : }
728 :
729 11277 : if (avail > 0) {
730 2 : queue_.push_back(buffer);
731 : } else {
732 11275 : session_->ReleaseBuffer(buffer);
733 11276 : offset_ = 0;
734 11276 : 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 436264 : bool TcpSession::IsSocketErrorHard(const error_code &ec) {
744 436264 : if (!ec)
745 423509 : return false;
746 12755 : if (ec == try_again)
747 11403 : return false;
748 1352 : if (ec == would_block)
749 0 : return false;
750 1352 : if (ec == in_progress)
751 0 : return false;
752 1352 : if (ec == interrupted)
753 0 : return false;
754 1352 : if (ec == network_down)
755 0 : return false;
756 1352 : if (ec == network_reset)
757 0 : return false;
758 1352 : if (ec == network_unreachable)
759 0 : return false;
760 1352 : if (ec == no_buffer_space)
761 0 : return false;
762 :
763 1352 : 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 456 : error_code TcpSession::SetSocketKeepaliveOptions(int keepalive_time,
804 : int keepalive_intvl, int keepalive_probes, int tcp_user_timeout_val) {
805 456 : error_code ec;
806 456 : socket_base::keep_alive keep_alive_option(true);
807 456 : socket()->set_option(keep_alive_option, ec);
808 456 : 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 434 : keepalive_idle_time keepalive_idle_time_option(keepalive_time);
816 434 : socket()->set_option(keepalive_idle_time_option, ec);
817 434 : 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 434 : keepalive_interval keepalive_interval_option(keepalive_intvl);
837 434 : socket()->set_option(keepalive_interval_option, ec);
838 434 : 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 434 : keepalive_count keepalive_count_option(keepalive_probes);
847 434 : socket()->set_option(keepalive_count_option, ec);
848 434 : 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 434 : tcp_user_timeout tcp_user_timeout_option(tcp_user_timeout_val);
857 434 : socket()->set_option(tcp_user_timeout_option, ec);
858 434 : 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 434 : return ec;
867 : }
868 :
869 5261 : error_code TcpSession::SetSocketOptions() {
870 5261 : error_code ec;
871 :
872 : //
873 : // Make socket write non-blocking
874 : //
875 5261 : socket()->non_blocking(true, ec);
876 5261 : 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 5261 : char *buffer_size_str = getenv("TCP_SESSION_SOCKET_BUFFER_SIZE");
883 5261 : 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 : }
|
