// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <stdint.h>
#include <string.h>
#include <threads.h>
#include <unistd.h>

#include <cobalt-client/cpp/histogram-internal.h>
#include <fbl/auto_call.h>
#include <fbl/string.h>
#include <fbl/string_printf.h>
#include <fuchsia/cobalt/c/fidl.h>
#include <lib/sync/completion.h>
#include <lib/zx/time.h>
#include <unittest/unittest.h>

namespace cobalt_client {
namespace internal {
namespace {

// Number of threads to use for multithreading sanity check.
constexpr uint32_t kThreads = 20;

// Scalar used in UpdateHistogramFn to calculate the number of operations
// per bucket.
constexpr uint64_t kUpdateScalar = 100;

constexpr uint64_t kMetricId = 5;

constexpr uint32_t kEncodingId = 1;

constexpr uint32_t kBuckets = 20;

constexpr char kName[] = "SomeName";

bool Contains(const fbl::Vector<uint64_t>& container, uint64_t val) {
    for (auto el : container) {
        if (el == val) {
            return true;
        }
    }
    return false;
}

ObservationValue MakeObservationValue(const char* name, Value value) {
    ObservationValue obs;
    obs.name.size = strlen(name) + 1;
    obs.name.data = const_cast<char*>(name);
    obs.value = value;
    obs.encoding_id = kEncodingId;
    return obs;
}

const fbl::Vector<ObservationValue>& GetMetadata() {
    static const char part_name[] = "part";

    static const fbl::Vector<ObservationValue> metadata = {
        MakeObservationValue(part_name, IntValue(24)),
        MakeObservationValue(part_name, DoubleValue(0.125)),
    };

    return metadata;
}

BaseHistogram MakeHistogram() {
    BaseHistogram histogram(kName, GetMetadata(), kBuckets, kMetricId, kEncodingId);
    return histogram;
}

struct CheckContentsFlushFn {
    void operator()(uint64_t metric_id, const fidl::VectorView<ObservationValue>& observations,
                    BaseHistogram::FlushCompleteFn complete_fn) const {
        const fbl::Vector<ObservationValue>& metadata = GetMetadata();
        auto complete_flush = fbl::MakeAutoCall(fbl::move(complete_fn));

        size_t expected_size = metadata.size() + 1;
        if (observations.count() < expected_size) {
            *error = fbl::StringPrintf("observations.count()(%lu) != expected_size(%lu)\n",
                                       observations.count(), expected_size);
            return;
        }

        fbl::Vector<size_t> visited_obs;
        for (size_t curr_meta = 0; curr_meta < metadata.size(); ++curr_meta) {
            const auto& meta_obs = metadata[curr_meta];
            bool found_match = false;
            for (size_t current_obs = 0; current_obs < metadata.size(); ++current_obs) {
                if (Contains(visited_obs, current_obs)) {
                    continue;
                }
                const ObservationValue& value = observations[current_obs];
                if (value.encoding_id != encoding_id) {
                    *error = fbl::StringPrintf(
                        "observations[%lu].encoding_id(%u) != expected_encoding_id(%u)\n",
                        current_obs, meta_obs.encoding_id, encoding_id);
                    return;
                }
                if (memcmp(meta_obs.name.data, value.name.data, value.name.size) == 0 &&
                    memcmp(&meta_obs.value, &value.value, sizeof(Value)) == 0) {
                    found_match = true;
                    visited_obs.push_back(current_obs);
                    break;
                }
            }
            if (!found_match) {
                *error = fbl::StringPrintf("metadata[%lu] is not in observations.\n", curr_meta);
                return;
            }
        }

        size_t histogram_index = metadata.size();
        const ObservationValue& hist_obs = observations[histogram_index];
        if (hist_obs.encoding_id != encoding_id) {
            *error =
                fbl::StringPrintf("observations[%lu].encoding_id(%u) != expected_encoding_id(%u)\n",
                                  histogram_index, hist_obs.encoding_id, encoding_id);
            return;
        }

        if (memcmp(histogram_name.c_str(), hist_obs.name.data, hist_obs.name.size) != 0) {
            *error = fbl::StringPrintf("observations[%lu].name(%*s) != histogram_name(%s)\n",
                                       histogram_index, static_cast<int>(hist_obs.name.size),
                                       hist_obs.name.data, histogram_name.c_str());
            return;
        }

        // Check bucket values.
        if (hist_obs.value.tag != fuchsia_cobalt_ValueTagint_bucket_distribution) {
            *error = fbl::StringPrintf(
                "observations[%lu].value not IntBucketDistribution. tag(%u) != %u\n",
                histogram_index, hist_obs.value.tag,
                fuchsia_cobalt_ValueTagint_bucket_distribution);
            return;
        }

        if (hist_obs.value.int_bucket_distribution.count != bucket_values.size()) {
            *error =
                fbl::StringPrintf("observations[%lu].value.int_bucket_distribution.count(%lu) "
                                  "!= bucket_values.size()(%lu)",
                                  histogram_index, hist_obs.value.int_bucket_distribution.count,
                                  bucket_values.size());
            return;
        }

        BucketDistributionEntry* buckets =
            static_cast<BucketDistributionEntry*>(hist_obs.value.int_bucket_distribution.data);
        for (size_t bucket_index = 0; bucket_index < bucket_values.size(); ++bucket_index) {
            bool index_found = false;
            for (size_t bucket = 0; bucket < bucket_values.size(); ++bucket) {
                if (buckets[bucket].index == bucket_index) {
                    index_found = true;
                    if (buckets[bucket].count != bucket_values[bucket_index]) {
                        *error = fbl::StringPrintf(
                            "bucket_value[%lu](%lu) != buckets[%lu].count(%lu), but index match!\n",
                            bucket_index, bucket_values[bucket], bucket, buckets[bucket].count);
                        return;
                    }
                    break;
                }
            }
            if (!index_found) {
                *error = fbl::StringPrintf(
                    "bucket at index %lu is missing from the observed buckets.\n", bucket_index);
                return;
            }
        }
    }

