// 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.

//

// This file is copied from

// https://github.com/apache/impala/blob/master/be/src/runtime/io/handle-cache.h

// and modified by Doris

#pragma once

#include <array>

#include <cstdint>

#include <list>

#include <map>

#include <memory>

#include <mutex>

#include <string>

#include <utility>

#include "common/status.h"

#include "io/fs/file_system.h"

#include "io/fs/hdfs.h"

#include "util/aligned_new.h"

#include "util/lru_multi_cache.inline.h"

#include "util/thread.h"

namespace doris::io {

/// This abstract class is a small wrapper around the hdfsFile handle and the file system

/// instance which is needed to close the file handle. The handle incorporates

/// the last modified time of the file when it was opened. This is used to distinguish

/// between file handles for files that can be updated or overwritten.

/// This is used only through its subclasses, CachedHdfsFileHandle and

/// ExclusiveHdfsFileHandle.

class HdfsFileHandle {

public:

    /// Destructor will close the file handle

    ~HdfsFileHandle();

    /// Init only sets file_size (from param or hdfsGetPathInfo), does NOT open the file.

    Status init(int64_t file_size);

    /// Lazily opens the file handle on first read. Thread-safe via std::call_once.

    Status ensure_open();

    hdfsFS fs() const { return _fs; }

    hdfsFile file() const { return _hdfs_file; }

    int64_t mtime() const { return _mtime; }

    int64_t file_size() const { return _file_size; }

protected:

    HdfsFileHandle(const hdfsFS& fs, const std::string& fname, int64_t mtime)

            : _fs(fs), _fname(fname), _mtime(mtime) {}

private:

    hdfsFS _fs;

    const std::string _fname;

    hdfsFile _hdfs_file = nullptr;

    int64_t _mtime;

    int64_t _file_size = -1;

    std::once_flag _open_once;

    Status _open_status;

};

/// CachedHdfsFileHandles are owned by the file handle cache and are used for no

/// other purpose.

class CachedHdfsFileHandle : public HdfsFileHandle {

public:

    CachedHdfsFileHandle(const hdfsFS& fs, const std::string& fname, int64_t mtime);

    ~CachedHdfsFileHandle();

};

/// ExclusiveHdfsFileHandles are used for all purposes where a CachedHdfsFileHandle

/// is not appropriate.

class ExclusiveHdfsFileHandle : public HdfsFileHandle {

public:

    ExclusiveHdfsFileHandle(const hdfsFS& fs, const std::string& fname, int64_t mtime)

            : HdfsFileHandle(fs, fname, mtime) {}

};

/// The FileHandleCache is a data structure that owns HdfsFileHandles to share between

/// threads. The HdfsFileHandles are hash partitioned across NUM_PARTITIONS partitions.

/// Each partition operates independently with its own locks, reducing contention

/// between concurrent threads. The `capacity` is split between the partitions and is

/// enforced independently.

///

/// Threads check out a file handle for exclusive access, released automatically by RAII

/// accessor. If the file handle is not already present in the cache or all file handles

/// for this file are checked out, the file handle is emplaced in the cache. The cache can

/// contain multiple file handles for the same file. If a file handle is checked out, it

/// cannot be evicted from the cache. In this case, a cache can exceed the specified

/// capacity.

///

/// Remote file systems could keep a connection as part of the file handle without support

/// for unbuffering. The file handle cache is not suitable for those systems, as the cache

/// size can exceed the limit on the number of concurrent connections. HDFS does not

/// maintain a connection in the file handle, S3A client supports unbuffering since

/// IMPALA-8428, so those do not have this restriction.

///

/// If there is a file handle in the cache and the underlying file is deleted,

/// the file handle might keep the file from being deleted at the OS level. This can

/// take up disk space and impact correctness. To avoid this, the cache will evict any

/// file handle that has been unused for longer than threshold specified by

/// `unused_handle_timeout_secs`. Eviction is disabled when the threshold is 0.

///

/// TODO: The cache should also evict file handles more aggressively if the file handle's

/// mtime is older than the file's current mtime.

class FileHandleCache {

private:

