ceph/src/msg/async/ProtocolV2.cc

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