1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2014 UnitedStack <haomai@unitedstack.com>
8 * Author: Haomai Wang <haomaiwang@gmail.com>
10 * This is free software; you can redistribute it and/or
11 * modify it under the terms of the GNU Lesser General Public
12 * License version 2.1, as published by the Free Software
13 * Foundation. See file COPYING.
19 #include "include/Context.h"
20 #include "common/errno.h"
21 #include "AsyncMessenger.h"
22 #include "AsyncConnection.h"
24 #include "messages/MOSDOp.h"
25 #include "messages/MOSDOpReply.h"
26 #include "common/EventTrace.h"
28 // Constant to limit starting sequence number to 2^31. Nothing special about it, just a big number. PLR
29 #define SEQ_MASK 0x7fffffff
31 #define dout_subsys ceph_subsys_ms
33 #define dout_prefix _conn_prefix(_dout)
34 ostream
& AsyncConnection::_conn_prefix(std::ostream
*_dout
) {
35 return *_dout
<< "-- " << async_msgr
->get_myinst().addr
<< " >> " << peer_addr
<< " conn(" << this
37 << " s=" << get_state_name(state
)
38 << " pgs=" << peer_global_seq
39 << " cs=" << connect_seq
40 << " l=" << policy
.lossy
45 // 1. Don't dispatch any event when closed! It may cause AsyncConnection alive even if AsyncMessenger dead
47 const int AsyncConnection::TCP_PREFETCH_MIN_SIZE
= 512;
48 const int ASYNC_COALESCE_THRESHOLD
= 256;
50 class C_time_wakeup
: public EventCallback
{
51 AsyncConnectionRef conn
;
54 explicit C_time_wakeup(AsyncConnectionRef c
): conn(c
) {}
55 void do_request(int fd_or_id
) override
{
56 conn
->wakeup_from(fd_or_id
);
60 class C_handle_read
: public EventCallback
{
61 AsyncConnectionRef conn
;
64 explicit C_handle_read(AsyncConnectionRef c
): conn(c
) {}
65 void do_request(int fd_or_id
) override
{
70 class C_handle_write
: public EventCallback
{
71 AsyncConnectionRef conn
;
74 explicit C_handle_write(AsyncConnectionRef c
): conn(c
) {}
75 void do_request(int fd
) override
{
80 class C_clean_handler
: public EventCallback
{
81 AsyncConnectionRef conn
;
83 explicit C_clean_handler(AsyncConnectionRef c
): conn(c
) {}
84 void do_request(int id
) override
{
90 class C_tick_wakeup
: public EventCallback
{
91 AsyncConnectionRef conn
;
94 explicit C_tick_wakeup(AsyncConnectionRef c
): conn(c
) {}
95 void do_request(int fd_or_id
) override
{
100 static void alloc_aligned_buffer(bufferlist
& data
, unsigned len
, unsigned off
)
102 // create a buffer to read into that matches the data alignment
104 if (off
& ~CEPH_PAGE_MASK
) {
107 head
= MIN(CEPH_PAGE_SIZE
- (off
& ~CEPH_PAGE_MASK
), left
);
108 data
.push_back(buffer::create(head
));
111 unsigned middle
= left
& CEPH_PAGE_MASK
;
113 data
.push_back(buffer::create_page_aligned(middle
));
117 data
.push_back(buffer::create(left
));
121 AsyncConnection::AsyncConnection(CephContext
*cct
, AsyncMessenger
*m
, DispatchQueue
*q
,
123 : Connection(cct
, m
), delay_state(NULL
), async_msgr(m
), conn_id(q
->get_id()),
124 logger(w
->get_perf_counter()), global_seq(0), connect_seq(0), peer_global_seq(0),
125 state(STATE_NONE
), state_after_send(STATE_NONE
), port(-1),
126 dispatch_queue(q
), can_write(WriteStatus::NOWRITE
),
127 keepalive(false), recv_buf(NULL
),
128 recv_max_prefetch(MAX(msgr
->cct
->_conf
->ms_tcp_prefetch_max_size
, TCP_PREFETCH_MIN_SIZE
)),
129 recv_start(0), recv_end(0),
130 last_active(ceph::coarse_mono_clock::now()),
131 inactive_timeout_us(cct
->_conf
->ms_tcp_read_timeout
*1000*1000),
132 got_bad_auth(false), authorizer(NULL
), replacing(false),
133 is_reset_from_peer(false), once_ready(false), state_buffer(NULL
), state_offset(0),
134 worker(w
), center(&w
->center
)
136 read_handler
= new C_handle_read(this);
137 write_handler
= new C_handle_write(this);
138 wakeup_handler
= new C_time_wakeup(this);
139 tick_handler
= new C_tick_wakeup(this);
140 memset(msgvec
, 0, sizeof(msgvec
));
141 // double recv_max_prefetch see "read_until"
142 recv_buf
= new char[2*recv_max_prefetch
];
143 state_buffer
= new char[4096];
144 logger
->inc(l_msgr_created_connections
);
147 AsyncConnection::~AsyncConnection()
149 assert(out_q
.empty());
150 assert(sent
.empty());
155 delete[] state_buffer
;
156 assert(!delay_state
);
159 void AsyncConnection::maybe_start_delay_thread()
162 auto pos
= async_msgr
->cct
->_conf
->get_val
<std::string
>("ms_inject_delay_type").find(ceph_entity_type_name(peer_type
));
163 if (pos
!= string::npos
) {
164 ldout(msgr
->cct
, 1) << __func__
<< " setting up a delay queue" << dendl
;
165 delay_state
= new DelayedDelivery(async_msgr
, center
, dispatch_queue
, conn_id
);
170 /* return -1 means `fd` occurs error or closed, it should be closed
171 * return 0 means EAGAIN or EINTR */
172 ssize_t
AsyncConnection::read_bulk(char *buf
, unsigned len
)
176 nread
= cs
.read(buf
, len
);
178 if (nread
== -EAGAIN
) {
180 } else if (nread
== -EINTR
) {
183 ldout(async_msgr
->cct
, 1) << __func__
<< " reading from fd=" << cs
.fd()
184 << " : "<< strerror(nread
) << dendl
;
187 } else if (nread
== 0) {
188 ldout(async_msgr
->cct
, 1) << __func__
<< " peer close file descriptor "
195 // return the remaining bytes, it may larger than the length of ptr
196 // else return < 0 means error
197 ssize_t
AsyncConnection::_try_send(bool more
)
199 if (async_msgr
->cct
->_conf
->ms_inject_socket_failures
&& cs
) {
200 if (rand() % async_msgr
->cct
->_conf
->ms_inject_socket_failures
== 0) {
201 ldout(async_msgr
->cct
, 0) << __func__
<< " injecting socket failure" << dendl
;
206 assert(center
->in_thread());
207 ssize_t r
= cs
.send(outcoming_bl
, more
);
209 ldout(async_msgr
->cct
, 1) << __func__
<< " send error: " << cpp_strerror(r
) << dendl
;
213 ldout(async_msgr
->cct
, 10) << __func__
<< " sent bytes " << r
214 << " remaining bytes " << outcoming_bl
.length() << dendl
;
216 if (!open_write
&& is_queued()) {
217 center
->create_file_event(cs
.fd(), EVENT_WRITABLE
, write_handler
);
221 if (open_write
&& !is_queued()) {
222 center
->delete_file_event(cs
.fd(), EVENT_WRITABLE
);
224 if (state_after_send
!= STATE_NONE
)
225 center
->dispatch_event_external(read_handler
);
228 return outcoming_bl
.length();
231 // Because this func will be called multi times to populate
232 // the needed buffer, so the passed in bufferptr must be the same.
233 // Normally, only "read_message" will pass existing bufferptr in
235 // And it will uses readahead method to reduce small read overhead,
236 // "recv_buf" is used to store read buffer
238 // return the remaining bytes, 0 means this buffer is finished
239 // else return < 0 means error
240 ssize_t
AsyncConnection::read_until(unsigned len
, char *p
)
242 ldout(async_msgr
->cct
, 25) << __func__
<< " len is " << len
<< " state_offset is "
243 << state_offset
<< dendl
;
245 if (async_msgr
->cct
->_conf
->ms_inject_socket_failures
&& cs
) {
246 if (rand() % async_msgr
->cct
->_conf
->ms_inject_socket_failures
== 0) {
247 ldout(async_msgr
->cct
, 0) << __func__
<< " injecting socket failure" << dendl
;
253 uint64_t left
= len
- state_offset
;
254 if (recv_end
> recv_start
) {
255 uint64_t to_read
= MIN(recv_end
- recv_start
, left
);
256 memcpy(p
, recv_buf
+recv_start
, to_read
);
257 recv_start
+= to_read
;
259 ldout(async_msgr
->cct
, 25) << __func__
<< " got " << to_read
<< " in buffer "
260 << " left is " << left
<< " buffer still has "
261 << recv_end
- recv_start
<< dendl
;
265 state_offset
+= to_read
;
268 recv_end
= recv_start
= 0;
269 /* nothing left in the prefetch buffer */
270 if (len
> recv_max_prefetch
) {
271 /* this was a large read, we don't prefetch for these */
273 r
= read_bulk(p
+state_offset
, left
);
274 ldout(async_msgr
->cct
, 25) << __func__
<< " read_bulk left is " << left
<< " got " << r
<< dendl
;
276 ldout(async_msgr
->cct
, 1) << __func__
<< " read failed" << dendl
;
278 } else if (r
== static_cast<int>(left
)) {
287 r
= read_bulk(recv_buf
+recv_end
, recv_max_prefetch
);
288 ldout(async_msgr
->cct
, 25) << __func__
<< " read_bulk recv_end is " << recv_end
289 << " left is " << left
<< " got " << r
<< dendl
;
291 ldout(async_msgr
->cct
, 1) << __func__
<< " read failed" << dendl
;
295 if (r
>= static_cast<int>(left
)) {
296 recv_start
= len
- state_offset
;
297 memcpy(p
+state_offset
, recv_buf
, recv_start
);
303 memcpy(p
+state_offset
, recv_buf
, recv_end
-recv_start
);
304 state_offset
+= (recv_end
- recv_start
);
305 recv_end
= recv_start
= 0;
307 ldout(async_msgr
->cct
, 25) << __func__
<< " need len " << len
<< " remaining "
308 << len
- state_offset
<< " bytes" << dendl
;
309 return len
- state_offset
;
312 void AsyncConnection::inject_delay() {
313 if (async_msgr
->cct
->_conf
->ms_inject_internal_delays
) {
314 ldout(async_msgr
->cct
, 10) << __func__
<< " sleep for " <<
315 async_msgr
->cct
->_conf
->ms_inject_internal_delays
<< dendl
;
317 t
.set_from_double(async_msgr
->cct
->_conf
->ms_inject_internal_delays
);
322 void AsyncConnection::process()
325 int prev_state
= state
;
326 #if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
327 utime_t ltt_recv_stamp
= ceph_clock_now();
329 bool need_dispatch_writer
= false;
330 std::lock_guard
<std::mutex
> l(lock
);
331 last_active
= ceph::coarse_mono_clock::now();
332 auto recv_start_time
= ceph::mono_clock::now();
334 ldout(async_msgr
->cct
, 20) << __func__
<< " prev state is " << get_state_name(prev_state
) << dendl
;
340 r
= read_until(sizeof(tag
), &tag
);
342 ldout(async_msgr
->cct
, 1) << __func__
<< " read tag failed" << dendl
;
348 if (tag
== CEPH_MSGR_TAG_KEEPALIVE
) {
349 ldout(async_msgr
->cct
, 20) << __func__
<< " got KEEPALIVE" << dendl
;
350 set_last_keepalive(ceph_clock_now());
351 } else if (tag
== CEPH_MSGR_TAG_KEEPALIVE2
) {
352 state
= STATE_OPEN_KEEPALIVE2
;
353 } else if (tag
== CEPH_MSGR_TAG_KEEPALIVE2_ACK
) {
354 state
= STATE_OPEN_KEEPALIVE2_ACK
;
355 } else if (tag
== CEPH_MSGR_TAG_ACK
) {
356 state
= STATE_OPEN_TAG_ACK
;
357 } else if (tag
== CEPH_MSGR_TAG_MSG
) {
358 state
= STATE_OPEN_MESSAGE_HEADER
;
359 } else if (tag
== CEPH_MSGR_TAG_CLOSE
) {
360 state
= STATE_OPEN_TAG_CLOSE
;
362 ldout(async_msgr
->cct
, 0) << __func__
<< " bad tag " << (int)tag
<< dendl
;
369 case STATE_OPEN_KEEPALIVE2
:
372 r
= read_until(sizeof(*t
), state_buffer
);
374 ldout(async_msgr
->cct
, 1) << __func__
<< " read keeplive timespec failed" << dendl
;
380 ldout(async_msgr
->cct
, 30) << __func__
<< " got KEEPALIVE2 tag ..." << dendl
;
381 t
= (ceph_timespec
*)state_buffer
;
382 utime_t kp_t
= utime_t(*t
);
384 _append_keepalive_or_ack(true, &kp_t
);
386 ldout(async_msgr
->cct
, 20) << __func__
<< " got KEEPALIVE2 " << kp_t
<< dendl
;
387 set_last_keepalive(ceph_clock_now());
388 need_dispatch_writer
= true;
393 case STATE_OPEN_KEEPALIVE2_ACK
:
396 r
= read_until(sizeof(*t
), state_buffer
);
398 ldout(async_msgr
->cct
, 1) << __func__
<< " read keeplive timespec failed" << dendl
;
404 t
= (ceph_timespec
*)state_buffer
;
405 set_last_keepalive_ack(utime_t(*t
));
406 ldout(async_msgr
->cct
, 20) << __func__
<< " got KEEPALIVE_ACK" << dendl
;
411 case STATE_OPEN_TAG_ACK
:
414 r
= read_until(sizeof(*seq
), state_buffer
);
416 ldout(async_msgr
->cct
, 1) << __func__
<< " read ack seq failed" << dendl
;
422 seq
= (ceph_le64
*)state_buffer
;
423 ldout(async_msgr
->cct
, 20) << __func__
<< " got ACK" << dendl
;
429 case STATE_OPEN_MESSAGE_HEADER
:
431 #if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
432 ltt_recv_stamp
= ceph_clock_now();
434 recv_stamp
= ceph_clock_now();
435 ldout(async_msgr
->cct
, 20) << __func__
<< " begin MSG" << dendl
;
436 ceph_msg_header header
;
437 ceph_msg_header_old oldheader
;
438 __u32 header_crc
= 0;
440 if (has_feature(CEPH_FEATURE_NOSRCADDR
))
441 len
= sizeof(header
);
443 len
= sizeof(oldheader
);
445 r
= read_until(len
, state_buffer
);
447 ldout(async_msgr
->cct
, 1) << __func__
<< " read message header failed" << dendl
;
453 ldout(async_msgr
->cct
, 20) << __func__
<< " got MSG header" << dendl
;
455 if (has_feature(CEPH_FEATURE_NOSRCADDR
)) {
456 header
= *((ceph_msg_header
*)state_buffer
);
457 if (msgr
->crcflags
& MSG_CRC_HEADER
)
458 header_crc
= ceph_crc32c(0, (unsigned char *)&header
,
459 sizeof(header
) - sizeof(header
.crc
));
461 oldheader
= *((ceph_msg_header_old
*)state_buffer
);
463 memcpy(&header
, &oldheader
, sizeof(header
));
464 header
.src
= oldheader
.src
.name
;
465 header
.reserved
= oldheader
.reserved
;
466 if (msgr
->crcflags
& MSG_CRC_HEADER
) {
467 header
.crc
= oldheader
.crc
;
468 header_crc
= ceph_crc32c(0, (unsigned char *)&oldheader
, sizeof(oldheader
) - sizeof(oldheader
.crc
));
472 ldout(async_msgr
->cct
, 20) << __func__
<< " got envelope type=" << header
.type
473 << " src " << entity_name_t(header
.src
)
474 << " front=" << header
.front_len
475 << " data=" << header
.data_len
476 << " off " << header
.data_off
<< dendl
;
479 if (msgr
->crcflags
& MSG_CRC_HEADER
&& header_crc
!= header
.crc
) {
480 ldout(async_msgr
->cct
,0) << __func__
<< " got bad header crc "
481 << header_crc
<< " != " << header
.crc
<< dendl
;
490 current_header
= header
;
491 state
= STATE_OPEN_MESSAGE_THROTTLE_MESSAGE
;
495 case STATE_OPEN_MESSAGE_THROTTLE_MESSAGE
:
497 if (policy
.throttler_messages
) {
498 ldout(async_msgr
->cct
, 10) << __func__
<< " wants " << 1 << " message from policy throttler "
499 << policy
.throttler_messages
->get_current() << "/"
500 << policy
.throttler_messages
->get_max() << dendl
;
501 if (!policy
.throttler_messages
->get_or_fail()) {
502 ldout(async_msgr
->cct
, 10) << __func__
<< " wants 1 message from policy throttle "
503 << policy
.throttler_messages
->get_current() << "/"
504 << policy
.throttler_messages
->get_max() << " failed, just wait." << dendl
;
505 // following thread pool deal with th full message queue isn't a
506 // short time, so we can wait a ms.
507 if (register_time_events
.empty())
508 register_time_events
.insert(center
->create_time_event(1000, wakeup_handler
));
513 state
= STATE_OPEN_MESSAGE_THROTTLE_BYTES
;
517 case STATE_OPEN_MESSAGE_THROTTLE_BYTES
:
519 cur_msg_size
= current_header
.front_len
+ current_header
.middle_len
+ current_header
.data_len
;
521 if (policy
.throttler_bytes
) {
522 ldout(async_msgr
->cct
, 10) << __func__
<< " wants " << cur_msg_size
<< " bytes from policy throttler "
523 << policy
.throttler_bytes
->get_current() << "/"
524 << policy
.throttler_bytes
->get_max() << dendl
;
525 if (!policy
.throttler_bytes
->get_or_fail(cur_msg_size
)) {
526 ldout(async_msgr
->cct
, 10) << __func__
<< " wants " << cur_msg_size
<< " bytes from policy throttler "
527 << policy
.throttler_bytes
->get_current() << "/"
528 << policy
.throttler_bytes
->get_max() << " failed, just wait." << dendl
;
529 // following thread pool deal with th full message queue isn't a
530 // short time, so we can wait a ms.
531 if (register_time_events
.empty())
532 register_time_events
.insert(center
->create_time_event(1000, wakeup_handler
));
538 state
= STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE
;
542 case STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE
:
545 if (!dispatch_queue
->dispatch_throttler
.get_or_fail(cur_msg_size
)) {
546 ldout(async_msgr
->cct
, 10) << __func__
<< " wants " << cur_msg_size
<< " bytes from dispatch throttle "
547 << dispatch_queue
->dispatch_throttler
.get_current() << "/"
548 << dispatch_queue
->dispatch_throttler
.get_max() << " failed, just wait." << dendl
;
549 // following thread pool deal with th full message queue isn't a
550 // short time, so we can wait a ms.
551 if (register_time_events
.empty())
552 register_time_events
.insert(center
->create_time_event(1000, wakeup_handler
));
557 throttle_stamp
= ceph_clock_now();
558 state
= STATE_OPEN_MESSAGE_READ_FRONT
;
562 case STATE_OPEN_MESSAGE_READ_FRONT
:
565 unsigned front_len
= current_header
.front_len
;
568 front
.push_back(buffer::create(front_len
));
570 r
= read_until(front_len
, front
.c_str());
572 ldout(async_msgr
->cct
, 1) << __func__
<< " read message front failed" << dendl
;
578 ldout(async_msgr
->cct
, 20) << __func__
<< " got front " << front
.length() << dendl
;
580 state
= STATE_OPEN_MESSAGE_READ_MIDDLE
;
583 case STATE_OPEN_MESSAGE_READ_MIDDLE
:
586 unsigned middle_len
= current_header
.middle_len
;
588 if (!middle
.length())
589 middle
.push_back(buffer::create(middle_len
));
591 r
= read_until(middle_len
, middle
.c_str());
593 ldout(async_msgr
->cct
, 1) << __func__
<< " read message middle failed" << dendl
;
598 ldout(async_msgr
->cct
, 20) << __func__
<< " got middle " << middle
.length() << dendl
;
601 state
= STATE_OPEN_MESSAGE_READ_DATA_PREPARE
;
604 case STATE_OPEN_MESSAGE_READ_DATA_PREPARE
:
607 unsigned data_len
= le32_to_cpu(current_header
.data_len
);
608 unsigned data_off
= le32_to_cpu(current_header
.data_off
);
611 map
<ceph_tid_t
,pair
<bufferlist
,int> >::iterator p
= rx_buffers
.find(current_header
.tid
);
612 if (p
!= rx_buffers
.end()) {
613 ldout(async_msgr
->cct
,10) << __func__
<< " seleting rx buffer v " << p
->second
.second
614 << " at offset " << data_off
615 << " len " << p
->second
.first
.length() << dendl
;
616 data_buf
= p
->second
.first
;
617 // make sure it's big enough
618 if (data_buf
.length() < data_len
)
619 data_buf
.push_back(buffer::create(data_len
- data_buf
.length()));
620 data_blp
= data_buf
.begin();
622 ldout(async_msgr
->cct
,20) << __func__
<< " allocating new rx buffer at offset " << data_off
<< dendl
;
623 alloc_aligned_buffer(data_buf
, data_len
, data_off
);
624 data_blp
= data_buf
.begin();
629 state
= STATE_OPEN_MESSAGE_READ_DATA
;
632 case STATE_OPEN_MESSAGE_READ_DATA
:
634 while (msg_left
> 0) {
635 bufferptr bp
= data_blp
.get_current_ptr();
636 unsigned read
= MIN(bp
.length(), msg_left
);
637 r
= read_until(read
, bp
.c_str());
639 ldout(async_msgr
->cct
, 1) << __func__
<< " read data error " << dendl
;
645 data_blp
.advance(read
);
646 data
.append(bp
, 0, read
);
653 state
= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
;
656 case STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
:
658 ceph_msg_footer footer
;
659 ceph_msg_footer_old old_footer
;
662 if (has_feature(CEPH_FEATURE_MSG_AUTH
))
663 len
= sizeof(footer
);
665 len
= sizeof(old_footer
);
667 r
= read_until(len
, state_buffer
);
669 ldout(async_msgr
->cct
, 1) << __func__
<< " read footer data error " << dendl
;
675 if (has_feature(CEPH_FEATURE_MSG_AUTH
)) {
676 footer
= *((ceph_msg_footer
*)state_buffer
);
678 old_footer
= *((ceph_msg_footer_old
*)state_buffer
);
679 footer
.front_crc
= old_footer
.front_crc
;
680 footer
.middle_crc
= old_footer
.middle_crc
;
681 footer
.data_crc
= old_footer
.data_crc
;
683 footer
.flags
= old_footer
.flags
;
685 int aborted
= (footer
.flags
& CEPH_MSG_FOOTER_COMPLETE
) == 0;
686 ldout(async_msgr
->cct
, 10) << __func__
<< " aborted = " << aborted
<< dendl
;
688 ldout(async_msgr
->cct
, 0) << __func__
<< " got " << front
.length() << " + " << middle
.length() << " + " << data
.length()
689 << " byte message.. ABORTED" << dendl
;
693 ldout(async_msgr
->cct
, 20) << __func__
<< " got " << front
.length() << " + " << middle
.length()
694 << " + " << data
.length() << " byte message" << dendl
;
695 Message
*message
= decode_message(async_msgr
->cct
, async_msgr
->crcflags
, current_header
, footer
,
696 front
, middle
, data
, this);
698 ldout(async_msgr
->cct
, 1) << __func__
<< " decode message failed " << dendl
;
703 // Check the signature if one should be present. A zero return indicates success. PLR
706 if (session_security
.get() == NULL
) {
707 ldout(async_msgr
->cct
, 10) << __func__
<< " no session security set" << dendl
;
709 if (session_security
->check_message_signature(message
)) {
710 ldout(async_msgr
->cct
, 0) << __func__
<< " Signature check failed" << dendl
;
715 message
->set_byte_throttler(policy
.throttler_bytes
);
716 message
->set_message_throttler(policy
.throttler_messages
);
718 // store reservation size in message, so we don't get confused
719 // by messages entering the dispatch queue through other paths.
720 message
->set_dispatch_throttle_size(cur_msg_size
);
722 message
->set_recv_stamp(recv_stamp
);
723 message
->set_throttle_stamp(throttle_stamp
);
724 message
->set_recv_complete_stamp(ceph_clock_now());
726 // check received seq#. if it is old, drop the message.
727 // note that incoming messages may skip ahead. this is convenient for the client
728 // side queueing because messages can't be renumbered, but the (kernel) client will
729 // occasionally pull a message out of the sent queue to send elsewhere. in that case
730 // it doesn't matter if we "got" it or not.
731 uint64_t cur_seq
= in_seq
;
732 if (message
->get_seq() <= cur_seq
) {
733 ldout(async_msgr
->cct
,0) << __func__
<< " got old message "
734 << message
->get_seq() << " <= " << cur_seq
<< " " << message
<< " " << *message
735 << ", discarding" << dendl
;
737 if (has_feature(CEPH_FEATURE_RECONNECT_SEQ
) && async_msgr
->cct
->_conf
->ms_die_on_old_message
)
738 assert(0 == "old msgs despite reconnect_seq feature");
741 if (message
->get_seq() > cur_seq
+ 1) {
742 ldout(async_msgr
->cct
, 0) << __func__
<< " missed message? skipped from seq "
743 << cur_seq
<< " to " << message
->get_seq() << dendl
;
744 if (async_msgr
->cct
->_conf
->ms_die_on_skipped_message
)
745 assert(0 == "skipped incoming seq");
748 message
->set_connection(this);
750 #if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
751 if (message
->get_type() == CEPH_MSG_OSD_OP
|| message
->get_type() == CEPH_MSG_OSD_OPREPLY
) {
752 utime_t ltt_processed_stamp
= ceph_clock_now();
753 double usecs_elapsed
= (ltt_processed_stamp
.to_nsec()-ltt_recv_stamp
.to_nsec())/1000;
755 if (message
->get_type() == CEPH_MSG_OSD_OP
)
756 OID_ELAPSED_WITH_MSG(message
, usecs_elapsed
, "TIME_TO_DECODE_OSD_OP", false);
758 OID_ELAPSED_WITH_MSG(message
, usecs_elapsed
, "TIME_TO_DECODE_OSD_OPREPLY", false);
762 // note last received message.
763 in_seq
= message
->get_seq();
764 ldout(async_msgr
->cct
, 5) << " rx " << message
->get_source() << " seq "
765 << message
->get_seq() << " " << message
766 << " " << *message
<< dendl
;
770 need_dispatch_writer
= true;
774 logger
->inc(l_msgr_recv_messages
);
775 logger
->inc(l_msgr_recv_bytes
, cur_msg_size
+ sizeof(ceph_msg_header
) + sizeof(ceph_msg_footer
));
777 async_msgr
->ms_fast_preprocess(message
);
778 auto fast_dispatch_time
= ceph::mono_clock::now();
779 logger
->tinc(l_msgr_running_recv_time
, fast_dispatch_time
- recv_start_time
);
781 utime_t release
= message
->get_recv_stamp();
782 double delay_period
= 0;
783 if (rand() % 10000 < async_msgr
->cct
->_conf
->ms_inject_delay_probability
* 10000.0) {
784 delay_period
= async_msgr
->cct
->_conf
->ms_inject_delay_max
* (double)(rand() % 10000) / 10000.0;
785 release
+= delay_period
;
786 ldout(async_msgr
->cct
, 1) << "queue_received will delay until " << release
<< " on "
787 << message
<< " " << *message
<< dendl
;
789 delay_state
->queue(delay_period
, release
, message
);
790 } else if (async_msgr
->ms_can_fast_dispatch(message
)) {
792 dispatch_queue
->fast_dispatch(message
);
793 recv_start_time
= ceph::mono_clock::now();
794 logger
->tinc(l_msgr_running_fast_dispatch_time
,
795 recv_start_time
- fast_dispatch_time
);
798 dispatch_queue
->enqueue(message
, message
->get_priority(), conn_id
);
804 case STATE_OPEN_TAG_CLOSE
:
806 ldout(async_msgr
->cct
, 20) << __func__
<< " got CLOSE" << dendl
;
813 ldout(async_msgr
->cct
, 20) << __func__
<< " enter STANDY" << dendl
;
820 ldout(async_msgr
->cct
, 20) << __func__
<< " enter none state" << dendl
;
826 ldout(async_msgr
->cct
, 20) << __func__
<< " socket closed" << dendl
;
832 ldout(async_msgr
->cct
, 1) << __func__
<< " enter wait state, failing" << dendl
;
838 if (_process_connection() < 0)
843 } while (prev_state
!= state
);
845 if (need_dispatch_writer
&& is_connected())
846 center
->dispatch_event_external(write_handler
);
848 logger
->tinc(l_msgr_running_recv_time
, ceph::mono_clock::now() - recv_start_time
);
855 ssize_t
AsyncConnection::_process_connection()
860 case STATE_WAIT_SEND
:
862 std::lock_guard
<std::mutex
> l(write_lock
);
863 if (!outcoming_bl
.length()) {
864 assert(state_after_send
);
865 state
= state_after_send
;
866 state_after_send
= STATE_NONE
;
871 case STATE_CONNECTING
:
873 assert(!policy
.server
);
875 // reset connect state variables
876 got_bad_auth
= false;
879 authorizer_buf
.clear();
880 memset(&connect_msg
, 0, sizeof(connect_msg
));
881 memset(&connect_reply
, 0, sizeof(connect_reply
));
883 global_seq
= async_msgr
->get_global_seq();
884 // close old socket. this is safe because we stopped the reader thread above.
886 center
->delete_file_event(cs
.fd(), EVENT_READABLE
|EVENT_WRITABLE
);
891 opts
.priority
= async_msgr
->get_socket_priority();
892 opts
.connect_bind_addr
= msgr
->get_myaddr();
893 r
= worker
->connect(get_peer_addr(), opts
, &cs
);
897 center
->create_file_event(cs
.fd(), EVENT_READABLE
, read_handler
);
898 state
= STATE_CONNECTING_RE
;
902 case STATE_CONNECTING_RE
:
904 r
= cs
.is_connected();
906 ldout(async_msgr
->cct
, 1) << __func__
<< " reconnect failed " << dendl
;
907 if (r
== -ECONNREFUSED
) {
908 ldout(async_msgr
->cct
, 2) << __func__
<< " connection refused!" << dendl
;
909 dispatch_queue
->queue_refused(this);
913 ldout(async_msgr
->cct
, 10) << __func__
<< " nonblock connect inprogress" << dendl
;
914 if (async_msgr
->get_stack()->nonblock_connect_need_writable_event())
915 center
->create_file_event(cs
.fd(), EVENT_WRITABLE
, read_handler
);
919 center
->delete_file_event(cs
.fd(), EVENT_WRITABLE
);
920 ldout(async_msgr
->cct
, 10) << __func__
<< " connect successfully, ready to send banner" << dendl
;
923 bl
.append(CEPH_BANNER
, strlen(CEPH_BANNER
));
926 state
= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
;
927 ldout(async_msgr
->cct
, 10) << __func__
<< " connect write banner done: "
928 << get_peer_addr() << dendl
;
930 state
= STATE_WAIT_SEND
;
931 state_after_send
= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
;
932 ldout(async_msgr
->cct
, 10) << __func__
<< " connect wait for write banner: "
933 << get_peer_addr() << dendl
;
941 case STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
:
943 entity_addr_t paddr
, peer_addr_for_me
;
945 unsigned banner_len
= strlen(CEPH_BANNER
);
946 unsigned need_len
= banner_len
+ sizeof(ceph_entity_addr
)*2;
947 r
= read_until(need_len
, state_buffer
);
949 ldout(async_msgr
->cct
, 1) << __func__
<< " read banner and identify addresses failed" << dendl
;
955 if (memcmp(state_buffer
, CEPH_BANNER
, banner_len
)) {
956 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol error (bad banner) on peer "
957 << get_peer_addr() << dendl
;
962 bl
.append(state_buffer
+banner_len
, sizeof(ceph_entity_addr
)*2);
963 bufferlist::iterator p
= bl
.begin();
966 ::decode(peer_addr_for_me
, p
);
967 } catch (const buffer::error
& e
) {
968 lderr(async_msgr
->cct
) << __func__
<< " decode peer addr failed " << dendl
;
971 ldout(async_msgr
->cct
, 20) << __func__
<< " connect read peer addr "
972 << paddr
<< " on socket " << cs
.fd() << dendl
;
973 if (peer_addr
!= paddr
) {
974 if (paddr
.is_blank_ip() && peer_addr
.get_port() == paddr
.get_port() &&
975 peer_addr
.get_nonce() == paddr
.get_nonce()) {
976 ldout(async_msgr
->cct
, 0) << __func__
<< " connect claims to be " << paddr
977 << " not " << peer_addr
978 << " - presumably this is the same node!" << dendl
;
980 ldout(async_msgr
->cct
, 10) << __func__
<< " connect claims to be "
981 << paddr
<< " not " << peer_addr
<< dendl
;
986 ldout(async_msgr
->cct
, 20) << __func__
<< " connect peer addr for me is " << peer_addr_for_me
<< dendl
;
988 async_msgr
->learned_addr(peer_addr_for_me
);
989 if (async_msgr
->cct
->_conf
->ms_inject_internal_delays
990 && async_msgr
->cct
->_conf
->ms_inject_socket_failures
) {
991 if (rand() % async_msgr
->cct
->_conf
->ms_inject_socket_failures
== 0) {
992 ldout(msgr
->cct
, 10) << __func__
<< " sleep for "
993 << async_msgr
->cct
->_conf
->ms_inject_internal_delays
<< dendl
;
995 t
.set_from_double(async_msgr
->cct
->_conf
->ms_inject_internal_delays
);
1001 if (state
!= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
) {
1002 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while learned_addr, mark_down or "
1003 << " replacing must be happened just now" << dendl
;
1007 ::encode(async_msgr
->get_myaddr(), myaddrbl
, 0); // legacy
1008 r
= try_send(myaddrbl
);
1010 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1011 ldout(async_msgr
->cct
, 10) << __func__
<< " connect sent my addr "
1012 << async_msgr
->get_myaddr() << dendl
;
1014 state
= STATE_WAIT_SEND
;
1015 state_after_send
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1016 ldout(async_msgr
->cct
, 10) << __func__
<< " connect send my addr done: "
1017 << async_msgr
->get_myaddr() << dendl
;
1019 ldout(async_msgr
->cct
, 2) << __func__
<< " connect couldn't write my addr, "
1020 << cpp_strerror(r
) << dendl
;
1027 case STATE_CONNECTING_SEND_CONNECT_MSG
:
1030 authorizer
= async_msgr
->get_authorizer(peer_type
, false);
1034 connect_msg
.features
= policy
.features_supported
;
1035 connect_msg
.host_type
= async_msgr
->get_myinst().name
.type();
1036 connect_msg
.global_seq
= global_seq
;
1037 connect_msg
.connect_seq
= connect_seq
;
1038 connect_msg
.protocol_version
= async_msgr
->get_proto_version(peer_type
, true);
1039 connect_msg
.authorizer_protocol
= authorizer
? authorizer
->protocol
: 0;
1040 connect_msg
.authorizer_len
= authorizer
? authorizer
->bl
.length() : 0;
1042 ldout(async_msgr
->cct
, 10) << __func__
<< " connect_msg.authorizer_len="
1043 << connect_msg
.authorizer_len
<< " protocol="
1044 << connect_msg
.authorizer_protocol
<< dendl
;
1045 connect_msg
.flags
= 0;
1047 connect_msg
.flags
|= CEPH_MSG_CONNECT_LOSSY
; // this is fyi, actually, server decides!
1048 bl
.append((char*)&connect_msg
, sizeof(connect_msg
));
1050 bl
.append(authorizer
->bl
.c_str(), authorizer
->bl
.length());
1052 ldout(async_msgr
->cct
, 10) << __func__
<< " connect sending gseq=" << global_seq
<< " cseq="
1053 << connect_seq
<< " proto=" << connect_msg
.protocol_version
<< dendl
;
1057 state
= STATE_CONNECTING_WAIT_CONNECT_REPLY
;
1058 ldout(async_msgr
->cct
,20) << __func__
<< " connect wrote (self +) cseq, waiting for reply" << dendl
;
1060 state
= STATE_WAIT_SEND
;
1061 state_after_send
= STATE_CONNECTING_WAIT_CONNECT_REPLY
;
1062 ldout(async_msgr
->cct
, 10) << __func__
<< " continue send reply " << dendl
;
1064 ldout(async_msgr
->cct
, 2) << __func__
<< " connect couldn't send reply "
1065 << cpp_strerror(r
) << dendl
;
1072 case STATE_CONNECTING_WAIT_CONNECT_REPLY
:
1074 r
= read_until(sizeof(connect_reply
), state_buffer
);
1076 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect reply failed" << dendl
;
1082 connect_reply
= *((ceph_msg_connect_reply
*)state_buffer
);
1084 ldout(async_msgr
->cct
, 20) << __func__
<< " connect got reply tag " << (int)connect_reply
.tag
1085 << " connect_seq " << connect_reply
.connect_seq
<< " global_seq "
1086 << connect_reply
.global_seq
<< " proto " << connect_reply
.protocol_version
1087 << " flags " << (int)connect_reply
.flags
<< " features "
1088 << connect_reply
.features
<< dendl
;
1089 state
= STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
;
1094 case STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
:
1096 bufferlist authorizer_reply
;
1097 if (connect_reply
.authorizer_len
) {
1098 ldout(async_msgr
->cct
, 10) << __func__
<< " reply.authorizer_len=" << connect_reply
.authorizer_len
<< dendl
;
1099 assert(connect_reply
.authorizer_len
< 4096);
1100 r
= read_until(connect_reply
.authorizer_len
, state_buffer
);
1102 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect reply authorizer failed" << dendl
;
1108 authorizer_reply
.append(state_buffer
, connect_reply
.authorizer_len
);
1110 if (connect_reply
.tag
== CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER
) {
1111 ldout(async_msgr
->cct
,10) << __func__
<< " connect got auth challenge" << dendl
;
1112 authorizer
->add_challenge(async_msgr
->cct
, authorizer_reply
);
1113 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1117 auto iter
= authorizer_reply
.begin();
1118 if (authorizer
&& !authorizer
->verify_reply(iter
)) {
1119 ldout(async_msgr
->cct
, 0) << __func__
<< " failed verifying authorize reply" << dendl
;
1123 r
= handle_connect_reply(connect_msg
, connect_reply
);
1127 // state must be changed!
1128 assert(state
!= STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
);
1132 case STATE_CONNECTING_WAIT_ACK_SEQ
:
1134 uint64_t newly_acked_seq
= 0;
1136 r
= read_until(sizeof(newly_acked_seq
), state_buffer
);
1138 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect ack seq failed" << dendl
;
1144 newly_acked_seq
= *((uint64_t*)state_buffer
);
1145 ldout(async_msgr
->cct
, 2) << __func__
<< " got newly_acked_seq " << newly_acked_seq
1146 << " vs out_seq " << out_seq
<< dendl
;
1147 discard_requeued_up_to(newly_acked_seq
);
1148 //while (newly_acked_seq > out_seq.read()) {
1149 // Message *m = _get_next_outgoing(NULL);
1151 // ldout(async_msgr->cct, 2) << __func__ << " discarding previously sent " << m->get_seq()
1152 // << " " << *m << dendl;
1153 // assert(m->get_seq() <= newly_acked_seq);
1159 uint64_t s
= in_seq
;
1160 bl
.append((char*)&s
, sizeof(s
));
1163 state
= STATE_CONNECTING_READY
;
1164 ldout(async_msgr
->cct
, 10) << __func__
<< " send in_seq done " << dendl
;
1166 state_after_send
= STATE_CONNECTING_READY
;
1167 state
= STATE_WAIT_SEND
;
1168 ldout(async_msgr
->cct
, 10) << __func__
<< " continue send in_seq " << dendl
;
1175 case STATE_CONNECTING_READY
:
1178 peer_global_seq
= connect_reply
.global_seq
;
1179 policy
.lossy
= connect_reply
.flags
& CEPH_MSG_CONNECT_LOSSY
;
1183 assert(connect_seq
== connect_reply
.connect_seq
);
1184 backoff
= utime_t();
1185 set_features((uint64_t)connect_reply
.features
& (uint64_t)connect_msg
.features
);
1186 ldout(async_msgr
->cct
, 10) << __func__
<< " connect success " << connect_seq
1187 << ", lossy = " << policy
.lossy
<< ", features "
1188 << get_features() << dendl
;
1190 // If we have an authorizer, get a new AuthSessionHandler to deal with ongoing security of the
1192 if (authorizer
!= NULL
) {
1193 session_security
.reset(
1194 get_auth_session_handler(async_msgr
->cct
,
1195 authorizer
->protocol
,
1196 authorizer
->session_key
,
1199 // We have no authorizer, so we shouldn't be applying security to messages in this AsyncConnection. PLR
1200 session_security
.reset();
1204 assert(delay_state
->ready());
1205 dispatch_queue
->queue_connect(this);
1206 async_msgr
->ms_deliver_handle_fast_connect(this);
1208 // make sure no pending tick timer
1210 center
->delete_time_event(last_tick_id
);
1211 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);
1213 // message may in queue between last _try_send and connection ready
1214 // write event may already notify and we need to force scheduler again
1216 can_write
= WriteStatus::CANWRITE
;
1218 center
->dispatch_event_external(write_handler
);
1219 write_lock
.unlock();
1220 maybe_start_delay_thread();
1224 case STATE_ACCEPTING
:
1227 center
->create_file_event(cs
.fd(), EVENT_READABLE
, read_handler
);
1229 bl
.append(CEPH_BANNER
, strlen(CEPH_BANNER
));
1231 ::encode(async_msgr
->get_myaddr(), bl
, 0); // legacy
1232 port
= async_msgr
->get_myaddr().get_port();
1233 ::encode(socket_addr
, bl
, 0); // legacy
1234 ldout(async_msgr
->cct
, 1) << __func__
<< " sd=" << cs
.fd() << " " << socket_addr
<< dendl
;
1238 state
= STATE_ACCEPTING_WAIT_BANNER_ADDR
;
1239 ldout(async_msgr
->cct
, 10) << __func__
<< " write banner and addr done: "
1240 << get_peer_addr() << dendl
;
1242 state
= STATE_WAIT_SEND
;
1243 state_after_send
= STATE_ACCEPTING_WAIT_BANNER_ADDR
;
1244 ldout(async_msgr
->cct
, 10) << __func__
<< " wait for write banner and addr: "
1245 << get_peer_addr() << dendl
;
1252 case STATE_ACCEPTING_WAIT_BANNER_ADDR
:
1255 entity_addr_t peer_addr
;
1257 r
= read_until(strlen(CEPH_BANNER
) + sizeof(ceph_entity_addr
), state_buffer
);
1259 ldout(async_msgr
->cct
, 1) << __func__
<< " read peer banner and addr failed" << dendl
;
1265 if (memcmp(state_buffer
, CEPH_BANNER
, strlen(CEPH_BANNER
))) {
1266 ldout(async_msgr
->cct
, 1) << __func__
<< " accept peer sent bad banner '" << state_buffer
1267 << "' (should be '" << CEPH_BANNER
<< "')" << dendl
;
1271 addr_bl
.append(state_buffer
+strlen(CEPH_BANNER
), sizeof(ceph_entity_addr
));
1273 bufferlist::iterator ti
= addr_bl
.begin();
1274 ::decode(peer_addr
, ti
);
1277 ldout(async_msgr
->cct
, 10) << __func__
<< " accept peer addr is " << peer_addr
<< dendl
;
1278 if (peer_addr
.is_blank_ip()) {
1279 // peer apparently doesn't know what ip they have; figure it out for them.
1280 int port
= peer_addr
.get_port();
1281 peer_addr
.u
= socket_addr
.u
;
1282 peer_addr
.set_port(port
);
1283 ldout(async_msgr
->cct
, 0) << __func__
<< " accept peer addr is really " << peer_addr
1284 << " (socket is " << socket_addr
<< ")" << dendl
;
1286 set_peer_addr(peer_addr
); // so that connection_state gets set up
1287 state
= STATE_ACCEPTING_WAIT_CONNECT_MSG
;
1291 case STATE_ACCEPTING_WAIT_CONNECT_MSG
:
1293 r
= read_until(sizeof(connect_msg
), state_buffer
);
1295 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect msg failed" << dendl
;
1301 connect_msg
= *((ceph_msg_connect
*)state_buffer
);
1302 state
= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
;
1306 case STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
:
1308 bufferlist authorizer_reply
;
1310 if (connect_msg
.authorizer_len
) {
1311 if (!authorizer_buf
.length())
1312 authorizer_buf
.push_back(buffer::create(connect_msg
.authorizer_len
));
1314 r
= read_until(connect_msg
.authorizer_len
, authorizer_buf
.c_str());
1316 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect authorizer failed" << dendl
;
1323 ldout(async_msgr
->cct
, 20) << __func__
<< " accept got peer connect_seq "
1324 << connect_msg
.connect_seq
<< " global_seq "
1325 << connect_msg
.global_seq
<< dendl
;
1326 set_peer_type(connect_msg
.host_type
);
1327 policy
= async_msgr
->get_policy(connect_msg
.host_type
);
1328 ldout(async_msgr
->cct
, 10) << __func__
<< " accept of host_type " << connect_msg
.host_type
1329 << ", policy.lossy=" << policy
.lossy
<< " policy.server="
1330 << policy
.server
<< " policy.standby=" << policy
.standby
1331 << " policy.resetcheck=" << policy
.resetcheck
<< dendl
;
1333 r
= handle_connect_msg(connect_msg
, authorizer_buf
, authorizer_reply
);
1337 // state is changed by "handle_connect_msg"
1338 assert(state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
);
1342 case STATE_ACCEPTING_WAIT_SEQ
:
1344 uint64_t newly_acked_seq
;
1345 r
= read_until(sizeof(newly_acked_seq
), state_buffer
);
1347 ldout(async_msgr
->cct
, 1) << __func__
<< " read ack seq failed" << dendl
;
1348 goto fail_registered
;
1353 newly_acked_seq
= *((uint64_t*)state_buffer
);
1354 ldout(async_msgr
->cct
, 2) << __func__
<< " accept get newly_acked_seq " << newly_acked_seq
<< dendl
;
1355 discard_requeued_up_to(newly_acked_seq
);
1356 state
= STATE_ACCEPTING_READY
;
1360 case STATE_ACCEPTING_READY
:
1362 ldout(async_msgr
->cct
, 20) << __func__
<< " accept done" << dendl
;
1364 memset(&connect_msg
, 0, sizeof(connect_msg
));
1367 assert(delay_state
->ready());
1368 // make sure no pending tick timer
1370 center
->delete_time_event(last_tick_id
);
1371 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);
1374 can_write
= WriteStatus::CANWRITE
;
1376 center
->dispatch_event_external(write_handler
);
1377 write_lock
.unlock();
1378 maybe_start_delay_thread();
1384 lderr(async_msgr
->cct
) << __func__
<< " bad state: " << state
<< dendl
;
1392 ldout(async_msgr
->cct
, 10) << "accept fault after register" << dendl
;
1399 int AsyncConnection::handle_connect_reply(ceph_msg_connect
&connect
, ceph_msg_connect_reply
&reply
)
1401 uint64_t feat_missing
;
1402 if (reply
.tag
== CEPH_MSGR_TAG_FEATURES
) {
1403 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol feature mismatch, my "
1404 << std::hex
<< connect
.features
<< " < peer "
1405 << reply
.features
<< " missing "
1406 << (reply
.features
& ~policy
.features_supported
)
1407 << std::dec
<< dendl
;
1411 if (reply
.tag
== CEPH_MSGR_TAG_BADPROTOVER
) {
1412 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol version mismatch, my "
1413 << connect
.protocol_version
<< " != " << reply
.protocol_version
1418 if (reply
.tag
== CEPH_MSGR_TAG_BADAUTHORIZER
) {
1419 ldout(async_msgr
->cct
,0) << __func__
<< " connect got BADAUTHORIZER" << dendl
;
1422 got_bad_auth
= true;
1424 authorizer
= async_msgr
->get_authorizer(peer_type
, true); // try harder
1425 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1427 if (reply
.tag
== CEPH_MSGR_TAG_RESETSESSION
) {
1428 ldout(async_msgr
->cct
, 0) << __func__
<< " connect got RESETSESSION" << dendl
;
1429 was_session_reset();
1430 // see was_session_reset
1431 outcoming_bl
.clear();
1432 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1434 if (reply
.tag
== CEPH_MSGR_TAG_RETRY_GLOBAL
) {
1435 global_seq
= async_msgr
->get_global_seq(reply
.global_seq
);
1436 ldout(async_msgr
->cct
, 5) << __func__
<< " connect got RETRY_GLOBAL "
1437 << reply
.global_seq
<< " chose new "
1438 << global_seq
<< dendl
;
1439 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1441 if (reply
.tag
== CEPH_MSGR_TAG_RETRY_SESSION
) {
1442 assert(reply
.connect_seq
> connect_seq
);
1443 ldout(async_msgr
->cct
, 5) << __func__
<< " connect got RETRY_SESSION "
1444 << connect_seq
<< " -> "
1445 << reply
.connect_seq
<< dendl
;
1446 connect_seq
= reply
.connect_seq
;
1447 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1449 if (reply
.tag
== CEPH_MSGR_TAG_WAIT
) {
1450 ldout(async_msgr
->cct
, 1) << __func__
<< " connect got WAIT (connection race)" << dendl
;
1454 feat_missing
= policy
.features_required
& ~(uint64_t)connect_reply
.features
;
1456 ldout(async_msgr
->cct
, 1) << __func__
<< " missing required features " << std::hex
1457 << feat_missing
<< std::dec
<< dendl
;
1461 if (reply
.tag
== CEPH_MSGR_TAG_SEQ
) {
1462 ldout(async_msgr
->cct
, 10) << __func__
<< " got CEPH_MSGR_TAG_SEQ, reading acked_seq and writing in_seq" << dendl
;
1463 state
= STATE_CONNECTING_WAIT_ACK_SEQ
;
1465 if (reply
.tag
== CEPH_MSGR_TAG_READY
) {
1466 ldout(async_msgr
->cct
, 10) << __func__
<< " got CEPH_MSGR_TAG_READY " << dendl
;
1467 state
= STATE_CONNECTING_READY
;
1476 ssize_t
AsyncConnection::handle_connect_msg(ceph_msg_connect
&connect
, bufferlist
&authorizer_bl
,
1477 bufferlist
&authorizer_reply
)
1480 ceph_msg_connect_reply reply
;
1481 bufferlist reply_bl
;
1483 memset(&reply
, 0, sizeof(reply
));
1484 reply
.protocol_version
= async_msgr
->get_proto_version(peer_type
, false);
1487 ldout(async_msgr
->cct
, 10) << __func__
<< " accept my proto " << reply
.protocol_version
1488 << ", their proto " << connect
.protocol_version
<< dendl
;
1489 if (connect
.protocol_version
!= reply
.protocol_version
) {
1490 return _reply_accept(CEPH_MSGR_TAG_BADPROTOVER
, connect
, reply
, authorizer_reply
);
1492 // require signatures for cephx?
1493 if (connect
.authorizer_protocol
== CEPH_AUTH_CEPHX
) {
1494 if (peer_type
== CEPH_ENTITY_TYPE_OSD
||
1495 peer_type
== CEPH_ENTITY_TYPE_MDS
||
1496 peer_type
== CEPH_ENTITY_TYPE_MGR
) {
1497 if (async_msgr
->cct
->_conf
->cephx_require_signatures
||
1498 async_msgr
->cct
->_conf
->cephx_cluster_require_signatures
) {
1499 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring MSG_AUTH feature bit for cluster" << dendl
;
1500 policy
.features_required
|= CEPH_FEATURE_MSG_AUTH
;
1502 if (async_msgr
->cct
->_conf
->cephx_require_version
>= 2 ||
1503 async_msgr
->cct
->_conf
->cephx_cluster_require_version
>= 2) {
1504 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring cephx v2 feature bit for cluster" << dendl
;
1505 policy
.features_required
|= CEPH_FEATUREMASK_CEPHX_V2
;
1508 if (async_msgr
->cct
->_conf
->cephx_require_signatures
||
1509 async_msgr
->cct
->_conf
->cephx_service_require_signatures
) {
1510 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring MSG_AUTH feature bit for service" << dendl
;
1511 policy
.features_required
|= CEPH_FEATURE_MSG_AUTH
;
1513 if (async_msgr
->cct
->_conf
->cephx_require_version
>= 2 ||
1514 async_msgr
->cct
->_conf
->cephx_service_require_version
>= 2) {
1515 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring cephx v2 feature bit for service" << dendl
;
1516 policy
.features_required
|= CEPH_FEATUREMASK_CEPHX_V2
;
1521 uint64_t feat_missing
= policy
.features_required
& ~(uint64_t)connect
.features
;
1523 ldout(async_msgr
->cct
, 1) << __func__
<< " peer missing required features "
1524 << std::hex
<< feat_missing
<< std::dec
<< dendl
;
1525 return _reply_accept(CEPH_MSGR_TAG_FEATURES
, connect
, reply
, authorizer_reply
);
1530 bool authorizer_valid
;
1531 bool need_challenge
= HAVE_FEATURE(connect
.features
, CEPHX_V2
);
1532 bool had_challenge
= (bool)authorizer_challenge
;
1533 if (!async_msgr
->verify_authorizer(
1534 this, peer_type
, connect
.authorizer_protocol
, authorizer_bl
,
1535 authorizer_reply
, authorizer_valid
, session_key
,
1536 need_challenge
? &authorizer_challenge
: nullptr) ||
1537 !authorizer_valid
) {
1540 if (need_challenge
&& !had_challenge
&& authorizer_challenge
) {
1541 ldout(async_msgr
->cct
,0) << __func__
<< ": challenging authorizer"
1543 assert(authorizer_reply
.length());
1544 tag
= CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER
;
1546 ldout(async_msgr
->cct
,0) << __func__
<< ": got bad authorizer" << dendl
;
1547 tag
= CEPH_MSGR_TAG_BADAUTHORIZER
;
1549 session_security
.reset();
1550 return _reply_accept(tag
, connect
, reply
, authorizer_reply
);
1553 // We've verified the authorizer for this AsyncConnection, so set up the session security structure. PLR
1554 ldout(async_msgr
->cct
, 10) << __func__
<< " accept setting up session_security." << dendl
;
1557 AsyncConnectionRef existing
= async_msgr
->lookup_conn(peer_addr
);
1562 if (state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
1563 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while accept, it must be mark_down" << dendl
;
1564 assert(state
== STATE_CLOSED
);
1568 if (existing
== this)
1571 // There is no possible that existing connection will acquire this
1572 // connection's lock
1573 existing
->lock
.lock(); // skip lockdep check (we are locking a second AsyncConnection here)
1575 if (existing
->state
== STATE_CLOSED
) {
1576 ldout(async_msgr
->cct
, 1) << __func__
<< " existing already closed." << dendl
;
1577 existing
->lock
.unlock();
1582 if (existing
->replacing
) {
1583 ldout(async_msgr
->cct
, 1) << __func__
<< " existing racing replace happened while replacing."
1584 << " existing_state=" << get_state_name(existing
->state
) << dendl
;
1585 reply
.global_seq
= existing
->peer_global_seq
;
1586 r
= _reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
);
1587 existing
->lock
.unlock();
1593 if (connect
.global_seq
< existing
->peer_global_seq
) {
1594 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
1595 << ".gseq " << existing
->peer_global_seq
<< " > "
1596 << connect
.global_seq
<< ", RETRY_GLOBAL" << dendl
;
1597 reply
.global_seq
= existing
->peer_global_seq
; // so we can send it below..
1598 existing
->lock
.unlock();
1599 return _reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
);
1601 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
1602 << ".gseq " << existing
->peer_global_seq
1603 << " <= " << connect
.global_seq
<< ", looks ok" << dendl
;
1606 if (existing
->policy
.lossy
) {
1607 ldout(async_msgr
->cct
, 0) << __func__
<< " accept replacing existing (lossy) channel (new one lossy="
1608 << policy
.lossy
<< ")" << dendl
;
1609 existing
->was_session_reset();
1613 ldout(async_msgr
->cct
, 0) << __func__
<< " accept connect_seq " << connect
.connect_seq
1614 << " vs existing csq=" << existing
->connect_seq
<< " existing_state="
1615 << get_state_name(existing
->state
) << dendl
;
1617 if (connect
.connect_seq
== 0 && existing
->connect_seq
> 0) {
1618 ldout(async_msgr
->cct
,0) << __func__
<< " accept peer reset, then tried to connect to us, replacing" << dendl
;
1619 // this is a hard reset from peer
1620 is_reset_from_peer
= true;
1621 if (policy
.resetcheck
)
1622 existing
->was_session_reset(); // this resets out_queue, msg_ and connect_seq #'s
1626 if (connect
.connect_seq
< existing
->connect_seq
) {
1627 // old attempt, or we sent READY but they didn't get it.
1628 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
<< ".cseq "
1629 << existing
->connect_seq
<< " > " << connect
.connect_seq
1630 << ", RETRY_SESSION" << dendl
;
1631 reply
.connect_seq
= existing
->connect_seq
+ 1;
1632 existing
->lock
.unlock();
1633 return _reply_accept(CEPH_MSGR_TAG_RETRY_SESSION
, connect
, reply
, authorizer_reply
);
1636 if (connect
.connect_seq
== existing
->connect_seq
) {
1637 // if the existing connection successfully opened, and/or
1638 // subsequently went to standby, then the peer should bump
1639 // their connect_seq and retry: this is not a connection race
1640 // we need to resolve here.
1641 if (existing
->state
== STATE_OPEN
||
1642 existing
->state
== STATE_STANDBY
) {
1643 ldout(async_msgr
->cct
, 10) << __func__
<< " accept connection race, existing " << existing
1644 << ".cseq " << existing
->connect_seq
<< " == "
1645 << connect
.connect_seq
<< ", OPEN|STANDBY, RETRY_SESSION" << dendl
;
1646 reply
.connect_seq
= existing
->connect_seq
+ 1;
1647 existing
->lock
.unlock();
1648 return _reply_accept(CEPH_MSGR_TAG_RETRY_SESSION
, connect
, reply
, authorizer_reply
);
1652 if (peer_addr
< async_msgr
->get_myaddr() || existing
->policy
.server
) {
1654 ldout(async_msgr
->cct
, 10) << __func__
<< " accept connection race, existing " << existing
1655 << ".cseq " << existing
->connect_seq
<< " == " << connect
.connect_seq
1656 << ", or we are server, replacing my attempt" << dendl
;
1659 // our existing outgoing wins
1660 ldout(async_msgr
->cct
,10) << __func__
<< " accept connection race, existing "
1661 << existing
<< ".cseq " << existing
->connect_seq
1662 << " == " << connect
.connect_seq
<< ", sending WAIT" << dendl
;
1663 assert(peer_addr
> async_msgr
->get_myaddr());
1664 existing
->lock
.unlock();
1665 return _reply_accept(CEPH_MSGR_TAG_WAIT
, connect
, reply
, authorizer_reply
);
1669 assert(connect
.connect_seq
> existing
->connect_seq
);
1670 assert(connect
.global_seq
>= existing
->peer_global_seq
);
1671 if (policy
.resetcheck
&& // RESETSESSION only used by servers; peers do not reset each other
1672 existing
->connect_seq
== 0) {
1673 ldout(async_msgr
->cct
, 0) << __func__
<< " accept we reset (peer sent cseq "
1674 << connect
.connect_seq
<< ", " << existing
<< ".cseq = "
1675 << existing
->connect_seq
<< "), sending RESETSESSION" << dendl
;
1676 existing
->lock
.unlock();
1677 return _reply_accept(CEPH_MSGR_TAG_RESETSESSION
, connect
, reply
, authorizer_reply
);
1681 ldout(async_msgr
->cct
, 10) << __func__
<< " accept peer sent cseq " << connect
.connect_seq
1682 << " > " << existing
->connect_seq
<< dendl
;
1685 else if (!replacing
&& connect
.connect_seq
> 0) {
1686 // we reset, and they are opening a new session
1687 ldout(async_msgr
->cct
, 0) << __func__
<< " accept we reset (peer sent cseq "
1688 << connect
.connect_seq
<< "), sending RESETSESSION" << dendl
;
1689 return _reply_accept(CEPH_MSGR_TAG_RESETSESSION
, connect
, reply
, authorizer_reply
);
1692 ldout(async_msgr
->cct
, 10) << __func__
<< " accept new session" << dendl
;
1699 ldout(async_msgr
->cct
, 10) << __func__
<< " accept replacing " << existing
<< dendl
;
1702 if (existing
->policy
.lossy
) {
1703 // disconnect from the Connection
1704 ldout(async_msgr
->cct
, 1) << __func__
<< " replacing on lossy channel, failing existing" << dendl
;
1706 existing
->dispatch_queue
->queue_reset(existing
.get());
1708 assert(can_write
== WriteStatus::NOWRITE
);
1709 existing
->write_lock
.lock();
1711 // reset the in_seq if this is a hard reset from peer,
1712 // otherwise we respect our original connection's value
1713 if (is_reset_from_peer
) {
1714 existing
->is_reset_from_peer
= true;
1717 center
->delete_file_event(cs
.fd(), EVENT_READABLE
|EVENT_WRITABLE
);
1719 if (existing
->delay_state
) {
1720 existing
->delay_state
->flush();
1721 assert(!delay_state
);
1723 existing
->reset_recv_state();
1725 auto temp_cs
= std::move(cs
);
1726 EventCenter
*new_center
= center
;
1727 Worker
*new_worker
= worker
;
1728 // avoid _stop shutdown replacing socket
1729 // queue a reset on the new connection, which we're dumping for the old
1732 dispatch_queue
->queue_reset(this);
1733 ldout(async_msgr
->cct
, 1) << __func__
<< " stop myself to swap existing" << dendl
;
1734 existing
->can_write
= WriteStatus::REPLACING
;
1735 existing
->replacing
= true;
1736 existing
->state_offset
= 0;
1737 // avoid previous thread modify event
1738 existing
->state
= STATE_NONE
;
1739 // Discard existing prefetch buffer in `recv_buf`
1740 existing
->recv_start
= existing
->recv_end
= 0;
1741 // there shouldn't exist any buffer
1742 assert(recv_start
== recv_end
);
1744 existing
->authorizer_challenge
.reset();
1746 auto deactivate_existing
= std::bind(
1747 [existing
, new_worker
, new_center
, connect
, reply
, authorizer_reply
](ConnectedSocket
&cs
) mutable {
1748 // we need to delete time event in original thread
1750 std::lock_guard
<std::mutex
> l(existing
->lock
);
1751 existing
->write_lock
.lock();
1752 existing
->requeue_sent();
1753 existing
->outcoming_bl
.clear();
1754 existing
->open_write
= false;
1755 existing
->write_lock
.unlock();
1756 if (existing
->state
== STATE_NONE
) {
1757 existing
->shutdown_socket();
1758 existing
->cs
= std::move(cs
);
1759 existing
->worker
->references
--;
1760 new_worker
->references
++;
1761 existing
->logger
= new_worker
->get_perf_counter();
1762 existing
->worker
= new_worker
;
1763 existing
->center
= new_center
;
1764 if (existing
->delay_state
)
1765 existing
->delay_state
->set_center(new_center
);
1766 } else if (existing
->state
== STATE_CLOSED
) {
1767 auto back_to_close
= std::bind(
1768 [](ConnectedSocket
&cs
) mutable { cs
.close(); }, std::move(cs
));
1769 new_center
->submit_to(
1770 new_center
->get_id(), std::move(back_to_close
), true);
1777 // Before changing existing->center, it may already exists some events in existing->center's queue.
1778 // Then if we mark down `existing`, it will execute in another thread and clean up connection.
1779 // Previous event will result in segment fault
1780 auto transfer_existing
= [existing
, connect
, reply
, authorizer_reply
]() mutable {
1781 std::lock_guard
<std::mutex
> l(existing
->lock
);
1782 if (existing
->state
== STATE_CLOSED
)
1784 assert(existing
->state
== STATE_NONE
);
1786 existing
->state
= STATE_ACCEPTING_WAIT_CONNECT_MSG
;
1787 existing
->center
->create_file_event(existing
->cs
.fd(), EVENT_READABLE
, existing
->read_handler
);
1788 reply
.global_seq
= existing
->peer_global_seq
;
1789 if (existing
->_reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
) < 0) {
1794 if (existing
->center
->in_thread())
1795 transfer_existing();
1797 existing
->center
->submit_to(
1798 existing
->center
->get_id(), std::move(transfer_existing
), true);
1799 }, std::move(temp_cs
));
1801 existing
->center
->submit_to(
1802 existing
->center
->get_id(), std::move(deactivate_existing
), true);
1803 existing
->write_lock
.unlock();
1804 existing
->lock
.unlock();
1807 existing
->lock
.unlock();
1810 connect_seq
= connect
.connect_seq
+ 1;
1811 peer_global_seq
= connect
.global_seq
;
1812 ldout(async_msgr
->cct
, 10) << __func__
<< " accept success, connect_seq = "
1813 << connect_seq
<< " in_seq=" << in_seq
<< ", sending READY" << dendl
;
1817 // if it is a hard reset from peer, we don't need a round-trip to negotiate in/out sequence
1818 if ((connect
.features
& CEPH_FEATURE_RECONNECT_SEQ
) && !is_reset_from_peer
) {
1819 reply
.tag
= CEPH_MSGR_TAG_SEQ
;
1820 next_state
= STATE_ACCEPTING_WAIT_SEQ
;
1822 reply
.tag
= CEPH_MSGR_TAG_READY
;
1823 next_state
= STATE_ACCEPTING_READY
;
1824 discard_requeued_up_to(0);
1825 is_reset_from_peer
= false;
1830 reply
.features
= policy
.features_supported
;
1831 reply
.global_seq
= async_msgr
->get_global_seq();
1832 reply
.connect_seq
= connect_seq
;
1834 reply
.authorizer_len
= authorizer_reply
.length();
1836 reply
.flags
= reply
.flags
| CEPH_MSG_CONNECT_LOSSY
;
1838 set_features((uint64_t)reply
.features
& (uint64_t)connect
.features
);
1839 ldout(async_msgr
->cct
, 10) << __func__
<< " accept features " << get_features() << dendl
;
1841 session_security
.reset(
1842 get_auth_session_handler(async_msgr
->cct
, connect
.authorizer_protocol
,
1843 session_key
, get_features()));
1845 reply_bl
.append((char*)&reply
, sizeof(reply
));
1847 if (reply
.authorizer_len
)
1848 reply_bl
.append(authorizer_reply
.c_str(), authorizer_reply
.length());
1850 if (reply
.tag
== CEPH_MSGR_TAG_SEQ
) {
1851 uint64_t s
= in_seq
;
1852 reply_bl
.append((char*)&s
, sizeof(s
));
1856 // Because "replacing" will prevent other connections preempt this addr,
1857 // it's safe that here we don't acquire Connection's lock
1858 r
= async_msgr
->accept_conn(this);
1865 ldout(async_msgr
->cct
, 1) << __func__
<< " existing race replacing process for addr=" << peer_addr
1866 << " just fail later one(this)" << dendl
;
1867 goto fail_registered
;
1869 if (state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
1870 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while accept_conn, it must be mark_down" << dendl
;
1871 assert(state
== STATE_CLOSED
|| state
== STATE_NONE
);
1872 goto fail_registered
;
1875 r
= try_send(reply_bl
);
1877 goto fail_registered
;
1880 dispatch_queue
->queue_accept(this);
1881 async_msgr
->ms_deliver_handle_fast_accept(this);
1886 ldout(async_msgr
->cct
, 2) << __func__
<< " accept write reply msg done" << dendl
;
1888 state
= STATE_WAIT_SEND
;
1889 state_after_send
= next_state
;
1895 ldout(async_msgr
->cct
, 10) << __func__
<< " accept fault after register" << dendl
;
1899 ldout(async_msgr
->cct
, 10) << __func__
<< " failed to accept." << dendl
;
1903 void AsyncConnection::_connect()
1905 ldout(async_msgr
->cct
, 10) << __func__
<< " csq=" << connect_seq
<< dendl
;
1907 state
= STATE_CONNECTING
;
1908 // rescheduler connection in order to avoid lock dep
1909 // may called by external thread(send_message)
1910 center
->dispatch_event_external(read_handler
);
1913 void AsyncConnection::accept(ConnectedSocket socket
, entity_addr_t
&addr
)
1915 ldout(async_msgr
->cct
, 10) << __func__
<< " sd=" << socket
.fd() << dendl
;
1916 assert(socket
.fd() >= 0);
1918 std::lock_guard
<std::mutex
> l(lock
);
1919 cs
= std::move(socket
);
1921 state
= STATE_ACCEPTING
;
1922 // rescheduler connection in order to avoid lock dep
1923 center
->dispatch_event_external(read_handler
);
1926 int AsyncConnection::send_message(Message
*m
)
1929 lgeneric_subdout(async_msgr
->cct
, ms
,
1930 1) << "-- " << async_msgr
->get_myaddr() << " --> "
1931 << get_peer_addr() << " -- "
1932 << *m
<< " -- " << m
<< " con "
1933 << m
->get_connection().get()
1936 // optimistic think it's ok to encode(actually may broken now)
1937 if (!m
->get_priority())
1938 m
->set_priority(async_msgr
->get_default_send_priority());
1940 m
->get_header().src
= async_msgr
->get_myname();
1941 m
->set_connection(this);
1943 if (m
->get_type() == CEPH_MSG_OSD_OP
)
1944 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OP_BEGIN", true);
1945 else if (m
->get_type() == CEPH_MSG_OSD_OPREPLY
)
1946 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OPREPLY_BEGIN", true);
1948 if (async_msgr
->get_myaddr() == get_peer_addr()) { //loopback connection
1949 ldout(async_msgr
->cct
, 20) << __func__
<< " " << *m
<< " local" << dendl
;
1950 std::lock_guard
<std::mutex
> l(write_lock
);
1951 if (can_write
!= WriteStatus::CLOSED
) {
1952 dispatch_queue
->local_delivery(m
, m
->get_priority());
1954 ldout(async_msgr
->cct
, 10) << __func__
<< " loopback connection closed."
1955 << " Drop message " << m
<< dendl
;
1961 last_active
= ceph::coarse_mono_clock::now();
1962 // we don't want to consider local message here, it's too lightweight which
1963 // may disturb users
1964 logger
->inc(l_msgr_send_messages
);
1967 uint64_t f
= get_features();
1969 // TODO: Currently not all messages supports reencode like MOSDMap, so here
1970 // only let fast dispatch support messages prepare message
1971 bool can_fast_prepare
= async_msgr
->ms_can_fast_dispatch(m
);
1972 if (can_fast_prepare
)
1973 prepare_send_message(f
, m
, bl
);
1975 std::lock_guard
<std::mutex
> l(write_lock
);
1976 // "features" changes will change the payload encoding
1977 if (can_fast_prepare
&& (can_write
== WriteStatus::NOWRITE
|| get_features() != f
)) {
1978 // ensure the correctness of message encoding
1980 m
->get_payload().clear();
1981 ldout(async_msgr
->cct
, 5) << __func__
<< " clear encoded buffer previous "
1982 << f
<< " != " << get_features() << dendl
;
1984 if (can_write
== WriteStatus::CLOSED
) {
1985 ldout(async_msgr
->cct
, 10) << __func__
<< " connection closed."
1986 << " Drop message " << m
<< dendl
;
1989 m
->trace
.event("async enqueueing message");
1990 out_q
[m
->get_priority()].emplace_back(std::move(bl
), m
);
1991 ldout(async_msgr
->cct
, 15) << __func__
<< " inline write is denied, reschedule m=" << m
<< dendl
;
1992 if (can_write
!= WriteStatus::REPLACING
)
1993 center
->dispatch_event_external(write_handler
);
1998 void AsyncConnection::requeue_sent()
2003 list
<pair
<bufferlist
, Message
*> >& rq
= out_q
[CEPH_MSG_PRIO_HIGHEST
];
2004 while (!sent
.empty()) {
2005 Message
* m
= sent
.back();
2007 ldout(async_msgr
->cct
, 10) << __func__
<< " " << *m
<< " for resend "
2008 << " (" << m
->get_seq() << ")" << dendl
;
2009 rq
.push_front(make_pair(bufferlist(), m
));
2014 void AsyncConnection::discard_requeued_up_to(uint64_t seq
)
2016 ldout(async_msgr
->cct
, 10) << __func__
<< " " << seq
<< dendl
;
2017 std::lock_guard
<std::mutex
> l(write_lock
);
2018 if (out_q
.count(CEPH_MSG_PRIO_HIGHEST
) == 0)
2020 list
<pair
<bufferlist
, Message
*> >& rq
= out_q
[CEPH_MSG_PRIO_HIGHEST
];
2021 while (!rq
.empty()) {
2022 pair
<bufferlist
, Message
*> p
= rq
.front();
2023 if (p
.second
->get_seq() == 0 || p
.second
->get_seq() > seq
)
2025 ldout(async_msgr
->cct
, 10) << __func__
<< " " << *(p
.second
) << " for resend seq " << p
.second
->get_seq()
2026 << " <= " << seq
<< ", discarding" << dendl
;
2032 out_q
.erase(CEPH_MSG_PRIO_HIGHEST
);
2036 * Tears down the AsyncConnection's message queues, and removes them from the DispatchQueue
2037 * Must hold write_lock prior to calling.
2039 void AsyncConnection::discard_out_queue()
2041 ldout(async_msgr
->cct
, 10) << __func__
<< " started" << dendl
;
2043 for (list
<Message
*>::iterator p
= sent
.begin(); p
!= sent
.end(); ++p
) {
2044 ldout(async_msgr
->cct
, 20) << __func__
<< " discard " << *p
<< dendl
;
2048 for (map
<int, list
<pair
<bufferlist
, Message
*> > >::iterator p
= out_q
.begin(); p
!= out_q
.end(); ++p
)
2049 for (list
<pair
<bufferlist
, Message
*> >::iterator r
= p
->second
.begin(); r
!= p
->second
.end(); ++r
) {
2050 ldout(async_msgr
->cct
, 20) << __func__
<< " discard " << r
->second
<< dendl
;
2056 int AsyncConnection::randomize_out_seq()
2058 if (get_features() & CEPH_FEATURE_MSG_AUTH
) {
2059 // Set out_seq to a random value, so CRC won't be predictable. Don't bother checking seq_error
2060 // here. We'll check it on the call. PLR
2062 int seq_error
= get_random_bytes((char *)&rand_seq
, sizeof(rand_seq
));
2063 rand_seq
&= SEQ_MASK
;
2064 lsubdout(async_msgr
->cct
, ms
, 10) << __func__
<< " randomize_out_seq " << rand_seq
<< dendl
;
2068 // previously, seq #'s always started at 0.
2074 void AsyncConnection::fault()
2076 if (state
== STATE_CLOSED
|| state
== STATE_NONE
) {
2077 ldout(async_msgr
->cct
, 10) << __func__
<< " connection is already closed" << dendl
;
2081 if (policy
.lossy
&& !(state
>= STATE_CONNECTING
&& state
< STATE_CONNECTING_READY
)) {
2082 ldout(async_msgr
->cct
, 1) << __func__
<< " on lossy channel, failing" << dendl
;
2084 dispatch_queue
->queue_reset(this);
2089 can_write
= WriteStatus::NOWRITE
;
2093 // queue delayed items immediately
2095 delay_state
->flush();
2096 // requeue sent items
2098 recv_start
= recv_end
= 0;
2101 is_reset_from_peer
= false;
2102 outcoming_bl
.clear();
2103 if (!once_ready
&& !is_queued() &&
2104 state
>=STATE_ACCEPTING
&& state
<= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
2105 ldout(async_msgr
->cct
, 10) << __func__
<< " with nothing to send and in the half "
2106 << " accept state just closed" << dendl
;
2107 write_lock
.unlock();
2109 dispatch_queue
->queue_reset(this);
2113 if (policy
.standby
&& !is_queued() && state
!= STATE_WAIT
) {
2114 ldout(async_msgr
->cct
, 10) << __func__
<< " with nothing to send, going to standby" << dendl
;
2115 state
= STATE_STANDBY
;
2116 write_lock
.unlock();
2120 write_lock
.unlock();
2121 if (!(state
>= STATE_CONNECTING
&& state
< STATE_CONNECTING_READY
) &&
2122 state
!= STATE_WAIT
) { // STATE_WAIT is coming from STATE_CONNECTING_*
2123 // policy maybe empty when state is in accept
2124 if (policy
.server
) {
2125 ldout(async_msgr
->cct
, 0) << __func__
<< " server, going to standby" << dendl
;
2126 state
= STATE_STANDBY
;
2128 ldout(async_msgr
->cct
, 0) << __func__
<< " initiating reconnect" << dendl
;
2130 state
= STATE_CONNECTING
;
2132 backoff
= utime_t();
2133 center
->dispatch_event_external(read_handler
);
2135 if (state
== STATE_WAIT
) {
2136 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_max_backoff
);
2137 } else if (backoff
== utime_t()) {
2138 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_initial_backoff
);
2141 if (backoff
> async_msgr
->cct
->_conf
->ms_max_backoff
)
2142 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_max_backoff
);
2145 state
= STATE_CONNECTING
;
2146 ldout(async_msgr
->cct
, 10) << __func__
<< " waiting " << backoff
<< dendl
;
2148 register_time_events
.insert(center
->create_time_event(
2149 backoff
.to_nsec()/1000, wakeup_handler
));
2153 void AsyncConnection::was_session_reset()
2155 ldout(async_msgr
->cct
,10) << __func__
<< " started" << dendl
;
2156 std::lock_guard
<std::mutex
> l(write_lock
);
2158 delay_state
->discard();
2159 dispatch_queue
->discard_queue(conn_id
);
2160 discard_out_queue();
2161 // note: we need to clear outcoming_bl here, but was_session_reset may be
2162 // called by other thread, so let caller clear this itself!
2163 // outcoming_bl.clear();
2165 dispatch_queue
->queue_remote_reset(this);
2167 if (randomize_out_seq()) {
2168 ldout(async_msgr
->cct
, 15) << __func__
<< " could not get random bytes to set seq number for session reset; set seq number to " << out_seq
<< dendl
;
2173 // it's safe to directly set 0, double locked
2176 can_write
= WriteStatus::NOWRITE
;
2179 void AsyncConnection::_stop()
2181 if (state
== STATE_CLOSED
)
2185 delay_state
->flush();
2187 ldout(async_msgr
->cct
, 2) << __func__
<< dendl
;
2188 std::lock_guard
<std::mutex
> l(write_lock
);
2191 dispatch_queue
->discard_queue(conn_id
);
2192 discard_out_queue();
2193 async_msgr
->unregister_conn(this);
2194 worker
->release_worker();
2196 state
= STATE_CLOSED
;
2198 can_write
= WriteStatus::CLOSED
;
2200 // Make sure in-queue events will been processed
2201 center
->dispatch_event_external(EventCallbackRef(new C_clean_handler(this)));
2204 void AsyncConnection::prepare_send_message(uint64_t features
, Message
*m
, bufferlist
&bl
)
2206 ldout(async_msgr
->cct
, 20) << __func__
<< " m" << " " << *m
<< dendl
;
2208 // associate message with Connection (for benefit of encode_payload)
2209 if (m
->empty_payload())
2210 ldout(async_msgr
->cct
, 20) << __func__
<< " encoding features "
2211 << features
<< " " << m
<< " " << *m
<< dendl
;
2213 ldout(async_msgr
->cct
, 20) << __func__
<< " half-reencoding features "
2214 << features
<< " " << m
<< " " << *m
<< dendl
;
2216 // encode and copy out of *m
2217 m
->encode(features
, msgr
->crcflags
);
2219 bl
.append(m
->get_payload());
2220 bl
.append(m
->get_middle());
2221 bl
.append(m
->get_data());
2224 ssize_t
AsyncConnection::write_message(Message
*m
, bufferlist
& bl
, bool more
)
2227 assert(center
->in_thread());
2228 m
->set_seq(++out_seq
);
2230 if (msgr
->crcflags
& MSG_CRC_HEADER
)
2231 m
->calc_header_crc();
2233 ceph_msg_header
& header
= m
->get_header();
2234 ceph_msg_footer
& footer
= m
->get_footer();
2236 // TODO: let sign_message could be reentry?
2237 // Now that we have all the crcs calculated, handle the
2238 // digital signature for the message, if the AsyncConnection has session
2239 // security set up. Some session security options do not
2240 // actually calculate and check the signature, but they should
2241 // handle the calls to sign_message and check_signature. PLR
2242 if (session_security
.get() == NULL
) {
2243 ldout(async_msgr
->cct
, 20) << __func__
<< " no session security" << dendl
;
2245 if (session_security
->sign_message(m
)) {
2246 ldout(async_msgr
->cct
, 20) << __func__
<< " failed to sign m="
2247 << m
<< "): sig = " << footer
.sig
<< dendl
;
2249 ldout(async_msgr
->cct
, 20) << __func__
<< " signed m=" << m
2250 << "): sig = " << footer
.sig
<< dendl
;
2254 unsigned original_bl_len
= outcoming_bl
.length();
2256 outcoming_bl
.append(CEPH_MSGR_TAG_MSG
);
2258 if (has_feature(CEPH_FEATURE_NOSRCADDR
)) {
2259 outcoming_bl
.append((char*)&header
, sizeof(header
));
2261 ceph_msg_header_old oldheader
;
2262 memcpy(&oldheader
, &header
, sizeof(header
));
2263 oldheader
.src
.name
= header
.src
;
2264 oldheader
.src
.addr
= get_peer_addr();
2265 oldheader
.orig_src
= oldheader
.src
;
2266 oldheader
.reserved
= header
.reserved
;
2267 oldheader
.crc
= ceph_crc32c(0, (unsigned char*)&oldheader
,
2268 sizeof(oldheader
) - sizeof(oldheader
.crc
));
2269 outcoming_bl
.append((char*)&oldheader
, sizeof(oldheader
));
2272 ldout(async_msgr
->cct
, 20) << __func__
<< " sending message type=" << header
.type
2273 << " src " << entity_name_t(header
.src
)
2274 << " front=" << header
.front_len
2275 << " data=" << header
.data_len
2276 << " off " << header
.data_off
<< dendl
;
2278 if ((bl
.length() <= ASYNC_COALESCE_THRESHOLD
) && (bl
.buffers().size() > 1)) {
2279 for (const auto &pb
: bl
.buffers()) {
2280 outcoming_bl
.append((char*)pb
.c_str(), pb
.length());
2283 outcoming_bl
.claim_append(bl
);
2286 // send footer; if receiver doesn't support signatures, use the old footer format
2287 ceph_msg_footer_old old_footer
;
2288 if (has_feature(CEPH_FEATURE_MSG_AUTH
)) {
2289 outcoming_bl
.append((char*)&footer
, sizeof(footer
));
2291 if (msgr
->crcflags
& MSG_CRC_HEADER
) {
2292 old_footer
.front_crc
= footer
.front_crc
;
2293 old_footer
.middle_crc
= footer
.middle_crc
;
2294 old_footer
.data_crc
= footer
.data_crc
;
2296 old_footer
.front_crc
= old_footer
.middle_crc
= 0;
2298 old_footer
.data_crc
= msgr
->crcflags
& MSG_CRC_DATA
? footer
.data_crc
: 0;
2299 old_footer
.flags
= footer
.flags
;
2300 outcoming_bl
.append((char*)&old_footer
, sizeof(old_footer
));
2303 m
->trace
.event("async writing message");
2304 ldout(async_msgr
->cct
, 20) << __func__
<< " sending " << m
->get_seq()
2305 << " " << m
<< dendl
;
2306 ssize_t total_send_size
= outcoming_bl
.length();
2307 ssize_t rc
= _try_send(more
);
2309 ldout(async_msgr
->cct
, 1) << __func__
<< " error sending " << m
<< ", "
2310 << cpp_strerror(rc
) << dendl
;
2311 } else if (rc
== 0) {
2312 logger
->inc(l_msgr_send_bytes
, total_send_size
- original_bl_len
);
2313 ldout(async_msgr
->cct
, 10) << __func__
<< " sending " << m
<< " done." << dendl
;
2315 logger
->inc(l_msgr_send_bytes
, total_send_size
- outcoming_bl
.length());
2316 ldout(async_msgr
->cct
, 10) << __func__
<< " sending " << m
<< " continuely." << dendl
;
2318 if (m
->get_type() == CEPH_MSG_OSD_OP
)
2319 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OP_END", false);
2320 else if (m
->get_type() == CEPH_MSG_OSD_OPREPLY
)
2321 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OPREPLY_END", false);
2327 void AsyncConnection::reset_recv_state()
2329 // clean up state internal variables and states
2330 if (state
>= STATE_CONNECTING_SEND_CONNECT_MSG
&&
2331 state
<= STATE_CONNECTING_READY
) {
2334 got_bad_auth
= false;
2337 if (state
> STATE_OPEN_MESSAGE_THROTTLE_MESSAGE
&&
2338 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
2339 && policy
.throttler_messages
) {
2340 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << 1
2341 << " message to policy throttler "
2342 << policy
.throttler_messages
->get_current() << "/"
2343 << policy
.throttler_messages
->get_max() << dendl
;
2344 policy
.throttler_messages
->put();
2346 if (state
> STATE_OPEN_MESSAGE_THROTTLE_BYTES
&&
2347 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
) {
2348 if (policy
.throttler_bytes
) {
2349 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << cur_msg_size
2350 << " bytes to policy throttler "
2351 << policy
.throttler_bytes
->get_current() << "/"
2352 << policy
.throttler_bytes
->get_max() << dendl
;
2353 policy
.throttler_bytes
->put(cur_msg_size
);
2356 if (state
> STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE
&&
2357 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
) {
2358 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << cur_msg_size
2359 << " bytes to dispatch_queue throttler "
2360 << dispatch_queue
->dispatch_throttler
.get_current() << "/"
2361 << dispatch_queue
->dispatch_throttler
.get_max() << dendl
;
2362 dispatch_queue
->dispatch_throttle_release(cur_msg_size
);
2366 void AsyncConnection::handle_ack(uint64_t seq
)
2368 ldout(async_msgr
->cct
, 15) << __func__
<< " got ack seq " << seq
<< dendl
;
2370 std::lock_guard
<std::mutex
> l(write_lock
);
2371 while (!sent
.empty() && sent
.front()->get_seq() <= seq
) {
2372 Message
* m
= sent
.front();
2374 ldout(async_msgr
->cct
, 10) << __func__
<< " got ack seq "
2375 << seq
<< " >= " << m
->get_seq() << " on "
2376 << m
<< " " << *m
<< dendl
;
2381 void AsyncConnection::DelayedDelivery::do_request(int id
)
2383 Message
*m
= nullptr;
2385 std::lock_guard
<std::mutex
> l(delay_lock
);
2386 register_time_events
.erase(id
);
2389 if (delay_queue
.empty())
2391 utime_t release
= delay_queue
.front().first
;
2392 m
= delay_queue
.front().second
;
2393 string delay_msg_type
= msgr
->cct
->_conf
->ms_inject_delay_msg_type
;
2394 utime_t now
= ceph_clock_now();
2395 if ((release
> now
&&
2396 (delay_msg_type
.empty() || m
->get_type_name() == delay_msg_type
))) {
2397 utime_t t
= release
- now
;
2400 delay_queue
.pop_front();
2402 if (msgr
->ms_can_fast_dispatch(m
)) {
2403 dispatch_queue
->fast_dispatch(m
);
2405 dispatch_queue
->enqueue(m
, m
->get_priority(), conn_id
);
2409 void AsyncConnection::DelayedDelivery::flush() {
2410 stop_dispatch
= true;
2412 center
->get_id(), [this] () mutable {
2413 std::lock_guard
<std::mutex
> l(delay_lock
);
2414 while (!delay_queue
.empty()) {
2415 Message
*m
= delay_queue
.front().second
;
2416 if (msgr
->ms_can_fast_dispatch(m
)) {
2417 dispatch_queue
->fast_dispatch(m
);
2419 dispatch_queue
->enqueue(m
, m
->get_priority(), conn_id
);
2421 delay_queue
.pop_front();
2423 for (auto i
: register_time_events
)
2424 center
->delete_time_event(i
);
2425 register_time_events
.clear();
2426 stop_dispatch
= false;
2430 void AsyncConnection::send_keepalive()
2432 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2433 std::lock_guard
<std::mutex
> l(write_lock
);
2434 if (can_write
!= WriteStatus::CLOSED
) {
2436 center
->dispatch_event_external(write_handler
);
2440 void AsyncConnection::mark_down()
2442 ldout(async_msgr
->cct
, 1) << __func__
<< dendl
;
2443 std::lock_guard
<std::mutex
> l(lock
);
2447 void AsyncConnection::_append_keepalive_or_ack(bool ack
, utime_t
*tp
)
2449 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2452 struct ceph_timespec ts
;
2453 tp
->encode_timeval(&ts
);
2454 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE2_ACK
);
2455 outcoming_bl
.append((char*)&ts
, sizeof(ts
));
2456 } else if (has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2
)) {
2457 struct ceph_timespec ts
;
2458 utime_t t
= ceph_clock_now();
2459 t
.encode_timeval(&ts
);
2460 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE2
);
2461 outcoming_bl
.append((char*)&ts
, sizeof(ts
));
2463 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE
);
2467 void AsyncConnection::handle_write()
2469 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2473 if (can_write
== WriteStatus::CANWRITE
) {
2475 _append_keepalive_or_ack();
2479 auto start
= ceph::mono_clock::now();
2483 Message
*m
= _get_next_outgoing(&data
);
2487 if (!policy
.lossy
) {
2492 more
= _has_next_outgoing();
2493 write_lock
.unlock();
2495 // send_message or requeue messages may not encode message
2497 prepare_send_message(get_features(), m
, data
);
2499 r
= write_message(m
, data
, more
);
2501 ldout(async_msgr
->cct
, 1) << __func__
<< " send msg failed" << dendl
;
2507 } while (can_write
== WriteStatus::CANWRITE
);
2508 write_lock
.unlock();
2510 uint64_t left
= ack_left
;
2514 outcoming_bl
.append(CEPH_MSGR_TAG_ACK
);
2515 outcoming_bl
.append((char*)&s
, sizeof(s
));
2516 ldout(async_msgr
->cct
, 10) << __func__
<< " try send msg ack, acked " << left
<< " messages" << dendl
;
2519 r
= _try_send(left
);
2520 } else if (is_queued()) {
2524 logger
->tinc(l_msgr_running_send_time
, ceph::mono_clock::now() - start
);
2526 ldout(async_msgr
->cct
, 1) << __func__
<< " send msg failed" << dendl
;
2530 write_lock
.unlock();
2533 if (state
== STATE_STANDBY
&& !policy
.server
&& is_queued()) {
2534 ldout(async_msgr
->cct
, 10) << __func__
<< " policy.server is false" << dendl
;
2536 } else if (cs
&& state
!= STATE_NONE
&& state
!= STATE_CONNECTING
&& state
!= STATE_CONNECTING_RE
&& state
!= STATE_CLOSED
) {
2539 ldout(async_msgr
->cct
, 1) << __func__
<< " send outcoming bl failed" << dendl
;
2540 write_lock
.unlock();
2546 write_lock
.unlock();
2558 void AsyncConnection::wakeup_from(uint64_t id
)
2561 register_time_events
.erase(id
);
2566 void AsyncConnection::tick(uint64_t id
)
2568 auto now
= ceph::coarse_mono_clock::now();
2569 ldout(async_msgr
->cct
, 20) << __func__
<< " last_id=" << last_tick_id
2570 << " last_active" << last_active
<< dendl
;
2571 std::lock_guard
<std::mutex
> l(lock
);
2573 auto idle_period
= std::chrono::duration_cast
<std::chrono::microseconds
>(now
- last_active
).count();
2574 if (inactive_timeout_us
< (uint64_t)idle_period
) {
2575 ldout(async_msgr
->cct
, 1) << __func__
<< " idle(" << idle_period
<< ") more than "
2576 << inactive_timeout_us
2577 << " us, mark self fault." << dendl
;
2579 } else if (is_connected()) {
2580 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);