use Xthreads in session butler.
[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
24 #ifndef PLATFORM_WINDOWS
25 #include <poll.h>
26 #endif
27
28 #include "pbd/error.h"
29 #include "pbd/pthread_utils.h"
30 #include "ardour/debug.h"
31 #include "ardour/butler.h"
32 #include "ardour/io.h"
33 #include "ardour/midi_diskstream.h"
34 #include "ardour/session.h"
35 #include "ardour/track.h"
36 #include "ardour/auditioner.h"
37
38 #include "i18n.h"
39
40 using namespace PBD;
41
42 namespace ARDOUR {
43
44 Butler::Butler(Session& s)
45         : SessionHandleRef (s)
46         , thread()
47         , have_thread (false)
48         , audio_dstream_capture_buffer_size(0)
49         , audio_dstream_playback_buffer_size(0)
50         , midi_dstream_buffer_size(0)
51         , pool_trash(16)
52         , _xthread (true)
53 {
54         g_atomic_int_set(&should_do_transport_work, 0);
55         SessionEvent::pool->set_trash (&pool_trash);
56
57         Config->ParameterChanged.connect_same_thread (*this, boost::bind (&Butler::config_changed, this, _1));
58 }
59
60 Butler::~Butler()
61 {
62         terminate_thread ();
63 }
64
65 void
66 Butler::config_changed (std::string p)
67 {
68         if (p == "playback-buffer-seconds") {
69                 /* size is in Samples, not bytes */
70                 audio_dstream_playback_buffer_size = (uint32_t) floor (Config->get_audio_playback_buffer_seconds() * _session.frame_rate());
71                 _session.adjust_playback_buffering ();
72         } else if (p == "capture-buffer-seconds") {
73                 audio_dstream_capture_buffer_size = (uint32_t) floor (Config->get_audio_capture_buffer_seconds() * _session.frame_rate());
74                 _session.adjust_capture_buffering ();
75         }
76 }
77
78 int
79 Butler::start_thread()
80 {
81         const float rate = (float)_session.frame_rate();
82
83         /* size is in Samples, not bytes */
84         audio_dstream_capture_buffer_size = (uint32_t) floor (Config->get_audio_capture_buffer_seconds() * rate);
85         audio_dstream_playback_buffer_size = (uint32_t) floor (Config->get_audio_playback_buffer_seconds() * rate);
86
87         /* size is in bytes
88          * XXX: Jack needs to tell us the MIDI buffer size
89          * (i.e. how many MIDI bytes we might see in a cycle)
90          */
91         midi_dstream_buffer_size = (uint32_t) floor (Config->get_midi_track_buffer_seconds() * rate);
92
93         MidiDiskstream::set_readahead_frames ((framecnt_t) (Config->get_midi_readahead() * rate));
94
95         should_run = false;
96
97         if (pthread_create_and_store ("disk butler", &thread, _thread_work, this)) {
98                 error << _("Session: could not create butler thread") << endmsg;
99                 return -1;
100         }
101
102         //pthread_detach (thread);
103         have_thread = true;
104         return 0;
105 }
106
107 void
108 Butler::terminate_thread ()
109 {
110         if (have_thread) {
111                 void* status;
112                 DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: ask butler to quit @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
113                 queue_request (Request::Quit);
114                 pthread_join (thread, &status);
115         }
116 }
117
118 void *
119 Butler::_thread_work (void* arg)
120 {
121         SessionEvent::create_per_thread_pool ("butler events", 4096);
122         pthread_set_name (X_("butler"));
123         return ((Butler *) arg)->thread_work ();
124 }
125
126 void *
127 Butler::thread_work ()
128 {
129         uint32_t err = 0;
130
131         bool disk_work_outstanding = false;
132         RouteList::iterator i;
133
134         while (true) {
135                 DEBUG_TRACE (DEBUG::Butler, string_compose ("%1 butler main loop, disk work outstanding ? %2 @ %3\n", DEBUG_THREAD_SELF, disk_work_outstanding, g_get_monotonic_time()));
136
137                 if(!disk_work_outstanding) {
138                         DEBUG_TRACE (DEBUG::Butler, string_compose ("%1 butler waits for requests @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
139
140                         char msg;
141                         /* empty the pipe of all current requests */
142                         if (_xthread.receive (msg, true) >= 0) {
143                                 Request::Type req = (Request::Type) msg;
144                                 switch (req) {
145
146                                         case Request::Run:
147                                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: butler asked to run @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
148                                                 should_run = true;
149                                                 break;
150
151                                         case Request::Pause:
152                                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: butler asked to pause @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
153                                                 should_run = false;
154                                                 break;
155
156                                         case Request::Quit:
157                                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: butler asked to quit @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
158                                                 return 0;
159                                                 abort(); /*NOTREACHED*/
160                                                 break;
161
162                                         default:
163                                                 break;
164                                 }
165                         }
166                 }
167
168                 
169           restart:
170                 DEBUG_TRACE (DEBUG::Butler, "at restart for disk work\n");
171                 disk_work_outstanding = false;
172
173                 if (transport_work_requested()) {
174                         DEBUG_TRACE (DEBUG::Butler, string_compose ("do transport work @ %1\n", g_get_monotonic_time()));
175                         _session.butler_transport_work ();
176                         DEBUG_TRACE (DEBUG::Butler, string_compose ("\ttransport work complete @ %1\n", g_get_monotonic_time()));
177                 }
178
179                 frameoffset_t audition_seek;
180                 if (should_run && _session.is_auditioning()
181                                 && (audition_seek = _session.the_auditioner()->seek_frame()) >= 0) {
182                         boost::shared_ptr<Track> tr = boost::dynamic_pointer_cast<Track> (_session.the_auditioner());
183                         DEBUG_TRACE (DEBUG::Butler, "seek the auditioner\n");
184                         tr->seek(audition_seek);
185                         _session.the_auditioner()->seek_response(audition_seek);
186                 }
187
188                 boost::shared_ptr<RouteList> rl = _session.get_routes();
189
190                 RouteList rl_with_auditioner = *rl;
191                 rl_with_auditioner.push_back (_session.the_auditioner());
192
193                 for (i = rl_with_auditioner.begin(); !transport_work_requested() && should_run && i != rl_with_auditioner.end(); ++i) {
194
195                         boost::shared_ptr<Track> tr = boost::dynamic_pointer_cast<Track> (*i);
196
197                         if (!tr) {
198                                 continue;
199                         }
200
201                         boost::shared_ptr<IO> io = tr->input ();
202
203                         if (io && !io->active()) {
204                                 /* don't read inactive tracks */
205                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("butler skips inactive track %1\n", tr->name()));
206                                 continue;
207                         }
208                         DEBUG_TRACE (DEBUG::Butler, string_compose ("butler refills %1, playback load = %2\n", tr->name(), tr->playback_buffer_load()));
209                         switch (tr->do_refill ()) {
210                         case 0:
211                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("\ttrack refill done %1\n", tr->name()));
212                                 break;
213                                 
214                         case 1:
215                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("\ttrack refill unfinished %1\n", tr->name()));
216                                 disk_work_outstanding = true;
217                                 break;
218
219                         default:
220                                 error << string_compose(_("Butler read ahead failure on dstream %1"), (*i)->name()) << endmsg;
221                                 std::cerr << string_compose(_("Butler read ahead failure on dstream %1"), (*i)->name()) << std::endl;
222                                 break;
223                         }
224
225                 }
226
227                 if (i != rl_with_auditioner.begin() && i != rl_with_auditioner.end()) {
228                         /* we didn't get to all the streams */
229                         disk_work_outstanding = true;
230                 }
231
232                 if (!err && transport_work_requested()) {
233                         DEBUG_TRACE (DEBUG::Butler, "transport work requested during refill, back to restart\n");
234                         goto restart;
235                 }
236
237                 for (i = rl->begin(); !transport_work_requested() && should_run && i != rl->end(); ++i) {
238                         // cerr << "write behind for " << (*i)->name () << endl;
239
240                         boost::shared_ptr<Track> tr = boost::dynamic_pointer_cast<Track> (*i);
241
242                         if (!tr) {
243                                 continue;
244                         }
245
246                         /* note that we still try to flush diskstreams attached to inactive routes
247                          */
248
249                         gint64 before, after;
250                         int ret;
251
252                         DEBUG_TRACE (DEBUG::Butler, string_compose ("butler flushes track %1 capture load %2\n", tr->name(), tr->capture_buffer_load()));
253                         before = g_get_monotonic_time ();
254                         ret = tr->do_flush (ButlerContext);
255                         after = g_get_monotonic_time ();
256                         switch (ret) {
257                         case 0:
258                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("\tflush complete for %1, %2 usecs\n", tr->name(), after - before));
259                                 break;
260                                 
261                         case 1:
262                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("\tflush not finished for %1, %2 usecs\n", tr->name(), after - before));
263                                 disk_work_outstanding = true;
264                                 break;
265
266                         default:
267                                 err++;
268                                 error << string_compose(_("Butler write-behind failure on dstream %1"), (*i)->name()) << endmsg;
269                                 std::cerr << string_compose(_("Butler write-behind failure on dstream %1"), (*i)->name()) << std::endl;
270                                 /* don't break - try to flush all streams in case they
271                                    are split across disks.
272                                 */
273                         }
274                 }
275
276                 if (err && _session.actively_recording()) {
277                         /* stop the transport and try to catch as much possible
278                            captured state as we can.
279                         */
280                         DEBUG_TRACE (DEBUG::Butler, "error occurred during recording - stop transport\n");
281                         _session.request_stop ();
282                 }
283
284                 if (i != rl->begin() && i != rl->end()) {
285                         /* we didn't get to all the streams */
286                         DEBUG_TRACE (DEBUG::Butler, "not all tracks processed, will need to go back for more\n");
287                         disk_work_outstanding = true;
288                 }
289
290                 if (!err && transport_work_requested()) {
291                         DEBUG_TRACE (DEBUG::Butler, "transport work requested during flush, back to restart\n");
292                         goto restart;
293                 }
294
295                 if (!disk_work_outstanding) {
296                         _session.refresh_disk_space ();
297                 }
298
299
300                 {
301                         Glib::Threads::Mutex::Lock lm (request_lock);
302
303                         if (should_run && (disk_work_outstanding || transport_work_requested())) {
304                                 DEBUG_TRACE (DEBUG::Butler, string_compose ("at end, should run %1 disk work %2 transport work %3 ... goto restart\n",
305                                                                             should_run, disk_work_outstanding, transport_work_requested()));
306                                 goto restart;
307                         }
308
309                         DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: butler signals pause @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
310                         paused.signal();
311                 }
312
313                 DEBUG_TRACE (DEBUG::Butler, "butler emptying pool trash\n");
314                 empty_pool_trash ();
315         }
316
317         return (0);
318 }
319
320 void
321 Butler::schedule_transport_work ()
322 {
323         g_atomic_int_inc (&should_do_transport_work);
324         summon ();
325 }
326
327 void
328 Butler::queue_request (Request::Type r)
329 {
330         char c = r;
331         _xthread.deliver (c);
332 }
333
334 void
335 Butler::summon ()
336 {
337         DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: summon butler to run @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
338         queue_request (Request::Run);
339 }
340
341 void
342 Butler::stop ()
343 {
344         Glib::Threads::Mutex::Lock lm (request_lock);
345         DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: asking butler to stop @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
346         queue_request (Request::Pause);
347         paused.wait(request_lock);
348 }
349
350 void
351 Butler::wait_until_finished ()
352 {
353         Glib::Threads::Mutex::Lock lm (request_lock);
354         DEBUG_TRACE (DEBUG::Butler, string_compose ("%1: waiting for butler to finish @ %2\n", DEBUG_THREAD_SELF, g_get_monotonic_time()));
355         queue_request (Request::Pause);
356         paused.wait(request_lock);
357 }
358
359 bool
360 Butler::transport_work_requested () const
361 {
362         return g_atomic_int_get(&should_do_transport_work);
363 }
364
365 void
366 Butler::empty_pool_trash ()
367 {
368         /* look in the trash, deleting empty pools until we come to one that is not empty */
369
370         RingBuffer<CrossThreadPool*>::rw_vector vec;
371         pool_trash.get_read_vector (&vec);
372
373         guint deleted = 0;
374
375         for (int i = 0; i < 2; ++i) {
376                 for (guint j = 0; j < vec.len[i]; ++j) {
377                         if (vec.buf[i][j]->empty()) {
378                                 delete vec.buf[i][j];
379                                 ++deleted;
380                         } else {
381                                 /* found a non-empty pool, so stop deleting */
382                                 if (deleted) {
383                                         pool_trash.increment_read_idx (deleted);
384                                 }
385                                 return;
386                         }
387                 }
388         }
389
390         if (deleted) {
391                 pool_trash.increment_read_idx (deleted);
392         }
393 }
394
395 void
396 Butler::drop_references ()
397 {
398         std::cerr << "Butler drops pool trash\n";
399         SessionEvent::pool->set_trash (0);
400 }
401
402
403 } // namespace ARDOUR
404