code.vuplus.com Git - vuplus_dvbapp/blob - lib/base/filepush.cpp

   1 #include <lib/base/filepush.h>
   2 #include <lib/base/eerror.h>
   3 #include <errno.h>
   4 #include <fcntl.h>
   5 #include <sys/ioctl.h>
   6
   7 #define PVR_COMMIT 1
   8
   9 eFilePushThread::eFilePushThread(): m_messagepump(eApp, 0)
  10 {
  11         m_stop = 0;
  12         m_sg = 0;
  13         flush();
  14         enablePVRCommit(0);
  15         CONNECT(m_messagepump.recv_msg, eFilePushThread::recvEvent);
  16 }
  17
  18 static void signal_handler(int x)
  19 {
  20 }
  21
  22 void eFilePushThread::thread()
  23 {
  24         off_t dest_pos = 0, source_pos = 0;
  25         size_t bytes_read = 0;
  26
  27         off_t current_span_offset = 0;
  28         size_t current_span_remaining = 0;
  29
  30         size_t written_since_last_sync = 0;
  31
  32         int already_empty = 0;
  33         eDebug("FILEPUSH THREAD START");
  34                 // this is a race. FIXME.
  35
  36                 /* we set the signal to not restart syscalls, so we can detect our signal. */
  37         struct sigaction act;
  38         act.sa_handler = signal_handler; // no, SIG_IGN doesn't do it. we want to receive the -EINTR
  39         act.sa_flags = 0;
  40         sigaction(SIGUSR1, &act, 0);
  41
  42         dest_pos = lseek(m_fd_dest, 0, SEEK_CUR);
  43         source_pos = lseek(m_fd_source, 0, SEEK_CUR);
  44
  45                 /* m_stop must be evaluated after each syscall. */
  46         while (!m_stop)
  47         {
  48                         /* first try flushing the bufptr */
  49                 if (m_buf_start != m_buf_end)
  50                 {
  51                                 // TODO: take care of boundaries.
  52                         int w = write(m_fd_dest, m_buffer + m_buf_start, m_buf_end - m_buf_start);
  53 //                      eDebug("wrote %d bytes", w);
  54                         if (w <= 0)
  55                         {
  56                                 if (errno == -EINTR)
  57                                         continue;
  58                                 break;
  59                                 // ... we would stop the thread
  60                         }
  61
  62                                 /* this should flush all written pages to disk. */
  63                         posix_fadvise(m_fd_dest, dest_pos, w, POSIX_FADV_DONTNEED);
  64
  65                         dest_pos += w;
  66                         written_since_last_sync += w;
  67
  68                         if (written_since_last_sync >= 512*1024)
  69                         {
  70                                 fdatasync(m_fd_dest);
  71                                 written_since_last_sync = 0;
  72                         }
  73
  74 //                      printf("FILEPUSH: wrote %d bytes\n", w);
  75                         m_buf_start += w;
  76                         continue;
  77                 }
  78
  79                         /* now fill our buffer. */
  80
  81                 if (m_sg && !current_span_remaining)
  82                 {
  83                         m_sg->getNextSourceSpan(source_pos, bytes_read, current_span_offset, current_span_remaining);
  84
  85                         if (source_pos != current_span_offset)
  86                                 source_pos = lseek(m_fd_source, current_span_offset, SEEK_SET);
  87                         bytes_read = 0;
  88                 }
  89
  90                 size_t maxread = sizeof(m_buffer);
  91
  92                         /* if we have a source span, don't read past the end */
  93                 if (m_sg && maxread > current_span_remaining)
  94                         maxread = current_span_remaining;
  95
  96                 m_buf_start = 0;
  97                 m_buf_end = 0;
  98
  99                 if (maxread)
 100                         m_buf_end = read(m_fd_source, m_buffer, maxread);
 101
 102                 if (m_buf_end < 0)
 103                 {
 104                         m_buf_end = 0;
 105                         if (errno == EINTR)
 106                                 continue;
 107                         if (errno == EOVERFLOW)
 108                         {
 109                                 eWarning("OVERFLOW while recording");
 110                                 continue;
 111                         }
 112                         eDebug("eFilePushThread *read error* (%m) - not yet handled");
 113                 }
 114                 if (m_buf_end == 0)
 115                 {
 116                                 /* on EOF, try COMMITting once. */
 117                         if (m_send_pvr_commit && !already_empty)
 118                         {
 119                                 eDebug("sending PVR commit");
 120                                 already_empty = 1;
 121                                 if (::ioctl(m_fd_dest, PVR_COMMIT) == EINTR)
 122                                         continue;
 123                                 eDebug("commit done");
 124                                                 /* well check again */
 125                                 continue;
 126                         }
 127                         sendEvent(evtEOF);
 128 #if 0
 129                         eDebug("FILEPUSH: end-of-file! (currently unhandled)");
 130                         if (!lseek(m_fd_source, 0, SEEK_SET))
 131                         {
 132                                 eDebug("(looping)");
 133                                 continue;
 134                         }
 135 #endif
 136                         break;
 137                 } else
 138                 {
 139                         source_pos += m_buf_end;
 140                         bytes_read += m_buf_end;
 141                         if (m_sg)
 142                                 current_span_remaining -= m_buf_end;
 143                         already_empty = 0;
 144                 }
 145 //              printf("FILEPUSH: read %d bytes\n", m_buf_end);
 146         }
 147
 148         eDebug("FILEPUSH THREAD STOP");
 149 }
 150
 151 void eFilePushThread::start(int fd_source, int fd_dest)
 152 {
 153         m_fd_source = fd_source;
 154         m_fd_dest = fd_dest;
 155         resume();
 156 }
 157
 158 void eFilePushThread::stop()
 159 {
 160         if (!thread_running()) /* FIXME: races */
 161                 return;
 162         m_stop = 1;
 163         sendSignal(SIGUSR1);
 164         kill();
 165 }
 166
 167 void eFilePushThread::pause()
 168 {
 169         stop();
 170 }
 171
 172 void eFilePushThread::seek(int whence, off_t where)
 173 {
 174         ::lseek(m_fd_source, where, whence);
 175 }
 176
 177 void eFilePushThread::resume()
 178 {
 179         m_stop = 0;
 180         run();
 181 }
 182
 183 void eFilePushThread::flush()
 184 {
 185         m_buf_start = m_buf_end = 0;
 186 }
 187
 188 void eFilePushThread::enablePVRCommit(int s)
 189 {
 190         m_send_pvr_commit = s;
 191 }
 192
 193 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
 194 {
 195         m_sg = sg;
 196 }
 197
 198 void eFilePushThread::sendEvent(int evt)
 199 {
 200         m_messagepump.send(evt);
 201 }
 202
 203 void eFilePushThread::recvEvent(const int &evt)
 204 {
 205         m_event(evt);
 206 }