msg/async, v2: add flags field to frame's epilogue.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index ebf0aa308cc8e554f4bd202f98f6af76e7eb2a7d..5cec45a4d9316121e7c0f8a355b0fcbf30f94cbd 100644 (file)
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -1391,16 +1391,16 @@ CtPtr ProtocolV2::handle_read_frame_epilogue_main(char *buffer, int r)
                   << FRAME_EPILOGUE_SIZE << dendl;
 
     ceph_assert(session_stream_handlers.rx);
-    ceph_assert(FRAME_EPILOGUE_SIZE == \
+    ceph_assert(sizeof(epilogue.integrity.auth_tag) == \
       session_stream_handlers.rx->get_extra_size_at_final());
 
     // I expect that ::temp_buffer is being used here.
-    ceph::bufferlist epilogue_bl;
-    epilogue_bl.push_back(buffer::create_static(FRAME_EPILOGUE_SIZE,
-        epilogue.auth_tag));
+    ceph::bufferlist auth_tag_bl;
+    auth_tag_bl.push_back(buffer::create_static(
+        sizeof(epilogue.integrity.auth_tag), epilogue.integrity.auth_tag));
     try {
       session_stream_handlers.rx->authenticated_decrypt_update_final(
-       std::move(epilogue_bl), segment_t::DEFAULT_ALIGNMENT);
+       std::move(auth_tag_bl), segment_t::DEFAULT_ALIGNMENT);
     } catch (ceph::crypto::onwire::MsgAuthError &e) {
       ldout(cct, 5) << __func__ << " message authentication failed: "
                    << e.what() << dendl;
@@ -1408,7 +1408,7 @@ CtPtr ProtocolV2::handle_read_frame_epilogue_main(char *buffer, int r)
     }
   } else {
     for (std::uint8_t idx = 0; idx < rx_segments_data.size(); idx++) {
-      const __u32 expected_crc = epilogue.crc_values[idx];
+      const __u32 expected_crc = epilogue.integrity.crc[idx];
       const __u32 calculated_crc = rx_segments_data[idx].crc32c(-1);
       if (expected_crc != calculated_crc) {
        ldout(cct, 5) << __func__ << " message integrity check failed: "
@@ -1425,6 +1425,13 @@ CtPtr ProtocolV2::handle_read_frame_epilogue_main(char *buffer, int r)
     }
   }
 
+  // verify the late abort bit
+  if (epilogue.flags & FRAME_FLAGS_LATE_ABORT) {
+    ldout(cct, 5) << __func__ << " got LATE_ABORT request. Dropping frame"
+                  << dendl;
+    return CONTINUE(read_frame);
+  }
+
   return handle_read_frame_dispatch();
 }
 

diff --git a/src/msg/async/frames_v2.h b/src/msg/async/frames_v2.h
index 85f333f5d55d3f73e2490eb79c34174fa0d668d6..d0f34b01d00e1a03dfb0581d8b52864e27c8748f 100644 (file)
--- a/src/msg/async/frames_v2.h
+++ b/src/msg/async/frames_v2.h
@@ -108,13 +108,23 @@ static_assert(std::is_standard_layout<preamble_block_t>::value);
 // at the end of each frame holds:
 //  * CRC32 for MAX_NUM_SEGMENTS -- in plain mode,
 //  * cipher-specific data (e.g. auth tag for AES-GCM).
-union epilogue_block_t {
-  char auth_tag[CRYPTO_BLOCK_SIZE];
-  std::array<__le32, MAX_NUM_SEGMENTS> crc_values;
-};
-static_assert(sizeof(epilogue_block_t) % CRYPTO_BLOCK_SIZE == 0);
+// Additionally, it consists flags field used currently to abort message
+// processing. It can be set by transceiver even very lately without any
+// need for introducing buffering.
+struct epilogue_block_t {
+  union {
+    char auth_tag[CRYPTO_BLOCK_SIZE];
+    std::array<__le32, MAX_NUM_SEGMENTS> crc;
+  } integrity;
+  static_assert(sizeof(integrity.auth_tag) == CRYPTO_BLOCK_SIZE);
+  static_assert(sizeof(integrity.auth_tag) == sizeof(integrity.crc));
+
+  __u8 flags;
+} __attribute__((packed));
 static_assert(std::is_standard_layout<epilogue_block_t>::value);
 
+static constexpr __u8 FRAME_FLAGS_LATE_ABORT = 0x1;
+
 
 static constexpr uint32_t FRAME_PREAMBLE_SIZE = sizeof(preamble_block_t);
 static constexpr uint32_t FRAME_EPILOGUE_SIZE = sizeof(epilogue_block_t);
@@ -171,7 +181,7 @@ public:
     ::memset(&epilogue, 0, sizeof(epilogue));
 
     ceph::bufferlist::const_iterator hdriter(&this->payload, FRAME_PREAMBLE_SIZE);
-    epilogue.crc_values[SegmentIndex::Frame::PAYLOAD] =
+    epilogue.integrity.crc[SegmentIndex::Frame::PAYLOAD] =
         hdriter.crc32c(hdriter.get_remaining(), -1);
     this->payload.append(reinterpret_cast<const char*>(&epilogue), sizeof(epilogue));
 
@@ -372,12 +382,14 @@ struct SignedEncryptedFrame : public PayloadFrame<T, Args...> {
       session_stream_handlers.tx->authenticated_encrypt_update(
           std::move(c.payload));
       c.payload = session_stream_handlers.tx->authenticated_encrypt_final();
+      // append zeroized epilogue flags. CLEANME.
+      c.payload.append_zero(1);
     } else {
       epilogue_block_t epilogue;
       ::memset(&epilogue, 0, sizeof(epilogue));
 
       ceph::bufferlist::const_iterator hdriter(&c.payload, FRAME_PREAMBLE_SIZE);
-      epilogue.crc_values[SegmentIndex::Frame::PAYLOAD] =
+      epilogue.integrity.crc[SegmentIndex::Frame::PAYLOAD] =
           hdriter.crc32c(hdriter.get_remaining(), -1);
       c.payload.append(reinterpret_cast<const char*>(&epilogue), sizeof(epilogue));
     }
@@ -636,14 +648,18 @@ struct MessageHeaderFrame
 
       // auth tag will be appended at the end
       f.payload = session_stream_handlers.tx->authenticated_encrypt_final();
+      // append zeroized epilogue flags. CLEANME.
+      f.payload.append_zero(1);
     } else {
       epilogue_block_t epilogue;
+      ::memset(&epilogue, 0, sizeof(epilogue));
+
       ceph::bufferlist::const_iterator hdriter(&f.payload, FRAME_PREAMBLE_SIZE);
-      epilogue.crc_values[SegmentIndex::Msg::HEADER] =
+      epilogue.integrity.crc[SegmentIndex::Msg::HEADER] =
           hdriter.crc32c(hdriter.get_remaining(), -1);
-      epilogue.crc_values[SegmentIndex::Msg::FRONT] = front.crc32c(-1),
-      epilogue.crc_values[SegmentIndex::Msg::MIDDLE] = middle.crc32c(-1),
-      epilogue.crc_values[SegmentIndex::Msg::DATA] = data.crc32c(-1),
+      epilogue.integrity.crc[SegmentIndex::Msg::FRONT] = front.crc32c(-1),
+      epilogue.integrity.crc[SegmentIndex::Msg::MIDDLE] = middle.crc32c(-1),
+      epilogue.integrity.crc[SegmentIndex::Msg::DATA] = data.crc32c(-1),
 
       f.payload.append(front);
       f.payload.append(middle);