    using CacheKey = std::pair<hdfsFS, std::pair<std::string, int64_t>>;

    /// Each partition operates independently, and thus has its own thread-safe cache.

    /// To avoid contention on the lock_ due to false sharing the partitions are

    /// aligned to cache line boundaries.

    struct FileHandleCachePartition : public CacheLineAligned {

        // The same HDFS path can be opened through different hdfsFS instances with

        // different authentication contexts, so the filesystem handle is part of the key.

        typedef LruMultiCache<CacheKey, CachedHdfsFileHandle> CacheType;

        CacheType cache;

    };

public:

    /// RAII accessor built over LruMultiCache::Accessor to handle metrics and unbuffering.

    /// Composition is used instead of inheritance to support the usage as in/out parameter

    class Accessor {

    public:

        Accessor();

        Accessor(FileHandleCachePartition::CacheType::Accessor&& cache_accessor);

        Accessor(Accessor&&) = default;

        Accessor& operator=(Accessor&&) = default;

        DISALLOW_COPY_AND_ASSIGN(Accessor);

        /// Handles metrics and unbuffering

        ~Accessor();

        /// Set function can be used if the Accessor is used as in/out parameter.

        void set(FileHandleCachePartition::CacheType::Accessor&& cache_accessor);

        /// Interface mimics LruMultiCache::Accessor's interface, handles metrics

        CachedHdfsFileHandle* get();

        void release();

        void destroy();

    private:

        FileHandleCachePartition::CacheType::Accessor _cache_accessor;

    };

    /// Instantiates the cache with `capacity` split evenly across NUM_PARTITIONS

    /// partitions. If the capacity does not split evenly, then the capacity is rounded

    /// up. The cache will age out any file handle that is unused for

    /// `unused_handle_timeout_secs` seconds. Age out is disabled if this is set to zero.

    FileHandleCache(size_t capacity, size_t num_partitions, uint64_t unused_handle_timeout_secs);

    /// Destructor is only called for backend tests

    ~FileHandleCache();

    /// Starts up a thread that monitors the age of file handles and evicts any that

    /// exceed the limit.

    Status init() WARN_UNUSED_RESULT;

    /// Get a file handle accessor from the cache for the specified filename (fname) and

    /// last modification time (mtime). This will hash the filename to determine

    /// which partition to use for this file handle.

    ///

    /// If 'require_new_handle' is false and the partition contains an available handle,

    /// an accessor is returned and cache_hit is set to true. Otherwise, the partition will

    /// emplace file handle, an accessor to it will be returned with cache_hit set to false.

    /// On failure, empty accessor will be returned. In either case, the partition may evict

    /// a file handle to make room for the new file handle.

    ///

    /// This obtains exclusive control over the returned file handle.

    Status get_file_handle(const hdfsFS& fs, const std::string& fname, int64_t mtime,

                           int64_t file_size, bool require_new_handle, Accessor* accessor,

                           bool* cache_hit) WARN_UNUSED_RESULT;

#ifdef BE_TEST

    static bool same_cache_key_for_test(const hdfsFS& lhs_fs, const std::string& lhs_fname,

                                        int64_t lhs_mtime, const hdfsFS& rhs_fs,

                                        const std::string& rhs_fname, int64_t rhs_mtime);

#endif

private:

    static CacheKey make_cache_key(const hdfsFS& fs, const std::string& fname, int64_t mtime) {

        return {fs, {fname, mtime}};

    }

    /// Periodic check to evict unused file handles. Only executed by _eviction_thread.

    void _evict_handles_loop();

    std::vector<FileHandleCachePartition> _cache_partitions;

    /// Maximum time before an unused file handle is aged out of the cache.

    /// Aging out is disabled if this is set to 0.

    uint64_t _unused_handle_timeout_secs;

    /// Thread to check for unused file handles to evict. This thread will exit when

    /// the _shut_down_promise is set.

    std::shared_ptr<Thread> _eviction_thread;

    std::atomic<bool> _is_shut_down = {false};

};

} // namespace doris::io