GB28181: Fix when camera restart, can not connect to SRS. #3944

trunk/src/app/srs_app_conn.cpp

@@ -239,6 +239,52 @@ void SrsResourceManager::remove(ISrsResource* c)
     removing_ = false;
 }
 
+void SrsResourceManager::erase(ISrsResource* c) 
+{
+    for (map<string, ISrsResource*>::iterator it = conns_name_.begin(); it != conns_name_.end();) {
+        if (c != it->second) {
+            ++it;
+        } else {
+            // Use C++98 style: https://stackoverflow.com/a/4636230
+            conns_name_.erase(it++);
+        }
+    }
+
+    for (map<string, ISrsResource*>::iterator it = conns_id_.begin(); it != conns_id_.end();) {
+        if (c != it->second) {
+            ++it;
+        } else {
+            // Use C++98 style: https://stackoverflow.com/a/4636230
+            conns_id_.erase(it++);
+        }
+    }
+
+    for (map<uint64_t, ISrsResource*>::iterator it = conns_fast_id_.begin(); it != conns_fast_id_.end();) {
+        if (c != it->second) {
+            ++it;
+        } else {
+            // Update the level-0 cache for fast-id.
+            uint64_t id = it->first;
+            SrsResourceFastIdItem* item = &conns_level0_cache_[(id | id>>32) % nn_level0_cache_];
+            item->nn_collisions--;
+            if (!item->nn_collisions) {
+                item->fast_id = 0;
+                item->available = false;
+            }
+
+            // Use C++98 style: https://stackoverflow.com/a/4636230
+            conns_fast_id_.erase(it++);
+        }
+    }
+
+    vector<ISrsResource*>::iterator it = std::find(conns_.begin(), conns_.end(), c);
+    if (it != conns_.end()) {
+        it = conns_.erase(it);
+    }
+    return;
+}
+
+
 void SrsResourceManager::do_remove(ISrsResource* c)
 {
     bool in_zombie = false;
@@ -354,46 +400,7 @@ void SrsResourceManager::do_clear()
 
 void SrsResourceManager::dispose(ISrsResource* c)
 {
-    for (map<string, ISrsResource*>::iterator it = conns_name_.begin(); it != conns_name_.end();) {
-        if (c != it->second) {
-            ++it;
-        } else {
-            // Use C++98 style: https://stackoverflow.com/a/4636230
-            conns_name_.erase(it++);
-        }
-    }
-
-    for (map<string, ISrsResource*>::iterator it = conns_id_.begin(); it != conns_id_.end();) {
-        if (c != it->second) {
-            ++it;
-        } else {
-            // Use C++98 style: https://stackoverflow.com/a/4636230
-            conns_id_.erase(it++);
-        }
-    }
-
-    for (map<uint64_t, ISrsResource*>::iterator it = conns_fast_id_.begin(); it != conns_fast_id_.end();) {
-        if (c != it->second) {
-            ++it;
-        } else {
-            // Update the level-0 cache for fast-id.
-            uint64_t id = it->first;
-            SrsResourceFastIdItem* item = &conns_level0_cache_[(id | id>>32) % nn_level0_cache_];
-            item->nn_collisions--;
-            if (!item->nn_collisions) {
-                item->fast_id = 0;
-                item->available = false;
-            }
-
-            // Use C++98 style: https://stackoverflow.com/a/4636230
-            conns_fast_id_.erase(it++);
-        }
-    }
-
-    vector<ISrsResource*>::iterator it = std::find(conns_.begin(), conns_.end(), c);
-    if (it != conns_.end()) {
-        it = conns_.erase(it);
-    }
+    erase(c);
 
     // We should copy all handlers, because it may change during callback.
     vector<ISrsDisposingHandler*> handlers = handlers_;

trunk/src/app/srs_app_conn.hpp

@@ -117,6 +117,7 @@ class SrsResourceManager : public ISrsCoroutineHandler, public ISrsResourceManag
 // Interface ISrsResourceManager
 public:
     virtual void remove(ISrsResource* c);
+    virtual void erase(ISrsResource* c);
 private:
     void do_remove(ISrsResource* c);
     void check_remove(ISrsResource* c, bool& in_zombie, bool& in_disposing);

trunk/src/app/srs_app_gb28181.cpp

@@ -76,7 +76,7 @@ SrsLazyGbSession::SrsLazyGbSession(SrsLazyObjectWrapper<SrsLazyGbSession>* wrapp
     sip_ = new SrsLazyObjectWrapper<SrsLazyGbSipTcpConn>();
     media_ = new SrsLazyObjectWrapper<SrsLazyGbMediaTcpConn>();
     muxer_ = new SrsGbMuxer(this);
-    state_ = SrsGbSessionStateInit;
+    reset();
 
     connecting_starttime_ = 0;
     connecting_timeout_ = 0;
@@ -400,6 +400,23 @@ std::string SrsLazyGbSession::desc()
     return "GBS";
 }
 
+void SrsLazyGbSession::set_to_delete()
+{
+    to_delete_ = true;
+    return;
+}
+
+bool SrsLazyGbSession::get_to_delete()
+{
+    return to_delete_;
+}
+
+void SrsLazyGbSession::reset() 
+{
+    state_ = SrsGbSessionStateInit;
+    to_delete_ = false;
+}
+
 SrsGbListener::SrsGbListener()
 {
     conf_ = NULL;
@@ -507,7 +524,7 @@ SrsLazyGbSipTcpConn::SrsLazyGbSipTcpConn(SrsLazyObjectWrapper<SrsLazyGbSipTcpCon
     conn_ = NULL;
     receiver_ = NULL;
     sender_ = NULL;
-
+    cond_ = srs_cond_new();
     trd_ = new SrsSTCoroutine("sip", this);
 }
 
@@ -521,6 +538,7 @@ SrsLazyGbSipTcpConn::~SrsLazyGbSipTcpConn()
     srs_freep(register_);
     srs_freep(invite_ok_);
     srs_freep(conf_);
+    srs_cond_destroy(cond_);
 }
 
 void SrsLazyGbSipTcpConn::setup(SrsConfDirective* conf, SrsTcpListener* sip, SrsTcpListener* media, srs_netfd_t stfd)
@@ -612,6 +630,11 @@ void SrsLazyGbSipTcpConn::enqueue_sip_message(SrsSipMessage* msg)
     sender_->enqueue(msg);
 }
 
+void SrsLazyGbSipTcpConn::on_sip_disconnect() {
+    session_->resource()->set_to_delete();
+    this->wake_up();
+}
+
 void SrsLazyGbSipTcpConn::drive_state(SrsSipMessage* msg)
 {
     srs_error_t err = srs_success;
@@ -622,6 +645,10 @@ void SrsLazyGbSipTcpConn::drive_state(SrsSipMessage* msg)
             srs_sip_state(ostate, state_).c_str()); \
     }
 
