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

package org.apache.cassandra.metrics;

import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.codahale.metrics.Clock;
import com.google.common.annotations.VisibleForTesting;

import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.utils.MonotonicClock;
import org.apache.cassandra.utils.ReflectionUtils;

import static java.lang.Math.exp;

/**
 * An alternative to Dropwizard Meter which implements the same kind of API.
 * it has more efficent mark operations and consumes less memory.
 * Only exponential decaying moving average is supported for 1/5/15-minutes rate values.
 * Tick logic is moved out from a mark operation and always executed in a background thread.
 * For better cache locality rate values are extracted to a common non thread-local array
 *  updated by a background thread in bulk.
 * Counter logic inside is implemented using @see ThreadLocalMetrics functionality.
 * NOTE: Dropwizard Meter is a class and there is no an interface for Dropwizard Meter logic,
 *   so we have to create an alternative hierarchy.
 */
public class ThreadLocalMeter extends com.codahale.metrics.Meter implements Meter
{
    private static final int INTERVAL_SEC = 5;
    private static final long TICK_INTERVAL_NS = TimeUnit.SECONDS.toNanos(INTERVAL_SEC);
    private static final double SECONDS_PER_MINUTE = 60.0;

    private static final double ONE_NS_IN_SEC = 1d / TimeUnit.SECONDS.toNanos(1);
    private static final int ONE_MINUTE = 1;
    private static final int FIVE_MINUTES = 5;
    private static final int FIFTEEN_MINUTES = 15;
    private static final double M1_ALPHA = 1 - exp(-INTERVAL_SEC / SECONDS_PER_MINUTE / ONE_MINUTE);
    private static final double M5_ALPHA = 1 - exp(-INTERVAL_SEC / SECONDS_PER_MINUTE / FIVE_MINUTES);
    private static final double M15_ALPHA = 1 - exp(-INTERVAL_SEC / SECONDS_PER_MINUTE / FIFTEEN_MINUTES);

    // each meter instance stores rate stats using 3 consecutive cells in rates array
    private static final int M1_RATE_OFFSET = 0;
    private static final int M5_RATE_OFFSET = 1;
    private static final int M15_RATE_OFFSET = 2;
    private static final int RATES_COUNT = 3;
    private static final double NON_INITIALIZED = Double.MIN_VALUE;

    private static final int BACKGROUND_TICK_INTERVAL_SEC = INTERVAL_SEC;
    private static final List<WeakReference<ThreadLocalMeter>> allMeters = new CopyOnWriteArrayList<>();

    /**
     * CASSANDRA-19332
     * If ticking would reduce even Long.MAX_VALUE in the 15 minute EWMA below this target then don't bother
     * ticking in a loop and instead reset all the EWMAs.
     */
    private static final double maxTickZeroTarget = 0.0001;
    private static final int maxTicks;

    static
    {
        int m3Ticks = 1;
        double emulatedM15Rate = 0.0;
        emulatedM15Rate = tickFifteenMinuteEWMA(emulatedM15Rate, Long.MAX_VALUE);
        do
        {
            emulatedM15Rate = tickFifteenMinuteEWMA(emulatedM15Rate, 0);
            m3Ticks++;
        }
        while (getRatePerSecond(emulatedM15Rate) > maxTickZeroTarget);
        maxTicks = m3Ticks;
    }

    private static final ScheduledFuture<?> backgroundTickingJob;

    /**
     * the method is provided for test purposes only,
     * to disable background activities and make tests more deterministic
     */
    @VisibleForTesting
    static void disableBackgroundTicking()
    {
        backgroundTickingJob.cancel(true);
    }

    static
    {
        backgroundTickingJob = ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(ThreadLocalMeter::tickAll,
                                                                 BACKGROUND_TICK_INTERVAL_SEC,
                                                                 BACKGROUND_TICK_INTERVAL_SEC,
                                                                 TimeUnit.SECONDS);
    }

    private static final Object ratesArrayGuard = new Object();

    // writes should be synchonized using ratesArrayGuard
    // 1) write during a copy when we need to extend rates array when a new Meter is created
    // 2) write during a tick
    private static volatile double[] rates = new double[RATES_COUNT * 16];
    static final AtomicInteger rateGroupIdGenerator = new AtomicInteger();

