Function queue_remove_entries

src/cls/queue/cls_queue_src.cc:469–529

}

int queue_remove_entries(cls_method_context_t hctx, const cls_queue_remove_op& op, cls_queue_head& head)
{
  //Queue is empty
  if ((head.front.offset == head.tail.offset) && (head.front.gen == head.tail.gen)) {
    return 0;
  }

  cls_queue_marker end_marker;
  end_marker.from_str(op.end_marker.c_str());

  CLS_LOG(5, "INFO: queue_remove_entries: op.end_marker = %s", end_marker.to_str().c_str());

  //Zero out the entries that have been removed, to reclaim storage space
  if (end_marker.offset > head.front.offset && end_marker.gen == head.front.gen) {
    uint64_t len = end_marker.offset - head.front.offset;
    if (len > 0) {
      auto ret = cls_cxx_write_zero(hctx, head.front.offset, len);
      if (ret < 0) {
        CLS_LOG(5, "INFO: queue_remove_entries: Failed to zero out entries");
        CLS_LOG(10, "INFO: queue_remove_entries: Start offset = %s", head.front.to_str().c_str());
        return ret;
      }
    }
  } else if ((head.front.offset >= end_marker.offset) && (end_marker.gen == head.front.gen + 1)) { //start offset > end offset
    uint64_t len = head.queue_size - head.front.offset;
    if (len > 0) {
      auto ret = cls_cxx_write_zero(hctx, head.front.offset, len);
      if (ret < 0) {
        CLS_LOG(5, "INFO: queue_remove_entries: Failed to zero out entries");
        CLS_LOG(10, "INFO: queue_remove_entries: Start offset = %s", head.front.to_str().c_str());
        return ret;
      }
    }
    len = end_marker.offset - head.max_head_size;
    if (len > 0) {
      auto ret = cls_cxx_write_zero(hctx, head.max_head_size, len);
      if (ret < 0) {
        CLS_LOG(5, "INFO: queue_remove_entries: Failed to zero out entries");
        CLS_LOG(10, "INFO: queue_remove_entries: Start offset = %lu", head.max_head_size);
        return ret;
      }
    }
  } else if ((head.front.offset == end_marker.offset) && (head.front.gen == end_marker.gen)) {
    //no-op
  } else {
    CLS_LOG(0, "INFO: queue_remove_entries: Invalid end marker: offset = %s, gen = %lu", end_marker.to_str().c_str(), end_marker.gen);
    return -EINVAL;
  }

  head.front = end_marker;

  // Check if it is the end, then wrap around
  if (head.front.offset == head.queue_size) {
    head.front.offset = head.max_head_size;
    head.front.gen += 1;
  }

  CLS_LOG(20, "INFO: queue_remove_entries: front offset is: %s and tail offset is %s", head.front.to_str().c_str(), head.tail.to_str().c_str());