#pragma once

#include <atomic>

#include <chrono>

#include <cstdint>

#include <list>

#include <memory>

#include <string>

#include "envoy/stats/stats_matcher.h"

#include "envoy/stats/tag.h"

#include "envoy/thread_local/thread_local.h"

#include "source/common/common/hash.h"

#include "source/common/common/thread_synchronizer.h"

#include "source/common/stats/allocator.h"

#include "source/common/stats/histogram_impl.h"

#include "source/common/stats/null_counter.h"

#include "source/common/stats/null_gauge.h"

#include "source/common/stats/null_text_readout.h"

#include "source/common/stats/symbol_table.h"

#include "source/common/stats/utility.h"

#include "absl/container/flat_hash_map.h"

#include "circllhist.h"

namespace Envoy {

namespace Stats {

/**

 * A histogram that is stored in TLS and used to record values per thread. This holds two

 * histograms, one to collect the values and other as backup that is used for merge process. The

 * swap happens during the merge process.

 */

class ThreadLocalHistogramImpl : public HistogramImplHelper {

public:

  ThreadLocalHistogramImpl(StatName name, Histogram::Unit unit, StatName tag_extracted_name,

                           const StatNameTagVector& stat_name_tags, SymbolTable& symbol_table,

                           absl::optional<uint32_t> bins);

  ~ThreadLocalHistogramImpl() override;

  void merge(histogram_t* target);

  /**

   * Called in the beginning of merge process. Swaps the histogram used for collection so that we do

   * not have to lock the histogram in high throughput TLS writes.

   */

    // This switches the current_active_ between 1 and 0.

  // Stats::Histogram

    // If at some point ThreadLocalHistogramImpl will hold a pointer to its parent we can just

    // return parent's unit here and not store it separately.

  void recordValue(uint64_t value) override;

  // Stats::Metric

private:

  const Histogram::Unit unit_;

  uint64_t current_active_{0};

  histogram_t* histograms_[2];

  std::atomic<bool> used_;

  const std::thread::id created_thread_id_;

  SymbolTable& symbol_table_;

};

using TlsHistogramSharedPtr = RefcountPtr<ThreadLocalHistogramImpl>;

class ThreadLocalStoreImpl;

/**

 * Log Linear Histogram implementation that is stored in the main thread.

 */

class ParentHistogramImpl : public MetricImpl<ParentHistogram> {

public:

  ParentHistogramImpl(StatName name, Histogram::Unit unit, ThreadLocalStoreImpl& parent,

                      StatName tag_extracted_name, const StatNameTagVector& stat_name_tags,

                      ConstSupportedBuckets& supported_buckets, absl::optional<uint32_t> bins,

                      uint64_t id);

  ~ParentHistogramImpl() override;

  void addTlsHistogram(const TlsHistogramSharedPtr& hist_ptr);

  // Stats::Histogram

  Histogram::Unit unit() const override;

  void recordValue(uint64_t value) override;

  /**

   * This method is called during the main stats flush process for each of the histograms. It

   * iterates through the TLS histograms and collects the histogram data of all of them

   * in to "interval_histogram". Then the collected "interval_histogram" is merged to a

   * "cumulative_histogram".

   */

  void merge() override;

  std::string quantileSummary() const override;

  std::string bucketSummary() const override;

  uint64_t cumulativeCountLessThanOrEqualToValue(double value) const override;

  // Stats::Metric

  SymbolTable& symbolTable() override;

  bool used() const override;

  void markUnused() override;

  bool hidden() const override;

  // RefcountInterface

  void incRefCount() override;

  bool decRefCount() override;

  // Indicates that the ThreadLocalStore is shutting down, so no need to clear its histogram_set_.

private:

  bool usedLockHeld() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(merge_lock_);

  std::vector<Stats::ParentHistogram::Bucket>

  detailedlBucketsHelper(const histogram_t& histogram) const;

  const Histogram::Unit unit_;

  const absl::optional<uint32_t> bins_;

  ThreadLocalStoreImpl& thread_local_store_;

  histogram_t* interval_histogram_;

  histogram_t* cumulative_histogram_;

  HistogramStatisticsImpl interval_statistics_;

  HistogramStatisticsImpl cumulative_statistics_;

  mutable Thread::MutexBasicLockable merge_lock_;

  std::list<TlsHistogramSharedPtr> tls_histograms_ ABSL_GUARDED_BY(merge_lock_);

