1 #include <lib/base/filepush.h>
2 #include <lib/base/eerror.h>
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)
14 m_send_pvr_commit = 0;
18 CONNECT(m_messagepump.recv_msg, eFilePushThread::recvEvent);
21 static void signal_handler(int x)
25 void eFilePushThread::thread()
27 setIoPrio(prio_class, prio);
29 off_t dest_pos = 0, source_pos = 0;
30 size_t bytes_read = 0;
32 off_t current_span_offset = 0;
33 size_t current_span_remaining = 0;
35 size_t written_since_last_sync = 0;
37 int already_empty = 0;
38 eDebug("FILEPUSH THREAD START");
40 /* we set the signal to not restart syscalls, so we can detect our signal. */
42 act.sa_handler = signal_handler; // no, SIG_IGN doesn't do it. we want to receive the -EINTR
44 sigaction(SIGUSR1, &act, 0);
48 source_pos = m_raw_source.lseek(0, SEEK_CUR);
50 /* m_stop must be evaluated after each syscall. */
53 /* first try flushing the bufptr */
54 if (m_buf_start != m_buf_end)
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);
64 eDebug("eFilePushThread WRITE ERROR");
65 sendEvent(evtWriteError);
67 // ... we would stop the thread
70 written_since_last_sync += w;
72 if (written_since_last_sync >= 512*1024)
74 int toflush = written_since_last_sync > 2*1024*1024 ?
75 2*1024*1024 : written_since_last_sync &~ 4095; // write max 2MB at once
76 dest_pos = lseek(m_fd_dest, 0, SEEK_CUR);
78 posix_fadvise(m_fd_dest, dest_pos, toflush, POSIX_FADV_DONTNEED);
79 written_since_last_sync -= toflush;
82 // printf("FILEPUSH: wrote %d bytes\n", w);
87 /* now fill our buffer. */
89 if (m_sg && !current_span_remaining)
91 m_sg->getNextSourceSpan(source_pos, bytes_read, current_span_offset, current_span_remaining);
93 if (source_pos != current_span_offset)
94 source_pos = m_raw_source.lseek(current_span_offset, SEEK_SET);
98 size_t maxread = sizeof(m_buffer);
100 /* if we have a source span, don't read past the end */
101 if (m_sg && maxread > current_span_remaining)
102 maxread = current_span_remaining;
108 m_buf_end = m_raw_source.read(m_buffer, maxread);
115 if (errno == EOVERFLOW)
117 eWarning("OVERFLOW while recording");
120 eDebug("eFilePushThread *read error* (%m) - not yet handled");
124 /* on EOF, try COMMITting once. */
125 if (m_send_pvr_commit && !already_empty)
127 eDebug("sending PVR commit");
129 if (::ioctl(m_fd_dest, PVR_COMMIT) < 0 && errno == EINTR)
131 eDebug("commit done");
132 /* well check again */
136 /* in stream_mode, we are sending EOF events
137 over and over until somebody responds.
139 in stream_mode, think of evtEOF as "buffer underrun occured". */
144 eDebug("reached EOF, but we are in stream mode. delaying 1 second.");
149 eDebug("FILEPUSH: end-of-file! (currently unhandled)");
150 if (!m_raw_source.lseek(0, SEEK_SET))
159 source_pos += m_buf_end;
160 bytes_read += m_buf_end;
162 current_span_remaining -= m_buf_end;
165 // printf("FILEPUSH: read %d bytes\n", m_buf_end);
167 fdatasync(m_fd_dest);
169 eDebug("FILEPUSH THREAD STOP");
172 void eFilePushThread::start(int fd_source, int fd_dest)
174 m_raw_source.setfd(fd_source);
179 int eFilePushThread::start(const char *filename, int fd_dest)
181 if (m_raw_source.open(filename) < 0)
188 void eFilePushThread::stop()
190 /* if we aren't running, don't bother stopping. */
196 // fixmee.. here we need a better solution to ensure
197 // that the thread context take notice of the signal
198 // even when no syscall is in progress
199 while(!sendSignal(SIGUSR1))
201 eDebug("send SIGUSR1 to thread context");
202 usleep(5000); // wait msek
207 void eFilePushThread::pause()
212 void eFilePushThread::seek(int whence, off_t where)
214 m_raw_source.lseek(where, whence);
217 void eFilePushThread::resume()
223 void eFilePushThread::flush()
225 m_buf_start = m_buf_end = 0;
228 void eFilePushThread::enablePVRCommit(int s)
230 m_send_pvr_commit = s;
233 void eFilePushThread::setStreamMode(int s)
238 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
243 void eFilePushThread::sendEvent(int evt)
245 m_messagepump.send(evt);
248 void eFilePushThread::recvEvent(const int &evt)
253 void eFilePushThread::filterRecordData(const unsigned char *data, int len)