<commit_before>// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include <tuple>

#include "runtime/io/handle-cache.h"
#include "util/hash-util.h"
#include "util/time.h"

#ifndef IMPALA_RUNTIME_DISK_IO_MGR_HANDLE_CACHE_INLINE_H
#define IMPALA_RUNTIME_DISK_IO_MGR_HANDLE_CACHE_INLINE_H

namespace impala {
namespace io {

HdfsFileHandle::HdfsFileHandle(const hdfsFS& fs, const char* fname,
    int64_t mtime)
    : fs_(fs), hdfs_file_(hdfsOpenFile(fs, fname, O_RDONLY, 0, 0, 0)), mtime_(mtime) {
  VLOG_FILE << "hdfsOpenFile() file=" << fname << " fid=" << hdfs_file_;
}

HdfsFileHandle::~HdfsFileHandle() {
  if (hdfs_file_ != nullptr && fs_ != nullptr) {
    VLOG_FILE << "hdfsCloseFile() fid=" << hdfs_file_;
    hdfsCloseFile(fs_, hdfs_file_);
  }
  fs_ = nullptr;
  hdfs_file_ = nullptr;
}

CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const char* fname,
    int64_t mtime)
  : HdfsFileHandle(fs, fname, mtime) {
  ImpaladMetrics::IO_MGR_NUM_CACHED_FILE_HANDLES->Increment(1L);
}

CachedHdfsFileHandle::~CachedHdfsFileHandle() {
  ImpaladMetrics::IO_MGR_NUM_CACHED_FILE_HANDLES->Increment(-1L);
}

FileHandleCache::FileHandleCache(size_t capacity,
      size_t num_partitions, uint64_t unused_handle_timeout_secs)
  : cache_partitions_(num_partitions),
  unused_handle_timeout_secs_(unused_handle_timeout_secs) {
  DCHECK_GT(num_partitions, 0);
  size_t remainder = capacity % num_partitions;
  size_t base_capacity = capacity / num_partitions;
  size_t partition_capacity = (remainder > 0 ? base_capacity + 1 : base_capacity);
  for (FileHandleCachePartition& p : cache_partitions_) {
    p.size = 0;
    p.capacity = partition_capacity;
  }
}

FileHandleCache::LruListEntry::LruListEntry(
    typename MapType::iterator map_entry_in)
     : map_entry(map_entry_in), timestamp_seconds(MonotonicSeconds()) {}

FileHandleCache::~FileHandleCache() {
  shut_down_promise_.Set(true);
  if (eviction_thread_ != nullptr) eviction_thread_->Join();
}

Status FileHandleCache::Init() {
  return Thread::Create("disk-io-mgr-handle-cache", "File Handle Timeout",
      &FileHandleCache::EvictHandlesLoop, this, &eviction_thread_);
}

CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
    const hdfsFS& fs, std::string* fname, int64_t mtime, bool require_new_handle,
    bool* cache_hit) {
  // Hash the key and get appropriate partition
  int index = HashUtil::Hash(fname->data(), fname->size(), 0) % cache_partitions_.size();
  FileHandleCachePartition& p = cache_partitions_[index];
  boost::lock_guard<SpinLock> g(p.lock);
  pair<typename MapType::iterator, typename MapType::iterator> range =
    p.cache.equal_range(*fname);

  // If this requires a new handle, skip to the creation codepath. Otherwise,
  // find an unused entry with the same mtime
  FileHandleEntry* ret_elem = nullptr;
  if (!require_new_handle) {
    while (range.first != range.second) {
      FileHandleEntry* elem = &range.first->second;
      if (!elem->in_use && elem->fh->mtime() == mtime) {
        // This element is currently in the lru_list, which means that lru_entry must
        // be an iterator pointing into the lru_list.
        DCHECK(elem->lru_entry != p.lru_list.end());
        // Remove the element from the lru_list and designate that it is not on
        // the lru_list by resetting its iterator to point to the end of the list.
        p.lru_list.erase(elem->lru_entry);
        elem->lru_entry = p.lru_list.end();
        ret_elem = elem;
        *cache_hit = true;
        break;
      }
      ++range.first;
    }
  }

  // There was no entry that was free or caller asked for a new handle
  if (!ret_elem) {
    *cache_hit = false;
    // Create a new entry and move it into the map
    CachedHdfsFileHandle* new_fh = new CachedHdfsFileHandle(fs, fname->data(), mtime);
    if (!new_fh->ok()) {
      delete new_fh;
      return nullptr;
    }
    FileHandleEntry entry(new_fh, p.lru_list);
    typename MapType::iterator new_it = p.cache.emplace_hint(range.second,
        *fname, std::move(entry));
    ret_elem = &new_it->second;
    ++p.size;
    if (p.size > p.capacity) EvictHandles(p);
  }

  DCHECK(ret_elem->fh.get() != nullptr);
  DCHECK(!ret_elem->in_use);
  ret_elem->in_use = true;
  return ret_elem->fh.get();
}

