Method RefreshCacheEntry

be/src/runtime/lib-cache.cc:381–439

}

Status LibCache::RefreshCacheEntry(const string& hdfs_lib_file, LibType type,
    time_t exp_mtime, const LibMap::iterator& iter, unique_lock<mutex>* entry_lock,
    LibCacheEntry** entry) {
  // Check if an error occurred on another thread while loading the library.
  {
    unique_lock<mutex> local_entry_lock((iter->second)->lock);
    if (!(iter->second)->loading_status.ok()) {
      // If loading_status is already set, the returned *entry should be nullptr.
      DCHECK(*entry == nullptr);
      return (iter->second)->loading_status;
    }
  }

  // Refresh the cache entry if needed. A refresh is needed if check_needs_refresh is set
  // (e.g., set by ddl) or if the exp_mtime argument is more recent.
  // If refreshed or an error occurred, remove the entry and set the returned entry to
  // nullptr.
  *entry = iter->second;
  if ((*entry)->check_needs_refresh || (*entry)->last_mod_time < exp_mtime) {
    // Check if file has been modified since loading the cached copy. If so, remove the
    // cached entry and create a new one.
    (*entry)->check_needs_refresh = false;
    hdfsFS hdfs_conn;
    Status status = HdfsFsCache::instance()->GetConnection(hdfs_lib_file, &hdfs_conn);
    if (!status.ok()) {
      RemoveEntryInternal(hdfs_lib_file, iter);
      *entry = nullptr;
      return status;
    }
    time_t fs_last_modified_time;
    status =
        GetLastModificationTime(hdfs_conn, hdfs_lib_file.c_str(), &fs_last_modified_time);

    // Check that the expected last_modified_time is the same as what's on the filesystem.
    if (status.ok() && exp_mtime >= 0 && fs_last_modified_time != exp_mtime) {
      status = Status(TErrorCode::LIB_VERSION_MISMATCH, hdfs_lib_file,
          fs_last_modified_time, exp_mtime);
    }
    if (!status.ok() || (*entry)->last_mod_time < fs_last_modified_time) {
      RemoveEntryInternal(hdfs_lib_file, iter);
      *entry = nullptr;
    }
    RETURN_IF_ERROR(status);
  }

  // No refresh needed, the entry can be used.
  if (*entry != nullptr) {
    // The cache level lock continues to be held while the entry lock is obtained
    // so that some other thread does not access the entry and delete it.
    unique_lock<mutex> local_entry_lock((*entry)->lock);
    entry_lock->swap(local_entry_lock);

    // Let the caller propagate any error that occurred when loading the entry.
    RETURN_IF_ERROR((*entry)->copy_file_status);
    DCHECK_EQ((*entry)->type, type) << (*entry)->local_path;
    DCHECK(!(*entry)->local_path.empty());
  }
  return Status::OK();