external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

Refine typo in app.

Browse source
This commit is contained in:
winlin 2019-04-28 09:08:05 +08:00
parent 4d25520f99
commit aac8a13f42
15 changed files with 1065 additions and 1962 deletions
Download patch file Download diff file Expand all files Collapse all files

35
trunk/src/app/srs_app_async_call.hpp
View file

@ -31,38 +31,29 @@
#include <srs_app_thread.hpp> #include <srs_app_thread.hpp>
/** // The async call for http hooks, for the http hooks will switch st-thread,
* the async call for http hooks, // so we must use isolate thread to avoid the thread corrupt,
* for the http hooks will switch st-thread, // for example, when dvr call http hooks, the video receive thread got
* so we must use isolate thread to avoid the thread corrupt, // a video and pass it to the dvr again.
* for example, when dvr call http hooks, the video receive thread got // Futhurmore, the aync call never block the main worker thread.
* a video and pass it to the dvr again.
* futhurmore, the aync call never block the main worker thread.
*/
class ISrsAsyncCallTask class ISrsAsyncCallTask
{ {
public: public:
ISrsAsyncCallTask(); ISrsAsyncCallTask();
virtual ~ISrsAsyncCallTask(); virtual ~ISrsAsyncCallTask();
public: public:
/** // Execute the task async.
* execute the task async. // This method is the actual execute method of task,
* this method is the actual execute method of task, // for example, to notify callback server.
* for example, to notify callback server.
*/
virtual srs_error_t call() = 0; virtual srs_error_t call() = 0;
/** // Convert task to string to describe it.
* convert task to string to describe it. // It's used for logger.
* used for logger.
*/
virtual std::string to_string() = 0; virtual std::string to_string() = 0;
}; };
/** // The async callback for dvr, callback and other async worker.
* the async callback for dvr, callback and other async worker. // When worker call with the task, the worker will do it in isolate thread.
* when worker call with the task, the worker will do it in isolate thread. // That is, the task is execute/call in async mode.
* that is, the task is execute/call in async mode.
*/
class SrsAsyncCallWorker : public ISrsCoroutineHandler class SrsAsyncCallWorker : public ISrsCoroutineHandler
{ {
private: private:

218
trunk/src/app/srs_app_bandwidth.hpp
View file

@ -36,85 +36,69 @@ class SrsRtmpServer;
class SrsKbpsLimit; class SrsKbpsLimit;
class ISrsProtocolStatistic; class ISrsProtocolStatistic;
/** // The bandwidth check/test sample.
* bandwidth check/test sample.
*/
class SrsBandwidthSample class SrsBandwidthSample
{ {
public: public:
/** // The plan, how long to do the test, in ms,
* the plan, how long to do the test, in ms, // if exceed the duration, abort the test.
* if exceed the duration, abort the test.
*/
int duration_ms; int duration_ms;
/** // The plan, interval for each check/test packet, in ms
* the plan, interval for each check/test packet, in ms
*/
int interval_ms; int interval_ms;
public: public:
/** // The actual test duration, in ms.
* the actual test duration, in ms.
*/
int actual_duration_ms; int actual_duration_ms;
/** // The actual test bytes
* the actual test bytes
*/
int bytes; int bytes;
/** // The actual test kbps
* the actual test kbps
*/
int kbps; int kbps;
public: public:
SrsBandwidthSample(); SrsBandwidthSample();
virtual ~SrsBandwidthSample(); virtual ~SrsBandwidthSample();
public: public:
/** // Update the bytes and actual duration, then calc the kbps.
* update the bytes and actual duration, then calc the kbps. // @param _bytes update the sample bytes.
* @param _bytes update the sample bytes. // @param _duration update the actual duration, in ms.
* @param _duration update the actual duration, in ms.
*/
virtual void calc_kbps(int _bytes, int _duration); virtual void calc_kbps(int _bytes, int _duration);
}; };
/** // The bandwidth test agent which provides the interfaces for bandwidth check.
* bandwidth test agent which provides the interfaces for bandwidth check. // 1. if vhost disabled bandwidth check, ignore.
* 1. if vhost disabled bandwidth check, ignore. // 2. otherwise, check the key, error if verify failed.
* 2. otherwise, check the key, error if verify failed. // 3. check the interval limit, error if bandwidth in the interval window.
* 3. check the interval limit, error if bandwidth in the interval window. // 4. check the bandwidth under the max kbps.
* 4. check the bandwidth under the max kbps. // 5. send the bandwidth data to client.
* 5. send the bandwidth data to client. // bandwidth workflow:
* bandwidth workflow: // +------------+ +----------+
* +------------+ +----------+ // | Client | | Server |
* | Client | | Server | // +-----+------+ +-----+----+
* +-----+------+ +-----+----+ // | |
* | | // | connect vhost------> | if vhost enable bandwidth,
* | connect vhost------> | if vhost enable bandwidth, // | <-----result(success) | do bandwidth check.
* | <-----result(success) | do bandwidth check. // | |
* | | // | <----call(start play) | onSrsBandCheckStartPlayBytes
* | <----call(start play) | onSrsBandCheckStartPlayBytes // | result(playing)-----> | onSrsBandCheckStartingPlayBytes
* | result(playing)-----> | onSrsBandCheckStartingPlayBytes // | <-------data(playing) | onSrsBandCheckStartingPlayBytes
* | <-------data(playing) | onSrsBandCheckStartingPlayBytes // | <-----call(stop play) | onSrsBandCheckStopPlayBytes
* | <-----call(stop play) | onSrsBandCheckStopPlayBytes // | result(stopped)-----> | onSrsBandCheckStoppedPlayBytes
* | result(stopped)-----> | onSrsBandCheckStoppedPlayBytes // | |
* | | // | <-call(start publish) | onSrsBandCheckStartPublishBytes
* | <-call(start publish) | onSrsBandCheckStartPublishBytes // | result(publishing)--> | onSrsBandCheckStartingPublishBytes
* | result(publishing)--> | onSrsBandCheckStartingPublishBytes // | data(publishing)(3)-> | onSrsBandCheckStartingPublishBytes
* | data(publishing)(3)-> | onSrsBandCheckStartingPublishBytes // | <--call(stop publish) | onSrsBandCheckStopPublishBytes
* | <--call(stop publish) | onSrsBandCheckStopPublishBytes // | result(stopped)(1)--> | onSrsBandCheckStoppedPublishBytes
* | result(stopped)(1)--> | onSrsBandCheckStoppedPublishBytes // | |
* | | // | <--------------report |
* | <--------------report | // | final(2)------------> | finalClientPacket
* | final(2)------------> | finalClientPacket // | <END> |
* | <END> | //
* // 1. when flash client, server never wait the stop publish response,
* 1. when flash client, server never wait the stop publish response, // for the flash client queue is fullfill with other packets.
* for the flash client queue is fullfill with other packets. // 2. when flash client, server never wait the final packet,
* 2. when flash client, server never wait the final packet, // for the flash client directly close when got report packet.
* for the flash client directly close when got report packet. // 3. for linux client, it will send the publish data then send a stop publish,
* 3. for linux client, it will send the publish data then send a stop publish, // for the linux client donot know when to stop the publish.
* for the linux client donot know when to stop the publish. // when server got publishing and stop publish, stop publish.
* when server got publishing and stop publish, stop publish.
*/
class SrsBandwidth class SrsBandwidth
{ {
private: private:
@ -124,81 +108,61 @@ public:
SrsBandwidth(); SrsBandwidth();
virtual ~SrsBandwidth(); virtual ~SrsBandwidth();
public: public:
/** // Do the bandwidth check.
* do the bandwidth check. // @param rtmp, server RTMP protocol object, send/recv RTMP packet to/from client.
* @param rtmp, server RTMP protocol object, send/recv RTMP packet to/from client. // @param io_stat, the underlayer io statistic, provides send/recv bytes count.
* @param io_stat, the underlayer io statistic, provides send/recv bytes count. // @param req, client request object, specifies the request info from client.
* @param req, client request object, specifies the request info from client. // @param local_ip, the ip of server which client connected at
* @param local_ip, the ip of server which client connected at
*/
virtual srs_error_t bandwidth_check(SrsRtmpServer* rtmp, ISrsProtocolStatistic* io_stat, SrsRequest* req, std::string local_ip); virtual srs_error_t bandwidth_check(SrsRtmpServer* rtmp, ISrsProtocolStatistic* io_stat, SrsRequest* req, std::string local_ip);
private: private:
/** // Used to process band width check from client.
* used to process band width check from client. // @param limit, the bandwidth limit object, to slowdown if exceed the kbps.
* @param limit, the bandwidth limit object, to slowdown if exceed the kbps.
*/
virtual srs_error_t do_bandwidth_check(SrsKbpsLimit* limit); virtual srs_error_t do_bandwidth_check(SrsKbpsLimit* limit);
// play check/test, downloading bandwidth kbps. // play check/test, downloading bandwidth kbps.
private: private:
/** // Start play/download bandwidth check/test,
* start play/download bandwidth check/test, // send start-play command to client, client must response starting-play
* send start-play command to client, client must response starting-play // to start the test.
* to start the test.
*/
virtual srs_error_t play_start(SrsBandwidthSample* sample, SrsKbpsLimit* limit); virtual srs_error_t play_start(SrsBandwidthSample* sample, SrsKbpsLimit* limit);
/** // Do play/download bandwidth check/test,
* do play/download bandwidth check/test, // server send call messages to client in specified time,
* server send call messages to client in specified time, // calc the time and bytes sent, then we got the kbps.
* calc the time and bytes sent, then we got the kbps.
*/
virtual srs_error_t play_checking(SrsBandwidthSample* sample, SrsKbpsLimit* limit); virtual srs_error_t play_checking(SrsBandwidthSample* sample, SrsKbpsLimit* limit);
/** // stop play/download bandwidth check/test,
* stop play/download bandwidth check/test, // send stop-play command to client, client must response stopped-play
* send stop-play command to client, client must response stopped-play // to stop the test.
* to stop the test.
*/
virtual srs_error_t play_stop(SrsBandwidthSample* sample, SrsKbpsLimit* limit); virtual srs_error_t play_stop(SrsBandwidthSample* sample, SrsKbpsLimit* limit);
// publish check/test, publishing bandwidth kbps. // publish check/test, publishing bandwidth kbps.
private: private:
/** // Start publish/upload bandwidth check/test,
* start publish/upload bandwidth check/test, // send start-publish command to client, client must response starting-publish
* send start-publish command to client, client must response starting-publish // to start the test.
* to start the test.
*/
virtual srs_error_t publish_start(SrsBandwidthSample* sample, SrsKbpsLimit* limit); virtual srs_error_t publish_start(SrsBandwidthSample* sample, SrsKbpsLimit* limit);
/** // Do publish/upload bandwidth check/test,
* do publish/upload bandwidth check/test, // client send call messages to client in specified time,
* client send call messages to client in specified time, // server calc the time and bytes received, then we got the kbps.
* server calc the time and bytes received, then we got the kbps. // @remark, for linux client, it will send a stop publish client, server will stop publishing.
* @remark, for linux client, it will send a stop publish client, server will stop publishing. // then enter the publish-stop stage with client.
* then enter the publish-stop stage with client. // @remark, for flash client, it will send many many call messages, that is,
* @remark, for flash client, it will send many many call messages, that is, // the send queue is fullfill with call messages, so we should never expect the
* the send queue is fullfill with call messages, so we should never expect the // response message in the publish-stop stage.
* response message in the publish-stop stage.
*/
virtual srs_error_t publish_checking(SrsBandwidthSample* sample, SrsKbpsLimit* limit); virtual srs_error_t publish_checking(SrsBandwidthSample* sample, SrsKbpsLimit* limit);
/** // Stop publish/upload bandwidth check/test,
* stop publish/upload bandwidth check/test, // send stop-publish command to client,
* send stop-publish command to client, // for linux client, always expect a stopped-publish response from client,
* for linux client, always expect a stopped-publish response from client, // for flash client, the sent queue is fullfill with publishing call messages,
* for flash client, the sent queue is fullfill with publishing call messages, // so server never expect the stopped-publish from it.
* so server never expect the stopped-publish from it.
*/
virtual srs_error_t publish_stop(SrsBandwidthSample* sample, SrsKbpsLimit* limit); virtual srs_error_t publish_stop(SrsBandwidthSample* sample, SrsKbpsLimit* limit);
private: private:
/** // Report and final packet
* report and final packet // report a finish packet, with the bytes/time/kbps bandwidth check/test result,
* report a finish packet, with the bytes/time/kbps bandwidth check/test result, // for linux client, server always expect a final packet from client,
* for linux client, server always expect a final packet from client, // for flash client, the sent queue is fullfill with publishing call messages,
* for flash client, the sent queue is fullfill with publishing call messages, // so server never expect the final packet from it.
* so server never expect the final packet from it.
*/
virtual srs_error_t do_final(SrsBandwidthSample& play_sample, SrsBandwidthSample& publish_sample, srs_utime_t start_time, srs_utime_t& end_time); virtual srs_error_t do_final(SrsBandwidthSample& play_sample, SrsBandwidthSample& publish_sample, srs_utime_t start_time, srs_utime_t& end_time);
}; };
/** // The kbps limit, if exceed the kbps, slow down.
* the kbps limit, if exceed the kbps, slow down.
*/
class SrsKbpsLimit class SrsKbpsLimit
{ {
private: private:
@ -208,17 +172,11 @@ public:
SrsKbpsLimit(SrsKbps* kbps, int limit_kbps); SrsKbpsLimit(SrsKbps* kbps, int limit_kbps);
virtual ~SrsKbpsLimit(); virtual ~SrsKbpsLimit();
public: public:
/** // Get the system limit kbps.
* get the system limit kbps.
*/
virtual int limit_kbps(); virtual int limit_kbps();
/** // Limit the recv bandwidth.
* limit the recv bandwidth.
*/
virtual void recv_limit(); virtual void recv_limit();
/** // Limit the send bandwidth.
* limit the send bandwidth.
*/
virtual void send_limit(); virtual void send_limit();
}; };

23
trunk/src/app/srs_app_caster_flv.hpp
View file

@ -46,9 +46,7 @@ class SrsSimpleRtmpClient;
#include <srs_app_http_conn.hpp> #include <srs_app_http_conn.hpp>
#include <srs_kernel_file.hpp> #include <srs_kernel_file.hpp>
/** // The stream caster for flv stream over HTTP POST.
* the stream caster for flv stream over HTTP POST.
*/
class SrsAppCasterFlv : virtual public ISrsTcpHandler class SrsAppCasterFlv : virtual public ISrsTcpHandler
, virtual public IConnectionManager, virtual public ISrsHttpHandler , virtual public IConnectionManager, virtual public ISrsHttpHandler
{ {
@ -62,20 +60,18 @@ public:
virtual ~SrsAppCasterFlv(); virtual ~SrsAppCasterFlv();
public: public:
virtual srs_error_t initialize(); virtual srs_error_t initialize();
// ISrsTcpHandler // interface ISrsTcpHandler
public: public:
virtual srs_error_t on_tcp_client(srs_netfd_t stfd); virtual srs_error_t on_tcp_client(srs_netfd_t stfd);
// IConnectionManager // interface IConnectionManager
public: public:
virtual void remove(ISrsConnection* c); virtual void remove(ISrsConnection* c);
// ISrsHttpHandler // interface ISrsHttpHandler
public: public:
virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r); virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
}; };
/** // The dynamic http connection, never drop the body.
* the dynamic http connection, never drop the body.
*/
class SrsDynamicHttpConn : public SrsHttpConn class SrsDynamicHttpConn : public SrsHttpConn
{ {
private: private:
@ -93,10 +89,7 @@ private:
virtual srs_error_t do_proxy(ISrsHttpResponseReader* rr, SrsFlvDecoder* dec); virtual srs_error_t do_proxy(ISrsHttpResponseReader* rr, SrsFlvDecoder* dec);
}; };
/** // The http wrapper for file reader, to read http post stream like a file.
* the http wrapper for file reader,
* to read http post stream like a file.
*/
class SrsHttpFileReader : public SrsFileReader class SrsHttpFileReader : public SrsFileReader
{ {
private: private:
@ -105,9 +98,7 @@ public:
SrsHttpFileReader(ISrsHttpResponseReader* h); SrsHttpFileReader(ISrsHttpResponseReader* h);
virtual ~SrsHttpFileReader(); virtual ~SrsHttpFileReader();
public: public:
/** // Open file reader, can open then close then open...
* open file reader, can open then close then open...
*/
virtual srs_error_t open(std::string file); virtual srs_error_t open(std::string file);
virtual void close(); virtual void close();
public: public:

1420
trunk/src/app/srs_app_config.hpp
View file

File diff suppressed because it is too large Load diff

86
trunk/src/app/srs_app_conn.hpp
View file

@ -36,48 +36,32 @@
class SrsWallClock; class SrsWallClock;
/** // The basic connection of SRS,
* the basic connection of SRS, // all connections accept from listener must extends from this base class,
* all connections accept from listener must extends from this base class, // server will add the connection to manager, and delete it when remove.
* server will add the connection to manager, and delete it when remove.
*/
class SrsConnection : virtual public ISrsConnection, virtual public ISrsCoroutineHandler class SrsConnection : virtual public ISrsConnection, virtual public ISrsCoroutineHandler
, virtual public ISrsKbpsDelta, virtual public ISrsReloadHandler , virtual public ISrsKbpsDelta, virtual public ISrsReloadHandler
{ {
protected: protected:
/** // Each connection start a green thread,
* each connection start a green thread, // when thread stop, the connection will be delete by server.
* when thread stop, the connection will be delete by server.
*/
SrsCoroutine* trd; SrsCoroutine* trd;
/** // The manager object to manage the connection.
* the manager object to manage the connection.
*/
IConnectionManager* manager; IConnectionManager* manager;
/** // The underlayer st fd handler.
* the underlayer st fd handler.
*/
srs_netfd_t stfd; srs_netfd_t stfd;
/** // The ip of client.
* the ip of client.
*/
std::string ip; std::string ip;
/** // The underlayer socket.
* the underlayer socket.
*/
SrsStSocket* skt; SrsStSocket* skt;
/** // The connection total kbps.
* connection total kbps. // not only the rtmp or http connection, all type of connection are
* not only the rtmp or http connection, all type of connection are // need to statistic the kbps of io.
* need to statistic the kbps of io. // The SrsStatistic will use it indirectly to statistic the bytes delta of current connection.
* the SrsStatistic will use it indirectly to statistic the bytes delta of current connection.
*/
SrsKbps* kbps; SrsKbps* kbps;
SrsWallClock* clk; SrsWallClock* clk;
/** // The create time in milliseconds.
* the create time in milliseconds. // for current connection to log self create time and calculate the living time.
* for current connection to log self create time and calculate the living time.
*/
int64_t create_time; int64_t create_time;
public: public:
SrsConnection(IConnectionManager* cm, srs_netfd_t c, std::string cip); SrsConnection(IConnectionManager* cm, srs_netfd_t c, std::string cip);
@ -86,19 +70,15 @@ public:
public: public:
virtual void remark(int64_t* in, int64_t* out); virtual void remark(int64_t* in, int64_t* out);
public: public:
/** // To dipose the connection.
* to dipose the connection.
*/
virtual void dispose(); virtual void dispose();
/** // Start the client green thread.
* start the client green thread. // when server get a client from listener,
* when server get a client from listener, // 1. server will create an concrete connection(for instance, RTMP connection),
* 1. server will create an concrete connection(for instance, RTMP connection), // 2. then add connection to its connection manager,
* 2. then add connection to its connection manager, // 3. start the client thread by invoke this start()
* 3. start the client thread by invoke this start() // when client cycle thread stop, invoke the on_thread_stop(), which will use server
* when client cycle thread stop, invoke the on_thread_stop(), which will use server // To remove the client by server->remove(this).
* to remove the client by server->remove(this).
*/
virtual srs_error_t start(); virtual srs_error_t start();
// Set socket option TCP_NODELAY. // Set socket option TCP_NODELAY.
virtual srs_error_t set_tcp_nodelay(bool v); virtual srs_error_t set_tcp_nodelay(bool v);
@ -106,25 +86,17 @@ public:
virtual srs_error_t set_socket_buffer(srs_utime_t buffer_v); virtual srs_error_t set_socket_buffer(srs_utime_t buffer_v);
// interface ISrsOneCycleThreadHandler // interface ISrsOneCycleThreadHandler
public: public:
/** // The thread cycle function,
* the thread cycle function, // when serve connection completed, terminate the loop which will terminate the thread,
* when serve connection completed, terminate the loop which will terminate the thread, // thread will invoke the on_thread_stop() when it terminated.
* thread will invoke the on_thread_stop() when it terminated.
*/
virtual srs_error_t cycle(); virtual srs_error_t cycle();
public: public:
/** // Get the srs id which identify the client.
* get the srs id which identify the client.
*/
virtual int srs_id(); virtual int srs_id();
/** // Set connection to expired.
* set connection to expired.
*/
virtual void expire(); virtual void expire();
protected: protected:
/** // For concrete connection to do the cycle.
* for concrete connection to do the cycle.
*/
virtual srs_error_t do_cycle() = 0; virtual srs_error_t do_cycle() = 0;
}; };

1
trunk/src/app/srs_app_coworkers.hpp
View file

@ -33,6 +33,7 @@ class SrsJsonAny;
class SrsRequest; class SrsRequest;
class SrsSource; class SrsSource;
// For origin cluster.
class SrsCoWorkers class SrsCoWorkers
{ {
private: private:

5
trunk/src/app/srs_app_heartbeat.hpp
View file

@ -26,10 +26,7 @@
#include <srs_core.hpp> #include <srs_core.hpp>
/** // The http heartbeat to api-server to notice api that the information of SRS.
* the http heartbeat to api-server to notice api
* that the information of SRS.
*/
class SrsHttpHeartbeat class SrsHttpHeartbeat
{ {
public: public:

196
trunk/src/app/srs_app_hls.hpp
View file

@ -50,12 +50,10 @@ class SrsTsMessageCache;
class SrsHlsSegment; class SrsHlsSegment;
class SrsTsContext; class SrsTsContext;
/** // The wrapper of m3u8 segment from specification:
* the wrapper of m3u8 segment from specification: //
* // 3.3.2. EXTINF
* 3.3.2. EXTINF // The EXTINF tag specifies the duration of a media segment.
* The EXTINF tag specifies the duration of a media segment.
*/
class SrsHlsSegment : public SrsFragment class SrsHlsSegment : public SrsFragment
{ {
public: public:
@ -63,7 +61,7 @@ public:
int sequence_no; int sequence_no;
// ts uri in m3u8. // ts uri in m3u8.
std::string uri; std::string uri;
// the underlayer file writer. // The underlayer file writer.
SrsFileWriter* writer; SrsFileWriter* writer;
// The TS context writer to write TS to file. // The TS context writer to write TS to file.
SrsTsContextWriter* tscw; SrsTsContextWriter* tscw;
@ -78,9 +76,7 @@ public:
void config_cipher(unsigned char* key,unsigned char* iv); void config_cipher(unsigned char* key,unsigned char* iv);
}; };
/** // The hls async call: on_hls
* the hls async call: on_hls
*/
class SrsDvrAsyncCallOnHls : public ISrsAsyncCallTask class SrsDvrAsyncCallOnHls : public ISrsAsyncCallTask
{ {
private: private:
@ -101,9 +97,7 @@ public:
virtual std::string to_string(); virtual std::string to_string();
}; };
/** // The hls async call: on_hls_notify
* the hls async call: on_hls_notify
*/
class SrsDvrAsyncCallOnHlsNotify : public ISrsAsyncCallTask class SrsDvrAsyncCallOnHlsNotify : public ISrsAsyncCallTask
{ {
private: private:
@ -118,14 +112,12 @@ public:
virtual std::string to_string(); virtual std::string to_string();
}; };
/** // Mux the HLS stream(m3u8 and ts files).
* muxer the HLS stream(m3u8 and ts files). // Generally, the m3u8 muxer only provides methods to open/close segments,
* generally, the m3u8 muxer only provides methods to open/close segments, // to flush video/audio, without any mechenisms.
* to flush video/audio, without any mechenisms. //
* // That is, user must use HlsCache, which will control the methods of muxer,
* that is, user must use HlsCache, which will control the methods of muxer, // and provides HLS mechenisms.
* and provides HLS mechenisms.
*/
class SrsHlsMuxer class SrsHlsMuxer
{ {
private: private:
@ -143,26 +135,26 @@ private:
srs_utime_t hls_window; srs_utime_t hls_window;
SrsAsyncCallWorker* async; SrsAsyncCallWorker* async;
private: private:
// whether use floor algorithm for timestamp. // Whether use floor algorithm for timestamp.
bool hls_ts_floor; bool hls_ts_floor;
// the deviation in piece to adjust the fragment to be more // The deviation in piece to adjust the fragment to be more
// bigger or smaller. // bigger or smaller.
int deviation_ts; int deviation_ts;
// the previous reap floor timestamp, // The previous reap floor timestamp,
// used to detect the dup or jmp or ts. // used to detect the dup or jmp or ts.
int64_t accept_floor_ts; int64_t accept_floor_ts;
int64_t previous_floor_ts; int64_t previous_floor_ts;
private: private:
// encrypted or not // Whether encrypted or not
bool hls_keys; bool hls_keys;
int hls_fragments_per_key; int hls_fragments_per_key;
// key file name // The key file name
std::string hls_key_file; std::string hls_key_file;
// key file path // The key file path
std::string hls_key_file_path; std::string hls_key_file_path;
// key file url // The key file url
std::string hls_key_url; std::string hls_key_url;
// key and iv. // The key and iv.
unsigned char key[16]; unsigned char key[16];
unsigned char iv[16]; unsigned char iv[16];
SrsFileWriter *writer; SrsFileWriter *writer;
@ -176,10 +168,8 @@ private:
SrsFragmentWindow* segments; SrsFragmentWindow* segments;
// The current writing segment. // The current writing segment.
SrsHlsSegment* current; SrsHlsSegment* current;
/** // The ts context, to keep cc continous between ts.
* the ts context, to keep cc continous between ts. // @see https://github.com/ossrs/srs/issues/375
* @see https://github.com/ossrs/srs/issues/375
*/
SrsTsContext* context; SrsTsContext* context;
public: public:
SrsHlsMuxer(); SrsHlsMuxer();
@ -192,48 +182,32 @@ public:
virtual srs_utime_t duration(); virtual srs_utime_t duration();
virtual int deviation(); virtual int deviation();
public: public:
/** // Initialize the hls muxer.
* initialize the hls muxer.
*/
virtual srs_error_t initialize(); virtual srs_error_t initialize();
/** // When publish, update the config for muxer.
* when publish, update the config for muxer.
*/
virtual srs_error_t update_config(SrsRequest* r, std::string entry_prefix, virtual srs_error_t update_config(SrsRequest* r, std::string entry_prefix,
std::string path, std::string m3u8_file, std::string ts_file, std::string path, std::string m3u8_file, std::string ts_file,
srs_utime_t fragment, srs_utime_t window, bool ts_floor, double aof_ratio, srs_utime_t fragment, srs_utime_t window, bool ts_floor, double aof_ratio,
bool cleanup, bool wait_keyframe, bool keys, int fragments_per_key, bool cleanup, bool wait_keyframe, bool keys, int fragments_per_key,
std::string key_file, std::string key_file_path, std::string key_url); std::string key_file, std::string key_file_path, std::string key_url);
/** // Open a new segment(a new ts file)
* open a new segment(a new ts file)
*/
virtual srs_error_t segment_open(); virtual srs_error_t segment_open();
virtual srs_error_t on_sequence_header(); virtual srs_error_t on_sequence_header();
/** // Whether segment overflow,
* whether segment overflow, // that is whether the current segment duration>=(the segment in config)
* that is whether the current segment duration>=(the segment in config)
*/
virtual bool is_segment_overflow(); virtual bool is_segment_overflow();
/** // Whether wait keyframe to reap the ts.
* whether wait keyframe to reap the ts.
*/
virtual bool wait_keyframe(); virtual bool wait_keyframe();
/** // Whether segment absolutely overflow, for pure audio to reap segment,
* whether segment absolutely overflow, for pure audio to reap segment, // that is whether the current segment duration>=2*(the segment in config)
* that is whether the current segment duration>=2*(the segment in config) // @see https://github.com/ossrs/srs/issues/151#issuecomment-71155184
* @see https://github.com/ossrs/srs/issues/151#issuecomment-71155184
*/
virtual bool is_segment_absolutely_overflow(); virtual bool is_segment_absolutely_overflow();
public: public:
/** // Whether current hls muxer is pure audio mode.
* whether current hls muxer is pure audio mode.
*/
virtual bool pure_audio(); virtual bool pure_audio();
virtual srs_error_t flush_audio(SrsTsMessageCache* cache); virtual srs_error_t flush_audio(SrsTsMessageCache* cache);
virtual srs_error_t flush_video(SrsTsMessageCache* cache); virtual srs_error_t flush_video(SrsTsMessageCache* cache);
/** // Close segment(ts).
* Close segment(ts).
*/
virtual srs_error_t segment_close(); virtual srs_error_t segment_close();
private: private:
virtual srs_error_t do_segment_close(); virtual srs_error_t do_segment_close();
@ -242,23 +216,21 @@ private:
virtual srs_error_t _refresh_m3u8(std::string m3u8_file); virtual srs_error_t _refresh_m3u8(std::string m3u8_file);
}; };
/** // The hls stream cache,
* hls stream cache, // use to cache hls stream and flush to hls muxer.
* use to cache hls stream and flush to hls muxer. //
* // When write stream to ts file:
* when write stream to ts file: // video frame will directly flush to M3u8Muxer,
* video frame will directly flush to M3u8Muxer, // audio frame need to cache, because it's small and flv tbn problem.
* audio frame need to cache, because it's small and flv tbn problem. //
* // Whatever, the Hls cache used to cache video/audio,
* whatever, the Hls cache used to cache video/audio, // and flush video/audio to m3u8 muxer if needed.
* and flush video/audio to m3u8 muxer if needed. //
* // About the flv tbn problem:
* about the flv tbn problem: // flv tbn is 1/1000, ts tbn is 1/90000,
* flv tbn is 1/1000, ts tbn is 1/90000, // when timestamp convert to flv tbn, it will loose precise,
* when timestamp convert to flv tbn, it will loose precise, // so we must gather audio frame together, and recalc the timestamp @see SrsTsAacJitter,
* so we must gather audio frame together, and recalc the timestamp @see SrsTsAacJitter, // we use a aac jitter to correct the audio pts.
* we use a aac jitter to correct the audio pts.
*/
class SrsHlsController class SrsHlsController
{ {
private: private:
@ -278,40 +250,28 @@ public:
virtual srs_utime_t duration(); virtual srs_utime_t duration();
virtual int deviation(); virtual int deviation();
public: public:
/** // When publish or unpublish stream.
* when publish or unpublish stream.
*/
virtual srs_error_t on_publish(SrsRequest* req); virtual srs_error_t on_publish(SrsRequest* req);
virtual srs_error_t on_unpublish(); virtual srs_error_t on_unpublish();
/** // When get sequence header,
* when get sequence header, // must write a #EXT-X-DISCONTINUITY to m3u8.
* must write a #EXT-X-DISCONTINUITY to m3u8. // @see: hls-m3u8-draft-pantos-http-live-streaming-12.txt
* @see: hls-m3u8-draft-pantos-http-live-streaming-12.txt // @see: 3.4.11. EXT-X-DISCONTINUITY
* @see: 3.4.11. EXT-X-DISCONTINUITY
*/
virtual srs_error_t on_sequence_header(); virtual srs_error_t on_sequence_header();
/** // write audio to cache, if need to flush, flush to muxer.
* write audio to cache, if need to flush, flush to muxer.
*/
virtual srs_error_t write_audio(SrsAudioFrame* frame, int64_t pts); virtual srs_error_t write_audio(SrsAudioFrame* frame, int64_t pts);
/** // write video to muxer.
* write video to muxer.
*/
virtual srs_error_t write_video(SrsVideoFrame* frame, int64_t dts); virtual srs_error_t write_video(SrsVideoFrame* frame, int64_t dts);
private: private:
/** // Reopen the muxer for a new hls segment,
* reopen the muxer for a new hls segment, // close current segment, open a new segment,
* close current segment, open a new segment, // then write the key frame to the new segment.
* then write the key frame to the new segment. // so, user must reap_segment then flush_video to hls muxer.
* so, user must reap_segment then flush_video to hls muxer.
*/
virtual srs_error_t reap_segment(); virtual srs_error_t reap_segment();
}; };
/** // Transmux RTMP stream to HLS(m3u8 and ts).
* Transmux RTMP stream to HLS(m3u8 and ts). // TODO: FIXME: add utest for hls.
* TODO: FIXME: add utest for hls.
*/
class SrsHls class SrsHls
{ {
private: private:
@ -338,31 +298,21 @@ public:
virtual void dispose(); virtual void dispose();
virtual srs_error_t cycle(); virtual srs_error_t cycle();
public: public:
/** // Initialize the hls by handler and source.
* initialize the hls by handler and source.
*/
virtual srs_error_t initialize(SrsOriginHub* h, SrsRequest* r); virtual srs_error_t initialize(SrsOriginHub* h, SrsRequest* r);
/** // Publish stream event, continue to write the m3u8,
* publish stream event, continue to write the m3u8, // for the muxer object not destroyed.
* for the muxer object not destroyed. // @param fetch_sequence_header whether fetch sequence from source.
* @param fetch_sequence_header whether fetch sequence from source.
*/
virtual srs_error_t on_publish(); virtual srs_error_t on_publish();
/** // The unpublish event, only close the muxer, donot destroy the
* the unpublish event, only close the muxer, donot destroy the // muxer, for when we continue to publish, the m3u8 will continue.
* muxer, for when we continue to publish, the m3u8 will continue.
*/
virtual void on_unpublish(); virtual void on_unpublish();
/** // Mux the audio packets to ts.
* mux the audio packets to ts. // @param shared_audio, directly ptr, copy it if need to save it.
* @param shared_audio, directly ptr, copy it if need to save it.
*/
virtual srs_error_t on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format); virtual srs_error_t on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format);
/** // Mux the video packets to ts.
* mux the video packets to ts. // @param shared_video, directly ptr, copy it if need to save it.
* @param shared_video, directly ptr, copy it if need to save it. // @param is_sps_pps whether the video is h.264 sps/pps.
* @param is_sps_pps whether the video is h.264 sps/pps.
*/
// TODO: FIXME: Remove param is_sps_pps. // TODO: FIXME: Remove param is_sps_pps.
virtual srs_error_t on_video(SrsSharedPtrMessage* shared_video, SrsFormat* format); virtual srs_error_t on_video(SrsSharedPtrMessage* shared_video, SrsFormat* format);
private: private:

19
trunk/src/app/srs_app_ingest.hpp
View file

@ -35,9 +35,7 @@ class SrsFFMPEG;
class SrsConfDirective; class SrsConfDirective;
class SrsPithyPrint; class SrsPithyPrint;
/** // Ingester ffmpeg object.
* ingester ffmpeg object.
*/
class SrsIngesterFFMPEG class SrsIngesterFFMPEG
{ {
private: private:
@ -50,9 +48,9 @@ public:
virtual ~SrsIngesterFFMPEG(); virtual ~SrsIngesterFFMPEG();
public: public:
virtual srs_error_t initialize(SrsFFMPEG* ff, std::string v, std::string i); virtual srs_error_t initialize(SrsFFMPEG* ff, std::string v, std::string i);
// the ingest uri, [vhost]/[ingest id] // The ingest uri, [vhost]/[ingest id]
virtual std::string uri(); virtual std::string uri();
// the alive in srs_utime_t. // The alive in srs_utime_t.
virtual srs_utime_t alive(); virtual srs_utime_t alive();
virtual bool equals(std::string v, std::string i); virtual bool equals(std::string v, std::string i);
virtual bool equals(std::string v); virtual bool equals(std::string v);
@ -64,11 +62,9 @@ public:
virtual void fast_stop(); virtual void fast_stop();
}; };
/** // Ingest file/stream/device,
* ingest file/stream/device, // encode with FFMPEG(optional),
* encode with FFMPEG(optional), // push to SRS(or any RTMP server) over RTMP.
* push to SRS(or any RTMP server) over RTMP.
*/
class SrsIngester : public ISrsCoroutineHandler, public ISrsReloadHandler class SrsIngester : public ISrsCoroutineHandler, public ISrsReloadHandler
{ {
private: private:
@ -76,8 +72,7 @@ private:
private: private:
SrsCoroutine* trd; SrsCoroutine* trd;
SrsPithyPrint* pprint; SrsPithyPrint* pprint;
// whether the ingesters are expired, // Whether the ingesters are expired, for example, the listen port changed,
// for example, the listen port changed,
// all ingesters must be restart. // all ingesters must be restart.
bool expired; bool expired;
public: public:

68
trunk/src/app/srs_app_kafka.hpp
View file

@ -42,14 +42,10 @@ class SrsKafkaProducer;
#ifdef SRS_AUTO_KAFKA #ifdef SRS_AUTO_KAFKA
/** // The partition messages cache.
* the partition messages cache.
*/
typedef std::vector<SrsJsonObject*> SrsKafkaPartitionCache; typedef std::vector<SrsJsonObject*> SrsKafkaPartitionCache;
/** // The kafka partition info.
* the kafka partition info.
*/
struct SrsKafkaPartition struct SrsKafkaPartition
{ {
private: private:
@ -75,9 +71,7 @@ private:
virtual void disconnect(); virtual void disconnect();
}; };
/** // The following is all types of kafka messages.
* the following is all types of kafka messages.
*/
class SrsKafkaMessage : public ISrsAsyncCallTask class SrsKafkaMessage : public ISrsAsyncCallTask
{ {
private: private:
@ -93,19 +87,17 @@ public:
virtual std::string to_string(); virtual std::string to_string();
}; };
/** // A message cache for kafka.
* a message cache for kafka.
*/
class SrsKafkaCache class SrsKafkaCache
{ {
public: public:
// the total partitions, // The total partitions,
// for the key to map to the parition by key%nb_partitions. // for the key to map to the parition by key%nb_partitions.
int nb_partitions; int nb_partitions;
private: private:
// total messages for all partitions. // Total messages for all partitions.
int count; int count;
// key is the partition id, value is the message set to write to this partition. // The key is the partition id, value is the message set to write to this partition.
// @remark, when refresh metadata, the partition will increase, // @remark, when refresh metadata, the partition will increase,
// so maybe some message will dispatch to new partition. // so maybe some message will dispatch to new partition.
std::map< int32_t, SrsKafkaPartitionCache*> cache; std::map< int32_t, SrsKafkaPartitionCache*> cache;
@ -115,37 +107,27 @@ public:
public: public:
virtual void append(int key, SrsJsonObject* obj); virtual void append(int key, SrsJsonObject* obj);
virtual int size(); virtual int size();
/** // Fetch out a available partition cache.
* fetch out a available partition cache. // @return true when got a key and pc; otherwise, false.
* @return true when got a key and pc; otherwise, false.
*/
virtual bool fetch(int* pkey, SrsKafkaPartitionCache** ppc); virtual bool fetch(int* pkey, SrsKafkaPartitionCache** ppc);
/** // Flush the specified partition cache.
* flush the specified partition cache.
*/
virtual srs_error_t flush(SrsKafkaPartition* partition, int key, SrsKafkaPartitionCache* pc); virtual srs_error_t flush(SrsKafkaPartition* partition, int key, SrsKafkaPartitionCache* pc);
}; };
/** // The kafka cluster interface.
* the kafka cluster interface.
*/
class ISrsKafkaCluster class ISrsKafkaCluster
{ {
public: public:
ISrsKafkaCluster(); ISrsKafkaCluster();
virtual ~ISrsKafkaCluster(); virtual ~ISrsKafkaCluster();
public: public:
/** // When got any client connect to SRS, notify kafka.
* when got any client connect to SRS, notify kafka. // @param key the partition map key, the client id or hash(ip).
* @param key the partition map key, the client id or hash(ip). // @param type the type of client.
* @param type the type of client. // @param ip the peer ip of client.
* @param ip the peer ip of client.
*/
virtual srs_error_t on_client(int key, SrsListenerType type, std::string ip) = 0; virtual srs_error_t on_client(int key, SrsListenerType type, std::string ip) = 0;
/** // When client close or disconnect for error.
* when client close or disconnect for error. // @param key the partition map key, the client id or hash(ip).
* @param key the partition map key, the client id or hash(ip).
*/
virtual srs_error_t on_close(int key) = 0; virtual srs_error_t on_close(int key) = 0;
}; };
@ -155,9 +137,7 @@ extern ISrsKafkaCluster* _srs_kafka;
extern srs_error_t srs_initialize_kafka(); extern srs_error_t srs_initialize_kafka();
extern void srs_dispose_kafka(); extern void srs_dispose_kafka();
/** // The kafka producer used to save log to kafka cluster.
* the kafka producer used to save log to kafka cluster.
*/
class SrsKafkaProducer : virtual public ISrsCoroutineHandler, virtual public ISrsKafkaCluster class SrsKafkaProducer : virtual public ISrsCoroutineHandler, virtual public ISrsKafkaCluster
{ {
private: private:
@ -183,12 +163,10 @@ public:
virtual void stop(); virtual void stop();
// internal: for worker to call task to send object. // internal: for worker to call task to send object.
public: public:
/** // Send json object to kafka cluster.
* send json object to kafka cluster. // The producer will aggregate message and send in kafka message set.
* the producer will aggregate message and send in kafka message set. // @param key the key to map to the partition, user can use cid or hash.
* @param key the key to map to the partition, user can use cid or hash. // @param obj the json object; user must never free it again.
* @param obj the json object; user must never free it again.
*/
virtual srs_error_t send(int key, SrsJsonObject* obj); virtual srs_error_t send(int key, SrsJsonObject* obj);
// interface ISrsKafkaCluster // interface ISrsKafkaCluster
public: public:
@ -201,7 +179,7 @@ private:
virtual void clear_metadata(); virtual void clear_metadata();
virtual srs_error_t do_cycle(); virtual srs_error_t do_cycle();
virtual srs_error_t request_metadata(); virtual srs_error_t request_metadata();
// set the metadata to invalid and refresh it. // Set the metadata to invalid and refresh it.
virtual void refresh_metadata(); virtual void refresh_metadata();
virtual srs_error_t flush(); virtual srs_error_t flush();
}; };

195
trunk/src/app/srs_app_server.hpp
View file

@ -55,7 +55,7 @@ class SrsKafkaProducer;
#endif #endif
class SrsCoroutineManager; class SrsCoroutineManager;
// listener type for server to identify the connection, // The listener type for server to identify the connection,
// that is, use different type to process the connection. // that is, use different type to process the connection.
enum SrsListenerType enum SrsListenerType
{ {
@ -73,9 +73,7 @@ enum SrsListenerType
SrsListenerFlv = 5, SrsListenerFlv = 5,
}; };
/** // A common tcp listener, for RTMP/HTTP server.
* the common tcp listener, for RTMP/HTTP server.
*/
class SrsListener class SrsListener
{ {
protected: protected:
@ -92,9 +90,7 @@ public:
virtual srs_error_t listen(std::string i, int p) = 0; virtual srs_error_t listen(std::string i, int p) = 0;
}; };
/** // A buffered TCP listener.
* tcp listener.
*/
class SrsBufferListener : virtual public SrsListener, virtual public ISrsTcpHandler class SrsBufferListener : virtual public SrsListener, virtual public ISrsTcpHandler
{ {
private: private:
@ -104,14 +100,12 @@ public:
virtual ~SrsBufferListener(); virtual ~SrsBufferListener();
public: public:
virtual srs_error_t listen(std::string ip, int port); virtual srs_error_t listen(std::string ip, int port);
// ISrsTcpHandler // interface ISrsTcpHandler
public: public:
virtual srs_error_t on_tcp_client(srs_netfd_t stfd); virtual srs_error_t on_tcp_client(srs_netfd_t stfd);
}; };
/** // A TCP listener, for rtsp server.
* the tcp listener, for rtsp server.
*/
class SrsRtspListener : virtual public SrsListener, virtual public ISrsTcpHandler class SrsRtspListener : virtual public SrsListener, virtual public ISrsTcpHandler
{ {
private: private:
@ -122,14 +116,12 @@ public:
virtual ~SrsRtspListener(); virtual ~SrsRtspListener();
public: public:
virtual srs_error_t listen(std::string i, int p); virtual srs_error_t listen(std::string i, int p);
// ISrsTcpHandler // interface ISrsTcpHandler
public: public:
virtual srs_error_t on_tcp_client(srs_netfd_t stfd); virtual srs_error_t on_tcp_client(srs_netfd_t stfd);
}; };
/** // A TCP listener, for flv stream server.
* the tcp listener, for flv stream server.
*/
class SrsHttpFlvListener : virtual public SrsListener, virtual public ISrsTcpHandler class SrsHttpFlvListener : virtual public SrsListener, virtual public ISrsTcpHandler
{ {
private: private:
@ -140,14 +132,12 @@ public:
virtual ~SrsHttpFlvListener(); virtual ~SrsHttpFlvListener();
public: public:
virtual srs_error_t listen(std::string i, int p); virtual srs_error_t listen(std::string i, int p);
// ISrsTcpHandler // interface ISrsTcpHandler
public: public:
virtual srs_error_t on_tcp_client(srs_netfd_t stfd); virtual srs_error_t on_tcp_client(srs_netfd_t stfd);
}; };
/** // A UDP listener, for udp server.
* the udp listener, for udp server.
*/
class SrsUdpStreamListener : public SrsListener class SrsUdpStreamListener : public SrsListener
{ {
protected: protected:
@ -160,9 +150,7 @@ public:
virtual srs_error_t listen(std::string i, int p); virtual srs_error_t listen(std::string i, int p);
}; };
/** // A UDP listener, for udp stream caster server.
* the udp listener, for udp stream caster server.
*/
class SrsUdpCasterListener : public SrsUdpStreamListener class SrsUdpCasterListener : public SrsUdpStreamListener
{ {
public: public:
@ -170,15 +158,13 @@ public:
virtual ~SrsUdpCasterListener(); virtual ~SrsUdpCasterListener();
}; };
/** // Convert signal to io,
* convert signal to io, // @see: st-1.9/docs/notes.html
* @see: st-1.9/docs/notes.html
*/
class SrsSignalManager : public ISrsCoroutineHandler class SrsSignalManager : public ISrsCoroutineHandler
{ {
private: private:
/* Per-process pipe which is used as a signal queue. */ // Per-process pipe which is used as a signal queue.
/* Up to PIPE_BUF/sizeof(int) signals can be queued up. */ // Up to PIPE_BUF/sizeof(int) signals can be queued up.
int sig_pipe[2]; int sig_pipe[2];
srs_netfd_t signal_read_stfd; srs_netfd_t signal_read_stfd;
private: private:
@ -194,43 +180,30 @@ public:
public: public:
virtual srs_error_t cycle(); virtual srs_error_t cycle();
private: private:
// global singleton instance // Global singleton instance
static SrsSignalManager* instance; static SrsSignalManager* instance;
/* Signal catching function. */ // Signal catching function.
/* Converts signal event to I/O event. */ // Converts signal event to I/O event.
static void sig_catcher(int signo); static void sig_catcher(int signo);
}; };
/** // A handler to the handle cycle in SRS RTMP server.
* the handler to the handle cycle in SRS RTMP server.
*/
class ISrsServerCycle class ISrsServerCycle
{ {
public: public:
ISrsServerCycle(); ISrsServerCycle();
virtual ~ISrsServerCycle(); virtual ~ISrsServerCycle();
public: public:
/** // Initialize the cycle handler.
* initialize the cycle handler.
*/
virtual srs_error_t initialize() = 0; virtual srs_error_t initialize() = 0;
/** // Do on_cycle while server doing cycle.
* do on_cycle while server doing cycle.
*/
virtual srs_error_t on_cycle() = 0; virtual srs_error_t on_cycle() = 0;
/** // Callback the handler when got client.
* callback the handler when got client.
*/
virtual srs_error_t on_accept_client(int max, int cur) = 0; virtual srs_error_t on_accept_client(int max, int cur) = 0;
}; };
/** // SRS RTMP server, initialize and listen, start connection service thread, destroy client.
* SRS RTMP server, initialize and listen, class SrsServer : virtual public ISrsReloadHandler, virtual public ISrsSourceHandler, virtual public IConnectionManager
* start connection service thread, destroy client.
*/
class SrsServer : virtual public ISrsReloadHandler
, virtual public ISrsSourceHandler
, virtual public IConnectionManager
{ {
private: private:
// TODO: FIXME: rename to http_api // TODO: FIXME: rename to http_api
@ -240,59 +213,41 @@ private:
SrsIngester* ingester; SrsIngester* ingester;
SrsCoroutineManager* conn_manager; SrsCoroutineManager* conn_manager;
private: private:
/** // The pid file fd, lock the file write when server is running.
* the pid file fd, lock the file write when server is running. // @remark the init.d script should cleanup the pid file, when stop service,
* @remark the init.d script should cleanup the pid file, when stop service, // for the server never delete the file; when system startup, the pid in pid file
* for the server never delete the file; when system startup, the pid in pid file // maybe valid but the process is not SRS, the init.d script will never start server.
* maybe valid but the process is not SRS, the init.d script will never start server.
*/
int pid_fd; int pid_fd;
/** // All connections, connection manager
* all connections, connection manager
*/
std::vector<SrsConnection*> conns; std::vector<SrsConnection*> conns;
/** // All listners, listener manager.
* all listners, listener manager.
*/
std::vector<SrsListener*> listeners; std::vector<SrsListener*> listeners;
/** // Signal manager which convert gignal to io message.
* signal manager which convert gignal to io message.
*/
SrsSignalManager* signal_manager; SrsSignalManager* signal_manager;
/** // Handle in server cycle.
* handle in server cycle.
*/
ISrsServerCycle* handler; ISrsServerCycle* handler;
/** // User send the signal, convert to variable.
* user send the signal, convert to variable.
*/
bool signal_reload; bool signal_reload;
bool signal_persistence_config; bool signal_persistence_config;
bool signal_gmc_stop; bool signal_gmc_stop;
bool signal_gracefully_quit; bool signal_gracefully_quit;
// parent pid for asprocess. // Parent pid for asprocess.
int ppid; int ppid;
public: public:
SrsServer(); SrsServer();
virtual ~SrsServer(); virtual ~SrsServer();
private: private:
/** // The destroy is for gmc to analysis the memory leak,
* the destroy is for gmc to analysis the memory leak, // if not destroy global/static data, the gmc will warning memory leak.
* if not destroy global/static data, the gmc will warning memory leak. // In service, server never destroy, directly exit when restart.
* in service, server never destroy, directly exit when restart.
*/
virtual void destroy(); virtual void destroy();
/** // When SIGTERM, SRS should do cleanup, for example,
* when SIGTERM, SRS should do cleanup, for example, // to stop all ingesters, cleanup HLS and dvr.
* to stop all ingesters, cleanup HLS and dvr.
*/
virtual void dispose(); virtual void dispose();
// server startup workflow, @see run_master() // server startup workflow, @see run_master()
public: public:
/** // Initialize server with callback handler ch.
* initialize server with callback handler ch. // @remark user must free the handler.
* @remark user must free the handler.
*/
virtual srs_error_t initialize(ISrsServerCycle* ch); virtual srs_error_t initialize(ISrsServerCycle* ch);
virtual srs_error_t initialize_st(); virtual srs_error_t initialize_st();
virtual srs_error_t initialize_signal(); virtual srs_error_t initialize_signal();
@ -304,62 +259,48 @@ public:
virtual srs_error_t cycle(); virtual srs_error_t cycle();
// server utilities. // server utilities.
public: public:
/** // The callback for signal manager got a signal.
* callback for signal manager got a signal. // The signal manager convert signal to io message,
* the signal manager convert signal to io message, // whatever, we will got the signo like the orignal signal(int signo) handler.
* whatever, we will got the signo like the orignal signal(int signo) handler. // @param signo the signal number from user, where:
* @param signo the signal number from user, where: // SRS_SIGNAL_GRACEFULLY_QUIT, the SIGTERM, dispose then quit.
* SRS_SIGNAL_GRACEFULLY_QUIT, the SIGTERM, dispose then quit. // SRS_SIGNAL_REOPEN_LOG, the SIGUSR1, reopen the log file.
* SRS_SIGNAL_REOPEN_LOG, the SIGUSR1, reopen the log file. // SRS_SIGNAL_RELOAD, the SIGHUP, reload the config.
* SRS_SIGNAL_RELOAD, the SIGHUP, reload the config. // SRS_SIGNAL_PERSISTENCE_CONFIG, application level signal, persistence config to file.
* SRS_SIGNAL_PERSISTENCE_CONFIG, application level signal, persistence config to file. // @remark, for SIGINT:
* @remark, for SIGINT: // no gmc, directly exit.
* no gmc, directly exit. // for gmc, set the variable signal_gmc_stop, the cycle will return and cleanup for gmc.
* for gmc, set the variable signal_gmc_stop, the cycle will return and cleanup for gmc. // @remark, maybe the HTTP RAW API will trigger the on_signal() also.
* @remark, maybe the HTTP RAW API will trigger the on_signal() also.
*/
virtual void on_signal(int signo); virtual void on_signal(int signo);
private: private:
/** // The server thread main cycle,
* the server thread main cycle, // update the global static data, for instance, the current time,
* update the global static data, for instance, the current time, // the cpu/mem/network statistic.
* the cpu/mem/network statistic.
*/
virtual srs_error_t do_cycle(); virtual srs_error_t do_cycle();
/** // listen at specified protocol.
* listen at specified protocol.
*/
virtual srs_error_t listen_rtmp(); virtual srs_error_t listen_rtmp();
virtual srs_error_t listen_http_api(); virtual srs_error_t listen_http_api();
virtual srs_error_t listen_http_stream(); virtual srs_error_t listen_http_stream();
virtual srs_error_t listen_stream_caster(); virtual srs_error_t listen_stream_caster();
/** // Close the listeners for specified type,
* close the listeners for specified type, // Remove the listen object from manager.
* remove the listen object from manager.
*/
virtual void close_listeners(SrsListenerType type); virtual void close_listeners(SrsListenerType type);
/** // Resample the server kbs.
* resample the server kbs.
*/
virtual void resample_kbps(); virtual void resample_kbps();
// internal only // For internal only
public: public:
/** // When listener got a fd, notice server to accept it.
* when listener got a fd, notice server to accept it. // @param type, the client type, used to create concrete connection,
* @param type, the client type, used to create concrete connection, // for instance RTMP connection to serve client.
* for instance RTMP connection to serve client. // @param stfd, the client fd in st boxed, the underlayer fd.
* @param stfd, the client fd in st boxed, the underlayer fd.
*/
virtual srs_error_t accept_client(SrsListenerType type, srs_netfd_t stfd); virtual srs_error_t accept_client(SrsListenerType type, srs_netfd_t stfd);
private: private:
virtual srs_error_t fd2conn(SrsListenerType type, srs_netfd_t stfd, SrsConnection** pconn); virtual srs_error_t fd2conn(SrsListenerType type, srs_netfd_t stfd, SrsConnection** pconn);
// IConnectionManager // interface IConnectionManager
public: public:
/** // A callback for connection to remove itself.
* callback for connection to remove itself. // When connection thread cycle terminated, callback this to delete connection.
* when connection thread cycle terminated, callback this to delete connection. // @see SrsConnection.on_thread_stop().
* @see SrsConnection.on_thread_stop().
*/
virtual void remove(ISrsConnection* c); virtual void remove(ISrsConnection* c);
// interface ISrsReloadHandler. // interface ISrsReloadHandler.
public: public:

396
trunk/src/app/srs_app_source.hpp
View file

@ -61,12 +61,10 @@ class SrsBuffer;
class SrsHds; class SrsHds;
#endif #endif
/** // The time jitter algorithm:
* the time jitter algorithm: // 1. full, to ensure stream start at zero, and ensure stream monotonically increasing.
* 1. full, to ensure stream start at zero, and ensure stream monotonically increasing. // 2. zero, only ensure sttream start at zero, ignore timestamp jitter.
* 2. zero, only ensure sttream start at zero, ignore timestamp jitter. // 3. off, disable the time jitter algorithm, like atc.
* 3. off, disable the time jitter algorithm, like atc.
*/
enum SrsRtmpJitterAlgorithm enum SrsRtmpJitterAlgorithm
{ {
SrsRtmpJitterAlgorithmFULL = 0x01, SrsRtmpJitterAlgorithmFULL = 0x01,
@ -75,10 +73,7 @@ enum SrsRtmpJitterAlgorithm
}; };
int _srs_time_jitter_string2int(std::string time_jitter); int _srs_time_jitter_string2int(std::string time_jitter);
/** // Time jitter detect and correct, to ensure the rtmp stream is monotonically.
* time jitter detect and correct,
* to ensure the rtmp stream is monotonically.
*/
class SrsRtmpJitter class SrsRtmpJitter
{ {
private: private:
@ -88,23 +83,16 @@ public:
SrsRtmpJitter(); SrsRtmpJitter();
virtual ~SrsRtmpJitter(); virtual ~SrsRtmpJitter();
public: public:
/** // detect the time jitter and correct it.
* detect the time jitter and correct it. // @param ag the algorithm to use for time jitter.
* @param ag the algorithm to use for time jitter.
*/
virtual srs_error_t correct(SrsSharedPtrMessage* msg, SrsRtmpJitterAlgorithm ag); virtual srs_error_t correct(SrsSharedPtrMessage* msg, SrsRtmpJitterAlgorithm ag);
/** // Get current client time, the last packet time.
* get current client time, the last packet time.
*/
virtual int64_t get_time(); virtual int64_t get_time();
}; };
#ifdef SRS_PERF_QUEUE_FAST_VECTOR #ifdef SRS_PERF_QUEUE_FAST_VECTOR
/** // To alloc and increase fixed space, fast remove and insert for msgs sender.
* to alloc and increase fixed space, // @see https://github.com/ossrs/srs/issues/251
* fast remove and insert for msgs sender.
* @see https://github.com/ossrs/srs/issues/251
*/
class SrsFastVector class SrsFastVector
{ {
private: private:
@ -127,10 +115,8 @@ public:
}; };
#endif #endif
/** // The message queue for the consumer(client), forwarder.
* the message queue for the consumer(client), forwarder. // We limit the size in seconds, drop old messages(the whole gop) if full.
* we limit the size in seconds, drop old messages(the whole gop) if full.
*/
class SrsMessageQueue class SrsMessageQueue
{ {
private: private:
@ -151,85 +137,63 @@ public:
SrsMessageQueue(bool ignore_shrink = false); SrsMessageQueue(bool ignore_shrink = false);
virtual ~SrsMessageQueue(); virtual ~SrsMessageQueue();
public: public:
/** // Get the size of queue.
* get the size of queue.
*/
virtual int size(); virtual int size();
/** // Get the duration of queue.
* get the duration of queue.
*/
virtual srs_utime_t duration(); virtual srs_utime_t duration();
/** // Set the queue size
* set the queue size // @param queue_size the queue size in srs_utime_t.
* @param queue_size the queue size in srs_utime_t.
*/
virtual void set_queue_size(srs_utime_t queue_size); virtual void set_queue_size(srs_utime_t queue_size);
public: public:
/** // Enqueue the message, the timestamp always monotonically.
* enqueue the message, the timestamp always monotonically. // @param msg, the msg to enqueue, user never free it whatever the return code.
* @param msg, the msg to enqueue, user never free it whatever the return code. // @param is_overflow, whether overflow and shrinked. NULL to ignore.
* @param is_overflow, whether overflow and shrinked. NULL to ignore.
*/
virtual srs_error_t enqueue(SrsSharedPtrMessage* msg, bool* is_overflow = NULL); virtual srs_error_t enqueue(SrsSharedPtrMessage* msg, bool* is_overflow = NULL);
/** // Get packets in consumer queue.
* get packets in consumer queue. // @pmsgs SrsSharedPtrMessage*[], used to store the msgs, user must alloc it.
* @pmsgs SrsSharedPtrMessage*[], used to store the msgs, user must alloc it. // @count the count in array, output param.
* @count the count in array, output param. // @max_count the max count to dequeue, must be positive.
* @max_count the max count to dequeue, must be positive.
*/
virtual srs_error_t dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count); virtual srs_error_t dump_packets(int max_count, SrsSharedPtrMessage** pmsgs, int& count);
/** // Dumps packets to consumer, use specified args.
* dumps packets to consumer, use specified args. // @remark the atc/tba/tbv/ag are same to SrsConsumer.enqueue().
* @remark the atc/tba/tbv/ag are same to SrsConsumer.enqueue().
*/
virtual srs_error_t dump_packets(SrsConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm ag); virtual srs_error_t dump_packets(SrsConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm ag);
private: private:
/** // Remove a gop from the front.
* remove a gop from the front. // if no iframe found, clear it.
* if no iframe found, clear it.
*/
virtual void shrink(); virtual void shrink();
public: public:
/** // clear all messages in queue.
* clear all messages in queue.
*/
virtual void clear(); virtual void clear();
}; };
/** // The wakable used for some object
* the wakable used for some object // which is waiting on cond.
* which is waiting on cond.
*/
class ISrsWakable class ISrsWakable
{ {
public: public:
ISrsWakable(); ISrsWakable();
virtual ~ISrsWakable(); virtual ~ISrsWakable();
public: public:
/** // when the consumer(for player) got msg from recv thread,
* when the consumer(for player) got msg from recv thread, // it must be processed for maybe it's a close msg, so the cond
* it must be processed for maybe it's a close msg, so the cond // wait must be wakeup.
* wait must be wakeup.
*/
virtual void wakeup() = 0; virtual void wakeup() = 0;
}; };
/** // The consumer for SrsSource, that is a play client.
* the consumer for SrsSource, that is a play client.
*/
class SrsConsumer : public ISrsWakable class SrsConsumer : public ISrsWakable
{ {
private: private:
SrsRtmpJitter* jitter; SrsRtmpJitter* jitter;
SrsSource* source; SrsSource* source;
SrsMessageQueue* queue; SrsMessageQueue* queue;
// the owner connection for debug, maybe NULL. // The owner connection for debug, maybe NULL.
SrsConnection* conn; SrsConnection* conn;
bool paused; bool paused;
// when source id changed, notice all consumers // when source id changed, notice all consumers
bool should_update_source_id; bool should_update_source_id;
#ifdef SRS_PERF_QUEUE_COND_WAIT #ifdef SRS_PERF_QUEUE_COND_WAIT
// the cond wait for mw. // The cond wait for mw.
// @see https://github.com/ossrs/srs/issues/251 // @see https://github.com/ossrs/srs/issues/251
srs_cond_t mw_wait; srs_cond_t mw_wait;
bool mw_waiting; bool mw_waiting;
@ -240,159 +204,109 @@ public:
SrsConsumer(SrsSource* s, SrsConnection* c); SrsConsumer(SrsSource* s, SrsConnection* c);
virtual ~SrsConsumer(); virtual ~SrsConsumer();
public: public:
/** // Set the size of queue.
* set the size of queue.
*/
virtual void set_queue_size(srs_utime_t queue_size); virtual void set_queue_size(srs_utime_t queue_size);
/** // when source id changed, notice client to print.
* when source id changed, notice client to print.
*/
virtual void update_source_id(); virtual void update_source_id();
public: public:
/** // Get current client time, the last packet time.
* get current client time, the last packet time.
*/
virtual int64_t get_time(); virtual int64_t get_time();
/** // Enqueue an shared ptr message.
* enqueue an shared ptr message. // @param shared_msg, directly ptr, copy it if need to save it.
* @param shared_msg, directly ptr, copy it if need to save it. // @param whether atc, donot use jitter correct if true.
* @param whether atc, donot use jitter correct if true. // @param ag the algorithm of time jitter.
* @param ag the algorithm of time jitter.
*/
virtual srs_error_t enqueue(SrsSharedPtrMessage* shared_msg, bool atc, SrsRtmpJitterAlgorithm ag); virtual srs_error_t enqueue(SrsSharedPtrMessage* shared_msg, bool atc, SrsRtmpJitterAlgorithm ag);
/** // Get packets in consumer queue.
* get packets in consumer queue. // @param msgs the msgs array to dump packets to send.
* @param msgs the msgs array to dump packets to send. // @param count the count in array, intput and output param.
* @param count the count in array, intput and output param. // @remark user can specifies the count to get specified msgs; 0 to get all if possible.
* @remark user can specifies the count to get specified msgs; 0 to get all if possible.
*/
virtual srs_error_t dump_packets(SrsMessageArray* msgs, int& count); virtual srs_error_t dump_packets(SrsMessageArray* msgs, int& count);
#ifdef SRS_PERF_QUEUE_COND_WAIT #ifdef SRS_PERF_QUEUE_COND_WAIT
/** // wait for messages incomming, atleast nb_msgs and in duration.
* wait for messages incomming, atleast nb_msgs and in duration. // @param nb_msgs the messages count to wait.
* @param nb_msgs the messages count to wait. // @param msgs_duration the messages duration to wait.
* @param msgs_duration the messages duration to wait.
*/
virtual void wait(int nb_msgs, srs_utime_t msgs_duration); virtual void wait(int nb_msgs, srs_utime_t msgs_duration);
#endif #endif
/** // when client send the pause message.
* when client send the pause message.
*/
virtual srs_error_t on_play_client_pause(bool is_pause); virtual srs_error_t on_play_client_pause(bool is_pause);
// ISrsWakable // ISrsWakable
public: public:
/** // when the consumer(for player) got msg from recv thread,
* when the consumer(for player) got msg from recv thread, // it must be processed for maybe it's a close msg, so the cond
* it must be processed for maybe it's a close msg, so the cond // wait must be wakeup.
* wait must be wakeup.
*/
virtual void wakeup(); virtual void wakeup();
}; };
/** // cache a gop of video/audio data,
* cache a gop of video/audio data, // delivery at the connect of flash player,
* delivery at the connect of flash player, // To enable it to fast startup.
* to enable it to fast startup.
*/
class SrsGopCache class SrsGopCache
{ {
private: private:
/** // if disabled the gop cache,
* if disabled the gop cache, // The client will wait for the next keyframe for h264,
* the client will wait for the next keyframe for h264, // and will be black-screen.
* and will be black-screen.
*/
bool enable_gop_cache; bool enable_gop_cache;
/** // The video frame count, avoid cache for pure audio stream.
* the video frame count, avoid cache for pure audio stream.
*/
int cached_video_count; int cached_video_count;
/** // when user disabled video when publishing, and gop cache enalbed,
* when user disabled video when publishing, and gop cache enalbed, // We will cache the audio/video for we already got video, but we never
* we will cache the audio/video for we already got video, but we never // know when to clear the gop cache, for there is no video in future,
* know when to clear the gop cache, for there is no video in future, // so we must guess whether user disabled the video.
* so we must guess whether user disabled the video. // when we got some audios after laster video, for instance, 600 audio packets,
* when we got some audios after laster video, for instance, 600 audio packets, // about 3s(26ms per packet) 115 audio packets, clear gop cache.
* about 3s(26ms per packet) 115 audio packets, clear gop cache. //
* // @remark, it is ok for performance, for when we clear the gop cache,
* @remark, it is ok for performance, for when we clear the gop cache, // gop cache is disabled for pure audio stream.
* gop cache is disabled for pure audio stream. // @see: https://github.com/ossrs/srs/issues/124
* @see: https://github.com/ossrs/srs/issues/124
*/
int audio_after_last_video_count; int audio_after_last_video_count;
/** // cached gop.
* cached gop.
*/
std::vector<SrsSharedPtrMessage*> gop_cache; std::vector<SrsSharedPtrMessage*> gop_cache;
public: public:
SrsGopCache(); SrsGopCache();
virtual ~SrsGopCache(); virtual ~SrsGopCache();
public: public:
/** // cleanup when system quit.
* cleanup when system quit.
*/
virtual void dispose(); virtual void dispose();
/** // To enable or disable the gop cache.
* to enable or disable the gop cache.
*/
virtual void set(bool v); virtual void set(bool v);
virtual bool enabled(); virtual bool enabled();
/** // only for h264 codec
* only for h264 codec // 1. cache the gop when got h264 video packet.
* 1. cache the gop when got h264 video packet. // 2. clear gop when got keyframe.
* 2. clear gop when got keyframe. // @param shared_msg, directly ptr, copy it if need to save it.
* @param shared_msg, directly ptr, copy it if need to save it.
*/
virtual srs_error_t cache(SrsSharedPtrMessage* shared_msg); virtual srs_error_t cache(SrsSharedPtrMessage* shared_msg);
/** // clear the gop cache.
* clear the gop cache.
*/
virtual void clear(); virtual void clear();
/** // dump the cached gop to consumer.
* dump the cached gop to consumer.
*/
virtual srs_error_t dump(SrsConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm jitter_algorithm); virtual srs_error_t dump(SrsConsumer* consumer, bool atc, SrsRtmpJitterAlgorithm jitter_algorithm);
/** // used for atc to get the time of gop cache,
* used for atc to get the time of gop cache, // The atc will adjust the sequence header timestamp to gop cache.
* the atc will adjust the sequence header timestamp to gop cache.
*/
virtual bool empty(); virtual bool empty();
/** // Get the start time of gop cache, in srs_utime_t.
* get the start time of gop cache, in srs_utime_t. // @return 0 if no packets.
* @return 0 if no packets.
*/
virtual srs_utime_t start_time(); virtual srs_utime_t start_time();
/** // whether current stream is pure audio,
* whether current stream is pure audio, // when no video in gop cache, the stream is pure audio right now.
* when no video in gop cache, the stream is pure audio right now.
*/
virtual bool pure_audio(); virtual bool pure_audio();
}; };
/** // The handler to handle the event of srs source.
* the handler to handle the event of srs source. // For example, the http flv streaming module handle the event and
* for example, the http flv streaming module handle the event and // mount http when rtmp start publishing.
* mount http when rtmp start publishing.
*/
class ISrsSourceHandler class ISrsSourceHandler
{ {
public: public:
ISrsSourceHandler(); ISrsSourceHandler();
virtual ~ISrsSourceHandler(); virtual ~ISrsSourceHandler();
public: public:
/** // when stream start publish, mount stream.
* when stream start publish, mount stream.
*/
virtual srs_error_t on_publish(SrsSource* s, SrsRequest* r) = 0; virtual srs_error_t on_publish(SrsSource* s, SrsRequest* r) = 0;
/** // when stream stop publish, unmount stream.
* when stream stop publish, unmount stream.
*/
virtual void on_unpublish(SrsSource* s, SrsRequest* r) = 0; virtual void on_unpublish(SrsSource* s, SrsRequest* r) = 0;
}; };
/** // The mix queue to correct the timestamp for mix_correct algorithm.
* the mix queue to correct the timestamp for mix_correct algorithm.
*/
class SrsMixQueue class SrsMixQueue
{ {
private: private:
@ -408,11 +322,9 @@ public:
virtual SrsSharedPtrMessage* pop(); virtual SrsSharedPtrMessage* pop();
}; };
/** // The hub for origin is a collection of utilities for origin only,
* The hub for origin is a collection of utilities for origin only, // For example, DVR, HLS, Forward and Transcode are only available for origin,
* for example, DVR, HLS, Forward and Transcode are only available for origin, // they are meanless for edge server.
* they are meanless for edge server.
*/
class SrsOriginHub : public ISrsReloadHandler class SrsOriginHub : public ISrsReloadHandler
{ {
private: private:
@ -437,7 +349,7 @@ private:
#endif #endif
// nginx-rtmp exec feature. // nginx-rtmp exec feature.
SrsNgExec* ng_exec; SrsNgExec* ng_exec;
// to forward stream to other servers // To forward stream to other servers
std::vector<SrsForwarder*> forwarders; std::vector<SrsForwarder*> forwarders;
public: public:
SrsOriginHub(); SrsOriginHub();
@ -447,10 +359,10 @@ public:
// @param r The request object, managed by source. // @param r The request object, managed by source.
virtual srs_error_t initialize(SrsSource* s, SrsRequest* r); virtual srs_error_t initialize(SrsSource* s, SrsRequest* r);
// Dispose the hub, release utilities resource, // Dispose the hub, release utilities resource,
// for example, delete all HLS pieces. // For example, delete all HLS pieces.
virtual void dispose(); virtual void dispose();
// Cycle the hub, process some regular events, // Cycle the hub, process some regular events,
// for example, dispose hls in cycle. // For example, dispose hls in cycle.
virtual srs_error_t cycle(); virtual srs_error_t cycle();
public: public:
// When got a parsed metadata. // When got a parsed metadata.
@ -466,9 +378,9 @@ public:
virtual void on_unpublish(); virtual void on_unpublish();
// Internal callback. // Internal callback.
public: public:
// for the SrsForwarder to callback to request the sequence headers. // For the SrsForwarder to callback to request the sequence headers.
virtual srs_error_t on_forwarder_start(SrsForwarder* forwarder); virtual srs_error_t on_forwarder_start(SrsForwarder* forwarder);
// for the SrsDvr to callback to request the sequence headers. // For the SrsDvr to callback to request the sequence headers.
virtual srs_error_t on_dvr_request_sh(); virtual srs_error_t on_dvr_request_sh();
// interface ISrsReloadHandler // interface ISrsReloadHandler
public: public:
@ -484,10 +396,8 @@ private:
virtual void destroy_forwarders(); virtual void destroy_forwarders();
}; };
/** // Each stream have optional meta(sps/pps in sequence header and metadata).
* Each stream have optional meta(sps/pps in sequence header and metadata). // This class cache and update the meta.
* This class cache and update the meta.
*/
class SrsMetaCache class SrsMetaCache
{ {
private: private:
@ -528,46 +438,35 @@ public:
virtual srs_error_t update_vsh(SrsSharedPtrMessage* msg); virtual srs_error_t update_vsh(SrsSharedPtrMessage* msg);
}; };
/** // live streaming source.
* live streaming source.
*/
class SrsSource : public ISrsReloadHandler class SrsSource : public ISrsReloadHandler
{ {
friend class SrsOriginHub; friend class SrsOriginHub;
private: private:
static std::map<std::string, SrsSource*> pool; static std::map<std::string, SrsSource*> pool;
public: public:
/** // create source when fetch from cache failed.
* create source when fetch from cache failed. // @param r the client request.
* @param r the client request. // @param h the event handler for source.
* @param h the event handler for source. // @param pps the matched source, if success never be NULL.
* @param pps the matched source, if success never be NULL.
*/
static srs_error_t fetch_or_create(SrsRequest* r, ISrsSourceHandler* h, SrsSource** pps); static srs_error_t fetch_or_create(SrsRequest* r, ISrsSourceHandler* h, SrsSource** pps);
private: private:
/** // Get the exists source, NULL when not exists.
* get the exists source, NULL when not exists. // update the request and return the exists source.
* update the request and return the exists source.
*/
static SrsSource* fetch(SrsRequest* r); static SrsSource* fetch(SrsRequest* r);
public: public:
/** // dispose and cycle all sources.
* dispose and cycle all sources.
*/
static void dispose_all(); static void dispose_all();
static srs_error_t cycle_all(); static srs_error_t cycle_all();
private: private:
static srs_error_t do_cycle_all(); static srs_error_t do_cycle_all();
public: public:
/** // when system exit, destroy the sources,
* when system exit, destroy the sources, // For gmc to analysis mem leaks.
* for gmc to analysis mem leaks.
*/
static void destroy(); static void destroy();
private: private:
// source id, // For publish, it's the publish client id.
// for publish, it's the publish client id. // For edge, it's the edge ingest id.
// for edge, it's the edge ingest id.
// when source id changed, for example, the edge reconnect, // when source id changed, for example, the edge reconnect,
// invoke the on_source_id_changed() to let all clients know. // invoke the on_source_id_changed() to let all clients know.
int _source_id; int _source_id;
@ -575,31 +474,29 @@ private:
int _pre_source_id; int _pre_source_id;
// deep copy of client request. // deep copy of client request.
SrsRequest* req; SrsRequest* req;
// to delivery stream to clients. // To delivery stream to clients.
std::vector<SrsConsumer*> consumers; std::vector<SrsConsumer*> consumers;
// the time jitter algorithm for vhost. // The time jitter algorithm for vhost.
SrsRtmpJitterAlgorithm jitter_algorithm; SrsRtmpJitterAlgorithm jitter_algorithm;
// for play, whether use interlaced/mixed algorithm to correct timestamp. // For play, whether use interlaced/mixed algorithm to correct timestamp.
bool mix_correct; bool mix_correct;
// the mix queue to implements the mix correct algorithm. // The mix queue to implements the mix correct algorithm.
SrsMixQueue* mix_queue; SrsMixQueue* mix_queue;
/** // For play, whether enabled atc.
* for play, whether enabled atc. // The atc(use absolute time and donot adjust time),
* atc whether atc(use absolute time and donot adjust time), // directly use msg time and donot adjust if atc is true,
* directly use msg time and donot adjust if atc is true, // otherwise, adjust msg time to start from 0 to make flash happy.
* otherwise, adjust msg time to start from 0 to make flash happy.
*/
bool atc; bool atc;
// whether stream is monotonically increase. // whether stream is monotonically increase.
bool is_monotonically_increase; bool is_monotonically_increase;
// the time of the packet we just got. // The time of the packet we just got.
int64_t last_packet_time; int64_t last_packet_time;
// the event handler. // The event handler.
ISrsSourceHandler* handler; ISrsSourceHandler* handler;
// edge control service // The edge control service
SrsPlayEdge* play_edge; SrsPlayEdge* play_edge;
SrsPublishEdge* publish_edge; SrsPublishEdge* publish_edge;
// gop cache for client fast startup. // The gop cache for client fast startup.
SrsGopCache* gop_cache; SrsGopCache* gop_cache;
// The hub for origin server. // The hub for origin server.
SrsOriginHub* hub; SrsOriginHub* hub;
@ -609,7 +506,7 @@ private:
// Whether source is avaiable for publishing. // Whether source is avaiable for publishing.
bool _can_publish; bool _can_publish;
// The last die time, when all consumers quit and no publisher, // The last die time, when all consumers quit and no publisher,
// we will remove the source when source die. // We will remove the source when source die.
srs_utime_t die_at; srs_utime_t die_at;
public: public:
SrsSource(); SrsSource();
@ -617,26 +514,23 @@ public:
public: public:
virtual void dispose(); virtual void dispose();
virtual srs_error_t cycle(); virtual srs_error_t cycle();
// remove source when expired. // Remove source when expired.
virtual bool expired(); virtual bool expired();
// initialize, get and setter.
public: public:
// initialize the hls with handlers. // Initialize the hls with handlers.
virtual srs_error_t initialize(SrsRequest* r, ISrsSourceHandler* h); virtual srs_error_t initialize(SrsRequest* r, ISrsSourceHandler* h);
// interface ISrsReloadHandler // interface ISrsReloadHandler
public: public:
virtual srs_error_t on_reload_vhost_play(std::string vhost); virtual srs_error_t on_reload_vhost_play(std::string vhost);
// for the tools callback
public: public:
// source id changed. // The source id changed.
virtual srs_error_t on_source_id_changed(int id); virtual srs_error_t on_source_id_changed(int id);
// get current source id. // Get current source id.
virtual int source_id(); virtual int source_id();
virtual int pre_source_id(); virtual int pre_source_id();
// Whether source is inactive, which means there is no publishing stream source. // Whether source is inactive, which means there is no publishing stream source.
// @remark For edge, it's inactive util stream has been pulled from origin. // @remark For edge, it's inactive util stream has been pulled from origin.
virtual bool inactive(); virtual bool inactive();
// logic data methods
public: public:
virtual bool can_publish(bool is_edge); virtual bool can_publish(bool is_edge);
virtual srs_error_t on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata); virtual srs_error_t on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
@ -650,33 +544,27 @@ private:
virtual srs_error_t on_video_imp(SrsSharedPtrMessage* video); virtual srs_error_t on_video_imp(SrsSharedPtrMessage* video);
public: public:
virtual srs_error_t on_aggregate(SrsCommonMessage* msg); virtual srs_error_t on_aggregate(SrsCommonMessage* msg);
/** // Publish stream event notify.
* publish stream event notify. // @param _req the request from client, the source will deep copy it,
* @param _req the request from client, the source will deep copy it, // for when reload the request of client maybe invalid.
* for when reload the request of client maybe invalid.
*/
virtual srs_error_t on_publish(); virtual srs_error_t on_publish();
virtual void on_unpublish(); virtual void on_unpublish();
// consumer methods
public: public:
/** // Create consumer and dumps packets in cache.
* create consumer and dumps packets in cache. // @param consumer, output the create consumer.
* @param consumer, output the create consumer. // @param ds, whether dumps the sequence header.
* @param ds, whether dumps the sequence header. // @param dm, whether dumps the metadata.
* @param dm, whether dumps the metadata. // @param dg, whether dumps the gop cache.
* @param dg, whether dumps the gop cache.
*/
virtual srs_error_t create_consumer(SrsConnection* conn, SrsConsumer*& consumer, bool ds = true, bool dm = true, bool dg = true); virtual srs_error_t create_consumer(SrsConnection* conn, SrsConsumer*& consumer, bool ds = true, bool dm = true, bool dg = true);
virtual void on_consumer_destroy(SrsConsumer* consumer); virtual void on_consumer_destroy(SrsConsumer* consumer);
virtual void set_cache(bool enabled); virtual void set_cache(bool enabled);
virtual SrsRtmpJitterAlgorithm jitter(); virtual SrsRtmpJitterAlgorithm jitter();
// internal
public: public:
// for edge, when publish edge stream, check the state // For edge, when publish edge stream, check the state
virtual srs_error_t on_edge_start_publish(); virtual srs_error_t on_edge_start_publish();
// for edge, proxy the publish // For edge, proxy the publish
virtual srs_error_t on_edge_proxy_publish(SrsCommonMessage* msg); virtual srs_error_t on_edge_proxy_publish(SrsCommonMessage* msg);
// for edge, proxy stop publish // For edge, proxy stop publish
virtual void on_edge_proxy_unpublish(); virtual void on_edge_proxy_unpublish();
public: public:
virtual std::string get_curr_origin(); virtual std::string get_curr_origin();

128
trunk/src/app/srs_app_statistic.hpp
View file

@ -48,9 +48,7 @@ public:
int nb_streams; int nb_streams;
int nb_clients; int nb_clients;
public: public:
/** // The vhost total kbps.
* vhost total kbps.
*/
SrsKbps* kbps; SrsKbps* kbps;
SrsWallClock* clk; SrsWallClock* clk;
public: public:
@ -73,19 +71,17 @@ public:
int nb_clients; int nb_clients;
uint64_t nb_frames; uint64_t nb_frames;
public: public:
/** // The stream total kbps.
* stream total kbps.
*/
SrsKbps* kbps; SrsKbps* kbps;
SrsWallClock* clk; SrsWallClock* clk;
public: public:
bool has_video; bool has_video;
SrsVideoCodecId vcodec; SrsVideoCodecId vcodec;
// profile_idc, ISO_IEC_14496-10-AVC-2003.pdf, page 45. // The profile_idc, ISO_IEC_14496-10-AVC-2003.pdf, page 45.
SrsAvcProfile avc_profile; SrsAvcProfile avc_profile;
// level_idc, ISO_IEC_14496-10-AVC-2003.pdf, page 45. // The level_idc, ISO_IEC_14496-10-AVC-2003.pdf, page 45.
SrsAvcLevel avc_level; SrsAvcLevel avc_level;
// the width and height in codec info. // The width and height in codec info.
int width; int width;
int height; int height;
public: public:
@ -93,12 +89,10 @@ public:
SrsAudioCodecId acodec; SrsAudioCodecId acodec;
SrsAudioSampleRate asample_rate; SrsAudioSampleRate asample_rate;
SrsAudioChannels asound_type; SrsAudioChannels asound_type;
/** // The audio specified
* audio specified // audioObjectType, in 1.6.2.1 AudioSpecificConfig, page 33,
* audioObjectType, in 1.6.2.1 AudioSpecificConfig, page 33, // 1.5.1.1 Audio object type definition, page 23,
* 1.5.1.1 Audio object type definition, page 23, // in ISO_IEC_14496-3-AAC-2001.pdf.
* in ISO_IEC_14496-3-AAC-2001.pdf.
*/
SrsAacObjectType aac_object; SrsAacObjectType aac_object;
public: public:
SrsStatisticStream(); SrsStatisticStream();
@ -106,13 +100,9 @@ public:
public: public:
virtual srs_error_t dumps(SrsJsonObject* obj); virtual srs_error_t dumps(SrsJsonObject* obj);
public: public:
/** // Publish the stream.
* publish the stream.
*/
virtual void publish(int cid); virtual void publish(int cid);
/** // Close the stream.
* close the stream.
*/
virtual void close(); virtual void close();
}; };
@ -136,24 +126,24 @@ class SrsStatistic
{ {
private: private:
static SrsStatistic *_instance; static SrsStatistic *_instance;
// the id to identify the sever. // The id to identify the sever.
int64_t _server_id; int64_t _server_id;
private: private:
// key: vhost id, value: vhost object. // The key: vhost id, value: vhost object.
std::map<int64_t, SrsStatisticVhost*> vhosts; std::map<int64_t, SrsStatisticVhost*> vhosts;
// key: vhost url, value: vhost Object. // The key: vhost url, value: vhost Object.
// @remark a fast index for vhosts. // @remark a fast index for vhosts.
std::map<std::string, SrsStatisticVhost*> rvhosts; std::map<std::string, SrsStatisticVhost*> rvhosts;
private: private:
// key: stream id, value: stream Object. // The key: stream id, value: stream Object.
std::map<int64_t, SrsStatisticStream*> streams; std::map<int64_t, SrsStatisticStream*> streams;
// key: stream url, value: stream Object. // The key: stream url, value: stream Object.
// @remark a fast index for streams. // @remark a fast index for streams.
std::map<std::string, SrsStatisticStream*> rstreams; std::map<std::string, SrsStatisticStream*> rstreams;
private: private:
// key: client id, value: stream object. // The key: client id, value: stream object.
std::map<int, SrsStatisticClient*> clients; std::map<int, SrsStatisticClient*> clients;
// server total kbps. // The server total kbps.
SrsKbps* kbps; SrsKbps* kbps;
SrsWallClock* clk; SrsWallClock* clk;
private: private:
@ -167,77 +157,51 @@ public:
virtual SrsStatisticStream* find_stream(int sid); virtual SrsStatisticStream* find_stream(int sid);
virtual SrsStatisticClient* find_client(int cid); virtual SrsStatisticClient* find_client(int cid);
public: public:
/** // When got video info for stream.
* when got video info for stream.
*/
virtual srs_error_t on_video_info(SrsRequest* req, SrsVideoCodecId vcodec, SrsAvcProfile avc_profile, virtual srs_error_t on_video_info(SrsRequest* req, SrsVideoCodecId vcodec, SrsAvcProfile avc_profile,
SrsAvcLevel avc_level, int width, int height); SrsAvcLevel avc_level, int width, int height);
/** // When got audio info for stream.
* when got audio info for stream.
*/
virtual srs_error_t on_audio_info(SrsRequest* req, SrsAudioCodecId acodec, SrsAudioSampleRate asample_rate, virtual srs_error_t on_audio_info(SrsRequest* req, SrsAudioCodecId acodec, SrsAudioSampleRate asample_rate,
SrsAudioChannels asound_type, SrsAacObjectType aac_object); SrsAudioChannels asound_type, SrsAacObjectType aac_object);
/** // When got videos, update the frames.
* When got videos, update the frames. // We only stat the total number of video frames.
* We only stat the total number of video frames.
*/
virtual srs_error_t on_video_frames(SrsRequest* req, int nb_frames); virtual srs_error_t on_video_frames(SrsRequest* req, int nb_frames);
/** // When publish stream.
* when publish stream. // @param req the request object of publish connection.
* @param req the request object of publish connection. // @param cid the cid of publish connection.
* @param cid the cid of publish connection.
*/
virtual void on_stream_publish(SrsRequest* req, int cid); virtual void on_stream_publish(SrsRequest* req, int cid);
/** // When close stream.
* when close stream.
*/
virtual void on_stream_close(SrsRequest* req); virtual void on_stream_close(SrsRequest* req);
public: public:
/** // When got a client to publish/play stream,
* when got a client to publish/play stream, // @param id, the client srs id.
* @param id, the client srs id. // @param req, the client request object.
* @param req, the client request object. // @param conn, the physical absract connection object.
* @param conn, the physical absract connection object. // @param type, the type of connection.
* @param type, the type of connection.
*/
virtual srs_error_t on_client(int id, SrsRequest* req, SrsConnection* conn, SrsRtmpConnType type); virtual srs_error_t on_client(int id, SrsRequest* req, SrsConnection* conn, SrsRtmpConnType type);
/** // Client disconnect
* client disconnect // @remark the on_disconnect always call, while the on_client is call when
* @remark the on_disconnect always call, while the on_client is call when // only got the request object, so the client specified by id maybe not
* only got the request object, so the client specified by id maybe not // exists in stat.
* exists in stat.
*/
virtual void on_disconnect(int id); virtual void on_disconnect(int id);
/** // Sample the kbps, add delta bytes of conn.
* sample the kbps, add delta bytes of conn. // Use kbps_sample() to get all result of kbps stat.
* use kbps_sample() to get all result of kbps stat.
*/
// TODO: FIXME: the add delta must use ISrsKbpsDelta interface instead. // TODO: FIXME: the add delta must use ISrsKbpsDelta interface instead.
virtual void kbps_add_delta(SrsConnection* conn); virtual void kbps_add_delta(SrsConnection* conn);
/** // Calc the result for all kbps.
* calc the result for all kbps. // @return the server kbps.
* @return the server kbps.
*/
virtual SrsKbps* kbps_sample(); virtual SrsKbps* kbps_sample();
public: public:
/** // Get the server id, used to identify the server.
* get the server id, used to identify the server. // For example, when restart, the server id must changed.
* for example, when restart, the server id must changed.
*/
virtual int64_t server_id(); virtual int64_t server_id();
/** // Dumps the vhosts to amf0 array.
* dumps the vhosts to amf0 array.
*/
virtual srs_error_t dumps_vhosts(SrsJsonArray* arr); virtual srs_error_t dumps_vhosts(SrsJsonArray* arr);
/** // Dumps the streams to amf0 array.
* dumps the streams to amf0 array.
*/
virtual srs_error_t dumps_streams(SrsJsonArray* arr); virtual srs_error_t dumps_streams(SrsJsonArray* arr);
/** // Dumps the clients to amf0 array
* dumps the clients to amf0 array // @param start the start index, from 0.
* @param start the start index, from 0. // @param count the max count of clients to dump.
* @param count the max count of clients to dump.
*/
virtual srs_error_t dumps_clients(SrsJsonArray* arr, int start, int count); virtual srs_error_t dumps_clients(SrsJsonArray* arr, int start, int count);
private: private:
virtual SrsStatisticVhost* create_vhost(SrsRequest* req); virtual SrsStatisticVhost* create_vhost(SrsRequest* req);

14
trunk/src/app/srs_app_thread.hpp
View file

@ -31,12 +31,10 @@
#include <srs_app_st.hpp> #include <srs_app_st.hpp>
#include <srs_service_conn.hpp> #include <srs_service_conn.hpp>
/** // The coroutine manager use a thread to delete a connection, which will stop the service
* The coroutine manager use a thread to delete a connection, which will stop the service // thread, for example, when the RTMP connection thread cycle terminated, it will notify
* thread, for example, when the RTMP connection thread cycle terminated, it will notify // the manager(the server) to remove the connection from list of server and push it to
* the manager(the server) to remove the connection from list of server and push it to // the manager thread to delete it, finally the thread of connection will stop.
* the manager thread to delete it, finally the thread of connection will stop.
*/
class SrsCoroutineManager : virtual public ISrsCoroutineHandler, virtual public IConnectionManager class SrsCoroutineManager : virtual public ISrsCoroutineHandler, virtual public IConnectionManager
{ {
private: private:
@ -48,10 +46,10 @@ public:
virtual ~SrsCoroutineManager(); virtual ~SrsCoroutineManager();
public: public:
srs_error_t start(); srs_error_t start();
// ISrsCoroutineHandler // interface ISrsCoroutineHandler
public: public:
virtual srs_error_t cycle(); virtual srs_error_t cycle();
// IConnectionManager // interface IConnectionManager
public: public:
virtual void remove(ISrsConnection* c); virtual void remove(ISrsConnection* c);
private: private:

223
trunk/src/app/srs_app_utility.hpp
View file

@ -41,51 +41,43 @@ class SrsKbps;
class SrsBuffer; class SrsBuffer;
class SrsJsonObject; class SrsJsonObject;
/** // Convert level in string to log level in int.
* convert level in string to log level in int. // @return the log level defined in SrsLogLevel.
* @return the log level defined in SrsLogLevel.
*/
extern SrsLogLevel srs_get_log_level(std::string level); extern SrsLogLevel srs_get_log_level(std::string level);
/** // Build the path according to vhost/app/stream, where replace variables:
* build the path according to vhost/app/stream, where replace variables: // [vhost], the vhost of stream.
* [vhost], the vhost of stream. // [app], the app of stream.
* [app], the app of stream. // [stream], the stream name of stream.
* [stream], the stream name of stream. // @return the replaced path.
* @return the replaced path.
*/
extern std::string srs_path_build_stream(std::string template_path, std::string vhost, std::string app, std::string stream); extern std::string srs_path_build_stream(std::string template_path, std::string vhost, std::string app, std::string stream);
/** // Build the path according to timestamp, where replace variables:
* build the path according to timestamp, where replace variables: // [2006], replace this const to current year.
* [2006], replace this const to current year. // [01], replace this const to current month.
* [01], replace this const to current month. // [02], replace this const to current date.
* [02], replace this const to current date. // [15], replace this const to current hour.
* [15], replace this const to current hour. // [04], repleace this const to current minute.
* [04], repleace this const to current minute. // [05], repleace this const to current second.
* [05], repleace this const to current second. // [999], repleace this const to current millisecond.
* [999], repleace this const to current millisecond. // [timestamp],replace this const to current UNIX timestamp in ms.
* [timestamp],replace this const to current UNIX timestamp in ms. // @return the replaced path.
* @return the replaced path.
*/
extern std::string srs_path_build_timestamp(std::string template_path); extern std::string srs_path_build_timestamp(std::string template_path);
/** // Kill the pid by SIGINT, then wait to quit,
* kill the pid by SIGINT, then wait to quit, // Kill the pid by SIGKILL again when exceed the timeout.
* kill the pid by SIGKILL again when exceed the timeout. // @param pid the pid to kill. ignore for -1. set to -1 when killed.
* @param pid the pid to kill. ignore for -1. set to -1 when killed. // @return an int error code.
* @return an int error code.
*/
extern srs_error_t srs_kill_forced(int& pid); extern srs_error_t srs_kill_forced(int& pid);
// current process resouce usage. // Current process resouce usage.
// @see: man getrusage // @see: man getrusage
class SrsRusage class SrsRusage
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// the time in ms when sample. // The time in ms when sample.
int64_t sample_time; int64_t sample_time;
public: public:
@ -95,21 +87,21 @@ public:
SrsRusage(); SrsRusage();
}; };
// get system rusage, use cache to avoid performance problem. // Get system rusage, use cache to avoid performance problem.
extern SrsRusage* srs_get_system_rusage(); extern SrsRusage* srs_get_system_rusage();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_system_rusage(); extern void srs_update_system_rusage();
// to stat the process info. // To stat the process info.
// @see: man 5 proc, /proc/[pid]/stat // @see: man 5 proc, /proc/[pid]/stat
class SrsProcSelfStat class SrsProcSelfStat
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// the time in ms when sample. // The time in ms when sample.
int64_t sample_time; int64_t sample_time;
// the percent of usage. 0.153 is 15.3%. // The percent of usage. 0.153 is 15.3%.
float percent; float percent;
// data of /proc/[pid]/stat // data of /proc/[pid]/stat
@ -256,49 +248,47 @@ public:
SrsProcSelfStat(); SrsProcSelfStat();
}; };
// to stat the cpu time. // To stat the cpu time.
// @see: man 5 proc, /proc/stat // @see: man 5 proc, /proc/stat
/** // about the cpu time, @see: http://stackoverflow.com/questions/16011677/calculating-cpu-usage-using-proc-files
* about the cpu time, @see: http://stackoverflow.com/questions/16011677/calculating-cpu-usage-using-proc-files // for example, for ossrs.net, a single cpu machine:
* for example, for ossrs.net, a single cpu machine: // [winlin@SRS ~]$ cat /proc/uptime && cat /proc/stat
* [winlin@SRS ~]$ cat /proc/uptime && cat /proc/stat // 5275153.01 4699624.99
* 5275153.01 4699624.99 // cpu 43506750 973 8545744 466133337 4149365 190852 804666 0 0
* cpu 43506750 973 8545744 466133337 4149365 190852 804666 0 0 // Where the uptime is 5275153.01s
* where the uptime is 5275153.01s // generally, USER_HZ sysconf(_SC_CLK_TCK)=100, which means the unit of /proc/stat is "1/100ths seconds"
* generally, USER_HZ sysconf(_SC_CLK_TCK)=100, which means the unit of /proc/stat is "1/100ths seconds" // that is, USER_HZ=1/100 seconds
* that is, USER_HZ=1/100 seconds // cpu total = 43506750+973+8545744+466133337+4149365+190852+804666+0+0 (USER_HZ)
* cpu total = 43506750+973+8545744+466133337+4149365+190852+804666+0+0 (USER_HZ) // = 523331687 (USER_HZ)
* = 523331687 (USER_HZ) // = 523331687 * 1/100 (seconds)
* = 523331687 * 1/100 (seconds) // = 5233316.87 seconds
* = 5233316.87 seconds // The cpu total seconds almost the uptime, the delta is more precise.
* the cpu total seconds almost the uptime, the delta is more precise. //
* // we run the command about 26minutes:
* we run the command about 26minutes: // [winlin@SRS ~]$ cat /proc/uptime && cat /proc/stat
* [winlin@SRS ~]$ cat /proc/uptime && cat /proc/stat // 5276739.83 4701090.76
* 5276739.83 4701090.76 // cpu 43514105 973 8548948 466278556 4150480 190899 804937 0 0
* cpu 43514105 973 8548948 466278556 4150480 190899 804937 0 0 // Where the uptime is 5276739.83s
* where the uptime is 5276739.83s // cpu total = 43514105+973+8548948+466278556+4150480+190899+804937+0+0 (USER_HZ)
* cpu total = 43514105+973+8548948+466278556+4150480+190899+804937+0+0 (USER_HZ) // = 523488898 (USER_HZ)
* = 523488898 (USER_HZ) // = 523488898 * 1/100 (seconds)
* = 523488898 * 1/100 (seconds) // = 5234888.98 seconds
* = 5234888.98 seconds // where:
* where: // uptime delta = 1586.82s
* uptime delta = 1586.82s // cpu total delta = 1572.11s
* cpu total delta = 1572.11s // The deviation is more smaller.
* the deviation is more smaller.
*/
class SrsProcSystemStat class SrsProcSystemStat
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// the time in ms when sample. // The time in ms when sample.
int64_t sample_time; int64_t sample_time;
// the percent of usage. 0.153 is 15.3%. // The percent of usage. 0.153 is 15.3%.
// the percent is in [0, 1], where 1 is 100%. // The percent is in [0, 1], where 1 is 100%.
// for multiple core cpu, max also is 100%. // for multiple core cpu, max also is 100%.
float percent; float percent;
// the total cpu time units // The total cpu time units
// @remark, zero for the previous total() is zero. // @remark, zero for the previous total() is zero.
// the usaged_cpu_delta = total_delta * percent // the usaged_cpu_delta = total_delta * percent
// previous cpu total = this->total() - total_delta // previous cpu total = this->total() - total_delta
@ -310,7 +300,7 @@ public:
// (1/100ths of a second on most architectures, use // (1/100ths of a second on most architectures, use
// sysconf(_SC_CLK_TCK) to obtain the right value) // sysconf(_SC_CLK_TCK) to obtain the right value)
// //
// the system spent in user mode, // The system spent in user mode,
unsigned long long user; unsigned long long user;
// user mode with low priority (nice), // user mode with low priority (nice),
unsigned long long nice; unsigned long long nice;
@ -341,18 +331,18 @@ public:
public: public:
SrsProcSystemStat(); SrsProcSystemStat();
// get total cpu units. // Get total cpu units.
int64_t total(); int64_t total();
}; };
// get system cpu stat, use cache to avoid performance problem. // Get system cpu stat, use cache to avoid performance problem.
extern SrsProcSelfStat* srs_get_self_proc_stat(); extern SrsProcSelfStat* srs_get_self_proc_stat();
// get system cpu stat, use cache to avoid performance problem. // Get system cpu stat, use cache to avoid performance problem.
extern SrsProcSystemStat* srs_get_system_proc_stat(); extern SrsProcSystemStat* srs_get_system_proc_stat();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_proc_stat(); extern void srs_update_proc_stat();
// stat disk iops // Stat disk iops
// @see: http://stackoverflow.com/questions/4458183/how-the-util-of-iostat-is-computed // @see: http://stackoverflow.com/questions/4458183/how-the-util-of-iostat-is-computed
// for total disk io, @see: cat /proc/vmstat |grep pgpg // for total disk io, @see: cat /proc/vmstat |grep pgpg
// for device disk io, @see: cat /proc/diskstats // for device disk io, @see: cat /proc/diskstats
@ -364,9 +354,9 @@ extern void srs_update_proc_stat();
class SrsDiskStat class SrsDiskStat
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// the time in ms when sample. // The time in ms when sample.
int64_t sample_time; int64_t sample_time;
// input(read) KBytes per seconds // input(read) KBytes per seconds
@ -382,10 +372,10 @@ public:
public: public:
// @see: cat /proc/vmstat // @see: cat /proc/vmstat
// the in(read) page count, pgpgin*1024 is the read bytes. // The in(read) page count, pgpgin*1024 is the read bytes.
// Total number of kilobytes the system paged in from disk per second. // Total number of kilobytes the system paged in from disk per second.
unsigned long pgpgin; unsigned long pgpgin;
// the out(write) page count, pgpgout*1024 is the write bytes. // The out(write) page count, pgpgout*1024 is the write bytes.
// Total number of kilobytes the system paged out to disk per second. // Total number of kilobytes the system paged out to disk per second.
unsigned long pgpgout; unsigned long pgpgout;
@ -415,7 +405,7 @@ public:
// Write I/O operations // Write I/O operations
unsigned int wr_ios; unsigned int wr_ios;
// Number of writes merged Reads and writes which are adjacent // Number of writes merged Reads and writes which are adjacent
// to each other may be merged for efficiency. Thus two 4K // To each other may be merged for efficiency. Thus two 4K
// reads may become one 8K read before it is ultimately // reads may become one 8K read before it is ultimately
// handed to the disk, and so it will be counted (and queued) // handed to the disk, and so it will be counted (and queued)
// as only one I/O. This field lets you know how often this was done. // as only one I/O. This field lets you know how often this was done.
@ -446,7 +436,7 @@ public:
// progress (field 9) times the number of milliseconds spent // progress (field 9) times the number of milliseconds spent
// doing I/O since the last update of this field. This can // doing I/O since the last update of this field. This can
// provide an easy measure of both I/O completion time and // provide an easy measure of both I/O completion time and
// the backlog that may be accumulating. // The backlog that may be accumulating.
// Average queue length // Average queue length
unsigned int aveq; unsigned int aveq;
@ -454,21 +444,21 @@ public:
SrsDiskStat(); SrsDiskStat();
}; };
// get disk stat, use cache to avoid performance problem. // Get disk stat, use cache to avoid performance problem.
extern SrsDiskStat* srs_get_disk_stat(); extern SrsDiskStat* srs_get_disk_stat();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_disk_stat(); extern void srs_update_disk_stat();
// stat system memory info // Stat system memory info
// @see: cat /proc/meminfo // @see: cat /proc/meminfo
class SrsMemInfo class SrsMemInfo
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// the time in ms when sample. // The time in ms when sample.
int64_t sample_time; int64_t sample_time;
// the percent of usage. 0.153 is 15.3%. // The percent of usage. 0.153 is 15.3%.
float percent_ram; float percent_ram;
float percent_swap; float percent_swap;
@ -495,9 +485,9 @@ public:
SrsMemInfo(); SrsMemInfo();
}; };
// get system meminfo, use cache to avoid performance problem. // Get system meminfo, use cache to avoid performance problem.
extern SrsMemInfo* srs_get_meminfo(); extern SrsMemInfo* srs_get_meminfo();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_meminfo(); extern void srs_update_meminfo();
// system cpu hardware info. // system cpu hardware info.
@ -506,7 +496,7 @@ extern void srs_update_meminfo();
class SrsCpuInfo class SrsCpuInfo
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// data of /proc/cpuinfo // data of /proc/cpuinfo
@ -520,14 +510,14 @@ public:
SrsCpuInfo(); SrsCpuInfo();
}; };
// get system cpu info, use cache to avoid performance problem. // Get system cpu info, use cache to avoid performance problem.
extern SrsCpuInfo* srs_get_cpuinfo(); extern SrsCpuInfo* srs_get_cpuinfo();
// platform(os, srs) uptime/load summary // The platform(os, srs) uptime/load summary
class SrsPlatformInfo class SrsPlatformInfo
{ {
public: public:
// whether the data is ok. // Whether the data is ok.
bool ok; bool ok;
// srs startup time, in ms. // srs startup time, in ms.
@ -551,22 +541,21 @@ public:
SrsPlatformInfo(); SrsPlatformInfo();
}; };
// get platform info, use cache to avoid performance problem. // Get platform info, use cache to avoid performance problem.
extern SrsPlatformInfo* srs_get_platform_info(); extern SrsPlatformInfo* srs_get_platform_info();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_platform_info(); extern void srs_update_platform_info();
// network device summary for each network device, // The network device summary for each network device, for example, eth0, eth1, ethN
// for example, eth0, eth1, ethN
class SrsNetworkDevices class SrsNetworkDevices
{ {
public: public:
// whether the network device is ok. // Whether the network device is ok.
bool ok; bool ok;
// 6-chars interfaces name // 6-chars interfaces name
char name[7]; char name[7];
// the sample time in ms. // The sample time in ms.
int64_t sample_time; int64_t sample_time;
public: public:
@ -594,20 +583,20 @@ public:
SrsNetworkDevices(); SrsNetworkDevices();
}; };
// get network devices info, use cache to avoid performance problem. // Get network devices info, use cache to avoid performance problem.
extern SrsNetworkDevices* srs_get_network_devices(); extern SrsNetworkDevices* srs_get_network_devices();
extern int srs_get_network_devices_count(); extern int srs_get_network_devices_count();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_network_devices(); extern void srs_update_network_devices();
// system connections, and srs rtmp network summary // The system connections, and srs rtmp network summary
class SrsNetworkRtmpServer class SrsNetworkRtmpServer
{ {
public: public:
// whether the network device is ok. // Whether the network device is ok.
bool ok; bool ok;
// the sample time in ms. // The sample time in ms.
int64_t sample_time; int64_t sample_time;
public: public:
@ -638,32 +627,32 @@ public:
SrsNetworkRtmpServer(); SrsNetworkRtmpServer();
}; };
// get network devices info, use cache to avoid performance problem. // Get network devices info, use cache to avoid performance problem.
extern SrsNetworkRtmpServer* srs_get_network_rtmp_server(); extern SrsNetworkRtmpServer* srs_get_network_rtmp_server();
// the deamon st-thread will update it. // The deamon st-thread will update it.
extern void srs_update_rtmp_server(int nb_conn, SrsKbps* kbps); extern void srs_update_rtmp_server(int nb_conn, SrsKbps* kbps);
// get local or peer ip. // Get local or peer ip.
// where local ip is the server ip which client connected. // Where local ip is the server ip which client connected.
extern std::string srs_get_local_ip(int fd); extern std::string srs_get_local_ip(int fd);
// get the local id port. // Get the local id port.
extern int srs_get_local_port(int fd); extern int srs_get_local_port(int fd);
// where peer ip is the client public ip which connected to server. // Where peer ip is the client public ip which connected to server.
extern std::string srs_get_peer_ip(int fd); extern std::string srs_get_peer_ip(int fd);
// whether string is digit number // Whether string is digit number
// is_digit("1234567890") === true // is_digit("1234567890") === true
// is_digit("0123456789") === false // is_digit("0123456789") === false
// is_digit("1234567890a") === false // is_digit("1234567890a") === false
// is_digit("a1234567890") === false // is_digit("a1234567890") === false
extern bool srs_is_digit_number(const std::string& str); extern bool srs_is_digit_number(const std::string& str);
// whether string is boolean // Whether string is boolean
// is_bool("true") == true // is_bool("true") == true
// is_bool("false") == true // is_bool("false") == true
// otherwise, false. // otherwise, false.
extern bool srs_is_boolean(const std::string& str); extern bool srs_is_boolean(const std::string& str);
// dump summaries for /api/v1/summaries. // Dump summaries for /api/v1/summaries.
extern void srs_api_dump_summaries(SrsJsonObject* obj); extern void srs_api_dump_summaries(SrsJsonObject* obj);
#endif #endif