    // we recycle and reuse rate group IDs
    // a set bit means the correspondent rate group id is available to use
    private static final BitSet freeRateGroupIdSet = new BitSet();

    private static int allocateRateGroupOffset()
    {
        int rateGroupId;
        synchronized (freeRateGroupIdSet)
        {
            rateGroupId = freeRateGroupIdSet.nextSetBit(0);
            if (rateGroupId >= 0)
                freeRateGroupIdSet.clear(rateGroupId);
        }
        if (rateGroupId < 0)
        {
            rateGroupId = rateGroupIdGenerator.getAndAdd(RATES_COUNT);
        }
        synchronized (ratesArrayGuard)
        {
            if (rates.length < rateGroupId + RATES_COUNT)
            {
                double[] newRates = new double[rateGroupId + RATES_COUNT];
                System.arraycopy(rates, 0, newRates, 0, rates.length);
                rates = newRates;
            }
            rates[rateGroupId +  M1_RATE_OFFSET] = NON_INITIALIZED;
            rates[rateGroupId +  M5_RATE_OFFSET] = NON_INITIALIZED;
            rates[rateGroupId + M15_RATE_OFFSET] = NON_INITIALIZED;
        }
        return rateGroupId;
    }

    static double getRateValue(int offset)
    {
        return rates[offset];
    }

    private static void setRateValue(int offset, double value)
    {
        rates[offset] = value;
    }

    private final int countMetricId;
    private final int uncountedMetricId;
    private final int rateGroupId;
    private final long startTime;
    private final MonotonicClock clock;
    private long lastTick;

    public ThreadLocalMeter()
    {
        this(MonotonicClock.Global.approxTime);
    }

    public ThreadLocalMeter(MonotonicClock clock)
    {
        // movingAverages is set to null to reduce metrics memory footprint
        super(null, Clock.defaultClock());
        this.clock = clock;
        this.startTime = this.clock.now();
        this.lastTick = this.startTime;
        this.countMetricId = ThreadLocalMetrics.allocateMetricId();
        this.uncountedMetricId = ThreadLocalMetrics.allocateMetricId();
        this.rateGroupId = allocateRateGroupOffset();
        allMeters.add(new WeakReference<>(this));
        ThreadLocalMetrics.destroyWhenUnreachable(this, new MeterCleaner(countMetricId, uncountedMetricId, rateGroupId));
        ReflectionUtils.setFieldToNull(this, com.codahale.metrics.Meter.class, "count"); // to reduce metrics memory footprint
    }

    private static class MeterCleaner implements ThreadLocalMetrics.MetricCleaner
    {
        private final int countMetricId;
        private final int uncountedMetricId;
        private final int rateGroupId;

        private MeterCleaner(int countMetricId, int uncountedMetricId, int rateGroupId)
        {
            this.countMetricId = countMetricId;
            this.uncountedMetricId = uncountedMetricId;
            this.rateGroupId = rateGroupId;
        }

        @Override
        public void clean()
        {
            recycleRateGroupId(rateGroupId);
            ThreadLocalMetrics.recycleMetricId(countMetricId);
            ThreadLocalMetrics.recycleMetricId(uncountedMetricId);
        }

        private static void recycleRateGroupId(int rateGroupId)
        {
            synchronized (freeRateGroupIdSet)
            {
                freeRateGroupIdSet.set(rateGroupId);
            }
        }
    }

    /**
     * Mark the occurrence of an event.
     */
    public void mark()
    {
        mark(1);
    }

    /**
     * Mark the occurrence of a given number of events.
     *
     * @param n the number of events
     */
    public void mark(long n)
    {
        // we retrieve the context once here to reduce thread local lookup overheads
        ThreadLocalMetrics context = ThreadLocalMetrics.get();
        context.addNonStatic(countMetricId, n);
        context.addNonStatic(uncountedMetricId, n);
    }

    @Override
    public long getCount()
    {
        return ThreadLocalMetrics.getCount(countMetricId);
    }

    @Override
    public double getFifteenMinuteRate()
    {
        return getRatePerSecond(getRateValue(rateGroupId + M15_RATE_OFFSET));
    }

