Ticket #1660: asyncdb.13130.patch

libs/libmythtv/NuppelVideoPlayer.cpp

@@ -5429 +5429 @@
     using_null_videoout = true;
 
     // clear out any existing seektables
-    m_playbackinfo->ClearPositionMap(MARK_KEYFRAME);
-    m_playbackinfo->ClearPositionMap(MARK_GOP_START);
-    m_playbackinfo->ClearPositionMap(MARK_GOP_BYFRAME);
+    m_playbackinfo->ClearPositionMap(MARK_KEYFRAME,true);
+    m_playbackinfo->ClearPositionMap(MARK_GOP_START,true);
+    m_playbackinfo->ClearPositionMap(MARK_GOP_BYFRAME,true);
 
     if (OpenFile() < 0)
         return(0);

libs/libmythtv/programinfo.h

@@ -237 +237 @@
     // Keyframe positions Map
     void GetPositionMap(frm_pos_map_t &, int type) const;
     void ClearPositionMap(int type) const;
+    void ClearPositionMap(int type, bool sync) const;
     void SetPositionMap(frm_pos_map_t &, int type,
                         long long min_frm = -1, long long max_frm = -1) const;
-    void SetPositionMapDelta(frm_pos_map_t &, int type) const;
+    void SetPositionMapDelta(frm_pos_map_t &, int type);
 
-
     // GUI stuff
     void showDetails(void) const;
     void EditRecording(void);
@@ -344 +344 @@
   private:
     bool ignoreBookmark;
     mutable class ScheduledRecording* record;
+    
+    QString inUseForWhat;
 