  bool merged_{false};

  std::atomic<bool> shutting_down_{false};

  std::atomic<uint32_t> ref_count_{0};

  const uint64_t id_; // Index into TlsCache::histogram_cache_.

};

using ParentHistogramImplSharedPtr = RefcountPtr<ParentHistogramImpl>;

/**

 * Store implementation with thread local caching. For design details see

 * https://github.com/envoyproxy/envoy/blob/main/source/docs/stats.md

 */

class ThreadLocalStoreImpl : Logger::Loggable<Logger::Id::stats>, public StoreRoot {

public:

  static const char DeleteScopeSync[];

  static const char IterateScopeSync[];

  static const char MainDispatcherCleanupSync[];

  ThreadLocalStoreImpl(Allocator& alloc);

  ~ThreadLocalStoreImpl() override;

  // Stats::Store

  std::vector<CounterSharedPtr> counters() const override;

  std::vector<GaugeSharedPtr> gauges() const override;

  std::vector<TextReadoutSharedPtr> textReadouts() const override;

  std::vector<ParentHistogramSharedPtr> histograms() const override;

  void forEachCounter(SizeFn f_size, StatFn<Counter> f_stat) const override;

  void forEachGauge(SizeFn f_size, StatFn<Gauge> f_stat) const override;

  void forEachTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const override;

  void forEachHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override;

  void forEachScope(SizeFn f_size, StatFn<const Scope> f_stat) const override;

  void evictUnused() override;

  // Stats::StoreRoot

  void setStatsMatcher(StatsMatcherPtr&& stats_matcher) override;

  void setHistogramSettings(HistogramSettingsConstPtr&& histogram_settings) override;

  void initializeThreading(Event::Dispatcher& main_thread_dispatcher,

                           ThreadLocal::Instance& tls) override;

  void shutdownThreading() override;

  void mergeHistograms(PostMergeCb merge_cb) override;

  void deliverHistogramToSinks(const Histogram& histogram, uint64_t value) override;

  Histogram& tlsHistogram(ParentHistogramImpl& parent, uint64_t id);

  void forEachSinkedCounter(SizeFn f_size, StatFn<Counter> f_stat) const override;

  void forEachSinkedGauge(SizeFn f_size, StatFn<Gauge> f_stat) const override;

  void forEachSinkedTextReadout(SizeFn f_size, StatFn<TextReadout> f_stat) const override;

  void forEachSinkedHistogram(SizeFn f_size, StatFn<ParentHistogram> f_stat) const override;

  void setSinkPredicates(std::unique_ptr<SinkPredicates>&& sink_predicates) override;

  /**

   * @return a thread synchronizer object used for controlling thread behavior in tests.

   */

  /**

   * @return a set of well known tag names; used to reduce symbol table churn.

   */

  bool decHistogramRefCount(ParentHistogramImpl& histogram, std::atomic<uint32_t>& ref_count);

  void releaseHistogramCrossThread(uint64_t histogram_id);

  void extractAndAppendTags(StatName name, StatNamePool& pool, StatNameTagVector& tags) override;

  void extractAndAppendTags(absl::string_view name, StatNamePool& pool,

                            StatNameTagVector& tags) override;

  void ensureOverflowStats(const ScopeStatsLimitSettings& limits);

private:

  friend class ThreadLocalStoreTestingPeer;

  template <class Stat> using StatRefMap = StatNameHashMap<std::reference_wrapper<Stat>>;

  struct TlsCacheEntry {

    // The counters, gauges and text readouts in the TLS cache are stored by reference,

    // depending on the CentralCache for backing store. This avoids a potential

    // contention-storm when destructing a scope, as the counter/gauge ref-count

    // decrement in allocator.cc needs to hold the single allocator mutex.

    StatRefMap<Counter> counters_;

    StatRefMap<Gauge> gauges_;

    StatRefMap<TextReadout> text_readouts_;

    // Histograms also require holding a mutex while decrementing reference

    // counts. The only difference from other stats is that the histogram_set_

    // lives in the ThreadLocalStore object, rather than in

    // Allocator. Histograms are removed from that set when all scopes

    // referencing the histogram are dropped. Each ParentHistogram has a unique

    // index, which is not re-used during the process lifetime.

    //

    // There is also a tls_histogram_cache_ in the TlsCache object, which is

    // not tied to a scope. It maps from parent histogram's unique index to

    // a TlsHistogram. This enables continuity between same-named histograms

