Merge pull request #469 from ksooo/fix-issue465-2

7.1.4: HTSP connection related fixes and cleanup

pvr.hts/addon.xml.in

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <addon
   id="pvr.hts"
-  version="7.1.3"
+  version="7.1.4"
   name="Tvheadend HTSP Client"
   provider-name="Adam Sutton, Sam Stenvall, Lars Op den Kamp, Kai Sommerfeld">
   <requires>@ADDON_DEPENDS@</requires>

pvr.hts/changelog.txt

@@ -1,3 +1,9 @@
+7.1.4
+- Fix deadlock while reconnecting to backend after a backend response timeout (Fixes #465)
+- Refactor subscription seek code
+- Refactor subscription weight code
+- Fix HTSP connection thread not ended while in supended state
+
 7.1.3
 - Withdraw v7.1.2 due to new deadlocks it introduces, for example during initial connect to backend.
 

src/Tvheadend.cpp

@@ -1493,31 +1493,8 @@ void CTvheadend::Disconnected()
 
 bool CTvheadend::Connected()
 {
-  /* Rebuild state */
-  for (auto* dmx : m_dmx)
-    dmx->Connected();
-
-  m_vfs->Connected();
-  m_timeRecordings.Connected();
-  m_autoRecordings.Connected();
-
-  /* Flag all async fields in case they've been deleted */
-  for (auto& entry : m_channels)
-    entry.second.SetDirty(true);
-  for (auto& entry : m_tags)
-    entry.second.SetDirty(true);
-  for (auto& entry : m_schedules)
-    entry.second.SetDirty(true);
-
-  {
-    CLockObject lock(m_mutex);
-
-    for (auto& entry : m_recordings)
-      entry.second.SetDirty(true);
-  }
-
-  /* Request Async data, first is channels */
-  m_asyncState.SetState(ASYNC_CHN);
+  /* Request Async data, first is init (which rebuilds state) */
+  m_asyncState.SetState(ASYNC_INIT);
 
   htsmsg_t* msg = htsmsg_create_map();
   if (Settings::GetInstance().GetAsyncEpg())
@@ -1902,6 +1879,7 @@ void CTvheadend::SyncCompleted()
   Logger::Log(LogLevel::LEVEL_INFO, "async updates initialised");
 
   /* The complete calls are probably redundant, but its a safety feature */
+  SyncInitCompleted();
   SyncChannelsCompleted();
   SyncDvrCompleted();
   SyncEpgCompleted();
@@ -1926,6 +1904,34 @@ void CTvheadend::SyncCompleted()
   }
 }
 