    fbl::Vector<uint64_t> bucket_values;
    fbl::String histogram_name;

    fbl::String* error;
    uint32_t encoding_id;
};

// Add an observation and verify the bucket is updated.
bool AddObservationTest() {
    BEGIN_TEST;
    BaseHistogram histogram = MakeHistogram();
    ASSERT_EQ(histogram.GetCount(/*bucket=*/10), 0);
    histogram.IncrementCount(/*bucket=*/10);
    ASSERT_EQ(histogram.GetCount(10), 1);
    END_TEST;
}

// Verifies the order and the data in the elements when flushed, everything in the histogram
// should be flushed. This is true only for single threaded environment, since operation reordering,
// might send stale version of the bucket, but everything will eventually be flushed, so eventually
// all data will handled in multi-threaded environment.
bool FlushTest() {
    BEGIN_TEST;
    fbl::String error;
    CheckContentsFlushFn handler;
    handler.histogram_name = kName;
    handler.encoding_id = kEncodingId;
    handler.error = &error;
    BaseHistogram histogram = MakeHistogram();
    // bucket[i] = i
    for (uint64_t bucket_index = 0; bucket_index < kBuckets; ++bucket_index) {
        for (uint64_t op = 0; op < bucket_index; ++op) {
            histogram.IncrementCount(/*bucket=*/bucket_index);
        }
        handler.bucket_values.push_back(bucket_index);
    }
    ASSERT_TRUE(histogram.Flush(fbl::move(handler)));
    ASSERT_TRUE(error.empty(), error.c_str());
    END_TEST;
}

bool FlushWhileFlushingTest() {
    BEGIN_TEST;
    BaseHistogram::FlushCompleteFn complete_cb;
    BaseHistogram histogram = MakeHistogram();

    ASSERT_TRUE(histogram.Flush([&complete_cb](uint64_t metric_id,
                                               const fidl::VectorView<ObservationValue>& obs,
                                               BaseHistogram::FlushCompleteFn complete_fn) {
        complete_cb = fbl::move(complete_fn);
    }));
    ASSERT_FALSE(histogram.Flush(BaseHistogram::FlushFn()));
    complete_cb();
    ASSERT_TRUE(histogram.Flush([](uint64_t, const fidl::VectorView<ObservationValue>&,
                                   BaseHistogram::FlushCompleteFn) {}));
    ASSERT_FALSE(histogram.Flush(BaseHistogram::FlushFn()));
    END_TEST;
}

struct UpdateHistogramFnArgs {
    BaseHistogram* histogram;
    sync_completion_t* completion;
};

int UpdateHistogramFn(void* void_args) {
    UpdateHistogramFnArgs* args = static_cast<UpdateHistogramFnArgs*>(void_args);

    // Wait for call.
    zx_status_t res = sync_completion_wait(args->completion, zx::sec(20).get());
    if (res != ZX_OK) {
        return thrd_error;
    }

    for (uint64_t bucket = 0; bucket < kBuckets; ++bucket) {
        for (uint64_t i = 0; i < kUpdateScalar * bucket; ++i) {
            args->histogram->IncrementCount(bucket);
        }
    }

    return thrd_success;
}

// Verify that incrementing each bucket |kUpdateScalar|*|bucket_index| times per thread in
// |kThreads| threads, have a consistent view of the data, which allows the histogram to be
// thread-safe through this operations.
bool MultiThreadCountOpsConsistencyTest() {
    BEGIN_TEST;
    BaseHistogram histogram = MakeHistogram();
    fbl::Vector<thrd_t> thread_ids;
    sync_completion_t wait_for_start;

    UpdateHistogramFnArgs args;
    args.histogram = &histogram;
    args.completion = &wait_for_start;

    // Initialize all threads then signal start, and join all fo them.
    thread_ids.reserve(kThreads);
    for (uint32_t i = 0; i < kThreads; ++i) {
        thread_ids.push_back(0);
    }

    for (auto& thread_id : thread_ids) {
        ASSERT_EQ(thrd_create(&thread_id, UpdateHistogramFn, &args), thrd_success);
    }
    sync_completion_signal(&wait_for_start);

    for (auto& thread_id : thread_ids) {
        int res = thrd_success;
        ASSERT_EQ(thrd_join(thread_id, &res), thrd_success);
        ASSERT_EQ(res, thrd_success);
    }

    // Verify the result is coherent.
    // bucket[i] = i * kUpdateScalar * kThreads
    for (uint64_t bucket_index = 0; bucket_index < kBuckets; ++bucket_index) {
        ASSERT_EQ(histogram.GetCount(bucket_index), bucket_index * kUpdateScalar * kThreads);
    }
    END_TEST;
}

struct WaitBeforeCompleteFlushHandler {
    void operator()(uint64_t metric_id, const fidl::VectorView<ObservationValue>& observations,
                    BaseHistogram::FlushCompleteFn complete_fn) {
        ZX_ASSERT(thread_ids != nullptr);
        ZX_ASSERT(flushing_thread != nullptr);

        for (auto thread_id : *thread_ids) {
            if (thread_id != thrd_current()) {
                int res;
                thrd_join(thread_id, &res);
            }
        }

        *flushing_thread = thrd_current();
        counter->Increment();
        complete_fn();
        sync_completion_signal(completion);
    }

