async_msgr: crc configuration in messenger

Add new header_crc and data_crc configuration booleans, and use
them consistently to govern whether CRC is performed in the
Message encode, decode, and transit paths.

Learn from 2ffacbe6efae05f5c426cae34882b3351a5ccfbe

Signed-off-by: Haomai Wang <haomaiwang@gmail.com>

diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 3a730edcf39b950b71ee3cda6376fcecc2ed5a8c..d2db0ebd0adf9db258f72d6966f437051fea21b8 100644 (file)
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -575,16 +575,19 @@ void AsyncConnection::process()
 
           if (has_feature(CEPH_FEATURE_NOSRCADDR)) {
             header = *((ceph_msg_header*)state_buffer);
-            header_crc = ceph_crc32c(0, (unsigned char *)&header,
-                                    sizeof(header) - sizeof(header.crc));
+            if (msgr->crcflags & MSG_CRC_HEADER)
+              header_crc = ceph_crc32c(0, (unsigned char *)&header,
+                                       sizeof(header) - sizeof(header.crc));
           } else {
             oldheader = *((ceph_msg_header_old*)state_buffer);
             // this is fugly
             memcpy(&header, &oldheader, sizeof(header));
             header.src = oldheader.src.name;
             header.reserved = oldheader.reserved;
-            header.crc = oldheader.crc;
-            header_crc = ceph_crc32c(0, (unsigned char *)&oldheader, sizeof(oldheader) - sizeof(oldheader.crc));
+            if (msgr->crcflags & MSG_CRC_HEADER) {
+              header.crc = oldheader.crc;
+              header_crc = ceph_crc32c(0, (unsigned char *)&oldheader, sizeof(oldheader) - sizeof(oldheader.crc));
+            }
           }
 
           ldout(async_msgr->cct, 20) << __func__ << " got envelope type=" << header.type
@@ -594,7 +597,8 @@ void AsyncConnection::process()
                               << " off " << header.data_off << dendl;
 
           // verify header crc
-          if (header_crc != header.crc) {
+          if (!(msgr->crcflags & MSG_CRC_HEADER)) {
+          } else if (header_crc != header.crc) {
             ldout(async_msgr->cct,0) << __func__ << " reader got bad header crc "
                               << header_crc << " != " << header.crc << dendl;
             goto fail;
@@ -765,9 +769,11 @@ void AsyncConnection::process()
             footer = *((ceph_msg_footer*)state_buffer);
           } else {
             old_footer = *((ceph_msg_footer_old*)state_buffer);
-            footer.front_crc = old_footer.front_crc;
-            footer.middle_crc = old_footer.middle_crc;
-            footer.data_crc = old_footer.data_crc;
+            if (msgr->crcflags & MSG_CRC_HEADER) {
+              footer.front_crc = old_footer.front_crc;
+              footer.middle_crc = old_footer.middle_crc;
+              footer.data_crc = old_footer.data_crc;
+            }
             footer.sig = 0;
             footer.flags = old_footer.flags;
           }
@@ -2177,8 +2183,12 @@ int AsyncConnection::write_message(ceph_msg_header& header, ceph_msg_footer& foo
     oldheader.src.addr = get_peer_addr();
     oldheader.orig_src = oldheader.src;
     oldheader.reserved = header.reserved;
-    oldheader.crc = ceph_crc32c(0, (unsigned char*)&oldheader,
-                                sizeof(oldheader) - sizeof(oldheader.crc));
+    if (msgr->crcflags & MSG_CRC_HEADER) {
+       oldheader.crc = ceph_crc32c(0, (unsigned char*)&oldheader,
+                                   sizeof(oldheader) - sizeof(oldheader.crc));
+    } else {
+       oldheader.crc = 0;
+    }
     bl.append((char*)&oldheader, sizeof(oldheader));
   }
 
@@ -2189,9 +2199,13 @@ int AsyncConnection::write_message(ceph_msg_header& header, ceph_msg_footer& foo
   if (has_feature(CEPH_FEATURE_MSG_AUTH)) {
     bl.append((char*)&footer, sizeof(footer));
   } else {
-    old_footer.front_crc = footer.front_crc;
-    old_footer.middle_crc = footer.middle_crc;
-    old_footer.data_crc = footer.data_crc;
+    if (msgr->crcflags & MSG_CRC_HEADER) {
+      old_footer.front_crc = footer.front_crc;
+      old_footer.middle_crc = footer.middle_crc;
+      old_footer.data_crc = footer.data_crc;
+    } else {
+       old_footer.front_crc = old_footer.middle_crc = old_footer.data_crc = 0;
+    }
     old_footer.flags = footer.flags;
     bl.append((char*)&old_footer, sizeof(old_footer));
   }