+    // if (state_ == SrsGbSipStateDisconnect) {
+    //     return;
+    // }
+
     //const char* mt = msg->type_ == HTTP_REQUEST ? "REQUEST" : "RESPONSE";
     //const char* mm = msg->type_ == HTTP_REQUEST ? http_method_str(msg->method_) : "Response";
     //int ms = msg->type_ == HTTP_REQUEST ? 200 : msg->status_;
@@ -866,13 +893,23 @@ bool SrsLazyGbSipTcpConn::is_bye()
     return state_ == SrsGbSipStateBye;
 }
 
+// bool SrsLazyGbSipTcpConn::is_disconnect()
+// {
+//     return state_ == SrsGbSipStateDisconnect;
+// }
+
 SrsGbSipState SrsLazyGbSipTcpConn::set_state(SrsGbSipState v)
 {
     SrsGbSipState state = state_;
     state_ = v;
     return state;
 }
 
+void SrsLazyGbSipTcpConn::wake_up() 
+{
+    srs_cond_signal(cond_);
+}
+
 const SrsContextId& SrsLazyGbSipTcpConn::get_id()
 {
     return trd_->cid();
@@ -909,6 +946,8 @@ srs_error_t SrsLazyGbSipTcpConn::cycle()
     // Interrupt the receiver and sender coroutine.
     receiver_->interrupt();
     sender_->interrupt();
+    // avoid bind session before resource destruction.
+    _srs_gb_manager->erase(session_);
 
     // Note that we added wrapper to manager, so we must free the wrapper, not this connection.
     SrsLazyObjectWrapper<SrsLazyGbSipTcpConn>* wrapper = wrapper_root_;
@@ -952,7 +991,11 @@ srs_error_t SrsLazyGbSipTcpConn::do_cycle()
         }
 
         // TODO: Handle other messages.
