MultiPutObject opt continue

Author: gavinhgchen

include/op/file_upload_task.h

@@ -82,8 +82,12 @@ 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;
+  void SetCheckCrc64(bool check_crc64) {
+    mb_check_crc64 = check_crc64;
+  }
+
+  uint64_t GetCrc64Value() const {
+    return m_crc64_value;
   }
 
  private:
@@ -109,7 +113,8 @@ class FileUploadTask : public Poco::Runnable {
   SSLCtxCallback m_ssl_ctx_cb;
   void *m_user_data;
 
-  uint64_t m_part_crc64;
+  bool mb_check_crc64;
+  uint64_t m_crc64_value;
 };
 
 }  // namespace qcloud_cos

include/op/object_op.h

@@ -20,6 +20,18 @@
 
 namespace qcloud_cos {
 
+class PartBufInfo {
+public:
+  unsigned char* buf;
+  size_t len;
+
+public:
+  PartBufInfo() {
+    buf = nullptr;
+    len = 0;
+  }
+};
+
 class FileUploadTask;
 class FileCopyTask;
 
@@ -447,7 +459,7 @@ class ObjectOp : public BaseOp {
                       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 crc64);
+                      bool check_crc64);
 
   void FillCopyTask(const std::string& upload_id, const std::string& host,
                     const std::string& path, uint64_t part_number,

src/op/file_upload_task.cpp

@@ -34,7 +34,8 @@ FileUploadTask::FileUploadTask(const std::string& full_url,
       m_ca_location(ca_location),
       m_ssl_ctx_cb(ssl_ctx_cb),
       m_user_data(user_data),
-      m_part_crc64(0) {}
+      mb_check_crc64(false),
+      m_crc64_value(0) {}
 
 FileUploadTask::FileUploadTask(
     const std::string& full_url,
@@ -61,7 +62,8 @@ FileUploadTask::FileUploadTask(
       m_ca_location(ca_location),
       m_ssl_ctx_cb(ssl_ctx_cb),
       m_user_data(user_data),
-      m_part_crc64(0) {}
+      mb_check_crc64(false),
+      m_crc64_value(0) {}
 
 FileUploadTask::FileUploadTask(
     const std::string& full_url,
@@ -88,7 +90,8 @@ FileUploadTask::FileUploadTask(
       m_ca_location(ca_location),
       m_ssl_ctx_cb(ssl_ctx_cb),
       m_user_data(user_data),
-      m_part_crc64(0) {}
+      mb_check_crc64(false),
+      m_crc64_value(0) {}
 
 void FileUploadTask::run() {
   m_resp = "";
@@ -158,8 +161,14 @@ void FileUploadTask::SetSslCtxCb(SSLCtxCallback cb, void *data) {
 
 void FileUploadTask::UploadTask() {
   std::string md5_str;
+  // 数据一致性校验采用crc64
+  if (mb_check_crc64) {
+    m_crc64_value = 0;
+    m_crc64_value = CRC64::CalcCRC(m_crc64_value, static_cast<void*>(m_data_buf_ptr), m_data_len);
+    SDK_LOG_DBG("Part Crc64: %" PRIu64, m_crc64_value);
+  }
   // 没有crc64则默认走md5校验
-  if (m_part_crc64 == 0) {
+  else {
   #ifdef USE_OPENSSL_MD5
     unsigned char digest[MD5_DIGEST_LENGTH];
     MD5((const unsigned char *)m_data_buf_ptr, m_data_len, digest);
@@ -174,6 +183,7 @@ void FileUploadTask::UploadTask() {
     dos.close();
     md5_str = Poco::DigestEngine::digestToHex(md5.digest());
   #endif
+    SDK_LOG_DBG("Part Md5: %s", md5_str.c_str());
   }
 
   int loop = 0;
@@ -210,18 +220,19 @@ void FileUploadTask::UploadTask() {
     }
 
     // crc64一致性校验
-    if (m_part_crc64 != 0) {
+    if (mb_check_crc64) {
       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) {
+          StringUtil::StringToUint64(c_itr->second) != m_crc64_value) {
         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_crc64_value, c_itr->second.c_str());
         m_is_task_success = false;
         continue;
       }
+      SDK_LOG_DBG("Part Crc64 Check Success.");
     } else {
       std::map<std::string, std::string>::const_iterator c_itr =
           m_resp_headers.find("ETag");
@@ -234,6 +245,7 @@ void FileUploadTask::UploadTask() {
         m_is_task_success = false;
         continue;
       }
+      SDK_LOG_DBG("Part Md5 Check Success.");
     }
 
     m_is_task_success = true;

src/op/object_op.cpp

@@ -763,6 +763,7 @@ CosResult ObjectOp::MultiUploadObject(const PutObjectByFileReq& req,
       !comp_resp.GetXCosHashCrc64Ecma().empty()) {
     uint64_t crc64_server_resp =
         StringUtil::StringToUint64(comp_resp.GetXCosHashCrc64Ecma());
+    SDK_LOG_DBG("File Crc64: %" PRIu64, crc64_origin);
     if (crc64_server_resp != crc64_origin) {
       std::string err_msg =
           "MultiUploadObject failed, crc64 check failed, crc64_origin: " +
@@ -1762,9 +1763,9 @@ CosResult ObjectOp::MultiThreadUpload(
     return result;
   }
 
-  unsigned char** file_content_buf = new unsigned char*[pool_size];
+  PartBufInfo *part_buf_info = new PartBufInfo[pool_size];
   for (int i = 0; i < pool_size; ++i) {
-    file_content_buf[i] = new unsigned char[(size_t)part_size];
+    part_buf_info[i].buf = new unsigned char[(size_t)part_size];
   }
 
   std::string dest_url = GetRealUrl(host, path, req.IsHttps());
@@ -1796,18 +1797,18 @@ CosResult ObjectOp::MultiThreadUpload(
       }
 
       for (; task_index < pool_size; ++task_index) {
-        fin.read((char*)file_content_buf[task_index], part_size);
+        fin.read((char *)(part_buf_info[task_index].buf), part_size);
         std::streamsize read_len = fin.gcount();
         if (read_len == 0 && fin.eof()) {
           SDK_LOG_DBG("read over, task_index: %d", task_index);
           break;
         }
+        part_buf_info[task_index].len = static_cast<size_t>(read_len);
 
         SDK_LOG_DBG("upload data, task_index=%d, file_size=%" PRIu64
                     ", offset=%" PRIu64 ", len=%" PRIu64,
                     task_index, file_size, offset, read_len);
 
-        uint64_t crc64_part = 0;
         // Check the resume
         FileUploadTask* ptask = pptaskArr[task_index];
 
@@ -1823,25 +1824,11 @@ 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(), crc64_part);
+          FillUploadTask(upload_id, host, path, part_buf_info[task_index].buf,
+                         read_len, part_number, ptask, req.SignHeaderHost(), req.CheckPartCrc64());
           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;
@@ -1887,6 +1874,24 @@ CosResult ObjectOp::MultiThreadUpload(
           task_fail_flag = true;
           break;
         }
+
+        // 根据每个part流式计算整个文件的crc64值
+        if (req.CheckCRC64()) {
+          // 如果已经计算了part的crc64值，只需要直接流式合并即可
+          if (ptask->GetCrc64Value() != 0) {
+            crc64_file = CRC64::CombineCRC(crc64_file, ptask->GetCrc64Value(), 
+                            static_cast<uintmax_t>(part_buf_info[task_index].len));
+            SDK_LOG_DBG("Combine Crc64: %" PRIu64 ", Part Crc64: %" PRIu64, 
+                        crc64_file, ptask->GetCrc64Value());
+          } else {
+            // 两种情况都有可能：
+            // 1、CheckPartCrc64()为false
+            // 2、此part是断点续传已经上传的part
+            crc64_file = CRC64::CalcCRC(crc64_file, static_cast<void *>(part_buf_info[task_index].buf), 
+                            part_buf_info[task_index].len);
+            SDK_LOG_DBG("Calc Crc64: %" PRIu64, crc64_file)
+          }
+        }
       }
 
       if (task_fail_flag) {
@@ -1907,9 +1912,9 @@ CosResult ObjectOp::MultiThreadUpload(
   delete[] pptaskArr;
 
   for (int i = 0; i < pool_size; ++i) {
-    delete[] file_content_buf[i];
+    delete[] part_buf_info[i].buf;
   }
-  delete[] file_content_buf;
+  delete[] part_buf_info;
 
   return result;
 }
@@ -2094,7 +2099,7 @@ 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 crc64) {
+                              bool sign_header_host, bool check_crc64) {
   std::map<std::string, std::string> req_params;
   req_params.insert(std::make_pair("uploadId", upload_id));
   req_params.insert(
@@ -2125,7 +2130,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);
+  task_ptr->SetCheckCrc64(check_crc64);
 }
 
 void ObjectOp::FillCopyTask(const std::string& upload_id,