    // List of all threads that were spawned to flush.
    fbl::Vector<thrd_t>* thread_ids;

    // Notify main thread that all threads finished while this thread was flushing.
    sync_completion_t* completion;

    // Counter for the number of times Flush was actually called.
    BaseCounter* counter;

    // Main thread will join this thread.
    thrd_t* flushing_thread;
};

struct FlushHistogramFnArgs {
    BaseHistogram* histogram;
    // FlushHandler
    WaitBeforeCompleteFlushHandler flush_handler;

    // Notify to start the threads together.
    sync_completion_t* start;
};

// This verifies that while a thread is flushing, all other threads fail to flush the contents
// of the histogram thus any action that needs to be taken over the observation buffer is safe,
// as long as.
int FlushHistogramFn(void* void_args) {
    FlushHistogramFnArgs* args = static_cast<FlushHistogramFnArgs*>(void_args);
    sync_completion_wait(args->start, zx::sec(20).get());
    args->histogram->Flush(fbl::move(args->flush_handler));
    return thrd_success;
}

bool MultiThreadFlushOpsConsistencyTest() {
    BEGIN_TEST;
    BaseHistogram histogram = MakeHistogram();
    FlushHistogramFnArgs args[kThreads];
    fbl::Vector<thrd_t> thread_ids;
    sync_completion_t wait_for_start, wait_for_completion;
    thrd_t flushing_thread;
    BaseCounter flushes;

    // Initialize all threads then signal start, and join all fo them.
    thread_ids.reserve(kThreads);
    for (uint32_t i = 0; i < kThreads; ++i) {
        thread_ids.push_back(0);
    }

    for (uint64_t thread = 0; thread < kThreads; ++thread) {
        auto& thread_id = thread_ids[thread];
        auto handler = WaitBeforeCompleteFlushHandler();
        handler.thread_ids = &thread_ids;
        handler.completion = &wait_for_completion;
        handler.counter = &flushes;
        handler.flushing_thread = &flushing_thread;
        args[thread].flush_handler = fbl::move(handler);
        args[thread].histogram = &histogram;
        args[thread].start = &wait_for_start;
        ASSERT_EQ(thrd_create(&thread_id, FlushHistogramFn, &args), thrd_success);
    }
    sync_completion_signal(&wait_for_start);
    sync_completion_wait(&wait_for_completion, zx::sec(20).get());

    // Wait for the flushing thread to finish.
    thrd_join(flushing_thread, nullptr);

    // Verify the number of times it was flushed, must be one.
    ASSERT_EQ(flushes.Load(), 1);

    // Verify that if we flush again it succeeds.
    ASSERT_TRUE(histogram.Flush([](uint64_t, const fidl::VectorView<ObservationValue>&,
                                   BaseHistogram::FlushCompleteFn) {}));

    ASSERT_FALSE(histogram.Flush([](uint64_t, const fidl::VectorView<ObservationValue>&,
                                    BaseHistogram::FlushCompleteFn) {}));
    END_TEST;
}

BEGIN_TEST_CASE(BaseHistogramTest)
RUN_TEST(AddObservationTest)
RUN_TEST(FlushTest)
RUN_TEST(FlushWhileFlushingTest)
RUN_TEST(MultiThreadCountOpsConsistencyTest)
RUN_TEST(MultiThreadFlushOpsConsistencyTest)
END_TEST_CASE(BaseHistogramTest)
} // namespace
} // namespace internal
} // namespace cobalt_client