-        srs_usleep(SRS_UTIME_NO_TIMEOUT);
+        int ret = srs_cond_timedwait(cond_, 10 * SRS_UTIME_SECONDS);
+        // waitup by signal and no new connection bind session.
+        if (ret == -1 && session_->resource()->get_to_delete()) {
+            return srs_error_new(ret, "errno:%d", errno);
+        }
     }
 
     return err;
@@ -974,6 +1017,10 @@ srs_error_t SrsLazyGbSipTcpConn::bind_session(SrsSipMessage* msg, SrsLazyObjectW
 
     // Find exists session for register, might be created by another object and still alive.
     SrsLazyObjectWrapper<SrsLazyGbSession>* session = dynamic_cast<SrsLazyObjectWrapper<SrsLazyGbSession>*>(_srs_gb_manager->find_by_id(device));
+    // if session ready to delete but not insert in zombies, reuse it.
+    if (session && session->resource()->get_to_delete()) {
+        session->resource()->reset();
+    }
     if (!session) {
         // Create new GB session.
         session = new SrsLazyObjectWrapper<SrsLazyGbSession>();
@@ -1048,6 +1095,9 @@ srs_error_t SrsLazyGbSipTcpReceiver::cycle()
     // TODO: FIXME: Notify SIP transport to cleanup.
     if (err != srs_success) {
         srs_error("SIP: Receive err %s", srs_error_desc(err).c_str());
+        if (sip_) {
+            sip_->on_sip_disconnect();
+        }
     }
 
     return err;

trunk/src/app/srs_app_gb28181.hpp

@@ -86,6 +86,7 @@ enum SrsGbSipState
     SrsGbSipStateReinviting,
     SrsGbSipStateStable,
     SrsGbSipStateBye,
+    // SrsGbSipStateDisconnect,
 };
 std::string srs_gb_sip_state(SrsGbSipState state);
 
@@ -116,6 +117,8 @@ class SrsLazyGbSession : public SrsLazyObject, public ISrsResource, public ISrsS
     srs_utime_t reinvite_wait_;
     // The number of timeout, dispose session if exceed.
     uint32_t nn_timeout_;
+    // The flag to delete
+    bool to_delete_;
 private:
     SrsAlonePithyPrint* ppp_;
     srs_utime_t startime_;
@@ -164,6 +167,10 @@ class SrsLazyGbSession : public SrsLazyObject, public ISrsResource, public ISrsS
 public:
     virtual const SrsContextId& get_id();
     virtual std::string desc();
+    void set_to_delete();
+    bool get_to_delete();
+    // reset sip state, if device reconnected before insert zombies
+    void reset();
 };
 
 // The SIP and Media listener for GB.
@@ -206,6 +213,7 @@ class SrsLazyGbSipTcpConn : public SrsLazyObject, public ISrsResource, public IS
     SrsLazyGbSipTcpReceiver* receiver_;
     SrsLazyGbSipTcpSender* sender_;
     SrsCoroutine* trd_;
+    srs_cond_t cond_;
 private:
     friend class SrsLazyObjectWrapper<SrsLazyGbSipTcpConn>;
     SrsLazyGbSipTcpConn(SrsLazyObjectWrapper<SrsLazyGbSipTcpConn>* wrapper_root);
@@ -224,6 +232,8 @@ class SrsLazyGbSipTcpConn : public SrsLazyObject, public ISrsResource, public IS
 public:
     // When got a SIP message.
     srs_error_t on_sip_message(SrsSipMessage* msg);
+    // When SIP connection lost.
+    void on_sip_disconnect();
     // Enqueue a SIP message to send, which might be a request or response.
     void enqueue_sip_message(SrsSipMessage* msg);
 private:
@@ -247,8 +257,10 @@ class SrsLazyGbSipTcpConn : public SrsLazyObject, public ISrsResource, public IS
     bool is_stable();
     // Whether SIP is bye bye.
     bool is_bye();
+    // bool is_disconnect();
 private:
     SrsGbSipState set_state(SrsGbSipState v);
+    void wake_up();
 // Interface ISrsResource
 public:
     virtual const SrsContextId& get_id();