-    QString inUseForWhat;
 };
 
 /** \class ProgramList

libs/libmythtv/programinfo.cpp

@@ -17 +17 @@
 #include "util.h"
 #include "mythcontext.h"
 #include "dialogbox.h"
+#include "asyncdb.h"
 #include "remoteutil.h"
 #include "jobqueue.h"
 #include "mythdbcon.h"
@@ -1828 +1829 @@
 {
     filesize = fsize;
 
-    MSqlQuery query(MSqlQuery::InitCon());
-    query.prepare("UPDATE recorded SET filesize = :FILESIZE"
-                  " WHERE chanid = :CHANID"
-                  " AND starttime = :STARTTIME ;");
-    query.bindValue(":FILESIZE", longLongToString(fsize));
-    query.bindValue(":CHANID", chanid);
-    query.bindValue(":STARTTIME", recstartts);
+    QString query = QString("UPDATE recorded SET filesize = %1"
+                            " WHERE chanid = %2"
+                            " AND starttime = \"%3\" ;")
+      .arg(longLongToString(fsize))
+      .arg(chanid)
+      .arg(recstartts.toString("yyyyMMddhhmmss"));
     
-    if (!query.exec() || !query.isActive())
-        MythContext::DBError("File size update", 
-                             query);
+    gAsyncDB->AddCommand(query);
 }
 
 /** \fn ProgramInfo::GetFilesize(void)
@@ -2605 +2603 @@
 
 void ProgramInfo::ClearPositionMap(int type) const
 {
+    QString query;
+  
+    if (isVideo)
+    {
+        query = QString("DELETE FROM filemarkup"
+                        " WHERE filename = \"%1\"")
+        .arg(pathname);
+    }
+    else
+    {
+        query = QString("DELETE FROM recordedseek"
+                      " WHERE chanid = %1"
+                      " AND starttime = \"%2\"")
+        .arg(chanid)
+        .arg(recstartts.toString("yyyyMMddhhmmss"));
+    }
+    query += QString(" AND type = %1 ;").arg(type);
+                               
+    /* Hand it to the async thread */
+    gAsyncDB->AddCommand(query);
+}
+
+void ProgramInfo::ClearPositionMap(int type, bool sync) const
+{
     MSqlQuery query(MSqlQuery::InitCon());
   
     if (isVideo)
@@ -2710 +2732 @@
 }
 
 void ProgramInfo::SetPositionMapDelta(frm_pos_map_t &posMap,
-                                      int type) const
+                                      int type)
 {
     QMap<long long, long long>::Iterator i;
-    MSqlQuery query(MSqlQuery::InitCon());
+    QString insert;
 
+    if( posMap.isEmpty() )
+      return;
+
+    if (isVideo)
+      {
+        insert = QString("INSERT INTO filemarkup"
+                         " (filename, mark, type, offset)"
+                         " VALUES");
+      } 
+    else 
+      {
+        insert = QString("INSERT INTO recordedseek"
+                         " (chanid, starttime, mark, type, offset)"
+                         " VALUES");
+      }
+    QString sep = " ";
+
+    /* Assemble these these updates into a single multi-row insert 
+     * statement -- much more efficient */
     for (i = posMap.begin(); i != posMap.end(); ++i)
     {
         long long frame = i.key();
         long long offset = i.data();
 
         if (isVideo)
-        {
-            query.prepare("INSERT INTO filemarkup"
-                          " (filename, mark, type, offset)"
-                          " VALUES"
-                          " ( :PATH , :MARK , :TYPE , :OFFSET );");
-            query.bindValue(":PATH", pathname);
-        }
-        else
-        {
-            query.prepare("INSERT INTO recordedseek"
-                          " (chanid, starttime, mark, type, offset)"
-                          " VALUES"
-                          " ( :CHANID , :STARTTIME , :MARK , :TYPE , :OFFSET );");
-            query.bindValue(":CHANID", chanid);
-            query.bindValue(":STARTTIME", recstartts);
-        }
-        query.bindValue(":MARK", frame);
-        query.bindValue(":TYPE", type);
-        query.bindValue(":OFFSET", offset);
-        
-        if (!query.exec() || !query.isActive())
-            MythContext::DBError("delta position map insert", 
-                                 query);
-    }
+          {
+            insert += sep + QString("( \"%1\", %2, %3, %4 )")
+              .arg(pathname).arg(frame).arg(type).arg(offset);
+          }
+          else
+          {
+            insert += sep + QString("( %1 , \"%2\" , %3 , %4 , %5 )")
+              .arg(chanid).arg(recstartts.toString("yyyyMMddhhmmss"))
+              .arg(frame).arg(type).arg(offset);
+          }
+        sep = ",";
+      }
+
+    insert += QString(";");
+    /* Hand it to the async thread */
+    gAsyncDB->AddCommand(insert);
 }
 
 /** \fn ProgramInfo::ReactivateRecording(void)

libs/libmyth/exitcodes.h

@@ -57 +57 @@
 #define BACKEND_BUGGY_EXIT_NO_CAP_CARD            GENERIC_EXIT_START-11
 #define BACKEND_BUGGY_EXIT_NO_CHAN_DATA           GENERIC_EXIT_START-12
 #define BACKEND_EXIT_START                        GENERIC_EXIT_START-12
+#define BACKEND_EXIT_ASYNCDB_ERROR                GENERIC_EXIT_START-13
 
 // mythtranscode
 #define TRANSCODE_EXIT_OK                         GENERIC_EXIT_OK

libs/libmyth/libmyth.pro

@@ -20 +20 @@
 HEADERS += langsettings.h audiooutputnull.h mythsocket.h
 HEADERS += DisplayResScreen.h util-x11.h mythdeque.h qmdcodec.h
 HEADERS += exitcodes.h virtualkeyboard.h mythobservable.h mythevent.h
+HEADERS += asyncdb.h
 HEADERS += mythexp.h mythpluginapi.h
 
 SOURCES += dialogbox.cpp lcddevice.cpp mythcontext.cpp mythwidgets.cpp 
@@ -33 +34 @@
 SOURCES += langsettings.cpp mythdbcon.cpp audiooutputnull.cpp
 SOURCES += DisplayResScreen.cpp util-x11.cpp qmdcodec.cpp
 SOURCES += virtualkeyboard.cpp mythobservable.cpp mythsocket.cpp
+SOURCES += asyncdb.cpp
 
 INCLUDEPATH += ../libmythsamplerate ../libmythsoundtouch ../.. ../
 DEPENDPATH += ../libmythsamplerate ../libmythsoundtouch ../ ../libmythui
@@ -59 +61 @@
 inc.files += visual.h volumebase.h output.h langsettings.h qmdcodec.h
 inc.files += exitcodes.h mythconfig.h mythconfig.mak virtualkeyboard.h
 inc.files += mythevent.h mythobservable.h mythsocket.h
+inc.files += asyncdb.h
 inc.files += mythexp.h mythpluginapi.h
 
 using_oss {

libs/libmyth/asyncdb.h

@@ +1 @@
+#ifndef ASYNCDB_H_
+#define ASYNCDB_H_
+
+// ANSI C headers
+#include <cstdio>
+#include <cstdlib>
+#include <cerrno>
+
+#include "mythdbcon.h"
+
+#include <qstring.h>
+
+using namespace std;
+
+class AsyncDB
+{
+ public:
+    AsyncDB(void);
+    ~AsyncDB();
+    bool Init(void);
+    void AddCommand(QString);  
+
+ private:
+    pthread_t thread;
+    QStringList list;
+    QMutex listLock;
+    int retry;
+    bool threadRunning;
+
+ protected:
+    static void *StartThread(void *);
+    void Worker(void);
+};
+
+extern AsyncDB *gAsyncDB;
+#endif

libs/libmyth/asyncdb.cpp

@@ +1 @@
+#include "asyncdb.h"
+#include <unistd.h>
+#include <sys/time.h>
+#include <sys/resource.h>
+#include <sys/mman.h>
+
+AsyncDB *gAsyncDB = NULL;
+#define ADB QString("AsyncDB: ")
+
+/** \class AsyncDB
+ *  \brief This class supports asynchronous database inserts.
+ *
+ *   This class allows us to toss those database queries for which
+ *   we do not require values or status to be run by a separate thread.
+ *   This helps the enocder threads to keep up with the datastream.
+ */
+
+/** \fn AsyncDB
+ *  \brief Initialize the class
+ */
+AsyncDB::AsyncDB(void):
+  retry(0), threadRunning(false)
+{
+  list.clear();
+}
+
+/** \fn Init()
+ *  \brief Starts the thread
+ *
+ *   Create a thread and return false if this fails
+ */
+bool AsyncDB::Init(void)
+{
+  int rv;
+  if(  ( rv = pthread_create(&thread, NULL, StartThread, this) ) ) {
+    VERBOSE(VB_IMPORTANT, ADB + QString("Can't start thread"));
+    return false;
+  }
+  return(true);
+}
+
+/** \fn StartThread(AsyncDB *)
+ *  \brief Runs the worker function
+ *
+ *   Drop the priority on this thread and invoke the worker function
+ */
+void *AsyncDB::StartThread(void *wotzit)
+{
+  AsyncDB *pi = (AsyncDB *)wotzit;
+  VERBOSE(VB_IMPORTANT,QString("Starting async db thread"));
+  // Loser priority, to avoid problems with recordings.
+  pi->threadRunning = true;
+  setpriority(PRIO_PROCESS, 0, 3);
+  do {
+    pi->Worker();
+  } while(pi->retry <= 1 );     // A second retry means we're hosed
+  pi->threadRunning = false;
+  return(NULL);
+}
+
+/** \fn AddCommand(QString)
+ *  \brief Adds a database command to the list
+ *
+ *  \param QString   The sql command
+ */
+void AsyncDB::AddCommand(QString cmd)
+{
+  if( threadRunning ) {
+    listLock.lock();
+    list.append(cmd);
+    listLock.unlock();
+  }
+}
+
+/** \fn ~AsyncDB
+ *  \brief Shut down the thread
+ *
+ *   Add a "done" command to the list to make the thread 
+ *   shutdown and reap it.
+ */
+AsyncDB::~AsyncDB(void)
+{
+  AddCommand(QString("done"));
+  pthread_join(thread, NULL);
+  VERBOSE(VB_IMPORTANT,QString("Ending async db thread"));
+}
+
+/** \fn Worker()
+ *  \brief Run the lists of commands
+ *
+ *   Swap the list for an empty one and call ListRunner
+ *   to execute the commands.
+ */
+void AsyncDB::Worker(void)
+{
+  bool done = false;
+  bool errored = false;
+  MSqlQuery query(MSqlQuery::InitCon());
+
+  while( ! done && ! errored ) {
+    if( list.empty() ) {
+      sleep(1);
+    } else {
+      listLock.lock();
+      QStringList mylist = list;
+      list.clear();
+      listLock.unlock();
+      for ( QStringList::Iterator it = mylist.begin(); it != mylist.end(); ++it ) {
+        if( *it == QString("done") ) {
+          done = true;
+          retry = 2;                    // Make calling function exit.
+          break;
+        }
+        if( errored ) {
+          /* If we experienece a database error, we stuff the commands back onto the list,
+             and exit this function. We get called again which refreshes the DB connection.
+             We count these retries, so we don't loop, but only do it once without some
+             success */
+          AddCommand(*it);
+        } else {
+          query.prepare(*it);
+          if (!query.exec() || !query.isActive()) {
+            MythContext::DBError("delta position map insert", 
+                                 query);
+            retry++;
+            errored = true;
+            AddCommand(*it);
+          } else
+            retry = 0;
+        }
+      }
+      mylist.clear(); 
+    }
+  }
+}
+
+

