Add SetCheckPartCrc64() for MultiPutObjectReq

Author: gavinhgchen

include/op/file_upload_task.h

@@ -82,6 +82,10 @@ class FileUploadTask : public Poco::Runnable {
   void SetCaLocation(const std::string& ca_location);
   void SetSslCtxCb(SSLCtxCallback cb, void *data);
 
+  void SetPartCrc64(uint64_t crc64) {
+    m_part_crc64 = crc64;
+  }
+
  private:
   std::string m_full_url;
   std::map<std::string, std::string> m_headers;
@@ -104,6 +108,8 @@ class FileUploadTask : public Poco::Runnable {
   std::string m_ca_location;
   SSLCtxCallback m_ssl_ctx_cb;
   void *m_user_data;
+
+  uint64_t m_part_crc64;
 };
 
 }  // namespace qcloud_cos

include/op/object_op.h

@@ -430,6 +430,7 @@ class ObjectOp : public BaseOp {
                     const std::vector<std::string>& already_exist_parts,
                     bool resume_flag, std::vector<std::string>* etags_ptr,
                     std::vector<uint64_t>* part_numbers_ptr,
+                    uint64_t& crc64_file,
                     const SharedTransferHandler& handler = nullptr,
                     bool change_backup_domain = false);
 
@@ -445,7 +446,8 @@ class ObjectOp : public BaseOp {
   void FillUploadTask(const std::string& upload_id, const std::string& host,
                       const std::string& path, unsigned char* file_content_buf,
                       uint64_t len, uint64_t part_number,
-                      FileUploadTask* task_ptr, bool sign_header_host);
+                      FileUploadTask* task_ptr, bool sign_header_host,
+                      uint64_t crc64);
 
   void FillCopyTask(const std::string& upload_id, const std::string& host,
                     const std::string& path, uint64_t part_number,

include/request/object_req.h

@@ -375,6 +375,7 @@ class PutObjectByFileReq : public PutObjectReq {
 #if defined(_WIN32)
     mb_is_widechar_path = false;
 #endif
+    mb_check_part_crc64 = false;
   }
 
   virtual ~PutObjectByFileReq() {}
@@ -392,6 +393,15 @@ class PutObjectByFileReq : public PutObjectReq {
     m_local_file_path = local_file_path;
   }
 
+  // MultiPutObjectReq: use crc64 instead of md5 for consistency check
+  void SetCheckPartCrc64(bool part_crc64) {
+    mb_check_part_crc64 = part_crc64;
+  }
+
+  bool CheckPartCrc64() const {
+    return mb_check_part_crc64;
+  }
+
   std::string GetLocalFilePath() const { return m_local_file_path; }
 
 #if defined(_WIN32)
@@ -411,6 +421,7 @@ class PutObjectByFileReq : public PutObjectReq {
 #if defined(_WIN32)
   bool mb_is_widechar_path;  // 标识文件路径是否为宽字符
 #endif
+  bool mb_check_part_crc64;
 };
 
 class DeleteObjectReq : public ObjectReq {

src/op/file_upload_task.cpp

@@ -6,6 +6,7 @@
 #include "util/http_sender.h"
 #include "util/string_util.h"
 #include "util/codec_util.h"
+#include "util/crc64.h"
 #ifdef USE_OPENSSL_MD5
 #include <openssl/md5.h>
 #endif
@@ -32,7 +33,8 @@ FileUploadTask::FileUploadTask(const std::string& full_url,
       m_verify_cert(verify_cert),
       m_ca_location(ca_location),
       m_ssl_ctx_cb(ssl_ctx_cb),
-      m_user_data(user_data) {}
+      m_user_data(user_data),
+      m_part_crc64(0) {}
 
 FileUploadTask::FileUploadTask(
     const std::string& full_url,
@@ -58,7 +60,8 @@ FileUploadTask::FileUploadTask(
       m_verify_cert(verify_cert),
       m_ca_location(ca_location),
       m_ssl_ctx_cb(ssl_ctx_cb),
-      m_user_data(user_data) {}
+      m_user_data(user_data),
+      m_part_crc64(0) {}
 
 FileUploadTask::FileUploadTask(
     const std::string& full_url,
@@ -84,7 +87,8 @@ FileUploadTask::FileUploadTask(
       m_verify_cert(verify_cert),
       m_ca_location(ca_location),
       m_ssl_ctx_cb(ssl_ctx_cb),
-      m_user_data(user_data) {}
+      m_user_data(user_data),
+      m_part_crc64(0) {}
 
 void FileUploadTask::run() {
   m_resp = "";
@@ -154,12 +158,13 @@ void FileUploadTask::SetSslCtxCb(SSLCtxCallback cb, void *data) {
 
 void FileUploadTask::UploadTask() {
   std::string md5_str;
-#ifdef USE_OPENSSL_MD5
-  unsigned char digest[MD5_DIGEST_LENGTH];
-  MD5((const unsigned char *)m_data_buf_ptr, m_data_len, digest);
-  md5_str = CodecUtil::DigestToHex(digest, MD5_DIGEST_LENGTH);
-#else
-  {
+  // 没有crc64则默认走md5校验
+  if (m_part_crc64 == 0) {
+  #ifdef USE_OPENSSL_MD5
+    unsigned char digest[MD5_DIGEST_LENGTH];
+    MD5((const unsigned char *)m_data_buf_ptr, m_data_len, digest);
+    md5_str = CodecUtil::DigestToHex(digest, MD5_DIGEST_LENGTH);
+  #else
     // 计算上传的md5
     Poco::MD5Engine md5;
     std::string body((const char*)m_data_buf_ptr, m_data_len);
@@ -168,8 +173,8 @@ void FileUploadTask::UploadTask() {
     Poco::StreamCopier::copyStream(istr, dos);
     dos.close();
     md5_str = Poco::DigestEngine::digestToHex(md5.digest());
+  #endif
   }
-#endif
 
   int loop = 0;
   do {
@@ -204,16 +209,31 @@ void FileUploadTask::UploadTask() {
       continue;
     }
 
-    std::map<std::string, std::string>::const_iterator c_itr =
-        m_resp_headers.find("ETag");
-    if (c_itr == m_resp_headers.end() ||
-        StringUtil::Trim(c_itr->second, "\"") != md5_str) {
-      SDK_LOG_ERR(
-          "Response etag is not correct, try again. Expect md5 is %s, but "
-          "return etag is %s.",
-          md5_str.c_str(), StringUtil::Trim(c_itr->second, "\"").c_str());
-      m_is_task_success = false;
-      continue;
+    // crc64一致性校验
+    if (m_part_crc64 != 0) {
+      std::map<std::string, std::string>::const_iterator c_itr =
+        m_resp_headers.find(kRespHeaderXCosHashCrc64Ecma);
+      if (c_itr == m_resp_headers.end() || 
+          StringUtil::StringToUint64(c_itr->second) != m_part_crc64) {
+        SDK_LOG_ERR(
+            "Response x-cos-hash-crc64ecma is not correct, try again. Expect crc64 is %" PRIu64 ", but "
+            "return crc64 is %s",
+            m_part_crc64, c_itr->second.c_str());
+        m_is_task_success = false;
+        continue;
+      }
+    } else {
+      std::map<std::string, std::string>::const_iterator c_itr =
+          m_resp_headers.find("ETag");
+      if (c_itr == m_resp_headers.end() ||
+          StringUtil::Trim(c_itr->second, "\"") != md5_str) {
+        SDK_LOG_ERR(
+            "Response etag is not correct, try again. Expect md5 is %s, but "
+            "return etag is %s.",
+            md5_str.c_str(), StringUtil::Trim(c_itr->second, "\"").c_str());
+        m_is_task_success = false;
+        continue;
+      }
     }
 
     m_is_task_success = true;

src/op/object_op.cpp

@@ -704,9 +704,11 @@ CosResult ObjectOp::MultiUploadObject(const PutObjectByFileReq& req,
   }
 
   CosResult upload_result;
+
+  uint64_t crc64_origin = 0;
   upload_result =
       MultiThreadUpload(req, resume_uploadid, already_exist_parts, resume_flag,
-                        &etags, &part_numbers, handler, change_backup_domain);
+                        &etags, &part_numbers, crc64_origin, handler, change_backup_domain);
   // Cancel way
   if (handler && !handler->ShouldContinue()) {
     SetResultAndLogError(upload_result, "Request canceled by user");
@@ -759,16 +761,6 @@ CosResult ObjectOp::MultiUploadObject(const PutObjectByFileReq& req,
   // check crc64 if needed
   if (req.CheckCRC64() && comp_result.IsSucc() &&
       !comp_resp.GetXCosHashCrc64Ecma().empty()) {
-    uint64_t crc64_origin = 0;
-#if defined(_WIN32)
-    if (req.IsWideCharPath()) {
-      crc64_origin = FileUtil::GetFileCrc64(req.GetWideCharLocalFilePath());
-    } else {
-      crc64_origin = FileUtil::GetFileCrc64(req.GetLocalFilePath());
-    }
-#else
-    crc64_origin = FileUtil::GetFileCrc64(req.GetLocalFilePath());
-#endif
     uint64_t crc64_server_resp =
         StringUtil::StringToUint64(comp_resp.GetXCosHashCrc64Ecma());
     if (crc64_server_resp != crc64_origin) {
@@ -1704,6 +1696,7 @@ CosResult ObjectOp::MultiThreadUpload(
     const std::vector<std::string>& already_exist_parts, bool resume_flag,
     std::vector<std::string>* etags_ptr,
     std::vector<uint64_t>* part_numbers_ptr,
+    uint64_t& crc64_file,
     const SharedTransferHandler& handler,
     bool change_backup_domain) {
   CosResult result;
@@ -1790,6 +1783,7 @@ CosResult ObjectOp::MultiThreadUpload(
 
   Poco::ThreadPool tp(pool_size);
 
+  crc64_file = 0;
   // 3. 多线程upload
   {
     uint64_t part_number = 1;
@@ -1813,6 +1807,7 @@ CosResult ObjectOp::MultiThreadUpload(
                     ", offset=%" PRIu64 ", len=%" PRIu64,
                     task_index, file_size, offset, read_len);
 
+        uint64_t crc64_part = 0;
         // Check the resume
         FileUploadTask* ptask = pptaskArr[task_index];
 
@@ -1828,11 +1823,25 @@ CosResult ObjectOp::MultiThreadUpload(
             handler->UpdateProgress(read_len);
           }
         } else {
+          // 计算每个part的crc64值
+          if (req.CheckPartCrc64()) {
+            crc64_part = CRC64::CalcCRC(crc64_part, static_cast<void*>(file_content_buf[task_index]), read_len);
+          }
           FillUploadTask(upload_id, host, path, file_content_buf[task_index],
-                         read_len, part_number, ptask, req.SignHeaderHost());
+                         read_len, part_number, ptask, req.SignHeaderHost(), crc64_part);
           tp.start(*ptask);
         }
 
+        // 根据每个part流式计算整个文件的crc64值
+        if (req.CheckCRC64()) {
+          // 如果已经计算了part的crc64值，只需要直接流式合并即可
+          if (crc64_part != 0) {
+            crc64_file = CRC64::CombineCRC(crc64_file, crc64_part, read_len);
+          } else {
+            crc64_file = CRC64::CalcCRC(crc64_file, static_cast<void*>(file_content_buf[task_index]), read_len);
+          }
+        }
+
         offset += read_len;
         part_numbers_ptr->push_back(part_number);
         ++part_number;
@@ -1987,8 +1996,10 @@ CosResult ObjectOp::SingleThreadUpload(
                     part_number, file_size, offset, read_len);
 
         // 提前计算整个文件的crc64，用于整个合并分块完成后做crc64校验
-        crc64 = CRC64::CalcCRC(crc64, static_cast<void*>(file_content_buf),
-                             static_cast<size_t>(read_len));
+        if (req.CheckCRC64()) {
+          crc64 = CRC64::CalcCRC(crc64, static_cast<void*>(file_content_buf),
+                              static_cast<size_t>(read_len));
+        }
 
         // Check the resume
 
@@ -2082,7 +2093,8 @@ uint64_t ObjectOp::GetContent(const std::string& src,
 void ObjectOp::FillUploadTask(const std::string& upload_id,
                               const std::string& host, const std::string& path,
                               unsigned char* file_content_buf, uint64_t len,
-                              uint64_t part_number, FileUploadTask* task_ptr, bool sign_header_host) {
+                              uint64_t part_number, FileUploadTask* task_ptr, 
+                              bool sign_header_host, uint64_t crc64) {
   std::map<std::string, std::string> req_params;
   req_params.insert(std::make_pair("uploadId", upload_id));
   req_params.insert(
@@ -2113,6 +2125,7 @@ void ObjectOp::FillUploadTask(const std::string& upload_id,
   task_ptr->AddHeaders(req_headers);
   task_ptr->SetUploadBuf(file_content_buf, len);
   task_ptr->SetPartNumber(part_number);
+  task_ptr->SetPartCrc64(crc64);
 }
 
 void ObjectOp::FillCopyTask(const std::string& upload_id,

unittest/src/object_op_test.cpp

@@ -2251,6 +2251,7 @@ TEST_F(ObjectOpTest, MultiPutObjectTest_OneStep) {
     // 2. 上传
     MultiPutObjectReq req(m_bucket_name, object_name, filename);
     req.SetXCosServerSideEncryption("AES256");
+    req.SetCheckPartCrc64(true);
     MultiPutObjectResp resp;
 
     CosResult result = m_client->MultiPutObject(req, &resp);
@@ -3430,6 +3431,7 @@ TEST_F(ObjectOpTest, MultiUploadVaryPartSizeAndThreadPoolSize) {
       MultiPutObjectReq multiupload_req(m_bucket_name, object_name, local_file);
       MultiPutObjectResp multiupload_resp;
       ASSERT_TRUE(multiupload_req.CheckCRC64());
+      multiupload_req.SetCheckPartCrc64(true);
 
       // upload object
       CosResult multiupload_result =