    @Override
    public double getFiveMinuteRate()
    {
        return getRatePerSecond(getRateValue(rateGroupId + M5_RATE_OFFSET));
    }

    @Override
    public double getOneMinuteRate()
    {
        return getRatePerSecond(getRateValue(rateGroupId + M1_RATE_OFFSET));
    }

    @Override
    public double getMeanRate()
    {
        long count = getCount();
        if (count == 0)
        {
            return 0.0;
        }
        else
        {
            long elapsed = clock.now() - startTime;
            return count / (elapsed * ONE_NS_IN_SEC);
        }
    }

    @VisibleForTesting
    static void tickAll()
    {
            List<WeakReference<ThreadLocalMeter>> emptyRefsToRemove = null;
            for (WeakReference<ThreadLocalMeter> threadLocalMeterRef : allMeters)
            {
                ThreadLocalMeter meter = threadLocalMeterRef.get();
                if (meter != null)
                {
                    meter.tickIfNessesary();
                }
                else
                {
                    if (emptyRefsToRemove == null)
                        emptyRefsToRemove = new ArrayList<>();
                    emptyRefsToRemove.add(threadLocalMeterRef);
                }
            }
            if (emptyRefsToRemove != null)
                allMeters.removeAll(emptyRefsToRemove);

    }

    private void tickIfNessesary()
    {
        long newTick = clock.now();
        long age = newTick - lastTick;
        if (age > TICK_INTERVAL_NS)
        {
            lastTick = newTick - age % TICK_INTERVAL_NS;
            long requiredTicks = age / TICK_INTERVAL_NS;
            tick(requiredTicks);
        }
    }
    private void tick(long requiredTicks)
    {
        synchronized (ratesArrayGuard)
        {
            if (requiredTicks >= maxTicks)
            {
                reset();
            }
            else if (requiredTicks > 0)
            {
                long count = ThreadLocalMetrics.getCountAndReset(uncountedMetricId);
                for (long i = 0; i < requiredTicks; i++)
                {
                    int m1Offset  = rateGroupId +  M1_RATE_OFFSET;
                    int m5Offset  = rateGroupId +  M5_RATE_OFFSET;
                    int m15Offset = rateGroupId + M15_RATE_OFFSET;
                    double m1Rate  = getRateValue(m1Offset);
                    double m5Rate  = getRateValue(m5Offset);
                    double m15Rate = getRateValue(m15Offset);
                    setRateValue(m1Offset,  tickOneMinuteEWMA(m1Rate, count));
                    setRateValue(m5Offset,  tickFiveMinuteEWMA(m5Rate, count));
                    setRateValue(m15Offset, tickFifteenMinuteEWMA(m15Rate, count));
                    count = 0;
                }
            }
        }
    }

    private static double tickOneMinuteEWMA(double oldRate, long count)
    {
        return tick(M1_ALPHA, oldRate, count);
    }

    private static double tickFiveMinuteEWMA(double oldRate, long count)
    {
        return tick(M5_ALPHA, oldRate, count);
    }

    private static double tickFifteenMinuteEWMA(double oldRate, long count)
    {
        return tick(M15_ALPHA, oldRate, count);
    }

    private static double tick(double alpha, double oldRate, long count)
    {
        double instantRate = (double) count / TICK_INTERVAL_NS;
        if (oldRate != NON_INITIALIZED)
            return oldRate + alpha * (instantRate - oldRate);
        else // init
            return instantRate;
    }
    private static double getRatePerSecond(double rate)
    {
        if (rate == NON_INITIALIZED)
            rate = 0.0;
        return rate * (double) TimeUnit.SECONDS.toNanos(1L);
    }

    /**
     * Set the rate to the smallest possible positive value. Used to avoid calling tick a large number of times.
     */
    private void reset()
    {
        ThreadLocalMetrics.getCountAndReset(uncountedMetricId);
        setRateValue(rateGroupId +  M1_RATE_OFFSET, Double.MIN_NORMAL);
        setRateValue(rateGroupId +  M5_RATE_OFFSET, Double.MIN_NORMAL);
        setRateValue(rateGroupId + M15_RATE_OFFSET, Double.MIN_NORMAL);
    }

    @VisibleForTesting
    static int getTickingMetersCount()
    {
        return allMeters.size();
    }
}
