[pull] develop from ossrs:develop

trunk/conf/full.conf

@@ -1648,6 +1648,10 @@ stream_caster {
     #       For gb28181 converter, listen at TCP port. for example, 9000.
     # @remark We always enable bundle for media streams at this port.
     listen 9000;
+
+    # SIP server for GB28181. The embedded SIP server is disabled by default, please use external SIP server
+    # such as [jsip](https://github.com/usnistgov/jsip) and there is a demo [srs-sip](https://github.com/ossrs/srs-sip)
+    # also base on it, for more information please see project [GB: External SIP](https://ossrs.io/lts/zh-cn/docs/v7/doc/gb28181#external-sip).
 }
 
 #

trunk/configure

@@ -263,7 +263,9 @@ MODULE_FILES=("srs_protocol_amf0" "srs_protocol_io" "srs_protocol_conn" "srs_pro
         "srs_protocol_rtmp_stack" "srs_protocol_utility" "srs_protocol_rtmp_msg_array" "srs_protocol_stream"
         "srs_protocol_raw_avc" "srs_protocol_http_stack" "srs_protocol_kbps" "srs_protocol_json"
         "srs_protocol_format" "srs_protocol_log" "srs_protocol_st" "srs_protocol_http_client"
-        "srs_protocol_http_conn" "srs_protocol_rtmp_conn" "srs_protocol_protobuf")
+        "srs_protocol_http_conn" "srs_protocol_rtmp_conn" "srs_protocol_protobuf"
+        "srs_protocol_http_stack_llhttp" "srs_protocol_http_stack_llhttpapi" 
+        "srs_protocol_http_stack_llhttpadapter" "srs_protocol_http_stack_llhttphttp")
 # Always include SRT protocol
 MODULE_FILES+=("srs_protocol_srt")
 ModuleLibIncs+=(${LibSRTRoot})

trunk/doc/CHANGELOG.md

@@ -7,6 +7,8 @@ The changelog for SRS.
 <a name="v7-changes"></a>
 
 ## SRS 7.0 Changelog