programs/mythtranscode/main.cpp

@@ -561 +561 @@
 {
     if (pginfo && ! mapfile)
     {
-        pginfo->ClearPositionMap(MARK_KEYFRAME);
-        pginfo->ClearPositionMap(MARK_GOP_START);
+        pginfo->ClearPositionMap(MARK_KEYFRAME, true);
+        pginfo->ClearPositionMap(MARK_GOP_START, true);
         pginfo->SetPositionMap(posMap, MARK_GOP_BYFRAME);
     }
     else if (mapfile)

programs/mythbackend/main.cpp

@@ -35 +35 @@
 #include "libmythtv/dbcheck.h"
 #include "libmythtv/jobqueue.h"
 #include "libmythtv/storagegroup.h"
+#include "libmyth/asyncdb.h"
 
 #include "mediaserver.h"
 #include "httpstatus.h"
@@ -199 +200 @@
 
 void cleanup(void) 
 {
+    delete gAsyncDB;
     delete gContext;
 
     if (sched)
@@ -518 +520 @@
     }    
     gContext->ActivateSettingsCache(true);
 
+    gAsyncDB = new AsyncDB();
+    if( gAsyncDB->Init() == false )
+      {
+        VERBOSE(VB_IMPORTANT, "Couldn't start async database thread");
+        return BACKEND_EXIT_ASYNCDB_ERROR;
+      }
+
     if (printsched || testsched)
     {
         gContext->SetBackend(false);