4bc52050b4a647cd22f0228f6068e37c6caf672f
[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 = m_raw_source.lseek(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                         posix_fadvise(m_fd_dest, dest_pos, w, POSIX_FADV_DONTNEED);
64
65                         dest_pos += w;
66                         written_since_last_sync += w;
67                         
68                         if (written_since_last_sync >= 2048*1024)
69                         {
70                                 fdatasync(m_fd_dest);
71                                 written_since_last_sync = 0;
72                         }
73
74 //                      printf("FILEPUSH: wrote %d bytes\n", w);
75                         m_buf_start += w;
76                         continue;
77                 }
78
79                         /* now fill our buffer. */
80                         
81                 if (m_sg && !current_span_remaining)
82                 {
83                         m_sg->getNextSourceSpan(source_pos, bytes_read, current_span_offset, current_span_remaining);
84
85                         if (source_pos != current_span_offset)
86                                 source_pos = m_raw_source.lseek(current_span_offset, SEEK_SET);
87                         bytes_read = 0;
88                 }
89                 
90                 size_t maxread = sizeof(m_buffer);
91                 
92                         /* if we have a source span, don't read past the end */
93                 if (m_sg && maxread > current_span_remaining)
94                         maxread = current_span_remaining;
95
96                 m_buf_start = 0;
97                 m_buf_end = 0;
98                 
99                 if (maxread)
100                         m_buf_end = m_raw_source.read(m_buffer, maxread);
101
102                 if (m_buf_end < 0)
103                 {
104                         m_buf_end = 0;
105                         if (errno == EINTR)
106                                 continue;
107                         if (errno == EOVERFLOW)
108                         {
109                                 eWarning("OVERFLOW while recording");
110                                 continue;
111                         }
112                         eDebug("eFilePushThread *read error* (%m) - not yet handled");
113                 }
114                 if (m_buf_end == 0)
115                 {
116                                 /* on EOF, try COMMITting once. */
117                         if (m_send_pvr_commit && !already_empty)
118                         {
119                                 eDebug("sending PVR commit");
120                                 already_empty = 1;
121                                 if (::ioctl(m_fd_dest, PVR_COMMIT) < 0 && errno == EINTR)
122                                         continue;
123                                 eDebug("commit done");
124                                                 /* well check again */
125                                 continue;
126                         }
127                         sendEvent(evtEOF);
128 #if 0
129                         eDebug("FILEPUSH: end-of-file! (currently unhandled)");
130                         if (!m_raw_source.lseek(0, SEEK_SET))
131                         {
132                                 eDebug("(looping)");
133                                 continue;
134                         }
135 #endif
136                         break;
137                 } else
138                 {
139                         source_pos += m_buf_end;
140                         bytes_read += m_buf_end;
141                         if (m_sg)
142                                 current_span_remaining -= m_buf_end;
143                         already_empty = 0;
144                 }
145 //              printf("FILEPUSH: read %d bytes\n", m_buf_end);
146         }
147         
148         eDebug("FILEPUSH THREAD STOP");
149 }
150
151 void eFilePushThread::start(int fd_source, int fd_dest)
152 {
153         m_raw_source.setfd(fd_source);
154         m_fd_dest = fd_dest;
155         resume();
156 }
157
158 int eFilePushThread::start(const char *filename, int fd_dest)
159 {
160         if (m_raw_source.open(filename) < 0)
161                 return -1;
162         m_fd_dest = fd_dest;
163         resume();
164         return 0;
165 }
166
167 void eFilePushThread::stop()
168 {
169         if (!thread_running()) /* FIXME: races */
170                 return;
171         m_stop = 1;
172         sendSignal(SIGUSR1);
173         kill();
174 }
175
176 void eFilePushThread::pause()
177 {
178         stop();
179 }
180
181 void eFilePushThread::seek(int whence, off_t where)
182 {
183         m_raw_source.lseek(where, whence);
184 }
185
186 void eFilePushThread::resume()
187 {
188         m_stop = 0;
189         run();
190 }
191
192 void eFilePushThread::flush()
193 {
194         m_buf_start = m_buf_end = 0;
195 }
196
197 void eFilePushThread::enablePVRCommit(int s)
198 {
199         m_send_pvr_commit = s;
200 }
201
202 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
203 {
204         m_sg = sg;
205 }
206
207 void eFilePushThread::sendEvent(int evt)
208 {
209         m_messagepump.send(evt);
210 }
211
212 void eFilePushThread::recvEvent(const int &evt)
213 {
214         m_event(evt);
215 }
216
217 void eFilePushThread::filterRecordData(const unsigned char *data, int len)
218 {
219         /* do nothing */
220 }
221