void FileHandleCache::ReleaseFileHandle(std::string* fname,
    CachedHdfsFileHandle* fh, bool destroy_handle) {
  DCHECK(fh != nullptr);
  // Hash the key and get appropriate partition
  int index = HashUtil::Hash(fname->data(), fname->size(), 0) % cache_partitions_.size();
  FileHandleCachePartition& p = cache_partitions_[index];
  boost::lock_guard<SpinLock> g(p.lock);
  pair<typename MapType::iterator, typename MapType::iterator> range =
    p.cache.equal_range(*fname);

  // TODO: This can be optimized by maintaining some state in the file handle about
  // its location in the map.
  typename MapType::iterator release_it = range.first;
  while (release_it != range.second) {
    FileHandleEntry* elem = &release_it->second;
    if (elem->fh.get() == fh) break;
    ++release_it;
  }<commit_msg>IMPALA-6638: Reduce file handle cache lock contention<commit_after>// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include <tuple>

#include "runtime/io/handle-cache.h"
#include "util/hash-util.h"
#include "util/time.h"

#ifndef IMPALA_RUNTIME_DISK_IO_MGR_HANDLE_CACHE_INLINE_H
#define IMPALA_RUNTIME_DISK_IO_MGR_HANDLE_CACHE_INLINE_H

namespace impala {
namespace io {

HdfsFileHandle::HdfsFileHandle(const hdfsFS& fs, const char* fname,
    int64_t mtime)
    : fs_(fs), hdfs_file_(hdfsOpenFile(fs, fname, O_RDONLY, 0, 0, 0)), mtime_(mtime) {
  VLOG_FILE << "hdfsOpenFile() file=" << fname << " fid=" << hdfs_file_;
}

HdfsFileHandle::~HdfsFileHandle() {
  if (hdfs_file_ != nullptr && fs_ != nullptr) {
    VLOG_FILE << "hdfsCloseFile() fid=" << hdfs_file_;
    hdfsCloseFile(fs_, hdfs_file_);
  }
  fs_ = nullptr;
  hdfs_file_ = nullptr;
}

CachedHdfsFileHandle::CachedHdfsFileHandle(const hdfsFS& fs, const char* fname,
    int64_t mtime)
  : HdfsFileHandle(fs, fname, mtime) {
  ImpaladMetrics::IO_MGR_NUM_CACHED_FILE_HANDLES->Increment(1L);
}

CachedHdfsFileHandle::~CachedHdfsFileHandle() {
  ImpaladMetrics::IO_MGR_NUM_CACHED_FILE_HANDLES->Increment(-1L);
}

FileHandleCache::FileHandleCache(size_t capacity,
      size_t num_partitions, uint64_t unused_handle_timeout_secs)
  : cache_partitions_(num_partitions),
  unused_handle_timeout_secs_(unused_handle_timeout_secs) {
  DCHECK_GT(num_partitions, 0);
  size_t remainder = capacity % num_partitions;
  size_t base_capacity = capacity / num_partitions;
  size_t partition_capacity = (remainder > 0 ? base_capacity + 1 : base_capacity);
  for (FileHandleCachePartition& p : cache_partitions_) {
    p.size = 0;
    p.capacity = partition_capacity;
  }
}

FileHandleCache::LruListEntry::LruListEntry(
    typename MapType::iterator map_entry_in)
     : map_entry(map_entry_in), timestamp_seconds(MonotonicSeconds()) {}

FileHandleCache::~FileHandleCache() {
  shut_down_promise_.Set(true);
  if (eviction_thread_ != nullptr) eviction_thread_->Join();
}

Status FileHandleCache::Init() {
  return Thread::Create("disk-io-mgr-handle-cache", "File Handle Timeout",
      &FileHandleCache::EvictHandlesLoop, this, &eviction_thread_);
}

CachedHdfsFileHandle* FileHandleCache::GetFileHandle(
    const hdfsFS& fs, std::string* fname, int64_t mtime, bool require_new_handle,
    bool* cache_hit) {
  // Hash the key and get appropriate partition
  int index = HashUtil::Hash(fname->data(), fname->size(), 0) % cache_partitions_.size();
  FileHandleCachePartition& p = cache_partitions_[index];

  // If this requires a new handle, skip to the creation codepath. Otherwise,
  // find an unused entry with the same mtime
  if (!require_new_handle) {
    boost::lock_guard<SpinLock> g(p.lock);
    pair<typename MapType::iterator, typename MapType::iterator> range =
      p.cache.equal_range(*fname);

    // When picking a cached entry, always follow the ordering of the map and
    // pick earlier entries first. This allows excessive entries for a file
    // to age out. For example, if there are three entries for a file and only
    // one is used at a time, only the first will be used and the other two
    // can age out.
    while (range.first != range.second) {
      FileHandleEntry* elem = &range.first->second;
      DCHECK(elem->fh.get() != nullptr);
      if (!elem->in_use && elem->fh->mtime() == mtime) {
        // This element is currently in the lru_list, which means that lru_entry must
        // be an iterator pointing into the lru_list.
        DCHECK(elem->lru_entry != p.lru_list.end());
        // Remove the element from the lru_list and designate that it is not on
        // the lru_list by resetting its iterator to point to the end of the list.
        p.lru_list.erase(elem->lru_entry);
        elem->lru_entry = p.lru_list.end();
        *cache_hit = true;
        elem->in_use = true;
        return elem->fh.get();
      }
      ++range.first;
    }
  }

  // There was no entry that was free or caller asked for a new handle
  // Opening a file handle requires talking to the NameNode, so construct
  // the file handle without holding the lock to reduce contention.
  *cache_hit = false;
  // Create a new file handle
  CachedHdfsFileHandle* new_fh = new CachedHdfsFileHandle(fs, fname->data(), mtime);
  if (!new_fh->ok()) {
    delete new_fh;
    return nullptr;
  }

  // Get the lock and create/move the new entry into the map
  // This entry is new and will be immediately used. Place it as the first entry
  // for this file in the multimap. The ordering is largely unimportant if all the
  // existing entries are in use. However, if require_new_handle is true, there may be
  // unused entries, so it would make more sense to insert the new entry at the front.
  boost::lock_guard<SpinLock> g(p.lock);
  pair<typename MapType::iterator, typename MapType::iterator> range =
      p.cache.equal_range(*fname);
  FileHandleEntry entry(new_fh, p.lru_list);
  typename MapType::iterator new_it = p.cache.emplace_hint(range.first,
      *fname, std::move(entry));
  ++p.size;
  if (p.size > p.capacity) EvictHandles(p);
  FileHandleEntry* new_elem = &new_it->second;
  DCHECK(!new_elem->in_use);
  new_elem->in_use = true;
  return new_elem->fh.get();
}

void FileHandleCache::ReleaseFileHandle(std::string* fname,
    CachedHdfsFileHandle* fh, bool destroy_handle) {
  DCHECK(fh != nullptr);
  // Hash the key and get appropriate partition
  int index = HashUtil::Hash(fname->data(), fname->size(), 0) % cache_partitions_.size();
  FileHandleCachePartition& p = cache_partitions_[index];
  boost::lock_guard<SpinLock> g(p.lock);
  pair<typename MapType::iterator, typename MapType::iterator> range =
    p.cache.equal_range(*fname);

  // TODO: This can be optimized by maintaining some state in the file handle about
  // its location in the map.
  typename MapType::iterator release_it = range.first;
  while (release_it != range.second) {
    FileHandleEntry* elem = &release_it->second;
    if (elem->fh.get() == fh) break;
    ++release_it;
  }
{'fraction_non_alphanumeric': 0.11154303847679492, 'fraction_numerical': 0.003966679888932963, 'mean_word_length': 4.062650602409638, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6526079', 'n_tokens_mistral': 4148, 'n_tokens_neox': 4090, 'n_words': 1144}