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

Revision 753, 60.1 kB (checked in by ppalmers, 16 years ago)

have separate threads for every handler

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