+void CTvheadend::SyncInitCompleted()
+{
+  /* check state engine */
+  if (m_asyncState.GetState() != ASYNC_INIT)
+    return;
+
+  /* Rebuild state */
+  for (auto* dmx : m_dmx)
+    dmx->RebuildState();
+
+  m_vfs->RebuildState();
+  m_timeRecordings.RebuildState();
+  m_autoRecordings.RebuildState();
+
+  /* Flag all async fields in case they've been deleted */
+  for (auto& entry : m_channels)
+    entry.second.SetDirty(true);
+  for (auto& entry : m_tags)
+    entry.second.SetDirty(true);
+  for (auto& entry : m_schedules)
+    entry.second.SetDirty(true);
+  for (auto& entry : m_recordings)
+    entry.second.SetDirty(true);
+
+  /* Next */
+  m_asyncState.SetState(ASYNC_CHN);
+}
+
 void CTvheadend::SyncChannelsCompleted()
 {
   /* check state engine */
@@ -1987,16 +1993,15 @@ void CTvheadend::SyncDvrCompleted()
 void CTvheadend::SyncEpgCompleted()
 {
   /* check state engine */
+  if (m_asyncState.GetState() != ASYNC_EPG)
+    return;
+
   if (!Settings::GetInstance().GetAsyncEpg())
   {
     m_asyncState.SetState(ASYNC_DONE);
     return;
   }
 
-  /* check state engine */
-  if (m_asyncState.GetState() != ASYNC_EPG)
-    return;
-
   /* Schedules */
   std::vector<std::pair<uint32_t, uint32_t>> deletedEvents;
   utilities::erase_if(m_schedules, [&](const ScheduleMapEntry& entry) {
@@ -2040,6 +2045,9 @@ void CTvheadend::SyncEpgCompleted()
 
 void CTvheadend::ParseTagAddOrUpdate(htsmsg_t* msg, bool bAdd)
 {
+  /* Rebuild state upon arrival of first async data */
+  SyncInitCompleted();
+
   /* Validate */
   uint32_t u32 = 0;
   if (htsmsg_get_u32(msg, "tagId", &u32))
@@ -2120,6 +2128,8 @@ void CTvheadend::ParseTagDelete(htsmsg_t* msg)
 
 void CTvheadend::ParseChannelAddOrUpdate(htsmsg_t* msg, bool bAdd)
 {
+  /* Rebuild state upon arrival of first async data */
+  SyncInitCompleted();
 
   /* Validate */
   uint32_t u32 = 0;
@@ -2243,7 +2253,10 @@ void CTvheadend::ParseChannelDelete(htsmsg_t* msg)
 
 void CTvheadend::ParseRecordingAddOrUpdate(htsmsg_t* msg, bool bAdd)
 {
-  /* Channels must be complete */
+  /* Rebuild state upon arrival of first async data */
+  SyncInitCompleted();
+
+  /* Channels complete */
   SyncChannelsCompleted();
 
   /* Validate */
@@ -2637,6 +2650,9 @@ void CTvheadend::ParseRecordingDelete(htsmsg_t* msg)
 
 bool CTvheadend::ParseEvent(htsmsg_t* msg, bool bAdd, Event& evt)
 {
+  /* Rebuild state upon arrival of first async data */
+  SyncInitCompleted();
+
   /* Recordings complete */
   SyncDvrCompleted();
 

src/Tvheadend.h

@@ -174,6 +174,7 @@ class ATTRIBUTE_HIDDEN CTvheadend : public kodi::addon::CInstancePVRClient,
   /*
    * Channel/Tags/Recordings/Events
    */
+  void SyncInitCompleted();
   void SyncChannelsCompleted();
   void SyncDvrCompleted();
   void SyncEpgCompleted();

src/tvheadend/AutoRecordings.cpp

@@ -31,7 +31,7 @@ AutoRecordings::~AutoRecordings()
 {
 }
 
-void AutoRecordings::Connected()
+void AutoRecordings::RebuildState()
 {
   /* Flag all async fields in case they've been deleted */
   for (auto& rec : m_autoRecordings)

src/tvheadend/AutoRecordings.h

@@ -33,7 +33,7 @@ class AutoRecordings
   ~AutoRecordings();
 
   /* state updates */
-  void Connected();
+  void RebuildState();
   void SyncDvrCompleted();
 
   /* data access */

src/tvheadend/HTSPConnection.cpp

@@ -325,6 +325,7 @@ bool HTSPConnection::ReadMessage()
     htsmsg_destroy(msg);
     return true;
   }
+
   Logger::Log(LogLevel::LEVEL_TRACE, "receive message [%s]", method);
 
   /* Pass (if return is true, message is finished) */
@@ -365,7 +366,8 @@ bool HTSPConnection::SendMessage0(const char* method, htsmsg_t* msg)
 
   if (c != static_cast<ssize_t>(len))
   {
-    Logger::Log(LogLevel::LEVEL_ERROR, "failed to write (%s)", m_socket->GetError().c_str());
+    Logger::Log(LogLevel::LEVEL_ERROR, "Command %s failed: failed to write (%s)", method,
+                m_socket->GetError().c_str());
     if (!m_suspended)
       Disconnect();
 
@@ -394,7 +396,7 @@ htsmsg_t* HTSPConnection::SendAndWait0(const char* method, htsmsg_t* msg, int iR
   if (!SendMessage0(method, msg))
   {
     m_messages.erase(seq);
-    Logger::Log(LogLevel::LEVEL_ERROR, "failed to transmit");
+    Logger::Log(LogLevel::LEVEL_ERROR, "Command %s failed: failed to transmit", method);
     return nullptr;
   }
 
@@ -581,7 +583,6 @@ void HTSPConnection::Register()
 
     /* Send Auth */
     Logger::Log(LogLevel::LEVEL_DEBUG, "sending auth");
-
     if (!SendAuth(user, pass))
     {
       SetState(PVR_CONNECTION_STATE_ACCESS_DENIED);
@@ -595,6 +596,7 @@ void HTSPConnection::Register()
 
     Logger::Log(LogLevel::LEVEL_DEBUG, "registered");
     SetState(PVR_CONNECTION_STATE_CONNECTED);
+
     m_ready = true;
     m_regCond.Broadcast();
     return;
@@ -603,7 +605,7 @@ void HTSPConnection::Register()
 fail:
   if (!m_suspended)
   {
-    /* Don't immediately reconnect (spare server CPU cycles)*/
+    /* Don't immediately reconnect (spare server CPU cycles) */
     Sleep(SLOW_RECONNECT_INTERVAL);
     Disconnect();
   }
@@ -643,14 +645,15 @@ void* HTSPConnection::Process()
       }
     }
 
-    while (m_suspended)
+    while (m_suspended && !IsStopped())
     {
-      Logger::Log(LogLevel::LEVEL_DEBUG, "suspended. Waiting for wakeup...");
-
       /* Wait for wakeup */
       Sleep(1000);
     }
 
+    if (IsStopped())
+      break;
+
     if (!log)
     {
       Logger::Log(LogLevel::LEVEL_DEBUG, "connecting to %s:%d", host.c_str(), port);
@@ -689,6 +692,7 @@ void* HTSPConnection::Process()
 
       continue;
     }
+
     Logger::Log(LogLevel::LEVEL_DEBUG, "connected");
     log = false;
     retryAttempt = 0;