705dbb5ea8ec88ceab99e3e50a1db684fa3f2d64
[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                         int w = write(m_fd_dest, m_buffer + m_buf_start, m_buf_end - m_buf_start);
53 //                      eDebug("wrote %d bytes", w);
54                         if (w <= 0)
55                         {
56                                 if (errno == -EINTR)
57                                         continue;
58                                 break;
59                                 // ... we would stop the thread
60                         }
61
62                                 /* this should flush all written pages to disk. */
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 >= 512*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 = lseek(m_fd_source, 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 = read(m_fd_source, 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) == 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 (!lseek(m_fd_source, 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_fd_source = fd_source;
154         m_fd_dest = fd_dest;
155         resume();
156 }
157
158 void eFilePushThread::stop()
159 {
160         if (!thread_running()) /* FIXME: races */
161                 return;
162         m_stop = 1;
163         sendSignal(SIGUSR1);
164         kill();
165 }
166
167 void eFilePushThread::pause()
168 {
169         stop();
170 }
171
172 void eFilePushThread::seek(int whence, off_t where)
173 {
174         ::lseek(m_fd_source, where, whence);
175 }
176
177 void eFilePushThread::resume()
178 {
179         m_stop = 0;
180         run();
181 }
182
183 void eFilePushThread::flush()
184 {
185         m_buf_start = m_buf_end = 0;
186 }
187
188 void eFilePushThread::enablePVRCommit(int s)
189 {
190         m_send_pvr_commit = s;
191 }
192
193 void eFilePushThread::setScatterGather(iFilePushScatterGather *sg)
194 {
195         m_sg = sg;
196 }
197
198 void eFilePushThread::sendEvent(int evt)
199 {
200         m_messagepump.send(evt);
201 }
202
203 void eFilePushThread::recvEvent(const int &evt)
204 {
205         m_event(evt);
206 }