MovingAverage.cpp

Go to the documentation of this file.

#include "MovingAverage.h"
#include "UnitTestFramework.h"
 
namespace cmlabs {
 
MovingAverage::MovingAverage(uint32 binTimeMS, uint16 binCount) {
    mutex.enter(200, __FUNCTION__);
    this->binTime = binTimeMS * 1000;
    this->binCount = binCount;
    entries = (StatEntry*)malloc(binCount*sizeof(StatEntry));
    memset(entries, 0, binCount*sizeof(StatEntry));
    currentBin = 0;
    currentBinStart = 0;
    totalValue = 0;
    totalCount = 0;
    mutex.leave();
}
 
MovingAverage::~MovingAverage() {
    mutex.enter(200, __FUNCTION__);
    free(entries);
    entries = NULL;
    mutex.leave();
}
 
bool MovingAverage::shiftBins(uint64 now) {
    // assume mutex is already locked
    // assume we already checked that data has been added
 
    int64 binShift = ( now - currentBinStart) / binTime;
    if (binShift < 0) {
        // This is a past time, don't change current bin
    }
    else if (binShift > binCount) {
        // We have missed all of our bins, reset everything except total
        // The computer was probably in sleep mode...
        currentBinStart = now;
        memset(entries, 0, binCount*sizeof(StatEntry));
        currentBin = 0;
    }
    else {
        // Shift the current bin forwards, wrap around as needed
        while (binShift > 0) {
            if (++currentBin >= binCount)
                currentBin = 0;
            entries[currentBin].count = 0;
            entries[currentBin].value = 0;
            binShift--;
            currentBinStart += binTime;
        }
    }
    return true;
}
 
 
bool MovingAverage::add(double val, uint32 count, uint64 time) {
    // We assume that time is close enough to 'now' so we don't have to do another GetTimeNow()
 
    mutex.enter(200, __FUNCTION__);
 
    if (!currentBinStart)
        currentBinStart = time;
    else {
        int64 binShift = ( time - currentBinStart) / binTime;
        if (binShift < 0) {
            // This is a past entry, put it into the old bin entry
            if (abs(binShift) < binCount) {
                int64 oldBin = (int64)currentBin + binShift; // negative binShift
                if (oldBin < 0)
                    oldBin += binCount;
                entries[(uint64)oldBin].value += val;
                entries[(uint64)oldBin].count += count;
                totalValue += val;
                totalCount += count;
                mutex.leave();
                return true;
            }
            else {
                // too old, just add to total
                totalValue += val;
                totalCount += count;
                mutex.leave();
                return true;
            }
        }
        else if (binShift > binCount) {
            // We have missed all of our bins, reset everything except total
            // The computer was probably in sleep mode...
            currentBinStart = time;
            memset(entries, 0, binCount*sizeof(StatEntry));
            currentBin = 0;
        }
        else {
            // Shift the current bin forwards, wrap around as needed
            while (binShift > 0) {
                if (++currentBin >= binCount)
                    currentBin = 0;
                entries[currentBin].count = 0;
                entries[currentBin].value = 0;
                binShift--;
                currentBinStart += binTime;
            }
        }
    }
 
    entries[currentBin].value += val;
    entries[currentBin].count += count;
    totalValue += val;
    totalCount += count;
 
    mutex.leave();
    return true;
}
 
bool MovingAverage::getAverage(uint32 ms, double& val, uint64& count, uint64 now) {
    if (!getSum(ms, val, count, now))
        return false;
 
    if (count)
        val = val / count;
    return true;
}
 
bool MovingAverage::getAverageMulti(
    uint32 ms1, uint32 ms2, uint32 ms3,
    double& avg1, uint64& count1,
    double& avg2, uint64& count2,
    double& avg3, uint64& count3, uint64 now) {
 
    double val1 = 0, val2 = 0, val3 = 0;
    if (!getSumMulti(ms1, ms2, ms3, val1, count1, val2, count2, val3, count3, now))
        return false;
 
    avg1 = avg2 = avg3 = 0;
    if (count1)
        avg1 = val1 / count1;
    if (count2)
        avg2 = val2 / count2;
    if (count3)
        avg3 = val3 / count3;
 
    return true;
}
 
bool MovingAverage::getSum(uint32 ms, double& val, uint64& count, uint64 now) {
    uint64 bins = ((uint64)ms*1000)/binTime;
    if (bins == 0)
        bins = 1;
    else if (bins > binCount-1)
        bins = binCount-1;
    
    val = 0;
    count = 0;
 
    mutex.enter(200, __FUNCTION__);
 
    if (!currentBinStart) {
        mutex.leave();
        return true;
    }
    else if (!shiftBins(now)) {
        mutex.leave();
        return false;
    }
 
    StatEntry* entry = entries + currentBin;
    // Never include the current bin as it is still being built
    if (entry == entries)
        entry = entries + binCount;
    entry--;
    for (uint64 n=0; n<bins; n++) {
        val += entry->value;
        count += entry->count;
        if (entry == entries)
            entry = entries + binCount;
        entry--;
    }
 
    mutex.leave();
    return true;
}
 
bool MovingAverage::getSumMulti(
    uint32 ms1, uint32 ms2, uint32 ms3,
    double& sum1, uint64& count1,
    double& sum2, uint64& count2,
    double& sum3, uint64& count3, uint64 now) {
 
    uint64 bins1 = ((uint64)ms1*1000)/binTime;
    if (bins1 == 0) bins1 = 1; else if (bins1 > binCount-1) bins1 = binCount-1;
    uint64 bins2 = ((uint64)ms2*1000)/binTime;
    if (bins2 == 0) bins2 = 1; else if (bins2 > binCount-1) bins2 = binCount-1;
    uint64 bins3 = ((uint64)ms3*1000)/binTime;
    if (bins3 == 0) bins3 = 1; else if (bins3 > binCount-1) bins3 = binCount-1;
    
    uint64 bins = clmax(clmax(bins1, bins2), bins3);
    sum1 = sum2 = sum3 = 0;
    count1 = count2 = count3 = 0;
 
    mutex.enter(200, __FUNCTION__);
 
    if (!currentBinStart) {
        mutex.leave();
        return true;
    }
    else if (!shiftBins(now)) {
        mutex.leave();
        return false;
    }
 
    StatEntry* entry = entries + currentBin;
    // Never include the current bin as it is still being built
    if (entry == entries)
        entry = entries + binCount;
    entry--;
    for (uint64 n=0; n<bins; n++) {
        if (n<bins1) {
            sum1 += entry->value;
            count1 += entry->count;
        }
        if (n<bins2) {
            sum2 += entry->value;
            count2 += entry->count;
        }
        if (n<bins3) {
            sum3 += entry->value;
            count3 += entry->count;
        }
        if (entry == entries)
            entry = entries + binCount;
        entry--;
    }
 
    mutex.leave();
    return true;
 
}
 
 
bool MovingAverage::getTotal(double& val, uint64& count) {
    mutex.enter(200, __FUNCTION__);
    val = totalValue;
    count = totalCount;
    mutex.leave();
    return true;
}
 
bool MovingAverage::getThroughput(uint32 ms, double& valPerSec, double& countPerSec, uint64 now) {
    double val;
    uint64 count;
    if (!getSum(ms, val, count, now))
        return false;
 
    double sec = (double)ms/1000.0;
    if (sec > 0) {
        valPerSec = val / sec;
        countPerSec = count / sec;
    }
    return true;
}
 
bool MovingAverage::getThroughputMulti(uint32 ms1, uint32 ms2, uint32 ms3, double& valPerSec1, double& countPerSec1, double& valPerSec2, double& countPerSec2, double& valPerSec3, double& countPerSec3, uint64 now ) {
 
    double val1 = 0, val2 = 0, val3 = 0;
    uint64 count1 = 0, count2 = 0, count3 = 0;
    if (!getSumMulti(ms1, ms2, ms3, val1, count1, val2, count2, val3, count3, now))
        return false;
 
    double sec = (double)ms1/1000.0;
    if (sec > 0) {
        valPerSec1 = val1 / sec;
        countPerSec1 = count1 / sec;
    }
    
    sec = (double)ms2/1000.0;
    if (sec > 0) {
        valPerSec2 = val2 / sec;
        countPerSec2 = count2 / sec;
    }
 
    sec = (double)ms3/1000.0;
    if (sec > 0) {
        valPerSec3 = val3 / sec;
        countPerSec3 = count3 / sec;
    }
 
    return true;
}
 
std::string MovingAverage::getPerfXML(uint32 binMS, uint32 binNum) {
    uint64 now = GetTimeNow();
    uint64 binSize = ((uint64)binMS * 1000) / binTime;
    uint64 bins = binNum * binSize;
    if (bins == 0)
        bins = 1;
    else if (bins > binCount - 1)
        bins = binCount - 1;
 
    uint32 actualBins = 0;
 
    mutex.enter(200, __FUNCTION__);
    std::string xml = "";
 
    if (currentBinStart && shiftBins(now)) {
        double val = 0;
        uint64 count = 0;
        uint32 binCounter = 0;
 
        StatEntry* entry = entries + currentBin;
        // Never include the current bin as it is still being built
        if (entry == entries)
            entry = entries + binCount;
        entry--;
        for (uint64 n = 0; n < bins; n++) {
            val += entry->value;
            count += entry->count;
            if (entry == entries)
                entry = entries + binCount;
            entry--;
            binCounter++;
            if (binCounter >= binSize) {
                if (count)
                    xml += utils::StringFormat("<count=\"%llu\" avg=\"%.6f\" total=\"%.6f\" />\n", count, val / count, val);
                else
                    xml += utils::StringFormat("<count=\"0\" avg=\"0\" total=\"0\" />\n");
                val = 0;
                count = 0;
                actualBins++;
                binCounter = 0;
            }
        }
        if (count) {
            xml += utils::StringFormat("<count=\"%llu\" avg=\"%.6f\" total=\"%.6f\" />\n", count, val / count, val);
            actualBins++;
        }
    }
 
    for (uint32 c = actualBins; c < binNum; c++)
        xml += utils::StringFormat("<count=\"0\" avg=\"0\" total=\"0\" />\n");
 
    mutex.leave();
    return utils::StringFormat("<performance>%s</performance>", xml.c_str());
}
 
std::string MovingAverage::getPerfJSON(uint32 binMS, uint32 binNum) {
    uint64 now = GetTimeNow();
    uint64 binSize = ((uint64)binMS * 1000) / binTime;
    uint64 bins = binNum * binSize;
    if (bins == 0)
        bins = 1;
    else if (bins > binCount - 1)
        bins = binCount - 1;
 
    uint32 actualBins = 0;
 
    mutex.enter(200, __FUNCTION__);
    std::string json = "";
 
    if (currentBinStart && shiftBins(now)) {
        double val = 0;
        uint64 count = 0;
        uint32 binCounter = 0;
 
        StatEntry* entry = entries + currentBin;
        // Never include the current bin as it is still being built
        if (entry == entries)
            entry = entries + binCount;
        entry--;
        for (uint64 n = 0; n < bins; n++) {
            val += entry->value;
            count += entry->count;
            if (entry == entries)
                entry = entries + binCount;
            entry--;
            binCounter++;
            if (binCounter >= binSize) {
                if (count)
                    json += utils::StringFormat("%s{\"count\":%llu ,\"avg\":%.6f, \"total\":%.6f}", json.length() ? "," : "", count, val / count, val);
                else
                    json += utils::StringFormat("%s{\"count\":0 ,\"avg\":0, \"total\":0}", json.length() ? "," : "");
                val = 0;
                count = 0;
                actualBins++;
                binCounter = 0;
            }
        }
        if (count) {
            json += utils::StringFormat("%s{\"count\":%llu ,\"avg\":%.6f, \"total\":%.6f}", json.length() ? "," : "", count, val / count, val);
            actualBins++;
        }
    }
 
    for (uint32 c = actualBins; c < binNum; c++)
        json += utils::StringFormat("%s{\"count\":0 ,\"avg\":0, \"total\":0}", json.length() ? "," : "");
 
    mutex.leave();
    return utils::StringFormat("[%s]", json.c_str());
}
 
 
//bool MovingAverage::getThroughputMultiAlt(uint32 ms1, uint32 ms2, uint32 ms3, double& valPerSec1, double& countPerSec1, double& valPerSec2, double& countPerSec2, double& valPerSec3, double& countPerSec3 ) {
//  if (!getThroughput(ms1, valPerSec1, countPerSec1))
//      return false;
//  if (!getThroughput(ms2, valPerSec2, countPerSec2))
//      return false;
//  if (!getThroughput(ms3, valPerSec3, countPerSec3))
//      return false;
//  return true;
//}
 
 
 
 
 
 
bool MovingAverage::UnitTest() {
    // binTime 100ms, 600 bins => 60s window
    MovingAverage stats(100, 60*10);
 
    uint32 innerCount = 100000;
    uint32 outerCount = 30;
 
    uint64 start, now = GetTimeNow();
    uint32 n, m;
    double val, v1, v2, v3;
    uint64 count;
    double c1, c2, c3;
 
    // 1. Add values and verify total bookkeeping
    unittest::progress(10, "timed add throughput");
    uint64 addStart = GetTimeNow();
    uint64 totalAdds = 0;
    for (n=0; n<outerCount; n++) {
        start = GetTimeNow();
        for (m=0; m<innerCount; m++) {
            stats.add(1024, 1, now);
            now += 10;
            totalAdds++;
        }
        unittest::detail("Time per timed add: %.3fus", ((double)GetTimeAge(start))/innerCount);
    }
    double addUs = (double)GetTimeAge(addStart);
    if (addUs > 0)
        unittest::metric("add_throughput", (double)totalAdds / addUs * 1e6, "ops/s", true);
    if (addUs > 0)
        unittest::metric("avg_add_latency", addUs / totalAdds, "us", false);
 
    // 2. Total must reflect every add (totals are never expired)
    unittest::progress(45, "verify totals");
    double totVal = 0;
    uint64 totCount = 0;
    if (!stats.getTotal(totVal, totCount)) {
        unittest::fail("MovingAverage test: getTotal failed\n");
        return false;
    }
    if (totCount != totalAdds) {
        unittest::fail("MovingAverage test: getTotal count %llu != expected %llu\n", totCount, totalAdds);
        return false;
    }
    if (totVal < (double)totalAdds * 1024.0 * 0.999 || totVal > (double)totalAdds * 1024.0 * 1.001) {
        unittest::fail("MovingAverage test: getTotal value %f != expected %f\n", totVal, (double)totalAdds * 1024.0);
        return false;
    }
 
    // NB: keep using the advanced simulated 'now' (the add loop stepped it ~30s
    // forward). Resetting it to real GetTimeNow() here would put the query
    // window before every entry, yielding a zero-count average.
 
    // 3. Average over a window: each entry has value 1024, so average must be ~1024
    unittest::progress(60, "average query");
    start = GetTimeNow();
    for (n=0; n<outerCount; n++) {
        if (!stats.getAverage(1000, val, count, now)) {
            unittest::fail("MovingAverage test: getAverage failed\n");
            return false;
        }
    }
    if (count == 0) {
        unittest::fail("MovingAverage test: getAverage returned zero count\n");
        return false;
    }
    if (val < 1023.0 || val > 1025.0) {
        unittest::fail("MovingAverage test: getAverage value %f not ~1024\n", val);
        return false;
    }
    unittest::metric("getaverage_latency", ((double)GetTimeAge(start))/outerCount, "us", false);
    unittest::detail("Average %f count: %llu", val, count);
 
    // 4. Sum over a window
    unittest::progress(75, "sum query");
    start = GetTimeNow();
    for (n=0; n<outerCount; n++) {
        if (!stats.getSum(1000, val, count, now)) {
            unittest::fail("MovingAverage test: getSum failed\n");
            return false;
        }
    }
    if (count == 0 || val <= 0) {
        unittest::fail("MovingAverage test: getSum returned empty window\n");
        return false;
    }
    // value/count must still average to ~1024
    if (val / (double)count < 1023.0 || val / (double)count > 1025.0) {
        unittest::fail("MovingAverage test: getSum avg %f not ~1024\n", val / (double)count);
        return false;
    }
    unittest::metric("getsum_latency", ((double)GetTimeAge(start))/outerCount, "us", false);
    unittest::detail("Sum %f count: %llu", val, count);
 
    // 5. Single-window throughput
    unittest::progress(85, "throughput query");
    start = GetTimeNow();
    for (n=0; n<outerCount; n++) {
        if (!stats.getThroughput(1000, v1, c1, now)) {
            unittest::fail("MovingAverage test: getThroughput failed\n");
            return false;
        }
    }
    if (v1 <= 0 || c1 <= 0) {
        unittest::fail("MovingAverage test: getThroughput non-positive rate\n");
        return false;
    }
    unittest::metric("getthroughput_latency", ((double)GetTimeAge(start))/outerCount, "us", false);
    unittest::detail("1sec %s %.3f msg/s", utils::BytifyRate(v1).c_str(), c1);
 
    // 6. Multi-window throughput
    unittest::progress(95, "multi throughput query");
    start = GetTimeNow();
    for (n=0; n<outerCount; n++) {
        if (!stats.getThroughputMulti(1000, 10000, 30000, v1, c1, v2, c2, v3, c3, now)) {
            unittest::fail("MovingAverage test: getThroughputMulti failed\n");
            return false;
        }
    }
    if (v1 <= 0 || v2 <= 0 || v3 <= 0) {
        unittest::fail("MovingAverage test: getThroughputMulti non-positive rate\n");
        return false;
    }
    unittest::metric("getthroughputmulti_latency", ((double)GetTimeAge(start))/outerCount, "us", false);
    unittest::detail("1sec %s %.3f msg/s 10sec %s %.3f msg/s 30sec %s %.3f msg/s",
        utils::BytifyRate(v1).c_str(), c1,
        utils::BytifyRate(v2).c_str(), c2,
        utils::BytifyRate(v3).c_str(), c3);
 
    unittest::progress(100, "done");
    return true;
}
 
void Register_MovingAverage_Tests() {
    UnitTestRunner::instance().registerTest("movingaverage", MovingAverage::UnitTest,
        "Moving average bin stats: add, sum, average, throughput", "core");
}
 
 
 
 
 
} // namespace cmlabs