void RGWPutACLs::execute()
{
  bufferlist bl;

  RGWAccessControlPolicy_S3 *policy = NULL;
  RGWACLXMLParser_S3 parser(s->cct);
  RGWAccessControlPolicy_S3 new_policy(s->cct);
  stringstream ss;
  char *new_data = NULL;
  rgw_obj obj;

  op_ret = 0; /* XXX redundant? */

  if (!parser.init()) {
    op_ret = -EINVAL;
    return;
  }


  RGWAccessControlPolicy* const existing_policy = \
    (s->object.empty() ? s->bucket_acl.get() : s->object_acl.get());

  owner = existing_policy->get_owner();

  op_ret = get_params();
  if (op_ret < 0) {
    if (op_ret == -ERANGE) {
      ldout(s->cct, 4) << "The size of request xml data is larger than the max limitation, data size = "
                       << s->length << dendl;
      op_ret = -ERR_MALFORMED_XML;
      s->err.message = "The XML you provided was larger than the maximum " +
                       std::to_string(s->cct->_conf->rgw_max_put_param_size) +
                       " bytes allowed.";
    }
    return;
  }

  ldout(s->cct, 15) << "read len=" << len << " data=" << (data ? data : "") << dendl;

  if (!s->canned_acl.empty() && len) {
    op_ret = -EINVAL;
    return;
  }

  if (!s->canned_acl.empty() || s->has_acl_header) {
    op_ret = get_policy_from_state(store, s, ss);
    if (op_ret < 0)
      return;

    new_data = strdup(ss.str().c_str());
    free(data);
    data = new_data;
    len = ss.str().size();
  }

  if (!parser.parse(data, len, 1)) {
    op_ret = -EINVAL;
    return;
  }
  policy = static_cast<RGWAccessControlPolicy_S3 *>(parser.find_first("AccessControlPolicy"));
  if (!policy) {
    op_ret = -EINVAL;
    return;
  }

  const RGWAccessControlList& req_acl = policy->get_acl();
  const multimap<string, ACLGrant>& req_grant_map = req_acl.get_grant_map();
#define ACL_GRANTS_MAX_NUM      100
  int max_num = s->cct->_conf->rgw_acl_grants_max_num;
  if (max_num < 0) {
    max_num = ACL_GRANTS_MAX_NUM;
  }

  int grants_num = req_grant_map.size();
  if (grants_num > max_num) {
    ldout(s->cct, 4) << "An acl can have up to "
                     << max_num
                     << " grants, request acl grants num: "
                     << grants_num << dendl;
    op_ret = -ERR_MALFORMED_ACL_ERROR;
    s->err.message = "The request is rejected, because the acl grants number you requested is larger than the maximum "
                     + std::to_string(max_num)
                     + " grants allowed in an acl.";
    return;
  }

  // forward bucket acl requests to meta master zone
  if (s->object.empty() && !store->is_meta_master()) {
    bufferlist in_data;
    // include acl data unless it was generated from a canned_acl
    if (s->canned_acl.empty()) {
      in_data.append(data, len);
    }
    op_ret = forward_request_to_master(s, NULL, store, in_data, NULL);
    if (op_ret < 0) {
      ldout(s->cct, 20) << __func__ << " forward_request_to_master returned ret=" << op_ret << dendl;
      return;
    }
  }

  if (s->cct->_conf->subsys.should_gather<ceph_subsys_rgw, 15>()) {
    ldout(s->cct, 15) << "Old AccessControlPolicy";
    policy->to_xml(*_dout);
    *_dout << dendl;
  }

  op_ret = policy->rebuild(store, &owner, new_policy);
  if (op_ret < 0)
    return;

  if (s->cct->_conf->subsys.should_gather<ceph_subsys_rgw, 15>()) {
    ldout(s->cct, 15) << "New AccessControlPolicy:";
    new_policy.to_xml(*_dout);
    *_dout << dendl;
  }

  new_policy.encode(bl);
  map<string, bufferlist> attrs;

  if (!s->object.empty()) {
    obj = rgw_obj(s->bucket, s->object);
    store->set_atomic(s->obj_ctx, obj);
    //if instance is empty, we should modify the latest object
    op_ret = modify_obj_attr(store, s, obj, RGW_ATTR_ACL, bl);
  } else {
    attrs = s->bucket_attrs;
    attrs[RGW_ATTR_ACL] = bl;
    op_ret = rgw_bucket_set_attrs(store, s->bucket_info, attrs, &s->bucket_info.objv_tracker);
  }
  if (op_ret == -ECANCELED) {
    op_ret = 0; /* lost a race, but it's ok because acls are immutable */
  }
}