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
982 << " (peer is possibly using public_bind_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 if (rand() % async_msgr
->cct
->_conf
->ms_inject_socket_failures
== 0) {
991 ldout(msgr
->cct
, 10) << __func__
<< " sleep for "
992 << async_msgr
->cct
->_conf
->ms_inject_internal_delays
<< dendl
;
994 t
.set_from_double(async_msgr
->cct
->_conf
->ms_inject_internal_delays
);
1000 if (state
!= STATE_CONNECTING_WAIT_BANNER_AND_IDENTIFY
) {
1001 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while learned_addr, mark_down or "
1002 << " replacing must be happened just now" << dendl
;
1006 ::encode(async_msgr
->get_myaddr(), myaddrbl
, 0); // legacy
1007 r
= try_send(myaddrbl
);
1009 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1010 ldout(async_msgr
->cct
, 10) << __func__
<< " connect sent my addr "
1011 << async_msgr
->get_myaddr() << dendl
;
1013 state
= STATE_WAIT_SEND
;
1014 state_after_send
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1015 ldout(async_msgr
->cct
, 10) << __func__
<< " connect send my addr done: "
1016 << async_msgr
->get_myaddr() << dendl
;
1018 ldout(async_msgr
->cct
, 2) << __func__
<< " connect couldn't write my addr, "
1019 << cpp_strerror(r
) << dendl
;
1026 case STATE_CONNECTING_SEND_CONNECT_MSG
:
1028 if (!got_bad_auth
) {
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
);
1109 bufferlist::iterator iter
= authorizer_reply
.begin();
1110 if (authorizer
&& !authorizer
->verify_reply(iter
)) {
1111 ldout(async_msgr
->cct
, 0) << __func__
<< " failed verifying authorize reply" << dendl
;
1115 r
= handle_connect_reply(connect_msg
, connect_reply
);
1119 // state must be changed!
1120 assert(state
!= STATE_CONNECTING_WAIT_CONNECT_REPLY_AUTH
);
1124 case STATE_CONNECTING_WAIT_ACK_SEQ
:
1126 uint64_t newly_acked_seq
= 0;
1128 r
= read_until(sizeof(newly_acked_seq
), state_buffer
);
1130 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect ack seq failed" << dendl
;
1136 newly_acked_seq
= *((uint64_t*)state_buffer
);
1137 ldout(async_msgr
->cct
, 2) << __func__
<< " got newly_acked_seq " << newly_acked_seq
1138 << " vs out_seq " << out_seq
<< dendl
;
1139 discard_requeued_up_to(newly_acked_seq
);
1140 //while (newly_acked_seq > out_seq.read()) {
1141 // Message *m = _get_next_outgoing(NULL);
1143 // ldout(async_msgr->cct, 2) << __func__ << " discarding previously sent " << m->get_seq()
1144 // << " " << *m << dendl;
1145 // assert(m->get_seq() <= newly_acked_seq);
1151 uint64_t s
= in_seq
;
1152 bl
.append((char*)&s
, sizeof(s
));
1155 state
= STATE_CONNECTING_READY
;
1156 ldout(async_msgr
->cct
, 10) << __func__
<< " send in_seq done " << dendl
;
1158 state_after_send
= STATE_CONNECTING_READY
;
1159 state
= STATE_WAIT_SEND
;
1160 ldout(async_msgr
->cct
, 10) << __func__
<< " continue send in_seq " << dendl
;
1167 case STATE_CONNECTING_READY
:
1170 peer_global_seq
= connect_reply
.global_seq
;
1171 policy
.lossy
= connect_reply
.flags
& CEPH_MSG_CONNECT_LOSSY
;
1175 assert(connect_seq
== connect_reply
.connect_seq
);
1176 backoff
= utime_t();
1177 set_features((uint64_t)connect_reply
.features
& (uint64_t)connect_msg
.features
);
1178 ldout(async_msgr
->cct
, 10) << __func__
<< " connect success " << connect_seq
1179 << ", lossy = " << policy
.lossy
<< ", features "
1180 << get_features() << dendl
;
1182 // If we have an authorizer, get a new AuthSessionHandler to deal with ongoing security of the
1184 if (authorizer
!= NULL
) {
1185 session_security
.reset(
1186 get_auth_session_handler(async_msgr
->cct
,
1187 authorizer
->protocol
,
1188 authorizer
->session_key
,
1191 // We have no authorizer, so we shouldn't be applying security to messages in this AsyncConnection. PLR
1192 session_security
.reset();
1196 assert(delay_state
->ready());
1197 dispatch_queue
->queue_connect(this);
1198 async_msgr
->ms_deliver_handle_fast_connect(this);
1200 // make sure no pending tick timer
1202 center
->delete_time_event(last_tick_id
);
1203 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);
1205 // message may in queue between last _try_send and connection ready
1206 // write event may already notify and we need to force scheduler again
1208 can_write
= WriteStatus::CANWRITE
;
1210 center
->dispatch_event_external(write_handler
);
1211 write_lock
.unlock();
1212 maybe_start_delay_thread();
1216 case STATE_ACCEPTING
:
1219 center
->create_file_event(cs
.fd(), EVENT_READABLE
, read_handler
);
1221 bl
.append(CEPH_BANNER
, strlen(CEPH_BANNER
));
1223 ::encode(async_msgr
->get_myaddr(), bl
, 0); // legacy
1224 port
= async_msgr
->get_myaddr().get_port();
1225 ::encode(socket_addr
, bl
, 0); // legacy
1226 ldout(async_msgr
->cct
, 1) << __func__
<< " sd=" << cs
.fd() << " " << socket_addr
<< dendl
;
1230 state
= STATE_ACCEPTING_WAIT_BANNER_ADDR
;
1231 ldout(async_msgr
->cct
, 10) << __func__
<< " write banner and addr done: "
1232 << get_peer_addr() << dendl
;
1234 state
= STATE_WAIT_SEND
;
1235 state_after_send
= STATE_ACCEPTING_WAIT_BANNER_ADDR
;
1236 ldout(async_msgr
->cct
, 10) << __func__
<< " wait for write banner and addr: "
1237 << get_peer_addr() << dendl
;
1244 case STATE_ACCEPTING_WAIT_BANNER_ADDR
:
1247 entity_addr_t peer_addr
;
1249 r
= read_until(strlen(CEPH_BANNER
) + sizeof(ceph_entity_addr
), state_buffer
);
1251 ldout(async_msgr
->cct
, 1) << __func__
<< " read peer banner and addr failed" << dendl
;
1257 if (memcmp(state_buffer
, CEPH_BANNER
, strlen(CEPH_BANNER
))) {
1258 ldout(async_msgr
->cct
, 1) << __func__
<< " accept peer sent bad banner '" << state_buffer
1259 << "' (should be '" << CEPH_BANNER
<< "')" << dendl
;
1263 addr_bl
.append(state_buffer
+strlen(CEPH_BANNER
), sizeof(ceph_entity_addr
));
1265 bufferlist::iterator ti
= addr_bl
.begin();
1266 ::decode(peer_addr
, ti
);
1269 ldout(async_msgr
->cct
, 10) << __func__
<< " accept peer addr is " << peer_addr
<< dendl
;
1270 if (peer_addr
.is_blank_ip()) {
1271 // peer apparently doesn't know what ip they have; figure it out for them.
1272 int port
= peer_addr
.get_port();
1273 peer_addr
.u
= socket_addr
.u
;
1274 peer_addr
.set_port(port
);
1275 ldout(async_msgr
->cct
, 0) << __func__
<< " accept peer addr is really " << peer_addr
1276 << " (socket is " << socket_addr
<< ")" << dendl
;
1278 set_peer_addr(peer_addr
); // so that connection_state gets set up
1279 state
= STATE_ACCEPTING_WAIT_CONNECT_MSG
;
1283 case STATE_ACCEPTING_WAIT_CONNECT_MSG
:
1285 r
= read_until(sizeof(connect_msg
), state_buffer
);
1287 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect msg failed" << dendl
;
1293 connect_msg
= *((ceph_msg_connect
*)state_buffer
);
1294 state
= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
;
1298 case STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
:
1300 bufferlist authorizer_reply
;
1302 if (connect_msg
.authorizer_len
) {
1303 if (!authorizer_buf
.length())
1304 authorizer_buf
.push_back(buffer::create(connect_msg
.authorizer_len
));
1306 r
= read_until(connect_msg
.authorizer_len
, authorizer_buf
.c_str());
1308 ldout(async_msgr
->cct
, 1) << __func__
<< " read connect authorizer failed" << dendl
;
1315 ldout(async_msgr
->cct
, 20) << __func__
<< " accept got peer connect_seq "
1316 << connect_msg
.connect_seq
<< " global_seq "
1317 << connect_msg
.global_seq
<< dendl
;
1318 set_peer_type(connect_msg
.host_type
);
1319 policy
= async_msgr
->get_policy(connect_msg
.host_type
);
1320 ldout(async_msgr
->cct
, 10) << __func__
<< " accept of host_type " << connect_msg
.host_type
1321 << ", policy.lossy=" << policy
.lossy
<< " policy.server="
1322 << policy
.server
<< " policy.standby=" << policy
.standby
1323 << " policy.resetcheck=" << policy
.resetcheck
<< dendl
;
1325 r
= handle_connect_msg(connect_msg
, authorizer_buf
, authorizer_reply
);
1329 // state is changed by "handle_connect_msg"
1330 assert(state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
);
1334 case STATE_ACCEPTING_WAIT_SEQ
:
1336 uint64_t newly_acked_seq
;
1337 r
= read_until(sizeof(newly_acked_seq
), state_buffer
);
1339 ldout(async_msgr
->cct
, 1) << __func__
<< " read ack seq failed" << dendl
;
1340 goto fail_registered
;
1345 newly_acked_seq
= *((uint64_t*)state_buffer
);
1346 ldout(async_msgr
->cct
, 2) << __func__
<< " accept get newly_acked_seq " << newly_acked_seq
<< dendl
;
1347 discard_requeued_up_to(newly_acked_seq
);
1348 state
= STATE_ACCEPTING_READY
;
1352 case STATE_ACCEPTING_READY
:
1354 ldout(async_msgr
->cct
, 20) << __func__
<< " accept done" << dendl
;
1356 memset(&connect_msg
, 0, sizeof(connect_msg
));
1359 assert(delay_state
->ready());
1360 // make sure no pending tick timer
1362 center
->delete_time_event(last_tick_id
);
1363 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);
1366 can_write
= WriteStatus::CANWRITE
;
1368 center
->dispatch_event_external(write_handler
);
1369 write_lock
.unlock();
1370 maybe_start_delay_thread();
1376 lderr(async_msgr
->cct
) << __func__
<< " bad state: " << state
<< dendl
;
1384 ldout(async_msgr
->cct
, 10) << "accept fault after register" << dendl
;
1391 int AsyncConnection::handle_connect_reply(ceph_msg_connect
&connect
, ceph_msg_connect_reply
&reply
)
1393 uint64_t feat_missing
;
1394 if (reply
.tag
== CEPH_MSGR_TAG_FEATURES
) {
1395 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol feature mismatch, my "
1396 << std::hex
<< connect
.features
<< " < peer "
1397 << reply
.features
<< " missing "
1398 << (reply
.features
& ~policy
.features_supported
)
1399 << std::dec
<< dendl
;
1403 if (reply
.tag
== CEPH_MSGR_TAG_BADPROTOVER
) {
1404 ldout(async_msgr
->cct
, 0) << __func__
<< " connect protocol version mismatch, my "
1405 << connect
.protocol_version
<< " != " << reply
.protocol_version
1410 if (reply
.tag
== CEPH_MSGR_TAG_BADAUTHORIZER
) {
1411 ldout(async_msgr
->cct
,0) << __func__
<< " connect got BADAUTHORIZER" << dendl
;
1414 got_bad_auth
= true;
1416 authorizer
= async_msgr
->get_authorizer(peer_type
, true); // try harder
1417 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1419 if (reply
.tag
== CEPH_MSGR_TAG_RESETSESSION
) {
1420 ldout(async_msgr
->cct
, 0) << __func__
<< " connect got RESETSESSION" << dendl
;
1421 was_session_reset();
1422 // see was_session_reset
1423 outcoming_bl
.clear();
1424 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1426 if (reply
.tag
== CEPH_MSGR_TAG_RETRY_GLOBAL
) {
1427 global_seq
= async_msgr
->get_global_seq(reply
.global_seq
);
1428 ldout(async_msgr
->cct
, 5) << __func__
<< " connect got RETRY_GLOBAL "
1429 << reply
.global_seq
<< " chose new "
1430 << global_seq
<< dendl
;
1431 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1433 if (reply
.tag
== CEPH_MSGR_TAG_RETRY_SESSION
) {
1434 assert(reply
.connect_seq
> connect_seq
);
1435 ldout(async_msgr
->cct
, 5) << __func__
<< " connect got RETRY_SESSION "
1436 << connect_seq
<< " -> "
1437 << reply
.connect_seq
<< dendl
;
1438 connect_seq
= reply
.connect_seq
;
1439 state
= STATE_CONNECTING_SEND_CONNECT_MSG
;
1441 if (reply
.tag
== CEPH_MSGR_TAG_WAIT
) {
1442 ldout(async_msgr
->cct
, 1) << __func__
<< " connect got WAIT (connection race)" << dendl
;
1446 feat_missing
= policy
.features_required
& ~(uint64_t)connect_reply
.features
;
1448 ldout(async_msgr
->cct
, 1) << __func__
<< " missing required features " << std::hex
1449 << feat_missing
<< std::dec
<< dendl
;
1453 if (reply
.tag
== CEPH_MSGR_TAG_SEQ
) {
1454 ldout(async_msgr
->cct
, 10) << __func__
<< " got CEPH_MSGR_TAG_SEQ, reading acked_seq and writing in_seq" << dendl
;
1455 state
= STATE_CONNECTING_WAIT_ACK_SEQ
;
1457 if (reply
.tag
== CEPH_MSGR_TAG_READY
) {
1458 ldout(async_msgr
->cct
, 10) << __func__
<< " got CEPH_MSGR_TAG_READY " << dendl
;
1459 state
= STATE_CONNECTING_READY
;
1468 ssize_t
AsyncConnection::handle_connect_msg(ceph_msg_connect
&connect
, bufferlist
&authorizer_bl
,
1469 bufferlist
&authorizer_reply
)
1472 ceph_msg_connect_reply reply
;
1473 bufferlist reply_bl
;
1475 memset(&reply
, 0, sizeof(reply
));
1476 reply
.protocol_version
= async_msgr
->get_proto_version(peer_type
, false);
1479 ldout(async_msgr
->cct
, 10) << __func__
<< " accept my proto " << reply
.protocol_version
1480 << ", their proto " << connect
.protocol_version
<< dendl
;
1481 if (connect
.protocol_version
!= reply
.protocol_version
) {
1482 return _reply_accept(CEPH_MSGR_TAG_BADPROTOVER
, connect
, reply
, authorizer_reply
);
1484 // require signatures for cephx?
1485 if (connect
.authorizer_protocol
== CEPH_AUTH_CEPHX
) {
1486 if (peer_type
== CEPH_ENTITY_TYPE_OSD
||
1487 peer_type
== CEPH_ENTITY_TYPE_MDS
) {
1488 if (async_msgr
->cct
->_conf
->cephx_require_signatures
||
1489 async_msgr
->cct
->_conf
->cephx_cluster_require_signatures
) {
1490 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring MSG_AUTH feature bit for cluster" << dendl
;
1491 policy
.features_required
|= CEPH_FEATURE_MSG_AUTH
;
1494 if (async_msgr
->cct
->_conf
->cephx_require_signatures
||
1495 async_msgr
->cct
->_conf
->cephx_service_require_signatures
) {
1496 ldout(async_msgr
->cct
, 10) << __func__
<< " using cephx, requiring MSG_AUTH feature bit for service" << dendl
;
1497 policy
.features_required
|= CEPH_FEATURE_MSG_AUTH
;
1501 uint64_t feat_missing
= policy
.features_required
& ~(uint64_t)connect
.features
;
1503 ldout(async_msgr
->cct
, 1) << __func__
<< " peer missing required features "
1504 << std::hex
<< feat_missing
<< std::dec
<< dendl
;
1505 return _reply_accept(CEPH_MSGR_TAG_FEATURES
, connect
, reply
, authorizer_reply
);
1510 bool authorizer_valid
;
1511 if (!async_msgr
->verify_authorizer(this, peer_type
, connect
.authorizer_protocol
, authorizer_bl
,
1512 authorizer_reply
, authorizer_valid
, session_key
) || !authorizer_valid
) {
1514 ldout(async_msgr
->cct
,0) << __func__
<< ": got bad authorizer" << dendl
;
1515 session_security
.reset();
1516 return _reply_accept(CEPH_MSGR_TAG_BADAUTHORIZER
, connect
, reply
, authorizer_reply
);
1519 // We've verified the authorizer for this AsyncConnection, so set up the session security structure. PLR
1520 ldout(async_msgr
->cct
, 10) << __func__
<< " accept setting up session_security." << dendl
;
1523 AsyncConnectionRef existing
= async_msgr
->lookup_conn(peer_addr
);
1528 if (state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
1529 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while accept, it must be mark_down" << dendl
;
1530 assert(state
== STATE_CLOSED
);
1534 if (existing
== this)
1537 // There is no possible that existing connection will acquire this
1538 // connection's lock
1539 existing
->lock
.lock(); // skip lockdep check (we are locking a second AsyncConnection here)
1541 if (existing
->state
== STATE_CLOSED
) {
1542 ldout(async_msgr
->cct
, 1) << __func__
<< " existing already closed." << dendl
;
1543 existing
->lock
.unlock();
1548 if (existing
->replacing
) {
1549 ldout(async_msgr
->cct
, 1) << __func__
<< " existing racing replace happened while replacing."
1550 << " existing_state=" << get_state_name(existing
->state
) << dendl
;
1551 reply
.global_seq
= existing
->peer_global_seq
;
1552 r
= _reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
);
1553 existing
->lock
.unlock();
1559 if (connect
.global_seq
< existing
->peer_global_seq
) {
1560 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
1561 << ".gseq " << existing
->peer_global_seq
<< " > "
1562 << connect
.global_seq
<< ", RETRY_GLOBAL" << dendl
;
1563 reply
.global_seq
= existing
->peer_global_seq
; // so we can send it below..
1564 existing
->lock
.unlock();
1565 return _reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
);
1567 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
1568 << ".gseq " << existing
->peer_global_seq
1569 << " <= " << connect
.global_seq
<< ", looks ok" << dendl
;
1572 if (existing
->policy
.lossy
) {
1573 ldout(async_msgr
->cct
, 0) << __func__
<< " accept replacing existing (lossy) channel (new one lossy="
1574 << policy
.lossy
<< ")" << dendl
;
1575 existing
->was_session_reset();
1579 ldout(async_msgr
->cct
, 0) << __func__
<< " accept connect_seq " << connect
.connect_seq
1580 << " vs existing csq=" << existing
->connect_seq
<< " existing_state="
1581 << get_state_name(existing
->state
) << dendl
;
1583 if (connect
.connect_seq
== 0 && existing
->connect_seq
> 0) {
1584 ldout(async_msgr
->cct
,0) << __func__
<< " accept peer reset, then tried to connect to us, replacing" << dendl
;
1585 // this is a hard reset from peer
1586 is_reset_from_peer
= true;
1587 if (policy
.resetcheck
)
1588 existing
->was_session_reset(); // this resets out_queue, msg_ and connect_seq #'s
1592 if (connect
.connect_seq
< existing
->connect_seq
) {
1593 // old attempt, or we sent READY but they didn't get it.
1594 ldout(async_msgr
->cct
, 10) << __func__
<< " accept existing " << existing
<< ".cseq "
1595 << existing
->connect_seq
<< " > " << connect
.connect_seq
1596 << ", RETRY_SESSION" << dendl
;
1597 reply
.connect_seq
= existing
->connect_seq
+ 1;
1598 existing
->lock
.unlock();
1599 return _reply_accept(CEPH_MSGR_TAG_RETRY_SESSION
, connect
, reply
, authorizer_reply
);
1602 if (connect
.connect_seq
== existing
->connect_seq
) {
1603 // if the existing connection successfully opened, and/or
1604 // subsequently went to standby, then the peer should bump
1605 // their connect_seq and retry: this is not a connection race
1606 // we need to resolve here.
1607 if (existing
->state
== STATE_OPEN
||
1608 existing
->state
== STATE_STANDBY
) {
1609 ldout(async_msgr
->cct
, 10) << __func__
<< " accept connection race, existing " << existing
1610 << ".cseq " << existing
->connect_seq
<< " == "
1611 << connect
.connect_seq
<< ", OPEN|STANDBY, RETRY_SESSION" << dendl
;
1612 reply
.connect_seq
= existing
->connect_seq
+ 1;
1613 existing
->lock
.unlock();
1614 return _reply_accept(CEPH_MSGR_TAG_RETRY_SESSION
, connect
, reply
, authorizer_reply
);
1618 if (peer_addr
< async_msgr
->get_myaddr() || existing
->policy
.server
) {
1620 ldout(async_msgr
->cct
, 10) << __func__
<< " accept connection race, existing " << existing
1621 << ".cseq " << existing
->connect_seq
<< " == " << connect
.connect_seq
1622 << ", or we are server, replacing my attempt" << dendl
;
1625 // our existing outgoing wins
1626 ldout(async_msgr
->cct
,10) << __func__
<< " accept connection race, existing "
1627 << existing
<< ".cseq " << existing
->connect_seq
1628 << " == " << connect
.connect_seq
<< ", sending WAIT" << dendl
;
1629 assert(peer_addr
> async_msgr
->get_myaddr());
1630 existing
->lock
.unlock();
1631 return _reply_accept(CEPH_MSGR_TAG_WAIT
, connect
, reply
, authorizer_reply
);
1635 assert(connect
.connect_seq
> existing
->connect_seq
);
1636 assert(connect
.global_seq
>= existing
->peer_global_seq
);
1637 if (policy
.resetcheck
&& // RESETSESSION only used by servers; peers do not reset each other
1638 existing
->connect_seq
== 0) {
1639 ldout(async_msgr
->cct
, 0) << __func__
<< " accept we reset (peer sent cseq "
1640 << connect
.connect_seq
<< ", " << existing
<< ".cseq = "
1641 << existing
->connect_seq
<< "), sending RESETSESSION" << dendl
;
1642 existing
->lock
.unlock();
1643 return _reply_accept(CEPH_MSGR_TAG_RESETSESSION
, connect
, reply
, authorizer_reply
);
1647 ldout(async_msgr
->cct
, 10) << __func__
<< " accept peer sent cseq " << connect
.connect_seq
1648 << " > " << existing
->connect_seq
<< dendl
;
1651 else if (!replacing
&& connect
.connect_seq
> 0) {
1652 // we reset, and they are opening a new session
1653 ldout(async_msgr
->cct
, 0) << __func__
<< " accept we reset (peer sent cseq "
1654 << connect
.connect_seq
<< "), sending RESETSESSION" << dendl
;
1655 return _reply_accept(CEPH_MSGR_TAG_RESETSESSION
, connect
, reply
, authorizer_reply
);
1658 ldout(async_msgr
->cct
, 10) << __func__
<< " accept new session" << dendl
;
1665 ldout(async_msgr
->cct
, 10) << __func__
<< " accept replacing " << existing
<< dendl
;
1668 if (existing
->policy
.lossy
) {
1669 // disconnect from the Connection
1670 ldout(async_msgr
->cct
, 1) << __func__
<< " replacing on lossy channel, failing existing" << dendl
;
1672 existing
->dispatch_queue
->queue_reset(existing
.get());
1674 assert(can_write
== WriteStatus::NOWRITE
);
1675 existing
->write_lock
.lock();
1677 // reset the in_seq if this is a hard reset from peer,
1678 // otherwise we respect our original connection's value
1679 if (is_reset_from_peer
) {
1680 existing
->is_reset_from_peer
= true;
1683 center
->delete_file_event(cs
.fd(), EVENT_READABLE
|EVENT_WRITABLE
);
1685 if (existing
->delay_state
) {
1686 existing
->delay_state
->flush();
1687 assert(!delay_state
);
1689 existing
->reset_recv_state();
1691 auto temp_cs
= std::move(cs
);
1692 EventCenter
*new_center
= center
;
1693 Worker
*new_worker
= worker
;
1694 // avoid _stop shutdown replacing socket
1695 // queue a reset on the new connection, which we're dumping for the old
1698 dispatch_queue
->queue_reset(this);
1699 ldout(async_msgr
->cct
, 1) << __func__
<< " stop myself to swap existing" << dendl
;
1700 existing
->can_write
= WriteStatus::REPLACING
;
1701 existing
->replacing
= true;
1702 existing
->state_offset
= 0;
1703 // avoid previous thread modify event
1704 existing
->state
= STATE_NONE
;
1705 // Discard existing prefetch buffer in `recv_buf`
1706 existing
->recv_start
= existing
->recv_end
= 0;
1707 // there shouldn't exist any buffer
1708 assert(recv_start
== recv_end
);
1710 auto deactivate_existing
= std::bind(
1711 [existing
, new_worker
, new_center
, connect
, reply
, authorizer_reply
](ConnectedSocket
&cs
) mutable {
1712 // we need to delete time event in original thread
1714 std::lock_guard
<std::mutex
> l(existing
->lock
);
1715 existing
->write_lock
.lock();
1716 existing
->requeue_sent();
1717 existing
->outcoming_bl
.clear();
1718 existing
->open_write
= false;
1719 existing
->write_lock
.unlock();
1720 if (existing
->state
== STATE_NONE
) {
1721 existing
->shutdown_socket();
1722 existing
->cs
= std::move(cs
);
1723 existing
->worker
->references
--;
1724 new_worker
->references
++;
1725 existing
->logger
= new_worker
->get_perf_counter();
1726 existing
->worker
= new_worker
;
1727 existing
->center
= new_center
;
1728 if (existing
->delay_state
)
1729 existing
->delay_state
->set_center(new_center
);
1730 } else if (existing
->state
== STATE_CLOSED
) {
1731 auto back_to_close
= std::bind(
1732 [](ConnectedSocket
&cs
) mutable { cs
.close(); }, std::move(cs
));
1733 new_center
->submit_to(
1734 new_center
->get_id(), std::move(back_to_close
), true);
1741 // Before changing existing->center, it may already exists some events in existing->center's queue.
1742 // Then if we mark down `existing`, it will execute in another thread and clean up connection.
1743 // Previous event will result in segment fault
1744 auto transfer_existing
= [existing
, connect
, reply
, authorizer_reply
]() mutable {
1745 std::lock_guard
<std::mutex
> l(existing
->lock
);
1746 if (existing
->state
== STATE_CLOSED
)
1748 assert(existing
->state
== STATE_NONE
);
1750 existing
->state
= STATE_ACCEPTING_WAIT_CONNECT_MSG
;
1751 existing
->center
->create_file_event(existing
->cs
.fd(), EVENT_READABLE
, existing
->read_handler
);
1752 reply
.global_seq
= existing
->peer_global_seq
;
1753 if (existing
->_reply_accept(CEPH_MSGR_TAG_RETRY_GLOBAL
, connect
, reply
, authorizer_reply
) < 0) {
1758 if (existing
->center
->in_thread())
1759 transfer_existing();
1761 existing
->center
->submit_to(
1762 existing
->center
->get_id(), std::move(transfer_existing
), true);
1763 }, std::move(temp_cs
));
1765 existing
->center
->submit_to(
1766 existing
->center
->get_id(), std::move(deactivate_existing
), true);
1767 existing
->write_lock
.unlock();
1768 existing
->lock
.unlock();
1771 existing
->lock
.unlock();
1774 connect_seq
= connect
.connect_seq
+ 1;
1775 peer_global_seq
= connect
.global_seq
;
1776 ldout(async_msgr
->cct
, 10) << __func__
<< " accept success, connect_seq = "
1777 << connect_seq
<< " in_seq=" << in_seq
<< ", sending READY" << dendl
;
1781 // if it is a hard reset from peer, we don't need a round-trip to negotiate in/out sequence
1782 if ((connect
.features
& CEPH_FEATURE_RECONNECT_SEQ
) && !is_reset_from_peer
) {
1783 reply
.tag
= CEPH_MSGR_TAG_SEQ
;
1784 next_state
= STATE_ACCEPTING_WAIT_SEQ
;
1786 reply
.tag
= CEPH_MSGR_TAG_READY
;
1787 next_state
= STATE_ACCEPTING_READY
;
1788 discard_requeued_up_to(0);
1789 is_reset_from_peer
= false;
1794 reply
.features
= policy
.features_supported
;
1795 reply
.global_seq
= async_msgr
->get_global_seq();
1796 reply
.connect_seq
= connect_seq
;
1798 reply
.authorizer_len
= authorizer_reply
.length();
1800 reply
.flags
= reply
.flags
| CEPH_MSG_CONNECT_LOSSY
;
1802 set_features((uint64_t)reply
.features
& (uint64_t)connect
.features
);
1803 ldout(async_msgr
->cct
, 10) << __func__
<< " accept features " << get_features() << dendl
;
1805 session_security
.reset(
1806 get_auth_session_handler(async_msgr
->cct
, connect
.authorizer_protocol
,
1807 session_key
, get_features()));
1809 reply_bl
.append((char*)&reply
, sizeof(reply
));
1811 if (reply
.authorizer_len
)
1812 reply_bl
.append(authorizer_reply
.c_str(), authorizer_reply
.length());
1814 if (reply
.tag
== CEPH_MSGR_TAG_SEQ
) {
1815 uint64_t s
= in_seq
;
1816 reply_bl
.append((char*)&s
, sizeof(s
));
1820 // Because "replacing" will prevent other connections preempt this addr,
1821 // it's safe that here we don't acquire Connection's lock
1822 r
= async_msgr
->accept_conn(this);
1829 ldout(async_msgr
->cct
, 1) << __func__
<< " existing race replacing process for addr=" << peer_addr
1830 << " just fail later one(this)" << dendl
;
1831 goto fail_registered
;
1833 if (state
!= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
1834 ldout(async_msgr
->cct
, 1) << __func__
<< " state changed while accept_conn, it must be mark_down" << dendl
;
1835 assert(state
== STATE_CLOSED
);
1836 goto fail_registered
;
1839 r
= try_send(reply_bl
);
1841 goto fail_registered
;
1844 dispatch_queue
->queue_accept(this);
1845 async_msgr
->ms_deliver_handle_fast_accept(this);
1850 ldout(async_msgr
->cct
, 2) << __func__
<< " accept write reply msg done" << dendl
;
1852 state
= STATE_WAIT_SEND
;
1853 state_after_send
= next_state
;
1859 ldout(async_msgr
->cct
, 10) << __func__
<< " accept fault after register" << dendl
;
1863 ldout(async_msgr
->cct
, 10) << __func__
<< " failed to accept." << dendl
;
1867 void AsyncConnection::_connect()
1869 ldout(async_msgr
->cct
, 10) << __func__
<< " csq=" << connect_seq
<< dendl
;
1871 state
= STATE_CONNECTING
;
1872 // rescheduler connection in order to avoid lock dep
1873 // may called by external thread(send_message)
1874 center
->dispatch_event_external(read_handler
);
1877 void AsyncConnection::accept(ConnectedSocket socket
, entity_addr_t
&addr
)
1879 ldout(async_msgr
->cct
, 10) << __func__
<< " sd=" << socket
.fd() << dendl
;
1880 assert(socket
.fd() >= 0);
1882 std::lock_guard
<std::mutex
> l(lock
);
1883 cs
= std::move(socket
);
1885 state
= STATE_ACCEPTING
;
1886 // rescheduler connection in order to avoid lock dep
1887 center
->dispatch_event_external(read_handler
);
1890 int AsyncConnection::send_message(Message
*m
)
1893 lgeneric_subdout(async_msgr
->cct
, ms
,
1894 1) << "-- " << async_msgr
->get_myaddr() << " --> "
1895 << get_peer_addr() << " -- "
1896 << *m
<< " -- " << m
<< " con "
1897 << m
->get_connection().get()
1900 // optimistic think it's ok to encode(actually may broken now)
1901 if (!m
->get_priority())
1902 m
->set_priority(async_msgr
->get_default_send_priority());
1904 m
->get_header().src
= async_msgr
->get_myname();
1905 m
->set_connection(this);
1907 if (m
->get_type() == CEPH_MSG_OSD_OP
)
1908 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OP_BEGIN", true);
1909 else if (m
->get_type() == CEPH_MSG_OSD_OPREPLY
)
1910 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OPREPLY_BEGIN", true);
1912 if (async_msgr
->get_myaddr() == get_peer_addr()) { //loopback connection
1913 ldout(async_msgr
->cct
, 20) << __func__
<< " " << *m
<< " local" << dendl
;
1914 std::lock_guard
<std::mutex
> l(write_lock
);
1915 if (can_write
!= WriteStatus::CLOSED
) {
1916 dispatch_queue
->local_delivery(m
, m
->get_priority());
1918 ldout(async_msgr
->cct
, 10) << __func__
<< " loopback connection closed."
1919 << " Drop message " << m
<< dendl
;
1925 last_active
= ceph::coarse_mono_clock::now();
1926 // we don't want to consider local message here, it's too lightweight which
1927 // may disturb users
1928 logger
->inc(l_msgr_send_messages
);
1931 uint64_t f
= get_features();
1933 // TODO: Currently not all messages supports reencode like MOSDMap, so here
1934 // only let fast dispatch support messages prepare message
1935 bool can_fast_prepare
= async_msgr
->ms_can_fast_dispatch(m
);
1936 if (can_fast_prepare
)
1937 prepare_send_message(f
, m
, bl
);
1939 std::lock_guard
<std::mutex
> l(write_lock
);
1940 // "features" changes will change the payload encoding
1941 if (can_fast_prepare
&& (can_write
== WriteStatus::NOWRITE
|| get_features() != f
)) {
1942 // ensure the correctness of message encoding
1944 m
->get_payload().clear();
1945 ldout(async_msgr
->cct
, 5) << __func__
<< " clear encoded buffer previous "
1946 << f
<< " != " << get_features() << dendl
;
1948 if (can_write
== WriteStatus::CLOSED
) {
1949 ldout(async_msgr
->cct
, 10) << __func__
<< " connection closed."
1950 << " Drop message " << m
<< dendl
;
1953 m
->trace
.event("async enqueueing message");
1954 out_q
[m
->get_priority()].emplace_back(std::move(bl
), m
);
1955 ldout(async_msgr
->cct
, 15) << __func__
<< " inline write is denied, reschedule m=" << m
<< dendl
;
1956 if (can_write
!= WriteStatus::REPLACING
)
1957 center
->dispatch_event_external(write_handler
);
1962 void AsyncConnection::requeue_sent()
1967 list
<pair
<bufferlist
, Message
*> >& rq
= out_q
[CEPH_MSG_PRIO_HIGHEST
];
1968 while (!sent
.empty()) {
1969 Message
* m
= sent
.back();
1971 ldout(async_msgr
->cct
, 10) << __func__
<< " " << *m
<< " for resend "
1972 << " (" << m
->get_seq() << ")" << dendl
;
1973 rq
.push_front(make_pair(bufferlist(), m
));
1978 void AsyncConnection::discard_requeued_up_to(uint64_t seq
)
1980 ldout(async_msgr
->cct
, 10) << __func__
<< " " << seq
<< dendl
;
1981 std::lock_guard
<std::mutex
> l(write_lock
);
1982 if (out_q
.count(CEPH_MSG_PRIO_HIGHEST
) == 0)
1984 list
<pair
<bufferlist
, Message
*> >& rq
= out_q
[CEPH_MSG_PRIO_HIGHEST
];
1985 while (!rq
.empty()) {
1986 pair
<bufferlist
, Message
*> p
= rq
.front();
1987 if (p
.second
->get_seq() == 0 || p
.second
->get_seq() > seq
)
1989 ldout(async_msgr
->cct
, 10) << __func__
<< " " << *(p
.second
) << " for resend seq " << p
.second
->get_seq()
1990 << " <= " << seq
<< ", discarding" << dendl
;
1996 out_q
.erase(CEPH_MSG_PRIO_HIGHEST
);
2000 * Tears down the AsyncConnection's message queues, and removes them from the DispatchQueue
2001 * Must hold write_lock prior to calling.
2003 void AsyncConnection::discard_out_queue()
2005 ldout(async_msgr
->cct
, 10) << __func__
<< " started" << dendl
;
2007 for (list
<Message
*>::iterator p
= sent
.begin(); p
!= sent
.end(); ++p
) {
2008 ldout(async_msgr
->cct
, 20) << __func__
<< " discard " << *p
<< dendl
;
2012 for (map
<int, list
<pair
<bufferlist
, Message
*> > >::iterator p
= out_q
.begin(); p
!= out_q
.end(); ++p
)
2013 for (list
<pair
<bufferlist
, Message
*> >::iterator r
= p
->second
.begin(); r
!= p
->second
.end(); ++r
) {
2014 ldout(async_msgr
->cct
, 20) << __func__
<< " discard " << r
->second
<< dendl
;
2020 int AsyncConnection::randomize_out_seq()
2022 if (get_features() & CEPH_FEATURE_MSG_AUTH
) {
2023 // Set out_seq to a random value, so CRC won't be predictable. Don't bother checking seq_error
2024 // here. We'll check it on the call. PLR
2026 int seq_error
= get_random_bytes((char *)&rand_seq
, sizeof(rand_seq
));
2027 rand_seq
&= SEQ_MASK
;
2028 lsubdout(async_msgr
->cct
, ms
, 10) << __func__
<< " randomize_out_seq " << rand_seq
<< dendl
;
2032 // previously, seq #'s always started at 0.
2038 void AsyncConnection::fault()
2040 if (state
== STATE_CLOSED
|| state
== STATE_NONE
) {
2041 ldout(async_msgr
->cct
, 10) << __func__
<< " connection is already closed" << dendl
;
2045 if (policy
.lossy
&& !(state
>= STATE_CONNECTING
&& state
< STATE_CONNECTING_READY
)) {
2046 ldout(async_msgr
->cct
, 1) << __func__
<< " on lossy channel, failing" << dendl
;
2048 dispatch_queue
->queue_reset(this);
2053 can_write
= WriteStatus::NOWRITE
;
2057 // queue delayed items immediately
2059 delay_state
->flush();
2060 // requeue sent items
2062 recv_start
= recv_end
= 0;
2065 is_reset_from_peer
= false;
2066 outcoming_bl
.clear();
2067 if (!once_ready
&& !is_queued() &&
2068 state
>=STATE_ACCEPTING
&& state
<= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH
) {
2069 ldout(async_msgr
->cct
, 10) << __func__
<< " with nothing to send and in the half "
2070 << " accept state just closed" << dendl
;
2071 write_lock
.unlock();
2073 dispatch_queue
->queue_reset(this);
2077 if (policy
.standby
&& !is_queued() && state
!= STATE_WAIT
) {
2078 ldout(async_msgr
->cct
, 10) << __func__
<< " with nothing to send, going to standby" << dendl
;
2079 state
= STATE_STANDBY
;
2080 write_lock
.unlock();
2084 write_lock
.unlock();
2085 if (!(state
>= STATE_CONNECTING
&& state
< STATE_CONNECTING_READY
) &&
2086 state
!= STATE_WAIT
) { // STATE_WAIT is coming from STATE_CONNECTING_*
2087 // policy maybe empty when state is in accept
2088 if (policy
.server
) {
2089 ldout(async_msgr
->cct
, 0) << __func__
<< " server, going to standby" << dendl
;
2090 state
= STATE_STANDBY
;
2092 ldout(async_msgr
->cct
, 0) << __func__
<< " initiating reconnect" << dendl
;
2094 state
= STATE_CONNECTING
;
2096 backoff
= utime_t();
2097 center
->dispatch_event_external(read_handler
);
2099 if (state
== STATE_WAIT
) {
2100 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_max_backoff
);
2101 } else if (backoff
== utime_t()) {
2102 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_initial_backoff
);
2105 if (backoff
> async_msgr
->cct
->_conf
->ms_max_backoff
)
2106 backoff
.set_from_double(async_msgr
->cct
->_conf
->ms_max_backoff
);
2109 state
= STATE_CONNECTING
;
2110 ldout(async_msgr
->cct
, 10) << __func__
<< " waiting " << backoff
<< dendl
;
2112 register_time_events
.insert(center
->create_time_event(
2113 backoff
.to_nsec()/1000, wakeup_handler
));
2117 void AsyncConnection::was_session_reset()
2119 ldout(async_msgr
->cct
,10) << __func__
<< " started" << dendl
;
2120 std::lock_guard
<std::mutex
> l(write_lock
);
2122 delay_state
->discard();
2123 dispatch_queue
->discard_queue(conn_id
);
2124 discard_out_queue();
2125 // note: we need to clear outcoming_bl here, but was_session_reset may be
2126 // called by other thread, so let caller clear this itself!
2127 // outcoming_bl.clear();
2129 dispatch_queue
->queue_remote_reset(this);
2131 if (randomize_out_seq()) {
2132 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
;
2137 // it's safe to directly set 0, double locked
2140 can_write
= WriteStatus::NOWRITE
;
2143 void AsyncConnection::_stop()
2145 if (state
== STATE_CLOSED
)
2149 delay_state
->flush();
2151 ldout(async_msgr
->cct
, 2) << __func__
<< dendl
;
2152 std::lock_guard
<std::mutex
> l(write_lock
);
2155 dispatch_queue
->discard_queue(conn_id
);
2156 discard_out_queue();
2157 async_msgr
->unregister_conn(this);
2158 worker
->release_worker();
2160 state
= STATE_CLOSED
;
2162 can_write
= WriteStatus::CLOSED
;
2164 // Make sure in-queue events will been processed
2165 center
->dispatch_event_external(EventCallbackRef(new C_clean_handler(this)));
2168 void AsyncConnection::prepare_send_message(uint64_t features
, Message
*m
, bufferlist
&bl
)
2170 ldout(async_msgr
->cct
, 20) << __func__
<< " m" << " " << *m
<< dendl
;
2172 // associate message with Connection (for benefit of encode_payload)
2173 if (m
->empty_payload())
2174 ldout(async_msgr
->cct
, 20) << __func__
<< " encoding features "
2175 << features
<< " " << m
<< " " << *m
<< dendl
;
2177 ldout(async_msgr
->cct
, 20) << __func__
<< " half-reencoding features "
2178 << features
<< " " << m
<< " " << *m
<< dendl
;
2180 // encode and copy out of *m
2181 m
->encode(features
, msgr
->crcflags
);
2183 bl
.append(m
->get_payload());
2184 bl
.append(m
->get_middle());
2185 bl
.append(m
->get_data());
2188 ssize_t
AsyncConnection::write_message(Message
*m
, bufferlist
& bl
, bool more
)
2191 assert(center
->in_thread());
2192 m
->set_seq(++out_seq
);
2194 if (msgr
->crcflags
& MSG_CRC_HEADER
)
2195 m
->calc_header_crc();
2197 ceph_msg_header
& header
= m
->get_header();
2198 ceph_msg_footer
& footer
= m
->get_footer();
2200 // TODO: let sign_message could be reentry?
2201 // Now that we have all the crcs calculated, handle the
2202 // digital signature for the message, if the AsyncConnection has session
2203 // security set up. Some session security options do not
2204 // actually calculate and check the signature, but they should
2205 // handle the calls to sign_message and check_signature. PLR
2206 if (session_security
.get() == NULL
) {
2207 ldout(async_msgr
->cct
, 20) << __func__
<< " no session security" << dendl
;
2209 if (session_security
->sign_message(m
)) {
2210 ldout(async_msgr
->cct
, 20) << __func__
<< " failed to sign m="
2211 << m
<< "): sig = " << footer
.sig
<< dendl
;
2213 ldout(async_msgr
->cct
, 20) << __func__
<< " signed m=" << m
2214 << "): sig = " << footer
.sig
<< dendl
;
2218 unsigned original_bl_len
= outcoming_bl
.length();
2220 outcoming_bl
.append(CEPH_MSGR_TAG_MSG
);
2222 if (has_feature(CEPH_FEATURE_NOSRCADDR
)) {
2223 outcoming_bl
.append((char*)&header
, sizeof(header
));
2225 ceph_msg_header_old oldheader
;
2226 memcpy(&oldheader
, &header
, sizeof(header
));
2227 oldheader
.src
.name
= header
.src
;
2228 oldheader
.src
.addr
= get_peer_addr();
2229 oldheader
.orig_src
= oldheader
.src
;
2230 oldheader
.reserved
= header
.reserved
;
2231 oldheader
.crc
= ceph_crc32c(0, (unsigned char*)&oldheader
,
2232 sizeof(oldheader
) - sizeof(oldheader
.crc
));
2233 outcoming_bl
.append((char*)&oldheader
, sizeof(oldheader
));
2236 ldout(async_msgr
->cct
, 20) << __func__
<< " sending message type=" << header
.type
2237 << " src " << entity_name_t(header
.src
)
2238 << " front=" << header
.front_len
2239 << " data=" << header
.data_len
2240 << " off " << header
.data_off
<< dendl
;
2242 if ((bl
.length() <= ASYNC_COALESCE_THRESHOLD
) && (bl
.buffers().size() > 1)) {
2243 std::list
<buffer::ptr
>::const_iterator pb
;
2244 for (pb
= bl
.buffers().begin(); pb
!= bl
.buffers().end(); ++pb
) {
2245 outcoming_bl
.append((char*)pb
->c_str(), pb
->length());
2248 outcoming_bl
.claim_append(bl
);
2251 // send footer; if receiver doesn't support signatures, use the old footer format
2252 ceph_msg_footer_old old_footer
;
2253 if (has_feature(CEPH_FEATURE_MSG_AUTH
)) {
2254 outcoming_bl
.append((char*)&footer
, sizeof(footer
));
2256 if (msgr
->crcflags
& MSG_CRC_HEADER
) {
2257 old_footer
.front_crc
= footer
.front_crc
;
2258 old_footer
.middle_crc
= footer
.middle_crc
;
2259 old_footer
.data_crc
= footer
.data_crc
;
2261 old_footer
.front_crc
= old_footer
.middle_crc
= 0;
2263 old_footer
.data_crc
= msgr
->crcflags
& MSG_CRC_DATA
? footer
.data_crc
: 0;
2264 old_footer
.flags
= footer
.flags
;
2265 outcoming_bl
.append((char*)&old_footer
, sizeof(old_footer
));
2268 m
->trace
.event("async writing message");
2269 logger
->inc(l_msgr_send_bytes
, outcoming_bl
.length() - original_bl_len
);
2270 ldout(async_msgr
->cct
, 20) << __func__
<< " sending " << m
->get_seq()
2271 << " " << m
<< dendl
;
2272 ssize_t rc
= _try_send(more
);
2274 ldout(async_msgr
->cct
, 1) << __func__
<< " error sending " << m
<< ", "
2275 << cpp_strerror(rc
) << dendl
;
2276 } else if (rc
== 0) {
2277 ldout(async_msgr
->cct
, 10) << __func__
<< " sending " << m
<< " done." << dendl
;
2279 ldout(async_msgr
->cct
, 10) << __func__
<< " sending " << m
<< " continuely." << dendl
;
2281 if (m
->get_type() == CEPH_MSG_OSD_OP
)
2282 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OP_END", false);
2283 else if (m
->get_type() == CEPH_MSG_OSD_OPREPLY
)
2284 OID_EVENT_TRACE_WITH_MSG(m
, "SEND_MSG_OSD_OPREPLY_END", false);
2290 void AsyncConnection::reset_recv_state()
2292 // clean up state internal variables and states
2293 if (state
>= STATE_CONNECTING_SEND_CONNECT_MSG
&&
2294 state
<= STATE_CONNECTING_READY
) {
2297 got_bad_auth
= false;
2300 if (state
> STATE_OPEN_MESSAGE_THROTTLE_MESSAGE
&&
2301 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
2302 && policy
.throttler_messages
) {
2303 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << 1
2304 << " message to policy throttler "
2305 << policy
.throttler_messages
->get_current() << "/"
2306 << policy
.throttler_messages
->get_max() << dendl
;
2307 policy
.throttler_messages
->put();
2309 if (state
> STATE_OPEN_MESSAGE_THROTTLE_BYTES
&&
2310 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
) {
2311 if (policy
.throttler_bytes
) {
2312 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << cur_msg_size
2313 << " bytes to policy throttler "
2314 << policy
.throttler_bytes
->get_current() << "/"
2315 << policy
.throttler_bytes
->get_max() << dendl
;
2316 policy
.throttler_bytes
->put(cur_msg_size
);
2319 if (state
> STATE_OPEN_MESSAGE_THROTTLE_DISPATCH_QUEUE
&&
2320 state
<= STATE_OPEN_MESSAGE_READ_FOOTER_AND_DISPATCH
) {
2321 ldout(async_msgr
->cct
, 10) << __func__
<< " releasing " << cur_msg_size
2322 << " bytes to dispatch_queue throttler "
2323 << dispatch_queue
->dispatch_throttler
.get_current() << "/"
2324 << dispatch_queue
->dispatch_throttler
.get_max() << dendl
;
2325 dispatch_queue
->dispatch_throttle_release(cur_msg_size
);
2329 void AsyncConnection::handle_ack(uint64_t seq
)
2331 ldout(async_msgr
->cct
, 15) << __func__
<< " got ack seq " << seq
<< dendl
;
2333 std::lock_guard
<std::mutex
> l(write_lock
);
2334 while (!sent
.empty() && sent
.front()->get_seq() <= seq
) {
2335 Message
* m
= sent
.front();
2337 ldout(async_msgr
->cct
, 10) << __func__
<< " got ack seq "
2338 << seq
<< " >= " << m
->get_seq() << " on "
2339 << m
<< " " << *m
<< dendl
;
2344 void AsyncConnection::DelayedDelivery::do_request(int id
)
2346 Message
*m
= nullptr;
2348 std::lock_guard
<std::mutex
> l(delay_lock
);
2349 register_time_events
.erase(id
);
2352 if (delay_queue
.empty())
2354 utime_t release
= delay_queue
.front().first
;
2355 m
= delay_queue
.front().second
;
2356 string delay_msg_type
= msgr
->cct
->_conf
->ms_inject_delay_msg_type
;
2357 utime_t now
= ceph_clock_now();
2358 if ((release
> now
&&
2359 (delay_msg_type
.empty() || m
->get_type_name() == delay_msg_type
))) {
2360 utime_t t
= release
- now
;
2363 delay_queue
.pop_front();
2365 if (msgr
->ms_can_fast_dispatch(m
)) {
2366 dispatch_queue
->fast_dispatch(m
);
2368 dispatch_queue
->enqueue(m
, m
->get_priority(), conn_id
);
2372 void AsyncConnection::DelayedDelivery::flush() {
2373 stop_dispatch
= true;
2375 center
->get_id(), [this] () mutable {
2376 std::lock_guard
<std::mutex
> l(delay_lock
);
2377 while (!delay_queue
.empty()) {
2378 Message
*m
= delay_queue
.front().second
;
2379 if (msgr
->ms_can_fast_dispatch(m
)) {
2380 dispatch_queue
->fast_dispatch(m
);
2382 dispatch_queue
->enqueue(m
, m
->get_priority(), conn_id
);
2384 delay_queue
.pop_front();
2386 for (auto i
: register_time_events
)
2387 center
->delete_time_event(i
);
2388 register_time_events
.clear();
2389 stop_dispatch
= false;
2393 void AsyncConnection::send_keepalive()
2395 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2396 std::lock_guard
<std::mutex
> l(write_lock
);
2397 if (can_write
!= WriteStatus::CLOSED
) {
2399 center
->dispatch_event_external(write_handler
);
2403 void AsyncConnection::mark_down()
2405 ldout(async_msgr
->cct
, 1) << __func__
<< dendl
;
2406 std::lock_guard
<std::mutex
> l(lock
);
2410 void AsyncConnection::_append_keepalive_or_ack(bool ack
, utime_t
*tp
)
2412 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2415 struct ceph_timespec ts
;
2416 tp
->encode_timeval(&ts
);
2417 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE2_ACK
);
2418 outcoming_bl
.append((char*)&ts
, sizeof(ts
));
2419 } else if (has_feature(CEPH_FEATURE_MSGR_KEEPALIVE2
)) {
2420 struct ceph_timespec ts
;
2421 utime_t t
= ceph_clock_now();
2422 t
.encode_timeval(&ts
);
2423 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE2
);
2424 outcoming_bl
.append((char*)&ts
, sizeof(ts
));
2426 outcoming_bl
.append(CEPH_MSGR_TAG_KEEPALIVE
);
2430 void AsyncConnection::handle_write()
2432 ldout(async_msgr
->cct
, 10) << __func__
<< dendl
;
2436 if (can_write
== WriteStatus::CANWRITE
) {
2438 _append_keepalive_or_ack();
2442 auto start
= ceph::mono_clock::now();
2446 Message
*m
= _get_next_outgoing(&data
);
2450 if (!policy
.lossy
) {
2455 more
= _has_next_outgoing();
2456 write_lock
.unlock();
2458 // send_message or requeue messages may not encode message
2460 prepare_send_message(get_features(), m
, data
);
2462 r
= write_message(m
, data
, more
);
2464 ldout(async_msgr
->cct
, 1) << __func__
<< " send msg failed" << dendl
;
2470 } while (can_write
== WriteStatus::CANWRITE
);
2471 write_lock
.unlock();
2473 uint64_t left
= ack_left
;
2477 outcoming_bl
.append(CEPH_MSGR_TAG_ACK
);
2478 outcoming_bl
.append((char*)&s
, sizeof(s
));
2479 ldout(async_msgr
->cct
, 10) << __func__
<< " try send msg ack, acked " << left
<< " messages" << dendl
;
2482 r
= _try_send(left
);
2483 } else if (is_queued()) {
2487 logger
->tinc(l_msgr_running_send_time
, ceph::mono_clock::now() - start
);
2489 ldout(async_msgr
->cct
, 1) << __func__
<< " send msg failed" << dendl
;
2493 write_lock
.unlock();
2496 if (state
== STATE_STANDBY
&& !policy
.server
&& is_queued()) {
2497 ldout(async_msgr
->cct
, 10) << __func__
<< " policy.server is false" << dendl
;
2499 } else if (cs
&& state
!= STATE_NONE
&& state
!= STATE_CONNECTING
&& state
!= STATE_CONNECTING_RE
&& state
!= STATE_CLOSED
) {
2502 ldout(async_msgr
->cct
, 1) << __func__
<< " send outcoming bl failed" << dendl
;
2503 write_lock
.unlock();
2509 write_lock
.unlock();
2521 void AsyncConnection::wakeup_from(uint64_t id
)
2524 register_time_events
.erase(id
);
2529 void AsyncConnection::tick(uint64_t id
)
2531 auto now
= ceph::coarse_mono_clock::now();
2532 ldout(async_msgr
->cct
, 20) << __func__
<< " last_id=" << last_tick_id
2533 << " last_active" << last_active
<< dendl
;
2534 std::lock_guard
<std::mutex
> l(lock
);
2536 auto idle_period
= std::chrono::duration_cast
<std::chrono::microseconds
>(now
- last_active
).count();
2537 if (inactive_timeout_us
< (uint64_t)idle_period
) {
2538 ldout(async_msgr
->cct
, 1) << __func__
<< " idle(" << idle_period
<< ") more than "
2539 << inactive_timeout_us
2540 << " us, mark self fault." << dendl
;
2542 } else if (is_connected()) {
2543 last_tick_id
= center
->create_time_event(inactive_timeout_us
, tick_handler
);