X-Git-Url: https://sigrok.org/gitweb/?p=pulseview.git;a=blobdiff_plain;f=pv%2Fdata%2Fdecoderstack.cpp;h=c3bf26d25699c12135707b40512414c9ee0ef4f3;hp=9b02f9e929cc41b9ca75709acc27f3793745c333;hb=8bd26d8b9c831b509ee3241ea4dac6f50c023622;hpb=13a3528cfa6ed4dc18c44aae2353ac145b0160cc diff --git a/pv/data/decoderstack.cpp b/pv/data/decoderstack.cpp index 9b02f9e9..c3bf26d2 100644 --- a/pv/data/decoderstack.cpp +++ b/pv/data/decoderstack.cpp @@ -33,11 +33,14 @@ #include #include #include +#include #include using boost::lock_guard; using boost::mutex; +using boost::optional; using boost::shared_ptr; +using boost::unique_lock; using std::deque; using std::make_pair; using std::max; @@ -55,20 +58,34 @@ namespace data { const double DecoderStack::DecodeMargin = 1.0; const double DecoderStack::DecodeThreshold = 0.2; const int64_t DecoderStack::DecodeChunkLength = 4096; +const unsigned int DecoderStack::DecodeNotifyPeriod = 65536; mutex DecoderStack::_global_decode_mutex; -DecoderStack::DecoderStack(const srd_decoder *const dec) : +DecoderStack::DecoderStack(pv::SigSession &session, + const srd_decoder *const dec) : + _session(session), + _sample_count(0), + _frame_complete(false), _samples_decoded(0) { + connect(&_session, SIGNAL(frame_began()), + this, SLOT(on_new_frame())); + connect(&_session, SIGNAL(data_received()), + this, SLOT(on_data_received())); + connect(&_session, SIGNAL(frame_ended()), + this, SLOT(on_frame_ended())); + _stack.push_back(shared_ptr( new decode::Decoder(dec))); } DecoderStack::~DecoderStack() { - _decode_thread.interrupt(); - _decode_thread.join(); + if (_decode_thread.joinable()) { + _decode_thread.interrupt(); + _decode_thread.join(); + } } const std::list< boost::shared_ptr >& @@ -99,13 +116,13 @@ void DecoderStack::remove(int index) int64_t DecoderStack::samples_decoded() const { - lock_guard decode_lock(_mutex); + lock_guard decode_lock(_output_mutex); return _samples_decoded; } std::vector DecoderStack::get_visible_rows() const { - lock_guard lock(_mutex); + lock_guard lock(_output_mutex); vector rows; @@ -140,7 +157,7 @@ void DecoderStack::get_annotation_subset( const Row &row, uint64_t start_sample, uint64_t end_sample) const { - lock_guard lock(_mutex); + lock_guard lock(_output_mutex); std::map::const_iterator iter = _rows.find(row); @@ -151,12 +168,14 @@ void DecoderStack::get_annotation_subset( QString DecoderStack::error_message() { - lock_guard lock(_mutex); + lock_guard lock(_output_mutex); return _error_message; } void DecoderStack::clear() { + _sample_count = 0; + _frame_complete = false; _samples_decoded = 0; _error_message = QString(); _rows.clear(); @@ -168,11 +187,21 @@ void DecoderStack::begin_decode() shared_ptr logic_signal; shared_ptr data; - _decode_thread.interrupt(); - _decode_thread.join(); + if (_decode_thread.joinable()) { + _decode_thread.interrupt(); + _decode_thread.join(); + } clear(); + // Check that all decoders have the required channels + BOOST_FOREACH(const shared_ptr &dec, _stack) + if (!dec->have_required_probes()) { + _error_message = tr("One or more required channels " + "have not been specified"); + return; + } + // Add classes BOOST_FOREACH (const shared_ptr &dec, _stack) { @@ -204,26 +233,32 @@ void DecoderStack::begin_decode() } } - // We get the logic data of the first probe in the list. + // We get the logic data of the first channel in the list. // This works because we are currently assuming all // LogicSignals have the same data/snapshot BOOST_FOREACH (const shared_ptr &dec, _stack) - if (dec && !dec->probes().empty() && - ((logic_signal = (*dec->probes().begin()).second)) && + if (dec && !dec->channels().empty() && + ((logic_signal = (*dec->channels().begin()).second)) && ((data = logic_signal->logic_data()))) break; if (!data) return; + // Check we have a snapshot of data + const deque< shared_ptr > &snapshots = + data->get_snapshots(); + if (snapshots.empty()) + return; + _snapshot = snapshots.front(); + // Get the samplerate and start time _start_time = data->get_start_time(); _samplerate = data->samplerate(); if (_samplerate == 0.0) _samplerate = 1.0; - _decode_thread = boost::thread(&DecoderStack::decode_proc, this, - data); + _decode_thread = boost::thread(&DecoderStack::decode_proc, this); } uint64_t DecoderStack::get_max_sample_count() const @@ -238,37 +273,71 @@ uint64_t DecoderStack::get_max_sample_count() const return max_sample_count; } -void DecoderStack::decode_proc(shared_ptr data) +optional DecoderStack::wait_for_data() const +{ + unique_lock input_lock(_input_mutex); + while(!boost::this_thread::interruption_requested() && + !_frame_complete && _samples_decoded >= _sample_count) + _input_cond.wait(input_lock); + return boost::make_optional( + !boost::this_thread::interruption_requested() && + (_samples_decoded < _sample_count || !_frame_complete), + _sample_count); +} + +void DecoderStack::decode_data( + const int64_t sample_count, const unsigned int unit_size, + srd_session *const session) { - srd_session *session; uint8_t chunk[DecodeChunkLength]; - srd_decoder_inst *prev_di = NULL; - assert(data); + const unsigned int chunk_sample_count = + DecodeChunkLength / _snapshot->unit_size(); - // Check we have a snapshot of data - const deque< shared_ptr > &snapshots = - data->get_snapshots(); - if (snapshots.empty()) - return; + for (int64_t i = 0; + !boost::this_thread::interruption_requested() && + i < sample_count; + i += chunk_sample_count) + { + lock_guard decode_lock(_global_decode_mutex); - // Check that all decoders have the required probes - BOOST_FOREACH(const shared_ptr &dec, _stack) - if (!dec->have_required_probes()) - return; + const int64_t chunk_end = min( + i + chunk_sample_count, sample_count); + _snapshot->get_samples(chunk, i, chunk_end); - const shared_ptr &snapshot = - snapshots.front(); - const int64_t sample_count = snapshot->get_sample_count(); - const unsigned int unit_size = snapshot->unit_size(); - const unsigned int chunk_sample_count = - DecodeChunkLength / unit_size; + if (srd_session_send(session, i, i + sample_count, chunk, + (chunk_end - i) * unit_size) != SRD_OK) { + _error_message = tr("Decoder reported an error"); + break; + } + + { + lock_guard lock(_output_mutex); + _samples_decoded = chunk_end; + } + + if (i % DecodeNotifyPeriod == 0) + new_decode_data(); + } + + new_decode_data(); +} + +void DecoderStack::decode_proc() +{ + optional sample_count; + srd_session *session; + srd_decoder_inst *prev_di = NULL; + + assert(_snapshot); // Create the session srd_session_new(&session); assert(session); // Create the decoders + const unsigned int unit_size = _snapshot->unit_size(); + BOOST_FOREACH(const shared_ptr &dec, _stack) { srd_decoder_inst *const di = dec->create_decoder_inst(session, unit_size); @@ -286,6 +355,12 @@ void DecoderStack::decode_proc(shared_ptr data) prev_di = di; } + // Get the intial sample count + { + unique_lock input_lock(_input_mutex); + sample_count = _sample_count = _snapshot->get_sample_count(); + } + // Start the session srd_session_metadata_set(session, SRD_CONF_SAMPLERATE, g_variant_new_uint64((uint64_t)_samplerate)); @@ -295,28 +370,9 @@ void DecoderStack::decode_proc(shared_ptr data) srd_session_start(session); - for (int64_t i = 0; - !boost::this_thread::interruption_requested() && - i < sample_count; - i += chunk_sample_count) - { - lock_guard decode_lock(_global_decode_mutex); - - const int64_t chunk_end = min( - i + chunk_sample_count, sample_count); - snapshot->get_samples(chunk, i, chunk_end); - - if (srd_session_send(session, i, i + sample_count, chunk, - (chunk_end - i) * unit_size) != SRD_OK) { - _error_message = tr("Decoder reported an error"); - break; - } - - { - lock_guard lock(_mutex); - _samples_decoded = chunk_end; - } - } + do { + decode_data(*sample_count, unit_size, session); + } while(_error_message.isEmpty() && (sample_count = wait_for_data())); // Destroy the session srd_session_destroy(session); @@ -330,7 +386,7 @@ void DecoderStack::annotation_callback(srd_proto_data *pdata, void *decoder) DecoderStack *const d = (DecoderStack*)decoder; assert(d); - lock_guard lock(d->_mutex); + lock_guard lock(d->_output_mutex); const Annotation a(pdata); @@ -363,8 +419,31 @@ void DecoderStack::annotation_callback(srd_proto_data *pdata, void *decoder) // Add the annotation (*row_iter).second.push_annotation(a); +} - d->new_decode_data(); +void DecoderStack::on_new_frame() +{ + begin_decode(); +} + +void DecoderStack::on_data_received() +{ + { + unique_lock lock(_input_mutex); + if (_snapshot) + _sample_count = _snapshot->get_sample_count(); + } + _input_cond.notify_one(); +} + +void DecoderStack::on_frame_ended() +{ + { + unique_lock lock(_input_mutex); + if (_snapshot) + _frame_complete = true; + } + _input_cond.notify_one(); } } // namespace data