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                         filterRecordData(m_buffer + m_buf_start, m_buf_end - m_buf_start);
  53                         int w = write(m_fd_dest, m_buffer + m_buf_start, m_buf_end - m_buf_start);
  54 //                      eDebug("wrote %d bytes", w);
  55                         if (w <= 0)
  56                         {
  57                                 if (errno == -EINTR)
  58                                         continue;
  59                                 break;
  60                                 // ... we would stop the thread
  61                         }
  62
  63                                 /* this should flush all written pages to disk. */
  64                         posix_fadvise(m_fd_dest, dest_pos, w, POSIX_FADV_DONTNEED);
  65
  66                         dest_pos += w;
  67                         written_since_last_sync += w;
  68
  69                         if (written_since_last_sync >= 512*1024)
  70                         {
  71                                 fdatasync(m_fd_dest);
  72                                 written_since_last_sync = 0;
  73                         }
  74
  75 //                      printf("FILEPUSH: wrote %d bytes\n", w);
  76                         m_buf_start += w;
  77                         continue;
  78                 }
  79
  80                         /* now fill our buffer. */
  81
  82                 if (m_sg && !current_span_remaining)
  83                 {
  84                         m_sg->getNextSourceSpan(source_pos, bytes_read, current_span_offset, current_span_remaining);
  85
  86                         if (source_pos != current_span_offset)
  87                                 source_pos = lseek(m_fd_source, current_span_offset, SEEK_SET);
  88                         bytes_read = 0;
  89                 }
  90
  91                 size_t maxread = sizeof(m_buffer);
  92
  93                         /* if we have a source span, don't read past the end */
  94                 if (m_sg && maxread > current_span_remaining)
  95                         maxread = current_span_remaining;
  96
  97                 m_buf_start = 0;
  98                 m_buf_end = 0;
  99
 100                 if (maxread)
 101                         m_buf_end = read(m_fd_source, m_buffer, maxread);
 102
 103                 if (m_buf_end < 0)
 104                 {
 105                         m_buf_end = 0;
 106                         if (errno == EINTR)
 107                                 continue;
 108                         if (errno == EOVERFLOW)
 109                         {
 110                                 eWarning("OVERFLOW while recording");
 111                                 continue;
 112                         }
 113                         eDebug("eFilePushThread *read error* (%m) - not yet handled");
 114                 }
 115                 if (m_buf_end == 0)
 116                 {
 117                                 /* on EOF, try COMMITting once. */
 118                         if (m_send_pvr_commit && !already_empty)
 119                         {
 120                                 eDebug("sending PVR commit");
 121                                 already_empty = 1;
 122                                 if (::ioctl(m_fd_dest, PVR_COMMIT) == EINTR)
 123                                         continue;
 124                                 eDebug("commit done");
 125                                                 /* well check again */
 126                                 continue;
 127                         }
 128                         sendEvent(evtEOF);
 129 #if 0
 130                         eDebug("FILEPUSH: end-of-file! (currently unhandled)");
 131                         if (!lseek(m_fd_source, 0, SEEK_SET))
 132                         {
 133                                 eDebug("(looping)");
 134                                 continue;
 135                         }
 136 #endif
 137                         break;
 138                 } else
 139                 {
 140                         source_pos += m_buf_end;
 141                         bytes_read += m_buf_end;
 142                         if (m_sg)
 143                                 current_span_remaining -= m_buf_end;
 144                         already_empty = 0;
 145                 }
 146 //              printf("FILEPUSH: read %d bytes\n", m_buf_end);
 147         }
 148
 149         eDebug("FILEPUSH THREAD STOP");
 150 }
 151
 152 void eFilePushThread::start(int fd_source, int fd_dest)
 153 {
 154         m_fd_source = fd_source;
 155         m_fd_dest = fd_dest;
 156         resume();
 157 }
 158
 159 void eFilePushThread::stop()
 160 {
 161         if (!thread_running()) /* FIXME: races */
 162                 return;
 163         m_stop = 1;
 164         sendSignal(SIGUSR1);
 165         kill();
 166 }
 167
 168 void eFilePushThread::pause()
 169 {
 170         stop();
 171 }
 172
 173 void eFilePushThread::seek(int whence, off_t where)
 174 {
 175         ::lseek(m_fd_source, where, whence);
 176 }
 177
 178 void eFilePushThread::resume()
 179 {
 180         m_stop = 0;
 181         run();
 182 }
 183
 184 void eFilePushThread::flush()
 185 {
 186         m_buf_start = m_buf_end = 0;
 187 }
 188
 189 void eFilePushThread::enablePVRCommit(int s)
 190 {
 191         m_send_pvr_commit = s;
 192 }
 193
 194 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
 195 {
 196         m_sg = sg;
 197 }
 198
 199 void eFilePushThread::sendEvent(int evt)
 200 {
 201         m_messagepump.send(evt);
 202 }
 203
 204 void eFilePushThread::recvEvent(const int &evt)
 205 {
 206         m_event(evt);
 207 }
 208
 209 void eFilePushThread::filterRecordData(const unsigned char *data, int len)
 210 {
 211         /* do nothing */
 212 }
 213