refine code, move the kbps to protocol

trunk/src/protocol/srs_protocol_kbps.cpp

@@ -0,0 +1,252 @@
+/*
+The MIT License (MIT)
+
+Copyright (c) 2013-2015 SRS(simple-rtmp-server)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+
+#include <srs_protocol_kbps.hpp>
+
+#include <srs_kernel_utility.hpp>
+
+SrsKbpsSample::SrsKbpsSample()
+{
+    bytes = time = 0;
+    kbps = 0;
+}
+
+SrsKbpsSlice::SrsKbpsSlice()
+{
+    io.in = NULL;
+    io.out = NULL;
+    last_bytes = io_bytes_base = starttime = bytes = delta_bytes = 0;
+}
+
+SrsKbpsSlice::~SrsKbpsSlice()
+{
+}
+
+int64_t SrsKbpsSlice::get_total_bytes()
+{
+    return bytes + last_bytes - io_bytes_base;
+}
+
+void SrsKbpsSlice::sample()
+{
+    int64_t now = srs_get_system_time_ms();
+    int64_t total_bytes = get_total_bytes();
+    
+    if (sample_30s.time <= 0) {
+        sample_30s.kbps = 0;
+        sample_30s.time = now;
+        sample_30s.bytes = total_bytes;
+    }
+    if (sample_1m.time <= 0) {
+        sample_1m.kbps = 0;
+        sample_1m.time = now;
+        sample_1m.bytes = total_bytes;
+    }
+    if (sample_5m.time <= 0) {
+        sample_5m.kbps = 0;
+        sample_5m.time = now;
+        sample_5m.bytes = total_bytes;
+    }
+    if (sample_60m.time <= 0) {
+        sample_60m.kbps = 0;
+        sample_60m.time = now;
+        sample_60m.bytes = total_bytes;
+    }
+    
+    if (now - sample_30s.time > 30 * 1000) {
+        sample_30s.kbps = (int)((total_bytes - sample_30s.bytes) * 8 / (now - sample_30s.time));
+        sample_30s.time = now;
+        sample_30s.bytes = total_bytes;
+    }
+    if (now - sample_1m.time > 60 * 1000) {
+        sample_1m.kbps = (int)((total_bytes - sample_1m.bytes) * 8 / (now - sample_1m.time));
+        sample_1m.time = now;
+        sample_1m.bytes = total_bytes;
+    }
+    if (now - sample_5m.time > 300 * 1000) {
+        sample_5m.kbps = (int)((total_bytes - sample_5m.bytes) * 8 / (now - sample_5m.time));
+        sample_5m.time = now;
+        sample_5m.bytes = total_bytes;
+    }
+    if (now - sample_60m.time > 3600 * 1000) {
+        sample_60m.kbps = (int)((total_bytes - sample_60m.bytes) * 8 / (now - sample_60m.time));
+        sample_60m.time = now;
+        sample_60m.bytes = total_bytes;
+    }
+}
+
+IKbpsDelta::IKbpsDelta()
+{
+}
+
+IKbpsDelta::~IKbpsDelta()
+{
+}
+
+SrsKbps::SrsKbps()
+{
+}
+
+SrsKbps::~SrsKbps()
+{
+}
+
+void SrsKbps::set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out)
+{
+    // set input stream
+    // now, set start time.
+    if (is.starttime == 0) {
+        is.starttime = srs_get_system_time_ms();
+    }
+    // save the old in bytes.
+    if (is.io.in) {
+        is.bytes += is.last_bytes - is.io_bytes_base;
+    }
+    // use new io.
+    is.io.in = in;
+    is.last_bytes = is.io_bytes_base = 0;
+    if (in) {
+        is.last_bytes = is.io_bytes_base = in->get_recv_bytes();
+    }
+    // resample
+    is.sample();
+    
+    // set output stream
+    // now, set start time.
+    if (os.starttime == 0) {
+        os.starttime = srs_get_system_time_ms();
+    }
+    // save the old in bytes.
+    if (os.io.out) {
+        os.bytes += os.last_bytes - os.io_bytes_base;
+    }
+    // use new io.
+    os.io.out = out;
+    os.last_bytes = os.io_bytes_base = 0;
+    if (out) {
+        os.last_bytes = os.io_bytes_base = out->get_send_bytes();
+    }
+    // resample
+    os.sample();
+}
+
+int SrsKbps::get_send_kbps()
+{
+    int64_t duration = srs_get_system_time_ms() - is.starttime;
+    if (duration <= 0) {
+        return 0;
+    }
+    int64_t bytes = get_send_bytes();
+    return (int)(bytes * 8 / duration);
+}
+
+int SrsKbps::get_recv_kbps()
+{
+    int64_t duration = srs_get_system_time_ms() - os.starttime;
+    if (duration <= 0) {
+        return 0;
+    }
+    int64_t bytes = get_recv_bytes();
+    return (int)(bytes * 8 / duration);
+}
+
+int SrsKbps::get_send_kbps_30s()
+{
+    return os.sample_30s.kbps;
+}
+
+int SrsKbps::get_recv_kbps_30s()
+{
+    return is.sample_30s.kbps;
+}
+
+int SrsKbps::get_send_kbps_5m()
+{
+    return os.sample_5m.kbps;
+}
+
+int SrsKbps::get_recv_kbps_5m()
+{
+    return is.sample_5m.kbps;
+}
+
+int64_t SrsKbps::get_send_bytes()
+{
+    return os.get_total_bytes();
+}
+
+int64_t SrsKbps::get_recv_bytes()
+{
+    return is.get_total_bytes();
+}
+
+void SrsKbps::resample()
+{
+    sample();
+}
+
+int64_t SrsKbps::get_send_bytes_delta()
+{
+    int64_t delta = os.get_total_bytes() - os.delta_bytes;
+    return delta;
+}
+
+int64_t SrsKbps::get_recv_bytes_delta()
+{
+    int64_t delta = is.get_total_bytes() - is.delta_bytes;
+    return delta;
+}
+
+void SrsKbps::cleanup()
+{
+    os.delta_bytes = os.get_total_bytes();
+    is.delta_bytes = is.get_total_bytes();
+}
+
+void SrsKbps::add_delta(IKbpsDelta* delta)
+{
+    srs_assert(delta);
+    
+    // update the total bytes
+    is.last_bytes += delta->get_recv_bytes_delta();
+    os.last_bytes += delta->get_send_bytes_delta();
+    
+    // we donot sample, please use sample() to do resample.
+}
+
+void SrsKbps::sample()
+{
+    // update the total bytes
+    if (os.io.out) {
+        os.last_bytes = os.io.out->get_send_bytes();
+    }
+    
+    if (is.io.in) {
+        is.last_bytes = is.io.in->get_recv_bytes();
+    }
+    
+    // resample
+    is.sample();
+    os.sample();
+}
+

trunk/src/protocol/srs_protocol_kbps.hpp

@@ -0,0 +1,237 @@
+/*
+The MIT License (MIT)
+
+Copyright (c) 2013-2015 SRS(simple-rtmp-server)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+
+#ifndef SRS_PROTOCOL_KBPS_HPP
+#define SRS_PROTOCOL_KBPS_HPP
+
+/*
+#include <srs_protocol_kbps.hpp>
+*/
+
+#include <srs_core.hpp>
+
+#include <srs_rtmp_io.hpp>
+
+/**
+* a kbps sample, for example, 1minute kbps, 
+* 10minute kbps sample.
+*/
+class SrsKbpsSample
+{
+public:
+    int64_t bytes;
+    int64_t time;
+    int kbps;
+public:
+    SrsKbpsSample();
+};
+
+/**
+* a slice of kbps statistic, for input or output.
+* a slice contains a set of sessions, which has a base offset of bytes,
+* where a slice is:
+*       starttime(oldest session startup time)
+*               bytes(total bytes of previous sessions)
+*               io_bytes_base(bytes offset of current session)
+*                       last_bytes(bytes of current session)
+* so, the total send bytes now is:
+*       send_bytes = bytes + last_bytes - io_bytes_base
+* so, the bytes sent duration current session is:
+*       send_bytes = last_bytes - io_bytes_base
+* @remark use set_io to start new session.
+* @remakr the slice is a data collection object driven by SrsKbps.
+*/
+class SrsKbpsSlice
+{
+private:
+    union slice_io {
+        ISrsProtocolStatistic* in;
+        ISrsProtocolStatistic* out;
+    };
+public:
+    // the slice io used for SrsKbps to invoke,
+    // the SrsKbpsSlice itself never use it.
+    slice_io io;
+    // session startup bytes
+    // @remark, use total_bytes() to get the total bytes of slice.
+    int64_t bytes;
+    // slice starttime, the first time to record bytes.
+    int64_t starttime;
+    // session startup bytes number for io when set it,
+    // the base offset of bytes for io.
+    int64_t io_bytes_base;
+    // last updated bytes number,
+    // cache for io maybe freed.
+    int64_t last_bytes;
+    // samples
+    SrsKbpsSample sample_30s;
+    SrsKbpsSample sample_1m;
+    SrsKbpsSample sample_5m;
+    SrsKbpsSample sample_60m;
+public:
+    // for the delta bytes.
+    int64_t delta_bytes;
+public:
+    SrsKbpsSlice();
+    virtual ~SrsKbpsSlice();
+public:
+    /**
+    * get current total bytes.
+    */
+    virtual int64_t get_total_bytes();
+    /**
+    * resample all samples.
+    */
+    virtual void sample();
+};
+
+/**
+* the interface which provices delta of bytes.
+* for a delta, for example, a live stream connection, we can got the delta by:
+*       IKbpsDelta* delta = ...;
+*       delta->resample();
+*       kbps->add_delta(delta);
+*       delta->cleanup();
+*/
+class IKbpsDelta
+{
+public:
+    IKbpsDelta();
+    virtual ~IKbpsDelta();
+public:
+    /**
+    * resample to generate the value of delta bytes.
+    */
+    virtual void resample() = 0;
+    /**
+    * get the send or recv bytes delta.
+    */
+    virtual int64_t get_send_bytes_delta() = 0;
+    virtual int64_t get_recv_bytes_delta() = 0;
+    /**
+    * cleanup the value of delta bytes.
+    */
+    virtual void cleanup() = 0;
+};
+
+/**
+ * to statistic the kbps of io.
+ * itself can be a statistic source, for example, used for SRS bytes stat.
+ * there are some usage scenarios:
+ * 1. connections to calc kbps by sample():
+ *       SrsKbps* kbps = ...;
+ *       kbps->set_io(in, out)
+ *       kbps->sample()
+ *       kbps->get_xxx_kbps().
+ *   the connections know how many bytes already send/recv.
+ * 2. server to calc kbps by add_delta():
+ *       SrsKbps* kbps = ...;
+ *       kbps->set_io(NULL, NULL)
+ *       for each connection in connections:
+ *           IKbpsDelta* delta = connection; // where connection implements IKbpsDelta
+ *           delta->resample()
+ *           kbps->add_delta(delta)
+ *           delta->cleanup()
+ *       kbps->sample()
+ *       kbps->get_xxx_kbps().
+ * 3. kbps used as IKbpsDelta, to provides delta bytes:
+ *      SrsKbps* kbps = ...;
+ *      kbps->set_io(in, out);
+ *      IKbpsDelta* delta = (IKbpsDelta*)kbps;
+ *      delta->resample();
+ *      printf("delta is %d/%d", delta->get_send_bytes_delta(), delta->get_recv_bytes_delta());
+ *      delta->cleanup();
+ *   the server never know how many bytes already send/recv, for the connection maybe closed.
+ */
+class SrsKbps : public virtual ISrsProtocolStatistic, public virtual IKbpsDelta
+{
+private:
+    SrsKbpsSlice is;
+    SrsKbpsSlice os;
+public:
+    SrsKbps();
+    virtual ~SrsKbps();
+public:
+    /**
+    * set io to start new session.
+    * set the underlayer reader/writer,
+    * if the io destroied, for instance, the forwarder reconnect,
+    * user must set the io of SrsKbps to NULL to continue to use the kbps object.
+    * @param in the input stream statistic. can be NULL.
+    * @param out the output stream statistic. can be NULL.
+    * @remark if in/out is NULL, use the cached data for kbps.
+    */
+    virtual void set_io(ISrsProtocolStatistic* in, ISrsProtocolStatistic* out);
+public:
+    /**
+    * get total kbps, duration is from the startup of io.
+    * @remark, use sample() to update data.
+    */
+    virtual int get_send_kbps();
+    virtual int get_recv_kbps();
+    // 30s
+    virtual int get_send_kbps_30s();
+    virtual int get_recv_kbps_30s();
+    // 5m
+    virtual int get_send_kbps_5m();
+    virtual int get_recv_kbps_5m();
+public:
+    /**
+    * get the total send/recv bytes, from the startup of the oldest io.
+    * @remark, use sample() to update data.
+    */
+    virtual int64_t get_send_bytes();
+    virtual int64_t get_recv_bytes();
+public:
+    /**
+    * resample to get the delta.
+    */
+    virtual void resample();
+    /**
+    * get the delta of send/recv bytes.
+    */
+    virtual int64_t get_send_bytes_delta();
+    virtual int64_t get_recv_bytes_delta();
+    /**
+    * cleanup the delta.
+    */
+    virtual void cleanup();
+public:
+    /**
+    * add delta to kbps clac mechenism.
+    * we donot know the total bytes, but know the delta, for instance, 
+    * for rtmp server to calc total bytes and kbps.
+    * @remark user must invoke sample() to calc result after invoke this method.
+    * @param delta, assert should never be NULL.
+    */
+    virtual void add_delta(IKbpsDelta* delta);
+    /**
+    * resample all samples, ignore if in/out is NULL.
+    * used for user to calc the kbps, to sample new kbps value.
+    * @remark if user, for instance, the rtmp server to calc the total bytes,
+    *       use the add_delta() is better solutions.
+    */
+    virtual void sample();
+};
+
+#endif