    // in same-named scopes. That scenario is common when re-creating scopes in

    // response to xDS.

    StatNameHashMap<ParentHistogramSharedPtr> parent_histograms_;

    // We keep a TLS cache of rejected stat names. This costs memory, but

    // reduces runtime overhead running the matcher. Moreover, once symbol

    // tables are integrated, rejection will need the fully elaborated string,

    // and it we need to take a global symbol-table lock to run. We keep this

    // StatName set here in the TLS cache to avoid taking a lock to compute

    // rejection.

    StatNameHashSet rejected_stats_;

  };

  struct CentralCacheEntry : public RefcountHelper {

    ~CentralCacheEntry();

    StatNameHashMap<CounterSharedPtr> counters_;

    StatNameHashMap<GaugeSharedPtr> gauges_;

    StatNameHashMap<ParentHistogramImplSharedPtr> histograms_;

    StatNameHashMap<TextReadoutSharedPtr> text_readouts_;

    StatNameStorageSet rejected_stats_;

    SymbolTable& symbol_table_;

  };

  using CentralCacheEntrySharedPtr = RefcountPtr<CentralCacheEntry>;

  struct ScopeImpl : public Scope {

    ScopeImpl(ThreadLocalStoreImpl& parent, StatName prefix, bool evictable,

              const ScopeStatsLimitSettings& limits = {},

              StatsMatcherSharedPtr scope_matcher = nullptr);

    ~ScopeImpl() override;

    // Stats::Scope

    Counter& counterFromStatNameWithTags(const StatName& name,

                                         StatNameTagVectorOptConstRef tags) override;

    Gauge& gaugeFromStatNameWithTags(const StatName& name, StatNameTagVectorOptConstRef tags,

                                     Gauge::ImportMode import_mode) override;

    Histogram& histogramFromStatNameWithTags(const StatName& name,

                                             StatNameTagVectorOptConstRef tags,

                                             Histogram::Unit unit) override;

    TextReadout& textReadoutFromStatNameWithTags(const StatName& name,

                                                 StatNameTagVectorOptConstRef tags) override;

    ScopeSharedPtr createScope(const std::string& name, bool evictable = false,

                               const ScopeStatsLimitSettings& limits = {},

                               StatsMatcherSharedPtr matcher = nullptr) override;

    ScopeSharedPtr scopeFromStatName(StatName name, bool evictable = false,

                                     const ScopeStatsLimitSettings& limits = {},

                                     StatsMatcherSharedPtr matcher = nullptr) override;

    bool iterateLockHeld(const IterateFn<Counter>& fn) const

    bool iterateLockHeld(const IterateFn<Gauge>& fn) const

    bool iterateLockHeld(const IterateFn<Histogram>& fn) const

    bool iterateLockHeld(const IterateFn<TextReadout>& fn) const

    // NOTE: The find methods assume that `name` is fully-qualified.

    // Implementations will not add the scope prefix.

    CounterOptConstRef findCounter(StatName name) const override;

    GaugeOptConstRef findGauge(StatName name) const override;

    HistogramOptConstRef findHistogram(StatName name) const override;

    TextReadoutOptConstRef findTextReadout(StatName name) const override;

    HistogramOptConstRef findHistogramLockHeld(StatName name) const;

    template <class StatType>

    using MakeStatFn = std::function<RefcountPtr<StatType>(

        Allocator&, StatName name, StatName tag_extracted_name, const StatNameTagVector& tags)>;

    /**

     * Makes a stat either by looking it up in the central cache,

     * generating it from the parent allocator, or as a last

     * result, creating it with the heap allocator.

     *

     * @param full_stat_name the full name of the stat with appended tags.

     * @param name_no_tags the full name of the stat (not tag extracted) without appended tags.

     * @param stat_name_tags the tags provided at creation time. If empty, tag extraction occurs.

     * @param central_cache_map a map from name to the desired object in the central cache.

     * @param make_stat a function to generate the stat object, called if it's not in cache.

     * @param tls_ref possibly null reference to a cache entry for this stat, which will be

     *     used if non-empty, or filled in if empty (and non-null).

     */

    template <class StatType>