+* v7.0, 2025-09-03, Merge [#4469](https://github.com/ossrs/srs/pull/4469): Upgrade HTTP parser from http-parser to llhttp. v7.0.77 (#4469)
+* v7.0, 2025-09-03, Merge [#4470](https://github.com/ossrs/srs/pull/4470): RTX: Fix race condition for timer. v7.0.76 (#4470)
 * v7.0, 2025-09-02, Merge [#4466](https://github.com/ossrs/srs/pull/4466): AI: GB28181: Remove embedded SIP server and enforce external SIP usage. v7.0.75 (#4466)
 * v7.0, 2025-09-01, Merge [#4465](https://github.com/ossrs/srs/pull/4465): AI: Replace SrsSharedPtrMessage with SrsMediaPacket for unified media packet handling. v7.0.74 (#4465)
 * v7.0, 2025-09-01, Merge [#4464](https://github.com/ossrs/srs/pull/4464): AI: Use shared ptr in RTMP message. v7.0.73 (#4464)

trunk/src/app/srs_app_rtc_conn.cpp

@@ -928,18 +928,29 @@ srs_error_t SrsRtcPlayStream::do_request_keyframe(uint32_t ssrc, SrsContextId ci
 
 SrsRtcPublishRtcpTimer::SrsRtcPublishRtcpTimer(SrsRtcPublishStream *p) : p_(p)
 {
+    lock_ = srs_mutex_new();
     _srs_shared_timer->timer1s()->subscribe(this);
 }
 
 SrsRtcPublishRtcpTimer::~SrsRtcPublishRtcpTimer()
 {
-    _srs_shared_timer->timer1s()->unsubscribe(this);
+    if (true) {
+        SrsLocker(lock_);
+        _srs_shared_timer->timer1s()->unsubscribe(this);
+    }
+    srs_mutex_destroy(lock_);
 }
 
 srs_error_t SrsRtcPublishRtcpTimer::on_timer(srs_utime_t interval)
 {
     srs_error_t err = srs_success;
 
+    // This is a very heavy function, and it may potentially cause a coroutine switch.
+    // Therefore, during this function, the 'this' pointer might become invalid because
+    // the object could be freed by another thread. As a result, we must lock the object
+    // to prevent it from being freed.
+    SrsLocker(lock_);
+
     ++_srs_pps_pub->sugar;
 
     if (!p_->is_started) {
@@ -964,18 +975,29 @@ srs_error_t SrsRtcPublishRtcpTimer::on_timer(srs_utime_t interval)
 
 SrsRtcPublishTwccTimer::SrsRtcPublishTwccTimer(SrsRtcPublishStream *p) : p_(p)
 {
+    lock_ = srs_mutex_new();
     _srs_shared_timer->timer100ms()->subscribe(this);
 }
 
 SrsRtcPublishTwccTimer::~SrsRtcPublishTwccTimer()
 {
-    _srs_shared_timer->timer100ms()->unsubscribe(this);
+    if (true) {
+        SrsLocker(lock_);
+        _srs_shared_timer->timer100ms()->unsubscribe(this);
+    }
+    srs_mutex_destroy(lock_);
 }
 
 srs_error_t SrsRtcPublishTwccTimer::on_timer(srs_utime_t interval)
 {
     srs_error_t err = srs_success;
 
+    // This is a very heavy function, and it may potentially cause a coroutine switch.
+    // Therefore, during this function, the 'this' pointer might become invalid because
+    // the object could be freed by another thread. As a result, we must lock the object
+    // to prevent it from being freed.
+    SrsLocker(lock_);
+
     ++_srs_pps_pub->sugar;
 
     if (!p_->is_started) {

trunk/src/app/srs_app_rtc_conn.hpp

@@ -306,6 +306,7 @@ class SrsRtcPublishRtcpTimer : public ISrsFastTimer
 {
 private:
     SrsRtcPublishStream *p_;
+    srs_mutex_t lock_;
 
 public:
     SrsRtcPublishRtcpTimer(SrsRtcPublishStream *p);
@@ -320,6 +321,7 @@ class SrsRtcPublishTwccTimer : public ISrsFastTimer
 {
 private:
     SrsRtcPublishStream *p_;
+    srs_mutex_t lock_;
 
 public:
     SrsRtcPublishTwccTimer(SrsRtcPublishStream *p);

trunk/src/core/srs_core_version7.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR 7
 #define VERSION_MINOR 0
-#define VERSION_REVISION 75
+#define VERSION_REVISION 77
 
 #endif

trunk/src/protocol/srs_protocol_http_conn.cpp

@@ -35,18 +35,15 @@ SrsHttpParser::~SrsHttpParser()
     srs_freep(header);
 }
 
-srs_error_t SrsHttpParser::initialize(enum http_parser_type type)
+srs_error_t SrsHttpParser::initialize(enum llhttp_type type)
 {
     srs_error_t err = srs_success;
 
     jsonp = false;
     type_ = type;
 
-    // Initialize the parser, however it's not necessary.
-    http_parser_init(&parser, type_);
-    parser.data = (void *)this;
-
-    memset(&settings, 0, sizeof(settings));
+    // Initialize the settings first
+    llhttp_settings_init(&settings);
     settings.on_message_begin = on_message_begin;
     settings.on_url = on_url;
     settings.on_header_field = on_header_field;
@@ -55,6 +52,10 @@ srs_error_t SrsHttpParser::initialize(enum http_parser_type type)
     settings.on_body = on_body;
     settings.on_message_complete = on_message_complete;
 
+    // Initialize the parser with settings
+    llhttp_init(&parser, type_, &settings);
+    parser.data = (void *)this;
+
     return err;
 }
 
@@ -71,7 +72,7 @@ srs_error_t SrsHttpParser::parse_message(ISrsReader *reader, ISrsHttpMessage **p
 
     // Reset parser data and state.
     state = SrsHttpParseStateInit;
-    memset(&hp_header, 0, sizeof(http_parser));
+    memset(&hp_header, 0, sizeof(llhttp_t));
     // We must reset the field name and value, because we may get a partial value in on_header_value.
     field_name = field_value = "";
     // Reset the url.
@@ -89,7 +90,8 @@ srs_error_t SrsHttpParser::parse_message(ISrsReader *reader, ISrsHttpMessage **p
     // when got next message, the whole next message is parsed as the body of previous one,
     // and the message fail.
     // @note You can comment the bellow line, the utest will fail.
-    http_parser_init(&parser, type_);
+    llhttp_reset(&parser);
+
     // Reset the parsed type.
     parsed_type_ = HTTP_BOTH;
     // callback object ptr.
@@ -106,8 +108,8 @@ srs_error_t SrsHttpParser::parse_message(ISrsReader *reader, ISrsHttpMessage **p
     SrsHttpMessage *msg = new SrsHttpMessage(reader, buffer);
 
     // Initialize the basic information.
-    msg->set_basic(hp_header.type, (http_method)hp_header.method, (http_status)hp_header.status_code, hp_header.content_length);
-    msg->set_header(header, http_should_keep_alive(&hp_header));
+    msg->set_basic(hp_header.type, (llhttp_method_t)hp_header.method, (llhttp_status_t)hp_header.status_code, hp_header.content_length);
+    msg->set_header(header, llhttp_should_keep_alive(&hp_header));
     // For HTTP response, no url.
     if (parsed_type_ != HTTP_RESPONSE && (err = msg->set_url(url, jsonp)) != srs_success) {
         srs_freep(msg);
@@ -126,19 +128,40 @@ srs_error_t SrsHttpParser::parse_message_imp(ISrsReader *reader)
 
     while (true) {
         if (buffer->size() > 0) {
-            ssize_t consumed = http_parser_execute(&parser, &settings, buffer->bytes(), buffer->size());
+            const char *data_start = buffer->bytes();
+            llhttp_errno_t code = llhttp_execute(&parser, data_start, buffer->size());
+
+            ssize_t consumed = 0;
+            if (code == HPE_OK) {
+                // No problem, all buffer should be consumed.
+                consumed = buffer->size();
+            } else if (code == HPE_PAUSED) {
+                // We only consume the header, not message or body.
+                const char *error_pos = llhttp_get_error_pos(&parser);
+                if (error_pos && error_pos < data_start) {
+                    return srs_error_new(ERROR_HTTP_PARSE_HEADER, "llhttp error_pos=%p < data_start=%p", error_pos, data_start);
+                }
 
-            // The error is set in http_errno.
-            enum http_errno code = HTTP_PARSER_ERRNO(&parser);
-            if (code != HPE_OK) {
+                if (error_pos && error_pos >= data_start) {
+                    consumed = error_pos - data_start;
+                }
+            }
+
+            // Check for errors (but allow certain conditions that are normal)
+            // HPE_OK: success
+            // HPE_PAUSED: we use to skip body
+            if (code != HPE_OK && code != HPE_PAUSED) {
                 return srs_error_new(ERROR_HTTP_PARSE_HEADER, "parse %dB, nparsed=%d, err=%d/%s %s",
-                                     buffer->size(), (int)consumed, code, http_errno_name(code), http_errno_description(code));
+                                     buffer->size(), (int)consumed, code, llhttp_errno_name(code),
+                                     llhttp_get_error_reason(&parser) ? llhttp_get_error_reason(&parser) : "");
             }
 
             srs_info("size=%d, nparsed=%d", buffer->size(), (int)consumed);
 
             // Only consume the header bytes.
-            buffer->read_slice(consumed);
+            if (consumed > 0) {
+                buffer->read_slice(consumed);
+            }
 
             // Done when header completed, never wait for body completed, because it maybe chunked.
             if (state >= SrsHttpParseStateHeaderComplete) {
@@ -161,7 +184,7 @@ srs_error_t SrsHttpParser::parse_message_imp(ISrsReader *reader)
     return err;
 }
 
-int SrsHttpParser::on_message_begin(http_parser *parser)
+int SrsHttpParser::on_message_begin(llhttp_t *parser)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
@@ -170,14 +193,14 @@ int SrsHttpParser::on_message_begin(http_parser *parser)
     obj->state = SrsHttpParseStateStart;
 
     // If we set to HTTP_BOTH, the type is detected and speicifed by parser.
-    obj->parsed_type_ = (http_parser_type)parser->type;
+    obj->parsed_type_ = (llhttp_type)parser->type;
 
     srs_info("***MESSAGE BEGIN***");
 
     return 0;
 }
 
-int SrsHttpParser::on_headers_complete(http_parser *parser)
+int SrsHttpParser::on_headers_complete(llhttp_t *parser)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
@@ -189,28 +212,29 @@ int SrsHttpParser::on_headers_complete(http_parser *parser)
     srs_info("***HEADERS COMPLETE***");
 
     // The return code of this callback:
-    //      0: Continue to process body.
-    //      1: Skip body, but continue to parse util all data parsed.
-    //      2: Upgrade and skip body and left message, because it is in a different protocol.
-    //      N: Error and failed as HPE_CB_headers_complete.
-    // We choose 2 because we only want to parse the header, not the body.
-    return 2;
+    //      `0`: Proceed normally.
+    //      `1`: Assume that request/response has no body, and proceed to parsing the next message.
+    //      `2`: Assume absence of body (as above) and make `llhttp_execute()` return `HPE_PAUSED_UPGRADE`.
+    //      `-1`: Error
+    //      `HPE_PAUSED`: Pause parsing and wait for user to call `llhttp_resume()`.
+    // We use HPE_PAUSED to skip body.
+    return HPE_PAUSED;
 }
 
-int SrsHttpParser::on_message_complete(http_parser *parser)
+int SrsHttpParser::on_message_complete(llhttp_t *parser)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
 
-    // save the parser when body parse completed.
+    // Note that we should never get here, because we always return HPE_PAUSED in on_headers_complete.
     obj->state = SrsHttpParseStateMessageComplete;
 
     srs_info("***MESSAGE COMPLETE***\n");
 
     return 0;
 }
 
-int SrsHttpParser::on_url(http_parser *parser, const char *at, size_t length)
+int SrsHttpParser::on_url(llhttp_t *parser, const char *at, size_t length)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
@@ -225,7 +249,7 @@ int SrsHttpParser::on_url(http_parser *parser, const char *at, size_t length)
     return 0;
 }
 
-int SrsHttpParser::on_header_field(http_parser *parser, const char *at, size_t length)
+int SrsHttpParser::on_header_field(llhttp_t *parser, const char *at, size_t length)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
@@ -243,7 +267,7 @@ int SrsHttpParser::on_header_field(http_parser *parser, const char *at, size_t l
     return 0;
 }
 
-int SrsHttpParser::on_header_value(http_parser *parser, const char *at, size_t length)
+int SrsHttpParser::on_header_value(llhttp_t *parser, const char *at, size_t length)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
@@ -256,7 +280,7 @@ int SrsHttpParser::on_header_value(http_parser *parser, const char *at, size_t l
     return 0;
 }
 
-int SrsHttpParser::on_body(http_parser *parser, const char *at, size_t length)
+int SrsHttpParser::on_body(llhttp_t *parser, const char *at, size_t length)
 {
     SrsHttpParser *obj = (SrsHttpParser *)parser->data;
     srs_assert(obj);
@@ -279,9 +303,9 @@ SrsHttpMessage::SrsHttpMessage(ISrsReader *reader, SrsFastStream *buffer) : ISrs
     jsonp = false;
 
     // As 0 is DELETE, so we use GET as default.
-    _method = (http_method)SRS_CONSTS_HTTP_GET;
+    _method = (llhttp_method_t)SRS_CONSTS_HTTP_GET;
     // 200 is ok.
-    _status = (http_status)SRS_CONSTS_HTTP_OK;
+    _status = (llhttp_status_t)SRS_CONSTS_HTTP_OK;
     // -1 means infinity chunked mode.
     _content_length = -1;
     // From HTTP/1.1, default to keep alive.
@@ -297,12 +321,15 @@ SrsHttpMessage::~SrsHttpMessage()
     srs_freep(_uri);
 }
 
-void SrsHttpMessage::set_basic(uint8_t type, http_method method, http_status status, int64_t content_length)
+void SrsHttpMessage::set_basic(uint8_t type, llhttp_method_t method, llhttp_status_t status, int64_t content_length)
 {
     type_ = type;
     _method = method;
     _status = status;
-    if (_content_length == -1) {
+
+    // We use -1 as uninitialized mode, while llhttp use 0, so we should only
+    // update it when it's not 0 and the message is not initialized.
+    if (_content_length == -1 && content_length) {
         _content_length = content_length;
     }
 }
@@ -455,7 +482,7 @@ string SrsHttpMessage::method_str()
         return jsonp_method;
     }
 
-    return http_method_str((http_method)_method);
+    return llhttp_method_name((llhttp_method_t)_method);
 }
 
 bool SrsHttpMessage::is_http_get()
@@ -1136,6 +1163,7 @@ srs_error_t SrsHttpResponseReader::read_chunked(void *data, size_t nb_data, ssiz
             // find the CRLF of chunk header end.
             char *start = buffer->bytes();
             char *end = start + buffer->size();
+
             for (char *p = start; p < end - 1; p++) {
                 if (p[0] == SRS_HTTP_CR && p[1] == SRS_HTTP_LF) {
                     // invalid chunk, ignore.