Suspend deletion of cross-thread pools until they are empty. Prevents crashes when...
[ardour.git] / libs / ardour / butler.cc
1 /*
2     Copyright (C) 1999-2009 Paul Davis
3
4     This program is free software; you can redistribute it and/or modify
5     it under the terms of the GNU General Public License as published by
6     the Free Software Foundation; either version 2 of the License, or
7     (at your option) any later version.
8
9     This program is distributed in the hope that it will be useful,
10     but WITHOUT ANY WARRANTY; without even the implied warranty of
11     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12     GNU General Public License for more details.
13
14     You should have received a copy of the GNU General Public License
15     along with this program; if not, write to the Free Software
16     Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
17
18 */
19
20 #include <errno.h>
21 #include <fcntl.h>
22 #include <unistd.h>
23 #include <poll.h>
24 #include "pbd/error.h"
25 #include "pbd/pthread_utils.h"
26 #include "ardour/butler.h"
27 #include "ardour/crossfade.h"
28 #include "ardour/io.h"
29 #include "ardour/midi_diskstream.h"
30 #include "ardour/session.h"
31
32 #include "i18n.h"
33
34 using namespace PBD;
35
36 static float _read_data_rate;
37 static float _write_data_rate;
38
39 namespace ARDOUR {
40
41 Butler::Butler(Session& s)
42         : SessionHandleRef (s)
43         , thread(0)
44         , audio_dstream_buffer_size(0)
45         , midi_dstream_buffer_size(0)
46         , pool_trash(16)
47 {
48         g_atomic_int_set(&should_do_transport_work, 0);
49         SessionEvent::pool->set_trash (&pool_trash);
50 }
51
52 Butler::~Butler()
53 {
54         terminate_thread ();
55 }
56
57 int
58 Butler::start_thread()
59 {
60         const float rate = (float)_session.frame_rate();
61
62         /* size is in Samples, not bytes */
63         audio_dstream_buffer_size = (uint32_t) floor (Config->get_audio_track_buffer_seconds() * rate);
64
65         /* size is in bytes
66          * XXX: Jack needs to tell us the MIDI buffer size
67          * (i.e. how many MIDI bytes we might see in a cycle)
68          */
69         midi_dstream_buffer_size = (uint32_t) floor (Config->get_midi_track_buffer_seconds() * rate);
70
71         MidiDiskstream::set_readahead_frames ((nframes_t)(Config->get_midi_readahead() * rate));
72
73         Crossfade::set_buffer_size (audio_dstream_buffer_size);
74
75         should_run = false;
76
77         if (pipe (request_pipe)) {
78                 error << string_compose(_("Cannot create transport request signal pipe (%1)"),
79                                 strerror (errno)) << endmsg;
80                 return -1;
81         }
82
83         if (fcntl (request_pipe[0], F_SETFL, O_NONBLOCK)) {
84                 error << string_compose(_("UI: cannot set O_NONBLOCK on butler request pipe (%1)"),
85                                 strerror (errno)) << endmsg;
86                 return -1;
87         }
88
89         if (fcntl (request_pipe[1], F_SETFL, O_NONBLOCK)) {
90                 error << string_compose(_("UI: cannot set O_NONBLOCK on butler request pipe (%1)"),
91                                 strerror (errno)) << endmsg;
92                 return -1;
93         }
94
95         if (pthread_create_and_store ("disk butler", &thread, _thread_work, this)) {
96                 error << _("Session: could not create butler thread") << endmsg;
97                 return -1;
98         }
99
100         //pthread_detach (thread);
101
102         return 0;
103 }
104
105 void
106 Butler::terminate_thread ()
107 {
108         if (thread) {
109                 void* status;
110                 const char c = Request::Quit;
111                 ::write (request_pipe[1], &c, 1);
112                 pthread_join (thread, &status);
113         }
114 }
115
116 void *
117 Butler::_thread_work (void* arg)
118 {
119         SessionEvent::create_per_thread_pool ("butler events", 64);
120         pthread_set_name (X_("butler"));
121         return ((Butler *) arg)->thread_work ();
122 }
123
124 void *
125 Butler::thread_work ()
126 {
127         uint32_t err = 0;
128         int32_t bytes;
129         bool compute_io;
130         microseconds_t begin, end;
131
132         struct pollfd pfd[1];
133         bool disk_work_outstanding = false;
134         Session::DiskstreamList::iterator i;
135
136         while (true) {
137                 pfd[0].fd = request_pipe[0];
138                 pfd[0].events = POLLIN|POLLERR|POLLHUP;
139
140                 if (poll (pfd, 1, (disk_work_outstanding ? 0 : -1)) < 0) {
141
142                         if (errno == EINTR) {
143                                 continue;
144                         }
145
146                         error << string_compose (_("poll on butler request pipe failed (%1)"),
147                                           strerror (errno))
148                               << endmsg;
149                         break;
150                 }
151
152                 if (pfd[0].revents & ~POLLIN) {
153                         error << string_compose (_("Error on butler thread request pipe: fd=%1 err=%2"), pfd[0].fd, pfd[0].revents) << endmsg;
154                         break;
155                 }
156
157                 if (pfd[0].revents & POLLIN) {
158
159                         char req;
160
161                         /* empty the pipe of all current requests */
162
163                         while (1) {
164                                 size_t nread = ::read (request_pipe[0], &req, sizeof (req));
165                                 if (nread == 1) {
166
167                                         switch ((Request::Type) req) {
168
169                                         case Request::Wake:
170                                                 break;
171
172                                         case Request::Run:
173                                                 should_run = true;
174                                                 break;
175
176                                         case Request::Pause:
177                                                 should_run = false;
178                                                 break;
179
180                                         case Request::Quit:
181                                                 pthread_exit_pbd (0);
182                                                 /*NOTREACHED*/
183                                                 break;
184
185                                         default:
186                                                 break;
187                                         }
188
189                                 } else if (nread == 0) {
190                                         break;
191                                 } else if (errno == EAGAIN) {
192                                         break;
193                                 } else {
194                                         fatal << _("Error reading from butler request pipe") << endmsg;
195                                         /*NOTREACHED*/
196                                 }
197                         }
198                 }
199
200                 if (transport_work_requested()) {
201                         _session.butler_transport_work ();
202                 }
203
204                 disk_work_outstanding = false;
205                 bytes = 0;
206                 compute_io = true;
207
208                 begin = get_microseconds();
209
210                 boost::shared_ptr<Session::DiskstreamList> dsl = _session.diskstream_list().reader ();
211
212 //              for (i = dsl->begin(); i != dsl->end(); ++i) {
213 //                      cerr << "BEFORE " << (*i)->name() << ": pb = " << (*i)->playback_buffer_load() << " cp = " << (*i)->capture_buffer_load() << endl;
214 //              }
215
216                 for (i = dsl->begin(); !transport_work_requested() && should_run && i != dsl->end(); ++i) {
217
218                         boost::shared_ptr<Diskstream> ds = *i;
219
220                         /* don't read inactive tracks */
221
222                         boost::shared_ptr<IO> io = ds->io();
223
224                         if (io && !io->active()) {
225                                 continue;
226                         }
227
228                         switch (ds->do_refill ()) {
229                         case 0:
230                                 bytes += ds->read_data_count();
231                                 break;
232                         case 1:
233                                 bytes += ds->read_data_count();
234                                 disk_work_outstanding = true;
235                                 break;
236
237                         default:
238                                 compute_io = false;
239                                 error << string_compose(_("Butler read ahead failure on dstream %1"), (*i)->name()) << endmsg;
240                                 break;
241                         }
242
243                 }
244
245                 if (i != dsl->begin() && i != dsl->end()) {
246                         /* we didn't get to all the streams */
247                         disk_work_outstanding = true;
248                 }
249
250                 if (!err && transport_work_requested()) {
251                         continue;
252                 }
253
254                 if (compute_io) {
255                         end = get_microseconds();
256                         if (end - begin > 0) {
257                                 _read_data_rate = (float) bytes / (float) (end - begin);
258                         } else {
259                                 _read_data_rate = 0; // infinity better
260                         }
261                 }
262
263                 bytes = 0;
264                 compute_io = true;
265                 begin = get_microseconds();
266
267                 for (i = dsl->begin(); !transport_work_requested() && should_run && i != dsl->end(); ++i) {
268                         // cerr << "write behind for " << (*i)->name () << endl;
269
270                         /* note that we still try to flush diskstreams attached to inactive routes
271                          */
272
273                         switch ((*i)->do_flush (ButlerContext)) {
274                         case 0:
275                                 bytes += (*i)->write_data_count();
276                                 break;
277                         case 1:
278                                 bytes += (*i)->write_data_count();
279                                 disk_work_outstanding = true;
280                                 break;
281
282                         default:
283                                 err++;
284                                 compute_io = false;
285                                 error << string_compose(_("Butler write-behind failure on dstream %1"), (*i)->name()) << endmsg;
286                                 /* don't break - try to flush all streams in case they
287                                    are split across disks.
288                                 */
289                         }
290                 }
291
292                 if (err && _session.actively_recording()) {
293                         /* stop the transport and try to catch as much possible
294                            captured state as we can.
295                         */
296                         _session.request_stop ();
297                 }
298
299                 if (i != dsl->begin() && i != dsl->end()) {
300                         /* we didn't get to all the streams */
301                         disk_work_outstanding = true;
302                 }
303
304                 if (!err && transport_work_requested()) {
305                         continue;
306                 }
307
308                 if (compute_io) {
309                         // there are no apparent users for this calculation?
310                         end = get_microseconds();
311                         if (end - begin > 0) {
312                                 _write_data_rate = (float) bytes / (float) (end - begin);
313                         } else {
314                                 _write_data_rate = 0; // Well, infinity would be better
315                         }
316                 }
317
318                 if (!disk_work_outstanding) {
319                         _session.refresh_disk_space ();
320                 }
321
322
323                 {
324                         Glib::Mutex::Lock lm (request_lock);
325
326                         if (should_run && (disk_work_outstanding || transport_work_requested())) {
327 //                              for (DiskstreamList::iterator i = dsl->begin(); i != dsl->end(); ++i) {
328 //                                      cerr << "AFTER " << (*i)->name() << ": pb = " << (*i)->playback_buffer_load() << " cp = " << (*i)->capture_buffer_load() << endl;
329 //                              }
330
331                                 continue;
332                         }
333
334                         paused.signal();
335                 }
336
337                 empty_pool_trash ();
338         }
339
340         pthread_exit_pbd (0);
341         /*NOTREACHED*/
342         return (0);
343 }
344
345 void
346 Butler::schedule_transport_work ()
347 {
348         g_atomic_int_inc (&should_do_transport_work);
349         summon ();
350 }
351
352 void
353 Butler::summon ()
354 {
355         char c = Request::Run;
356         ::write (request_pipe[1], &c, 1);
357 }
358
359 void
360 Butler::stop ()
361 {
362         Glib::Mutex::Lock lm (request_lock);
363         char c = Request::Pause;
364         ::write (request_pipe[1], &c, 1);
365         paused.wait(request_lock);
366 }
367
368 void
369 Butler::wait_until_finished ()
370 {
371         Glib::Mutex::Lock lm (request_lock);
372         char c = Request::Wake;
373         ::write (request_pipe[1], &c, 1);
374         paused.wait(request_lock);
375 }
376
377 bool
378 Butler::transport_work_requested () const
379 {
380         return g_atomic_int_get(&should_do_transport_work);
381 }
382
383 float
384 Butler::read_data_rate () const
385 {
386         /* disk i/o in excess of 10000MB/sec indicate the buffer cache
387            in action. ignore it.
388         */
389         return _read_data_rate > 10485.7600000f ? 0.0f : _read_data_rate;
390 }
391
392 float
393 Butler::write_data_rate () const
394 {
395         /* disk i/o in excess of 10000MB/sec indicate the buffer cache
396            in action. ignore it.
397         */
398         return _write_data_rate > 10485.7600000f ? 0.0f : _write_data_rate;
399 }
400
401 void
402 Butler::empty_pool_trash ()
403 {
404         /* look in the trash, deleting empty pools until we come to one that is not empty */
405         
406         RingBuffer<CrossThreadPool*>::rw_vector vec;
407         pool_trash.get_read_vector (&vec);
408
409         guint deleted = 0;
410         
411         for (int i = 0; i < 2; ++i) {
412                 for (guint j = 0; j < vec.len[i]; ++j) {
413                         if (vec.buf[i][j]->empty()) {
414                                 delete vec.buf[i][j];
415                                 ++deleted;
416                         } else {
417                                 /* found a non-empty pool, so stop deleting */
418                                 if (deleted) {
419                                         pool_trash.increment_read_idx (deleted);
420                                 }
421                                 return;
422                         }
423                 }
424         }
425
426         if (deleted) {
427                 pool_trash.increment_read_idx (deleted);
428         }
429 }
430
431 } // namespace ARDOUR
432