Function cls_2pc_queue_remove_entries

src/cls/2pc_queue/cls_2pc_queue.cc:651–703

}

static int cls_2pc_queue_remove_entries(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  auto in_iter = in->cbegin();
  cls_2pc_queue_remove_op rem_2pc_op;
  try {
    decode(rem_2pc_op, in_iter);
  } catch (ceph::buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_2pc_queue_remove_entries: failed to decode entry: %s", err.what());
    return -EINVAL;
  }

  cls_queue_head head;
  auto ret = queue_read_head(hctx, head);
  if (ret < 0) {
    return ret;
  }

  // Old RGW is running, and it sent cls_queue_remove_op instead of cls_2pc_queue_remove_op
  if (rem_2pc_op.entries_to_remove == 0) {
    CLS_LOG(10, "INFO: cls_2pc_queue_remove_entries: incompatible RGW with rados, counting entries to remove...");
    cls_queue_list_op list_op;
    list_op.max = std::numeric_limits<uint64_t>::max(); // max length because endmarker is the stopping condition.
    list_op.end_marker = rem_2pc_op.end_marker;
    ret = cls_2pc_queue_count_entries(hctx, list_op, head, rem_2pc_op.entries_to_remove);
    if (ret < 0) {
      CLS_LOG(1, "ERROR: cls_2pc_queue_remove_entries: returned: %d", ret);
      return ret;
    }
    CLS_LOG(10, "INFO: cls_2pc_queue_remove_entries: counted: %u", rem_2pc_op.entries_to_remove);
  }

  cls_queue_remove_op rem_op;
  rem_op.end_marker = std::move(rem_2pc_op.end_marker);
  ret = queue_remove_entries(hctx, rem_op, head);
  if (ret < 0) {
    return ret;
  }

  cls_2pc_urgent_data urgent_data;
  try {
    auto in_iter = head.bl_urgent_data.cbegin();
    decode(urgent_data, in_iter);
  } catch (ceph::buffer::error& err) {
    CLS_LOG(1, "ERROR: cls_2pc_queue_remove_entries: failed to decode header of queue: %s", err.what());
    return -EINVAL;
  }
  urgent_data.committed_entries -= rem_2pc_op.entries_to_remove;
  // write back head
  head.bl_urgent_data.clear();
  encode(urgent_data, head.bl_urgent_data);

  return queue_write_head(hctx, head);
}

CLS_INIT(2pc_queue)
{