Method SubmitStoreTask

be/src/runtime/io/data-cache.cc:1438–1475

}

bool DataCache::SubmitStoreTask(const std::string& filename, int64_t mtime,
    int64_t offset, const uint8_t* buffer, int64_t buffer_len) {
  const int64_t charge_len = BitUtil::RoundUp(buffer_len, PAGE_SIZE);
  if (UNLIKELY(charge_len > per_partition_capacity_)) return false;

  // Tries to increase the current_buffer_size_ by buffer_len before allocate buffer.
  // If new size exceeds store_buffer_capacity_, return false and current_buffer_size_ is
  // not changed.
  while (true) {
    int64_t current_size = current_buffer_size_.Load();
    int64_t new_size = current_size + buffer_len;
    if (UNLIKELY(new_size > store_buffer_capacity_)) {
      VLOG(2) << Substitute("Failed to create store task due to buffer size limitation, "
          "current buffer size: $0 size limitation: $1 require: $2",
          current_size, store_buffer_capacity_, buffer_len);
      ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_ASYNC_WRITES_DROPPED_BYTES->
          Increment(buffer_len);
      ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_ASYNC_WRITES_DROPPED_ENTRIES->Increment(1);
      return false;
    }
    if (LIKELY(current_buffer_size_.CompareAndSwap(current_size, new_size))) {
      ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_ASYNC_WRITES_OUTSTANDING_BYTES->SetValue(
          current_buffer_size_.Load());
      break;
    }
  }

  DCHECK(buffer != nullptr);
  // TODO: Should we use buffer pool instead of piecemeal memory allocate?
  uint8_t* task_buffer = new uint8_t[buffer_len];
  memcpy(task_buffer, buffer, buffer_len);

  const StoreTask* task =
      new StoreTask(filename, mtime, offset, task_buffer, buffer_len, this);
  storer_pool_->Offer(StoreTaskHandle(task));
  ImpaladMetrics::IO_MGR_REMOTE_DATA_CACHE_NUM_ASYNC_WRITES_SUBMITTED->Increment(1);
  return true;
}

void DataCache::CompleteStoreTask(const StoreTask& task) {
  current_buffer_size_.Add(-task.buffer_len());