root/trunk/libffado/src/libstreaming/StreamProcessorManager.cpp

Revision 807, 38.1 kB (checked in by ppalmers, 14 years ago)

more reliability things

Line 
1 /*
2  * Copyright (C) 2005-2007 by Pieter Palmers
3  *
4  * This file is part of FFADO
5  * FFADO = Free Firewire (pro-)audio drivers for linux
6  *
7  * FFADO is based upon FreeBoB.
8  *
9  * This program is free software: you can redistribute it and/or modify
10  * it under the terms of the GNU General Public License as published by
11  * the Free Software Foundation, either version 3 of the License, or
12  * (at your option) any later version.
13  *
14  * This program is distributed in the hope that it will be useful,
15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17  * GNU General Public License for more details.
18  *
19  * You should have received a copy of the GNU General Public License
20  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
21  *
22  */
23 #include "config.h"
24
25 #include "StreamProcessorManager.h"
26 #include "generic/StreamProcessor.h"
27 #include "generic/Port.h"
28 #include "libieee1394/cycletimer.h"
29
30 #include "libutil/Time.h"
31
32 #include <errno.h>
33 #include <assert.h>
34 #include <math.h>
35
36 namespace Streaming {
37
38 IMPL_DEBUG_MODULE( StreamProcessorManager, StreamProcessorManager, DEBUG_LEVEL_VERBOSE );
39
40 StreamProcessorManager::StreamProcessorManager()
41     : m_is_slave( false )
42     , m_SyncSource(NULL)
43     , m_nb_buffers( 0 )
44     , m_period( 0 )
45     , m_nominal_framerate ( 0 )
46     , m_xrun_happened( false )
47     , m_xruns(0)
48     , m_nbperiods(0)
49 {
50     addOption(Util::OptionContainer::Option("slaveMode",false));
51 }
52
53 StreamProcessorManager::StreamProcessorManager(unsigned int period, unsigned int framerate, unsigned int nb_buffers)
54     : m_is_slave( false )
55     , m_SyncSource(NULL)
56     , m_nb_buffers(nb_buffers)
57     , m_period(period)
58     , m_nominal_framerate ( framerate )
59     , m_xruns(0)
60     , m_xrun_happened( false )
61     , m_nbperiods(0)
62 {
63     addOption(Util::OptionContainer::Option("slaveMode",false));
64 }
65
66 StreamProcessorManager::~StreamProcessorManager() {
67 }
68
69 /**
70  * Registers \ref processor with this manager.
71  *
72  * also registers it with the isohandlermanager
73  *
74  * be sure to call isohandlermanager->init() first!
75  * and be sure that the processors are also ->init()'ed
76  *
77  * @param processor
78  * @return true if successfull
79  */
80 bool StreamProcessorManager::registerProcessor(StreamProcessor *processor)
81 {
82     debugOutput( DEBUG_LEVEL_VERBOSE, "Registering processor (%p)\n",processor);
83     assert(processor);
84     if (processor->getType() == StreamProcessor::ePT_Receive) {
85         processor->setVerboseLevel(getDebugLevel()); // inherit debug level
86         m_ReceiveProcessors.push_back(processor);
87         return true;
88     }
89
90     if (processor->getType() == StreamProcessor::ePT_Transmit) {
91         processor->setVerboseLevel(getDebugLevel()); // inherit debug level
92         m_TransmitProcessors.push_back(processor);
93         return true;
94     }
95
96     debugFatal("Unsupported processor type!\n");
97     return false;
98 }
99
100 bool StreamProcessorManager::unregisterProcessor(StreamProcessor *processor)
101 {
102     debugOutput( DEBUG_LEVEL_VERBOSE, "Unregistering processor (%p)\n",processor);
103     assert(processor);
104
105     if (processor->getType()==StreamProcessor::ePT_Receive) {
106
107         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
108               it != m_ReceiveProcessors.end();
109               ++it )
110         {
111             if ( *it == processor ) {
112                 if (*it == m_SyncSource) {
113                     debugOutput(DEBUG_LEVEL_VERBOSE, "unregistering sync source\n");
114                     m_SyncSource = NULL;
115                 }
116                 m_ReceiveProcessors.erase(it);
117                 return true;
118             }
119         }
120     }
121
122     if (processor->getType()==StreamProcessor::ePT_Transmit) {
123         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
124               it != m_TransmitProcessors.end();
125               ++it )
126         {
127             if ( *it == processor ) {
128                 if (*it == m_SyncSource) {
129                     debugOutput(DEBUG_LEVEL_VERBOSE, "unregistering sync source\n");
130                     m_SyncSource = NULL;
131                 }
132                 m_TransmitProcessors.erase(it);
133                 return true;
134             }
135         }
136     }
137
138     debugFatal("Processor (%p) not found!\n",processor);
139     return false; //not found
140 }
141
142 bool StreamProcessorManager::setSyncSource(StreamProcessor *s) {
143     debugOutput( DEBUG_LEVEL_VERBOSE, "Setting sync source to (%p)\n", s);
144     m_SyncSource=s;
145     return true;
146 }
147
148 bool StreamProcessorManager::prepare() {
149
150     debugOutput( DEBUG_LEVEL_VERBOSE, "Preparing...\n");
151
152     m_is_slave=false;
153     if(!getOption("slaveMode", m_is_slave)) {
154         debugWarning("Could not retrieve slaveMode parameter, defaulting to false\n");
155     }
156
157     // if no sync source is set, select one here
158     if(m_SyncSource == NULL) {
159        debugWarning("Sync Source is not set. Defaulting to first StreamProcessor.\n");
160     }
161
162     // FIXME: put into separate method
163     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
164           it != m_ReceiveProcessors.end();
165           ++it )
166     {
167         if(m_SyncSource == NULL) {
168             debugWarning(" => Sync Source is %p.\n", *it);
169             m_SyncSource = *it;
170         }
171     }
172     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
173           it != m_TransmitProcessors.end();
174           ++it )
175     {
176         if(m_SyncSource == NULL) {
177             debugWarning(" => Sync Source is %p.\n", *it);
178             m_SyncSource = *it;
179         }
180     }
181
182     // now do the actual preparation of the SP's
183     debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare Receive processors...\n");
184     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
185         it != m_ReceiveProcessors.end();
186         ++it ) {
187
188         if(!(*it)->setOption("slaveMode", m_is_slave)) {
189             debugOutput(DEBUG_LEVEL_VERBOSE, " note: could not set slaveMode option for (%p)...\n",(*it));
190         }
191
192         if(!(*it)->prepare()) {
193             debugFatal(  " could not prepare (%p)...\n",(*it));
194             return false;
195         }
196     }
197     debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare Transmit processors...\n");
198     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
199         it != m_TransmitProcessors.end();
200         ++it ) {
201         if(!(*it)->setOption("slaveMode", m_is_slave)) {
202             debugOutput(DEBUG_LEVEL_VERBOSE, " note: could not set slaveMode option for (%p)...\n",(*it));
203         }
204         if(!(*it)->prepare()) {
205             debugFatal( " could not prepare (%p)...\n",(*it));
206             return false;
207         }
208     }
209
210     // if there are no stream processors registered,
211     // fail
212     if (m_ReceiveProcessors.size() + m_TransmitProcessors.size() == 0) {
213         debugFatal("No stream processors registered, can't do anything usefull\n");
214         return false;
215     }
216     return true;
217 }
218
219 bool StreamProcessorManager::startDryRunning() {
220     debugOutput( DEBUG_LEVEL_VERBOSE, "Putting StreamProcessor streams into dry-running state...\n");
221     debugOutput( DEBUG_LEVEL_VERBOSE, " Schedule start dry-running...\n");
222     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
223             it != m_ReceiveProcessors.end();
224             ++it ) {
225         if (!(*it)->isDryRunning()) {
226             if(!(*it)->scheduleStartDryRunning(-1)) {
227                 debugError("Could not put SP %p into the dry-running state\n", *it);
228                 return false;
229             }
230         } else {
231             debugOutput( DEBUG_LEVEL_VERBOSE, " SP %p already dry-running...\n", *it);
232         }
233     }
234     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
235             it != m_TransmitProcessors.end();
236             ++it ) {
237         if (!(*it)->isDryRunning()) {
238             if(!(*it)->scheduleStartDryRunning(-1)) {
239                 debugError("Could not put SP %p into the dry-running state\n", *it);
240                 return false;
241             }
242         } else {
243             debugOutput( DEBUG_LEVEL_VERBOSE, " SP %p already dry-running...\n", *it);
244         }
245     }
246     debugOutput( DEBUG_LEVEL_VERBOSE, " Waiting for all SP's to be dry-running...\n");
247     // wait for the syncsource to start running.
248     // that will block the waitForPeriod call until everyone has started (theoretically)
249     int cnt = STREAMPROCESSORMANAGER_CYCLES_FOR_DRYRUN; // by then it should have started
250     bool all_dry_running = false;
251     while (!all_dry_running && cnt) {
252         all_dry_running = true;
253         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
254                 it != m_ReceiveProcessors.end();
255                 ++it ) {
256             all_dry_running &= (*it)->isDryRunning();
257         }
258         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
259                 it != m_TransmitProcessors.end();
260                 ++it ) {
261             all_dry_running &= (*it)->isDryRunning();
262         }
263
264         SleepRelativeUsec(125);
265         cnt--;
266     }
267     if(cnt==0) {
268         debugOutput(DEBUG_LEVEL_VERBOSE, " Timeout waiting for the SP's to start dry-running\n");
269         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
270                 it != m_ReceiveProcessors.end();
271                 ++it ) {
272             debugOutput( DEBUG_LEVEL_VERBOSE, " %s SP %p has state %s\n",
273                 (*it)->getTypeString(), *it, (*it)->getStateString());
274         }
275         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
276                 it != m_TransmitProcessors.end();
277                 ++it ) {
278             debugOutput( DEBUG_LEVEL_VERBOSE, " %s SP %p has state %s\n",
279                 (*it)->getTypeString(), *it, (*it)->getStateString());
280         }
281         return false;
282     }
283     debugOutput( DEBUG_LEVEL_VERBOSE, " StreamProcessor streams dry-running...\n");
284     return true;
285 }
286
287 bool StreamProcessorManager::syncStartAll() {
288     if(m_SyncSource == NULL) return false;
289     // figure out when to get the SP's running.
290     // the xmit SP's should also know the base timestamp
291     // streams should be aligned here
292
293     // now find out how long we have to delay the wait operation such that
294     // the received frames will all be presented to the SP
295     debugOutput( DEBUG_LEVEL_VERBOSE, "Finding minimal sync delay...\n");
296     int max_of_min_delay = 0;
297     int min_delay = 0;
298     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
299             it != m_ReceiveProcessors.end();
300             ++it ) {
301         min_delay = (*it)->getMaxFrameLatency();
302         if(min_delay > max_of_min_delay) max_of_min_delay = min_delay;
303     }
304
305     // add some processing margin. This only shifts the time
306     // at which the buffer is transfer()'ed. This makes things somewhat
307     // more robust. It should be noted though that shifting the transfer
308     // time to a later time instant also causes the xmit buffer fill to be
309     // lower on average.
310     max_of_min_delay += STREAMPROCESSORMANAGER_SIGNAL_DELAY_TICKS;
311     debugOutput( DEBUG_LEVEL_VERBOSE, " sync delay = %d ticks (%03us %04uc %04ut)...\n",
312         max_of_min_delay,
313         (unsigned int)TICKS_TO_SECS(max_of_min_delay),
314         (unsigned int)TICKS_TO_CYCLES(max_of_min_delay),
315         (unsigned int)TICKS_TO_OFFSET(max_of_min_delay));
316     m_SyncSource->setSyncDelay(max_of_min_delay);
317
318     //STEP X: when we implement such a function, we can wait for a signal from the devices that they
319     //        have aquired lock
320     //debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for device(s) to indicate clock sync lock...\n");
321     //sleep(2); // FIXME: be smarter here
322
323     // make sure that we are dry-running long enough for the
324     // DLL to have a decent sync (FIXME: does the DLL get updated when dry-running)?
325     debugOutput( DEBUG_LEVEL_VERBOSE, "Waiting for sync...\n");
326     int nb_sync_runs=20;
327     int64_t time_till_next_period;
328     while(nb_sync_runs--) { // or while not sync-ed?
329         // check if we were woken up too soon
330         time_till_next_period = m_SyncSource->getTimeUntilNextPeriodSignalUsecs();
331         debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "waiting for %d usecs...\n", time_till_next_period);
332         if(time_till_next_period > 0) {
333             // wait for the period
334             SleepRelativeUsec(time_till_next_period);
335         }
336     }
337
338     debugOutput( DEBUG_LEVEL_VERBOSE, "Propagate sync info...\n");
339     // FIXME: in the SPM it would be nice to have system time instead of
340     //        1394 time
341
342     // we now should have decent sync info on the sync source
343     // determine a point in time where the system should start
344     // figure out where we are now
345     uint64_t time_of_first_sample = m_SyncSource->getTimeAtPeriod();
346     debugOutput( DEBUG_LEVEL_VERBOSE, " sync at TS=%011llu (%03us %04uc %04ut)...\n",
347         time_of_first_sample,
348         (unsigned int)TICKS_TO_SECS(time_of_first_sample),
349         (unsigned int)TICKS_TO_CYCLES(time_of_first_sample),
350         (unsigned int)TICKS_TO_OFFSET(time_of_first_sample));
351
352     // start wet-running in STREAMPROCESSORMANAGER_CYCLES_FOR_STARTUP cycles
353     // this is the time window we have to setup all SP's such that they
354     // can start wet-running correctly.
355     time_of_first_sample = addTicks(time_of_first_sample,
356                                     STREAMPROCESSORMANAGER_CYCLES_FOR_STARTUP * TICKS_PER_CYCLE);
357
358     debugOutput( DEBUG_LEVEL_VERBOSE, "  => first sample at TS=%011llu (%03us %04uc %04ut)...\n",
359         time_of_first_sample,
360         (unsigned int)TICKS_TO_SECS(time_of_first_sample),
361         (unsigned int)TICKS_TO_CYCLES(time_of_first_sample),
362         (unsigned int)TICKS_TO_OFFSET(time_of_first_sample));
363
364     // we should start wet-running the transmit SP's some cycles in advance
365     // such that we know it is wet-running when it should output its first sample
366     uint64_t time_to_start_xmit = substractTicks(time_of_first_sample,
367                                                  STREAMPROCESSORMANAGER_PRESTART_CYCLES_FOR_XMIT * TICKS_PER_CYCLE);
368
369     uint64_t time_to_start_recv = substractTicks(time_of_first_sample,
370                                                  STREAMPROCESSORMANAGER_PRESTART_CYCLES_FOR_RECV * TICKS_PER_CYCLE);
371     debugOutput( DEBUG_LEVEL_VERBOSE, "  => xmit starts at  TS=%011llu (%03us %04uc %04ut)...\n",
372         time_to_start_xmit,
373         (unsigned int)TICKS_TO_SECS(time_to_start_xmit),
374         (unsigned int)TICKS_TO_CYCLES(time_to_start_xmit),
375         (unsigned int)TICKS_TO_OFFSET(time_to_start_xmit));
376     debugOutput( DEBUG_LEVEL_VERBOSE, "  => recv starts at  TS=%011llu (%03us %04uc %04ut)...\n",
377         time_to_start_recv,
378         (unsigned int)TICKS_TO_SECS(time_to_start_recv),
379         (unsigned int)TICKS_TO_CYCLES(time_to_start_recv),
380         (unsigned int)TICKS_TO_OFFSET(time_to_start_recv));
381
382     // at this point the buffer head timestamp of the transmit buffers can be set
383     // this is the presentation time of the first sample in the buffer
384     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
385           it != m_TransmitProcessors.end();
386           ++it ) {
387         (*it)->setBufferHeadTimestamp(time_of_first_sample);
388     }
389
390     // STEP X: switch SP's over to the running state
391     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
392           it != m_ReceiveProcessors.end();
393           ++it ) {
394         if(!(*it)->scheduleStartRunning(time_to_start_recv)) {
395             debugError("%p->scheduleStartRunning(%11llu) failed\n", *it, time_to_start_recv);
396             return false;
397         }
398     }
399     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
400           it != m_TransmitProcessors.end();
401           ++it ) {
402         if(!(*it)->scheduleStartRunning(time_to_start_xmit)) {
403             debugError("%p->scheduleStartRunning(%11llu) failed\n", *it, time_to_start_xmit);
404             return false;
405         }
406     }
407     // wait for the syncsource to start running.
408     // that will block the waitForPeriod call until everyone has started (theoretically)
409     int cnt = STREAMPROCESSORMANAGER_CYCLES_FOR_STARTUP * 20; // by then it should have started
410     while (!m_SyncSource->isRunning() && cnt) {
411         SleepRelativeUsec(125);
412         cnt--;
413     }
414     if(cnt==0) {
415         debugOutput(DEBUG_LEVEL_VERBOSE, " Timeout waiting for the SyncSource to get started\n");
416         return false;
417     }
418
419     // now align the received streams
420     if(!alignReceivedStreams()) {
421         debugError("Could not align streams\n");
422         return false;
423     }
424     debugOutput( DEBUG_LEVEL_VERBOSE, " StreamProcessor streams running...\n");
425     return true;
426 }
427
428 bool
429 StreamProcessorManager::alignReceivedStreams()
430 {
431     if(m_SyncSource == NULL) return false;
432     debugOutput( DEBUG_LEVEL_VERBOSE, "Aligning received streams...\n");
433     unsigned int nb_sync_runs;
434     unsigned int nb_rcv_sp = m_ReceiveProcessors.size();
435     int64_t diff_between_streams[nb_rcv_sp];
436     int64_t diff;
437
438     unsigned int i;
439
440     unsigned int periods_per_align_try = (STREAMPROCESSORMANAGER_ALIGN_AVERAGE_TIME_MSEC * getNominalRate());
441     periods_per_align_try /= 1000;
442     periods_per_align_try /= getPeriodSize();
443     debugOutput( DEBUG_LEVEL_VERBOSE, " averaging over %u periods...\n", periods_per_align_try);
444
445     bool aligned = false;
446     int cnt = STREAMPROCESSORMANAGER_NB_ALIGN_TRIES;
447     while (!aligned && cnt--) {
448         nb_sync_runs = periods_per_align_try;
449         while(nb_sync_runs) {
450             debugOutput( DEBUG_LEVEL_VERY_VERBOSE, " check (%d)...\n", nb_sync_runs);
451             if(!waitForPeriod()) {
452                 debugWarning("xrun while aligning streams...\n");
453                 return false;
454             };
455
456             i = 0;
457             for ( i = 0; i < nb_rcv_sp; i++) {
458                 StreamProcessor *s = m_ReceiveProcessors.at(i);
459                 diff = diffTicks(m_SyncSource->getTimeAtPeriod(), s->getTimeAtPeriod());
460                 debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "  offset between SyncSP %p and SP %p is %lld ticks...\n",
461                     m_SyncSource, s, diff);
462                 if ( nb_sync_runs == periods_per_align_try ) {
463                     diff_between_streams[i] = diff;
464                 } else {
465                     diff_between_streams[i] += diff;
466                 }
467             }
468             if(!transferSilence()) {
469                 debugError("Could not transfer silence\n");
470                 return false;
471             }
472             nb_sync_runs--;
473         }
474         // calculate the average offsets
475         debugOutput( DEBUG_LEVEL_VERBOSE, " Average offsets:\n");
476         int diff_between_streams_frames[nb_rcv_sp];
477         aligned = true;
478         for ( i = 0; i < nb_rcv_sp; i++) {
479             StreamProcessor *s = m_ReceiveProcessors.at(i);
480
481             diff_between_streams[i] /= periods_per_align_try;
482             diff_between_streams_frames[i] = (int)roundf(diff_between_streams[i] / s->getTicksPerFrame());
483             debugOutput( DEBUG_LEVEL_VERBOSE, "   avg offset between SyncSP %p and SP %p is %lld ticks, %d frames...\n",
484                 m_SyncSource, s, diff_between_streams[i], diff_between_streams_frames[i]);
485
486             aligned &= (diff_between_streams_frames[i] == 0);
487
488             // reposition the stream
489             if(!s->shiftStream(diff_between_streams_frames[i])) {
490                 debugError("Could not shift SP %p %d frames\n", s, diff_between_streams_frames[i]);
491                 return false;
492             }
493         }
494         if (!aligned) {
495             debugOutput(DEBUG_LEVEL_VERBOSE, "Streams not aligned, doing new round...\n");
496         }
497     }
498     if (cnt == 0) {
499         debugError("Align failed\n");
500         return false;
501     }
502     return true;
503 }
504
505 bool StreamProcessorManager::start() {
506     debugOutput( DEBUG_LEVEL_VERBOSE, "Starting Processors...\n");
507
508     // put all SP's into dry-running state
509     if (!startDryRunning()) {
510         debugFatal("Could not put SP's in dry-running state\n");
511         return false;
512     }
513
514     // start all SP's synchonized
515     if (!syncStartAll()) {
516         debugFatal("Could not syncStartAll...\n");
517         return false;
518     }
519     return true;
520 }
521
522 bool StreamProcessorManager::stop() {
523     debugOutput( DEBUG_LEVEL_VERBOSE, "Stopping...\n");
524
525     debugOutput( DEBUG_LEVEL_VERBOSE, " scheduling stop for all SP's...\n");
526     // switch SP's over to the dry-running state
527     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
528           it != m_ReceiveProcessors.end();
529           ++it ) {
530         if((*it)->isRunning()) {
531             if(!(*it)->scheduleStopRunning(-1)) {
532                 debugError("%p->scheduleStopRunning(-1) failed\n", *it);
533                 return false;
534             }
535         }
536     }
537     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
538           it != m_TransmitProcessors.end();
539           ++it ) {
540         if((*it)->isRunning()) {
541             if(!(*it)->scheduleStopRunning(-1)) {
542                 debugError("%p->scheduleStopRunning(-1) failed\n", *it);
543                 return false;
544             }
545         }
546     }
547     // wait for the SP's to get into the dry-running state
548     int cnt = 2000;
549     bool ready = false;
550     while (!ready && cnt) {
551         ready = true;
552         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
553             it != m_ReceiveProcessors.end();
554             ++it ) {
555             ready &= ((*it)->isDryRunning() || (*it)->isStopped());
556         }
557         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
558             it != m_TransmitProcessors.end();
559             ++it ) {
560             ready &= ((*it)->isDryRunning() || (*it)->isStopped());
561         }
562         SleepRelativeUsec(125);
563         cnt--;
564     }
565     if(cnt==0) {
566         debugOutput(DEBUG_LEVEL_VERBOSE, " Timeout waiting for the SP's to start dry-running\n");
567         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
568             it != m_ReceiveProcessors.end();
569             ++it ) {
570             (*it)->dumpInfo();
571         }
572         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
573             it != m_TransmitProcessors.end();
574             ++it ) {
575             (*it)->dumpInfo();
576         }
577         return false;
578     }
579
580     // switch SP's over to the stopped state
581     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
582           it != m_ReceiveProcessors.end();
583           ++it ) {
584         if(!(*it)->scheduleStopDryRunning(-1)) {
585             debugError("%p->scheduleStopDryRunning(-1) failed\n", *it);
586             return false;
587         }
588     }
589     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
590           it != m_TransmitProcessors.end();
591           ++it ) {
592         if(!(*it)->scheduleStopDryRunning(-1)) {
593             debugError("%p->scheduleStopDryRunning(-1) failed\n", *it);
594             return false;
595         }
596     }
597     // wait for the SP's to get into the running state
598     cnt = 200;
599     ready = false;
600     while (!ready && cnt) {
601         ready = true;
602         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
603             it != m_ReceiveProcessors.end();
604             ++it ) {
605             ready &= (*it)->isStopped();
606         }
607         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
608             it != m_TransmitProcessors.end();
609             ++it ) {
610             ready &= (*it)->isStopped();
611         }
612         SleepRelativeUsec(125);
613         cnt--;
614     }
615     if(cnt==0) {
616         debugOutput(DEBUG_LEVEL_VERBOSE, " Timeout waiting for the SP's to stop\n");
617         return false;
618     }
619     return true;
620 }
621
622 /**
623  * Called upon Xrun events. This brings all StreamProcessors back
624  * into their starting state, and then carries on streaming. This should
625  * have the same effect as restarting the whole thing.
626  *
627  * @return true if successful, false otherwise
628  */
629 bool StreamProcessorManager::handleXrun() {
630
631     debugOutput( DEBUG_LEVEL_VERBOSE, "Handling Xrun ...\n");
632
633     dumpInfo();
634
635     /*
636      * Reset means:
637      * 1) Disabling the SP's, so that they don't process any packets
638      *    note: the isomanager does keep on delivering/requesting them
639      * 2) Bringing all buffers & streamprocessors into a know state
640      *    - Clear all capture buffers
641      *    - Put nb_periods*period_size of null frames into the playback buffers
642      * 3) Re-enable the SP's
643      */
644
645     // put all SP's back into dry-running state
646     if (!startDryRunning()) {
647         debugFatal("Could not put SP's in dry-running state\n");
648         return false;
649     }
650
651     debugOutput( DEBUG_LEVEL_VERBOSE, "Restarting StreamProcessors...\n");
652     // start all SP's synchonized
653     if (!syncStartAll()) {
654         debugFatal("Could not syncStartAll...\n");
655         return false;
656     }
657
658     debugOutput( DEBUG_LEVEL_VERBOSE, "Xrun handled...\n");
659
660     return true;
661 }
662
663 /**
664  * @brief Waits until the next period of samples is ready
665  *
666  * This function does not return until a full period of samples is (or should be)
667  * ready to be transferred.
668  *
669  * @return true if the period is ready, false if an xrun occurred
670  */
671 bool StreamProcessorManager::waitForPeriod() {
672     if(m_SyncSource == NULL) return false;
673     bool xrun_occurred = false;
674     bool period_not_ready = true;
675
676     while(period_not_ready) {
677         debugOutput( DEBUG_LEVEL_VERBOSE, "waiting for period (%d frames in buffer)...\n", m_SyncSource->getBufferFill());
678         if(!m_SyncSource->waitForSignal()) {
679             debugError("Error waiting for signal\n");
680             return false;
681         }
682
683         unsigned int bufferfill = m_SyncSource->getBufferFill();
684         period_not_ready = bufferfill < m_period;
685
686 #ifdef DEBUG
687         if(period_not_ready) {
688             debugOutput(DEBUG_LEVEL_VERBOSE, "period is not ready (bufferfill: %u)\n", bufferfill);
689         } else {
690             debugOutput(DEBUG_LEVEL_VERBOSE, "period is ready (bufferfill: %u)\n", bufferfill);
691         }
692 #endif
693
694         // check for underruns on the ISO side,
695         // those should make us bail out of the wait loop
696         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
697             it != m_ReceiveProcessors.end();
698             ++it ) {
699             // a xrun has occurred on the Iso side
700             xrun_occurred |= (*it)->xrunOccurred();
701         }
702         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
703             it != m_TransmitProcessors.end();
704             ++it ) {
705             // a xrun has occurred on the Iso side
706             xrun_occurred |= (*it)->xrunOccurred();
707         }
708         if(xrun_occurred) break;
709         // FIXME: make sure we also exit this loop when something else happens (e.g. signal, iso error)
710     }
711
712     // we save the 'ideal' time of the transfer at this point,
713     // because we can have interleaved read - process - write
714     // cycles making that we modify a receiving stream's buffer
715     // before we get to writing.
716     // NOTE: before waitForPeriod() is called again, both the transmit
717     //       and the receive processors should have done their transfer.
718     m_time_of_transfer = m_SyncSource->getTimeAtPeriod();
719     debugOutput( DEBUG_LEVEL_VERBOSE, "transfer at %llu ticks...\n",
720         m_time_of_transfer);
721
722     // this is to notify the client of the delay that we introduced by waiting
723     m_delayed_usecs = - m_SyncSource->getTimeUntilNextPeriodSignalUsecs();
724     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "delayed for %d usecs...\n", m_delayed_usecs);
725
726 #ifdef DEBUG
727     int rcv_bf=0, xmt_bf=0;
728     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
729         it != m_ReceiveProcessors.end();
730         ++it ) {
731         rcv_bf = (*it)->getBufferFill();
732     }
733     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
734         it != m_TransmitProcessors.end();
735         ++it ) {
736         xmt_bf = (*it)->getBufferFill();
737     }
738     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "XF at %011llu ticks, RBF=%d, XBF=%d, SUM=%d...\n",
739         m_time_of_transfer, rcv_bf, xmt_bf, rcv_bf+xmt_bf);
740
741     // check if xruns occurred on the Iso side.
742     // also check if xruns will occur should we transfer() now
743     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
744           it != m_ReceiveProcessors.end();
745           ++it ) {
746
747         if ((*it)->xrunOccurred()) {
748             debugWarning("Xrun on RECV SP %p due to ISO side xrun\n",*it);
749             (*it)->dumpInfo();
750         }
751         if (!((*it)->canClientTransferFrames(m_period))) {
752             debugWarning("Xrun on RECV SP %p due to buffer side xrun\n",*it);
753             (*it)->dumpInfo();
754         }
755     }
756     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
757           it != m_TransmitProcessors.end();
758           ++it ) {
759         if ((*it)->xrunOccurred()) {
760             debugWarning("Xrun on XMIT SP %p due to ISO side xrun\n",*it);
761         }
762         if (!((*it)->canClientTransferFrames(m_period))) {
763             debugWarning("Xrun on XMIT SP %p due to buffer side xrun\n",*it);
764         }
765     }
766 #endif
767
768     m_nbperiods++;
769     // now we can signal the client that we are (should be) ready
770     return !xrun_occurred;
771 }
772
773 /**
774  * @brief Transfer one period of frames for both receive and transmit StreamProcessors
775  *
776  * Transfers one period of frames from the client side to the Iso side and vice versa.
777  *
778  * @return true if successful, false otherwise (indicates xrun).
779  */
780 bool StreamProcessorManager::transfer() {
781     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Transferring period...\n");
782     bool retval=true;
783     retval &= transfer(StreamProcessor::ePT_Receive);
784     retval &= transfer(StreamProcessor::ePT_Transmit);
785     return retval;
786 }
787
788 /**
789  * @brief Transfer one period of frames for either the receive or transmit StreamProcessors
790  *
791  * Transfers one period of frames from the client side to the Iso side or vice versa.
792  *
793  * @param t The processor type to tranfer for (receive or transmit)
794  * @return true if successful, false otherwise (indicates xrun).
795  */
796 bool StreamProcessorManager::transfer(enum StreamProcessor::eProcessorType t) {
797     if(m_SyncSource == NULL) return false;
798     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "transfer(%d) at TS=%011llu (%03us %04uc %04ut)...\n",
799         t, m_time_of_transfer,
800         (unsigned int)TICKS_TO_SECS(m_time_of_transfer),
801         (unsigned int)TICKS_TO_CYCLES(m_time_of_transfer),
802         (unsigned int)TICKS_TO_OFFSET(m_time_of_transfer));
803
804     bool retval = true;
805     // a static cast could make sure that there is no performance
806     // penalty for the virtual functions (to be checked)
807     if (t==StreamProcessor::ePT_Receive) {
808         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
809                 it != m_ReceiveProcessors.end();
810                 ++it ) {
811             if(!(*it)->getFrames(m_period, m_time_of_transfer)) {
812                     debugWarning("could not getFrames(%u, %11llu) from stream processor (%p)\n",
813                             m_period, m_time_of_transfer,*it);
814                 retval &= false; // buffer underrun
815             }
816         }
817     } else {
818         // FIXME: in the SPM it would be nice to have system time instead of
819         //        1394 time
820         float rate = m_SyncSource->getTicksPerFrame();
821         int64_t one_ringbuffer_in_ticks=(int64_t)(((float)(m_nb_buffers * m_period)) * rate);
822
823         // the data we are putting into the buffer is intended to be transmitted
824         // one ringbuffer size after it has been received
825         int64_t transmit_timestamp = addTicks(m_time_of_transfer, one_ringbuffer_in_ticks);
826
827         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
828                 it != m_TransmitProcessors.end();
829                 ++it ) {
830             // FIXME: in the SPM it would be nice to have system time instead of
831             //        1394 time
832             if(!(*it)->putFrames(m_period, transmit_timestamp)) {
833                 debugWarning("could not putFrames(%u,%llu) to stream processor (%p)\n",
834                         m_period, transmit_timestamp, *it);
835                 retval &= false; // buffer underrun
836             }
837         }
838     }
839     return retval;
840 }
841
842 /**
843  * @brief Transfer one period of silence for both receive and transmit StreamProcessors
844  *
845  * Transfers one period of silence to the Iso side for transmit SP's
846  * or dump one period of frames for receive SP's
847  *
848  * @return true if successful, false otherwise (indicates xrun).
849  */
850 bool StreamProcessorManager::transferSilence() {
851     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "Transferring silent period...\n");
852     bool retval=true;
853     retval &= transferSilence(StreamProcessor::ePT_Receive);
854     retval &= transferSilence(StreamProcessor::ePT_Transmit);
855     return retval;
856 }
857
858 /**
859  * @brief Transfer one period of silence for either the receive or transmit StreamProcessors
860  *
861  * Transfers one period of silence to the Iso side for transmit SP's
862  * or dump one period of frames for receive SP's
863  *
864  * @param t The processor type to tranfer for (receive or transmit)
865  * @return true if successful, false otherwise (indicates xrun).
866  */
867 bool StreamProcessorManager::transferSilence(enum StreamProcessor::eProcessorType t) {
868     if(m_SyncSource == NULL) return false;
869     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "transferSilence(%d) at TS=%011llu (%03us %04uc %04ut)...\n",
870         t, m_time_of_transfer,
871         (unsigned int)TICKS_TO_SECS(m_time_of_transfer),
872         (unsigned int)TICKS_TO_CYCLES(m_time_of_transfer),
873         (unsigned int)TICKS_TO_OFFSET(m_time_of_transfer));
874
875     bool retval = true;
876     // a static cast could make sure that there is no performance
877     // penalty for the virtual functions (to be checked)
878     if (t==StreamProcessor::ePT_Receive) {
879         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
880                 it != m_ReceiveProcessors.end();
881                 ++it ) {
882             if(!(*it)->dropFrames(m_period, m_time_of_transfer)) {
883                     debugWarning("could not dropFrames(%u, %11llu) from stream processor (%p)\n",
884                             m_period, m_time_of_transfer,*it);
885                 retval &= false; // buffer underrun
886             }
887         }
888     } else {
889         // FIXME: in the SPM it would be nice to have system time instead of
890         //        1394 time
891         float rate = m_SyncSource->getTicksPerFrame();
892         int64_t one_ringbuffer_in_ticks=(int64_t)(((float)(m_nb_buffers * m_period)) * rate);
893
894         // the data we are putting into the buffer is intended to be transmitted
895         // one ringbuffer size after it has been received
896         int64_t transmit_timestamp = addTicks(m_time_of_transfer, one_ringbuffer_in_ticks);
897
898         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
899                 it != m_TransmitProcessors.end();
900                 ++it ) {
901             // FIXME: in the SPM it would be nice to have system time instead of
902             //        1394 time
903             if(!(*it)->putSilenceFrames(m_period, transmit_timestamp)) {
904                 debugWarning("could not putSilenceFrames(%u,%llu) to stream processor (%p)\n",
905                         m_period, transmit_timestamp, *it);
906                 retval &= false; // buffer underrun
907             }
908         }
909     }
910     return retval;
911 }
912
913 void StreamProcessorManager::dumpInfo() {
914     debugOutputShort( DEBUG_LEVEL_NORMAL, "----------------------------------------------------\n");
915     debugOutputShort( DEBUG_LEVEL_NORMAL, "Dumping StreamProcessorManager information...\n");
916     debugOutputShort( DEBUG_LEVEL_NORMAL, "Period count: %6d\n", m_nbperiods);
917
918     debugOutputShort( DEBUG_LEVEL_NORMAL, " Receive processors...\n");
919     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
920         it != m_ReceiveProcessors.end();
921         ++it ) {
922         (*it)->dumpInfo();
923     }
924
925     debugOutputShort( DEBUG_LEVEL_NORMAL, " Transmit processors...\n");
926     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
927         it != m_TransmitProcessors.end();
928         ++it ) {
929         (*it)->dumpInfo();
930     }
931
932     debugOutputShort( DEBUG_LEVEL_NORMAL, "----------------------------------------------------\n");
933
934 }
935
936 void StreamProcessorManager::setVerboseLevel(int l) {
937     setDebugLevel(l);
938
939     debugOutput( DEBUG_LEVEL_VERBOSE, " Receive processors...\n");
940     for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
941         it != m_ReceiveProcessors.end();
942         ++it ) {
943         (*it)->setVerboseLevel(l);
944     }
945
946     debugOutput( DEBUG_LEVEL_VERBOSE, " Transmit processors...\n");
947     for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
948         it != m_TransmitProcessors.end();
949         ++it ) {
950         (*it)->setVerboseLevel(l);
951     }
952 }
953
954
955 int StreamProcessorManager::getPortCount(enum Port::E_PortType type, enum Port::E_Direction direction) {
956     int count=0;
957
958     if (direction == Port::E_Capture) {
959         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
960             it != m_ReceiveProcessors.end();
961             ++it ) {
962             count += (*it)->getPortCount(type);
963         }
964     } else {
965         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
966             it != m_TransmitProcessors.end();
967             ++it ) {
968             count += (*it)->getPortCount(type);
969         }
970     }
971     return count;
972 }
973
974 int StreamProcessorManager::getPortCount(enum Port::E_Direction direction) {
975     int count=0;
976
977     if (direction == Port::E_Capture) {
978         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
979             it != m_ReceiveProcessors.end();
980             ++it ) {
981             count += (*it)->getPortCount();
982         }
983     } else {
984         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
985             it != m_TransmitProcessors.end();
986             ++it ) {
987             count += (*it)->getPortCount();
988         }
989     }
990     return count;
991 }
992
993 // TODO: implement a port map here, instead of the loop
994
995 Port* StreamProcessorManager::getPortByIndex(int idx, enum Port::E_Direction direction) {
996     int count=0;
997     int prevcount=0;
998
999     if (direction == Port::E_Capture) {
1000         for ( StreamProcessorVectorIterator it = m_ReceiveProcessors.begin();
1001             it != m_ReceiveProcessors.end();
1002             ++it ) {
1003             count += (*it)->getPortCount();
1004             if (count > idx) {
1005                 return (*it)->getPortAtIdx(idx-prevcount);
1006             }
1007             prevcount=count;
1008         }
1009     } else {
1010         for ( StreamProcessorVectorIterator it = m_TransmitProcessors.begin();
1011             it != m_TransmitProcessors.end();
1012             ++it ) {
1013             count += (*it)->getPortCount();
1014             if (count > idx) {
1015                 return (*it)->getPortAtIdx(idx-prevcount);
1016             }
1017             prevcount=count;
1018         }
1019     }
1020     return NULL;
1021 }
1022
1023 bool StreamProcessorManager::setThreadParameters(bool rt, int priority) {
1024     m_thread_realtime=rt;
1025     m_thread_priority=priority;
1026     return true;
1027 }
1028
1029
1030 } // end of namespace
Note: See TracBrowser for help on using the browser.