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);
65 // ... we would stop the thread
68 written_since_last_sync += w;
70 if (written_since_last_sync >= 512*1024)
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);
76 posix_fadvise(m_fd_dest, dest_pos, toflush, POSIX_FADV_DONTNEED);
77 written_since_last_sync -= toflush;
80 // printf("FILEPUSH: wrote %d bytes\n", w);
85 /* now fill our buffer. */
87 if (m_sg && !current_span_remaining)
89 m_sg->getNextSourceSpan(source_pos, bytes_read, current_span_offset, current_span_remaining);
91 if (source_pos != current_span_offset)
92 source_pos = m_raw_source.lseek(current_span_offset, SEEK_SET);
96 size_t maxread = sizeof(m_buffer);
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;
106 m_buf_end = m_raw_source.read(m_buffer, maxread);
113 if (errno == EOVERFLOW)
115 eWarning("OVERFLOW while recording");
118 eDebug("eFilePushThread *read error* (%m) - not yet handled");
122 /* on EOF, try COMMITting once. */
123 if (m_send_pvr_commit && !already_empty)
125 eDebug("sending PVR commit");
127 if (::ioctl(m_fd_dest, PVR_COMMIT) < 0 && errno == EINTR)
129 eDebug("commit done");
130 /* well check again */
134 /* in stream_mode, we are sending EOF events
135 over and over until somebody responds.
137 in stream_mode, think of evtEOF as "buffer underrun occured". */
142 eDebug("reached EOF, but we are in stream mode. delaying 1 second.");
147 eDebug("FILEPUSH: end-of-file! (currently unhandled)");
148 if (!m_raw_source.lseek(0, SEEK_SET))
157 source_pos += m_buf_end;
158 bytes_read += m_buf_end;
160 current_span_remaining -= m_buf_end;
163 // printf("FILEPUSH: read %d bytes\n", m_buf_end);
165 fdatasync(m_fd_dest);
167 eDebug("FILEPUSH THREAD STOP");
170 void eFilePushThread::start(int fd_source, int fd_dest)
172 m_raw_source.setfd(fd_source);
177 int eFilePushThread::start(const char *filename, int fd_dest)
179 if (m_raw_source.open(filename) < 0)
186 void eFilePushThread::stop()
188 /* if we aren't running, don't bother stopping. */
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))
199 eDebug("send SIGUSR1 to thread context");
200 usleep(5000); // wait msek
205 void eFilePushThread::pause()
210 void eFilePushThread::seek(int whence, off_t where)
212 m_raw_source.lseek(where, whence);
215 void eFilePushThread::resume()
221 void eFilePushThread::flush()
223 m_buf_start = m_buf_end = 0;
226 void eFilePushThread::enablePVRCommit(int s)
228 m_send_pvr_commit = s;
231 void eFilePushThread::setStreamMode(int s)
236 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
241 void eFilePushThread::sendEvent(int evt)
243 m_messagepump.send(evt);
246 void eFilePushThread::recvEvent(const int &evt)
251 void eFilePushThread::filterRecordData(const unsigned char *data, int len)