root/trunk/libffado/src/libstreaming/generic/StreamProcessor.cpp

Revision 748, 57.5 kB (checked in by ppalmers, 15 years ago)

try to reorganize things such that less information is duplicated

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
24 #include "StreamProcessor.h"
25 #include "libieee1394/cycletimer.h"
26 #include "../StreamProcessorManager.h"
27
28 #include "libutil/Atomic.h"
29
30 #include <assert.h>
31 #include <math.h>
32
33 namespace Streaming {
34
35 IMPL_DEBUG_MODULE( StreamProcessor, StreamProcessor, DEBUG_LEVEL_VERBOSE );
36
37 StreamProcessor::StreamProcessor(FFADODevice &parent, enum eProcessorType type)
38     : m_parent( parent )
39     , m_processor_type ( type )
40     , m_channel( -1 )
41     , m_handler( NULL )
42     , m_state( ePS_Created )
43     , m_next_state( ePS_Invalid )
44     , m_cycle_to_switch_state( 0 )
45     , m_scratch_buffer( NULL )
46     , m_scratch_buffer_size_bytes( 0 )
47     , m_manager( NULL )
48     , m_ticks_per_frame( 0 )
49     , m_last_cycle( -1 )
50     , m_sync_delay( 0 )
51     , m_in_xrun( false )
52     , m_last_timestamp(0)
53     , m_last_timestamp2(0)
54     , m_dropped(0)
55 {
56     // create the timestamped buffer and register ourselves as its client
57     m_data_buffer = new Util::TimestampedBuffer(this);
58 }
59
60 StreamProcessor::~StreamProcessor() {
61     if (m_data_buffer) delete m_data_buffer;
62     if (m_scratch_buffer) delete[] m_scratch_buffer;
63 }
64
65 uint64_t StreamProcessor::getTimeNow() {
66     return m_parent.get1394Service().getCycleTimerTicks();
67 }
68
69 int StreamProcessor::getMaxFrameLatency() {
70     if (getType() == ePT_Receive) {
71         return (int)(m_handler->getWakeupInterval() * TICKS_PER_CYCLE);
72     } else {
73         return (int)(m_handler->getWakeupInterval() * TICKS_PER_CYCLE);
74     }
75 }
76
77 unsigned int
78 StreamProcessor::getNominalPacketsNeeded(unsigned int nframes)
79 {
80     unsigned int nominal_frames_per_second = m_manager->getNominalRate();
81     uint64_t nominal_ticks_per_frame = TICKS_PER_SECOND / nominal_frames_per_second;
82     uint64_t nominal_ticks = nominal_ticks_per_frame * nframes;
83     uint64_t nominal_packets = nominal_ticks / TICKS_PER_CYCLE;
84     return nominal_packets;
85 }
86
87 unsigned int
88 StreamProcessor::getPacketsPerPeriod()
89 {
90     return getNominalPacketsNeeded(m_manager->getPeriodSize());
91 }
92
93
94 /***********************************************
95  * Buffer management and manipulation          *
96  ***********************************************/
97 int StreamProcessor::getBufferFill() {
98     return m_data_buffer->getBufferFill();
99 }
100
101 int64_t
102 StreamProcessor::getTimeUntilNextPeriodSignalUsecs()
103 {
104     uint64_t time_at_period=getTimeAtPeriod();
105
106     // we delay the period signal with the sync delay
107     // this makes that the period signals lag a little compared to reality
108     // ISO buffering causes the packets to be received at max
109     // m_handler->getWakeupInterval() later than the time they were received.
110     // hence their payload is available this amount of time later. However, the
111     // period boundary is predicted based upon earlier samples, and therefore can
112     // pass before these packets are processed. Adding this extra term makes that
113     // the period boundary is signalled later
114     time_at_period = addTicks(time_at_period, m_manager->getSyncSource().getSyncDelay());
115
116     uint64_t cycle_timer=m_parent.get1394Service().getCycleTimerTicks();
117
118     // calculate the time until the next period
119     int32_t until_next=diffTicks(time_at_period,cycle_timer);
120
121     debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "=> TAP=%11llu, CTR=%11llu, UTN=%11ld\n",
122         time_at_period, cycle_timer, until_next
123         );
124
125     // now convert to usecs
126     // don't use the mapping function because it only works
127     // for absolute times, not the relative time we are
128     // using here (which can also be negative).
129     return (int64_t)(((float)until_next) / TICKS_PER_USEC);
130 }
131
132 void
133 StreamProcessor::setSyncDelay(int d) {
134     debugOutput(DEBUG_LEVEL_VERBOSE, "Setting SP %p SyncDelay to %d ticks\n", this, d);
135     m_sync_delay = d;
136 }
137
138 uint64_t
139 StreamProcessor::getTimeAtPeriodUsecs()
140 {
141     return (uint64_t)((float)getTimeAtPeriod() * TICKS_PER_USEC);
142 }
143
144 uint64_t
145 StreamProcessor::getTimeAtPeriod()
146 {
147     if (getType() == ePT_Receive) {
148         ffado_timestamp_t next_period_boundary=m_data_buffer->getTimestampFromHead(m_manager->getPeriodSize());
149    
150         #ifdef DEBUG
151         ffado_timestamp_t ts;
152         signed int fc;
153         m_data_buffer->getBufferTailTimestamp(&ts,&fc);
154    
155         debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "=> NPD="TIMESTAMP_FORMAT_SPEC", LTS="TIMESTAMP_FORMAT_SPEC", FC=%5u, TPF=%f\n",
156             next_period_boundary, ts, fc, getTicksPerFrame()
157             );
158         #endif
159         return (uint64_t)next_period_boundary;
160     } else {
161         ffado_timestamp_t next_period_boundary=m_data_buffer->getTimestampFromTail((m_manager->getNbBuffers()-1) * m_manager->getPeriodSize());
162    
163         #ifdef DEBUG
164         ffado_timestamp_t ts;
165         signed int fc;
166         m_data_buffer->getBufferTailTimestamp(&ts,&fc);
167    
168         debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "=> NPD="TIMESTAMP_FORMAT_SPEC", LTS="TIMESTAMP_FORMAT_SPEC", FC=%5u, TPF=%f\n",
169             next_period_boundary, ts, fc, getTicksPerFrame()
170             );
171         #endif
172         return (uint64_t)next_period_boundary;
173     }
174 }
175
176 float
177 StreamProcessor::getTicksPerFrame()
178 {
179     assert(m_data_buffer != NULL);
180     return m_data_buffer->getRate();
181 }
182
183 bool
184 StreamProcessor::canClientTransferFrames(unsigned int nbframes)
185 {
186     bool can_transfer;
187     unsigned int fc = m_data_buffer->getFrameCounter();
188     if (getType() == ePT_Receive) {
189         can_transfer = (fc >= nbframes);
190     } else {
191         // there has to be enough space to put the frames in
192         can_transfer = m_data_buffer->getBufferSize() - fc > nbframes;
193         // or the buffer is transparent
194         can_transfer |= m_data_buffer->isTransparent();
195     }
196    
197     #ifdef DEBUG
198     if (!can_transfer) {
199         debugWarning("(%p, %s) cannot transfer since fc == %u, nbframes == %u\n",
200             this, ePTToString(getType()), fc, nbframes);
201     }
202     #endif
203    
204     return can_transfer;
205 }
206
207 /***********************************************
208  * I/O API                                     *
209  ***********************************************/
210
211 // Packet transfer API
212 enum raw1394_iso_disposition
213 StreamProcessor::putPacket(unsigned char *data, unsigned int length,
214                            unsigned char channel, unsigned char tag, unsigned char sy,
215                            unsigned int cycle, unsigned int dropped) {
216     if(m_last_cycle == -1) {
217         debugOutput(DEBUG_LEVEL_VERBOSE, "Handler for %s SP %p is alive (cycle = %u)\n", getTypeString(), this, cycle);
218     }
219
220     int dropped_cycles = 0;
221     if (m_last_cycle != (int)cycle && m_last_cycle != -1) {
222         dropped_cycles = diffCycles(cycle, m_last_cycle) - 1;
223         if (dropped_cycles < 0) debugWarning("(%p) dropped < 1 (%d)\n", this, dropped_cycles);
224         if (dropped_cycles > 0) {
225             debugWarning("(%p) dropped %d packets on cycle %u\n", this, dropped_cycles, cycle);
226             m_dropped += dropped_cycles;
227         }
228     }
229     m_last_cycle = cycle;
230
231     // bypass based upon state
232     if (m_state == ePS_Invalid) {
233         debugError("Should not have state %s\n", ePSToString(m_state) );
234         return RAW1394_ISO_ERROR;
235     }
236     if (m_state == ePS_Created) {
237         return RAW1394_ISO_DEFER;
238     }
239
240     // store the previous timestamp
241     m_last_timestamp2 = m_last_timestamp;
242
243     // NOTE: synchronized switching is restricted to a 0.5 sec span (4000 cycles)
244     //       it happens on the first 'good' cycle for the wait condition
245     //       or on the first received cycle that is received afterwards (might be a problem)
246
247     // check whether we are waiting for a stream to be disabled
248     if(m_state == ePS_WaitingForStreamDisable) {
249         // we then check whether we have to switch on this cycle
250         if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
251             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning\n");
252             m_next_state = ePS_DryRunning;
253             if (!updateState()) { // we are allowed to change the state directly
254                 debugError("Could not update state!\n");
255                 return RAW1394_ISO_ERROR;
256             }
257         } else {
258             // not time to disable yet
259         }
260         // the received data can be discarded while waiting for the stream
261         // to be disabled
262         return RAW1394_ISO_OK;
263     }
264
265     // check whether we are waiting for a stream to be enabled
266     else if(m_state == ePS_WaitingForStreamEnable) {
267         // we then check whether we have to switch on this cycle
268         if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
269             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to Running\n");
270             m_next_state = ePS_Running;
271             if (!updateState()) { // we are allowed to change the state directly
272                 debugError("Could not update state!\n");
273                 return RAW1394_ISO_ERROR;
274             }
275         } else {
276             // not time to enable yet
277         }
278         // we are dryRunning hence data should be processed in any case
279     }
280
281     // check the packet header
282     enum eChildReturnValue result = processPacketHeader(data, length, channel, tag, sy, cycle, dropped_cycles);
283     if (result == eCRV_OK) {
284         debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "RECV: CY=%04u TS=%011llu\n",
285                 cycle, m_last_timestamp);
286         // update some accounting
287         m_last_good_cycle = cycle;
288         m_last_dropped = dropped_cycles;
289
290         // check whether we are waiting for a stream to startup
291         // this requires that the packet is good
292         if(m_state == ePS_WaitingForStream) {
293             // since we have a packet with an OK header,
294             // we can indicate that the stream started up
295
296             // we then check whether we have to switch on this cycle
297             if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
298                 debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning due to good packet\n");
299                 // hence go to the dryRunning state
300                 m_next_state = ePS_DryRunning;
301                 if (!updateState()) { // we are allowed to change the state directly
302                     debugError("Could not update state!\n");
303                     return RAW1394_ISO_ERROR;
304                 }
305             } else {
306                 // not time (yet) to switch state
307             }
308             // in both cases we don't want to process the data
309             return RAW1394_ISO_OK;
310         }
311
312         // check whether a state change has been requested
313         // note that only the wait state changes are synchronized with the cycles
314         else if(m_state != m_next_state) {
315             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
316                                              ePSToString(m_state), ePSToString(m_next_state));
317             // execute the requested change
318             if (!updateState()) { // we are allowed to change the state directly
319                 debugError("Could not update state!\n");
320                 return RAW1394_ISO_ERROR;
321             }
322         }
323
324         // handle dropped cycles
325         if(dropped_cycles) {
326             // they represent a discontinuity in the timestamps, and hence are
327             // to be dealt with
328             debugWarning("(%p) Correcting timestamp for dropped cycles, discarding packet...\n", this);
329             m_data_buffer->setBufferTailTimestamp(m_last_timestamp);
330             if (m_state == ePS_Running) {
331                 // this is an xrun situation
332                 m_in_xrun = true;
333                 debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to WaitingForStreamDisable due to dropped packet xrun\n");
334                 m_cycle_to_switch_state = cycle + 1; // switch in the next cycle
335                 m_next_state = ePS_WaitingForStreamDisable;
336                 // execute the requested change
337                 if (!updateState()) { // we are allowed to change the state directly
338                     debugError("Could not update state!\n");
339                     return RAW1394_ISO_ERROR;
340                 }
341                 return RAW1394_ISO_DEFER;
342             }
343         }
344
345         // for all states that reach this we are allowed to
346         // do protocol specific data reception
347         enum eChildReturnValue result2 = processPacketData(data, length, channel, tag, sy, cycle, dropped_cycles);
348
349         // if an xrun occured, switch to the dryRunning state and
350         // allow for the xrun to be picked up
351         if (result2 == eCRV_XRun) {
352             m_in_xrun = true;
353             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to WaitingForStreamDisable due to data xrun\n");
354             m_cycle_to_switch_state = cycle+1; // switch in the next cycle
355             m_next_state = ePS_WaitingForStreamDisable;
356             // execute the requested change
357             if (!updateState()) { // we are allowed to change the state directly
358                 debugError("Could not update state!\n");
359                 return RAW1394_ISO_ERROR;
360             }
361             return RAW1394_ISO_DEFER;
362         } else if(result2 == eCRV_OK) {
363             // no problem here
364             return RAW1394_ISO_OK;
365         } else {
366             debugError("Invalid response\n");
367             return RAW1394_ISO_ERROR;
368         }
369     } else if(result == eCRV_Invalid) {
370         // apparently we don't have to do anything when the packets are not valid
371         return RAW1394_ISO_OK;
372     } else {
373         debugError("Invalid response\n");
374         return RAW1394_ISO_ERROR;
375     }
376     debugError("reached the unreachable\n");
377     return RAW1394_ISO_ERROR;
378 }
379
380 enum raw1394_iso_disposition
381 StreamProcessor::getPacket(unsigned char *data, unsigned int *length,
382                            unsigned char *tag, unsigned char *sy,
383                            int cycle, unsigned int dropped, unsigned int max_length) {
384     if (cycle<0) {
385         *tag = 0;
386         *sy = 0;
387         *length = 0;
388         return RAW1394_ISO_OK;
389     }
390
391     if(m_last_cycle == -1) {
392         debugOutput(DEBUG_LEVEL_VERBOSE, "Handler for %s SP %p is alive (cycle = %d)\n", getTypeString(), this, cycle);
393     }
394
395     int dropped_cycles = 0;
396     if (m_last_cycle != cycle && m_last_cycle != -1) {
397         dropped_cycles = diffCycles(cycle, m_last_cycle) - 1;
398         if (dropped_cycles < 0) debugWarning("(%p) dropped < 1 (%d)\n", this, dropped_cycles);
399         if (dropped_cycles > 0) {
400             debugWarning("(%p) dropped %d packets on cycle %u (last_cycle=%u, dropped=%d)\n", this, dropped_cycles, cycle, m_last_cycle, dropped);
401             m_dropped += dropped_cycles;
402         }
403     }
404     if (cycle >= 0) {
405         m_last_cycle = cycle;
406     }
407
408     // bypass based upon state
409     if (m_state == ePS_Invalid) {
410         debugError("Should not have state %s\n", ePSToString(m_state) );
411         return RAW1394_ISO_ERROR;
412     }
413     if (m_state == ePS_Created) {
414         *tag = 0;
415         *sy = 0;
416         *length = 0;
417         return RAW1394_ISO_DEFER;
418     }
419
420     // normal processing
421     // note that we can't use getCycleTimer directly here,
422     // because packets are queued in advance. This means that
423     // we the packet we are constructing will be sent out
424     // on 'cycle', not 'now'.
425     unsigned int ctr = m_parent.get1394Service().getCycleTimer();
426     int now_cycles = (int)CYCLE_TIMER_GET_CYCLES(ctr);
427
428     // the difference between the cycle this
429     // packet is intended for and 'now'
430     int cycle_diff = diffCycles(cycle, now_cycles);
431
432     #ifdef DEBUG
433     if(cycle_diff < 0) {
434         debugWarning("Requesting packet for cycle %04d which is in the past (now=%04dcy)\n",
435             cycle, now_cycles);
436     }
437     #endif
438
439     // store the previous timestamp
440     m_last_timestamp2 = m_last_timestamp;
441
442     // NOTE: synchronized switching is restricted to a 0.5 sec span (4000 cycles)
443     //       it happens on the first 'good' cycle for the wait condition
444     //       or on the first received cycle that is received afterwards (might be a problem)
445
446     // check whether we are waiting for a stream to be disabled
447     if(m_state == ePS_WaitingForStreamDisable) {
448         // we then check whether we have to switch on this cycle
449         if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
450             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning\n");
451             m_next_state = ePS_DryRunning;
452             if (!updateState()) { // we are allowed to change the state directly
453                 debugError("Could not update state!\n");
454                 return RAW1394_ISO_ERROR;
455             }
456         } else {
457             // not time to disable yet
458         }
459     }
460     // check whether we are waiting for a stream to be enabled
461     else if(m_state == ePS_WaitingForStreamEnable) {
462         // we then check whether we have to switch on this cycle
463         if (diffCycles(cycle, m_cycle_to_switch_state) >= 0) {
464             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to Running\n");
465             m_next_state = ePS_Running;
466             if (!updateState()) { // we are allowed to change the state directly
467                 debugError("Could not update state!\n");
468                 return RAW1394_ISO_ERROR;
469             }
470         } else {
471             // not time to enable yet
472         }
473         // we are dryRunning hence data should be processed in any case
474     }
475     // check whether we are waiting for a stream to startup
476     else if(m_state == ePS_WaitingForStream) {
477         // as long as the cycle parameter is not in sync with
478         // the current time, the stream is considered not
479         // to be 'running'
480         // we then check whether we have to switch on this cycle
481         if ((cycle_diff >= 0) && (diffCycles(cycle, m_cycle_to_switch_state) >= 0)) {
482             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to DryRunning due to good packet\n");
483             // hence go to the dryRunning state
484             m_next_state = ePS_DryRunning;
485             if (!updateState()) { // we are allowed to change the state directly
486                 debugError("Could not update state!\n");
487                 return RAW1394_ISO_ERROR;
488             }
489         } else {
490             // not time (yet) to switch state
491         }
492     }
493     else if(m_state == ePS_Running) {
494         // check the packet header
495         enum eChildReturnValue result = generatePacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length);
496         if (result == eCRV_Packet) {
497             debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "XMIT: CY=%04u TS=%011llu\n",
498                     cycle, m_last_timestamp);
499             // update some accounting
500             m_last_good_cycle = cycle;
501             m_last_dropped = dropped_cycles;
502
503             // check whether a state change has been requested
504             // note that only the wait state changes are synchronized with the cycles
505             if(m_state != m_next_state) {
506                 debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
507                                                 ePSToString(m_state), ePSToString(m_next_state));
508                 // execute the requested change
509                 if (!updateState()) { // we are allowed to change the state directly
510                     debugError("Could not update state!\n");
511                     return RAW1394_ISO_ERROR;
512                 }
513             }
514
515             enum eChildReturnValue result2 = generatePacketData(data, length, tag, sy, cycle, dropped_cycles, max_length);
516             // if an xrun occured, switch to the dryRunning state and
517             // allow for the xrun to be picked up
518             if (result2 == eCRV_XRun) {
519                 debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to WaitingForStreamDisable due to data xrun\n");
520                 m_in_xrun = true;
521                 m_cycle_to_switch_state = cycle+1; // switch in the next cycle
522                 m_next_state = ePS_WaitingForStreamDisable;
523                 // execute the requested change
524                 if (!updateState()) { // we are allowed to change the state directly
525                     debugError("Could not update state!\n");
526                     return RAW1394_ISO_ERROR;
527                 }
528                 goto send_empty_packet;
529             }
530             return RAW1394_ISO_OK;
531         } else if (result == eCRV_XRun) { // pick up the possible xruns
532             debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state to WaitingForStreamDisable due to header xrun\n");
533             m_in_xrun = true;
534             m_cycle_to_switch_state = cycle+1; // switch in the next cycle
535             m_next_state = ePS_WaitingForStreamDisable;
536             // execute the requested change
537             if (!updateState()) { // we are allowed to change the state directly
538                 debugError("Could not update state!\n");
539                 return RAW1394_ISO_ERROR;
540             }
541         } else if ((result == eCRV_EmptyPacket) || (result == eCRV_Again)) {
542             if(m_state != m_next_state) {
543                 debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
544                                                 ePSToString(m_state), ePSToString(m_next_state));
545                 // execute the requested change
546                 if (!updateState()) { // we are allowed to change the state directly
547                     debugError("Could not update state!\n");
548                     return RAW1394_ISO_ERROR;
549                 }
550             }
551             goto send_empty_packet;
552 //         } else if (result == eCRV_Again) {
553 //             debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "have to retry cycle %d\n", cycle);
554 //             if(m_state != m_next_state) {
555 //                 debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
556 //                                                 ePSToString(m_state), ePSToString(m_next_state));
557 //                 // execute the requested change
558 //                 if (!updateState()) { // we are allowed to change the state directly
559 //                     debugError("Could not update state!\n");
560 //                     return RAW1394_ISO_ERROR;
561 //                 }
562 //             }
563             // force some delay
564 //             usleep(125);
565 //             return RAW1394_ISO_AGAIN;
566         } else {
567             debugError("Invalid return value: %d\n", result);
568             return RAW1394_ISO_ERROR;
569         }
570     }
571     // we are not running, so send an empty packet
572     // we should generate a valid packet any time
573 send_empty_packet:
574     // note that only the wait state changes are synchronized with the cycles
575     if(m_state != m_next_state) {
576         debugOutput(DEBUG_LEVEL_VERBOSE, "Should update state from %s to %s\n",
577                                         ePSToString(m_state), ePSToString(m_next_state));
578         // execute the requested change
579         if (!updateState()) { // we are allowed to change the state directly
580             debugError("Could not update state!\n");
581             return RAW1394_ISO_ERROR;
582         }
583     }
584
585     debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "XMIT EMPTY: CY=%04u\n", cycle);
586     generateSilentPacketHeader(data, length, tag, sy, cycle, dropped_cycles, max_length);
587     generateSilentPacketData(data, length, tag, sy, cycle, dropped_cycles, max_length);
588     return RAW1394_ISO_DEFER;
589 }
590
591
592 // Frame Transfer API
593 /**
594  * Transfer a block of frames from the event buffer to the port buffers
595  * @param nbframes number of frames to transfer
596  * @param ts the timestamp that the LAST frame in the block should have
597  * @return
598  */
599 bool StreamProcessor::getFrames(unsigned int nbframes, int64_t ts) {
600     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "%p.getFrames(%d, %11llu)", nbframes, ts);
601     assert( getType() == ePT_Receive );
602     if(isDryRunning()) return getFramesDry(nbframes, ts);
603     else return getFramesWet(nbframes, ts);
604 }
605
606 bool StreamProcessor::getFramesWet(unsigned int nbframes, int64_t ts) {
607 // FIXME: this should be done somewhere else
608 #ifdef DEBUG
609     uint64_t ts_expected;
610     signed int fc;
611     int32_t lag_ticks;
612     float lag_frames;
613
614     // in order to sync up multiple received streams, we should
615     // use the ts parameter. It specifies the time of the block's
616     // last sample.
617     float srate = m_manager->getSyncSource().getTicksPerFrame();
618     assert(srate != 0.0);
619     int64_t this_block_length_in_ticks = (int64_t)(((float)nbframes) * srate);
620
621     ffado_timestamp_t ts_head_tmp;
622     m_data_buffer->getBufferHeadTimestamp(&ts_head_tmp, &fc);
623     ts_expected = addTicks((uint64_t)ts_head_tmp, this_block_length_in_ticks);
624
625     lag_ticks = diffTicks(ts, ts_expected);
626     lag_frames = (((float)lag_ticks) / srate);
627     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "stream (%p): drifts %6d ticks = %10.5f frames (rate=%10.5f), %lld, %llu, %d\n",
628                  this, lag_ticks, lag_frames, srate, ts, ts_expected, fc);
629     if (lag_frames >= 1.0) {
630         // the stream lags
631         debugWarning( "stream (%p): lags  with %6d ticks = %10.5f frames (rate=%10.5f), %lld, %llu, %d\n",
632                       this, lag_ticks, lag_frames, srate, ts, ts_expected, fc);
633     } else if (lag_frames <= -1.0) {
634         // the stream leads
635         debugWarning( "stream (%p): leads with %6d ticks = %10.5f frames (rate=%10.5f), %lld, %llu, %d\n",
636                       this, lag_ticks, lag_frames, srate, ts, ts_expected, fc);
637     }
638 #endif
639     // ask the buffer to process nbframes of frames
640     // using it's registered client's processReadBlock(),
641     // which should be ours
642     m_data_buffer->blockProcessReadFrames(nbframes);
643     return true;
644 }
645
646 bool StreamProcessor::getFramesDry(unsigned int nbframes, int64_t ts)
647 {
648     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "stream (%p): dry run %d frames (@ ts=%lld)\n",
649                  this, nbframes, ts);
650     // dry run on this side means that we put silence in all enabled ports
651     // since there is do data put into the ringbuffer in the dry-running state
652     return provideSilenceBlock(nbframes, 0);
653 }
654
655 bool
656 StreamProcessor::dropFrames(unsigned int nbframes, int64_t ts)
657 {
658     debugOutput(DEBUG_LEVEL_VERY_VERBOSE, "StreamProcessor::dropFrames(%d, %lld)\n", nbframes, ts);
659     return m_data_buffer->dropFrames(nbframes);
660 }
661
662 bool StreamProcessor::putFrames(unsigned int nbframes, int64_t ts)
663 {
664     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "%p.putFrames(%d, %11llu)", nbframes, ts);
665     assert( getType() == ePT_Transmit );
666     if(isDryRunning()) return putFramesDry(nbframes, ts);
667     else return putFramesWet(nbframes, ts);
668 }
669
670 bool
671 StreamProcessor::putFramesWet(unsigned int nbframes, int64_t ts)
672 {
673     debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, "StreamProcessor::putFramesWet(%d, %llu)\n", nbframes, ts);
674     // transfer the data
675     m_data_buffer->blockProcessWriteFrames(nbframes, ts);
676     debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, " New timestamp: %llu\n", ts);
677     return true; // FIXME: what about failure?
678 }
679
680 bool
681 StreamProcessor::putFramesDry(unsigned int nbframes, int64_t ts)
682 {
683     debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, "StreamProcessor::putFramesDry(%d, %llu)\n", nbframes, ts);
684     // do nothing
685     return true;
686 }
687
688 bool
689 StreamProcessor::putSilenceFrames(unsigned int nbframes, int64_t ts)
690 {
691     debugOutput(DEBUG_LEVEL_ULTRA_VERBOSE, "StreamProcessor::putSilenceFrames(%d, %llu)\n", nbframes, ts);
692
693     size_t bytes_per_frame = getEventSize() * getEventsPerFrame();
694     unsigned int scratch_buffer_size_frames = m_scratch_buffer_size_bytes / bytes_per_frame;
695
696     if (nbframes > scratch_buffer_size_frames) {
697         debugError("nframes (%u) > scratch_buffer_size_frames (%u)\n",
698                    nbframes, scratch_buffer_size_frames);
699     }
700
701     assert(m_scratch_buffer);
702     if(!transmitSilenceBlock((char *)m_scratch_buffer, nbframes, 0)) {
703         debugError("Could not prepare silent block\n");
704         return false;
705     }
706     if(!m_data_buffer->writeFrames(nbframes, (char *)m_scratch_buffer, ts)) {
707         debugError("Could not write silent block\n");
708         return false;
709     }
710     return true;
711 }
712
713 bool
714 StreamProcessor::shiftStream(int nbframes)
715 {
716     if(nbframes == 0) return true;
717     if(nbframes > 0) {
718         return m_data_buffer->dropFrames(nbframes);
719     } else {
720         bool result = true;
721         while(nbframes++) {
722             result &= m_data_buffer->writeDummyFrame();
723         }
724         return result;
725     }
726 }
727
728 /**
729  * @brief write silence events to the stream ringbuffers.
730  */
731 bool StreamProcessor::provideSilenceBlock(unsigned int nevents, unsigned int offset)
732 {
733     bool no_problem=true;
734     for ( PortVectorIterator it = m_PeriodPorts.begin();
735           it != m_PeriodPorts.end();
736           ++it ) {
737         if((*it)->isDisabled()) {continue;};
738
739         //FIXME: make this into a static_cast when not DEBUG?
740         Port *port=dynamic_cast<Port *>(*it);
741
742         switch(port->getPortType()) {
743
744         case Port::E_Audio:
745             if(provideSilenceToPort(static_cast<AudioPort *>(*it), offset, nevents)) {
746                 debugWarning("Could not put silence into to port %s",(*it)->getName().c_str());
747                 no_problem=false;
748             }
749             break;
750         // midi is a packet based port, don't process
751         //    case MotuPortInfo::E_Midi:
752         //        break;
753
754         default: // ignore
755             break;
756         }
757     }
758     return no_problem;
759 }
760
761 int
762 StreamProcessor::provideSilenceToPort(
763                        AudioPort *p, unsigned int offset, unsigned int nevents)
764 {
765     unsigned int j=0;
766     switch(p->getDataType()) {
767         default:
768         case Port::E_Int24:
769             {
770                 quadlet_t *buffer=(quadlet_t *)(p->getBufferAddress());
771                 assert(nevents + offset <= p->getBufferSize());
772                 buffer+=offset;
773
774                 for(j = 0; j < nevents; j += 1) { // decode max nsamples
775                     *(buffer)=0;
776                     buffer++;
777                 }
778             }
779             break;
780         case Port::E_Float:
781             {
782                 float *buffer=(float *)(p->getBufferAddress());
783                 assert(nevents + offset <= p->getBufferSize());
784                 buffer+=offset;
785
786                 for(j = 0; j < nevents; j += 1) { // decode max nsamples
787                     *buffer = 0.0;
788                     buffer++;
789                 }
790             }
791             break;
792     }
793     return 0;
794 }
795
796 /***********************************************
797  * State related API                           *
798  ***********************************************/
799 bool StreamProcessor::init()
800 {
801     debugOutput( DEBUG_LEVEL_VERY_VERBOSE, "init...\n");
802
803     // initialization can be done without requesting it
804     // from the packet loop
805     m_next_state = ePS_Created;
806     return true;
807 }
808
809 bool StreamProcessor::prepare()
810 {
811     debugOutput( DEBUG_LEVEL_VERBOSE, "Prepare SP (%p)...\n", this);
812     if(!m_manager) {
813         debugFatal("Not attached to a manager!\n");
814         return false;
815     }
816
817     // make the scratch buffer one period of frames long
818     m_scratch_buffer_size_bytes = m_manager->getPeriodSize() * getEventsPerFrame() * getEventSize();
819     debugOutput( DEBUG_LEVEL_VERBOSE, " Allocate scratch buffer of %d quadlets\n");
820     if(m_scratch_buffer) delete[] m_scratch_buffer;
821     m_scratch_buffer = new byte_t[m_scratch_buffer_size_bytes];
822     if(m_scratch_buffer == NULL) {
823         debugFatal("Could not allocate scratch buffer\n");
824         return false;
825     }
826
827     if (!prepareChild()) {
828         debugFatal("Could not prepare child\n");
829         return false;
830     }
831
832     debugOutput( DEBUG_LEVEL_VERBOSE, "Prepared for:\n");
833     debugOutput( DEBUG_LEVEL_VERBOSE, " Samplerate: %d\n",
834              m_manager->getNominalRate());
835     debugOutput( DEBUG_LEVEL_VERBOSE, " PeriodSize: %d, NbBuffers: %d\n",
836              m_manager->getPeriodSize(), m_manager->getNbBuffers());
837     debugOutput( DEBUG_LEVEL_VERBOSE, " Port: %d, Channel: %d\n",
838              getPort(), m_channel);
839
840     // initialization can be done without requesting it
841     // from the packet loop
842     m_next_state = ePS_Stopped;
843     return updateState();
844 }
845
846 bool
847 StreamProcessor::scheduleStateTransition(enum eProcessorState state, uint64_t time_instant)
848 {
849     // first set the time, since in the packet loop we first check m_state == m_next_state before
850     // using the time
851     m_cycle_to_switch_state = TICKS_TO_CYCLES(time_instant);
852     m_next_state = state;
853     return true;
854 }
855
856 bool
857 StreamProcessor::waitForState(enum eProcessorState state, unsigned int timeout_ms)
858 {
859     debugOutput(DEBUG_LEVEL_VERBOSE, "Waiting for state %s\n", ePSToString(state));
860     int cnt = timeout_ms;
861     while (m_state != state && cnt) {
862         usleep(1000);
863         cnt--;
864     }
865     if(cnt==0) {
866         debugOutput(DEBUG_LEVEL_VERBOSE, " Timeout\n");
867         return false;
868     }
869     return true;
870 }
871
872 bool StreamProcessor::scheduleStartDryRunning(int64_t t) {
873     uint64_t tx;
874     if (t < 0) {
875         tx = addTicks(m_parent.get1394Service().getCycleTimerTicks(), 200 * TICKS_PER_CYCLE);
876     } else {
877         tx = t;
878     }
879     debugOutput(DEBUG_LEVEL_VERBOSE,"for %s SP (%p)\n", ePTToString(getType()), this);
880     uint64_t now = m_parent.get1394Service().getCycleTimerTicks();
881     debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011llu (%03us %04uc %04ut)\n",
882                           now,
883                           (unsigned int)TICKS_TO_SECS(now),
884                           (unsigned int)TICKS_TO_CYCLES(now),
885                           (unsigned int)TICKS_TO_OFFSET(now));
886     debugOutput(DEBUG_LEVEL_VERBOSE,"  Start at              : %011llu (%03us %04uc %04ut)\n",
887                           tx,
888                           (unsigned int)TICKS_TO_SECS(tx),
889                           (unsigned int)TICKS_TO_CYCLES(tx),
890                           (unsigned int)TICKS_TO_OFFSET(tx));
891     if (m_state == ePS_Stopped) {
892         return scheduleStateTransition(ePS_WaitingForStream, tx);
893     } else if (m_state == ePS_Running) {
894         return scheduleStateTransition(ePS_WaitingForStreamDisable, tx);
895     } else {
896         debugError("Cannot switch to ePS_DryRunning from %s\n", ePSToString(m_state));
897         return false;
898     }
899 }
900
901 bool StreamProcessor::scheduleStartRunning(int64_t t) {
902     uint64_t tx;
903     if (t < 0) {
904         tx = addTicks(m_parent.get1394Service().getCycleTimerTicks(), 200 * TICKS_PER_CYCLE);
905     } else {
906         tx = t;
907     }
908     debugOutput(DEBUG_LEVEL_VERBOSE,"for %s SP (%p)\n", ePTToString(getType()), this);
909     uint64_t now = m_parent.get1394Service().getCycleTimerTicks();
910     debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011llu (%03us %04uc %04ut)\n",
911                           now,
912                           (unsigned int)TICKS_TO_SECS(now),
913                           (unsigned int)TICKS_TO_CYCLES(now),
914                           (unsigned int)TICKS_TO_OFFSET(now));
915     debugOutput(DEBUG_LEVEL_VERBOSE,"  Start at              : %011llu (%03us %04uc %04ut)\n",
916                           tx,
917                           (unsigned int)TICKS_TO_SECS(tx),
918                           (unsigned int)TICKS_TO_CYCLES(tx),
919                           (unsigned int)TICKS_TO_OFFSET(tx));
920     return scheduleStateTransition(ePS_WaitingForStreamEnable, tx);
921 }
922
923 bool StreamProcessor::scheduleStopDryRunning(int64_t t) {
924     uint64_t tx;
925     if (t < 0) {
926         tx = addTicks(m_parent.get1394Service().getCycleTimerTicks(), 200 * TICKS_PER_CYCLE);
927     } else {
928         tx = t;
929     }
930     debugOutput(DEBUG_LEVEL_VERBOSE,"for %s SP (%p)\n", ePTToString(getType()), this);
931     uint64_t now = m_parent.get1394Service().getCycleTimerTicks();
932     debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011llu (%03us %04uc %04ut)\n",
933                           now,
934                           (unsigned int)TICKS_TO_SECS(now),
935                           (unsigned int)TICKS_TO_CYCLES(now),
936                           (unsigned int)TICKS_TO_OFFSET(now));
937     debugOutput(DEBUG_LEVEL_VERBOSE,"  Stop at               : %011llu (%03us %04uc %04ut)\n",
938                           tx,
939                           (unsigned int)TICKS_TO_SECS(tx),
940                           (unsigned int)TICKS_TO_CYCLES(tx),
941                           (unsigned int)TICKS_TO_OFFSET(tx));
942     return scheduleStateTransition(ePS_Stopped, tx);
943 }
944
945 bool StreamProcessor::scheduleStopRunning(int64_t t) {
946     uint64_t tx;
947     if (t < 0) {
948         tx = addTicks(m_parent.get1394Service().getCycleTimerTicks(), 200 * TICKS_PER_CYCLE);
949     } else {
950         tx = t;
951     }
952     debugOutput(DEBUG_LEVEL_VERBOSE,"for %s SP (%p)\n", ePTToString(getType()), this);
953     uint64_t now = m_parent.get1394Service().getCycleTimerTicks();
954     debugOutput(DEBUG_LEVEL_VERBOSE,"  Now                   : %011llu (%03us %04uc %04ut)\n",
955                           now,
956                           (unsigned int)TICKS_TO_SECS(now),
957                           (unsigned int)TICKS_TO_CYCLES(now),
958                           (unsigned int)TICKS_TO_OFFSET(now));
959     debugOutput(DEBUG_LEVEL_VERBOSE,"  Stop at               : %011llu (%03us %04uc %04ut)\n",
960                           tx,
961                           (unsigned int)TICKS_TO_SECS(tx),
962                           (unsigned int)TICKS_TO_CYCLES(tx),
963                           (unsigned int)TICKS_TO_OFFSET(tx));
964     return scheduleStateTransition(ePS_WaitingForStreamDisable, tx);
965 }
966
967 bool StreamProcessor::startDryRunning(int64_t t) {
968     if(!scheduleStartDryRunning(t)) {
969         debugError("Could not schedule transition\n");
970         return false;
971     }
972     if(!waitForState(ePS_DryRunning, 2000)) {
973         debugError(" Timeout while waiting for %s\n", ePSToString(ePS_DryRunning));
974         return false;
975     }
976     return true;
977 }
978
979 bool StreamProcessor::startRunning(int64_t t) {
980     if(!scheduleStartRunning(t)) {
981         debugError("Could not schedule transition\n");
982         return false;
983     }
984     if(!waitForState(ePS_Running, 2000)) {
985         debugError(" Timeout while waiting for %s\n", ePSToString(ePS_Running));
986         return false;
987     }
988     return true;
989 }
990
991 bool StreamProcessor::stopDryRunning(int64_t t) {
992     if(!scheduleStopDryRunning(t)) {
993         debugError("Could not schedule transition\n");
994         return false;
995     }
996     if(!waitForState(ePS_Stopped, 2000)) {
997         debugError(" Timeout while waiting for %s\n", ePSToString(ePS_Stopped));
998         return false;
999     }
1000     return true;
1001 }
1002
1003 bool StreamProcessor::stopRunning(int64_t t) {
1004     if(!scheduleStopRunning(t)) {
1005         debugError("Could not schedule transition\n");
1006         return false;
1007     }
1008     if(!waitForState(ePS_DryRunning, 2000)) {
1009         debugError(" Timeout while waiting for %s\n", ePSToString(ePS_DryRunning));
1010         return false;
1011     }
1012     return true;
1013 }
1014
1015
1016 // internal state API
1017
1018 /**
1019  * @brief Enter the ePS_Stopped state
1020  * @return true if successful, false if not
1021  *
1022  * @pre none
1023  *
1024  * @post the buffer and the isostream are ready for use.
1025  * @post all dynamic structures have been allocated successfully
1026  * @post the buffer is transparent and empty, and all parameters are set
1027  *       to the correct initial/nominal values.
1028  *
1029  */
1030 bool
1031 StreamProcessor::doStop()
1032 {
1033     float ticks_per_frame;
1034     unsigned int ringbuffer_size_frames = (m_manager->getNbBuffers() + 1) * m_manager->getPeriodSize();
1035
1036     debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
1037     bool result = true;
1038
1039     switch(m_state) {
1040         case ePS_Created:
1041             assert(m_data_buffer);
1042             // object just created
1043             result = m_data_buffer->init();
1044
1045             // prepare the framerate estimate
1046             ticks_per_frame = (TICKS_PER_SECOND*1.0) / ((float)m_manager->getNominalRate());
1047             m_ticks_per_frame = ticks_per_frame;
1048             debugOutput(DEBUG_LEVEL_VERBOSE,"Initializing remote ticks/frame to %f\n", ticks_per_frame);
1049
1050             // initialize internal buffer
1051             result &= m_data_buffer->setBufferSize(ringbuffer_size_frames);
1052
1053             result &= m_data_buffer->setEventSize( getEventSize() );
1054             result &= m_data_buffer->setEventsPerFrame( getEventsPerFrame() );
1055             if(getType() == ePT_Receive) {
1056                 result &= m_data_buffer->setUpdatePeriod( getNominalFramesPerPacket() );
1057             } else {
1058                 result &= m_data_buffer->setUpdatePeriod( m_manager->getPeriodSize() );
1059             }
1060             result &= m_data_buffer->setNominalRate(ticks_per_frame);
1061             result &= m_data_buffer->setWrapValue(128L*TICKS_PER_SECOND);
1062             result &= m_data_buffer->prepare(); // FIXME: the name
1063
1064             // set the parameters of ports we can:
1065             // we want the audio ports to be period buffered,
1066             // and the midi ports to be packet buffered
1067             for ( PortVectorIterator it = m_Ports.begin();
1068                 it != m_Ports.end();
1069                 ++it )
1070             {
1071                 debugOutput(DEBUG_LEVEL_VERBOSE, "Setting up port %s\n",(*it)->getName().c_str());
1072                 if(!(*it)->setBufferSize(m_manager->getPeriodSize())) {
1073                     debugFatal("Could not set buffer size to %d\n",m_manager->getPeriodSize());
1074                     return false;
1075                 }
1076                 switch ((*it)->getPortType()) {
1077                     case Port::E_Audio:
1078                         if(!(*it)->setSignalType(Port::E_PeriodSignalled)) {
1079                             debugFatal("Could not set signal type to PeriodSignalling");
1080                             return false;
1081                         }
1082                         // buffertype and datatype are dependant on the API
1083                         debugWarning("---------------- ! Doing hardcoded dummy setup ! --------------\n");
1084                         // buffertype and datatype are dependant on the API
1085                         if(!(*it)->setBufferType(Port::E_PointerBuffer)) {
1086                             debugFatal("Could not set buffer type");
1087                             return false;
1088                         }
1089                         if(!(*it)->useExternalBuffer(true)) {
1090                             debugFatal("Could not set external buffer usage");
1091                             return false;
1092                         }
1093                         if(!(*it)->setDataType(Port::E_Float)) {
1094                             debugFatal("Could not set data type");
1095                             return false;
1096                         }
1097                         break;
1098                     case Port::E_Midi:
1099                         if(!(*it)->setSignalType(Port::E_PacketSignalled)) {
1100                             debugFatal("Could not set signal type to PacketSignalling");
1101                             return false;
1102                         }
1103                         // buffertype and datatype are dependant on the API
1104                         debugWarning("---------------- ! Doing hardcoded test setup ! --------------\n");
1105                         // buffertype and datatype are dependant on the API
1106                         if(!(*it)->setBufferType(Port::E_RingBuffer)) {
1107                             debugFatal("Could not set buffer type");
1108                             return false;
1109                         }
1110                         if(!(*it)->setDataType(Port::E_MidiEvent)) {
1111                             debugFatal("Could not set data type");
1112                             return false;
1113                         }
1114                         break;
1115                     default:
1116                         debugWarning("Unsupported port type specified\n");
1117                         break;
1118                 }
1119             }
1120             // the API specific settings of the ports should already be set,
1121             // as this is called from the processorManager->prepare()
1122             // so we can init the ports
1123             result &= PortManager::initPorts();
1124
1125             break;
1126         case ePS_DryRunning:
1127             // what to do here?
1128             break;
1129         default:
1130             debugError("Entry from invalid state: %s\n", ePSToString(m_state));
1131             return false;
1132     }
1133
1134     result &= m_data_buffer->clearBuffer(); // FIXME: don't like the reset() name
1135     // make the buffer transparent
1136     m_data_buffer->setTransparent(true);
1137
1138     // reset all ports
1139     result &= PortManager::preparePorts();
1140
1141     m_state = ePS_Stopped;
1142     #ifdef DEBUG
1143     if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
1144         debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
1145         dumpInfo();
1146     }
1147     #endif
1148     return result;
1149 }
1150
1151 /**
1152  * @brief Enter the ePS_WaitingForStream state
1153  * @return true if successful, false if not
1154  *
1155  * @pre all dynamic data structures are allocated successfully
1156  *
1157  * @post
1158  *
1159  */
1160 bool
1161 StreamProcessor::doWaitForRunningStream()
1162 {
1163     debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
1164     switch(m_state) {
1165         case ePS_Stopped:
1166             // we have to start waiting for an incoming stream
1167             // this basically means nothing, the state change will
1168             // be picked up by the packet iterator
1169             break;
1170         default:
1171             debugError("Entry from invalid state: %s\n", ePSToString(m_state));
1172             return false;
1173     }
1174     m_state = ePS_WaitingForStream;
1175     #ifdef DEBUG
1176     if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
1177         debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
1178         dumpInfo();
1179     }
1180     #endif
1181     return true;
1182 }
1183
1184 /**
1185  * @brief Enter the ePS_DryRunning state
1186  * @return true if successful, false if not
1187  *
1188  * @pre
1189  *
1190  * @post
1191  *
1192  */
1193 bool
1194 StreamProcessor::doDryRunning()
1195 {
1196     bool result = true;
1197     debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
1198     switch(m_state) {
1199         case ePS_WaitingForStream:
1200             // a running stream has been detected
1201             debugOutput(DEBUG_LEVEL_VERBOSE, "StreamProcessor %p started dry-running at cycle %d\n", this, m_last_cycle);
1202             if (getType() == ePT_Receive) {
1203                 // this to ensure that there is no discontinuity when starting to
1204                 // update the DLL based upon the received packets
1205                 m_data_buffer->setBufferTailTimestamp(m_last_timestamp);
1206             } else {
1207                 // FIXME: PC=master mode will have to do something here I guess...
1208             }
1209             break;
1210         case ePS_WaitingForStreamDisable:
1211             result &= m_data_buffer->clearBuffer(); // FIXME: don't like the reset() name
1212             m_data_buffer->setTransparent(true);
1213             break;
1214         default:
1215             debugError("Entry from invalid state: %s\n", ePSToString(m_state));
1216             return false;
1217     }
1218     m_state = ePS_DryRunning;
1219     #ifdef DEBUG
1220     if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
1221         debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
1222         dumpInfo();
1223     }
1224     #endif
1225     return result;
1226 }
1227
1228 /**
1229  * @brief Enter the ePS_WaitingForStreamEnable state
1230  * @return true if successful, false if not
1231  *
1232  * @pre
1233  *
1234  * @post
1235  *
1236  */
1237 bool
1238 StreamProcessor::doWaitForStreamEnable()
1239 {
1240     debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
1241     unsigned int ringbuffer_size_frames;
1242     switch(m_state) {
1243         case ePS_DryRunning:
1244             // we have to start waiting for an incoming stream
1245             // this basically means nothing, the state change will
1246             // be picked up by the packet iterator
1247
1248             if(!m_data_buffer->clearBuffer()) { // FIXME: don't like the reset() name
1249                 debugError("Could not reset data buffer\n");
1250                 return false;
1251             }
1252             if (getType() == ePT_Transmit) {
1253                 ringbuffer_size_frames = m_manager->getNbBuffers() * m_manager->getPeriodSize();
1254                 debugOutput(DEBUG_LEVEL_VERBOSE, "Prefill transmit SP %p with %u frames\n", this, ringbuffer_size_frames);
1255                 // prefill the buffer
1256                 if(!transferSilence(ringbuffer_size_frames)) {
1257                     debugFatal("Could not prefill transmit stream\n");
1258                     return false;
1259                 }
1260             }
1261
1262             break;
1263         default:
1264             debugError("Entry from invalid state: %s\n", ePSToString(m_state));
1265             return false;
1266     }
1267     m_state = ePS_WaitingForStreamEnable;
1268     #ifdef DEBUG
1269     if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
1270         debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
1271         dumpInfo();
1272     }
1273     #endif
1274     return true;
1275 }
1276
1277 /**
1278  * @brief Enter the ePS_Running state
1279  * @return true if successful, false if not
1280  *
1281  * @pre
1282  *
1283  * @post
1284  *
1285  */
1286 bool
1287 StreamProcessor::doRunning()
1288 {
1289     bool result = true;
1290     debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
1291     switch(m_state) {
1292         case ePS_WaitingForStreamEnable:
1293             // a running stream has been detected
1294             debugOutput(DEBUG_LEVEL_VERBOSE, "StreamProcessor %p started running at cycle %d\n",
1295                                              this, m_last_cycle);
1296             m_in_xrun = false;
1297             m_data_buffer->setTransparent(false);
1298             break;
1299         default:
1300             debugError("Entry from invalid state: %s\n", ePSToString(m_state));
1301             return false;
1302     }
1303     m_state = ePS_Running;
1304     #ifdef DEBUG
1305     if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
1306         debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
1307         dumpInfo();
1308     }
1309     #endif
1310     return result;
1311 }
1312
1313 /**
1314  * @brief Enter the ePS_WaitingForStreamDisable state
1315  * @return true if successful, false if not
1316  *
1317  * @pre
1318  *
1319  * @post
1320  *
1321  */
1322 bool
1323 StreamProcessor::doWaitForStreamDisable()
1324 {
1325     debugOutput(DEBUG_LEVEL_VERBOSE, "Enter from state: %s\n", ePSToString(m_state));
1326     switch(m_state) {
1327         case ePS_Running:
1328             // the thread will do the transition
1329             break;
1330         default:
1331             debugError("Entry from invalid state: %s\n", ePSToString(m_state));
1332             return false;
1333     }
1334     m_state = ePS_WaitingForStreamDisable;
1335     #ifdef DEBUG
1336     if (getDebugLevel() >= DEBUG_LEVEL_VERBOSE) {
1337         debugOutput(DEBUG_LEVEL_VERBOSE, "State switch complete, dumping SP info...\n");
1338         dumpInfo();
1339     }
1340     #endif
1341     return true;
1342 }
1343
1344 /**
1345  * @brief Updates the state machine and calls the necessary transition functions
1346  * @return true if successful, false if not
1347  */
1348 bool StreamProcessor::updateState() {
1349     bool result = false;
1350     // copy the current state locally since it could change value,
1351     // and that's something we don't want to happen inbetween tests
1352     // if m_next_state changes during this routine, we know for sure
1353     // that the previous state change was at least attempted correctly.
1354     enum eProcessorState next_state = m_next_state;
1355
1356     debugOutput(DEBUG_LEVEL_VERBOSE, "Do state transition: %s => %s\n",
1357         ePSToString(m_state), ePSToString(next_state));
1358
1359     if (m_state == next_state) {
1360         debugWarning("ignoring identity state update from/to %s\n", ePSToString(m_state) );
1361         return true;
1362     }
1363
1364     // after creation, only initialization is allowed
1365     if (m_state == ePS_Created) {
1366         if(next_state != ePS_Stopped) {
1367             goto updateState_exit_with_error;
1368         }
1369         // do init here
1370         result = doStop();
1371         if (result) return true;
1372         else goto updateState_exit_change_failed;
1373     }
1374
1375     // after initialization, only WaitingForRunningStream is allowed
1376     if (m_state == ePS_Stopped) {
1377         if(next_state != ePS_WaitingForStream) {
1378             goto updateState_exit_with_error;
1379         }
1380         result = doWaitForRunningStream();
1381         if (result) return true;
1382         else goto updateState_exit_change_failed;
1383     }
1384
1385     // after WaitingForStream, only ePS_DryRunning is allowed
1386     // this means that the stream started running
1387     if (m_state == ePS_WaitingForStream) {
1388         if(next_state != ePS_DryRunning) {
1389             goto updateState_exit_with_error;
1390         }
1391         result = doDryRunning();
1392         if (result) return true;
1393         else goto updateState_exit_change_failed;
1394     }
1395
1396     // from ePS_DryRunning we can go to:
1397     //   - ePS_Stopped if something went wrong during DryRunning
1398     //   - ePS_WaitingForStreamEnable if there is a requested to enable
1399     if (m_state == ePS_DryRunning) {
1400         if((next_state != ePS_Stopped) &&
1401            (next_state != ePS_WaitingForStreamEnable)) {
1402             goto updateState_exit_with_error;
1403         }
1404         if (next_state == ePS_Stopped) {
1405             result = doStop();
1406         } else {
1407             result = doWaitForStreamEnable();
1408         }
1409         if (result) return true;
1410         else goto updateState_exit_change_failed;
1411     }
1412
1413     // from ePS_WaitingForStreamEnable we can go to:
1414     //   - ePS_DryRunning if something went wrong while waiting
1415     //   - ePS_Running if the stream enabled correctly
1416     if (m_state == ePS_WaitingForStreamEnable) {
1417         if((next_state != ePS_DryRunning) &&
1418            (next_state != ePS_Running)) {
1419             goto updateState_exit_with_error;
1420         }
1421         if (next_state == ePS_Stopped) {
1422             result = doDryRunning();
1423         } else {
1424             result = doRunning();
1425         }
1426         if (result) return true;
1427         else goto updateState_exit_change_failed;
1428     }
1429
1430     // from ePS_Running we can only start waiting for a disabled stream
1431     if (m_state == ePS_Running) {
1432         if(next_state != ePS_WaitingForStreamDisable) {
1433             goto updateState_exit_with_error;
1434         }
1435         result = doWaitForStreamDisable();
1436         if (result) return true;
1437         else goto updateState_exit_change_failed;
1438     }
1439
1440     // from ePS_WaitingForStreamDisable we can go to DryRunning
1441     if (m_state == ePS_WaitingForStreamDisable) {
1442         if(next_state != ePS_DryRunning) {
1443             goto updateState_exit_with_error;
1444         }
1445         result = doDryRunning();
1446         if (result) return true;
1447         else goto updateState_exit_change_failed;
1448     }
1449
1450     // if we arrive here there is an error
1451 updateState_exit_with_error:
1452     debugError("Invalid state transition: %s => %s\n",
1453         ePSToString(m_state), ePSToString(next_state));
1454     return false;
1455 updateState_exit_change_failed:
1456     debugError("State transition failed: %s => %s\n",
1457         ePSToString(m_state), ePSToString(next_state));
1458     return false;
1459 }
1460
1461 /***********************************************
1462  * Helper routines                             *
1463  ***********************************************/
1464 bool
1465 StreamProcessor::transferSilence(unsigned int nframes)
1466 {
1467     bool retval;
1468     signed int fc;
1469     ffado_timestamp_t ts_tail_tmp;
1470
1471     // prepare a buffer of silence
1472     char *dummybuffer = (char *)calloc(sizeof(quadlet_t), nframes * getEventsPerFrame());
1473     transmitSilenceBlock(dummybuffer, nframes, 0);
1474
1475     m_data_buffer->getBufferTailTimestamp(&ts_tail_tmp, &fc);
1476     if (fc != 0) {
1477         debugWarning("Prefilling a buffer that already contains %d frames\n", fc);
1478     }
1479
1480     // add the silence data to the ringbuffer
1481     if(m_data_buffer->preloadFrames(nframes, dummybuffer, true)) {
1482         retval = true;
1483     } else {
1484         debugWarning("Could not write to event buffer\n");
1485         retval = false;
1486     }
1487     free(dummybuffer);
1488     return retval;
1489 }
1490
1491 /**
1492  * @brief convert a eProcessorState to a string
1493  * @param s the state
1494  * @return a char * describing the state
1495  */
1496 const char *
1497 StreamProcessor::ePSToString(enum eProcessorState s) {
1498     switch (s) {
1499         case ePS_Invalid: return "ePS_Invalid";
1500         case ePS_Created: return "ePS_Created";
1501         case ePS_Stopped: return "ePS_Stopped";
1502         case ePS_WaitingForStream: return "ePS_WaitingForStream";
1503         case ePS_DryRunning: return "ePS_DryRunning";
1504         case ePS_WaitingForStreamEnable: return "ePS_WaitingForStreamEnable";
1505         case ePS_Running: return "ePS_Running";
1506         case ePS_WaitingForStreamDisable: return "ePS_WaitingForStreamDisable";
1507         default: return "error: unknown state";
1508     }
1509 }
1510
1511 /**
1512  * @brief convert a eProcessorType to a string
1513  * @param t the type
1514  * @return a char * describing the state
1515  */
1516 const char *
1517 StreamProcessor::ePTToString(enum eProcessorType t) {
1518     switch (t) {
1519         case ePT_Receive: return "Receive";
1520         case ePT_Transmit: return "Transmit";
1521         default: return "error: unknown type";
1522     }
1523 }
1524
1525 /***********************************************
1526  * Debug                                       *
1527  ***********************************************/
1528 void
1529 StreamProcessor::dumpInfo()
1530 {
1531     debugOutputShort( DEBUG_LEVEL_NORMAL, " StreamProcessor %p information\n", this);
1532     debugOutputShort( DEBUG_LEVEL_NORMAL, "  Port, Channel  : %d, %d\n", getPort(), m_channel);
1533     debugOutputShort( DEBUG_LEVEL_NORMAL, "  StreamProcessor info:\n");
1534     if (m_handler) {
1535         uint64_t now = m_parent.get1394Service().getCycleTimerTicks();
1536         debugOutputShort( DEBUG_LEVEL_NORMAL, "  Now                   : %011llu (%03us %04uc %04ut)\n",
1537                           now,
1538                           (unsigned int)TICKS_TO_SECS(now),
1539                           (unsigned int)TICKS_TO_CYCLES(now),
1540                           (unsigned int)TICKS_TO_OFFSET(now));
1541     }
1542     debugOutputShort( DEBUG_LEVEL_NORMAL, "  Xruns                 : %s\n", (m_in_xrun ? "True":"False"));
1543     debugOutputShort( DEBUG_LEVEL_NORMAL, "  State                 : %s\n", ePSToString(m_state));
1544     debugOutputShort( DEBUG_LEVEL_NORMAL, "   Next state           : %s\n", ePSToString(m_next_state));
1545     debugOutputShort( DEBUG_LEVEL_NORMAL, "    transition at       : %u\n", m_cycle_to_switch_state);
1546     debugOutputShort( DEBUG_LEVEL_NORMAL, "  Buffer                : %p\n", m_data_buffer);
1547     debugOutputShort( DEBUG_LEVEL_NORMAL, "  Nominal framerate     : %u\n", m_manager->getNominalRate());
1548     debugOutputShort( DEBUG_LEVEL_NORMAL, "  Device framerate      : Sync: %f, Buffer %f\n",
1549         24576000.0/m_manager->getSyncSource().m_data_buffer->getRate(),
1550         24576000.0/m_data_buffer->getRate()
1551         );
1552
1553     m_data_buffer->dumpInfo();
1554 }
1555
1556 void
1557 StreamProcessor::setVerboseLevel(int l) {
1558     setDebugLevel(l);
1559     PortManager::setVerboseLevel(l);
1560     m_data_buffer->setVerboseLevel(l);
1561 }
1562
1563 } // end of namespace
Note: See TracBrowser for help on using the browser.