    StatType& safeMakeStat(StatName full_stat_name, StatName name_no_tags,

                           const absl::optional<StatNameTagVector>& stat_name_tags,

                           StatNameHashMap<RefcountPtr<StatType>>& central_cache_map,

                           StatsMatcher::FastResult fast_reject_result,

                           StatNameStorageSet& central_rejected_stats,

                           MakeStatFn<StatType> make_stat, StatRefMap<StatType>* tls_cache,

                           StatNameHashSet* tls_rejected_stats, StatType& null_stat);

    template <class StatType>

    using StatTypeOptConstRef = absl::optional<std::reference_wrapper<const StatType>>;

    /**

     * Looks up an existing stat, populating the local cache if necessary. Does

     * not check the TLS or rejects, and does not create a stat if it does not

     * exist.

     *

     * @param name the full name of the stat (not tag extracted).

     * @param central_cache_map a map from name to the desired object in the central cache.

     * @return a reference to the stat, if it exists.

     */

    template <class StatType>

    StatTypeOptConstRef<StatType>

    findStatLockHeld(StatName name,

    // Returns the central cache, asserting that the parent lock is held.

    //

    // When a ThreadLocalStore method takes lock_ and then accesses

    // scope->central_cache_, the analysis system cannot understand that the

    // scope's parent_.lock_ is held, so we assert that here.

    const CentralCacheEntrySharedPtr& centralCacheLockHeld() const

    CentralCacheEntrySharedPtr&

    // Returns the central cache, bypassing thread analysis.

    //

    // This is used only when passing references to maps held in the central

    // cache to safeMakeStat, which takes the lock only if those maps are

    // actually referenced, due to the lookup missing the TLS cache.

    const CentralCacheEntrySharedPtr&

    // Returns the effective matcher for this scope: scope-level if set, else store-level.

    const uint64_t scope_id_;

    ThreadLocalStoreImpl& parent_;

    const bool evictable_{};

    const ScopeStatsLimitSettings limits_;

    StatsMatcherSharedPtr scope_matcher_;

  private:

    StatNameStorage prefix_;

    mutable CentralCacheEntrySharedPtr central_cache_ ABSL_GUARDED_BY(parent_.lock_);

  };

  struct TlsCache : public ThreadLocal::ThreadLocalObject {

    TlsCacheEntry& insertScope(uint64_t scope_id);

    void eraseScopes(const std::vector<uint64_t>& scope_ids);

    void eraseHistograms(const std::vector<uint64_t>& histograms);

    // The TLS scope cache is keyed by scope ID. This is used to avoid complex circular references

    // during scope destruction. An ID is required vs. using the address of the scope pointer

    // because it's possible that the memory allocator will recycle the scope pointer immediately

    // upon destruction, leading to a situation in which a new scope with the same address is used

    // to reference the cache, and then subsequently cache flushed, leaving nothing in the central

    // store. See the overview for more information. This complexity is required for lockless

    // operation in the fast path.

    absl::flat_hash_map<uint64_t, TlsCacheEntry> scope_cache_;

    // Maps from histogram ID (monotonically increasing) to a TLS histogram.

    absl::flat_hash_map<uint64_t, TlsHistogramSharedPtr> tls_histogram_cache_;

  };

  using ScopeImplSharedPtr = std::shared_ptr<ScopeImpl>;

  /**

   * assertLocked exists to help compiler figure out that lock_ and scope->parent_.lock_ is

   * actually the same lock known under two different names. This function requires lock_ to

   * be held when it's called and at the same time it is annotated as if it checks in runtime

   * that scope->parent_.lock_ is held. It does not actually perform any runtime checks, because

   * those aren't needed since we know that scope->parent_ refers to ThreadLockStoreImpl and

   * therefore scope->parent_.lock is the same as lock_.

   */

  void assertLocked(const ScopeImpl& scope) const ABSL_EXCLUSIVE_LOCKS_REQUIRED(lock_)

  /**

   * Calls fn_lock_held for every scope with, lock_ held. This avoids iterate/destruct

   * races for scopes.

   *

   * @param fn_lock_held function to be called, with lock_ held, on every scope, until

   *   fn_lock_held() returns false.

   * @return true if the iteration completed with fn_lock_held never returning false.

   */

  bool iterateScopesLockHeld(const std::function<bool(const ScopeImplSharedPtr&)> fn) const

      ABSL_EXCLUSIVE_LOCKS_REQUIRED(lock_);

  // The Store versions of iterate cover all the scopes in the store.

  std::string getTagsForName(const std::string& name, TagVector& tags) const;

  void clearScopesFromCaches();

  void clearHistogramsFromCaches();

