| 211 | } |
| 212 | |
| 213 | void ProtocolV1::send_message(Message *m) { |
| 214 | ceph::buffer::list bl; |
| 215 | uint64_t f = connection->get_features(); |
| 216 | |
| 217 | // TODO: Currently not all messages supports reencode like MOSDMap, so here |
| 218 | // only let fast dispatch support messages prepare message |
| 219 | bool can_fast_prepare = messenger->ms_can_fast_dispatch(m); |
| 220 | bool is_prepared = false; |
| 221 | if (can_fast_prepare && f) { |
| 222 | prepare_send_message(f, m, bl); |
| 223 | is_prepared = true; |
| 224 | } |
| 225 | |
| 226 | std::lock_guard<std::mutex> l(connection->write_lock); |
| 227 | // "features" changes will change the payload encoding |
| 228 | if (can_fast_prepare && |
| 229 | (can_write == WriteStatus::NOWRITE || connection->get_features() != f)) { |
| 230 | // ensure the correctness of message encoding |
| 231 | bl.clear(); |
| 232 | m->clear_payload(); |
| 233 | ldout(cct, 5) << __func__ << " clear encoded buffer previous " << f |
| 234 | << " != " << connection->get_features() << dendl; |
| 235 | } |
| 236 | if (can_write == WriteStatus::CLOSED) { |
| 237 | ldout(cct, 10) << __func__ << " connection closed." |
| 238 | << " Drop message " << m << dendl; |
| 239 | m->put(); |
| 240 | } else { |
| 241 | m->queue_start = ceph::mono_clock::now(); |
| 242 | m->trace.event("async enqueueing message"); |
| 243 | out_q[m->get_priority()].emplace_back(out_q_entry_t{ |
| 244 | std::move(bl), m, is_prepared}); |
| 245 | ldout(cct, 15) << __func__ << " inline write is denied, reschedule m=" << m |
| 246 | << dendl; |
| 247 | if (can_write != WriteStatus::REPLACING && !write_in_progress) { |
| 248 | write_in_progress = true; |
| 249 | connection->center->dispatch_event_external(connection->write_handler); |
| 250 | } |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | void ProtocolV1::prepare_send_message(uint64_t features, Message *m, |
| 255 | ceph::buffer::list &bl) { |
nothing calls this directly
no test coverage detected