Added a threadsafe speedometer class.

2 files changed, 163 insertions, 0 deletions

diff --git a/src/unstable/speedometer.cpp b/src/unstable/speedometer.cpp
new file mode 100644
index 0000000..75e424e
--- /dev/null
+++ b/src/unstable/speedometer.cpp
@@ -0,0 +1,120 @@
+#include "bu/speedometer.h"
+#include "bu/mutexlocker.h"
+#include "bu/sio.h"
+
+#include <sys/time.h>
+#include <string.h>
+
+Bu::Speedometer::Speedometer( int64_t iBucketMs, int32_t iBucketCount ) :
+    aiBucket( NULL ),
+    iTotal( 0 ),
+    iCapacity( iBucketCount ),
+    iFill( 0 ),
+    iStart( 0 ),
+    iCurrent( 0 ),
+    iBucketMs( iBucketMs ),
+    iStartTime( -1 ),
+    iWindowSizeMs( iBucketMs*iBucketCount ),
+    iLastUpdate( -1 )
+{
+    aiBucket = new uint32_t[iCapacity];
+    memset( aiBucket, 0, sizeof(uint32_t)*iCapacity );
+}
+
+Bu::Speedometer::~Speedometer()
+{
+    delete[] aiBucket;
+    aiBucket = NULL;
+}
+
+void Bu::Speedometer::addEvent()
+{
+    Bu::MutexLocker l( mLock );
+    int64_t iNow = getCurrentTime();
+    _updateBuckets( iNow );
+    if( iStartTime < 0 )
+    {
+        iStartTime = iNow;
+        iFill = 1;
+        aiBucket[iCurrent] = 1;
+        iTotal = 1;
+    }
+    else
+    {
+        aiBucket[iCurrent]++;
+        iTotal++;
+    }
+    iLastUpdate = iNow;
+}
+
+void Bu::Speedometer::debug()
+{
+    Bu::MutexLocker l( mLock );
+    Bu::print("%1 of %2: %3:%4: ").
+        arg( iFill ).arg( iCapacity ).arg( iStart ).arg( iCurrent );
+    for( int j = 0; j < iFill; j++ )
+    {
+        Bu::print("%1 | ").arg( aiBucket[(j+iStart)%iCapacity] );
+    }
+    Bu::println(" total = %1").arg( iTotal );
+}
+
+void Bu::Speedometer::_updateBuckets( int64_t iNow )
+{
+    if( iNow > iLastUpdate+iWindowSizeMs )
+    {
+        iLastUpdate = iStartTime = -1;
+        iFill = 0;
+        iStart = iCurrent = 0;
+        aiBucket[iCurrent] = 0;
+        iTotal = 0;
+    }
+    else 
+    {
+        int64_t iSkip = (iNow-(iStartTime+iBucketMs*iFill));
+        if( iSkip < 0 )
+            return;
+        iSkip = (iSkip/iBucketMs) + ((iSkip%iBucketMs)?(1):(0));
+
+        for( int64_t j = 0; j < iSkip; j++ )
+        {
+            iCurrent = (iCurrent+1)%iCapacity;
+            if( iFill < iCapacity )
+            {
+                iFill++;
+            }
+            else
+            {
+                iTotal -= aiBucket[iStart];
+                aiBucket[iStart] = 0;
+                iStart = (iStart+1)%iCapacity;
+                iStartTime += iBucketMs;
+            }
+        }
+    }
+}
+
+double Bu::Speedometer::getSpeed()
+{
+    Bu::MutexLocker l( mLock );
+    _updateBuckets( getCurrentTime() );
+
+    if( iStartTime < 0 )
+        return 0.0;
+
+    return (double)iTotal / (double)(iFill*iBucketMs);
+}
+
+int32_t Bu::Speedometer::getTotalEvents()
+{
+    Bu::MutexLocker l( mLock );
+    return iTotal;
+}
+
+int64_t Bu::Speedometer::getCurrentTime() const
+{
+    struct timeval tv;
+    gettimeofday( &tv, NULL );
+    return tv.tv_sec*1000 + tv.tv_usec/1000;
+}
+
diff --git a/src/unstable/speedometer.h b/src/unstable/speedometer.h
new file mode 100644
index 0000000..84d21ed
--- /dev/null
+++ b/src/unstable/speedometer.h
@@ -0,0 +1,43 @@
+#ifndef BU_SPEEDOMETER_H
+#define BU_SPEEDOMETER_H
+
+#include "bu/mutex.h"
+
+#include <stdint.h>
+
+namespace Bu
+{
+    class Speedometer
+    {
+    public:
+        Speedometer( int64_t iBucketMs, int32_t iBucketCount );
+        virtual ~Speedometer();
+
+        void addEvent();
+        double getSpeed();
+        int32_t getTotalEvents();
+        void debug();
+
+    private:
+        void _updateBuckets( int64_t iNow );
+        /**
+         * Gets the current time in milliseconds.
+         */
+        int64_t getCurrentTime() const;
+
+    private:
+        uint32_t *aiBucket;
+        uint32_t iTotal;
+        int32_t iCapacity;
+        int32_t iFill;
+        int32_t iStart;
+        int32_t iCurrent;
+        int64_t iBucketMs;
+        int64_t iStartTime;
+        int64_t iWindowSizeMs;
+        int64_t iLastUpdate;
+        mutable Bu::Mutex mLock;
+    };
+}
+
+#endif