  void releaseScopeCrossThread(ScopeImpl* scope);

  void mergeInternal(PostMergeCb merge_cb);

  bool slowRejects(StatsMatcher::FastResult fast_reject_result, StatName name) const;

  StatsMatcher::FastResult fastRejects(StatName name) const;

  template <class StatMapClass, class StatListClass>

  void removeRejectedStats(StatMapClass& map, StatListClass& list);

  template <class StatSharedPtr>

  void removeRejectedStats(StatNameHashMap<StatSharedPtr>& map,

                           std::function<void(const StatSharedPtr&)> f_deletion);

  bool checkAndRememberRejection(StatName name, StatsMatcher::FastResult fast_reject_result,

                                 StatNameStorageSet& central_rejected_stats,

                                 StatNameHashSet* tls_rejected_stats, const StatsMatcher& matcher);

  void addScope(std::shared_ptr<ScopeImpl>& new_scope);

  OptRef<SinkPredicates> sink_predicates_;

  Allocator& alloc_;

  Event::Dispatcher* main_thread_dispatcher_{};

  using TlsCacheSlot = ThreadLocal::TypedSlotPtr<TlsCache>;

  ThreadLocal::TypedSlotPtr<TlsCache> tls_cache_;

  mutable Thread::MutexBasicLockable lock_;

  absl::flat_hash_map<ScopeImpl*, std::weak_ptr<ScopeImpl>> scopes_ ABSL_GUARDED_BY(lock_);

  ScopeSharedPtr default_scope_;

  std::vector<std::reference_wrapper<Sink>> timer_sinks_;

  TagProducerPtr tag_producer_;

  StatsMatcherPtr stats_matcher_;

  HistogramSettingsConstPtr histogram_settings_;

  std::atomic<bool> threading_ever_initialized_{false};

  std::atomic<bool> shutting_down_{false};

  std::atomic<bool> merge_in_progress_{false};

  OptRef<ThreadLocal::Instance> tls_;

  NullCounterImpl null_counter_;

  NullGaugeImpl null_gauge_;

  NullHistogramImpl null_histogram_;

  NullTextReadoutImpl null_text_readout_;

  mutable Thread::ThreadSynchronizer sync_;

  std::atomic<uint64_t> next_scope_id_{0};

  uint64_t next_histogram_id_ ABSL_GUARDED_BY(hist_mutex_) = 0;

  StatNameSetPtr well_known_tags_;

  mutable Thread::MutexBasicLockable hist_mutex_;

  StatSet<ParentHistogramImpl> histogram_set_ ABSL_GUARDED_BY(hist_mutex_);

  StatSet<ParentHistogramImpl> sinked_histograms_ ABSL_GUARDED_BY(hist_mutex_);

  // Retain storage for deleted stats; these are no longer in maps because the

  // matcher-pattern was established after they were created. Since the stats

  // are held by reference in code that expects them to be there, we can't

  // actually delete the stats.

  //

  // It seems like it would be better to have each client that expects a stat

  // to exist to hold it as (e.g.) a CounterSharedPtr rather than a Counter&

  // but that would be fairly complex to change.

  std::vector<HistogramSharedPtr> deleted_histograms_ ABSL_GUARDED_BY(lock_);

  // Scope IDs and central cache entries that are queued for cross-scope release.

  // Because there can be a large number of scopes, all of which are released at once,

  // (e.g. when a scope is deleted), it is more efficient to batch their cleanup,

  // which would otherwise entail a post() per scope per thread.

  std::vector<uint64_t> scopes_to_cleanup_ ABSL_GUARDED_BY(lock_);

  std::vector<CentralCacheEntrySharedPtr> central_cache_entries_to_cleanup_ ABSL_GUARDED_BY(lock_);

  // Histograms IDs that are queued for cross-scope release. Because there

  // can be a large number of histograms, all of which are released at once,

  // (e.g. when a scope is deleted), it is likely more efficient to batch their

  // cleanup, which would otherwise entail a post() per histogram per thread.

  std::vector<uint64_t> histograms_to_cleanup_ ABSL_GUARDED_BY(hist_mutex_);

  CounterSharedPtr counters_overflow_;

  CounterSharedPtr gauges_overflow_;

  CounterSharedPtr histograms_overflow_;

};

using ThreadLocalStoreImplPtr = std::unique_ptr<ThreadLocalStoreImpl>;

} // namespace Stats

} // namespace Envoy