]> git.proxmox.com Git - ceph.git/blame - ceph/src/msg/async/ProtocolV2.cc
projects / ceph.git / blame
? search:
summary | shortlog | log | commit | commitdiff | tree
blob | blame (incremental) | history | HEAD
Import ceph 15.2.8
[ceph.git] / ceph / src / msg / async / ProtocolV2.cc
CommitLineData
11fdf7f2
TL		 1// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2// vim: ts=8 sw=2 smarttab
3
4#include <type_traits>
5
6#include "ProtocolV2.h"
7#include "AsyncMessenger.h"
8
9#include "common/EventTrace.h"
10#include "common/ceph_crypto.h"
11#include "common/errno.h"
12#include "include/random.h"
13#include "auth/AuthClient.h"
14#include "auth/AuthServer.h"
15
16#define dout_subsys ceph_subsys_ms
17#undef dout_prefix
18#define dout_prefix _conn_prefix(_dout)
19ostream &ProtocolV2::_conn_prefix(std::ostream *_dout) {
20 return *_dout << "--2- " << messenger->get_myaddrs() << " >> "
21 << *connection->peer_addrs << " conn(" << connection << " "
22 << this
23 << " " << ceph_con_mode_name(auth_meta->con_mode)
24 << " :" << connection->port
25 << " s=" << get_state_name(state) << " pgs=" << peer_global_seq
26 << " cs=" << connect_seq << " l=" << connection->policy.lossy
f6b5b4d7
TL		 27 << " rev1=" << HAVE_MSGR2_FEATURE(peer_supported_features,
28 REVISION_1)
11fdf7f2
TL		 29 << " rx=" << session_stream_handlers.rx.get()
30 << " tx=" << session_stream_handlers.tx.get()
31 << ").";
32}
33
34using namespace ceph::msgr::v2;
35
36using CtPtr = Ct<ProtocolV2> *;
37using CtRef = Ct<ProtocolV2> &;
38
39void ProtocolV2::run_continuation(CtPtr pcontinuation) {
40 if (pcontinuation) {
41 run_continuation(*pcontinuation);
42 }
43}
44
45void ProtocolV2::run_continuation(CtRef continuation) {
46 try {
47 CONTINUATION_RUN(continuation)
48 } catch (const buffer::error &e) {
49 lderr(cct) << __func__ << " failed decoding of frame header: " << e
50 << dendl;
51 _fault();
52 } catch (const ceph::crypto::onwire::MsgAuthError &e) {
53 lderr(cct) << __func__ << " " << e.what() << dendl;
54 _fault();
55 } catch (const DecryptionError &) {
56 lderr(cct) << __func__ << " failed to decrypt frame payload" << dendl;
57 }
58}
59
60#define WRITE(B, D, C) write(D, CONTINUATION(C), B)
61
62#define READ(L, C) read(CONTINUATION(C), buffer::ptr_node::create(buffer::create(L)))
63
64#define READ_RXBUF(B, C) read(CONTINUATION(C), B)
65
66#ifdef UNIT_TESTS_BUILT
67
68#define INTERCEPT(S) { \
69if(connection->interceptor) { \
70 auto a = connection->interceptor->intercept(connection, (S)); \
71 if (a == Interceptor::ACTION::FAIL) { \
72 return _fault(); \
73 } else if (a == Interceptor::ACTION::STOP) { \
74 stop(); \
75 connection->dispatch_queue->queue_reset(connection); \
76 return nullptr; \
77 }}}
78
79#else
80#define INTERCEPT(S)
81#endif
82
83ProtocolV2::ProtocolV2(AsyncConnection *connection)
84 : Protocol(2, connection),
85 state(NONE),
f6b5b4d7 86 peer_supported_features(0),
11fdf7f2
TL		 87 client_cookie(0),
88 server_cookie(0),
89 global_seq(0),
90 connect_seq(0),
91 peer_global_seq(0),
92 message_seq(0),
93 reconnecting(false),
94 replacing(false),
95 can_write(false),
96 bannerExchangeCallback(nullptr),
f6b5b4d7
TL		 97 tx_frame_asm(&session_stream_handlers, false),
98 rx_frame_asm(&session_stream_handlers, false),
11fdf7f2
TL		 99 next_tag(static_cast<Tag>(0)),
100 keepalive(false) {
101}
102
103ProtocolV2::~ProtocolV2() {
104}
105
106void ProtocolV2::connect() {
107 ldout(cct, 1) << __func__ << dendl;
108 state = START_CONNECT;
109 pre_auth.enabled = true;
110}
111
112void ProtocolV2::accept() {
113 ldout(cct, 1) << __func__ << dendl;
114 state = START_ACCEPT;
115}
116
117bool ProtocolV2::is_connected() { return can_write; }
118
119/*
120 * Tears down the message queues, and removes them from the
121 * DispatchQueue Must hold write_lock prior to calling.
122 */
123void ProtocolV2::discard_out_queue() {
124 ldout(cct, 10) << __func__ << " started" << dendl;
125
126 for (list<Message *>::iterator p = sent.begin(); p != sent.end(); ++p) {
127 ldout(cct, 20) << __func__ << " discard " << *p << dendl;
128 (*p)->put();
129 }
130 sent.clear();
131 for (auto& [ prio, entries ] : out_queue) {
132 static_cast<void>(prio);
133 for (auto& entry : entries) {
134 ldout(cct, 20) << __func__ << " discard " << *entry.m << dendl;
135 entry.m->put();
136 }
137 }
138 out_queue.clear();
494da23a 139 write_in_progress = false;
11fdf7f2
TL		 140}
141
142void ProtocolV2::reset_session() {
143 ldout(cct, 1) << __func__ << dendl;
144
145 std::lock_guard<std::mutex> l(connection->write_lock);
146 if (connection->delay_state) {
147 connection->delay_state->discard();
148 }
149
150 connection->dispatch_queue->discard_queue(connection->conn_id);
151 discard_out_queue();
9f95a23c 152 connection->outgoing_bl.clear();
11fdf7f2
TL		 153
154 connection->dispatch_queue->queue_remote_reset(connection);
155
156 out_seq = 0;
157 in_seq = 0;
158 client_cookie = 0;
159 server_cookie = 0;
160 connect_seq = 0;
161 peer_global_seq = 0;
162 message_seq = 0;
163 ack_left = 0;
164 can_write = false;
165}
166
167void ProtocolV2::stop() {
168 ldout(cct, 1) << __func__ << dendl;
169 if (state == CLOSED) {
170 return;
171 }
172
173 if (connection->delay_state) connection->delay_state->flush();
174
175 std::lock_guard<std::mutex> l(connection->write_lock);
176
177 reset_recv_state();
178 discard_out_queue();
179
180 connection->_stop();
181
182 can_write = false;
183 state = CLOSED;
184}
185
186void ProtocolV2::fault() { _fault(); }
187
188void ProtocolV2::requeue_sent() {
494da23a 189 write_in_progress = false;
11fdf7f2
TL		 190 if (sent.empty()) {
191 return;
192 }
193
194 auto& rq = out_queue[CEPH_MSG_PRIO_HIGHEST];
195 out_seq -= sent.size();
196 while (!sent.empty()) {
197 Message *m = sent.back();
198 sent.pop_back();
199 ldout(cct, 5) << __func__ << " requeueing message m=" << m
200 << " seq=" << m->get_seq() << " type=" << m->get_type() << " "
201 << *m << dendl;
9f95a23c 202 m->clear_payload();
11fdf7f2
TL		 203 rq.emplace_front(out_queue_entry_t{false, m});
204 }
205}
206
207uint64_t ProtocolV2::discard_requeued_up_to(uint64_t out_seq, uint64_t seq) {
208 ldout(cct, 10) << __func__ << " " << seq << dendl;
209 std::lock_guard<std::mutex> l(connection->write_lock);
210 if (out_queue.count(CEPH_MSG_PRIO_HIGHEST) == 0) {
211 return seq;
212 }
213 auto& rq = out_queue[CEPH_MSG_PRIO_HIGHEST];
214 uint64_t count = out_seq;
215 while (!rq.empty()) {
216 Message* const m = rq.front().m;
217 if (m->get_seq() == 0 || m->get_seq() > seq) break;
218 ldout(cct, 5) << __func__ << " discarding message m=" << m
219 << " seq=" << m->get_seq() << " ack_seq=" << seq << " "
220 << *m << dendl;
221 m->put();
222 rq.pop_front();
223 count++;
224 }
225 if (rq.empty()) out_queue.erase(CEPH_MSG_PRIO_HIGHEST);
226 return count;
227}
228
9f95a23c
TL		 229void ProtocolV2::reset_security() {
230 ldout(cct, 5) << __func__ << dendl;
231
494da23a 232 auth_meta.reset(new AuthConnectionMeta);
494da23a 233 session_stream_handlers.rx.reset(nullptr);
9f95a23c 234 session_stream_handlers.tx.reset(nullptr);
494da23a 235 pre_auth.rxbuf.clear();
9f95a23c
TL		 236 pre_auth.txbuf.clear();
237}
238
239// it's expected the `write_lock` is held while calling this method.
240void ProtocolV2::reset_recv_state() {
241 ldout(cct, 5) << __func__ << dendl;
242
243 if (!connection->center->in_thread()) {
244 // execute in the same thread that uses the rx/tx handlers. We need
245 // to do the warp because holding `write_lock` is not enough as
246 // `write_event()` unlocks it just before calling `write_message()`.
247 // `submit_to()` here is NOT blocking.
248 connection->center->submit_to(connection->center->get_id(), [this] {
249 ldout(cct, 5) << "reset_recv_state (warped) reseting crypto handlers"
250 << dendl;
251 // Possibly unnecessary. See the comment in `deactivate_existing`.
252 std::lock_guard<std::mutex> l(connection->lock);
253 std::lock_guard<std::mutex> wl(connection->write_lock);
254 reset_security();
255 }, /* always_async = */true);
256 } else {
257 reset_security();
258 }
11fdf7f2
TL		 259
260 // clean read and write callbacks
261 connection->pendingReadLen.reset();
262 connection->writeCallback.reset();
263
264 next_tag = static_cast<Tag>(0);
265
266 reset_throttle();
267}
268
269size_t ProtocolV2::get_current_msg_size() const {
f6b5b4d7 270 ceph_assert(rx_frame_asm.get_num_segments() > 0);
11fdf7f2
TL		 271 size_t sum = 0;
272 // we don't include SegmentIndex::Msg::HEADER.
f6b5b4d7
TL		 273 for (size_t i = 1; i < rx_frame_asm.get_num_segments(); i++) {
274 sum += rx_frame_asm.get_segment_logical_len(i);
11fdf7f2
TL		 275 }
276 return sum;
277}
278
279void ProtocolV2::reset_throttle() {
280 if (state > THROTTLE_MESSAGE && state <= THROTTLE_DONE &&
281 connection->policy.throttler_messages) {
282 ldout(cct, 10) << __func__ << " releasing " << 1
283 << " message to policy throttler "
284 << connection->policy.throttler_messages->get_current()
285 << "/" << connection->policy.throttler_messages->get_max()
286 << dendl;
287 connection->policy.throttler_messages->put();
288 }
289 if (state > THROTTLE_BYTES && state <= THROTTLE_DONE) {
290 if (connection->policy.throttler_bytes) {
291 const size_t cur_msg_size = get_current_msg_size();
292 ldout(cct, 10) << __func__ << " releasing " << cur_msg_size
293 << " bytes to policy throttler "
294 << connection->policy.throttler_bytes->get_current() << "/"
295 << connection->policy.throttler_bytes->get_max() << dendl;
296 connection->policy.throttler_bytes->put(cur_msg_size);
297 }
298 }
299 if (state > THROTTLE_DISPATCH_QUEUE && state <= THROTTLE_DONE) {
300 const size_t cur_msg_size = get_current_msg_size();
301 ldout(cct, 10)
302 << __func__ << " releasing " << cur_msg_size
303 << " bytes to dispatch_queue throttler "
304 << connection->dispatch_queue->dispatch_throttler.get_current() << "/"
305 << connection->dispatch_queue->dispatch_throttler.get_max() << dendl;
306 connection->dispatch_queue->dispatch_throttle_release(cur_msg_size);
307 }
308}
309
310CtPtr ProtocolV2::_fault() {
311 ldout(cct, 10) << __func__ << dendl;
312
313 if (state == CLOSED || state == NONE) {
314 ldout(cct, 10) << __func__ << " connection is already closed" << dendl;
315 return nullptr;
316 }
317
318 if (connection->policy.lossy &&
319 !(state >= START_CONNECT && state <= SESSION_RECONNECTING)) {
320 ldout(cct, 2) << __func__ << " on lossy channel, failing" << dendl;
321 stop();
322 connection->dispatch_queue->queue_reset(connection);
323 return nullptr;
324 }
325
326 connection->write_lock.lock();
327
328 can_write = false;
329 // requeue sent items
330 requeue_sent();
331
332 if (out_queue.empty() && state >= START_ACCEPT &&
333 state <= SESSION_ACCEPTING && !replacing) {
334 ldout(cct, 2) << __func__ << " with nothing to send and in the half "
335 << " accept state just closed" << dendl;
336 connection->write_lock.unlock();
337 stop();
338 connection->dispatch_queue->queue_reset(connection);
339 return nullptr;
340 }
341
342 replacing = false;
343 connection->fault();
344 reset_recv_state();
345
346 reconnecting = false;
347
348 if (connection->policy.standby && out_queue.empty() && !keepalive &&
349 state != WAIT) {
350 ldout(cct, 1) << __func__ << " with nothing to send, going to standby"
351 << dendl;
352 state = STANDBY;
353 connection->write_lock.unlock();
354 return nullptr;
355 }
356 if (connection->policy.server) {
357 ldout(cct, 1) << __func__ << " server, going to standby, even though i have stuff queued" << dendl;
358 state = STANDBY;
359 connection->write_lock.unlock();
360 return nullptr;
361 }
362
363 connection->write_lock.unlock();
364
365 if (!(state >= START_CONNECT && state <= SESSION_RECONNECTING) &&
366 state != WAIT &&
367 state != SESSION_ACCEPTING /* due to connection race */) {
368 // policy maybe empty when state is in accept
369 if (connection->policy.server) {
370 ldout(cct, 1) << __func__ << " server, going to standby" << dendl;
371 state = STANDBY;
372 } else {
373 ldout(cct, 1) << __func__ << " initiating reconnect" << dendl;
374 connect_seq++;
375 global_seq = messenger->get_global_seq();
376 state = START_CONNECT;
377 pre_auth.enabled = true;
378 connection->state = AsyncConnection::STATE_CONNECTING;
379 }
380 backoff = utime_t();
381 connection->center->dispatch_event_external(connection->read_handler);
382 } else {
383 if (state == WAIT) {
384 backoff.set_from_double(cct->_conf->ms_max_backoff);
385 } else if (backoff == utime_t()) {
386 backoff.set_from_double(cct->_conf->ms_initial_backoff);
387 } else {
388 backoff += backoff;
389 if (backoff > cct->_conf->ms_max_backoff)
390 backoff.set_from_double(cct->_conf->ms_max_backoff);
391 }
392
393 if (server_cookie) {
394 connect_seq++;
395 }
396
397 global_seq = messenger->get_global_seq();
398 state = START_CONNECT;
399 pre_auth.enabled = true;
400 connection->state = AsyncConnection::STATE_CONNECTING;
401 ldout(cct, 1) << __func__ << " waiting " << backoff << dendl;
402 // woke up again;
403 connection->register_time_events.insert(
404 connection->center->create_time_event(backoff.to_nsec() / 1000,
405 connection->wakeup_handler));
406 }
407 return nullptr;
408}
409
410void ProtocolV2::prepare_send_message(uint64_t features,
411 Message *m) {
412 ldout(cct, 20) << __func__ << " m=" << *m << dendl;
413
414 // associate message with Connection (for benefit of encode_payload)
9f95a23c
TL		 415 ldout(cct, 20) << __func__ << (m->empty_payload() ? " encoding features " : " half-reencoding features ")
416 << features << " " << m << " " << *m << dendl;
11fdf7f2
TL		 417
418 // encode and copy out of *m
419 m->encode(features, 0);
420}
421
422void ProtocolV2::send_message(Message *m) {
423 uint64_t f = connection->get_features();
424
425 // TODO: Currently not all messages supports reencode like MOSDMap, so here
426 // only let fast dispatch support messages prepare message
427 const bool can_fast_prepare = messenger->ms_can_fast_dispatch(m);
428 if (can_fast_prepare) {
429 prepare_send_message(f, m);
430 }
431
432 std::lock_guard<std::mutex> l(connection->write_lock);
433 bool is_prepared = can_fast_prepare;
434 // "features" changes will change the payload encoding
435 if (can_fast_prepare && (!can_write || connection->get_features() != f)) {
436 // ensure the correctness of message encoding
437 m->clear_payload();
438 is_prepared = false;
439 ldout(cct, 10) << __func__ << " clear encoded buffer previous " << f
440 << " != " << connection->get_features() << dendl;
441 }
442 if (state == CLOSED) {
443 ldout(cct, 10) << __func__ << " connection closed."
444 << " Drop message " << m << dendl;
445 m->put();
446 } else {
447 ldout(cct, 5) << __func__ << " enqueueing message m=" << m
448 << " type=" << m->get_type() << " " << *m << dendl;
9f95a23c 449 m->queue_start = ceph::mono_clock::now();
11fdf7f2
TL		 450 m->trace.event("async enqueueing message");
451 out_queue[m->get_priority()].emplace_back(
452 out_queue_entry_t{is_prepared, m});
453 ldout(cct, 15) << __func__ << " inline write is denied, reschedule m=" << m
454 << dendl;
494da23a
TL		 455 if (((!replacing && can_write) || state == STANDBY) && !write_in_progress) {
456 write_in_progress = true;
11fdf7f2
TL		 457 connection->center->dispatch_event_external(connection->write_handler);
458 }
459 }
460}
461
462void ProtocolV2::send_keepalive() {
463 ldout(cct, 10) << __func__ << dendl;
464 std::lock_guard<std::mutex> l(connection->write_lock);
465 if (state != CLOSED) {
466 keepalive = true;
467 connection->center->dispatch_event_external(connection->write_handler);
468 }
469}
470
471void ProtocolV2::read_event() {
472 ldout(cct, 20) << __func__ << dendl;
473
474 switch (state) {
475 case START_CONNECT:
476 run_continuation(CONTINUATION(start_client_banner_exchange));
477 break;
478 case START_ACCEPT:
479 run_continuation(CONTINUATION(start_server_banner_exchange));
480 break;
481 case READY:
482 run_continuation(CONTINUATION(read_frame));
483 break;
484 case THROTTLE_MESSAGE:
485 run_continuation(CONTINUATION(throttle_message));
486 break;
487 case THROTTLE_BYTES:
488 run_continuation(CONTINUATION(throttle_bytes));
489 break;
490 case THROTTLE_DISPATCH_QUEUE:
491 run_continuation(CONTINUATION(throttle_dispatch_queue));
492 break;
493 default:
494 break;
495 }
496}
497
498ProtocolV2::out_queue_entry_t ProtocolV2::_get_next_outgoing() {
499 out_queue_entry_t out_entry;
500
501 if (!out_queue.empty()) {
502 auto it = out_queue.rbegin();
503 auto& entries = it->second;
504 ceph_assert(!entries.empty());
505 out_entry = entries.front();
506 entries.pop_front();
507 if (entries.empty()) {
508 out_queue.erase(it->first);
509 }
510 }
511 return out_entry;
512}
513
514ssize_t ProtocolV2::write_message(Message *m, bool more) {
515 FUNCTRACE(cct);
516 ceph_assert(connection->center->in_thread());
517 m->set_seq(++out_seq);
518
519 connection->lock.lock();
520 uint64_t ack_seq = in_seq;
521 ack_left = 0;
522 connection->lock.unlock();
523
524 ceph_msg_header &header = m->get_header();
525 ceph_msg_footer &footer = m->get_footer();
526
527 ceph_msg_header2 header2{header.seq, header.tid,
528 header.type, header.priority,
529 header.version,
eafe8130
TL		 530 init_le32(0), header.data_off,
531 init_le64(ack_seq),
11fdf7f2
TL		 532 footer.flags, header.compat_version,
533 header.reserved};
534
535 auto message = MessageFrame::Encode(
536 header2,
537 m->get_payload(),
538 m->get_middle(),
539 m->get_data());
801d1391
TL		 540 if (!append_frame(message)) {
541 m->put();
542 return -EILSEQ;
543 }
11fdf7f2
TL		 544
545 ldout(cct, 5) << __func__ << " sending message m=" << m
546 << " seq=" << m->get_seq() << " " << *m << dendl;
547
548 m->trace.event("async writing message");
549 ldout(cct, 20) << __func__ << " sending m=" << m << " seq=" << m->get_seq()
550 << " src=" << entity_name_t(messenger->get_myname())
551 << " off=" << header2.data_off
552 << dendl;
9f95a23c 553 ssize_t total_send_size = connection->outgoing_bl.length();
11fdf7f2
TL		 554 ssize_t rc = connection->_try_send(more);
555 if (rc < 0) {
556 ldout(cct, 1) << __func__ << " error sending " << m << ", "
557 << cpp_strerror(rc) << dendl;
558 } else {
559 connection->logger->inc(
9f95a23c 560 l_msgr_send_bytes, total_send_size - connection->outgoing_bl.length());
11fdf7f2
TL		 561 ldout(cct, 10) << __func__ << " sending " << m
562 << (rc ? " continuely." : " done.") << dendl;
563 }
9f95a23c
TL		 564
565#if defined(WITH_EVENTTRACE)
11fdf7f2
TL		 566 if (m->get_type() == CEPH_MSG_OSD_OP)
567 OID_EVENT_TRACE_WITH_MSG(m, "SEND_MSG_OSD_OP_END", false);
568 else if (m->get_type() == CEPH_MSG_OSD_OPREPLY)
569 OID_EVENT_TRACE_WITH_MSG(m, "SEND_MSG_OSD_OPREPLY_END", false);
9f95a23c 570#endif
11fdf7f2
TL		 571 m->put();
572
573 return rc;
574}
575
801d1391
TL		 576template <class F>
577bool ProtocolV2::append_frame(F& frame) {
578 ceph::bufferlist bl;
579 try {
f6b5b4d7 580 bl = frame.get_buffer(tx_frame_asm);
801d1391
TL		 581 } catch (ceph::crypto::onwire::TxHandlerError &e) {
582 ldout(cct, 1) << __func__ << " " << e.what() << dendl;
583 return false;
584 }
f6b5b4d7
TL		 585
586 ldout(cct, 25) << __func__ << " assembled frame " << bl.length()
587 << " bytes " << tx_frame_asm << dendl;
801d1391
TL		 588 connection->outgoing_bl.append(bl);
589 return true;
11fdf7f2
TL		 590}
591
592void ProtocolV2::handle_message_ack(uint64_t seq) {
593 if (connection->policy.lossy) { // lossy connections don't keep sent messages
594 return;
595 }
596
597 ldout(cct, 15) << __func__ << " seq=" << seq << dendl;
598
599 // trim sent list
600 static const int max_pending = 128;
601 int i = 0;
602 Message *pending[max_pending];
9f95a23c 603 auto now = ceph::mono_clock::now();
11fdf7f2
TL		 604 connection->write_lock.lock();
605 while (!sent.empty() && sent.front()->get_seq() <= seq && i < max_pending) {
606 Message *m = sent.front();
607 sent.pop_front();
608 pending[i++] = m;
609 ldout(cct, 10) << __func__ << " got ack seq " << seq
610 << " >= " << m->get_seq() << " on " << m << " " << *m
611 << dendl;
612 }
613 connection->write_lock.unlock();
9f95a23c 614 connection->logger->tinc(l_msgr_handle_ack_lat, ceph::mono_clock::now() - now);
11fdf7f2
TL		 615 for (int k = 0; k < i; k++) {
616 pending[k]->put();
617 }
618}
619
620void ProtocolV2::write_event() {
621 ldout(cct, 10) << __func__ << dendl;
622 ssize_t r = 0;
623
624 connection->write_lock.lock();
625 if (can_write) {
626 if (keepalive) {
801d1391
TL		 627 ldout(cct, 10) << __func__ << " appending keepalive" << dendl;
628 auto keepalive_frame = KeepAliveFrame::Encode();
629 if (!append_frame(keepalive_frame)) {
630 connection->write_lock.unlock();
631 connection->lock.lock();
632 fault();
633 connection->lock.unlock();
634 return;
635 }
11fdf7f2
TL		 636 keepalive = false;
637 }
638
639 auto start = ceph::mono_clock::now();
640 bool more;
641 do {
642 const auto out_entry = _get_next_outgoing();
643 if (!out_entry.m) {
644 break;
645 }
646
647 if (!connection->policy.lossy) {
648 // put on sent list
649 sent.push_back(out_entry.m);
650 out_entry.m->get();
651 }
652 more = !out_queue.empty();
653 connection->write_lock.unlock();
654
655 // send_message or requeue messages may not encode message
656 if (!out_entry.is_prepared) {
657 prepare_send_message(connection->get_features(), out_entry.m);
658 }
659
9f95a23c
TL		 660 if (out_entry.m->queue_start != ceph::mono_time()) {
661 connection->logger->tinc(l_msgr_send_messages_queue_lat,
662 ceph::mono_clock::now() -
663 out_entry.m->queue_start);
664 }
665
11fdf7f2
TL		 666 r = write_message(out_entry.m, more);
667
668 connection->write_lock.lock();
669 if (r == 0) {
670 ;
671 } else if (r < 0) {
672 ldout(cct, 1) << __func__ << " send msg failed" << dendl;
673 break;
9f95a23c
TL		 674 } else if (r > 0) {
675 // Outbound message in-progress, thread will be re-awoken
676 // when the outbound socket is writeable again
11fdf7f2 677 break;
9f95a23c 678 }
11fdf7f2 679 } while (can_write);
494da23a 680 write_in_progress = false;
11fdf7f2
TL		 681
682 // if r > 0 mean data still lefted, so no need _try_send.
683 if (r == 0) {
684 uint64_t left = ack_left;
685 if (left) {
11fdf7f2
TL		 686 ldout(cct, 10) << __func__ << " try send msg ack, acked " << left
687 << " messages" << dendl;
801d1391
TL		 688 auto ack_frame = AckFrame::Encode(in_seq);
689 if (append_frame(ack_frame)) {
690 ack_left -= left;
691 left = ack_left;
692 r = connection->_try_send(left);
693 } else {
694 r = -EILSEQ;
695 }
11fdf7f2
TL		 696 } else if (is_queued()) {
697 r = connection->_try_send();
698 }
699 }
700 connection->write_lock.unlock();
701
702 connection->logger->tinc(l_msgr_running_send_time,
703 ceph::mono_clock::now() - start);
704 if (r < 0) {
705 ldout(cct, 1) << __func__ << " send msg failed" << dendl;
706 connection->lock.lock();
707 fault();
708 connection->lock.unlock();
709 return;
710 }
711 } else {
494da23a 712 write_in_progress = false;
11fdf7f2
TL		 713 connection->write_lock.unlock();
714 connection->lock.lock();
715 connection->write_lock.lock();
716 if (state == STANDBY && !connection->policy.server && is_queued()) {
717 ldout(cct, 10) << __func__ << " policy.server is false" << dendl;
718 if (server_cookie) { // only increment connect_seq if there is a session
719 connect_seq++;
720 }
721 connection->_connect();
722 } else if (connection->cs && state != NONE && state != CLOSED &&
723 state != START_CONNECT) {
724 r = connection->_try_send();
725 if (r < 0) {
726 ldout(cct, 1) << __func__ << " send outcoming bl failed" << dendl;
727 connection->write_lock.unlock();
728 fault();
729 connection->lock.unlock();
730 return;
731 }
732 }
733 connection->write_lock.unlock();
734 connection->lock.unlock();
735 }
736}
737
738bool ProtocolV2::is_queued() {
739 return !out_queue.empty() || connection->is_queued();
740}
741
11fdf7f2
TL		 742CtPtr ProtocolV2::read(CONTINUATION_RXBPTR_TYPE<ProtocolV2> &next,
743 rx_buffer_t &&buffer) {
744 const auto len = buffer->length();
745 const auto buf = buffer->c_str();
746 next.node = std::move(buffer);
747 ssize_t r = connection->read(len, buf,
748 [&next, this](char *buffer, int r) {
749 if (unlikely(pre_auth.enabled) && r >= 0) {
750 pre_auth.rxbuf.append(*next.node);
751 ceph_assert(!cct->_conf->ms_die_on_bug ||
f91f0fd5 752 pre_auth.rxbuf.length() < 10000000);
11fdf7f2
TL		 753 }
754 next.r = r;
755 run_continuation(next);
756 });
757 if (r <= 0) {
758 // error or done synchronously
759 if (unlikely(pre_auth.enabled) && r >= 0) {
760 pre_auth.rxbuf.append(*next.node);
761 ceph_assert(!cct->_conf->ms_die_on_bug ||
f91f0fd5 762 pre_auth.rxbuf.length() < 10000000);
11fdf7f2
TL		 763 }
764 next.r = r;
765 return &next;
766 }
767
768 return nullptr;
769}
770
771template <class F>
772CtPtr ProtocolV2::write(const std::string &desc,
773 CONTINUATION_TYPE<ProtocolV2> &next,
774 F &frame) {
801d1391
TL		 775 ceph::bufferlist bl;
776 try {
f6b5b4d7 777 bl = frame.get_buffer(tx_frame_asm);
801d1391
TL		 778 } catch (ceph::crypto::onwire::TxHandlerError &e) {
779 ldout(cct, 1) << __func__ << " " << e.what() << dendl;
780 return _fault();
781 }
f6b5b4d7
TL		 782
783 ldout(cct, 25) << __func__ << " assembled frame " << bl.length()
784 << " bytes " << tx_frame_asm << dendl;
11fdf7f2
TL		 785 return write(desc, next, bl);
786}
787
788CtPtr ProtocolV2::write(const std::string &desc,
789 CONTINUATION_TYPE<ProtocolV2> &next,
790 bufferlist &buffer) {
791 if (unlikely(pre_auth.enabled)) {
792 pre_auth.txbuf.append(buffer);
793 ceph_assert(!cct->_conf->ms_die_on_bug ||
f91f0fd5 794 pre_auth.txbuf.length() < 10000000);
11fdf7f2
TL		 795 }
796
797 ssize_t r =
798 connection->write(buffer, [&next, desc, this](int r) {
799 if (r < 0) {
800 ldout(cct, 1) << __func__ << " " << desc << " write failed r=" << r
801 << " (" << cpp_strerror(r) << ")" << dendl;
802 connection->inject_delay();
803 _fault();
804 }
805 run_continuation(next);
806 });
807
808 if (r < 0) {
809 ldout(cct, 1) << __func__ << " " << desc << " write failed r=" << r
810 << " (" << cpp_strerror(r) << ")" << dendl;
811 return _fault();
812 } else if (r == 0) {
813 next.setParams();
814 return &next;
815 }
816
817 return nullptr;
818}
819
820CtPtr ProtocolV2::_banner_exchange(CtRef callback) {
821 ldout(cct, 20) << __func__ << dendl;
822 bannerExchangeCallback = &callback;
823
824 bufferlist banner_payload;
825 encode((uint64_t)CEPH_MSGR2_SUPPORTED_FEATURES, banner_payload, 0);
826 encode((uint64_t)CEPH_MSGR2_REQUIRED_FEATURES, banner_payload, 0);
827
828 bufferlist bl;
829 bl.append(CEPH_BANNER_V2_PREFIX, strlen(CEPH_BANNER_V2_PREFIX));
830 encode((uint16_t)banner_payload.length(), bl, 0);
831 bl.claim_append(banner_payload);
832
833 INTERCEPT(state == BANNER_CONNECTING ? 3 : 4);
834
835 return WRITE(bl, "banner", _wait_for_peer_banner);
836}
837
838CtPtr ProtocolV2::_wait_for_peer_banner() {
9f95a23c 839 unsigned banner_len = strlen(CEPH_BANNER_V2_PREFIX) + sizeof(ceph_le16);
11fdf7f2
TL		 840 return READ(banner_len, _handle_peer_banner);
841}
842
843CtPtr ProtocolV2::_handle_peer_banner(rx_buffer_t &&buffer, int r) {
844 ldout(cct, 20) << __func__ << " r=" << r << dendl;
845
846 if (r < 0) {
847 ldout(cct, 1) << __func__ << " read peer banner failed r=" << r << " ("
848 << cpp_strerror(r) << ")" << dendl;
849 return _fault();
850 }
851
852 unsigned banner_prefix_len = strlen(CEPH_BANNER_V2_PREFIX);
853
854 if (memcmp(buffer->c_str(), CEPH_BANNER_V2_PREFIX, banner_prefix_len)) {
855 if (memcmp(buffer->c_str(), CEPH_BANNER, strlen(CEPH_BANNER)) == 0) {
856 lderr(cct) << __func__ << " peer " << *connection->peer_addrs
857 << " is using msgr V1 protocol" << dendl;
858 return _fault();
859 }
860 ldout(cct, 1) << __func__ << " accept peer sent bad banner" << dendl;
861 return _fault();
862 }
863
864 uint16_t payload_len;
865 bufferlist bl;
866 buffer->set_offset(banner_prefix_len);
9f95a23c 867 buffer->set_length(sizeof(ceph_le16));
11fdf7f2
TL		 868 bl.push_back(std::move(buffer));
869 auto ti = bl.cbegin();
870 try {
871 decode(payload_len, ti);
872 } catch (const buffer::error &e) {
873 lderr(cct) << __func__ << " decode banner payload len failed " << dendl;
874 return _fault();
875 }
876
877 INTERCEPT(state == BANNER_CONNECTING ? 5 : 6);
878
879 return READ(payload_len, _handle_peer_banner_payload);
880}
881
882CtPtr ProtocolV2::_handle_peer_banner_payload(rx_buffer_t &&buffer, int r) {
883 ldout(cct, 20) << __func__ << " r=" << r << dendl;
884
885 if (r < 0) {
886 ldout(cct, 1) << __func__ << " read peer banner payload failed r=" << r
887 << " (" << cpp_strerror(r) << ")" << dendl;
888 return _fault();
889 }
890
891 uint64_t peer_supported_features;
892 uint64_t peer_required_features;
893
894 bufferlist bl;
895 bl.push_back(std::move(buffer));
896 auto ti = bl.cbegin();
897 try {
898 decode(peer_supported_features, ti);
899 decode(peer_required_features, ti);
900 } catch (const buffer::error &e) {
901 lderr(cct) << __func__ << " decode banner payload failed " << dendl;
902 return _fault();
903 }
904
905 ldout(cct, 1) << __func__ << " supported=" << std::hex
906 << peer_supported_features << " required=" << std::hex
907 << peer_required_features << std::dec << dendl;
908
909 // Check feature bit compatibility
910
911 uint64_t supported_features = CEPH_MSGR2_SUPPORTED_FEATURES;
912 uint64_t required_features = CEPH_MSGR2_REQUIRED_FEATURES;
913
914 if ((required_features & peer_supported_features) != required_features) {
915 ldout(cct, 1) << __func__ << " peer does not support all required features"
916 << " required=" << std::hex << required_features
917 << " supported=" << std::hex << peer_supported_features
918 << std::dec << dendl;
919 stop();
920 connection->dispatch_queue->queue_reset(connection);
921 return nullptr;
922 }
923 if ((supported_features & peer_required_features) != peer_required_features) {
924 ldout(cct, 1) << __func__ << " we do not support all peer required features"
925 << " required=" << std::hex << peer_required_features
926 << " supported=" << supported_features << std::dec << dendl;
927 stop();
928 connection->dispatch_queue->queue_reset(connection);
929 return nullptr;
930 }
931
f6b5b4d7
TL		 932 this->peer_supported_features = peer_supported_features;
933 if (peer_required_features == 0) {
11fdf7f2
TL		 934 this->connection_features = msgr2_required;
935 }
936
f6b5b4d7
TL		 937 // if the peer supports msgr2.1, switch to it
938 bool is_rev1 = HAVE_MSGR2_FEATURE(peer_supported_features, REVISION_1);
939 tx_frame_asm.set_is_rev1(is_rev1);
940 rx_frame_asm.set_is_rev1(is_rev1);
11fdf7f2
TL		 941
942 if (state == BANNER_CONNECTING) {
943 state = HELLO_CONNECTING;
944 }
945 else {
946 ceph_assert(state == BANNER_ACCEPTING);
947 state = HELLO_ACCEPTING;
948 }
949
950 auto hello = HelloFrame::Encode(messenger->get_mytype(),
951 connection->target_addr);
952
953 INTERCEPT(state == HELLO_CONNECTING ? 7 : 8);
954
955 return WRITE(hello, "hello frame", read_frame);
956}
957
958CtPtr ProtocolV2::handle_hello(ceph::bufferlist &payload)
959{
960 ldout(cct, 20) << __func__
961 << " payload.length()=" << payload.length() << dendl;
962
963 if (state != HELLO_CONNECTING && state != HELLO_ACCEPTING) {
964 lderr(cct) << __func__ << " not in hello exchange state!" << dendl;
965 return _fault();
966 }
967
968 auto hello = HelloFrame::Decode(payload);
969
970 ldout(cct, 5) << __func__ << " received hello:"
971 << " peer_type=" << (int)hello.entity_type()
972 << " peer_addr_for_me=" << hello.peer_addr() << dendl;
973
81eedcae
TL		 974 sockaddr_storage ss;
975 socklen_t len = sizeof(ss);
976 getsockname(connection->cs.fd(), (sockaddr *)&ss, &len);
977 ldout(cct, 5) << __func__ << " getsockname says I am " << (sockaddr *)&ss
978 << " when talking to " << connection->target_addr << dendl;
979
11fdf7f2
TL		 980 if (connection->get_peer_type() == -1) {
981 connection->set_peer_type(hello.entity_type());
982
983 ceph_assert(state == HELLO_ACCEPTING);
984 connection->policy = messenger->get_policy(hello.entity_type());
985 ldout(cct, 10) << __func__ << " accept of host_type "
986 << (int)hello.entity_type()
987 << ", policy.lossy=" << connection->policy.lossy
988 << " policy.server=" << connection->policy.server
989 << " policy.standby=" << connection->policy.standby
990 << " policy.resetcheck=" << connection->policy.resetcheck
991 << dendl;
992 } else {
993 ceph_assert(state == HELLO_CONNECTING);
994 if (connection->get_peer_type() != hello.entity_type()) {
995 ldout(cct, 1) << __func__ << " connection peer type does not match what"
996 << " peer advertises " << connection->get_peer_type()
997 << " != " << (int)hello.entity_type() << dendl;
998 stop();
999 connection->dispatch_queue->queue_reset(connection);
1000 return nullptr;
1001 }
1002 }
1003
81eedcae
TL		 1004 if (messenger->get_myaddrs().empty() ||
1005 messenger->get_myaddrs().front().is_blank_ip()) {
1006 entity_addr_t a;
1007 if (cct->_conf->ms_learn_addr_from_peer) {
1008 ldout(cct, 1) << __func__ << " peer " << connection->target_addr
1009 << " says I am " << hello.peer_addr() << " (socket says "
1010 << (sockaddr*)&ss << ")" << dendl;
1011 a = hello.peer_addr();
1012 } else {
1013 ldout(cct, 1) << __func__ << " socket to " << connection->target_addr
1014 << " says I am " << (sockaddr*)&ss
1015 << " (peer says " << hello.peer_addr() << ")" << dendl;
1016 a.set_sockaddr((sockaddr *)&ss);
1017 }
1018 a.set_type(entity_addr_t::TYPE_MSGR2); // anything but NONE; learned_addr ignores this
1019 a.set_port(0);
1020 connection->lock.unlock();
1021 messenger->learned_addr(a);
1022 if (cct->_conf->ms_inject_internal_delays &&
1023 cct->_conf->ms_inject_socket_failures) {
1024 if (rand() % cct->_conf->ms_inject_socket_failures == 0) {
1025 ldout(cct, 10) << __func__ << " sleep for "
1026 << cct->_conf->ms_inject_internal_delays << dendl;
1027 utime_t t;
1028 t.set_from_double(cct->_conf->ms_inject_internal_delays);
1029 t.sleep();
1030 }
1031 }
1032 connection->lock.lock();
1033 if (state != HELLO_CONNECTING) {
1034 ldout(cct, 1) << __func__
1035 << " state changed while learned_addr, mark_down or "
1036 << " replacing must be happened just now" << dendl;
1037 return nullptr;
1038 }
1039 }
1040
1041
1042
11fdf7f2
TL		 1043 CtPtr callback;
1044 callback = bannerExchangeCallback;
1045 bannerExchangeCallback = nullptr;
1046 ceph_assert(callback);
1047 return callback;
1048}
1049
1050CtPtr ProtocolV2::read_frame() {
1051 if (state == CLOSED) {
1052 return nullptr;
1053 }
1054
1055 ldout(cct, 20) << __func__ << dendl;
f6b5b4d7
TL		 1056 rx_preamble.clear();
1057 rx_epilogue.clear();
1058 rx_segments_data.clear();
1059
1060 return READ(rx_frame_asm.get_preamble_onwire_len(),
1061 handle_read_frame_preamble_main);
11fdf7f2
TL		 1062}
1063
1064CtPtr ProtocolV2::handle_read_frame_preamble_main(rx_buffer_t &&buffer, int r) {
1065 ldout(cct, 20) << __func__ << " r=" << r << dendl;
1066
1067 if (r < 0) {
f6b5b4d7 1068 ldout(cct, 1) << __func__ << " read frame preamble failed r=" << r
11fdf7f2
TL		 1069 << " (" << cpp_strerror(r) << ")" << dendl;
1070 return _fault();
1071 }
1072
f6b5b4d7 1073 rx_preamble.push_back(std::move(buffer));
11fdf7f2
TL		 1074
1075 ldout(cct, 30) << __func__ << " preamble\n";
f6b5b4d7 1076 rx_preamble.hexdump(*_dout);
11fdf7f2
TL		 1077 *_dout << dendl;
1078
f6b5b4d7
TL		 1079 try {
1080 next_tag = rx_frame_asm.disassemble_preamble(rx_preamble);
1081 } catch (FrameError& e) {
1082 ldout(cct, 1) << __func__ << " " << e.what() << dendl;
1083 return _fault();
1084 } catch (ceph::crypto::onwire::MsgAuthError&) {
1085 ldout(cct, 1) << __func__ << "bad auth tag" << dendl;
1086 return _fault();
1087 }
11fdf7f2 1088
f6b5b4d7
TL		 1089 ldout(cct, 25) << __func__ << " disassembled preamble " << rx_frame_asm
1090 << dendl;
11fdf7f2 1091
f6b5b4d7 1092 if (session_stream_handlers.rx) {
11fdf7f2 1093 ldout(cct, 30) << __func__ << " preamble after decrypt\n";
f6b5b4d7 1094 rx_preamble.hexdump(*_dout);
11fdf7f2
TL		 1095 *_dout << dendl;
1096 }
1097
11fdf7f2
TL		 1098 // does it need throttle?
1099 if (next_tag == Tag::MESSAGE) {
1100 if (state != READY) {
1101 lderr(cct) << __func__ << " not in ready state!" << dendl;
1102 return _fault();
1103 }
1104 state = THROTTLE_MESSAGE;
1105 return CONTINUE(throttle_message);
1106 } else {
1107 return read_frame_segment();
1108 }
1109}
1110
1111CtPtr ProtocolV2::handle_read_frame_dispatch() {
1112 ldout(cct, 10) << __func__
1113 << " tag=" << static_cast<uint32_t>(next_tag) << dendl;
1114
1115 switch (next_tag) {
1116 case Tag::HELLO:
1117 case Tag::AUTH_REQUEST:
1118 case Tag::AUTH_BAD_METHOD:
1119 case Tag::AUTH_REPLY_MORE:
1120 case Tag::AUTH_REQUEST_MORE:
1121 case Tag::AUTH_DONE:
1122 case Tag::AUTH_SIGNATURE:
1123 case Tag::CLIENT_IDENT:
1124 case Tag::SERVER_IDENT:
1125 case Tag::IDENT_MISSING_FEATURES:
1126 case Tag::SESSION_RECONNECT:
1127 case Tag::SESSION_RESET:
1128 case Tag::SESSION_RETRY:
1129 case Tag::SESSION_RETRY_GLOBAL:
1130 case Tag::SESSION_RECONNECT_OK:
1131 case Tag::KEEPALIVE2:
1132 case Tag::KEEPALIVE2_ACK:
1133 case Tag::ACK:
1134 case Tag::WAIT:
1135 return handle_frame_payload();
1136 case Tag::MESSAGE:
1137 return handle_message();
1138 default: {
1139 lderr(cct) << __func__
1140 << " received unknown tag=" << static_cast<uint32_t>(next_tag)
1141 << dendl;
1142 return _fault();
1143 }
1144 }
1145
1146 return nullptr;
1147}
1148
1149CtPtr ProtocolV2::read_frame_segment() {
f6b5b4d7
TL		 1150 size_t seg_idx = rx_segments_data.size();
1151 ldout(cct, 20) << __func__ << " seg_idx=" << seg_idx << dendl;
1152 rx_segments_data.emplace_back();
1153
1154 uint32_t onwire_len = rx_frame_asm.get_segment_onwire_len(seg_idx);
1155 if (onwire_len == 0) {
1156 return _handle_read_frame_segment();
1157 }
11fdf7f2 1158
11fdf7f2 1159 rx_buffer_t rx_buffer;
f6b5b4d7 1160 uint16_t align = rx_frame_asm.get_segment_align(seg_idx);
11fdf7f2
TL		 1161 try {
1162 rx_buffer = buffer::ptr_node::create(buffer::create_aligned(
f6b5b4d7 1163 onwire_len, align));
11fdf7f2
TL		 1164 } catch (std::bad_alloc&) {
1165 // Catching because of potential issues with satisfying alignment.
f6b5b4d7
TL		 1166 ldout(cct, 1) << __func__ << " can't allocate aligned rx_buffer"
1167 << " len=" << onwire_len
1168 << " align=" << align
1169 << dendl;
11fdf7f2
TL		 1170 return _fault();
1171 }
1172
1173 return READ_RXBUF(std::move(rx_buffer), handle_read_frame_segment);
1174}
1175
1176CtPtr ProtocolV2::handle_read_frame_segment(rx_buffer_t &&rx_buffer, int r) {
1177 ldout(cct, 20) << __func__ << " r=" << r << dendl;
1178
1179 if (r < 0) {
1180 ldout(cct, 1) << __func__ << " read frame segment failed r=" << r << " ("
1181 << cpp_strerror(r) << ")" << dendl;
1182 return _fault();
1183 }
1184
11fdf7f2 1185 rx_segments_data.back().push_back(std::move(rx_buffer));
f6b5b4d7
TL		 1186 return _handle_read_frame_segment();
1187}
11fdf7f2 1188
f6b5b4d7
TL		 1189CtPtr ProtocolV2::_handle_read_frame_segment() {
1190 if (rx_segments_data.size() == rx_frame_asm.get_num_segments()) {
11fdf7f2 1191 // OK, all segments planned to read are read. Can go with epilogue.
f6b5b4d7
TL		 1192 uint32_t epilogue_onwire_len = rx_frame_asm.get_epilogue_onwire_len();
1193 if (epilogue_onwire_len == 0) {
1194 return _handle_read_frame_epilogue_main();
1195 }
1196 return READ(epilogue_onwire_len, handle_read_frame_epilogue_main);
11fdf7f2 1197 }
f6b5b4d7
TL		 1198 // TODO: for makeshift only. This will be more generic and throttled
1199 return read_frame_segment();
11fdf7f2
TL		 1200}
1201
1202CtPtr ProtocolV2::handle_frame_payload() {
1203 ceph_assert(!rx_segments_data.empty());
1204 auto& payload = rx_segments_data.back();
1205
1206 ldout(cct, 30) << __func__ << "\n";
1207 payload.hexdump(*_dout);
1208 *_dout << dendl;
1209
1210 switch (next_tag) {
1211 case Tag::HELLO:
1212 return handle_hello(payload);
1213 case Tag::AUTH_REQUEST:
1214 return handle_auth_request(payload);
1215 case Tag::AUTH_BAD_METHOD:
1216 return handle_auth_bad_method(payload);
1217 case Tag::AUTH_REPLY_MORE:
1218 return handle_auth_reply_more(payload);
1219 case Tag::AUTH_REQUEST_MORE:
1220 return handle_auth_request_more(payload);
1221 case Tag::AUTH_DONE:
1222 return handle_auth_done(payload);
1223 case Tag::AUTH_SIGNATURE:
1224 return handle_auth_signature(payload);
1225 case Tag::CLIENT_IDENT:
1226 return handle_client_ident(payload);
1227 case Tag::SERVER_IDENT:
1228 return handle_server_ident(payload);
1229 case Tag::IDENT_MISSING_FEATURES:
1230 return handle_ident_missing_features(payload);
1231 case Tag::SESSION_RECONNECT:
1232 return handle_reconnect(payload);
1233 case Tag::SESSION_RESET:
1234 return handle_session_reset(payload);
1235 case Tag::SESSION_RETRY:
1236 return handle_session_retry(payload);
1237 case Tag::SESSION_RETRY_GLOBAL:
1238 return handle_session_retry_global(payload);
1239 case Tag::SESSION_RECONNECT_OK:
1240 return handle_reconnect_ok(payload);
1241 case Tag::KEEPALIVE2:
1242 return handle_keepalive2(payload);
1243 case Tag::KEEPALIVE2_ACK:
1244 return handle_keepalive2_ack(payload);
1245 case Tag::ACK:
1246 return handle_message_ack(payload);
1247 case Tag::WAIT:
1248 return handle_wait(payload);
1249 default:
1250 ceph_abort();
1251 }
1252 return nullptr;
1253}
1254
1255CtPtr ProtocolV2::ready() {
1256 ldout(cct, 25) << __func__ << dendl;
1257
1258 reconnecting = false;
1259 replacing = false;
1260
1261 // make sure no pending tick timer
1262 if (connection->last_tick_id) {
1263 connection->center->delete_time_event(connection->last_tick_id);
1264 }
1265 connection->last_tick_id = connection->center->create_time_event(
1266 connection->inactive_timeout_us, connection->tick_handler);
1267
1268 {
1269 std::lock_guard<std::mutex> l(connection->write_lock);
1270 can_write = true;
1271 if (!out_queue.empty()) {
1272 connection->center->dispatch_event_external(connection->write_handler);
1273 }
1274 }
1275
1276 connection->maybe_start_delay_thread();
1277
1278 state = READY;
1279 ldout(cct, 1) << __func__ << " entity=" << peer_name << " client_cookie="
1280 << std::hex << client_cookie << " server_cookie="
1281 << server_cookie << std::dec << " in_seq=" << in_seq
1282 << " out_seq=" << out_seq << dendl;
1283
1284 INTERCEPT(15);
1285
1286 return CONTINUE(read_frame);
1287}
1288
1289CtPtr ProtocolV2::handle_read_frame_epilogue_main(rx_buffer_t &&buffer, int r)
1290{
1291 ldout(cct, 20) << __func__ << " r=" << r << dendl;
1292
1293 if (r < 0) {
f6b5b4d7
TL		 1294 ldout(cct, 1) << __func__ << " read frame epilogue failed r=" << r
1295 << " (" << cpp_strerror(r) << ")" << dendl;
11fdf7f2
TL		 1296 return _fault();
1297 }
1298
f6b5b4d7
TL		 1299 rx_epilogue.push_back(std::move(buffer));
1300 return _handle_read_frame_epilogue_main();
1301}
11fdf7f2 1302
f6b5b4d7
TL		 1303CtPtr ProtocolV2::_handle_read_frame_epilogue_main() {
1304 bool aborted;
1305 try {
1306 rx_frame_asm.disassemble_first_segment(rx_preamble, rx_segments_data[0]);
1307 aborted = !rx_frame_asm.disassemble_remaining_segments(
1308 rx_segments_data.data(), rx_epilogue);
1309 } catch (FrameError& e) {
1310 ldout(cct, 1) << __func__ << " " << e.what() << dendl;
1311 return _fault();
1312 } catch (ceph::crypto::onwire::MsgAuthError&) {
1313 ldout(cct, 1) << __func__ << "bad auth tag" << dendl;
1314 return _fault();
11fdf7f2
TL		 1315 }
1316
1317 // we do have a mechanism that allows transmitter to start sending message
1318 // and abort after putting entire data field on wire. This will be used by
1319 // the kernel client to avoid unnecessary buffering.
f6b5b4d7 1320 if (aborted) {
11fdf7f2
TL		 1321 reset_throttle();
1322 state = READY;
1323 return CONTINUE(read_frame);
11fdf7f2 1324 }
f6b5b4d7 1325 return handle_read_frame_dispatch();
11fdf7f2
TL		 1326}
1327
1328CtPtr ProtocolV2::handle_message() {
1329 ldout(cct, 20) << __func__ << dendl;
1330 ceph_assert(state == THROTTLE_DONE);
1331
9f95a23c
TL		 1332#if defined(WITH_EVENTTRACE)
1333 utime_t ltt_recv_stamp = ceph_clock_now();
11fdf7f2
TL		 1334#endif
1335 recv_stamp = ceph_clock_now();
1336
11fdf7f2 1337 const size_t cur_msg_size = get_current_msg_size();
f6b5b4d7 1338 auto msg_frame = MessageFrame::Decode(rx_segments_data);
11fdf7f2
TL		 1339
1340 // XXX: paranoid copy just to avoid oops
1341 ceph_msg_header2 current_header = msg_frame.header();
1342
1343 ldout(cct, 5) << __func__
1344 << " got " << msg_frame.front_len()
1345 << " + " << msg_frame.middle_len()
1346 << " + " << msg_frame.data_len()
1347 << " byte message."
1348 << " envelope type=" << current_header.type
1349 << " src " << peer_name
1350 << " off " << current_header.data_off
1351 << dendl;
1352
1353 INTERCEPT(16);
1354 ceph_msg_header header{current_header.seq,
1355 current_header.tid,
1356 current_header.type,
1357 current_header.priority,
1358 current_header.version,
eafe8130
TL		 1359 init_le32(msg_frame.front_len()),
1360 init_le32(msg_frame.middle_len()),
1361 init_le32(msg_frame.data_len()),
11fdf7f2
TL		 1362 current_header.data_off,
1363 peer_name,
1364 current_header.compat_version,
1365 current_header.reserved,
eafe8130
TL		 1366 init_le32(0)};
1367 ceph_msg_footer footer{init_le32(0), init_le32(0),
1368 init_le32(0), init_le64(0), current_header.flags};
11fdf7f2
TL		 1369
1370 Message *message = decode_message(cct, 0, header, footer,
1371 msg_frame.front(),
1372 msg_frame.middle(),
1373 msg_frame.data(),
1374 connection);
1375 if (!message) {
1376 ldout(cct, 1) << __func__ << " decode message failed " << dendl;
1377 return _fault();
1378 } else {
1379 state = READ_MESSAGE_COMPLETE;
1380 }
1381
1382 INTERCEPT(17);
1383
1384 message->set_byte_throttler(connection->policy.throttler_bytes);
1385 message->set_message_throttler(connection->policy.throttler_messages);
1386
1387 // store reservation size in message, so we don't get confused
1388 // by messages entering the dispatch queue through other paths.
1389 message->set_dispatch_throttle_size(cur_msg_size);
1390
1391 message->set_recv_stamp(recv_stamp);
1392 message->set_throttle_stamp(throttle_stamp);
1393 message->set_recv_complete_stamp(ceph_clock_now());
1394
1395 // check received seq#. if it is old, drop the message.
1396 // note that incoming messages may skip ahead. this is convenient for the
1397 // client side queueing because messages can't be renumbered, but the (kernel)
1398 // client will occasionally pull a message out of the sent queue to send
1399 // elsewhere. in that case it doesn't matter if we "got" it or not.
1400 uint64_t cur_seq = in_seq;
1401 if (message->get_seq() <= cur_seq) {
1402 ldout(cct, 0) << __func__ << " got old message " << message->get_seq()
1403 << " <= " << cur_seq << " " << message << " " << *message
1404 << ", discarding" << dendl;
1405 message->put();
1406 if (connection->has_feature(CEPH_FEATURE_RECONNECT_SEQ) &&
1407 cct->_conf->ms_die_on_old_message) {
1408 ceph_assert(0 == "old msgs despite reconnect_seq feature");
1409 }
1410 return nullptr;
1411 }
1412 if (message->get_seq() > cur_seq + 1) {
1413 ldout(cct, 0) << __func__ << " missed message? skipped from seq "
1414 << cur_seq << " to " << message->get_seq() << dendl;
1415 if (cct->_conf->ms_die_on_skipped_message) {
1416 ceph_assert(0 == "skipped incoming seq");
1417 }
1418 }
1419
9f95a23c 1420#if defined(WITH_EVENTTRACE)
11fdf7f2
TL		 1421 if (message->get_type() == CEPH_MSG_OSD_OP ||
1422 message->get_type() == CEPH_MSG_OSD_OPREPLY) {
1423 utime_t ltt_processed_stamp = ceph_clock_now();
1424 double usecs_elapsed =
1425 (ltt_processed_stamp.to_nsec() - ltt_recv_stamp.to_nsec()) / 1000;
1426 ostringstream buf;
1427 if (message->get_type() == CEPH_MSG_OSD_OP)
1428 OID_ELAPSED_WITH_MSG(message, usecs_elapsed, "TIME_TO_DECODE_OSD_OP",
1429 false);
1430 else
1431 OID_ELAPSED_WITH_MSG(message, usecs_elapsed, "TIME_TO_DECODE_OSD_OPREPLY",
1432 false);
1433 }
1434#endif
1435
1436 // note last received message.
1437 in_seq = message->get_seq();
1438 ldout(cct, 5) << __func__ << " received message m=" << message
1439 << " seq=" << message->get_seq()
1440 << " from=" << message->get_source() << " type=" << header.type
1441 << " " << *message << dendl;
1442
1443 bool need_dispatch_writer = false;
1444 if (!connection->policy.lossy) {
1445 ack_left++;
1446 need_dispatch_writer = true;
1447 }
1448
1449 state = READY;
1450
9f95a23c
TL		 1451 ceph::mono_time fast_dispatch_time;
1452
1453 if (connection->is_blackhole()) {
1454 ldout(cct, 10) << __func__ << " blackhole " << *message << dendl;
1455 message->put();
1456 goto out;
1457 }
1458
11fdf7f2 1459 connection->logger->inc(l_msgr_recv_messages);
f6b5b4d7
TL		 1460 connection->logger->inc(l_msgr_recv_bytes,
1461 rx_frame_asm.get_frame_onwire_len());
11fdf7f2
TL		 1462
1463 messenger->ms_fast_preprocess(message);
9f95a23c 1464 fast_dispatch_time = ceph::mono_clock::now();
11fdf7f2 1465 connection->logger->tinc(l_msgr_running_recv_time,
9f95a23c 1466 fast_dispatch_time - connection->recv_start_time);
11fdf7f2
TL		 1467 if (connection->delay_state) {
1468 double delay_period = 0;
1469 if (rand() % 10000 < cct->_conf->ms_inject_delay_probability * 10000.0) {
1470 delay_period =
1471 cct->_conf->ms_inject_delay_max * (double)(rand() % 10000) / 10000.0;
1472 ldout(cct, 1) << "queue_received will delay after "
1473 << (ceph_clock_now() + delay_period) << " on " << message
1474 << " " << *message << dendl;
1475 }
1476 connection->delay_state->queue(delay_period, message);
1477 } else if (messenger->ms_can_fast_dispatch(message)) {
1478 connection->lock.unlock();
1479 connection->dispatch_queue->fast_dispatch(message);
1480 connection->recv_start_time = ceph::mono_clock::now();
1481 connection->logger->tinc(l_msgr_running_fast_dispatch_time,
1482 connection->recv_start_time - fast_dispatch_time);
1483 connection->lock.lock();
9f95a23c
TL		 1484 // we might have been reused by another connection
1485 // let's check if that is the case
1486 if (state != READY) {
1487 // yes, that was the case, let's do nothing
1488 return nullptr;
1489 }
11fdf7f2
TL		 1490 } else {
1491 connection->dispatch_queue->enqueue(message, message->get_priority(),
1492 connection->conn_id);
1493 }
1494
1495 handle_message_ack(current_header.ack_seq);
1496
9f95a23c 1497 out:
11fdf7f2
TL		 1498 if (need_dispatch_writer && connection->is_connected()) {
1499 connection->center->dispatch_event_external(connection->write_handler);
1500 }
1501
1502 return CONTINUE(read_frame);
1503}
1504
1505
1506CtPtr ProtocolV2::throttle_message() {
1507 ldout(cct, 20) << __func__ << dendl;
1508
1509 if (connection->policy.throttler_messages) {
1510 ldout(cct, 10) << __func__ << " wants " << 1
1511 << " message from policy throttler "
1512 << connection->policy.throttler_messages->get_current()
1513 << "/" << connection->policy.throttler_messages->get_max()
1514 << dendl;
1515 if (!connection->policy.throttler_messages->get_or_fail()) {
1516 ldout(cct, 10) << __func__ << " wants 1 message from policy throttle "
1517 << connection->policy.throttler_messages->get_current()
1518 << "/" << connection->policy.throttler_messages->get_max()
1519 << " failed, just wait." << dendl;
1520 // following thread pool deal with th full message queue isn't a
1521 // short time, so we can wait a ms.
1522 if (connection->register_time_events.empty()) {
1523 connection->register_time_events.insert(
1524 connection->center->create_time_event(1000,
1525 connection->wakeup_handler));
1526 }
1527 return nullptr;
1528 }
1529 }
1530
1531 state = THROTTLE_BYTES;
1532 return CONTINUE(throttle_bytes);
1533}
1534
1535CtPtr ProtocolV2::throttle_bytes() {
1536 ldout(cct, 20) << __func__ << dendl;
1537
1538 const size_t cur_msg_size = get_current_msg_size();
1539 if (cur_msg_size) {
1540 if (connection->policy.throttler_bytes) {
1541 ldout(cct, 10) << __func__ << " wants " << cur_msg_size
1542 << " bytes from policy throttler "
1543 << connection->policy.throttler_bytes->get_current() << "/"
1544 << connection->policy.throttler_bytes->get_max() << dendl;
1545 if (!connection->policy.throttler_bytes->get_or_fail(cur_msg_size)) {
1546 ldout(cct, 10) << __func__ << " wants " << cur_msg_size
1547 << " bytes from policy throttler "
1548 << connection->policy.throttler_bytes->get_current()
1549 << "/" << connection->policy.throttler_bytes->get_max()
1550 << " failed, just wait." << dendl;
1551 // following thread pool deal with th full message queue isn't a
1552 // short time, so we can wait a ms.
1553 if (connection->register_time_events.empty()) {
1554 connection->register_time_events.insert(
1555 connection->center->create_time_event(
1556 1000, connection->wakeup_handler));
1557 }
1558 return nullptr;
1559 }
1560 }
1561 }
1562
1563 state = THROTTLE_DISPATCH_QUEUE;
1564 return CONTINUE(throttle_dispatch_queue);
1565}
1566
1567CtPtr ProtocolV2::throttle_dispatch_queue() {
1568 ldout(cct, 20) << __func__ << dendl;
1569
1570 const size_t cur_msg_size = get_current_msg_size();
1571 if (cur_msg_size) {
1572 if (!connection->dispatch_queue->dispatch_throttler.get_or_fail(
1573 cur_msg_size)) {
1574 ldout(cct, 10)
1575 << __func__ << " wants " << cur_msg_size
1576 << " bytes from dispatch throttle "
1577 << connection->dispatch_queue->dispatch_throttler.get_current() << "/"
1578 << connection->dispatch_queue->dispatch_throttler.get_max()
1579 << " failed, just wait." << dendl;
1580 // following thread pool deal with th full message queue isn't a
1581 // short time, so we can wait a ms.
1582 if (connection->register_time_events.empty()) {
1583 connection->register_time_events.insert(
1584 connection->center->create_time_event(1000,
1585 connection->wakeup_handler));
1586 }
1587 return nullptr;
1588 }
1589 }
1590
1591 throttle_stamp = ceph_clock_now();
1592 state = THROTTLE_DONE;
1593
1594 return read_frame_segment();
1595}
1596
1597CtPtr ProtocolV2::handle_keepalive2(ceph::bufferlist &payload)
1598{
1599 ldout(cct, 20) << __func__
1600 << " payload.length()=" << payload.length() << dendl;
1601
1602 if (state != READY) {
1603 lderr(cct) << __func__ << " not in ready state!" << dendl;
1604 return _fault();
1605 }
1606
1607 auto keepalive_frame = KeepAliveFrame::Decode(payload);
1608
1609 ldout(cct, 30) << __func__ << " got KEEPALIVE2 tag ..." << dendl;
1610
1611 connection->write_lock.lock();
801d1391
TL		 1612 auto keepalive_ack_frame = KeepAliveFrameAck::Encode(keepalive_frame.timestamp());
1613 if (!append_frame(keepalive_ack_frame)) {
1614 connection->write_lock.unlock();
1615 return _fault();
1616 }
11fdf7f2
TL		 1617 connection->write_lock.unlock();
1618
1619 ldout(cct, 20) << __func__ << " got KEEPALIVE2 "
1620 << keepalive_frame.timestamp() << dendl;
1621 connection->set_last_keepalive(ceph_clock_now());
1622
1623 if (is_connected()) {
1624 connection->center->dispatch_event_external(connection->write_handler);
1625 }
1626
1627 return CONTINUE(read_frame);
1628}
1629
1630CtPtr ProtocolV2::handle_keepalive2_ack(ceph::bufferlist &payload)
1631{
1632 ldout(cct, 20) << __func__
1633 << " payload.length()=" << payload.length() << dendl;
1634
1635 if (state != READY) {
1636 lderr(cct) << __func__ << " not in ready state!" << dendl;
1637 return _fault();
1638 }
1639
1640 auto keepalive_ack_frame = KeepAliveFrameAck::Decode(payload);
1641 connection->set_last_keepalive_ack(keepalive_ack_frame.timestamp());
1642 ldout(cct, 20) << __func__ << " got KEEPALIVE_ACK" << dendl;
1643
1644 return CONTINUE(read_frame);
1645}
1646
1647CtPtr ProtocolV2::handle_message_ack(ceph::bufferlist &payload)
1648{
1649 ldout(cct, 20) << __func__
1650 << " payload.length()=" << payload.length() << dendl;
1651
1652 if (state != READY) {
1653 lderr(cct) << __func__ << " not in ready state!" << dendl;
1654 return _fault();
1655 }
1656
1657 auto ack = AckFrame::Decode(payload);
1658 handle_message_ack(ack.seq());
1659 return CONTINUE(read_frame);
1660}
1661
1662/* Client Protocol Methods */
1663
1664CtPtr ProtocolV2::start_client_banner_exchange() {
1665 ldout(cct, 20) << __func__ << dendl;
1666
1667 INTERCEPT(1);
1668
1669 state = BANNER_CONNECTING;
1670
1671 global_seq = messenger->get_global_seq();
1672
1673 return _banner_exchange(CONTINUATION(post_client_banner_exchange));
1674}
1675
1676CtPtr ProtocolV2::post_client_banner_exchange() {
1677 ldout(cct, 20) << __func__ << dendl;
1678
1679 state = AUTH_CONNECTING;
1680
1681 return send_auth_request();
1682}
1683
1684CtPtr ProtocolV2::send_auth_request(std::vector<uint32_t> &allowed_methods) {
9f95a23c 1685 ceph_assert(messenger->auth_client);
11fdf7f2
TL		 1686 ldout(cct, 20) << __func__ << " peer_type " << (int)connection->peer_type
1687 << " auth_client " << messenger->auth_client << dendl;
11fdf7f2
TL		 1688
1689 bufferlist bl;
1690 vector<uint32_t> preferred_modes;
1691 auto am = auth_meta;
1692 connection->lock.unlock();
1693 int r = messenger->auth_client->get_auth_request(
1694 connection, am.get(),
1695 &am->auth_method, &preferred_modes, &bl);
1696 connection->lock.lock();
1697 if (state != AUTH_CONNECTING) {
1698 ldout(cct, 1) << __func__ << " state changed!" << dendl;
1699 return _fault();
1700 }
1701 if (r < 0) {
1702 ldout(cct, 0) << __func__ << " get_initial_auth_request returned " << r
1703 << dendl;
1704 stop();
1705 connection->dispatch_queue->queue_reset(connection);
1706 return nullptr;
1707 }
1708
1709 INTERCEPT(9);
1710
1711 auto frame = AuthRequestFrame::Encode(auth_meta->auth_method, preferred_modes,
1712 bl);
1713 return WRITE(frame, "auth request", read_frame);
1714}
1715
1716CtPtr ProtocolV2::handle_auth_bad_method(ceph::bufferlist &payload) {
1717 ldout(cct, 20) << __func__
1718 << " payload.length()=" << payload.length() << dendl;
1719
1720 if (state != AUTH_CONNECTING) {
1721 lderr(cct) << __func__ << " not in auth connect state!" << dendl;
1722 return _fault();
1723 }
1724
1725 auto bad_method = AuthBadMethodFrame::Decode(payload);
1726 ldout(cct, 1) << __func__ << " method=" << bad_method.method()
1727 << " result " << cpp_strerror(bad_method.result())
1728 << ", allowed methods=" << bad_method.allowed_methods()
1729 << ", allowed modes=" << bad_method.allowed_modes()
1730 << dendl;
1731 ceph_assert(messenger->auth_client);
1732 auto am = auth_meta;
1733 connection->lock.unlock();
1734 int r = messenger->auth_client->handle_auth_bad_method(
1735 connection,
1736 am.get(),
1737 bad_method.method(), bad_method.result(),
1738 bad_method.allowed_methods(),
1739 bad_method.allowed_modes());
1740 connection->lock.lock();
1741 if (state != AUTH_CONNECTING || r < 0) {
1742 return _fault();
1743 }
1744 return send_auth_request(bad_method.allowed_methods());
1745}
1746
1747CtPtr ProtocolV2::handle_auth_reply_more(ceph::bufferlist &payload)
1748{
1749 ldout(cct, 20) << __func__
1750 << " payload.length()=" << payload.length() << dendl;
1751
1752 if (state != AUTH_CONNECTING) {
1753 lderr(cct) << __func__ << " not in auth connect state!" << dendl;
1754 return _fault();
1755 }
1756
1757 auto auth_more = AuthReplyMoreFrame::Decode(payload);
1758 ldout(cct, 5) << __func__
1759 << " auth reply more len=" << auth_more.auth_payload().length()
1760 << dendl;
1761 ceph_assert(messenger->auth_client);
1762 ceph::bufferlist reply;
1763 auto am = auth_meta;
1764 connection->lock.unlock();
1765 int r = messenger->auth_client->handle_auth_reply_more(
1766 connection, am.get(), auth_more.auth_payload(), &reply);
1767 connection->lock.lock();
1768 if (state != AUTH_CONNECTING) {
1769 ldout(cct, 1) << __func__ << " state changed!" << dendl;
1770 return _fault();
1771 }
1772 if (r < 0) {
1773 lderr(cct) << __func__ << " auth_client handle_auth_reply_more returned "
1774 << r << dendl;
1775 return _fault();
1776 }
1777 auto more_reply = AuthRequestMoreFrame::Encode(reply);
1778 return WRITE(more_reply, "auth request more", read_frame);
1779}
1780
1781CtPtr ProtocolV2::handle_auth_done(ceph::bufferlist &payload)
1782{
1783 ldout(cct, 20) << __func__
1784 << " payload.length()=" << payload.length() << dendl;
1785
1786 if (state != AUTH_CONNECTING) {
1787 lderr(cct) << __func__ << " not in auth connect state!" << dendl;
1788 return _fault();
1789 }
1790
1791 auto auth_done = AuthDoneFrame::Decode(payload);
1792
1793 ceph_assert(messenger->auth_client);
1794 auto am = auth_meta;
1795 connection->lock.unlock();
1796 int r = messenger->auth_client->handle_auth_done(
1797 connection,
1798 am.get(),
1799 auth_done.global_id(),
1800 auth_done.con_mode(),
1801 auth_done.auth_payload(),
1802 &am->session_key,
1803 &am->connection_secret);
1804 connection->lock.lock();
1805 if (state != AUTH_CONNECTING) {
1806 ldout(cct, 1) << __func__ << " state changed!" << dendl;
1807 return _fault();
1808 }
1809 if (r < 0) {
1810 return _fault();
1811 }
1812 auth_meta->con_mode = auth_done.con_mode();
f6b5b4d7
TL		 1813 bool is_rev1 = HAVE_MSGR2_FEATURE(peer_supported_features, REVISION_1);
1814 session_stream_handlers = ceph::crypto::onwire::rxtx_t::create_handler_pair(
1815 cct, *auth_meta, /*new_nonce_format=*/is_rev1, /*crossed=*/false);
11fdf7f2
TL		 1816
1817 state = AUTH_CONNECTING_SIGN;
1818
1819 const auto sig = auth_meta->session_key.empty() ? sha256_digest_t() :
1820 auth_meta->session_key.hmac_sha256(cct, pre_auth.rxbuf);
1821 auto sig_frame = AuthSignatureFrame::Encode(sig);
1822 pre_auth.enabled = false;
1823 pre_auth.rxbuf.clear();
1824 return WRITE(sig_frame, "auth signature", read_frame);
1825}
1826
1827CtPtr ProtocolV2::finish_client_auth() {
1828 if (!server_cookie) {
1829 ceph_assert(connect_seq == 0);
1830 state = SESSION_CONNECTING;
1831 return send_client_ident();
1832 } else { // reconnecting to previous session
1833 state = SESSION_RECONNECTING;
1834 ceph_assert(connect_seq > 0);
1835 return send_reconnect();
1836 }
1837}
1838
1839CtPtr ProtocolV2::send_client_ident() {
1840 ldout(cct, 20) << __func__ << dendl;
1841
1842 if (!connection->policy.lossy && !client_cookie) {
1843 client_cookie = ceph::util::generate_random_number<uint64_t>(1, -1ll);
1844 }
1845
1846 uint64_t flags = 0;
1847 if (connection->policy.lossy) {
1848 flags |= CEPH_MSG_CONNECT_LOSSY;
1849 }
1850
11fdf7f2
TL		 1851 auto client_ident = ClientIdentFrame::Encode(
1852 messenger->get_myaddrs(),
1853 connection->target_addr,
1854 messenger->get_myname().num(),
1855 global_seq,
1856 connection->policy.features_supported,
1857 connection->policy.features_required | msgr2_required,
1858 flags,
1859 client_cookie);
1860
1861 ldout(cct, 5) << __func__ << " sending identification: "
1862 << "addrs=" << messenger->get_myaddrs()
1863 << " target=" << connection->target_addr
1864 << " gid=" << messenger->get_myname().num()
1865 << " global_seq=" << global_seq
1866 << " features_supported=" << std::hex
1867 << connection->policy.features_supported
1868 << " features_required="
1869 << (connection->policy.features_required | msgr2_required)
1870 << " flags=" << flags
1871 << " cookie=" << client_cookie << std::dec << dendl;
1872
1873 INTERCEPT(11);
1874
1875 return WRITE(client_ident, "client ident", read_frame);
1876}
1877
1878CtPtr ProtocolV2::send_reconnect() {
1879 ldout(cct, 20) << __func__ << dendl;
1880
1881 auto reconnect = ReconnectFrame::Encode(messenger->get_myaddrs(),
1882 client_cookie,
1883 server_cookie,
1884 global_seq,
1885 connect_seq,
1886 in_seq);
1887
1888 ldout(cct, 5) << __func__ << " reconnect to session: client_cookie="
1889 << std::hex << client_cookie << " server_cookie="
1890 << server_cookie << std::dec
1891 << " gs=" << global_seq << " cs=" << connect_seq
1892 << " ms=" << in_seq << dendl;
1893
1894 INTERCEPT(13);
1895
1896 return WRITE(reconnect, "reconnect", read_frame);
1897}
1898
1899CtPtr ProtocolV2::handle_ident_missing_features(ceph::bufferlist &payload)
1900{
1901 ldout(cct, 20) << __func__
1902 << " payload.length()=" << payload.length() << dendl;
1903
1904 if (state != SESSION_CONNECTING) {
1905 lderr(cct) << __func__ << " not in session connect state!" << dendl;
1906 return _fault();
1907 }
1908
1909 auto ident_missing =
1910 IdentMissingFeaturesFrame::Decode(payload);
1911 lderr(cct) << __func__
1912 << " client does not support all server features: " << std::hex
1913 << ident_missing.features() << std::dec << dendl;
1914
1915 return _fault();
1916}
1917
1918CtPtr ProtocolV2::handle_session_reset(ceph::bufferlist &payload)
1919{
1920 ldout(cct, 20) << __func__
1921 << " payload.length()=" << payload.length() << dendl;
1922
1923 if (state != SESSION_RECONNECTING) {
1924 lderr(cct) << __func__ << " not in session reconnect state!" << dendl;
1925 return _fault();
1926 }
1927
1928 auto reset = ResetFrame::Decode(payload);
1929
1930 ldout(cct, 1) << __func__ << " received session reset full=" << reset.full()
1931 << dendl;
1932 if (reset.full()) {
1933 reset_session();
1934 } else {
1935 server_cookie = 0;
1936 connect_seq = 0;
1937 in_seq = 0;
1938 }
1939
1940 state = SESSION_CONNECTING;
1941 return send_client_ident();
1942}
1943
1944CtPtr ProtocolV2::handle_session_retry(ceph::bufferlist &payload)
1945{
1946 ldout(cct, 20) << __func__
1947 << " payload.length()=" << payload.length() << dendl;
1948
1949 if (state != SESSION_RECONNECTING) {
1950 lderr(cct) << __func__ << " not in session reconnect state!" << dendl;
1951 return _fault();
1952 }
1953
1954 auto retry = RetryFrame::Decode(payload);
1955 connect_seq = retry.connect_seq() + 1;
1956
1957 ldout(cct, 1) << __func__
1958 << " received session retry connect_seq=" << retry.connect_seq()
1959 << ", inc to cs=" << connect_seq << dendl;
1960
1961 return send_reconnect();
1962}
1963
1964CtPtr ProtocolV2::handle_session_retry_global(ceph::bufferlist &payload)
1965{
1966 ldout(cct, 20) << __func__
1967 << " payload.length()=" << payload.length() << dendl;
1968
1969 if (state != SESSION_RECONNECTING) {
1970 lderr(cct) << __func__ << " not in session reconnect state!" << dendl;
1971 return _fault();
1972 }
1973
1974 auto retry = RetryGlobalFrame::Decode(payload);
1975 global_seq = messenger->get_global_seq(retry.global_seq());
1976
1977 ldout(cct, 1) << __func__ << " received session retry global global_seq="
1978 << retry.global_seq() << ", choose new gs=" << global_seq
1979 << dendl;
1980
1981 return send_reconnect();
1982}
1983
1984CtPtr ProtocolV2::handle_wait(ceph::bufferlist &payload) {
1985 ldout(cct, 20) << __func__
1986 << " received WAIT (connection race)"
1987 << " payload.length()=" << payload.length()
1988 << dendl;
1989
1990 if (state != SESSION_CONNECTING && state != SESSION_RECONNECTING) {
1991 lderr(cct) << __func__ << " not in session (re)connect state!" << dendl;
1992 return _fault();
1993 }
1994
1995 state = WAIT;
1996 WaitFrame::Decode(payload);
1997 return _fault();
1998}
1999
2000CtPtr ProtocolV2::handle_reconnect_ok(ceph::bufferlist &payload)
2001{
2002 ldout(cct, 20) << __func__
2003 << " payload.length()=" << payload.length() << dendl;
2004
2005 if (state != SESSION_RECONNECTING) {
2006 lderr(cct) << __func__ << " not in session reconnect state!" << dendl;
2007 return _fault();
2008 }
2009
2010 auto reconnect_ok = ReconnectOkFrame::Decode(payload);
2011 ldout(cct, 5) << __func__
2012 << " reconnect accepted: sms=" << reconnect_ok.msg_seq()
2013 << dendl;
2014
2015 out_seq = discard_requeued_up_to(out_seq, reconnect_ok.msg_seq());
2016
2017 backoff = utime_t();
2018 ldout(cct, 10) << __func__ << " reconnect success " << connect_seq
2019 << ", lossy = " << connection->policy.lossy << ", features "
2020 << connection->get_features() << dendl;
2021
2022 if (connection->delay_state) {
2023 ceph_assert(connection->delay_state->ready());
2024 }
2025
2026 connection->dispatch_queue->queue_connect(connection);
2027 messenger->ms_deliver_handle_fast_connect(connection);
2028
2029 return ready();
2030}
2031
2032CtPtr ProtocolV2::handle_server_ident(ceph::bufferlist &payload)
2033{
2034 ldout(cct, 20) << __func__
2035 << " payload.length()=" << payload.length() << dendl;
2036
2037 if (state != SESSION_CONNECTING) {
2038 lderr(cct) << __func__ << " not in session connect state!" << dendl;
2039 return _fault();
2040 }
2041
2042 auto server_ident = ServerIdentFrame::Decode(payload);
2043 ldout(cct, 5) << __func__ << " received server identification:"
2044 << " addrs=" << server_ident.addrs()
2045 << " gid=" << server_ident.gid()
2046 << " global_seq=" << server_ident.global_seq()
2047 << " features_supported=" << std::hex
2048 << server_ident.supported_features()
2049 << " features_required=" << server_ident.required_features()
f6b5b4d7
TL		 2050 << " flags=" << server_ident.flags()
2051 << " cookie=" << server_ident.cookie() << std::dec << dendl;
11fdf7f2
TL		 2052
2053 // is this who we intended to talk to?
2054 // be a bit forgiving here, since we may be connecting based on addresses parsed out
2055 // of mon_host or something.
2056 if (!server_ident.addrs().contains(connection->target_addr)) {
2057 ldout(cct,1) << __func__ << " peer identifies as " << server_ident.addrs()
2058 << ", does not include " << connection->target_addr << dendl;
2059 return _fault();
2060 }
2061
2062 server_cookie = server_ident.cookie();
2063
2064 connection->set_peer_addrs(server_ident.addrs());
2065 peer_name = entity_name_t(connection->get_peer_type(), server_ident.gid());
2066 connection->set_features(server_ident.supported_features() &
2067 connection->policy.features_supported);
2068 peer_global_seq = server_ident.global_seq();
2069
2070 connection->policy.lossy = server_ident.flags() & CEPH_MSG_CONNECT_LOSSY;
2071
2072 backoff = utime_t();
2073 ldout(cct, 10) << __func__ << " connect success " << connect_seq
2074 << ", lossy = " << connection->policy.lossy << ", features "
2075 << connection->get_features() << dendl;
2076
2077 if (connection->delay_state) {
2078 ceph_assert(connection->delay_state->ready());
2079 }
2080
2081 connection->dispatch_queue->queue_connect(connection);
2082 messenger->ms_deliver_handle_fast_connect(connection);
2083
2084 return ready();
2085}
2086
2087/* Server Protocol Methods */
2088
2089CtPtr ProtocolV2::start_server_banner_exchange() {
2090 ldout(cct, 20) << __func__ << dendl;
2091
2092 INTERCEPT(2);
2093
2094 state = BANNER_ACCEPTING;
2095
2096 return _banner_exchange(CONTINUATION(post_server_banner_exchange));
2097}
2098
2099CtPtr ProtocolV2::post_server_banner_exchange() {
2100 ldout(cct, 20) << __func__ << dendl;
2101
2102 state = AUTH_ACCEPTING;
2103
2104 return CONTINUE(read_frame);
2105}
2106
2107CtPtr ProtocolV2::handle_auth_request(ceph::bufferlist &payload) {
2108 ldout(cct, 20) << __func__ << " payload.length()=" << payload.length()
2109 << dendl;
2110
2111 if (state != AUTH_ACCEPTING) {
2112 lderr(cct) << __func__ << " not in auth accept state!" << dendl;
2113 return _fault();
2114 }
2115
2116 auto request = AuthRequestFrame::Decode(payload);
2117 ldout(cct, 10) << __func__ << " AuthRequest(method=" << request.method()
2118 << ", preferred_modes=" << request.preferred_modes()
2119 << ", payload_len=" << request.auth_payload().length() << ")"
2120 << dendl;
2121 auth_meta->auth_method = request.method();
2122 auth_meta->con_mode = messenger->auth_server->pick_con_mode(
2123 connection->get_peer_type(), auth_meta->auth_method,
2124 request.preferred_modes());
2125 if (auth_meta->con_mode == CEPH_CON_MODE_UNKNOWN) {
2126 return _auth_bad_method(-EOPNOTSUPP);
2127 }
2128 return _handle_auth_request(request.auth_payload(), false);
2129}
2130
2131CtPtr ProtocolV2::_auth_bad_method(int r)
2132{
2133 ceph_assert(r < 0);
2134 std::vector<uint32_t> allowed_methods;
2135 std::vector<uint32_t> allowed_modes;
2136 messenger->auth_server->get_supported_auth_methods(
2137 connection->get_peer_type(), &allowed_methods, &allowed_modes);
2138 ldout(cct, 1) << __func__ << " auth_method " << auth_meta->auth_method
2139 << " r " << cpp_strerror(r)
2140 << ", allowed_methods " << allowed_methods
2141 << ", allowed_modes " << allowed_modes
2142 << dendl;
2143 auto bad_method = AuthBadMethodFrame::Encode(auth_meta->auth_method, r,
2144 allowed_methods, allowed_modes);
2145 return WRITE(bad_method, "bad auth method", read_frame);
2146}
2147
2148CtPtr ProtocolV2::_handle_auth_request(bufferlist& auth_payload, bool more)
2149{
2150 if (!messenger->auth_server) {
2151 return _fault();
2152 }
2153 bufferlist reply;
2154 auto am = auth_meta;
2155 connection->lock.unlock();
2156 int r = messenger->auth_server->handle_auth_request(
2157 connection, am.get(),
2158 more, am->auth_method, auth_payload,
2159 &reply);
2160 connection->lock.lock();
2161 if (state != AUTH_ACCEPTING && state != AUTH_ACCEPTING_MORE) {
2162 ldout(cct, 1) << __func__
2163 << " state changed while accept, it must be mark_down"
2164 << dendl;
2165 ceph_assert(state == CLOSED);
2166 return _fault();
2167 }
2168 if (r == 1) {
2169 INTERCEPT(10);
2170 state = AUTH_ACCEPTING_SIGN;
2171
2172 auto auth_done = AuthDoneFrame::Encode(connection->peer_global_id,
2173 auth_meta->con_mode,
2174 reply);
2175 return WRITE(auth_done, "auth done", finish_auth);
2176 } else if (r == 0) {
2177 state = AUTH_ACCEPTING_MORE;
2178
2179 auto more = AuthReplyMoreFrame::Encode(reply);
2180 return WRITE(more, "auth reply more", read_frame);
2181 } else if (r == -EBUSY) {
2182 // kick the client and maybe they'll come back later
2183 return _fault();
2184 } else {
2185 return _auth_bad_method(r);
2186 }
2187}
2188
2189CtPtr ProtocolV2::finish_auth()
2190{
2191 ceph_assert(auth_meta);
2192 // TODO: having a possibility to check whether we're server or client could
2193 // allow reusing finish_auth().
f6b5b4d7
TL		 2194 bool is_rev1 = HAVE_MSGR2_FEATURE(peer_supported_features, REVISION_1);
2195 session_stream_handlers = ceph::crypto::onwire::rxtx_t::create_handler_pair(
2196 cct, *auth_meta, /*new_nonce_format=*/is_rev1, /*crossed=*/true);
11fdf7f2
TL		 2197
2198 const auto sig = auth_meta->session_key.empty() ? sha256_digest_t() :
2199 auth_meta->session_key.hmac_sha256(cct, pre_auth.rxbuf);
2200 auto sig_frame = AuthSignatureFrame::Encode(sig);
2201 pre_auth.enabled = false;
2202 pre_auth.rxbuf.clear();
2203 return WRITE(sig_frame, "auth signature", read_frame);
2204}
2205
2206CtPtr ProtocolV2::handle_auth_request_more(ceph::bufferlist &payload)
2207{
2208 ldout(cct, 20) << __func__
2209 << " payload.length()=" << payload.length() << dendl;
2210
2211 if (state != AUTH_ACCEPTING_MORE) {
2212 lderr(cct) << __func__ << " not in auth accept more state!" << dendl;
2213 return _fault();
2214 }
2215
2216 auto auth_more = AuthRequestMoreFrame::Decode(payload);
2217 return _handle_auth_request(auth_more.auth_payload(), true);
2218}
2219
2220CtPtr ProtocolV2::handle_auth_signature(ceph::bufferlist &payload)
2221{
2222 ldout(cct, 20) << __func__
2223 << " payload.length()=" << payload.length() << dendl;
2224
2225 if (state != AUTH_ACCEPTING_SIGN && state != AUTH_CONNECTING_SIGN) {
2226 lderr(cct) << __func__
2227 << " pre-auth verification signature seen in wrong state!"
2228 << dendl;
2229 return _fault();
2230 }
2231
2232 auto sig_frame = AuthSignatureFrame::Decode(payload);
2233
2234 const auto actual_tx_sig = auth_meta->session_key.empty() ?
2235 sha256_digest_t() : auth_meta->session_key.hmac_sha256(cct, pre_auth.txbuf);
2236 if (sig_frame.signature() != actual_tx_sig) {
2237 ldout(cct, 2) << __func__ << " pre-auth signature mismatch"
2238 << " actual_tx_sig=" << actual_tx_sig
2239 << " sig_frame.signature()=" << sig_frame.signature()
2240 << dendl;
2241 return _fault();
2242 } else {
2243 ldout(cct, 20) << __func__ << " pre-auth signature success"
2244 << " sig_frame.signature()=" << sig_frame.signature()
2245 << dendl;
2246 pre_auth.txbuf.clear();
2247 }
2248
2249 if (state == AUTH_ACCEPTING_SIGN) {
2250 // server had sent AuthDone and client responded with correct pre-auth
2251 // signature. we can start accepting new sessions/reconnects.
2252 state = SESSION_ACCEPTING;
2253 return CONTINUE(read_frame);
2254 } else if (state == AUTH_CONNECTING_SIGN) {
2255 // this happened at client side
2256 return finish_client_auth();
2257 } else {
9f95a23c 2258 ceph_abort("state corruption");
11fdf7f2
TL		 2259 }
2260}
2261
2262CtPtr ProtocolV2::handle_client_ident(ceph::bufferlist &payload)
2263{
2264 ldout(cct, 20) << __func__
2265 << " payload.length()=" << payload.length() << dendl;
2266
2267 if (state != SESSION_ACCEPTING) {
2268 lderr(cct) << __func__ << " not in session accept state!" << dendl;
2269 return _fault();
2270 }
2271
2272 auto client_ident = ClientIdentFrame::Decode(payload);
2273
2274 ldout(cct, 5) << __func__ << " received client identification:"
2275 << " addrs=" << client_ident.addrs()
2276 << " target=" << client_ident.target_addr()
2277 << " gid=" << client_ident.gid()
2278 << " global_seq=" << client_ident.global_seq()
2279 << " features_supported=" << std::hex
2280 << client_ident.supported_features()
2281 << " features_required=" << client_ident.required_features()
2282 << " flags=" << client_ident.flags()
2283 << " cookie=" << client_ident.cookie() << std::dec << dendl;
2284
2285 if (client_ident.addrs().empty() ||
2286 client_ident.addrs().front() == entity_addr_t()) {
2287 ldout(cct,5) << __func__ << " oops, client_ident.addrs() is empty" << dendl;
2288 return _fault(); // a v2 peer should never do this
2289 }
2290 if (!messenger->get_myaddrs().contains(client_ident.target_addr())) {
2291 ldout(cct,5) << __func__ << " peer is trying to reach "
2292 << client_ident.target_addr()
2293 << " which is not us (" << messenger->get_myaddrs() << ")"
2294 << dendl;
2295 return _fault();
2296 }
2297
2298 connection->set_peer_addrs(client_ident.addrs());
2299 connection->target_addr = connection->_infer_target_addr(client_ident.addrs());
2300
2301 peer_name = entity_name_t(connection->get_peer_type(), client_ident.gid());
2302 connection->set_peer_id(client_ident.gid());
2303
2304 client_cookie = client_ident.cookie();
2305
2306 uint64_t feat_missing =
2307 (connection->policy.features_required | msgr2_required) &
2308 ~(uint64_t)client_ident.supported_features();
2309 if (feat_missing) {
2310 ldout(cct, 1) << __func__ << " peer missing required features " << std::hex
2311 << feat_missing << std::dec << dendl;
2312 auto ident_missing_features =
2313 IdentMissingFeaturesFrame::Encode(feat_missing);
2314
2315 return WRITE(ident_missing_features, "ident missing features", read_frame);
2316 }
2317
2318 connection_features =
2319 client_ident.supported_features() & connection->policy.features_supported;
2320
2321 peer_global_seq = client_ident.global_seq();
2322
9f95a23c
TL		 2323 if (connection->policy.server &&
2324 connection->policy.lossy &&
2325 !connection->policy.register_lossy_clients) {
2326 // incoming lossy client, no need to register this connection
2327 } else {
2328 // Looks good so far, let's check if there is already an existing connection
2329 // to this peer.
2330 connection->lock.unlock();
2331 AsyncConnectionRef existing = messenger->lookup_conn(
2332 *connection->peer_addrs);
2333
2334 if (existing &&
2335 existing->protocol->proto_type != 2) {
2336 ldout(cct,1) << __func__ << " existing " << existing << " proto "
2337 << existing->protocol.get() << " version is "
2338 << existing->protocol->proto_type << ", marking down"
2339 << dendl;
2340 existing->mark_down();
2341 existing = nullptr;
2342 }
11fdf7f2 2343
9f95a23c 2344 connection->inject_delay();
11fdf7f2 2345
9f95a23c
TL		 2346 connection->lock.lock();
2347 if (state != SESSION_ACCEPTING) {
2348 ldout(cct, 1) << __func__
2349 << " state changed while accept, it must be mark_down"
2350 << dendl;
2351 ceph_assert(state == CLOSED);
2352 return _fault();
2353 }
11fdf7f2 2354
9f95a23c
TL		 2355 if (existing) {
2356 return handle_existing_connection(existing);
2357 }
11fdf7f2
TL		 2358 }
2359
2360 // if everything is OK reply with server identification
2361 return send_server_ident();
2362}
2363
2364CtPtr ProtocolV2::handle_reconnect(ceph::bufferlist &payload)
2365{
2366 ldout(cct, 20) << __func__
2367 << " payload.length()=" << payload.length() << dendl;
2368
2369 if (state != SESSION_ACCEPTING) {
2370 lderr(cct) << __func__ << " not in session accept state!" << dendl;
2371 return _fault();
2372 }
2373
2374 auto reconnect = ReconnectFrame::Decode(payload);
2375
2376 ldout(cct, 5) << __func__
2377 << " received reconnect:"
2378 << " client_cookie=" << std::hex << reconnect.client_cookie()
2379 << " server_cookie=" << reconnect.server_cookie() << std::dec
2380 << " gs=" << reconnect.global_seq()
2381 << " cs=" << reconnect.connect_seq()
2382 << " ms=" << reconnect.msg_seq()
2383 << dendl;
2384
2385 // Should we check if one of the ident.addrs match connection->target_addr
2386 // as we do in ProtocolV1?
2387 connection->set_peer_addrs(reconnect.addrs());
2388 connection->target_addr = connection->_infer_target_addr(reconnect.addrs());
2389 peer_global_seq = reconnect.global_seq();
2390
2391 connection->lock.unlock();
2392 AsyncConnectionRef existing = messenger->lookup_conn(*connection->peer_addrs);
2393
2394 if (existing &&
2395 existing->protocol->proto_type != 2) {
2396 ldout(cct,1) << __func__ << " existing " << existing << " proto "
2397 << existing->protocol.get() << " version is "
2398 << existing->protocol->proto_type << ", marking down" << dendl;
2399 existing->mark_down();
2400 existing = nullptr;
2401 }
2402
2403 connection->inject_delay();
2404
2405 connection->lock.lock();
2406 if (state != SESSION_ACCEPTING) {
2407 ldout(cct, 1) << __func__
2408 << " state changed while accept, it must be mark_down"
2409 << dendl;
2410 ceph_assert(state == CLOSED);
2411 return _fault();
2412 }
2413
2414 if (!existing) {
2415 // there is no existing connection therefore cannot reconnect to previous
2416 // session
2417 ldout(cct, 0) << __func__
2418 << " no existing connection exists, reseting client" << dendl;
2419 auto reset = ResetFrame::Encode(true);
2420 return WRITE(reset, "session reset", read_frame);
2421 }
2422
2423 std::lock_guard<std::mutex> l(existing->lock);
2424
2425 ProtocolV2 *exproto = dynamic_cast<ProtocolV2 *>(existing->protocol.get());
2426 if (!exproto) {
2427 ldout(cct, 1) << __func__ << " existing=" << existing << dendl;
2428 ceph_assert(false);
2429 }
2430
2431 if (exproto->state == CLOSED) {
2432 ldout(cct, 5) << __func__ << " existing " << existing
2433 << " already closed. Reseting client" << dendl;
2434 auto reset = ResetFrame::Encode(true);
2435 return WRITE(reset, "session reset", read_frame);
2436 }
2437
2438 if (exproto->replacing) {
2439 ldout(cct, 1) << __func__
2440 << " existing racing replace happened while replacing."
2441 << " existing=" << existing << dendl;
2442 auto retry = RetryGlobalFrame::Encode(exproto->peer_global_seq);
2443 return WRITE(retry, "session retry", read_frame);
2444 }
2445
2446 if (exproto->client_cookie != reconnect.client_cookie()) {
2447 ldout(cct, 1) << __func__ << " existing=" << existing
2448 << " client cookie mismatch, I must have reseted:"
2449 << " cc=" << std::hex << exproto->client_cookie
2450 << " rcc=" << reconnect.client_cookie()
2451 << ", reseting client." << std::dec
2452 << dendl;
2453 auto reset = ResetFrame::Encode(connection->policy.resetcheck);
2454 return WRITE(reset, "session reset", read_frame);
2455 } else if (exproto->server_cookie == 0) {
2456 // this happens when:
2457 // - a connects to b
2458 // - a sends client_ident
2459 // - b gets client_ident, sends server_ident and sets cookie X
2460 // - connection fault
2461 // - b reconnects to a with cookie X, connect_seq=1
2462 // - a has cookie==0
2463 ldout(cct, 1) << __func__ << " I was a client and didn't received the"
2464 << " server_ident. Asking peer to resume session"
2465 << " establishment" << dendl;
2466 auto reset = ResetFrame::Encode(false);
2467 return WRITE(reset, "session reset", read_frame);
2468 }
2469
2470 if (exproto->peer_global_seq > reconnect.global_seq()) {
2471 ldout(cct, 5) << __func__
2472 << " stale global_seq: sgs=" << exproto->peer_global_seq
2473 << " cgs=" << reconnect.global_seq()
2474 << ", ask client to retry global" << dendl;
2475 auto retry = RetryGlobalFrame::Encode(exproto->peer_global_seq);
2476
2477 INTERCEPT(18);
2478
2479 return WRITE(retry, "session retry", read_frame);
2480 }
2481
2482 if (exproto->connect_seq > reconnect.connect_seq()) {
2483 ldout(cct, 5) << __func__
2484 << " stale connect_seq scs=" << exproto->connect_seq
2485 << " ccs=" << reconnect.connect_seq()
2486 << " , ask client to retry" << dendl;
2487 auto retry = RetryFrame::Encode(exproto->connect_seq);
2488 return WRITE(retry, "session retry", read_frame);
2489 }
2490
2491 if (exproto->connect_seq == reconnect.connect_seq()) {
2492 // reconnect race: both peers are sending reconnect messages
2493 if (existing->peer_addrs->msgr2_addr() >
2494 messenger->get_myaddrs().msgr2_addr() &&
2495 !existing->policy.server) {
2496 // the existing connection wins
2497 ldout(cct, 1)
2498 << __func__
2499 << " reconnect race detected, this connection loses to existing="
2500 << existing << dendl;
2501
2502 auto wait = WaitFrame::Encode();
2503 return WRITE(wait, "wait", read_frame);
2504 } else {
2505 // this connection wins
2506 ldout(cct, 1) << __func__
2507 << " reconnect race detected, replacing existing="
2508 << existing << " socket by this connection's socket"
2509 << dendl;
2510 }
2511 }
2512
2513 ldout(cct, 1) << __func__ << " reconnect to existing=" << existing << dendl;
2514
2515 reconnecting = true;
2516
2517 // everything looks good
2518 exproto->connect_seq = reconnect.connect_seq();
2519 exproto->message_seq = reconnect.msg_seq();
2520
2521 return reuse_connection(existing, exproto);
2522}
2523
9f95a23c 2524CtPtr ProtocolV2::handle_existing_connection(const AsyncConnectionRef& existing) {
11fdf7f2
TL		 2525 ldout(cct, 20) << __func__ << " existing=" << existing << dendl;
2526
2527 std::lock_guard<std::mutex> l(existing->lock);
2528
2529 ProtocolV2 *exproto = dynamic_cast<ProtocolV2 *>(existing->protocol.get());
2530 if (!exproto) {
2531 ldout(cct, 1) << __func__ << " existing=" << existing << dendl;
2532 ceph_assert(false);
2533 }
2534
2535 if (exproto->state == CLOSED) {
2536 ldout(cct, 1) << __func__ << " existing " << existing << " already closed."
2537 << dendl;
2538 return send_server_ident();
2539 }
2540
2541 if (exproto->replacing) {
2542 ldout(cct, 1) << __func__
2543 << " existing racing replace happened while replacing."
2544 << " existing=" << existing << dendl;
2545 auto wait = WaitFrame::Encode();
2546 return WRITE(wait, "wait", read_frame);
2547 }
2548
2549 if (exproto->peer_global_seq > peer_global_seq) {
2550 ldout(cct, 1) << __func__ << " this is a stale connection, peer_global_seq="
2551 << peer_global_seq
2552 << " existing->peer_global_seq=" << exproto->peer_global_seq
2553 << ", stopping this connection." << dendl;
2554 stop();
2555 connection->dispatch_queue->queue_reset(connection);
2556 return nullptr;
2557 }
2558
2559 if (existing->policy.lossy) {
2560 // existing connection can be thrown out in favor of this one
2561 ldout(cct, 1)
2562 << __func__ << " existing=" << existing
2563 << " is a lossy channel. Stopping existing in favor of this connection"
2564 << dendl;
2565 existing->protocol->stop();
2566 existing->dispatch_queue->queue_reset(existing.get());
2567 return send_server_ident();
2568 }
2569
2570 if (exproto->server_cookie && exproto->client_cookie &&
2571 exproto->client_cookie != client_cookie) {
2572 // Found previous session
2573 // peer has reseted and we're going to reuse the existing connection
2574 // by replacing the communication socket
2575 ldout(cct, 1) << __func__ << " found previous session existing=" << existing
2576 << ", peer must have reseted." << dendl;
2577 if (connection->policy.resetcheck) {
2578 exproto->reset_session();
2579 }
2580 return reuse_connection(existing, exproto);
2581 }
2582
2583 if (exproto->client_cookie == client_cookie) {
2584 // session establishment interrupted between client_ident and server_ident,
2585 // continuing...
2586 ldout(cct, 1) << __func__ << " found previous session existing=" << existing
2587 << ", continuing session establishment." << dendl;
2588 return reuse_connection(existing, exproto);
2589 }
2590
2591 if (exproto->state == READY || exproto->state == STANDBY) {
2592 ldout(cct, 1) << __func__ << " existing=" << existing
2593 << " is READY/STANDBY, lets reuse it" << dendl;
2594 return reuse_connection(existing, exproto);
2595 }
2596
2597 // Looks like a connection race: server and client are both connecting to
2598 // each other at the same time.
2599 if (connection->peer_addrs->msgr2_addr() <
2600 messenger->get_myaddrs().msgr2_addr() ||
2601 existing->policy.server) {
2602 // this connection wins
2603 ldout(cct, 1) << __func__
2604 << " connection race detected, replacing existing="
2605 << existing << " socket by this connection's socket" << dendl;
2606 return reuse_connection(existing, exproto);
2607 } else {
2608 // the existing connection wins
2609 ldout(cct, 1)
2610 << __func__
2611 << " connection race detected, this connection loses to existing="
2612 << existing << dendl;
2613 ceph_assert(connection->peer_addrs->msgr2_addr() >
2614 messenger->get_myaddrs().msgr2_addr());
2615
2616 // make sure we follow through with opening the existing
2617 // connection (if it isn't yet open) since we know the peer
2618 // has something to send to us.
2619 existing->send_keepalive();
2620 auto wait = WaitFrame::Encode();
2621 return WRITE(wait, "wait", read_frame);
2622 }
2623}
2624
9f95a23c 2625CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
11fdf7f2
TL		 2626 ProtocolV2 *exproto) {
2627 ldout(cct, 20) << __func__ << " existing=" << existing
2628 << " reconnect=" << reconnecting << dendl;
2629
2630 connection->inject_delay();
2631
2632 std::lock_guard<std::mutex> l(existing->write_lock);
2633
2634 connection->center->delete_file_event(connection->cs.fd(),
2635 EVENT_READABLE | EVENT_WRITABLE);
2636
2637 if (existing->delay_state) {
2638 existing->delay_state->flush();
2639 ceph_assert(!connection->delay_state);
2640 }
2641 exproto->reset_recv_state();
2642 exproto->pre_auth.enabled = false;
2643
2644 if (!reconnecting) {
f6b5b4d7
TL		 2645 exproto->peer_supported_features = peer_supported_features;
2646 exproto->tx_frame_asm.set_is_rev1(tx_frame_asm.get_is_rev1());
2647 exproto->rx_frame_asm.set_is_rev1(rx_frame_asm.get_is_rev1());
2648
11fdf7f2
TL		 2649 exproto->client_cookie = client_cookie;
2650 exproto->peer_name = peer_name;
2651 exproto->connection_features = connection_features;
2652 existing->set_features(connection_features);
2653 }
2654 exproto->peer_global_seq = peer_global_seq;
2655
9f95a23c 2656 ceph_assert(connection->center->in_thread());
11fdf7f2
TL		 2657 auto temp_cs = std::move(connection->cs);
2658 EventCenter *new_center = connection->center;
2659 Worker *new_worker = connection->worker;
9f95a23c
TL		 2660 // we can steal the session_stream_handlers under the assumption
2661 // this happens in the event center's thread as there should be
2662 // no user outside its boundaries (simlarly to e.g. outgoing_bl).
2663 auto temp_stream_handlers = std::move(session_stream_handlers);
2664 exproto->auth_meta = auth_meta;
11fdf7f2
TL		 2665
2666 ldout(messenger->cct, 5) << __func__ << " stop myself to swap existing"
2667 << dendl;
494da23a 2668
11fdf7f2
TL		 2669 // avoid _stop shutdown replacing socket
2670 // queue a reset on the new connection, which we're dumping for the old
2671 stop();
2672
2673 connection->dispatch_queue->queue_reset(connection);
2674
2675 exproto->can_write = false;
494da23a 2676 exproto->write_in_progress = false;
11fdf7f2
TL		 2677 exproto->reconnecting = reconnecting;
2678 exproto->replacing = true;
11fdf7f2
TL		 2679 existing->state_offset = 0;
2680 // avoid previous thread modify event
2681 exproto->state = NONE;
2682 existing->state = AsyncConnection::STATE_NONE;
2683 // Discard existing prefetch buffer in `recv_buf`
2684 existing->recv_start = existing->recv_end = 0;
2685 // there shouldn't exist any buffer
2686 ceph_assert(connection->recv_start == connection->recv_end);
2687
2688 auto deactivate_existing = std::bind(
9f95a23c
TL		 2689 [ existing,
2690 new_worker,
2691 new_center,
2692 exproto,
2693 temp_stream_handlers=std::move(temp_stream_handlers)
2694 ](ConnectedSocket &cs) mutable {
11fdf7f2
TL		 2695 // we need to delete time event in original thread
2696 {
2697 std::lock_guard<std::mutex> l(existing->lock);
2698 existing->write_lock.lock();
2699 exproto->requeue_sent();
9f95a23c
TL		 2700 // XXX: do we really need the locking for `outgoing_bl`? There is
2701 // a comment just above its definition saying "lockfree, only used
2702 // in own thread". I'm following lockfull schema just in the case.
2703 // From performance point of view it should be fine – this happens
2704 // far away from hot paths.
2705 existing->outgoing_bl.clear();
11fdf7f2 2706 existing->open_write = false;
9f95a23c 2707 exproto->session_stream_handlers = std::move(temp_stream_handlers);
11fdf7f2
TL		 2708 existing->write_lock.unlock();
2709 if (exproto->state == NONE) {
2710 existing->shutdown_socket();
2711 existing->cs = std::move(cs);
2712 existing->worker->references--;
2713 new_worker->references++;
2714 existing->logger = new_worker->get_perf_counter();
2715 existing->worker = new_worker;
2716 existing->center = new_center;
2717 if (existing->delay_state)
2718 existing->delay_state->set_center(new_center);
2719 } else if (exproto->state == CLOSED) {
2720 auto back_to_close = std::bind(
2721 [](ConnectedSocket &cs) mutable { cs.close(); }, std::move(cs));
2722 new_center->submit_to(new_center->get_id(),
2723 std::move(back_to_close), true);
2724 return;
2725 } else {
2726 ceph_abort();
2727 }
2728 }
2729
2730 // Before changing existing->center, it may already exists some
2731 // events in existing->center's queue. Then if we mark down
2732 // `existing`, it will execute in another thread and clean up
2733 // connection. Previous event will result in segment fault
2734 auto transfer_existing = [existing, exproto]() mutable {
2735 std::lock_guard<std::mutex> l(existing->lock);
2736 if (exproto->state == CLOSED) return;
2737 ceph_assert(exproto->state == NONE);
2738
2739 exproto->state = SESSION_ACCEPTING;
81eedcae
TL		 2740 // we have called shutdown_socket above
2741 ceph_assert(existing->last_tick_id == 0);
2742 // restart timer since we are going to re-build connection
2743 existing->last_connect_started = ceph::coarse_mono_clock::now();
2744 existing->last_tick_id = existing->center->create_time_event(
2745 existing->connect_timeout_us, existing->tick_handler);
11fdf7f2
TL		 2746 existing->state = AsyncConnection::STATE_CONNECTION_ESTABLISHED;
2747 existing->center->create_file_event(existing->cs.fd(), EVENT_READABLE,
2748 existing->read_handler);
2749 if (!exproto->reconnecting) {
2750 exproto->run_continuation(exproto->send_server_ident());
2751 } else {
2752 exproto->run_continuation(exproto->send_reconnect_ok());
2753 }
2754 };
2755 if (existing->center->in_thread())
2756 transfer_existing();
2757 else
2758 existing->center->submit_to(existing->center->get_id(),
2759 std::move(transfer_existing), true);
2760 },
2761 std::move(temp_cs));
2762
2763 existing->center->submit_to(existing->center->get_id(),
2764 std::move(deactivate_existing), true);
2765 return nullptr;
2766}
2767
2768CtPtr ProtocolV2::send_server_ident() {
2769 ldout(cct, 20) << __func__ << dendl;
2770
2771 // this is required for the case when this connection is being replaced
2772 out_seq = discard_requeued_up_to(out_seq, 0);
2773 in_seq = 0;
2774
2775 if (!connection->policy.lossy) {
2776 server_cookie = ceph::util::generate_random_number<uint64_t>(1, -1ll);
2777 }
2778
2779 uint64_t flags = 0;
2780 if (connection->policy.lossy) {
2781 flags = flags | CEPH_MSG_CONNECT_LOSSY;
2782 }
2783
2784 uint64_t gs = messenger->get_global_seq();
2785 auto server_ident = ServerIdentFrame::Encode(
2786 messenger->get_myaddrs(),
2787 messenger->get_myname().num(),
2788 gs,
2789 connection->policy.features_supported,
2790 connection->policy.features_required | msgr2_required,
2791 flags,
2792 server_cookie);
2793
2794 ldout(cct, 5) << __func__ << " sending identification:"
2795 << " addrs=" << messenger->get_myaddrs()
2796 << " gid=" << messenger->get_myname().num()
2797 << " global_seq=" << gs << " features_supported=" << std::hex
2798 << connection->policy.features_supported
2799 << " features_required="
2800 << (connection->policy.features_required | msgr2_required)
f6b5b4d7
TL		 2801 << " flags=" << flags
2802 << " cookie=" << server_cookie << std::dec << dendl;
11fdf7f2
TL		 2803
2804 connection->lock.unlock();
2805 // Because "replacing" will prevent other connections preempt this addr,
2806 // it's safe that here we don't acquire Connection's lock
2807 ssize_t r = messenger->accept_conn(connection);
2808
2809 connection->inject_delay();
2810
2811 connection->lock.lock();
2812
2813 if (r < 0) {
2814 ldout(cct, 1) << __func__ << " existing race replacing process for addr = "
2815 << connection->peer_addrs->msgr2_addr()
2816 << " just fail later one(this)" << dendl;
2817 connection->inject_delay();
2818 return _fault();
2819 }
2820 if (state != SESSION_ACCEPTING) {
2821 ldout(cct, 1) << __func__
2822 << " state changed while accept_conn, it must be mark_down"
2823 << dendl;
2824 ceph_assert(state == CLOSED || state == NONE);
2825 messenger->unregister_conn(connection);
2826 connection->inject_delay();
2827 return _fault();
2828 }
2829
2830 connection->set_features(connection_features);
2831
2832 // notify
2833 connection->dispatch_queue->queue_accept(connection);
2834 messenger->ms_deliver_handle_fast_accept(connection);
2835
2836 INTERCEPT(12);
2837
2838 return WRITE(server_ident, "server ident", server_ready);
2839}
2840
2841CtPtr ProtocolV2::server_ready() {
2842 ldout(cct, 20) << __func__ << dendl;
2843
2844 if (connection->delay_state) {
2845 ceph_assert(connection->delay_state->ready());
2846 }
2847
2848 return ready();
2849}
2850
2851CtPtr ProtocolV2::send_reconnect_ok() {
2852 ldout(cct, 20) << __func__ << dendl;
2853
2854 out_seq = discard_requeued_up_to(out_seq, message_seq);
2855
2856 uint64_t ms = in_seq;
2857 auto reconnect_ok = ReconnectOkFrame::Encode(ms);
2858
2859 ldout(cct, 5) << __func__ << " sending reconnect_ok: msg_seq=" << ms << dendl;
2860
2861 connection->lock.unlock();
2862 // Because "replacing" will prevent other connections preempt this addr,
2863 // it's safe that here we don't acquire Connection's lock
2864 ssize_t r = messenger->accept_conn(connection);
2865
2866 connection->inject_delay();
2867
2868 connection->lock.lock();
2869
2870 if (r < 0) {
2871 ldout(cct, 1) << __func__ << " existing race replacing process for addr = "
2872 << connection->peer_addrs->msgr2_addr()
2873 << " just fail later one(this)" << dendl;
2874 connection->inject_delay();
2875 return _fault();
2876 }
2877 if (state != SESSION_ACCEPTING) {
2878 ldout(cct, 1) << __func__
2879 << " state changed while accept_conn, it must be mark_down"
2880 << dendl;
2881 ceph_assert(state == CLOSED || state == NONE);
2882 messenger->unregister_conn(connection);
2883 connection->inject_delay();
2884 return _fault();
2885 }
2886
2887 // notify
2888 connection->dispatch_queue->queue_accept(connection);
2889 messenger->ms_deliver_handle_fast_accept(connection);
2890
2891 INTERCEPT(14);
2892
2893 return WRITE(reconnect_ok, "reconnect ok", server_ready);
2894}
Ceph