add possibliity to filter recorded data
[openblackhole/openblackhole-enigma2.git] / 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 char *data, int len)
210 {
211         /* do nothing */
212 }
213