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
);
801 // clean up local buffer references
810 case STATE_OPEN_TAG_CLOSE
:
812 ldout(async_msgr
->cct
, 20) << __func__
<< " got CLOSE" << dendl
;
819 ldout(async_msgr
->cct
, 20) << __func__
<< " enter STANDY" << dendl
;
826 ldout(async_msgr
->cct
, 20) << __func__
<< " enter none state" << dendl
;
832 ldout(async_msgr
->cct
, 20) << __func__
<< " socket closed" << dendl
;
838 ldout(async_msgr
->cct
, 1) << __func__
<< " enter wait state, failing" << dendl
;
844 if (_process_connection() < 0)
849 } while (prev_state
!= state
);
851 if (need_dispatch_writer
&& is_connected())
852 center
->dispatch_event_external(write_handler
);
854 logger
->tinc(l_msgr_running_recv_time
, ceph::mono_clock::now() - recv_start_time
);
861 ssize_t
AsyncConnection::_process_connection()
866 case STATE_WAIT_SEND
:
868 std::lock_guard
<std::mutex
> l(write_lock
);
869 if (!outcoming_bl
.length()) {
870 assert(state_after_send
);
871 state
= state_after_send
;
872 state_after_send
= STATE_NONE
;
877 case STATE_CONNECTING
:
879 assert(!policy
.server
);
881 // reset connect state variables
882 got_bad_auth
= false;
885 authorizer_buf
.clear();
886 memset(&connect_msg
, 0, sizeof(connect_msg
));
887 memset(&connect_reply
, 0, sizeof(connect_reply
));
889 global_seq
= async_msgr
->get_global_seq();
890 // close old socket. this is safe because we stopped the reader thread above.
892 center
->delete_file_event(cs
.fd(), EVENT_READABLE
|EVENT_WRITABLE
);
897 opts
.priority
= async_msgr
->get_socket_priority();
898 opts
.connect_bind_addr
= msgr
->get_myaddr();
899 r
= worker
->connect(get_peer_addr(), opts
, &cs
);
903 center
->create_file_event(cs
.fd(), EVENT_READABLE
, read_handler
);
904 state
= STATE_CONNECTING_RE
;
908 case STATE_CONNECTING_RE
:
910 r
= cs
.is_connected();
912 ldout(async_msgr
->cct
, 1) << __func__
<< " reconnect failed " << dendl
;
913 if (r
== -ECONNREFUSED
) {
914 ldout(async_msgr
->cct
, 2) << __func__
<< " connection refused!" << dendl
;
915 dispatch_queue
->queue_refused(this);
919 ldout(async_msgr
->cct
, 10) << __func__
<< " nonblock connect inprogress" << dendl
;
920 if (async_msgr
->get_stack()->nonblock_connect_need_writable_event())
921 center
->create_file_event(cs
.fd(), EVENT_WRITABLE
, read_handler
);
925 center
->delete_file_event(cs
.fd(), EVENT_WRITABLE
);
926 ldout(async_msgr
->cct
, 10) << __func__
<< " connect successfully, ready to send banner" << dendl
;
929 bl
.append(CEPH_BANNER
, strlen(CEPH_BANNER
));
932 state
= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
;
933 ldout(async_msgr
->cct
, 10) << __func__
<< " connect write banner done: "
934 << get_peer_addr() << dendl
;
936 state
= STATE_WAIT_SEND
;
937 state_after_send
= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
;
938 ldout(async_msgr
->cct
, 10) << __func__
<< " connect wait for write banner: "
939 << get_peer_addr() << dendl
;
947 case STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
:
949 entity_addr_t paddr
, peer_addr_for_me
;
951 unsigned banner_len
= strlen(CEPH_BANNER
);
952 unsigned need_len
= banner_len
+ sizeof(ceph_entity_addr
)*2;
953 r
= read_until(need_len
, state_buffer
);
955 ldout(async_msgr
->cct
, 1) << __func__
<< " read banner and identify addresses failed" << dendl
;
961 if (memcmp(state_buffer
, CEPH_BANNER
, banner_len
)) {
962 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol error (bad banner) on peer "
963 << get_peer_addr() << dendl
;
968 bl
.append(state_buffer
+banner_len
, sizeof(ceph_entity_addr
)*2);
969 bufferlist::iterator p
= bl
.begin();
972 ::decode(peer_addr_for_me
, p
);
973 } catch (const buffer::error
& e
) {
974 lderr(async_msgr
->cct
) << __func__
<< " decode peer addr failed " << dendl
;
977 ldout(async_msgr
->cct
, 20) << __func__
<< " connect read peer addr "
978 << paddr
<< " on socket " << cs
.fd() << dendl
;
979 if (peer_addr
!= paddr
) {
980 if (paddr
.is_blank_ip() && peer_addr
.get_port() == paddr
.get_port() &&
981 peer_addr
.get_nonce() == paddr
.get_nonce()) {
982 ldout(async_msgr
->cct
, 0) << __func__
<< " connect claims to be " << paddr
983 << " not " << peer_addr
984 << " - presumably this is the same node!" << dendl
;
986 ldout(async_msgr
->cct
, 10) << __func__
<< " connect claims to be "
987 << paddr
<< " not " << peer_addr
<< dendl
;
992 ldout(async_msgr
->cct
, 20) << __func__
<< " connect peer addr for me is " << peer_addr_for_me
<< dendl
;
994 async_msgr
->learned_addr(peer_addr_for_me
);
995 if (async_msgr
->cct
->_conf
->ms_inject_internal_delays
996 && async_msgr
->cct
->_conf
->ms_inject_socket_failures
) {
997 if (rand() % async_msgr
->cct
->_conf
->ms_inject_socket_failures
== 0) {
998 ldout(msgr
->cct
, 10) << __func__
<< " sleep for "
999 << async_msgr
->cct
->_conf
->ms_inject_internal_delays
<< dendl
;
1001 t
.set_from_double(async_msgr
->cct
->_conf
->ms_inject_internal_delays
);
1007 if (state
!= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
) {
1008 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while learned_addr, mark_down or "
1009 << " replacing must be happened just now" << dendl
;
1013 ::encode(async_msgr
->get_myaddr(), myaddrbl
, 0); // legacy
1014 r
= try_send(myaddrbl
);
1016 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1017 ldout(async_msgr
->cct
, 10) << __func__
<< " connect sent my addr "
1018 << async_msgr
->get_myaddr() << dendl
;
1020 state
= STATE_WAIT_SEND
;
1021 state_after_send
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1022 ldout(async_msgr
->cct
, 10) << __func__
<< " connect send my addr done: "
1023 << async_msgr
->get_myaddr() << dendl
;
1025 ldout(async_msgr
->cct
, 2) << __func__
<< " connect couldn't write my addr, "
1026 << cpp_strerror(r
) << dendl
;
1033 case STATE_CONNECTING_SEND_CONNECT_MSG
:
1036 authorizer
= async_msgr
->get_authorizer(peer_type
, false);
1040 connect_msg
.features
= policy
.features_supported
;
1041 connect_msg
.host_type
= async_msgr
->get_myinst().name
.type();
1042 connect_msg
.global_seq
= global_seq
;
1043 connect_msg
.connect_seq
= connect_seq
;
1044 connect_msg
.protocol_version
= async_msgr
->get_proto_version(peer_type
, true);
1045 connect_msg
.authorizer_protocol
= authorizer
? authorizer
->protocol
: 0;
1046 connect_msg
.authorizer_len
= authorizer
? authorizer
->bl
.length() : 0;
1048 ldout(async_msgr
->cct
, 10) << __func__
<< " connect_msg.authorizer_len="
1049 << connect_msg
.authorizer_len
<< " protocol="
1050 << connect_msg
.authorizer_protocol
<< dendl
;
1051 connect_msg
.flags
= 0;
1053 connect_msg
.flags
|= CEPH_MSG_CONNECT_LOSSY
; // this is fyi, actually, server decides!
1054 bl
.append((char*)&connect_msg
, sizeof(connect_msg
));
1056 bl
.append(authorizer
->bl
.c_str(), authorizer
->bl
.length());
1058 ldout(async_msgr
->cct
, 10) << __func__
<< " connect sending gseq=" << global_seq
<< " cseq="
1059 << connect_seq
<< " proto=" << connect_msg
.protocol_version
<< dendl
;
1063 state
= STATE_CONNECTING_WAIT_CONNECT_REPLY
;
1064 ldout(async_msgr
->cct
,20) << __func__
<< " connect wrote (self +) cseq, waiting for reply" << dendl
;
1066 state
= STATE_WAIT_SEND
;
1067 state_after_send
= STATE_CONNECTING_WAIT_CONNECT_REPLY
;
1068 ldout(async_msgr
->cct
, 10) << __func__
<< " continue send reply " << dendl
;
1070 ldout(async_msgr
->cct
, 2) << __func__
<< " connect couldn't send reply "
1071 << cpp_strerror(r
) << dendl
;
1078 case STATE_CONNECTING_WAIT_CONNECT_REPLY
:
1080 r
= read_until(sizeof(connect_reply
), state_buffer
);
1082 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect reply failed" << dendl
;
1088 connect_reply
= *((ceph_msg_connect_reply
*)state_buffer
);
1090 ldout(async_msgr
->cct
, 20) << __func__
<< " connect got reply tag " << (int)connect_reply
.tag
1091 << " connect_seq " << connect_reply
.connect_seq
<< " global_seq "
1092 << connect_reply
.global_seq
<< " proto " << connect_reply
.protocol_version
1093 << " flags " << (int)connect_reply
.flags
<< " features "
1094 << connect_reply
.features
<< dendl
;
1095 state
= STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
;
1100 case STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
:
1102 bufferlist authorizer_reply
;
1103 if (connect_reply
.authorizer_len
) {
1104 ldout(async_msgr
->cct
, 10) << __func__
<< " reply.authorizer_len=" << connect_reply
.authorizer_len
<< dendl
;
1105 assert(connect_reply
.authorizer_len
< 4096);
1106 r
= read_until(connect_reply
.authorizer_len
, state_buffer
);
1108 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect reply authorizer failed" << dendl
;
1114 authorizer_reply
.append(state_buffer
, connect_reply
.authorizer_len
);
1116 if (connect_reply
.tag
== CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER
) {
1117 ldout(async_msgr
->cct
,10) << __func__
<< " connect got auth challenge" << dendl
;
1118 authorizer
->add_challenge(async_msgr
->cct
, authorizer_reply
);
1119 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1123 auto iter
= authorizer_reply
.begin();
1124 if (authorizer
&& !authorizer
->verify_reply(iter
)) {
1125 ldout(async_msgr
->cct
, 0) << __func__
<< " failed verifying authorize reply" << dendl
;
1129 r
= handle_connect_reply(connect_msg
, connect_reply
);
1133 // state must be changed!
1134 assert(state
!= STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
);
1138 case STATE_CONNECTING_WAIT_ACK_SEQ
:
1140 uint64_t newly_acked_seq
= 0;
1142 r
= read_until(sizeof(newly_acked_seq
), state_buffer
);
1144 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect ack seq failed" << dendl
;
1150 newly_acked_seq
= *((uint64_t*)state_buffer
);
1151 ldout(async_msgr
->cct
, 2) << __func__
<< " got newly_acked_seq " << newly_acked_seq
1152 << " vs out_seq " << out_seq
<< dendl
;
1153 discard_requeued_up_to(newly_acked_seq
);
1154 //while (newly_acked_seq > out_seq.read()) {
1155 // Message *m = _get_next_outgoing(NULL);
1157 // ldout(async_msgr->cct, 2) << __func__ << " discarding previously sent " << m->get_seq()
1158 // << " " << *m << dendl;
1159 // assert(m->get_seq() <= newly_acked_seq);
1165 uint64_t s
= in_seq
;
1166 bl
.append((char*)&s
, sizeof(s
));
1169 state
= STATE_CONNECTING_READY
;
1170 ldout(async_msgr
->cct
, 10) << __func__
<< " send in_seq done " << dendl
;
1172 state_after_send
= STATE_CONNECTING_READY
;
1173 state
= STATE_WAIT_SEND
;
1174 ldout(async_msgr
->cct
, 10) << __func__
<< " continue send in_seq " << dendl
;
1181 case STATE_CONNECTING_READY
:
1184 peer_global_seq
= connect_reply
.global_seq
;
1185 policy
.lossy
= connect_reply
.flags
& CEPH_MSG_CONNECT_LOSSY
;
1189 assert(connect_seq
== connect_reply
.connect_seq
);
1190 backoff
= utime_t();
1191 set_features((uint64_t)connect_reply
.features
& (uint64_t)connect_msg
.features
);
1192 ldout(async_msgr
->cct
, 10) << __func__
<< " connect success " << connect_seq
1193 << ", lossy = " << policy
.lossy
<< ", features "
1194 << get_features() << dendl
;
1196 // If we have an authorizer, get a new AuthSessionHandler to deal with ongoing security of the
1198 if (authorizer
!= NULL
) {
1199 session_security
.reset(
1200 get_auth_session_handler(async_msgr
->cct
,
1201 authorizer
->protocol
,
1202 authorizer
->session_key
,
1205 // We have no authorizer, so we shouldn't be applying security to messages in this AsyncConnection. PLR
1206 session_security
.reset();
1210 assert(delay_state
->ready());
1211 dispatch_queue
->queue_connect(this);
1212 async_msgr
->ms_deliver_handle_fast_connect(this);
1214 // make sure no pending tick timer
1216 center
->delete_time_event(last_tick_id
);
1217 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);
1219 // message may in queue between last _try_send and connection ready
1220 // write event may already notify and we need to force scheduler again
1222 can_write
= WriteStatus::CANWRITE
;
1224 center
->dispatch_event_external(write_handler
);
1225 write_lock
.unlock();
1226 maybe_start_delay_thread();
1230 case STATE_ACCEPTING
:
1233 center
->create_file_event(cs
.fd(), EVENT_READABLE
, read_handler
);
1235 bl
.append(CEPH_BANNER
, strlen(CEPH_BANNER
));
1237 ::encode(async_msgr
->get_myaddr(), bl
, 0); // legacy
1238 port
= async_msgr
->get_myaddr().get_port();
1239 ::encode(socket_addr
, bl
, 0); // legacy
1240 ldout(async_msgr
->cct
, 1) << __func__
<< " sd=" << cs
.fd() << " " << socket_addr
<< dendl
;
1244 state
= STATE_ACCEPTING_WAIT_BANNER_ADDR
;
1245 ldout(async_msgr
->cct
, 10) << __func__
<< " write banner and addr done: "
1246 << get_peer_addr() << dendl
;
1248 state
= STATE_WAIT_SEND
;
1249 state_after_send
= STATE_ACCEPTING_WAIT_BANNER_ADDR
;
1250 ldout(async_msgr
->cct
, 10) << __func__
<< " wait for write banner and addr: "
1251 << get_peer_addr() << dendl
;
1258 case STATE_ACCEPTING_WAIT_BANNER_ADDR
:
1261 entity_addr_t peer_addr
;
1263 r
= read_until(strlen(CEPH_BANNER
) + sizeof(ceph_entity_addr
), state_buffer
);
1265 ldout(async_msgr
->cct
, 1) << __func__
<< " read peer banner and addr failed" << dendl
;
1271 if (memcmp(state_buffer
, CEPH_BANNER
, strlen(CEPH_BANNER
))) {
1272 ldout(async_msgr
->cct
, 1) << __func__
<< " accept peer sent bad banner '" << state_buffer
1273 << "' (should be '" << CEPH_BANNER
<< "')" << dendl
;
1277 addr_bl
.append(state_buffer
+strlen(CEPH_BANNER
), sizeof(ceph_entity_addr
));
1279 bufferlist::iterator ti
= addr_bl
.begin();
1280 ::decode(peer_addr
, ti
);
1283 ldout(async_msgr
->cct
, 10) << __func__
<< " accept peer addr is " << peer_addr
<< dendl
;
1284 if (peer_addr
.is_blank_ip()) {
1285 // peer apparently doesn't know what ip they have; figure it out for them.
1286 int port
= peer_addr
.get_port();
1287 peer_addr
.u
= socket_addr
.u
;
1288 peer_addr
.set_port(port
);
1289 ldout(async_msgr
->cct
, 0) << __func__
<< " accept peer addr is really " << peer_addr
1290 << " (socket is " << socket_addr
<< ")" << dendl
;
1292 set_peer_addr(peer_addr
); // so that connection_state gets set up
1293 state
= STATE_ACCEPTING_WAIT_CONNECT_MSG
;
1297 case STATE_ACCEPTING_WAIT_CONNECT_MSG
:
1299 r
= read_until(sizeof(connect_msg
), state_buffer
);
1301 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect msg failed" << dendl
;
1307 connect_msg
= *((ceph_msg_connect
*)state_buffer
);
1308 state
= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
;
1312 case STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
:
1314 bufferlist authorizer_reply
;
1316 if (connect_msg
.authorizer_len
) {
1317 if (!authorizer_buf
.length())
1318 authorizer_buf
.push_back(buffer::create(connect_msg
.authorizer_len
));
1320 r
= read_until(connect_msg
.authorizer_len
, authorizer_buf
.c_str());
1322 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect authorizer failed" << dendl
;
1329 ldout(async_msgr
->cct
, 20) << __func__
<< " accept got peer connect_seq "
1330 << connect_msg
.connect_seq
<< " global_seq "
1331 << connect_msg
.global_seq
<< dendl
;
1332 set_peer_type(connect_msg
.host_type
);
1333 policy
= async_msgr
->get_policy(connect_msg
.host_type
);
1334 ldout(async_msgr
->cct
, 10) << __func__
<< " accept of host_type " << connect_msg
.host_type
1335 << ", policy.lossy=" << policy
.lossy
<< " policy.server="
1336 << policy
.server
<< " policy.standby=" << policy
.standby
1337 << " policy.resetcheck=" << policy
.resetcheck
<< dendl
;
1339 r
= handle_connect_msg(connect_msg
, authorizer_buf
, authorizer_reply
);
1343 // state is changed by "handle_connect_msg"
1344 assert(state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
);
1348 case STATE_ACCEPTING_WAIT_SEQ
:
1350 uint64_t newly_acked_seq
;
1351 r
= read_until(sizeof(newly_acked_seq
), state_buffer
);
1353 ldout(async_msgr
->cct
, 1) << __func__
<< " read ack seq failed" << dendl
;
1354 goto fail_registered
;
1359 newly_acked_seq
= *((uint64_t*)state_buffer
);
1360 ldout(async_msgr
->cct
, 2) << __func__
<< " accept get newly_acked_seq " << newly_acked_seq
<< dendl
;
1361 discard_requeued_up_to(newly_acked_seq
);
1362 state
= STATE_ACCEPTING_READY
;
1366 case STATE_ACCEPTING_READY
:
1368 ldout(async_msgr
->cct
, 20) << __func__
<< " accept done" << dendl
;
1370 memset(&connect_msg
, 0, sizeof(connect_msg
));
1373 assert(delay_state
->ready());
1374 // make sure no pending tick timer
1376 center
->delete_time_event(last_tick_id
);
1377 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);
1380 can_write
= WriteStatus::CANWRITE
;
1382 center
->dispatch_event_external(write_handler
);
1383 write_lock
.unlock();
1384 maybe_start_delay_thread();
1390 lderr(async_msgr
->cct
) << __func__
<< " bad state: " << state
<< dendl
;
1398 ldout(async_msgr
->cct
, 10) << "accept fault after register" << dendl
;
1405 int AsyncConnection::handle_connect_reply(ceph_msg_connect
&connect
, ceph_msg_connect_reply
&reply
)
1407 uint64_t feat_missing
;
1408 if (reply
.tag
== CEPH_MSGR_TAG_FEATURES
) {
1409 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol feature mismatch, my "
1410 << std::hex
<< connect
.features
<< " < peer "
1411 << reply
.features
<< " missing "
1412 << (reply
.features
& ~policy
.features_supported
)
1413 << std::dec
<< dendl
;
1417 if (reply
.tag
== CEPH_MSGR_TAG_BADPROTOVER
) {
1418 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol version mismatch, my "
1419 << connect
.protocol_version
<< " != " << reply
.protocol_version
1424 if (reply
.tag
== CEPH_MSGR_TAG_BADAUTHORIZER
) {
1425 ldout(async_msgr
->cct
,0) << __func__
<< " connect got BADAUTHORIZER" << dendl
;
1428 got_bad_auth
= true;
1430 authorizer
= async_msgr
->get_authorizer(peer_type
, true); // try harder
1431 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1433 if (reply
.tag
== CEPH_MSGR_TAG_RESETSESSION
) {
1434 ldout(async_msgr
->cct
, 0) << __func__
<< " connect got RESETSESSION" << dendl
;
1435 was_session_reset();
1436 // see was_session_reset
1437 outcoming_bl
.clear();
1438 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1440 if (reply
.tag
== CEPH_MSGR_TAG_RETRY_GLOBAL
) {
1441 global_seq
= async_msgr
->get_global_seq(reply
.global_seq
);
1442 ldout(async_msgr
->cct
, 5) << __func__
<< " connect got RETRY_GLOBAL "
1443 << reply
.global_seq
<< " chose new "
1444 << global_seq
<< dendl
;
1445 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1447 if (reply
.tag
== CEPH_MSGR_TAG_RETRY_SESSION
) {
1448 assert(reply
.connect_seq
> connect_seq
);
1449 ldout(async_msgr
->cct
, 5) << __func__
<< " connect got RETRY_SESSION "
1450 << connect_seq
<< " -> "
1451 << reply
.connect_seq
<< dendl
;
1452 connect_seq
= reply
.connect_seq
;
1453 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1455 if (reply
.tag
== CEPH_MSGR_TAG_WAIT
) {
1456 ldout(async_msgr
->cct
, 1) << __func__
<< " connect got WAIT (connection race)" << dendl
;
1460 feat_missing
= policy
.features_required
& ~(uint64_t)connect_reply
.features
;
1462 ldout(async_msgr
->cct
, 1) << __func__
<< " missing required features " << std::hex
1463 << feat_missing
<< std::dec
<< dendl
;
1467 if (reply
.tag
== CEPH_MSGR_TAG_SEQ
) {
1468 ldout(async_msgr
->cct
, 10) << __func__
<< " got CEPH_MSGR_TAG_SEQ, reading acked_seq and writing in_seq" << dendl
;
1469 state
= STATE_CONNECTING_WAIT_ACK_SEQ
;
1471 if (reply
.tag
== CEPH_MSGR_TAG_READY
) {
1472 ldout(async_msgr
->cct
, 10) << __func__
<< " got CEPH_MSGR_TAG_READY " << dendl
;
1473 state
= STATE_CONNECTING_READY
;
1482 ssize_t
AsyncConnection::handle_connect_msg(ceph_msg_connect
&connect
, bufferlist
&authorizer_bl
,
1483 bufferlist
&authorizer_reply
)
1486 ceph_msg_connect_reply reply
;
1487 bufferlist reply_bl
;
1489 memset(&reply
, 0, sizeof(reply
));
1490 reply
.protocol_version
= async_msgr
->get_proto_version(peer_type
, false);
1493 ldout(async_msgr
->cct
, 10) << __func__
<< " accept my proto " << reply
.protocol_version
1494 << ", their proto " << connect
.protocol_version
<< dendl
;
1495 if (connect
.protocol_version
!= reply
.protocol_version
) {
1496 return _reply_accept(CEPH_MSGR_TAG_BADPROTOVER
, connect
, reply
, authorizer_reply
);
1498 // require signatures for cephx?
1499 if (connect
.authorizer_protocol
== CEPH_AUTH_CEPHX
) {
1500 if (peer_type
== CEPH_ENTITY_TYPE_OSD
||
1501 peer_type
== CEPH_ENTITY_TYPE_MDS
||
1502 peer_type
== CEPH_ENTITY_TYPE_MGR
) {
1503 if (async_msgr
->cct
->_conf
->cephx_require_signatures
||
1504 async_msgr
->cct
->_conf
->cephx_cluster_require_signatures
) {
1505 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring MSG_AUTH feature bit for cluster" << dendl
;
1506 policy
.features_required
|= CEPH_FEATURE_MSG_AUTH
;
1508 if (async_msgr
->cct
->_conf
->cephx_require_version
>= 2 ||
1509 async_msgr
->cct
->_conf
->cephx_cluster_require_version
>= 2) {
1510 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring cephx v2 feature bit for cluster" << dendl
;
1511 policy
.features_required
|= CEPH_FEATUREMASK_CEPHX_V2
;
1514 if (async_msgr
->cct
->_conf
->cephx_require_signatures
||
1515 async_msgr
->cct
->_conf
->cephx_service_require_signatures
) {
1516 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring MSG_AUTH feature bit for service" << dendl
;
1517 policy
.features_required
|= CEPH_FEATURE_MSG_AUTH
;
1519 if (async_msgr
->cct
->_conf
->cephx_require_version
>= 2 ||
1520 async_msgr
->cct
->_conf
->cephx_service_require_version
>= 2) {
1521 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring cephx v2 feature bit for service" << dendl
;
1522 policy
.features_required
|= CEPH_FEATUREMASK_CEPHX_V2
;
1527 uint64_t feat_missing
= policy
.features_required
& ~(uint64_t)connect
.features
;
1529 ldout(async_msgr
->cct
, 1) << __func__
<< " peer missing required features "
1530 << std::hex
<< feat_missing
<< std::dec
<< dendl
;
1531 return _reply_accept(CEPH_MSGR_TAG_FEATURES
, connect
, reply
, authorizer_reply
);
1536 bool authorizer_valid
;
1537 bool need_challenge
= HAVE_FEATURE(connect
.features
, CEPHX_V2
);
1538 bool had_challenge
= (bool)authorizer_challenge
;
1539 if (!async_msgr
->verify_authorizer(
1540 this, peer_type
, connect
.authorizer_protocol
, authorizer_bl
,
1541 authorizer_reply
, authorizer_valid
, session_key
,
1542 need_challenge
? &authorizer_challenge
: nullptr) ||
1543 !authorizer_valid
) {
1546 if (need_challenge
&& !had_challenge
&& authorizer_challenge
) {
1547 ldout(async_msgr
->cct
,10) << __func__
<< ": challenging authorizer"
1549 assert(authorizer_reply
.length());
1550 tag
= CEPH_MSGR_TAG_CHALLENGE_AUTHORIZER
;
1552 ldout(async_msgr
->cct
,0) << __func__
<< ": got bad authorizer" << dendl
;
1553 tag
= CEPH_MSGR_TAG_BADAUTHORIZER
;
1555 session_security
.reset();
1556 return _reply_accept(tag
, connect
, reply
, authorizer_reply
);
1559 // We've verified the authorizer for this AsyncConnection, so set up the session security structure. PLR
1560 ldout(async_msgr
->cct
, 10) << __func__
<< " accept setting up session_security." << dendl
;
1563 AsyncConnectionRef existing
= async_msgr
->lookup_conn(peer_addr
);
1568 if (state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
1569 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while accept, it must be mark_down" << dendl
;
1570 assert(state
== STATE_CLOSED
);
1574 if (existing
== this)
1577 // There is no possible that existing connection will acquire this
1578 // connection's lock
1579 existing
->lock
.lock(); // skip lockdep check (we are locking a second AsyncConnection here)
1581 if (existing
->state
== STATE_CLOSED
) {
1582 ldout(async_msgr
->cct
, 1) << __func__
<< " existing already closed." << dendl
;
1583 existing
->lock
.unlock();
1588 if (existing
->replacing
) {
1589 ldout(async_msgr
->cct
, 1) << __func__
<< " existing racing replace happened while replacing."
1590 << " existing_state=" << get_state_name(existing
->state
) << dendl
;
1591 reply
.global_seq
= existing
->peer_global_seq
;
1592 r
= _reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
);
1593 existing
->lock
.unlock();
1599 if (connect
.global_seq
< existing
->peer_global_seq
) {
1600 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
1601 << ".gseq " << existing
->peer_global_seq
<< " > "
1602 << connect
.global_seq
<< ", RETRY_GLOBAL" << dendl
;
1603 reply
.global_seq
= existing
->peer_global_seq
; // so we can send it below..
1604 existing
->lock
.unlock();
1605 return _reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
);
1607 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
1608 << ".gseq " << existing
->peer_global_seq
1609 << " <= " << connect
.global_seq
<< ", looks ok" << dendl
;
1612 if (existing
->policy
.lossy
) {
1613 ldout(async_msgr
->cct
, 0) << __func__
<< " accept replacing existing (lossy) channel (new one lossy="
1614 << policy
.lossy
<< ")" << dendl
;
1615 existing
->was_session_reset();
1619 ldout(async_msgr
->cct
, 0) << __func__
<< " accept connect_seq " << connect
.connect_seq
1620 << " vs existing csq=" << existing
->connect_seq
<< " existing_state="
1621 << get_state_name(existing
->state
) << dendl
;
1623 if (connect
.connect_seq
== 0 && existing
->connect_seq
> 0) {
1624 ldout(async_msgr
->cct
,0) << __func__
<< " accept peer reset, then tried to connect to us, replacing" << dendl
;
1625 // this is a hard reset from peer
1626 is_reset_from_peer
= true;
1627 if (policy
.resetcheck
)
1628 existing
->was_session_reset(); // this resets out_queue, msg_ and connect_seq #'s
1632 if (connect
.connect_seq
< existing
->connect_seq
) {
1633 // old attempt, or we sent READY but they didn't get it.
1634 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
<< ".cseq "
1635 << existing
->connect_seq
<< " > " << connect
.connect_seq
1636 << ", RETRY_SESSION" << dendl
;
1637 reply
.connect_seq
= existing
->connect_seq
+ 1;
1638 existing
->lock
.unlock();
1639 return _reply_accept(CEPH_MSGR_TAG_RETRY_SESSION
, connect
, reply
, authorizer_reply
);
1642 if (connect
.connect_seq
== existing
->connect_seq
) {
1643 // if the existing connection successfully opened, and/or
1644 // subsequently went to standby, then the peer should bump
1645 // their connect_seq and retry: this is not a connection race
1646 // we need to resolve here.
1647 if (existing
->state
== STATE_OPEN
||
1648 existing
->state
== STATE_STANDBY
) {
1649 ldout(async_msgr
->cct
, 10) << __func__
<< " accept connection race, existing " << existing
1650 << ".cseq " << existing
->connect_seq
<< " == "
1651 << connect
.connect_seq
<< ", OPEN|STANDBY, RETRY_SESSION" << dendl
;
1652 reply
.connect_seq
= existing
->connect_seq
+ 1;
1653 existing
->lock
.unlock();
1654 return _reply_accept(CEPH_MSGR_TAG_RETRY_SESSION
, connect
, reply
, authorizer_reply
);
1658 if (peer_addr
< async_msgr
->get_myaddr() || existing
->policy
.server
) {
1660 ldout(async_msgr
->cct
, 10) << __func__
<< " accept connection race, existing " << existing
1661 << ".cseq " << existing
->connect_seq
<< " == " << connect
.connect_seq
1662 << ", or we are server, replacing my attempt" << dendl
;
1665 // our existing outgoing wins
1666 ldout(async_msgr
->cct
,10) << __func__
<< " accept connection race, existing "
1667 << existing
<< ".cseq " << existing
->connect_seq
1668 << " == " << connect
.connect_seq
<< ", sending WAIT" << dendl
;
1669 assert(peer_addr
> async_msgr
->get_myaddr());
1670 existing
->lock
.unlock();
1671 return _reply_accept(CEPH_MSGR_TAG_WAIT
, connect
, reply
, authorizer_reply
);
1675 assert(connect
.connect_seq
> existing
->connect_seq
);
1676 assert(connect
.global_seq
>= existing
->peer_global_seq
);
1677 if (policy
.resetcheck
&& // RESETSESSION only used by servers; peers do not reset each other
1678 existing
->connect_seq
== 0) {
1679 ldout(async_msgr
->cct
, 0) << __func__
<< " accept we reset (peer sent cseq "
1680 << connect
.connect_seq
<< ", " << existing
<< ".cseq = "
1681 << existing
->connect_seq
<< "), sending RESETSESSION" << dendl
;
1682 existing
->lock
.unlock();
1683 return _reply_accept(CEPH_MSGR_TAG_RESETSESSION
, connect
, reply
, authorizer_reply
);
1687 ldout(async_msgr
->cct
, 10) << __func__
<< " accept peer sent cseq " << connect
.connect_seq
1688 << " > " << existing
->connect_seq
<< dendl
;
1691 else if (!replacing
&& connect
.connect_seq
> 0) {
1692 // we reset, and they are opening a new session
1693 ldout(async_msgr
->cct
, 0) << __func__
<< " accept we reset (peer sent cseq "
1694 << connect
.connect_seq
<< "), sending RESETSESSION" << dendl
;
1695 return _reply_accept(CEPH_MSGR_TAG_RESETSESSION
, connect
, reply
, authorizer_reply
);
1698 ldout(async_msgr
->cct
, 10) << __func__
<< " accept new session" << dendl
;
1705 ldout(async_msgr
->cct
, 10) << __func__
<< " accept replacing " << existing
<< dendl
;
1708 if (existing
->policy
.lossy
) {
1709 // disconnect from the Connection
1710 ldout(async_msgr
->cct
, 1) << __func__
<< " replacing on lossy channel, failing existing" << dendl
;
1712 existing
->dispatch_queue
->queue_reset(existing
.get());
1714 assert(can_write
== WriteStatus::NOWRITE
);
1715 existing
->write_lock
.lock();
1717 // reset the in_seq if this is a hard reset from peer,
1718 // otherwise we respect our original connection's value
1719 if (is_reset_from_peer
) {
1720 existing
->is_reset_from_peer
= true;
1723 center
->delete_file_event(cs
.fd(), EVENT_READABLE
|EVENT_WRITABLE
);
1725 if (existing
->delay_state
) {
1726 existing
->delay_state
->flush();
1727 assert(!delay_state
);
1729 existing
->reset_recv_state();
1731 auto temp_cs
= std::move(cs
);
1732 EventCenter
*new_center
= center
;
1733 Worker
*new_worker
= worker
;
1734 // avoid _stop shutdown replacing socket
1735 // queue a reset on the new connection, which we're dumping for the old
1738 dispatch_queue
->queue_reset(this);
1739 ldout(async_msgr
->cct
, 1) << __func__
<< " stop myself to swap existing" << dendl
;
1740 existing
->can_write
= WriteStatus::REPLACING
;
1741 existing
->replacing
= true;
1742 existing
->state_offset
= 0;
1743 // avoid previous thread modify event
1744 existing
->state
= STATE_NONE
;
1745 // Discard existing prefetch buffer in `recv_buf`
1746 existing
->recv_start
= existing
->recv_end
= 0;
1747 // there shouldn't exist any buffer
1748 assert(recv_start
== recv_end
);
1750 existing
->authorizer_challenge
.reset();
1752 auto deactivate_existing
= std::bind(
1753 [existing
, new_worker
, new_center
, connect
, reply
, authorizer_reply
](ConnectedSocket
&cs
) mutable {
1754 // we need to delete time event in original thread
1756 std::lock_guard
<std::mutex
> l(existing
->lock
);
1757 existing
->write_lock
.lock();
1758 existing
->requeue_sent();
1759 existing
->outcoming_bl
.clear();
1760 existing
->open_write
= false;
1761 existing
->write_lock
.unlock();
1762 if (existing
->state
== STATE_NONE
) {
1763 existing
->shutdown_socket();
1764 existing
->cs
= std::move(cs
);
1765 existing
->worker
->references
--;
1766 new_worker
->references
++;
1767 existing
->logger
= new_worker
->get_perf_counter();
1768 existing
->worker
= new_worker
;
1769 existing
->center
= new_center
;
1770 if (existing
->delay_state
)
1771 existing
->delay_state
->set_center(new_center
);
1772 } else if (existing
->state
== STATE_CLOSED
) {
1773 auto back_to_close
= std::bind(
1774 [](ConnectedSocket
&cs
) mutable { cs
.close(); }, std::move(cs
));
1775 new_center
->submit_to(
1776 new_center
->get_id(), std::move(back_to_close
), true);
1783 // Before changing existing->center, it may already exists some events in existing->center's queue.
1784 // Then if we mark down `existing`, it will execute in another thread and clean up connection.
1785 // Previous event will result in segment fault
1786 auto transfer_existing
= [existing
, connect
, reply
, authorizer_reply
]() mutable {
1787 std::lock_guard
<std::mutex
> l(existing
->lock
);
1788 if (existing
->state
== STATE_CLOSED
)
1790 assert(existing
->state
== STATE_NONE
);
1792 existing
->state
= STATE_ACCEPTING_WAIT_CONNECT_MSG
;
1793 existing
->center
->create_file_event(existing
->cs
.fd(), EVENT_READABLE
, existing
->read_handler
);
1794 reply
.global_seq
= existing
->peer_global_seq
;
1795 if (existing
->_reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
) < 0) {
1800 if (existing
->center
->in_thread())
1801 transfer_existing();
1803 existing
->center
->submit_to(
1804 existing
->center
->get_id(), std::move(transfer_existing
), true);
1805 }, std::move(temp_cs
));
1807 existing
->center
->submit_to(
1808 existing
->center
->get_id(), std::move(deactivate_existing
), true);
1809 existing
->write_lock
.unlock();
1810 existing
->lock
.unlock();
1813 existing
->lock
.unlock();
1816 connect_seq
= connect
.connect_seq
+ 1;
1817 peer_global_seq
= connect
.global_seq
;
1818 ldout(async_msgr
->cct
, 10) << __func__
<< " accept success, connect_seq = "
1819 << connect_seq
<< " in_seq=" << in_seq
<< ", sending READY" << dendl
;
1823 // if it is a hard reset from peer, we don't need a round-trip to negotiate in/out sequence
1824 if ((connect
.features
& CEPH_FEATURE_RECONNECT_SEQ
) && !is_reset_from_peer
) {
1825 reply
.tag
= CEPH_MSGR_TAG_SEQ
;
1826 next_state
= STATE_ACCEPTING_WAIT_SEQ
;
1828 reply
.tag
= CEPH_MSGR_TAG_READY
;
1829 next_state
= STATE_ACCEPTING_READY
;
1830 discard_requeued_up_to(0);
1831 is_reset_from_peer
= false;
1836 reply
.features
= policy
.features_supported
;
1837 reply
.global_seq
= async_msgr
->get_global_seq();
1838 reply
.connect_seq
= connect_seq
;
1840 reply
.authorizer_len
= authorizer_reply
.length();
1842 reply
.flags
= reply
.flags
| CEPH_MSG_CONNECT_LOSSY
;
1844 set_features((uint64_t)reply
.features
& (uint64_t)connect
.features
);
1845 ldout(async_msgr
->cct
, 10) << __func__
<< " accept features " << get_features() << dendl
;
1847 session_security
.reset(
1848 get_auth_session_handler(async_msgr
->cct
, connect
.authorizer_protocol
,
1849 session_key
, get_features()));
1851 reply_bl
.append((char*)&reply
, sizeof(reply
));
1853 if (reply
.authorizer_len
)
1854 reply_bl
.append(authorizer_reply
.c_str(), authorizer_reply
.length());
1856 if (reply
.tag
== CEPH_MSGR_TAG_SEQ
) {
1857 uint64_t s
= in_seq
;
1858 reply_bl
.append((char*)&s
, sizeof(s
));
1862 // Because "replacing" will prevent other connections preempt this addr,
1863 // it's safe that here we don't acquire Connection's lock
1864 r
= async_msgr
->accept_conn(this);
1871 ldout(async_msgr
->cct
, 1) << __func__
<< " existing race replacing process for addr=" << peer_addr
1872 << " just fail later one(this)" << dendl
;
1873 goto fail_registered
;
1875 if (state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
1876 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while accept_conn, it must be mark_down" << dendl
;
1877 assert(state
== STATE_CLOSED
|| state
== STATE_NONE
);
1878 goto fail_registered
;
1881 r
= try_send(reply_bl
);
1883 goto fail_registered
;
1886 dispatch_queue
->queue_accept(this);
1887 async_msgr
->ms_deliver_handle_fast_accept(this);
1892 ldout(async_msgr
->cct
, 2) << __func__
<< " accept write reply msg done" << dendl
;
1894 state
= STATE_WAIT_SEND
;
1895 state_after_send
= next_state
;
1901 ldout(async_msgr
->cct
, 10) << __func__
<< " accept fault after register" << dendl
;
1905 ldout(async_msgr
->cct
, 10) << __func__
<< " failed to accept." << dendl
;
1909 void AsyncConnection::_connect()
1911 ldout(async_msgr
->cct
, 10) << __func__
<< " csq=" << connect_seq
<< dendl
;
1913 state
= STATE_CONNECTING
;
1914 // rescheduler connection in order to avoid lock dep
1915 // may called by external thread(send_message)
1916 center
->dispatch_event_external(read_handler
);
1919 void AsyncConnection::accept(ConnectedSocket socket
, entity_addr_t
&addr
)
1921 ldout(async_msgr
->cct
, 10) << __func__
<< " sd=" << socket
.fd() << dendl
;
1922 assert(socket
.fd() >= 0);
1924 std::lock_guard
<std::mutex
> l(lock
);
1925 cs
= std::move(socket
);
1927 state
= STATE_ACCEPTING
;
1928 // rescheduler connection in order to avoid lock dep
1929 center
->dispatch_event_external(read_handler
);
1932 int AsyncConnection::send_message(Message
*m
)
1935 lgeneric_subdout(async_msgr
->cct
, ms
,
1936 1) << "-- " << async_msgr
->get_myaddr() << " --> "
1937 << get_peer_addr() << " -- "
1938 << *m
<< " -- " << m
<< " con "
1939 << m
->get_connection().get()
1942 // optimistic think it's ok to encode(actually may broken now)
1943 if (!m
->get_priority())
1944 m
->set_priority(async_msgr
->get_default_send_priority());
1946 m
->get_header().src
= async_msgr
->get_myname();
1947 m
->set_connection(this);
1949 if (m
->get_type() == CEPH_MSG_OSD_OP
)
1950 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OP_BEGIN", true);
1951 else if (m
->get_type() == CEPH_MSG_OSD_OPREPLY
)
1952 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OPREPLY_BEGIN", true);
1954 if (async_msgr
->get_myaddr() == get_peer_addr()) { //loopback connection
1955 ldout(async_msgr
->cct
, 20) << __func__
<< " " << *m
<< " local" << dendl
;
1956 std::lock_guard
<std::mutex
> l(write_lock
);
1957 if (can_write
!= WriteStatus::CLOSED
) {
1958 dispatch_queue
->local_delivery(m
, m
->get_priority());
1960 ldout(async_msgr
->cct
, 10) << __func__
<< " loopback connection closed."
1961 << " Drop message " << m
<< dendl
;
1967 last_active
= ceph::coarse_mono_clock::now();
1968 // we don't want to consider local message here, it's too lightweight which
1969 // may disturb users
1970 logger
->inc(l_msgr_send_messages
);
1973 uint64_t f
= get_features();
1975 // TODO: Currently not all messages supports reencode like MOSDMap, so here
1976 // only let fast dispatch support messages prepare message
1977 bool can_fast_prepare
= async_msgr
->ms_can_fast_dispatch(m
);
1978 if (can_fast_prepare
)
1979 prepare_send_message(f
, m
, bl
);
1981 std::lock_guard
<std::mutex
> l(write_lock
);
1982 // "features" changes will change the payload encoding
1983 if (can_fast_prepare
&& (can_write
== WriteStatus::NOWRITE
|| get_features() != f
)) {
1984 // ensure the correctness of message encoding
1986 m
->get_payload().clear();
1987 ldout(async_msgr
->cct
, 5) << __func__
<< " clear encoded buffer previous "
1988 << f
<< " != " << get_features() << dendl
;
1990 if (can_write
== WriteStatus::CLOSED
) {
1991 ldout(async_msgr
->cct
, 10) << __func__
<< " connection closed."
1992 << " Drop message " << m
<< dendl
;
1995 m
->trace
.event("async enqueueing message");
1996 out_q
[m
->get_priority()].emplace_back(std::move(bl
), m
);
1997 ldout(async_msgr
->cct
, 15) << __func__
<< " inline write is denied, reschedule m=" << m
<< dendl
;
1998 if (can_write
!= WriteStatus::REPLACING
)
1999 center
->dispatch_event_external(write_handler
);
2004 void AsyncConnection::requeue_sent()
2009 list
<pair
<bufferlist
, Message
*> >& rq
= out_q
[CEPH_MSG_PRIO_HIGHEST
];
2010 while (!sent
.empty()) {
2011 Message
* m
= sent
.back();
2013 ldout(async_msgr
->cct
, 10) << __func__
<< " " << *m
<< " for resend "
2014 << " (" << m
->get_seq() << ")" << dendl
;
2015 rq
.push_front(make_pair(bufferlist(), m
));
2020 void AsyncConnection::discard_requeued_up_to(uint64_t seq
)
2022 ldout(async_msgr
->cct
, 10) << __func__
<< " " << seq
<< dendl
;
2023 std::lock_guard
<std::mutex
> l(write_lock
);
2024 if (out_q
.count(CEPH_MSG_PRIO_HIGHEST
) == 0)
2026 list
<pair
<bufferlist
, Message
*> >& rq
= out_q
[CEPH_MSG_PRIO_HIGHEST
];
2027 while (!rq
.empty()) {
2028 pair
<bufferlist
, Message
*> p
= rq
.front();
2029 if (p
.second
->get_seq() == 0 || p
.second
->get_seq() > seq
)
2031 ldout(async_msgr
->cct
, 10) << __func__
<< " " << *(p
.second
) << " for resend seq " << p
.second
->get_seq()
2032 << " <= " << seq
<< ", discarding" << dendl
;
2038 out_q
.erase(CEPH_MSG_PRIO_HIGHEST
);
2042 * Tears down the AsyncConnection's message queues, and removes them from the DispatchQueue
2043 * Must hold write_lock prior to calling.
2045 void AsyncConnection::discard_out_queue()
2047 ldout(async_msgr
->cct
, 10) << __func__
<< " started" << dendl
;
2049 for (list
<Message
*>::iterator p
= sent
.begin(); p
!= sent
.end(); ++p
) {
2050 ldout(async_msgr
->cct
, 20) << __func__
<< " discard " << *p
<< dendl
;
2054 for (map
<int, list
<pair
<bufferlist
, Message
*> > >::iterator p
= out_q
.begin(); p
!= out_q
.end(); ++p
)
2055 for (list
<pair
<bufferlist
, Message
*> >::iterator r
= p
->second
.begin(); r
!= p
->second
.end(); ++r
) {
2056 ldout(async_msgr
->cct
, 20) << __func__
<< " discard " << r
->second
<< dendl
;
2062 int AsyncConnection::randomize_out_seq()
2064 if (get_features() & CEPH_FEATURE_MSG_AUTH
) {
2065 // Set out_seq to a random value, so CRC won't be predictable. Don't bother checking seq_error
2066 // here. We'll check it on the call. PLR
2068 int seq_error
= get_random_bytes((char *)&rand_seq
, sizeof(rand_seq
));
2069 rand_seq
&= SEQ_MASK
;
2070 lsubdout(async_msgr
->cct
, ms
, 10) << __func__
<< " randomize_out_seq " << rand_seq
<< dendl
;
2074 // previously, seq #'s always started at 0.
2080 void AsyncConnection::fault()
2082 if (state
== STATE_CLOSED
|| state
== STATE_NONE
) {
2083 ldout(async_msgr
->cct
, 10) << __func__
<< " connection is already closed" << dendl
;
2087 if (policy
.lossy
&& !(state
>= STATE_CONNECTING
&& state
< STATE_CONNECTING_READY
)) {
2088 ldout(async_msgr
->cct
, 1) << __func__
<< " on lossy channel, failing" << dendl
;
2090 dispatch_queue
->queue_reset(this);
2095 can_write
= WriteStatus::NOWRITE
;
2099 // queue delayed items immediately
2101 delay_state
->flush();
2102 // requeue sent items
2104 recv_start
= recv_end
= 0;
2107 is_reset_from_peer
= false;
2108 outcoming_bl
.clear();
2109 if (!once_ready
&& !is_queued() &&
2110 state
>=STATE_ACCEPTING
&& state
<= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
2111 ldout(async_msgr
->cct
, 10) << __func__
<< " with nothing to send and in the half "
2112 << " accept state just closed" << dendl
;
2113 write_lock
.unlock();
2115 dispatch_queue
->queue_reset(this);
2119 if (policy
.standby
&& !is_queued() && state
!= STATE_WAIT
) {
2120 ldout(async_msgr
->cct
, 10) << __func__
<< " with nothing to send, going to standby" << dendl
;
2121 state
= STATE_STANDBY
;
2122 write_lock
.unlock();
2126 write_lock
.unlock();
2127 if (!(state
>= STATE_CONNECTING
&& state
< STATE_CONNECTING_READY
) &&
2128 state
!= STATE_WAIT
) { // STATE_WAIT is coming from STATE_CONNECTING_*
2129 // policy maybe empty when state is in accept
2130 if (policy
.server
) {
2131 ldout(async_msgr
->cct
, 0) << __func__
<< " server, going to standby" << dendl
;
2132 state
= STATE_STANDBY
;
2134 ldout(async_msgr
->cct
, 0) << __func__
<< " initiating reconnect" << dendl
;
2136 state
= STATE_CONNECTING
;
2138 backoff
= utime_t();
2139 center
->dispatch_event_external(read_handler
);
2141 if (state
== STATE_WAIT
) {
2142 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_max_backoff
);
2143 } else if (backoff
== utime_t()) {
2144 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_initial_backoff
);
2147 if (backoff
> async_msgr
->cct
->_conf
->ms_max_backoff
)
2148 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_max_backoff
);
2151 state
= STATE_CONNECTING
;
2152 ldout(async_msgr
->cct
, 10) << __func__
<< " waiting " << backoff
<< dendl
;
2154 register_time_events
.insert(center
->create_time_event(
2155 backoff
.to_nsec()/1000, wakeup_handler
));
2159 void AsyncConnection::was_session_reset()
2161 ldout(async_msgr
->cct
,10) << __func__
<< " started" << dendl
;
2162 std::lock_guard
<std::mutex
> l(write_lock
);
2164 delay_state
->discard();
2165 dispatch_queue
->discard_queue(conn_id
);
2166 discard_out_queue();
2167 // note: we need to clear outcoming_bl here, but was_session_reset may be
2168 // called by other thread, so let caller clear this itself!
2169 // outcoming_bl.clear();
2171 dispatch_queue
->queue_remote_reset(this);
2173 if (randomize_out_seq()) {
2174 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
;
2179 // it's safe to directly set 0, double locked
2182 can_write
= WriteStatus::NOWRITE
;
2185 void AsyncConnection::_stop()
2187 if (state
== STATE_CLOSED
)
2191 delay_state
->flush();
2193 ldout(async_msgr
->cct
, 2) << __func__
<< dendl
;
2194 std::lock_guard
<std::mutex
> l(write_lock
);
2197 dispatch_queue
->discard_queue(conn_id
);
2198 discard_out_queue();
2199 async_msgr
->unregister_conn(this);
2200 worker
->release_worker();
2202 state
= STATE_CLOSED
;
2204 can_write
= WriteStatus::CLOSED
;
2206 // Make sure in-queue events will been processed
2207 center
->dispatch_event_external(EventCallbackRef(new C_clean_handler(this)));
2210 void AsyncConnection::prepare_send_message(uint64_t features
, Message
*m
, bufferlist
&bl
)
2212 ldout(async_msgr
->cct
, 20) << __func__
<< " m" << " " << *m
<< dendl
;
2214 // associate message with Connection (for benefit of encode_payload)
2215 if (m
->empty_payload())
2216 ldout(async_msgr
->cct
, 20) << __func__
<< " encoding features "
2217 << features
<< " " << m
<< " " << *m
<< dendl
;
2219 ldout(async_msgr
->cct
, 20) << __func__
<< " half-reencoding features "
2220 << features
<< " " << m
<< " " << *m
<< dendl
;
2222 // encode and copy out of *m
2223 m
->encode(features
, msgr
->crcflags
);
2225 bl
.append(m
->get_payload());
2226 bl
.append(m
->get_middle());
2227 bl
.append(m
->get_data());
2230 ssize_t
AsyncConnection::write_message(Message
*m
, bufferlist
& bl
, bool more
)
2233 assert(center
->in_thread());
2234 m
->set_seq(++out_seq
);
2236 if (msgr
->crcflags
& MSG_CRC_HEADER
)
2237 m
->calc_header_crc();
2239 ceph_msg_header
& header
= m
->get_header();
2240 ceph_msg_footer
& footer
= m
->get_footer();
2242 // TODO: let sign_message could be reentry?
2243 // Now that we have all the crcs calculated, handle the
2244 // digital signature for the message, if the AsyncConnection has session
2245 // security set up. Some session security options do not
2246 // actually calculate and check the signature, but they should
2247 // handle the calls to sign_message and check_signature. PLR
2248 if (session_security
.get() == NULL
) {
2249 ldout(async_msgr
->cct
, 20) << __func__
<< " no session security" << dendl
;
2251 if (session_security
->sign_message(m
)) {
2252 ldout(async_msgr
->cct
, 20) << __func__
<< " failed to sign m="
2253 << m
<< "): sig = " << footer
.sig
<< dendl
;
2255 ldout(async_msgr
->cct
, 20) << __func__
<< " signed m=" << m
2256 << "): sig = " << footer
.sig
<< dendl
;
2260 unsigned original_bl_len
= outcoming_bl
.length();
2262 outcoming_bl
.append(CEPH_MSGR_TAG_MSG
);
2264 if (has_feature(CEPH_FEATURE_NOSRCADDR
)) {
2265 outcoming_bl
.append((char*)&header
, sizeof(header
));
2267 ceph_msg_header_old oldheader
;
2268 memcpy(&oldheader
, &header
, sizeof(header
));
2269 oldheader
.src
.name
= header
.src
;
2270 oldheader
.src
.addr
= get_peer_addr();
2271 oldheader
.orig_src
= oldheader
.src
;
2272 oldheader
.reserved
= header
.reserved
;
2273 oldheader
.crc
= ceph_crc32c(0, (unsigned char*)&oldheader
,
2274 sizeof(oldheader
) - sizeof(oldheader
.crc
));
2275 outcoming_bl
.append((char*)&oldheader
, sizeof(oldheader
));
2278 ldout(async_msgr
->cct
, 20) << __func__
<< " sending message type=" << header
.type
2279 << " src " << entity_name_t(header
.src
)
2280 << " front=" << header
.front_len
2281 << " data=" << header
.data_len
2282 << " off " << header
.data_off
<< dendl
;
2284 if ((bl
.length() <= ASYNC_COALESCE_THRESHOLD
) && (bl
.buffers().size() > 1)) {
2285 for (const auto &pb
: bl
.buffers()) {
2286 outcoming_bl
.append((char*)pb
.c_str(), pb
.length());
2289 outcoming_bl
.claim_append(bl
);
2292 // send footer; if receiver doesn't support signatures, use the old footer format
2293 ceph_msg_footer_old old_footer
;
2294 if (has_feature(CEPH_FEATURE_MSG_AUTH
)) {
2295 outcoming_bl
.append((char*)&footer
, sizeof(footer
));
2297 if (msgr
->crcflags
& MSG_CRC_HEADER
) {
2298 old_footer
.front_crc
= footer
.front_crc
;
2299 old_footer
.middle_crc
= footer
.middle_crc
;
2300 old_footer
.data_crc
= footer
.data_crc
;
2302 old_footer
.front_crc
= old_footer
.middle_crc
= 0;
2304 old_footer
.data_crc
= msgr
->crcflags
& MSG_CRC_DATA
? footer
.data_crc
: 0;
2305 old_footer
.flags
= footer
.flags
;
2306 outcoming_bl
.append((char*)&old_footer
, sizeof(old_footer
));
2309 m
->trace
.event("async writing message");
2310 ldout(async_msgr
->cct
, 20) << __func__
<< " sending " << m
->get_seq()
2311 << " " << m
<< dendl
;
2312 ssize_t total_send_size
= outcoming_bl
.length();
2313 ssize_t rc
= _try_send(more
);
2315 ldout(async_msgr
->cct
, 1) << __func__
<< " error sending " << m
<< ", "
2316 << cpp_strerror(rc
) << dendl
;
2317 } else if (rc
== 0) {
2318 logger
->inc(l_msgr_send_bytes
, total_send_size
- original_bl_len
);
2319 ldout(async_msgr
->cct
, 10) << __func__
<< " sending " << m
<< " done." << dendl
;
2321 logger
->inc(l_msgr_send_bytes
, total_send_size
- outcoming_bl
.length());
2322 ldout(async_msgr
->cct
, 10) << __func__
<< " sending " << m
<< " continuely." << dendl
;
2324 if (m
->get_type() == CEPH_MSG_OSD_OP
)
2325 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OP_END", false);
2326 else if (m
->get_type() == CEPH_MSG_OSD_OPREPLY
)
2327 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OPREPLY_END", false);
2333 void AsyncConnection::reset_recv_state()
2335 // clean up state internal variables and states
2336 if (state
>= STATE_CONNECTING_SEND_CONNECT_MSG
&&
2337 state
<= STATE_CONNECTING_READY
) {
2340 got_bad_auth
= false;
2343 if (state
> STATE_OPEN_MESSAGE_THROTTLE_MESSAGE
&&
2344 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
2345 && policy
.throttler_messages
) {
2346 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << 1
2347 << " message to policy throttler "
2348 << policy
.throttler_messages
->get_current() << "/"
2349 << policy
.throttler_messages
->get_max() << dendl
;
2350 policy
.throttler_messages
->put();
2352 if (state
> STATE_OPEN_MESSAGE_THROTTLE_BYTES
&&
2353 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
) {
2354 if (policy
.throttler_bytes
) {
2355 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << cur_msg_size
2356 << " bytes to policy throttler "
2357 << policy
.throttler_bytes
->get_current() << "/"
2358 << policy
.throttler_bytes
->get_max() << dendl
;
2359 policy
.throttler_bytes
->put(cur_msg_size
);
2362 if (state
> STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE
&&
2363 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
) {
2364 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << cur_msg_size
2365 << " bytes to dispatch_queue throttler "
2366 << dispatch_queue
->dispatch_throttler
.get_current() << "/"
2367 << dispatch_queue
->dispatch_throttler
.get_max() << dendl
;
2368 dispatch_queue
->dispatch_throttle_release(cur_msg_size
);
2372 void AsyncConnection::handle_ack(uint64_t seq
)
2374 ldout(async_msgr
->cct
, 15) << __func__
<< " got ack seq " << seq
<< dendl
;
2376 std::lock_guard
<std::mutex
> l(write_lock
);
2377 while (!sent
.empty() && sent
.front()->get_seq() <= seq
) {
2378 Message
* m
= sent
.front();
2380 ldout(async_msgr
->cct
, 10) << __func__
<< " got ack seq "
2381 << seq
<< " >= " << m
->get_seq() << " on "
2382 << m
<< " " << *m
<< dendl
;
2387 void AsyncConnection::DelayedDelivery::do_request(int id
)
2389 Message
*m
= nullptr;
2391 std::lock_guard
<std::mutex
> l(delay_lock
);
2392 register_time_events
.erase(id
);
2395 if (delay_queue
.empty())
2397 utime_t release
= delay_queue
.front().first
;
2398 m
= delay_queue
.front().second
;
2399 string delay_msg_type
= msgr
->cct
->_conf
->ms_inject_delay_msg_type
;
2400 utime_t now
= ceph_clock_now();
2401 if ((release
> now
&&
2402 (delay_msg_type
.empty() || m
->get_type_name() == delay_msg_type
))) {
2403 utime_t t
= release
- now
;
2406 delay_queue
.pop_front();
2408 if (msgr
->ms_can_fast_dispatch(m
)) {
2409 dispatch_queue
->fast_dispatch(m
);
2411 dispatch_queue
->enqueue(m
, m
->get_priority(), conn_id
);
2415 void AsyncConnection::DelayedDelivery::flush() {
2416 stop_dispatch
= true;
2418 center
->get_id(), [this] () mutable {
2419 std::lock_guard
<std::mutex
> l(delay_lock
);
2420 while (!delay_queue
.empty()) {
2421 Message
*m
= delay_queue
.front().second
;
2422 if (msgr
->ms_can_fast_dispatch(m
)) {
2423 dispatch_queue
->fast_dispatch(m
);
2425 dispatch_queue
->enqueue(m
, m
->get_priority(), conn_id
);
2427 delay_queue
.pop_front();
2429 for (auto i
: register_time_events
)
2430 center
->delete_time_event(i
);
2431 register_time_events
.clear();
2432 stop_dispatch
= false;
2436 void AsyncConnection::send_keepalive()
2438 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2439 std::lock_guard
<std::mutex
> l(write_lock
);
2440 if (can_write
!= WriteStatus::CLOSED
) {
2442 center
->dispatch_event_external(write_handler
);
2446 void AsyncConnection::mark_down()
2448 ldout(async_msgr
->cct
, 1) << __func__
<< dendl
;
2449 std::lock_guard
<std::mutex
> l(lock
);
2453 void AsyncConnection::_append_keepalive_or_ack(bool ack
, utime_t
*tp
)
2455 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2458 struct ceph_timespec ts
;
2459 tp
->encode_timeval(&ts
);
2460 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE2_ACK
);
2461 outcoming_bl
.append((char*)&ts
, sizeof(ts
));
2462 } else if (has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2
)) {
2463 struct ceph_timespec ts
;
2464 utime_t t
= ceph_clock_now();
2465 t
.encode_timeval(&ts
);
2466 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE2
);
2467 outcoming_bl
.append((char*)&ts
, sizeof(ts
));
2469 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE
);
2473 void AsyncConnection::handle_write()
2475 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2479 if (can_write
== WriteStatus::CANWRITE
) {
2481 _append_keepalive_or_ack();
2485 auto start
= ceph::mono_clock::now();
2489 Message
*m
= _get_next_outgoing(&data
);
2493 if (!policy
.lossy
) {
2498 more
= _has_next_outgoing();
2499 write_lock
.unlock();
2501 // send_message or requeue messages may not encode message
2503 prepare_send_message(get_features(), m
, data
);
2505 r
= write_message(m
, data
, more
);
2507 ldout(async_msgr
->cct
, 1) << __func__
<< " send msg failed" << dendl
;
2513 } while (can_write
== WriteStatus::CANWRITE
);
2514 write_lock
.unlock();
2516 uint64_t left
= ack_left
;
2520 outcoming_bl
.append(CEPH_MSGR_TAG_ACK
);
2521 outcoming_bl
.append((char*)&s
, sizeof(s
));
2522 ldout(async_msgr
->cct
, 10) << __func__
<< " try send msg ack, acked " << left
<< " messages" << dendl
;
2525 r
= _try_send(left
);
2526 } else if (is_queued()) {
2530 logger
->tinc(l_msgr_running_send_time
, ceph::mono_clock::now() - start
);
2532 ldout(async_msgr
->cct
, 1) << __func__
<< " send msg failed" << dendl
;
2536 write_lock
.unlock();
2539 if (state
== STATE_STANDBY
&& !policy
.server
&& is_queued()) {
2540 ldout(async_msgr
->cct
, 10) << __func__
<< " policy.server is false" << dendl
;
2542 } else if (cs
&& state
!= STATE_NONE
&& state
!= STATE_CONNECTING
&& state
!= STATE_CONNECTING_RE
&& state
!= STATE_CLOSED
) {
2545 ldout(async_msgr
->cct
, 1) << __func__
<< " send outcoming bl failed" << dendl
;
2546 write_lock
.unlock();
2552 write_lock
.unlock();
2564 void AsyncConnection::wakeup_from(uint64_t id
)
2567 register_time_events
.erase(id
);
2572 void AsyncConnection::tick(uint64_t id
)
2574 auto now
= ceph::coarse_mono_clock::now();
2575 ldout(async_msgr
->cct
, 20) << __func__
<< " last_id=" << last_tick_id
2576 << " last_active" << last_active
<< dendl
;
2577 std::lock_guard
<std::mutex
> l(lock
);
2579 auto idle_period
= std::chrono::duration_cast
<std::chrono::microseconds
>(now
- last_active
).count();
2580 if (inactive_timeout_us
< (uint64_t)idle_period
) {
2581 ldout(async_msgr
->cct
, 1) << __func__
<< " idle(" << idle_period
<< ") more than "
2582 << inactive_timeout_us
2583 << " us, mark self fault." << dendl
;
2585 } else if (is_connected()) {
2586 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);