void RGWBulkUploadOp::execute()
{
  ceph::bufferlist buffer(64 * 1024);

  ldout(s->cct, 20) << "bulk upload: start" << dendl;

  /* Create an instance of stream-abstracting class. Having this indirection
   * allows for easy introduction of decompressors like gzip and bzip2. */
  auto stream = create_stream();
  if (! stream) {
    return;
  }

  /* Handling the $UPLOAD_PATH accordingly to the Swift's Bulk middleware. See: 
   * https://github.com/openstack/swift/blob/2.13.0/swift/common/middleware/bulk.py#L31-L41 */
  std::string bucket_path, file_prefix;
  std::tie(bucket_path, file_prefix) = handle_upload_path(s);

  auto status = rgw::tar::StatusIndicator::create();
  do {
    op_ret = stream->get_exactly(rgw::tar::BLOCK_SIZE, buffer);
    if (op_ret < 0) {
      ldout(s->cct, 2) << "bulk upload: cannot read header" << dendl;
      return;
    }

    /* We need to re-interpret the buffer as a TAR block. Exactly two blocks
     * must be tracked to detect out end-of-archive. It occurs when both of
     * them are empty (zeroed). Tracing this particular inter-block dependency
     * is responsibility of the rgw::tar::StatusIndicator class. */
    boost::optional<rgw::tar::HeaderView> header;
    std::tie(status, header) = rgw::tar::interpret_block(status, buffer);

    if (! status.empty() && header) {
      /* This specific block isn't empty (entirely zeroed), so we can parse
       * it as a TAR header and dispatch. At the moment we do support only
       * regular files and directories. Everything else (symlinks, devices)
       * will be ignored but won't cease the whole upload. */
      switch (header->get_filetype()) {
        case rgw::tar::FileType::NORMAL_FILE: {
          ldout(s->cct, 2) << "bulk upload: handling regular file" << dendl;

          boost::string_ref filename = bucket_path.empty() ? header->get_filename() : \
                            file_prefix + header->get_filename().to_string();
          auto body = AlignedStreamGetter(0, header->get_filesize(),
                                          rgw::tar::BLOCK_SIZE, *stream);
          op_ret = handle_file(filename,
                               header->get_filesize(),
                               body);
          if (! op_ret) {
            /* Only regular files counts. */
            num_created++;
          } else {
            failures.emplace_back(op_ret, filename.to_string());
          }
          break;
        }
        case rgw::tar::FileType::DIRECTORY: {
          ldout(s->cct, 2) << "bulk upload: handling regular directory" << dendl;

          boost::string_ref dirname = bucket_path.empty() ? header->get_filename() : bucket_path;
          op_ret = handle_dir(dirname);
          if (op_ret < 0 && op_ret != -ERR_BUCKET_EXISTS) {
            failures.emplace_back(op_ret, dirname.to_string());
          }
          break;
        }
        default: {
          /* Not recognized. Skip. */
          op_ret = 0;
          break;
        }
      }

      /* In case of any problems with sub-request authorization Swift simply
       * terminates whole upload immediately. */
      if (boost::algorithm::contains(std::initializer_list<int>{ op_ret },
                                     terminal_errors)) {
        ldout(s->cct, 2) << "bulk upload: terminating due to ret=" << op_ret
                         << dendl;
        break;
      }
    } else {
      ldout(s->cct, 2) << "bulk upload: an empty block" << dendl;
      op_ret = 0;
    }

    buffer.clear();
  } while (! status.eof());

  return;
}