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 <lib/base/nconfig.h>
   4 #include <errno.h>
   5 #include <fcntl.h>
   6 #include <sys/ioctl.h>
   7 #include <sys/vfs.h>
   8 #if 0
   9 #include <dirent.h>
  10 #else
  11 #include <sys/types.h>
  12 #endif
  13
  14 #define PVR_COMMIT 1
  15
  16 #define MAJORSD_        8
  17 #define MAJORMMCBLK     179
  18 #define LIMIT_FILESIZE_NOHDD    2*1024*1024*1024LL      // 2GBytes
  19
  20 //FILE *f = fopen("/log.ts", "wb");
  21 static bool g_is_diskfull = false;
  22
  23 eFilePushThread::eFilePushThread(int io_prio_class, int io_prio_level, int blocksize)
  24         :prio_class(io_prio_class), prio(io_prio_level), m_messagepump(eApp, 0)
  25 {
  26         m_stop = 0;
  27         m_sg = 0;
  28         m_send_pvr_commit = 0;
  29         m_stream_mode = 0;
  30         m_blocksize = blocksize;
  31         flush();
  32         enablePVRCommit(0);
  33         CONNECT(m_messagepump.recv_msg, eFilePushThread::recvEvent);
  34         m_hdd_connected = false;
  35 }
  36
  37 static void signal_handler(int x)
  38 {
  39 }
  40
  41 void eFilePushThread::thread()
  42 {
  43         setIoPrio(prio_class, prio);
  44
  45         off_t dest_pos = 0;
  46         size_t bytes_read = 0;
  47
  48         off_t current_span_offset = 0;
  49         size_t current_span_remaining = 0;
  50
  51         size_t written_since_last_sync = 0;
  52
  53         std::string tspath;
  54         if(ePythonConfigQuery::getConfigValue("config.usage.timeshift_path", tspath) == -1) {
  55                 eDebug("could not query ts path from config");
  56         }
  57         tspath.append("/");
  58
  59 #if 0
  60         DIR *tsdir_info;
  61         struct dirent *tsdir_entry;
  62         tsdir_info = opendir("/sys/block");
  63         if (tsdir_info != NULL) {
  64                 m_hdd_connected = false;
  65                 while (tsdir_entry = readdir(tsdir_info)) {
  66                         if (strncmp(tsdir_entry->d_name, "sd", 2) == 0) {
  67                                 eDebug("HDD found: %s", tsdir_entry->d_name);
  68                                 m_hdd_connected = true;
  69                                 break;
  70                         }
  71                 }
  72         }
  73 #else
  74         struct stat tspath_st;
  75         if (stat(tspath.c_str(), &tspath_st) == 0) {
  76                 if (major(tspath_st.st_dev) == MAJORSD_) {
  77                         eDebug("Timeshift location on HDD!");
  78                         m_hdd_connected = true;
  79                 } else if (major(tspath_st.st_dev) == MAJORMMCBLK) {
  80                         eDebug("Timeshift location on eMMC!");
  81                         m_hdd_connected = false;
  82                 }
  83         }
  84 #endif
  85
  86         eDebug("FILEPUSH THREAD START");
  87
  88                 /* we set the signal to not restart syscalls, so we can detect our signal. */
  89         struct sigaction act;
  90         act.sa_handler = signal_handler; // no, SIG_IGN doesn't do it. we want to receive the -EINTR
  91         act.sa_flags = 0;
  92         sigaction(SIGUSR1, &act, 0);
  93
  94         hasStarted();
  95
  96                 /* m_stop must be evaluated after each syscall. */
  97         while (!m_stop)
  98         {
  99                         /* first try flushing the bufptr */
 100                 if (m_buf_start != m_buf_end)
 101                 {
 102                                 /* filterRecordData wants to work on multiples of blocksize.
 103                                    if it returns a negative result, it means that this many bytes should be skipped
 104                                    *in front* of the buffer. Then, it will be called again. with the newer, shorter buffer.
 105                                    if filterRecordData wants to skip more data then currently available, it must do that internally.
 106                                    Skipped bytes will also not be output.
 107
 108                                    if it returns a positive result, that means that only these many bytes should be used
 109                                    in the buffer.
 110
 111                                    In either case, current_span_remaining is given as a reference and can be modified. (Of course it
 112                                    doesn't make sense to decrement it to a non-zero value unless you return 0 because that would just
 113                                    skip some data). This is probably a very special application for fast-forward, where the current
 114                                    span is to be cancelled after a complete iframe has been output.
 115
 116                                    we always call filterRecordData with our full buffer (otherwise we couldn't easily strip from the end)
 117
 118                                    we filter data only once, of course, but it might not get immediately written.
 119                                    that's what m_filter_end is for - it points to the start of the unfiltered data.
 120                                 */
 121
 122                         int filter_res;
 123
 124                         do
 125                         {
 126                                 filter_res = filterRecordData(m_buffer + m_filter_end, m_buf_end - m_filter_end, current_span_remaining);
 127
 128                                 if (filter_res < 0)
 129                                 {
 130                                         eDebug("[eFilePushThread] filterRecordData re-syncs and skips %d bytes", -filter_res);
 131                                         m_buf_start = m_filter_end + -filter_res;  /* this will also drop unwritten data */
 132                                         ASSERT(m_buf_start <= m_buf_end); /* otherwise filterRecordData skipped more data than available. */
 133                                         continue; /* try again */
 134                                 }
 135
 136                                         /* adjust end of buffer to strip dropped tail bytes */
 137                                 m_buf_end = m_filter_end + filter_res;
 138                                         /* mark data as filtered. */
 139                                 m_filter_end = m_buf_end;
 140                         } while (0);
 141
 142                         ASSERT(m_filter_end == m_buf_end);
 143
 144                         if (m_buf_start == m_buf_end)
 145                                 continue;
 146
 147                                 /* now write out data. it will be 'aligned' (according to filterRecordData).
 148                                    absolutely forbidden is to return EINTR and consume a non-aligned number of bytes.
 149                                 */
 150                         int w = write(m_fd_dest, m_buffer + m_buf_start, m_buf_end - m_buf_start);
 151 //                      fwrite(m_buffer + m_buf_start, 1, m_buf_end - m_buf_start, f);
 152 //                      eDebug("wrote %d bytes", w);
 153                         if (w <= 0)
 154                         {
 155                                 if (w < 0 && (errno == EINTR || errno == EAGAIN || errno == EBUSY))
 156                                         continue;
 157                                 eDebug("eFilePushThread WRITE ERROR");
 158                                 sendEvent(evtWriteError);
 159
 160                                 struct statfs fs;
 161                                 if (statfs(tspath.c_str(), &fs) < 0) {
 162                                         eDebug("statfs failed!");
 163                                 }
 164                                 if ((off_t)fs.f_bavail < 1) {
 165                                         eDebug("not enough diskspace for timeshift!");
 166                                         g_is_diskfull = true;
 167                                 }
 168                                 break;
 169                                 // ... we would stop the thread
 170                         }
 171
 172                         written_since_last_sync += w;
 173
 174                         if (written_since_last_sync >= 512*1024)
 175                         {
 176                                 int toflush = written_since_last_sync > 2*1024*1024 ?
 177                                         2*1024*1024 : written_since_last_sync &~ 4095; // write max 2MB at once
 178                                 dest_pos = lseek(m_fd_dest, 0, SEEK_CUR);
 179                                 dest_pos -= toflush;
 180                                 posix_fadvise(m_fd_dest, dest_pos, toflush, POSIX_FADV_DONTNEED);
 181                                 written_since_last_sync -= toflush;
 182                         }
 183
 184 //                      printf("FILEPUSH: wrote %d bytes\n", w);
 185                         m_buf_start += w;
 186                         continue;
 187                 }
 188
 189                 if (!m_hdd_connected) {
 190                         struct stat limit_filesize;
 191                         if (fstat(m_fd_dest, &limit_filesize) == 0) {
 192                                 if (limit_filesize.st_size > LIMIT_FILESIZE_NOHDD) {
 193                                         eDebug("eFilePushThread %lld > %lld LIMIT FILESIZE", limit_filesize.st_size, LIMIT_FILESIZE_NOHDD);
 194                                         sendEvent(evtWriteError);
 195
 196                                         g_is_diskfull = true;
 197                                         break;
 198                                 }
 199                         }
 200                 }
 201
 202                         /* now fill our buffer. */
 203
 204                 if (m_sg && !current_span_remaining)
 205                 {
 206                         m_sg->getNextSourceSpan(m_current_position, bytes_read, current_span_offset, current_span_remaining);
 207                         ASSERT(!(current_span_remaining % m_blocksize));
 208                         m_current_position = current_span_offset;
 209                         bytes_read = 0;
 210                 }
 211
 212                 size_t maxread = sizeof(m_buffer);
 213
 214                         /* if we have a source span, don't read past the end */
 215                 if (m_sg && maxread > current_span_remaining)
 216                         maxread = current_span_remaining;
 217
 218                         /* align to blocksize */
 219                 maxread -= maxread % m_blocksize;
 220
 221                 m_buf_start = 0;
 222                 m_filter_end = 0;
 223                 m_buf_end = 0;
 224
 225                 if (maxread)
 226                         m_buf_end = m_source->read(m_current_position, m_buffer, maxread);
 227
 228                 if (m_buf_end < 0)
 229                 {
 230                         m_buf_end = 0;
 231                         if (errno == EINTR || errno == EBUSY || errno == EAGAIN)
 232                                 continue;
 233                         if (errno == EOVERFLOW)
 234                         {
 235                                 eWarning("OVERFLOW while recording");
 236                                 continue;
 237                         }
 238                         eDebug("eFilePushThread *read error* (%m) - not yet handled");
 239                 }
 240
 241                         /* a read might be mis-aligned in case of a short read. */
 242                 int d = m_buf_end % m_blocksize;
 243                 if (d)
 244                         m_buf_end -= d;
 245
 246                 if (m_buf_end == 0)
 247                 {
 248                                 /* on EOF, try COMMITting once. */
 249                         if (m_send_pvr_commit)
 250                         {
 251                                 struct pollfd pfd;
 252                                 pfd.fd = m_fd_dest;
 253                                 pfd.events = POLLIN;
 254                                 switch (poll(&pfd, 1, 250)) // wait for 250ms
 255                                 {
 256                                         case 0:
 257                                                 eDebug("wait for driver eof timeout");
 258                                                 continue;
 259                                         case 1:
 260                                                 eDebug("wait for driver eof ok");
 261                                                 break;
 262                                         default:
 263                                                 eDebug("wait for driver eof aborted by signal");
 264                                                 continue;
 265                                 }
 266                         }
 267
 268                                 /* in stream_mode, we are sending EOF events
 269                                    over and over until somebody responds.
 270
 271                                    in stream_mode, think of evtEOF as "buffer underrun occured". */
 272                         sendEvent(evtEOF);
 273
 274                         if (m_stream_mode)
 275                         {
 276                                 eDebug("reached EOF, but we are in stream mode. delaying 1 second.");
 277                                 sleep(1);
 278                                 continue;
 279                         }
 280                         break;
 281                 } else
 282                 {
 283                         m_current_position += m_buf_end;
 284                         bytes_read += m_buf_end;
 285                         if (m_sg)
 286                                 current_span_remaining -= m_buf_end;
 287                 }
 288 //              printf("FILEPUSH: read %d bytes\n", m_buf_end);
 289
 290                 if (g_is_diskfull) {
 291                         sendEvent(evtUser+3);
 292                         g_is_diskfull = false;
 293                 }
 294         }
 295         fdatasync(m_fd_dest);
 296
 297         eDebug("FILEPUSH THREAD STOP");
 298 }
 299
 300 void eFilePushThread::start(int fd, int fd_dest)
 301 {
 302         eRawFile *f = new eRawFile();
 303         ePtr<iTsSource> source = f;
 304         f->setfd(fd);
 305         start(source, fd_dest);
 306 }
 307
 308 int eFilePushThread::start(const char *file, int fd_dest)
 309 {
 310         eRawFile *f = new eRawFile();
 311         ePtr<iTsSource> source = f;
 312         if (f->open(file) < 0)
 313                 return -1;
 314         start(source, fd_dest);
 315         return 0;
 316 }
 317
 318 void eFilePushThread::start(ePtr<iTsSource> &source, int fd_dest)
 319 {
 320         m_source = source;
 321         m_fd_dest = fd_dest;
 322         m_current_position = 0;
 323         resume();
 324 }
 325
 326 void eFilePushThread::stop()
 327 {
 328                 /* if we aren't running, don't bother stopping. */
 329         if (!sync())
 330                 return;
 331
 332         m_stop = 1;
 333
 334         eDebug("stopping thread."); /* just do it ONCE. it won't help to do this more than once. */
 335         sendSignal(SIGUSR1);
 336         kill(0);
 337 }
 338
 339 void eFilePushThread::pause()
 340 {
 341         stop();
 342 }
 343
 344 void eFilePushThread::resume()
 345 {
 346         m_stop = 0;
 347         run();
 348 }
 349
 350 void eFilePushThread::flush()
 351 {
 352         m_buf_start = m_buf_end = m_filter_end = 0;
 353 }
 354
 355 void eFilePushThread::enablePVRCommit(int s)
 356 {
 357         m_send_pvr_commit = s;
 358 }
 359
 360 void eFilePushThread::setStreamMode(int s)
 361 {
 362         m_stream_mode = s;
 363 }
 364
 365 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
 366 {
 367         m_sg = sg;
 368 }
 369
 370 void eFilePushThread::sendEvent(int evt)
 371 {
 372         m_messagepump.send(evt);
 373 }
 374
 375 void eFilePushThread::recvEvent(const int &evt)
 376 {
 377         m_event(evt);
 378 }
 379
 380 int eFilePushThread::filterRecordData(const unsigned char *data, int len, size_t &current_span_remaining)
 381 {
 382         return len;
 383 }