add stream-mode to filepush, which will retry on eof until stream grows

[enigma2.git] / lib / base / filepush.cpp

diff --git a/lib/base/filepush.cpp b/lib/base/filepush.cpp
index 0f41b3917cf09eeff98bf3260206e00c1840d284..92c580c22d43c22cb6c05ce2915b507ce6a8cc69 100644 (file)
--- a/lib/base/filepush.cpp
+++ b/lib/base/filepush.cpp
@@ -6,10 +6,13 @@
 
 #define PVR_COMMIT 1
 
-eFilePushThread::eFilePushThread(): m_messagepump(eApp, 0)
+eFilePushThread::eFilePushThread(int io_prio_class, int io_prio_level)
+       :prio_class(io_prio_class), prio(io_prio_level), m_messagepump(eApp, 0)
 {
        m_stop = 0;
        m_sg = 0;
+       m_send_pvr_commit = 0;
+       m_stream_mode = 0;
        flush();
        enablePVRCommit(0);
        CONNECT(m_messagepump.recv_msg, eFilePushThread::recvEvent);
@@ -21,6 +24,8 @@ static void signal_handler(int x)
 
 void eFilePushThread::thread()
 {
+       setIoPrio(prio_class, prio);
+
        off_t dest_pos = 0, source_pos = 0;
        size_t bytes_read = 0;
        
@@ -28,10 +33,9 @@ void eFilePushThread::thread()
        size_t current_span_remaining = 0;
        
        size_t written_since_last_sync = 0;
-       
+
        int already_empty = 0;
        eDebug("FILEPUSH THREAD START");
-               // this is a race. FIXME.
        
                /* we set the signal to not restart syscalls, so we can detect our signal. */
        struct sigaction act;
@@ -39,7 +43,8 @@ void eFilePushThread::thread()
        act.sa_flags = 0;
        sigaction(SIGUSR1, &act, 0);
        
-       dest_pos = lseek(m_fd_dest, 0, SEEK_CUR);
+       hasStarted();
+       
        source_pos = m_raw_source.lseek(0, SEEK_CUR);
        
                /* m_stop must be evaluated after each syscall. */
@@ -54,22 +59,22 @@ void eFilePushThread::thread()
 //                     eDebug("wrote %d bytes", w);
                        if (w <= 0)
                        {
-                               if (errno == -EINTR)
+                               if (errno == EINTR)
                                        continue;
                                break;
                                // ... we would stop the thread
                        }
 
-                               /* this should flush all written pages to disk. */
-                       posix_fadvise(m_fd_dest, dest_pos, w, POSIX_FADV_DONTNEED);
-                       
-                       dest_pos += w;
                        written_since_last_sync += w;
-                       
-                       if (written_since_last_sync >= 2048*1024)
+
+                       if (written_since_last_sync >= 512*1024)
                        {
-                               fdatasync(m_fd_dest);
-                               written_since_last_sync = 0;
+                               int toflush = written_since_last_sync > 2*1024*1024 ?
+                                       2*1024*1024 : written_since_last_sync &~ 4095; // write max 2MB at once
+                               dest_pos = lseek(m_fd_dest, 0, SEEK_CUR);
+                               dest_pos -= toflush;
+                               posix_fadvise(m_fd_dest, dest_pos, toflush, POSIX_FADV_DONTNEED);
+                               written_since_last_sync -= toflush;
                        }
 
 //                     printf("FILEPUSH: wrote %d bytes\n", w);
@@ -119,13 +124,25 @@ void eFilePushThread::thread()
                        {
                                eDebug("sending PVR commit");
                                already_empty = 1;
-                               if (::ioctl(m_fd_dest, PVR_COMMIT) == EINTR)
+                               if (::ioctl(m_fd_dest, PVR_COMMIT) < 0 && errno == EINTR)
                                        continue;
                                eDebug("commit done");
                                                /* well check again */
                                continue;
                        }
+                       
+                               /* in stream_mode, we are sending EOF events 
+                                  over and over until somebody responds.
+                                  
+                                  in stream_mode, think of evtEOF as "buffer underrun occured". */
                        sendEvent(evtEOF);
+
+                       if (m_stream_mode)
+                       {
+                               eDebug("reached EOF, but we are in stream mode. delaying 1 second.");
+                               sleep(1);
+                               continue;
+                       }
 #if 0
                        eDebug("FILEPUSH: end-of-file! (currently unhandled)");
                        if (!m_raw_source.lseek(0, SEEK_SET))
@@ -145,7 +162,8 @@ void eFilePushThread::thread()
                }
 //             printf("FILEPUSH: read %d bytes\n", m_buf_end);
        }
-       
+       fdatasync(m_fd_dest);
+
        eDebug("FILEPUSH THREAD STOP");
 }
 
@@ -167,10 +185,20 @@ int eFilePushThread::start(const char *filename, int fd_dest)
 
 void eFilePushThread::stop()
 {
-       if (!thread_running()) /* FIXME: races */
+               /* if we aren't running, don't bother stopping. */
+       if (!sync())
                return;
+
        m_stop = 1;
-       sendSignal(SIGUSR1);
+
+       // fixmee.. here we need a better solution to ensure
+       // that the thread context take notice of the signal
+       // even when no syscall is in progress
+       while(!sendSignal(SIGUSR1))
+       {
+               eDebug("send SIGUSR1 to thread context");
+               usleep(5000); // wait msek
+       }
        kill();
 }
 
@@ -200,6 +228,11 @@ void eFilePushThread::enablePVRCommit(int s)
        m_send_pvr_commit = s;
 }
 
+void eFilePushThread::setStreamMode(int s)
+{
+       m_stream_mode = s;
+}
+
 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
 {
        m_sg = sg;