void RGWPutObj::execute()
{
  RGWPutObjProcessor *processor = NULL;
  RGWPutObjDataProcessor *filter = nullptr;
  std::unique_ptr<RGWPutObjDataProcessor> encrypt;
  char supplied_md5_bin[CEPH_CRYPTO_MD5_DIGESTSIZE + 1];
  char supplied_md5[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 1];
  char calc_md5[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 1];
  unsigned char m[CEPH_CRYPTO_MD5_DIGESTSIZE];
  MD5 hash;
  bufferlist bl, aclbl, bs;
  int len;
  bool multipart;
  
  off_t fst;
  off_t lst;
  const auto& compression_type = store->get_zone_params().get_compression_type(
      s->bucket_info.placement_rule);
  CompressorRef plugin;
  boost::optional<RGWPutObj_Compress> compressor;

  bool need_calc_md5 = (dlo_manifest == NULL) && (slo_info == NULL);
  perfcounter->inc(l_rgw_put);
  op_ret = -EINVAL;
  if (s->object.empty()) {
    goto done;
  }

  if (!s->bucket_exists) {
    op_ret = -ERR_NO_SUCH_BUCKET;
    return;
  }


  op_ret = get_system_versioning_params(s, &olh_epoch, &version_id);
  if (op_ret < 0) {
    ldout(s->cct, 20) << "get_system_versioning_params() returned ret="
		      << op_ret << dendl;
    goto done;
  }

  if (supplied_md5_b64) {
    need_calc_md5 = true;

    ldout(s->cct, 15) << "supplied_md5_b64=" << supplied_md5_b64 << dendl;
    op_ret = ceph_unarmor(supplied_md5_bin, &supplied_md5_bin[CEPH_CRYPTO_MD5_DIGESTSIZE + 1],
                       supplied_md5_b64, supplied_md5_b64 + strlen(supplied_md5_b64));
    ldout(s->cct, 15) << "ceph_armor ret=" << op_ret << dendl;
    if (op_ret != CEPH_CRYPTO_MD5_DIGESTSIZE) {
      op_ret = -ERR_INVALID_DIGEST;
      goto done;
    }

    buf_to_hex((const unsigned char *)supplied_md5_bin, CEPH_CRYPTO_MD5_DIGESTSIZE, supplied_md5);
    ldout(s->cct, 15) << "supplied_md5=" << supplied_md5 << dendl;
  }

  if (!chunked_upload) { /* with chunked upload we don't know how big is the upload.
                            we also check sizes at the end anyway */
    op_ret = store->check_quota(s->bucket_owner.get_id(), s->bucket,
				user_quota, bucket_quota, s->content_length);
    if (op_ret < 0) {
      ldout(s->cct, 20) << "check_quota() returned ret=" << op_ret << dendl;
      goto done;
    }
    op_ret = store->check_bucket_shards(s->bucket_info, s->bucket, bucket_quota);
    if (op_ret < 0) {
      ldout(s->cct, 20) << "check_bucket_shards() returned ret=" << op_ret << dendl;
      goto done;
    }
  }

  if (supplied_etag) {
    strncpy(supplied_md5, supplied_etag, sizeof(supplied_md5) - 1);
    supplied_md5[sizeof(supplied_md5) - 1] = '\0';
  }

  processor = select_processor(*static_cast<RGWObjectCtx *>(s->obj_ctx), &multipart);

  // no filters by default
  filter = processor;

  /* Handle object versioning of Swift API. */
  if (! multipart) {
    rgw_obj obj(s->bucket, s->object);
    op_ret = store->swift_versioning_copy(*static_cast<RGWObjectCtx *>(s->obj_ctx),
                                          s->bucket_owner.get_id(),
                                          s->bucket_info,
                                          obj);
    if (op_ret < 0) {
      goto done;
    }
  }

  op_ret = processor->prepare(store, NULL);
  if (op_ret < 0) {
    ldout(s->cct, 20) << "processor->prepare() returned ret=" << op_ret
		      << dendl;
    goto done;
  }

  if ((! copy_source.empty()) && !copy_source_range) {
    rgw_obj_key obj_key(copy_source_object_name, copy_source_version_id);
    rgw_obj obj(copy_source_bucket_info.bucket, obj_key.name);

    RGWObjState *astate;
    op_ret = store->get_obj_state(static_cast<RGWObjectCtx *>(s->obj_ctx),
                                  copy_source_bucket_info, obj, &astate, true, false);
    if (op_ret < 0) {
      ldout(s->cct, 0) << "ERROR: get copy source obj state returned with error" << op_ret << dendl;
      goto done;
    }
    if (!astate->exists){
      op_ret = -ENOENT;
      goto done;
    }
    lst = astate->accounted_size - 1;
  } else {
    lst = copy_source_range_lst;
  }

  fst = copy_source_range_fst;

  op_ret = get_encrypt_filter(&encrypt, filter);
  if (op_ret < 0) {
    goto done;
  }
  if (encrypt != nullptr) {
    filter = encrypt.get();
  } else {
    //no encryption, we can try compression
    if (compression_type != "none") {
      plugin = get_compressor_plugin(s, compression_type);
      if (!plugin) {
        ldout(s->cct, 1) << "Cannot load plugin for compression type "
            << compression_type << dendl;
      } else {
        compressor.emplace(s->cct, plugin, filter);
        filter = &*compressor;
      }
    }
  }

  tracepoint(rgw_op, before_data_transfer, s->req_id.c_str());
  do {
    bufferlist data;
    if (fst > lst)
      break;
    if (copy_source.empty()) {
      len = get_data(data);
    } else {
      uint64_t cur_lst = min(fst + s->cct->_conf->rgw_max_chunk_size - 1, lst);
      op_ret = get_data(fst, cur_lst, data);
      if (op_ret < 0)
        goto done;
      len = data.length();
      s->content_length += len;
      fst += len;
    }
    if (len < 0) {
      op_ret = len;
      ldout(s->cct, 20) << "get_data() returned ret=" << op_ret << dendl;
      goto done;
    }

    if (need_calc_md5) {
      hash.Update((const unsigned char *)data.c_str(), data.length());
    }

    /* update torrrent */
    torrent.update(data);

    /* do we need this operation to be synchronous? if we're dealing with an object with immutable
     * head, e.g., multipart object we need to make sure we're the first one writing to this object
     */
    bool need_to_wait = (ofs == 0) && multipart;

    bufferlist orig_data;

    if (need_to_wait) {
      orig_data = data;
    }

    op_ret = put_data_and_throttle(filter, data, ofs, need_to_wait);
    if (op_ret < 0) {
      if (op_ret != -EEXIST) {
        ldout(s->cct, 20) << "processor->thottle_data() returned ret="
			  << op_ret << dendl;
        goto done;
      }
      /* need_to_wait == true and op_ret == -EEXIST */
      ldout(s->cct, 5) << "NOTICE: processor->throttle_data() returned -EEXIST, need to restart write" << dendl;

      /* restore original data */
      data.swap(orig_data);

      /* restart processing with different oid suffix */

      dispose_processor(processor);
      processor = select_processor(*static_cast<RGWObjectCtx *>(s->obj_ctx), &multipart);
      filter = processor;

      string oid_rand;
      char buf[33];
      gen_rand_alphanumeric(store->ctx(), buf, sizeof(buf) - 1);
      oid_rand.append(buf);

      op_ret = processor->prepare(store, &oid_rand);
      if (op_ret < 0) {
        ldout(s->cct, 0) << "ERROR: processor->prepare() returned "
			 << op_ret << dendl;
        goto done;
      }

      op_ret = get_encrypt_filter(&encrypt, filter);
      if (op_ret < 0) {
        goto done;
      }
      if (encrypt != nullptr) {
        filter = encrypt.get();
      } else {
        if (compressor) {
          compressor.emplace(s->cct, plugin, filter);
          filter = &*compressor;
        }
      }
      op_ret = put_data_and_throttle(filter, data, ofs, false);
      if (op_ret < 0) {
        goto done;
      }
    }

    ofs += len;
  } while (len > 0);
  tracepoint(rgw_op, after_data_transfer, s->req_id.c_str(), ofs);

  {
    bufferlist flush;
    op_ret = put_data_and_throttle(filter, flush, ofs, false);
    if (op_ret < 0) {
      goto done;
    }
  }

  if (!chunked_upload && ofs != s->content_length) {
    op_ret = -ERR_REQUEST_TIMEOUT;
    goto done;
  }
  s->obj_size = ofs;

  perfcounter->inc(l_rgw_put_b, s->obj_size);

  op_ret = do_aws4_auth_completion();
  if (op_ret < 0) {
    goto done;
  }

  op_ret = store->check_quota(s->bucket_owner.get_id(), s->bucket,
                              user_quota, bucket_quota, s->obj_size);
  if (op_ret < 0) {
    ldout(s->cct, 20) << "second check_quota() returned op_ret=" << op_ret << dendl;
    goto done;
  }

  op_ret = store->check_bucket_shards(s->bucket_info, s->bucket, bucket_quota);
  if (op_ret < 0) {
    ldout(s->cct, 20) << "check_bucket_shards() returned ret=" << op_ret << dendl;
    goto done;
  }

  hash.Final(m);

  if (compressor && compressor->is_compressed()) {
    bufferlist tmp;
    RGWCompressionInfo cs_info;
    cs_info.compression_type = plugin->get_type_name();
    cs_info.orig_size = s->obj_size;
    cs_info.blocks = move(compressor->get_compression_blocks());
    encode(cs_info, tmp);
    attrs[RGW_ATTR_COMPRESSION] = tmp;
    ldout(s->cct, 20) << "storing " << RGW_ATTR_COMPRESSION
        << " with type=" << cs_info.compression_type
        << ", orig_size=" << cs_info.orig_size
        << ", blocks=" << cs_info.blocks.size() << dendl;
  }

  buf_to_hex(m, CEPH_CRYPTO_MD5_DIGESTSIZE, calc_md5);

  etag = calc_md5;

  if (supplied_md5_b64 && strcmp(calc_md5, supplied_md5)) {
    op_ret = -ERR_BAD_DIGEST;
    goto done;
  }

  policy.encode(aclbl);
  emplace_attr(RGW_ATTR_ACL, std::move(aclbl));

  if (dlo_manifest) {
    op_ret = encode_dlo_manifest_attr(dlo_manifest, attrs);
    if (op_ret < 0) {
      ldout(s->cct, 0) << "bad user manifest: " << dlo_manifest << dendl;
      goto done;
    }
    complete_etag(hash, &etag);
    ldout(s->cct, 10) << __func__ << ": calculated md5 for user manifest: " << etag << dendl;
  }

  if (slo_info) {
    bufferlist manifest_bl;
    encode(*slo_info, manifest_bl);
    emplace_attr(RGW_ATTR_SLO_MANIFEST, std::move(manifest_bl));

    hash.Update((unsigned char *)slo_info->raw_data, slo_info->raw_data_len);
    complete_etag(hash, &etag);
    ldout(s->cct, 10) << __func__ << ": calculated md5 for user manifest: " << etag << dendl;
  }

  if (supplied_etag && etag.compare(supplied_etag) != 0) {
    op_ret = -ERR_UNPROCESSABLE_ENTITY;
    goto done;
  }
  bl.append(etag.c_str(), etag.size());
  emplace_attr(RGW_ATTR_ETAG, std::move(bl));

  populate_with_generic_attrs(s, attrs);
  op_ret = rgw_get_request_metadata(s->cct, s->info, attrs);
  if (op_ret < 0) {
    goto done;
  }
  encode_delete_at_attr(delete_at, attrs);
  encode_obj_tags_attr(obj_tags.get(), attrs);

  /* Add a custom metadata to expose the information whether an object
   * is an SLO or not. Appending the attribute must be performed AFTER
   * processing any input from user in order to prohibit overwriting. */
  if (slo_info) {
    bufferlist slo_userindicator_bl;
    slo_userindicator_bl.append("True", 4);
    emplace_attr(RGW_ATTR_SLO_UINDICATOR, std::move(slo_userindicator_bl));
  }

  tracepoint(rgw_op, processor_complete_enter, s->req_id.c_str());
  op_ret = processor->complete(s->obj_size, etag, &mtime, real_time(), attrs,
                               (delete_at ? *delete_at : real_time()), if_match, if_nomatch,
                               (user_data.empty() ? nullptr : &user_data));
  tracepoint(rgw_op, processor_complete_exit, s->req_id.c_str());

  // only atomic upload will upate version_id here
  if (!multipart) 
    version_id = (static_cast<RGWPutObjProcessor_Atomic *>(processor))->get_version_id();

  /* produce torrent */
  if (s->cct->_conf->rgw_torrent_flag && (ofs == torrent.get_data_len()))
  {
    torrent.init(s, store);
    torrent.set_create_date(mtime);
    op_ret =  torrent.complete();
    if (0 != op_ret)
    {
      ldout(s->cct, 0) << "ERROR: torrent.handle_data() returned " << op_ret << dendl;
      goto done;
    }
  }

done:
  dispose_processor(processor);
  perfcounter->tinc(l_rgw_put_lat, s->time_elapsed());
}