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(int io_prio_class, int io_prio_level)
  10         :prio_class(io_prio_class), prio(io_prio_level), m_messagepump(eApp, 0)
  11 {
  12         m_stop = 0;
  13         m_sg = 0;
  14         m_send_pvr_commit = 0;
  15         m_stream_mode = 0;
  16         flush();
  17         enablePVRCommit(0);
  18         CONNECT(m_messagepump.recv_msg, eFilePushThread::recvEvent);
  19 }
  20
  21 static void signal_handler(int x)
  22 {
  23 }
  24
  25 void eFilePushThread::thread()
  26 {
  27         setIoPrio(prio_class, prio);
  28
  29         off_t dest_pos = 0, source_pos = 0;
  30         size_t bytes_read = 0;
  31
  32         off_t current_span_offset = 0;
  33         size_t current_span_remaining = 0;
  34
  35         size_t written_since_last_sync = 0;
  36
  37         int already_empty = 0;
  38         eDebug("FILEPUSH THREAD START");
  39
  40                 /* we set the signal to not restart syscalls, so we can detect our signal. */
  41         struct sigaction act;
  42         act.sa_handler = signal_handler; // no, SIG_IGN doesn't do it. we want to receive the -EINTR
  43         act.sa_flags = 0;
  44         sigaction(SIGUSR1, &act, 0);
  45
  46         hasStarted();
  47
  48         source_pos = m_raw_source.lseek(0, SEEK_CUR);
  49
  50                 /* m_stop must be evaluated after each syscall. */
  51         while (!m_stop)
  52         {
  53                         /* first try flushing the bufptr */
  54                 if (m_buf_start != m_buf_end)
  55                 {
  56                                 // TODO: take care of boundaries.
  57                         filterRecordData(m_buffer + m_buf_start, m_buf_end - m_buf_start);
  58                         int w = write(m_fd_dest, m_buffer + m_buf_start, m_buf_end - m_buf_start);
  59 //                      eDebug("wrote %d bytes", w);
  60                         if (w <= 0)
  61                         {
  62                                 if (errno == EINTR)
  63                                         continue;
  64                                 break;
  65                                 // ... we would stop the thread
  66                         }
  67
  68                         written_since_last_sync += w;
  69
  70                         if (written_since_last_sync >= 512*1024)
  71                         {
  72                                 int toflush = written_since_last_sync > 2*1024*1024 ?
  73                                         2*1024*1024 : written_since_last_sync &~ 4095; // write max 2MB at once
  74                                 dest_pos = lseek(m_fd_dest, 0, SEEK_CUR);
  75                                 dest_pos -= toflush;
  76                                 posix_fadvise(m_fd_dest, dest_pos, toflush, POSIX_FADV_DONTNEED);
  77                                 written_since_last_sync -= toflush;
  78                         }
  79
  80 //                      printf("FILEPUSH: wrote %d bytes\n", w);
  81                         m_buf_start += w;
  82                         continue;
  83                 }
  84
  85                         /* now fill our buffer. */
  86
  87                 if (m_sg && !current_span_remaining)
  88                 {
  89                         m_sg->getNextSourceSpan(source_pos, bytes_read, current_span_offset, current_span_remaining);
  90
  91                         if (source_pos != current_span_offset)
  92                                 source_pos = m_raw_source.lseek(current_span_offset, SEEK_SET);
  93                         bytes_read = 0;
  94                 }
  95
  96                 size_t maxread = sizeof(m_buffer);
  97
  98                         /* if we have a source span, don't read past the end */
  99                 if (m_sg && maxread > current_span_remaining)
 100                         maxread = current_span_remaining;
 101
 102                 m_buf_start = 0;
 103                 m_buf_end = 0;
 104
 105                 if (maxread)
 106                         m_buf_end = m_raw_source.read(m_buffer, maxread);
 107
 108                 if (m_buf_end < 0)
 109                 {
 110                         m_buf_end = 0;
 111                         if (errno == EINTR)
 112                                 continue;
 113                         if (errno == EOVERFLOW)
 114                         {
 115                                 eWarning("OVERFLOW while recording");
 116                                 continue;
 117                         }
 118                         eDebug("eFilePushThread *read error* (%m) - not yet handled");
 119                 }
 120                 if (m_buf_end == 0)
 121                 {
 122                                 /* on EOF, try COMMITting once. */
 123                         if (m_send_pvr_commit && !already_empty)
 124                         {
 125                                 eDebug("sending PVR commit");
 126                                 already_empty = 1;
 127                                 if (::ioctl(m_fd_dest, PVR_COMMIT) < 0 && errno == EINTR)
 128                                         continue;
 129                                 eDebug("commit done");
 130                                                 /* well check again */
 131                                 continue;
 132                         }
 133
 134                                 /* in stream_mode, we are sending EOF events
 135                                    over and over until somebody responds.
 136
 137                                    in stream_mode, think of evtEOF as "buffer underrun occured". */
 138                         sendEvent(evtEOF);
 139
 140                         if (m_stream_mode)
 141                         {
 142                                 eDebug("reached EOF, but we are in stream mode. delaying 1 second.");
 143                                 sleep(1);
 144                                 continue;
 145                         }
 146 #if 0
 147                         eDebug("FILEPUSH: end-of-file! (currently unhandled)");
 148                         if (!m_raw_source.lseek(0, SEEK_SET))
 149                         {
 150                                 eDebug("(looping)");
 151                                 continue;
 152                         }
 153 #endif
 154                         break;
 155                 } else
 156                 {
 157                         source_pos += m_buf_end;
 158                         bytes_read += m_buf_end;
 159                         if (m_sg)
 160                                 current_span_remaining -= m_buf_end;
 161                         already_empty = 0;
 162                 }
 163 //              printf("FILEPUSH: read %d bytes\n", m_buf_end);
 164         }
 165         fdatasync(m_fd_dest);
 166
 167         eDebug("FILEPUSH THREAD STOP");
 168 }
 169
 170 void eFilePushThread::start(int fd_source, int fd_dest)
 171 {
 172         m_raw_source.setfd(fd_source);
 173         m_fd_dest = fd_dest;
 174         resume();
 175 }
 176
 177 int eFilePushThread::start(const char *filename, int fd_dest)
 178 {
 179         if (m_raw_source.open(filename) < 0)
 180                 return -1;
 181         m_fd_dest = fd_dest;
 182         resume();
 183         return 0;
 184 }
 185
 186 void eFilePushThread::stop()
 187 {
 188                 /* if we aren't running, don't bother stopping. */
 189         if (!sync())
 190                 return;
 191
 192         m_stop = 1;
 193
 194         // fixmee.. here we need a better solution to ensure
 195         // that the thread context take notice of the signal
 196         // even when no syscall is in progress
 197         while(!sendSignal(SIGUSR1))
 198         {
 199                 eDebug("send SIGUSR1 to thread context");
 200                 usleep(5000); // wait msek
 201         }
 202         kill();
 203 }
 204
 205 void eFilePushThread::pause()
 206 {
 207         stop();
 208 }
 209
 210 void eFilePushThread::seek(int whence, off_t where)
 211 {
 212         m_raw_source.lseek(where, whence);
 213 }
 214
 215 void eFilePushThread::resume()
 216 {
 217         m_stop = 0;
 218         run();
 219 }
 220
 221 void eFilePushThread::flush()
 222 {
 223         m_buf_start = m_buf_end = 0;
 224 }
 225
 226 void eFilePushThread::enablePVRCommit(int s)
 227 {
 228         m_send_pvr_commit = s;
 229 }
 230
 231 void eFilePushThread::setStreamMode(int s)
 232 {
 233         m_stream_mode = s;
 234 }
 235
 236 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
 237 {
 238         m_sg = sg;
 239 }
 240
 241 void eFilePushThread::sendEvent(int evt)
 242 {
 243         m_messagepump.send(evt);
 244 }
 245
 246 void eFilePushThread::recvEvent(const int &evt)
 247 {
 248         m_event(evt);
 249 }
 250
 251 void eFilePushThread::filterRecordData(const unsigned char *data, int len)
 252 {
 253         /* do nothing */
 254 }
 255