You will instantiate this class when you want to decode a single utterance using the online-decoding setup for neural nets. More...

#include <online-nnet2-decoding-threaded.h>

Collaboration diagram for SingleUtteranceNnet2DecoderThreaded:

[legend]

Public Member Functions
	SingleUtteranceNnet2DecoderThreaded (const OnlineNnet2DecodingThreadedConfig &config, const TransitionModel &tmodel, const nnet2::AmNnet &am_nnet, const fst::Fst< fst::StdArc > &fst, const OnlineNnet2FeaturePipelineInfo &feature_info, const OnlineIvectorExtractorAdaptationState &adaptation_state, const OnlineCmvnState &cmvn_state)

void	AcceptWaveform (BaseFloat samp_freq, const VectorBase< BaseFloat > &wave_part)
	You call this to provide this class with more waveform to decode. More...

int32	NumWaveformPiecesPending ()
	Returns the number of pieces of waveform that are still waiting to be processed. More...

void	InputFinished ()
	You call this to inform the class that no more waveform will be provided; this allows it to flush out the last few frames of features, and is necessary if you want to call Wait() to wait until all decoding is done. More...

void	TerminateDecoding ()
	You can call this if you don't want the decoding to proceed further with this utterance. More...

void	Wait ()
	This call will block until all the data has been decoded; it must only be called after either InputFinished() has been called or TerminateDecoding() has been called; otherwise, to call it is an error. More...

void	FinalizeDecoding ()
	Finalizes the decoding. More...

int32	NumFramesReceivedApprox () const
	Returns approximately (ignoring end effects), the number of frames of data that we expect given the amount of data that the pipeline has received via AcceptWaveform(). More...

int32	NumFramesDecoded () const
	Returns the number of frames currently decoded. More...

void	GetLattice (bool end_of_utterance, CompactLattice clat, BaseFloat final_relative_cost) const
	Gets the lattice. More...

void	GetBestPath (bool end_of_utterance, Lattice best_path, BaseFloat final_relative_cost) const
	Outputs an FST corresponding to the single best path through the current lattice. More...

bool	EndpointDetected (const OnlineEndpointConfig &config)
	This function calls EndpointDetected from online-endpoint.h, with the required arguments. More...

void	GetAdaptationState (OnlineIvectorExtractorAdaptationState *adaptation_state)
	Outputs the adaptation state of the feature pipeline to "adaptation_state". More...

void	GetCmvnState (OnlineCmvnState *cmvn_state)
	Outputs the OnlineCmvnState of the feature pipeline to "cmvn_stat". More...

BaseFloat	GetRemainingWaveform (Vector< BaseFloat > *waveform_out) const
	Gets the remaining, un-decoded part of the waveform and returns the sample rate. More...

	~SingleUtteranceNnet2DecoderThreaded ()

Private Member Functions
void	AbortAllThreads (bool error)

void	WaitForAllThreads ()

bool	RunNnetEvaluationInternal ()

void	ProcessLoglikes (const CuVector< BaseFloat > &log_inv_prior, CuMatrixBase< BaseFloat > *loglikes)

bool	FeatureComputation (int32 num_frames_output)

bool	RunDecoderSearchInternal ()

Static Private Member Functions
static void	RunNnetEvaluation (SingleUtteranceNnet2DecoderThreaded *me)

static void	RunDecoderSearch (SingleUtteranceNnet2DecoderThreaded *me)

Private Attributes
OnlineNnet2DecodingThreadedConfig	config_

const nnet2::AmNnet &	am_nnet_

const TransitionModel &	tmodel_

BaseFloat	sampling_rate_

int64	num_samples_received_

bool	input_finished_

std::deque< Vector< BaseFloat > *>	input_waveform_

ThreadSynchronizer	waveform_synchronizer_

OnlineNnet2FeaturePipeline	feature_pipeline_

std::mutex	feature_pipeline_mutex_

std::deque< Vector< BaseFloat > *>	processed_waveform_

int64	num_samples_discarded_

OnlineSilenceWeighting	silence_weighting_

std::mutex	silence_weighting_mutex_

DecodableMatrixMappedOffset	decodable_

int32	num_frames_decoded_

ThreadSynchronizer	decodable_synchronizer_

LatticeFasterOnlineDecoder	decoder_

std::mutex	decoder_mutex_

std::thread	threads_ [2]

bool	abort_

bool	error_

Detailed Description

You will instantiate this class when you want to decode a single utterance using the online-decoding setup for neural nets.

Each time this class is created, it creates three background threads, and the feature extraction, neural net evaluation, and search aspects of decoding all happen in different threads. Note: we assume that all calls to its public interface happen from a single thread.

Definition at line 190 of file online-nnet2-decoding-threaded.h.

Constructor & Destructor Documentation

◆ SingleUtteranceNnet2DecoderThreaded()

SingleUtteranceNnet2DecoderThreaded	(	const OnlineNnet2DecodingThreadedConfig &	config,
		const TransitionModel &	tmodel,
		const nnet2::AmNnet &	am_nnet,
		const fst::Fst< fst::StdArc > &	fst,
		const OnlineNnet2FeaturePipelineInfo &	feature_info,
		const OnlineIvectorExtractorAdaptationState &	adaptation_state,
		const OnlineCmvnState &	cmvn_state
	)

Definition at line 112 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::decoder_, SingleUtteranceNnet2DecoderThreaded::feature_pipeline_, LatticeFasterDecoderTpl< FST, Token >::InitDecoding(), SingleUtteranceNnet2DecoderThreaded::RunDecoderSearch(), SingleUtteranceNnet2DecoderThreaded::RunNnetEvaluation(), OnlineNnet2FeaturePipeline::SetAdaptationState(), OnlineNnet2FeaturePipeline::SetCmvnState(), and SingleUtteranceNnet2DecoderThreaded::threads_.

                                       :
   config_(config), am_nnet_(am_nnet), tmodel_(tmodel), sampling_rate_(0.0),
   num_samples_received_(0), input_finished_(false),
   feature_pipeline_(feature_info),
   num_samples_discarded_(0),
   silence_weighting_(tmodel, feature_info.silence_weighting_config),
   decodable_(tmodel),
   num_frames_decoded_(0), decoder_(fst, config_.decoder_opts),
   abort_(false), error_(false) {
   // if the user supplies an adaptation state that was not freshly initialized,
   // it means that we take the adaptation state from the previous
   // utterance(s)... this only makes sense if theose previous utterance(s) are
   // believed to be from the same speaker.
   feature_pipeline_.SetAdaptationState(adaptation_state);
   feature_pipeline_.SetCmvnState(cmvn_state);
   // spawn threads.
   threads_[0] = std::thread(RunNnetEvaluation, this);
   decoder_.InitDecoding();
   threads_[1] = std::thread(RunDecoderSearch, this);
 }

◆ ~SingleUtteranceNnet2DecoderThreaded()

~SingleUtteranceNnet2DecoderThreaded ( )

Definition at line 141 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::abort_, SingleUtteranceNnet2DecoderThreaded::AbortAllThreads(), SingleUtteranceNnet2DecoderThreaded::input_waveform_, SingleUtteranceNnet2DecoderThreaded::processed_waveform_, and SingleUtteranceNnet2DecoderThreaded::WaitForAllThreads().

                                                                           {
   if (!abort_) {
     // If we have not already started the process of aborting the threads, do so now.
     bool error = false;
     AbortAllThreads(error);
   }
   // join all the threads (this avoids leaving zombie threads around, or threads
   // that might be accessing deconstructed object).
   WaitForAllThreads();
   while (!input_waveform_.empty()) {
     delete input_waveform_.front();
     input_waveform_.pop_front();
   }
   while (!processed_waveform_.empty()) {
     delete processed_waveform_.front();
     processed_waveform_.pop_front();
   }
 }

Member Function Documentation

◆ AbortAllThreads()

void AbortAllThreads ( bool error )

private

Definition at line 344 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::abort_, SingleUtteranceNnet2DecoderThreaded::decodable_synchronizer_, SingleUtteranceNnet2DecoderThreaded::error_, ThreadSynchronizer::SetAbort(), and SingleUtteranceNnet2DecoderThreaded::waveform_synchronizer_.

Referenced by SingleUtteranceNnet2DecoderThreaded::RunDecoderSearch(), SingleUtteranceNnet2DecoderThreaded::RunNnetEvaluation(), SingleUtteranceNnet2DecoderThreaded::TerminateDecoding(), and SingleUtteranceNnet2DecoderThreaded::~SingleUtteranceNnet2DecoderThreaded().

                                                                     {
   abort_ = true;
   if (error)
     error_ = true;
   waveform_synchronizer_.SetAbort();
   decodable_synchronizer_.SetAbort();
 }

◆ AcceptWaveform()

void AcceptWaveform	(	BaseFloat	samp_freq,
		const VectorBase< BaseFloat > &	wave_part
	)

You call this to provide this class with more waveform to decode.

This call is, for all practical purposes, non-blocking.

Definition at line 160 of file online-nnet2-decoding-threaded.cc.

References VectorBase< Real >::Dim(), SingleUtteranceNnet2DecoderThreaded::input_waveform_, KALDI_ASSERT, KALDI_ERR, ThreadSynchronizer::kProducer, ThreadSynchronizer::Lock(), SingleUtteranceNnet2DecoderThreaded::num_samples_received_, SingleUtteranceNnet2DecoderThreaded::sampling_rate_, ThreadSynchronizer::UnlockSuccess(), and SingleUtteranceNnet2DecoderThreaded::waveform_synchronizer_.

                                             {
   if (sampling_rate_ <= 0.0)
     sampling_rate_ = sampling_rate;
   else {
     KALDI_ASSERT(sampling_rate == sampling_rate_);
   }
   num_samples_received_ += wave_part.Dim();
 
   if (wave_part.Dim() == 0) return;
   if (!waveform_synchronizer_.Lock(ThreadSynchronizer::kProducer)) {
     KALDI_ERR << "Failure locking mutex: decoding aborted.";
   }
 
   Vector<BaseFloat> *new_part = new Vector<BaseFloat>(wave_part);
   input_waveform_.push_back(new_part);
   // we always unlock with success because there is no buffer size limitation
   // for the waveform so no reason why we might wait.
   waveform_synchronizer_.UnlockSuccess(ThreadSynchronizer::kProducer);
 }

◆ EndpointDetected()

bool EndpointDetected ( const OnlineEndpointConfig & config )

This function calls EndpointDetected from online-endpoint.h, with the required arguments.

Definition at line 651 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::decoder_, SingleUtteranceNnet2DecoderThreaded::decoder_mutex_, kaldi::EndpointDetected(), SingleUtteranceNnet2DecoderThreaded::feature_pipeline_, OnlineNnet2FeaturePipeline::FrameShiftInSeconds(), and SingleUtteranceNnet2DecoderThreaded::tmodel_.

                                         {
   std::lock_guard<std::mutex> lock(decoder_mutex_);
   return kaldi::EndpointDetected(config, tmodel_,
                                  feature_pipeline_.FrameShiftInSeconds(),
                                  decoder_);
 }

◆ FeatureComputation()

bool FeatureComputation ( int32 num_frames_output )

private

Definition at line 412 of file online-nnet2-decoding-threaded.cc.

Referenced by SingleUtteranceNnet2DecoderThreaded::RunNnetEvaluationInternal().

                                {
 
   int32 num_frames_ready = feature_pipeline_.NumFramesReady(),
       num_frames_usable = num_frames_ready - num_frames_consumed;
   bool features_done = feature_pipeline_.IsLastFrame(num_frames_ready - 1);
   KALDI_ASSERT(num_frames_usable >= 0);
   if (features_done) {
     return true;  // nothing to do. (but not an error).
   } else {
     if (num_frames_usable >= config_.nnet_batch_size)
       return true;  // We don't need more data yet.
 
     // Now try to get more data, if we can.
     if (!waveform_synchronizer_.Lock(ThreadSynchronizer::kConsumer)) {
       return false;
     }
     // we've got the lock.
     if (input_waveform_.empty()) {  // we got no data
       if (input_finished_ &&
           !feature_pipeline_.IsLastFrame(feature_pipeline_.NumFramesReady()-1)) {
         // the main thread called InputFinished() and set input_finished_, and
         // we haven't yet registered that fact.  This is progress so
         // unlock with UnlockSuccess().
         feature_pipeline_.InputFinished();
         return waveform_synchronizer_.UnlockSuccess(ThreadSynchronizer::kConsumer);
       } else {
         // there is no progress.  Unlock with UnlockFailure() so the next call to
         // waveform_synchronizer_.Lock() will lock.
         return waveform_synchronizer_.UnlockFailure(ThreadSynchronizer::kConsumer);
       }
     } else {  // we got some data.  Only take enough of the waveform to
               // give us a maximum nnet batch size of frames to decode.
       while (num_frames_usable < config_.nnet_batch_size &&
              !input_waveform_.empty()) {
         feature_pipeline_.AcceptWaveform(sampling_rate_, *input_waveform_.front());
         processed_waveform_.push_back(input_waveform_.front());
         input_waveform_.pop_front();
         num_frames_ready = feature_pipeline_.NumFramesReady();
         num_frames_usable = num_frames_ready - num_frames_consumed;
       }
       // Delete already-processed pieces of waveform if we have already decoded
       // those frames.  (If not already decoded, we keep them around for the
       // sake of GetRemainingWaveform()).
       int32 samples_shift_per_frame =
           sampling_rate_ * feature_pipeline_.FrameShiftInSeconds();
       while (!processed_waveform_.empty() &&
              num_samples_discarded_ + processed_waveform_.front()->Dim() <
              samples_shift_per_frame * num_frames_decoded_) {
         num_samples_discarded_ += processed_waveform_.front()->Dim();
         delete processed_waveform_.front();
         processed_waveform_.pop_front();
       }
       return waveform_synchronizer_.UnlockSuccess(ThreadSynchronizer::kConsumer);
     }
   }
 }

◆ FinalizeDecoding()

void FinalizeDecoding ( )

Finalizes the decoding.

Cleans up and prunes remaining tokens, so the final lattice is faster to obtain. May not be called unless either InputFinished() or TerminateDecoding() has been called. If InputFinished() was called, it calls Wait() to ensure that the decoding has finished (it's not an error if you already called Wait()).

Definition at line 228 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::decoder_, LatticeFasterDecoderTpl< FST, Token >::FinalizeDecoding(), KALDI_ERR, and SingleUtteranceNnet2DecoderThreaded::threads_.

                                                            {
   if (threads_[0].joinable()) {
     KALDI_ERR << "It is an error to call FinalizeDecoding before Wait().";
   }
   decoder_.FinalizeDecoding();
 }

◆ GetAdaptationState()

void GetAdaptationState ( OnlineIvectorExtractorAdaptationState * adaptation_state )

Outputs the adaptation state of the feature pipeline to "adaptation_state".

This mostly stores stats for iVector estimation, and will generally be called at the end of an utterance, assuming it's a scenario where each speaker is seen for more than one utterance. You may only call this function after either calling TerminateDecoding() or InputFinished, and then Wait(). Otherwise it is an error.

Definition at line 283 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::feature_pipeline_, SingleUtteranceNnet2DecoderThreaded::feature_pipeline_mutex_, and OnlineNnet2FeaturePipeline::GetAdaptationState().

                                                              {
   std::lock_guard<std::mutex> lock(feature_pipeline_mutex_);
   // If this blocks, it shouldn't be for very long.
   feature_pipeline_.GetAdaptationState(adaptation_state);
 }

◆ GetBestPath()

void GetBestPath	(	bool	end_of_utterance,
		Lattice *	best_path,
		BaseFloat *	final_relative_cost
	)		const

Outputs an FST corresponding to the single best path through the current lattice.

If "use_final_probs" is true AND we reached the final-state of the graph then it will include those as final-probs, else it will treat all final-probs as one. If no frames have been decoded yet, it will set best_path to a lattice with a single state that is final and with unit weight (no cost). The output to final_relative_cost (if non-NULL) is a number >= 0 that's closer to 0 if a final-state were close to the best-likelihood state active on the last frame, at the time we got the best path.

Definition at line 323 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::decoder_, SingleUtteranceNnet2DecoderThreaded::decoder_mutex_, LatticeFasterDecoderTpl< FST, Token >::FinalRelativeCost(), LatticeFasterOnlineDecoderTpl< FST >::GetBestPath(), LatticeFasterDecoderTpl< FST, Token >::NumFramesDecoded(), and LatticeWeightTpl< BaseFloat >::One().

                                           {
   std::lock_guard<std::mutex> lock(decoder_mutex_);
   if (decoder_.NumFramesDecoded() == 0) {
     // It's possible that this if-statement is not necessary because we'd get this
     // anyway if we just called GetBestPath on the decoder.
     best_path->DeleteStates();
     best_path->SetFinal(best_path->AddState(),
                         LatticeWeight::One());
     if (final_relative_cost != NULL)
       *final_relative_cost = std::numeric_limits<BaseFloat>::infinity();
   } else {
     decoder_.GetBestPath(best_path,
                          end_of_utterance);
     if (final_relative_cost != NULL)
       *final_relative_cost = decoder_.FinalRelativeCost();
   }
 }

◆ GetCmvnState()

void GetCmvnState ( OnlineCmvnState * cmvn_state )

Outputs the OnlineCmvnState of the feature pipeline to "cmvn_stat".

This stores cmvn stats for the non-iVector features, and will be called at the end of an utterance, assuming it's a scenario where each speaker is seen for more than one utterance. You may only call this function after either calling TerminateDecoding() or InputFinished, and then Wait(). Otherwise it is an error.

Definition at line 290 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::feature_pipeline_, SingleUtteranceNnet2DecoderThreaded::feature_pipeline_mutex_, and OnlineNnet2FeaturePipeline::GetCmvnState().

                                  {
   std::lock_guard<std::mutex> lock(feature_pipeline_mutex_);
   // If this blocks, it shouldn't be for very long.
   feature_pipeline_.GetCmvnState(cmvn_state);
 }

◆ GetLattice()

void GetLattice	(	bool	end_of_utterance,
		CompactLattice *	clat,
		BaseFloat *	final_relative_cost
	)		const

Gets the lattice.

The output lattice has any acoustic scaling in it (which will typically be desirable in an online-decoding context); if you want an un-scaled lattice, scale it using ScaleLattice() with the inverse of the acoustic weight. "end_of_utterance" will be true if you want the final-probs to be included. If this is at the end of the utterance, you might want to first call FinalizeDecoding() first; this will make this call return faster. If no frames have been decoded yet, it will set clat to a lattice with a single state that is final and with unit weight (no cost or alignment). The output to final_relative_cost (if non-NULL) is a number >= 0 that's closer to 0 if a final-state was close to the best-likelihood state active on the last frame, at the time we obtained the lattice.

Definition at line 297 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::config_, SingleUtteranceNnet2DecoderThreaded::decoder_, SingleUtteranceNnet2DecoderThreaded::decoder_mutex_, OnlineNnet2DecodingThreadedConfig::decoder_opts, LatticeFasterDecoderConfig::det_opts, LatticeFasterDecoderConfig::determinize_lattice, fst::DeterminizeLatticePhonePrunedWrapper(), LatticeFasterDecoderTpl< FST, Token >::FinalRelativeCost(), LatticeFasterDecoderTpl< FST, Token >::GetRawLattice(), KALDI_ERR, LatticeFasterDecoderConfig::lattice_beam, LatticeFasterDecoderTpl< FST, Token >::NumFramesDecoded(), CompactLatticeWeightTpl< WeightType, IntType >::One(), and SingleUtteranceNnet2DecoderThreaded::tmodel_.

                                           {
   clat->DeleteStates();
   decoder_mutex_.lock();
   if (final_relative_cost != NULL)
     *final_relative_cost = decoder_.FinalRelativeCost();
   if (decoder_.NumFramesDecoded() == 0) {
     decoder_mutex_.unlock();
     clat->SetFinal(clat->AddState(),
                    CompactLatticeWeight::One());
     return;
   }
   Lattice raw_lat;
   decoder_.GetRawLattice(&raw_lat, end_of_utterance);
   decoder_mutex_.unlock();
 
   if (!config_.decoder_opts.determinize_lattice)
     KALDI_ERR << "--determinize-lattice=false option is not supported at the moment";
 
   BaseFloat lat_beam = config_.decoder_opts.lattice_beam;
   DeterminizeLatticePhonePrunedWrapper(
       tmodel_, &raw_lat, lat_beam, clat, config_.decoder_opts.det_opts);
 }

◆ GetRemainingWaveform()

BaseFloat GetRemainingWaveform ( Vector< BaseFloat > * waveform_out ) const

Gets the remaining, un-decoded part of the waveform and returns the sample rate.

May only be called after Wait(), and it only makes sense to call this if you called TerminateDecoding() before Wait(). The idea is that you can then provide this un-decoded piece of waveform to another decoder.

Definition at line 235 of file online-nnet2-decoding-threaded.cc.

References VectorBase< Real >::Dim(), SingleUtteranceNnet2DecoderThreaded::feature_pipeline_, OnlineNnet2FeaturePipeline::FrameShiftInSeconds(), rnnlm::i, SingleUtteranceNnet2DecoderThreaded::input_waveform_, KALDI_ASSERT, KALDI_ERR, kaldi::kUndefined, SingleUtteranceNnet2DecoderThreaded::num_frames_decoded_, SingleUtteranceNnet2DecoderThreaded::num_samples_discarded_, SingleUtteranceNnet2DecoderThreaded::processed_waveform_, VectorBase< Real >::Range(), Vector< Real >::Resize(), SingleUtteranceNnet2DecoderThreaded::sampling_rate_, and SingleUtteranceNnet2DecoderThreaded::threads_.

                                        {
   if (threads_[0].joinable()) {
     KALDI_ERR << "It is an error to call GetRemainingWaveform before Wait().";
   }
   int64 num_samples_stored = 0;  // number of samples we still have.
   std::vector< Vector<BaseFloat>* > all_pieces;
   std::deque< Vector<BaseFloat>* >::const_iterator iter;
   for (iter = processed_waveform_.begin(); iter != processed_waveform_.end();
        ++iter) {
     num_samples_stored += (*iter)->Dim();
     all_pieces.push_back(*iter);
   }
   for (iter = input_waveform_.begin(); iter != input_waveform_.end(); ++iter) {
     num_samples_stored += (*iter)->Dim();
     all_pieces.push_back(*iter);
   }
   int64 samples_shift_per_frame =
       sampling_rate_ * feature_pipeline_.FrameShiftInSeconds();
   int64 num_samples_to_discard = samples_shift_per_frame * num_frames_decoded_;
   KALDI_ASSERT(num_samples_to_discard >= num_samples_discarded_);
 
   // num_samp_discard is how many samples we must discard from our stored
   // samples.
   int64 num_samp_discard = num_samples_to_discard - num_samples_discarded_,
       num_samp_keep = num_samples_stored - num_samp_discard;
   KALDI_ASSERT(num_samp_discard <= num_samples_stored && num_samp_keep >= 0);
   waveform->Resize(num_samp_keep, kUndefined);
   int32 offset = 0; // offset in output waveform.  assume output waveform is no
                     // larger than int32.
   for (size_t i = 0; i < all_pieces.size(); i++) {
     Vector<BaseFloat> *this_piece = all_pieces[i];
     int32 this_dim = this_piece->Dim();
     if (num_samp_discard >= this_dim) {
       num_samp_discard -= this_dim;
     } else {
       // normal case is num_samp_discard = 0.
       int32 this_dim_keep = this_dim - num_samp_discard;
       waveform->Range(offset, this_dim_keep).CopyFromVec(
           this_piece->Range(num_samp_discard, this_dim_keep));
       offset += this_dim_keep;
       num_samp_discard = 0;
     }
   }
   KALDI_ASSERT(offset == num_samp_keep && num_samp_discard == 0);
   return sampling_rate_;
 }

◆ InputFinished()

void InputFinished ( )

You call this to inform the class that no more waveform will be provided; this allows it to flush out the last few frames of features, and is necessary if you want to call Wait() to wait until all decoding is done.

After calling InputFinished() you cannot call AcceptWaveform any more.

Definition at line 203 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::input_finished_, KALDI_ASSERT, KALDI_ERR, ThreadSynchronizer::kProducer, ThreadSynchronizer::Lock(), ThreadSynchronizer::UnlockSuccess(), and SingleUtteranceNnet2DecoderThreaded::waveform_synchronizer_.

                                                         {
   // setting input_finished_ = true informs the feature-processing pipeline
   // to expect no more input, and to flush out the last few frames if there
   // is any latency in the pipeline (e.g. due to pitch).
   if (!waveform_synchronizer_.Lock(ThreadSynchronizer::kProducer)) {
     KALDI_ERR << "Failure locking mutex: decoding aborted.";
   }
   KALDI_ASSERT(!input_finished_ && "InputFinished called twice");
   input_finished_ = true;
   waveform_synchronizer_.UnlockSuccess(ThreadSynchronizer::kProducer);
 }

◆ NumFramesDecoded()

int32 NumFramesDecoded ( ) const

Returns the number of frames currently decoded.

Caution: don't rely on the lattice having exactly this number if you get it after this call, as it may increase after this– unless you've already called either TerminateDecoding() or InputFinished(), followed by Wait().

Definition at line 352 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::decoder_, SingleUtteranceNnet2DecoderThreaded::decoder_mutex_, and LatticeFasterDecoderTpl< FST, Token >::NumFramesDecoded().

                                                                   {
   std::lock_guard<std::mutex> lock(decoder_mutex_);
   return decoder_.NumFramesDecoded();
 }

◆ NumFramesReceivedApprox()

int32 NumFramesReceivedApprox ( ) const

Returns *approximately* (ignoring end effects), the number of frames of data that we expect given the amount of data that the pipeline has received via AcceptWaveform().

(ignores small end effects). This might be useful in application code to compare with NumFramesDecoded() and gauge how much latency there is.

Definition at line 198 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::feature_pipeline_, OnlineNnet2FeaturePipeline::FrameShiftInSeconds(), SingleUtteranceNnet2DecoderThreaded::num_samples_received_, and SingleUtteranceNnet2DecoderThreaded::sampling_rate_.

                                                                          {
   return num_samples_received_ /
       (sampling_rate_ * feature_pipeline_.FrameShiftInSeconds());
 }

◆ NumWaveformPiecesPending()

int32 NumWaveformPiecesPending ( )

Returns the number of pieces of waveform that are still waiting to be processed.

This may be useful for calling code to judge whether to supply more waveform or to wait.

Definition at line 182 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::input_waveform_, KALDI_ERR, ThreadSynchronizer::kProducer, ThreadSynchronizer::Lock(), ThreadSynchronizer::UnlockSuccess(), and SingleUtteranceNnet2DecoderThreaded::waveform_synchronizer_.

                                                                     {
   // Note RE locking: what we really want here is just to lock the mutex.  As a
   // side effect, because of the way the synchronizer code works, it will also
   // increment the semaphore and might wake up the consumer thread.  This will
   // possibly make it do a little useless work (go around a loop once), but
   // won't really do any harm.  Perhaps we should have implemented a version of
   // the Lock function that takes no arguments.
   if (!waveform_synchronizer_.Lock(ThreadSynchronizer::kProducer)) {
     KALDI_ERR << "Failure locking mutex: decoding aborted.";
   }
   int32 ans = input_waveform_.size();
   waveform_synchronizer_.UnlockSuccess(ThreadSynchronizer::kProducer);
   return ans;
 }

◆ ProcessLoglikes()

void ProcessLoglikes	(	const CuVector< BaseFloat > &	log_inv_prior,
		CuMatrixBase< BaseFloat > *	loglikes
	)

private

Definition at line 395 of file online-nnet2-decoding-threaded.cc.

References OnlineNnet2DecodingThreadedConfig::acoustic_scale, CuMatrixBase< Real >::AddVecToRows(), CuMatrixBase< Real >::ApplyFloor(), CuMatrixBase< Real >::ApplyLog(), SingleUtteranceNnet2DecoderThreaded::config_, CuMatrixBase< Real >::NumRows(), and CuMatrixBase< Real >::Scale().

Referenced by SingleUtteranceNnet2DecoderThreaded::RunNnetEvaluationInternal().

                                           {
   if (cu_loglikes->NumRows() != 0) {
     cu_loglikes->ApplyFloor(1.0e-20);
     cu_loglikes->ApplyLog();
     // take the log-posteriors and turn them into pseudo-log-likelihoods by
     // dividing by the pdf priors; then scale by the acoustic scale.
     cu_loglikes->AddVecToRows(1.0, log_inv_prior);
     cu_loglikes->Scale(config_.acoustic_scale);
   }
 }

◆ RunDecoderSearch()

void RunDecoderSearch ( SingleUtteranceNnet2DecoderThreaded * me )

staticprivate

Definition at line 371 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::abort_, SingleUtteranceNnet2DecoderThreaded::AbortAllThreads(), KALDI_ERR, KALDI_WARN, and SingleUtteranceNnet2DecoderThreaded::RunDecoderSearchInternal().

Referenced by SingleUtteranceNnet2DecoderThreaded::SingleUtteranceNnet2DecoderThreaded().

                                              {
   try {
     if (!me->RunDecoderSearchInternal() && !me->abort_)
       KALDI_ERR << "Returned abnormally and abort was not called";
   } catch(const std::exception &e) {
     KALDI_WARN << "Caught exception: " << e.what();
     // if an error happened in one thread, we need to make sure the other threads can exit too.
     bool error = true;
     me->AbortAllThreads(error);
   }
 }

◆ RunDecoderSearchInternal()

bool RunDecoderSearchInternal ( )

private

Definition at line 615 of file online-nnet2-decoding-threaded.cc.

Referenced by SingleUtteranceNnet2DecoderThreaded::RunDecoderSearch().

                                                                    {
   int32 num_frames_decoded = 0;  // this is just a copy of decoder_->NumFramesDecoded();
   while (true) {  // decode at most one frame each loop.
     if (!decodable_synchronizer_.Lock(ThreadSynchronizer::kConsumer))
       return false; // AbortAllThreads() called.
     if (decodable_.NumFramesReady() <= num_frames_decoded) {
       // no frames available to decode.
       KALDI_ASSERT(decodable_.NumFramesReady() == num_frames_decoded);
       if (decodable_.IsLastFrame(num_frames_decoded - 1)) {
         decodable_synchronizer_.UnlockSuccess(ThreadSynchronizer::kConsumer);
         return true;  // exit from this thread; we're done.
       } else {
         // we were not able to advance the decoding due to no available
         // input.  The next call will ensure that the next call to
         // decodable_synchronizer_.Lock() will wait.
         if (!decodable_synchronizer_.UnlockFailure(ThreadSynchronizer::kConsumer))
           return false;
       }
     } else {
       // Decode at most config_.decode_batch_size frames (e.g. 1 or 2).
       decoder_mutex_.lock();
       decoder_.AdvanceDecoding(&decodable_, config_.decode_batch_size);
       num_frames_decoded = decoder_.NumFramesDecoded();
       if (silence_weighting_.Active()) {
         std::lock_guard<std::mutex> lock(silence_weighting_mutex_);
         // the next function does not trace back all the way; it's very fast.
         silence_weighting_.ComputeCurrentTraceback(decoder_);
       }
       decoder_mutex_.unlock();
       num_frames_decoded_ = num_frames_decoded;
       if (!decodable_synchronizer_.UnlockSuccess(ThreadSynchronizer::kConsumer))
         return false;
     }
   }
 }

◆ RunNnetEvaluation()

void RunNnetEvaluation ( SingleUtteranceNnet2DecoderThreaded * me )

staticprivate

Definition at line 357 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::abort_, SingleUtteranceNnet2DecoderThreaded::AbortAllThreads(), KALDI_ERR, KALDI_WARN, and SingleUtteranceNnet2DecoderThreaded::RunNnetEvaluationInternal().

Referenced by SingleUtteranceNnet2DecoderThreaded::SingleUtteranceNnet2DecoderThreaded().

                                              {
   try {
     if (!me->RunNnetEvaluationInternal() && !me->abort_)
       KALDI_ERR << "Returned abnormally and abort was not called";
   } catch(const std::exception &e) {
     KALDI_WARN << "Caught exception: " << e.what();
     // if an error happened in one thread, we need to make sure the other
     // threads can exit too.
     bool error = true;
     me->AbortAllThreads(error);
   }
 }

◆ RunNnetEvaluationInternal()

bool RunNnetEvaluationInternal ( )

private

Definition at line 470 of file online-nnet2-decoding-threaded.cc.

Referenced by SingleUtteranceNnet2DecoderThreaded::RunNnetEvaluation().

                                                                     {
   // if any of the Lock/Unlock functions return false, it's because AbortAllThreads()
   // was called.
 
   // This object is responsible for keeping track of the context, and avoiding
   // re-computing things we've already computed.
   bool pad_input = true;
   nnet2::NnetOnlineComputer computer(am_nnet_.GetNnet(), pad_input);
 
   // we declare the following as CuVector just to enable GPU support, but
   // we expect this code to be run on CPU in the normal case.
   CuVector<BaseFloat> log_inv_prior(am_nnet_.Priors());
   log_inv_prior.ApplyFloor(1.0e-20);  // should have no effect.
   log_inv_prior.ApplyLog();
   log_inv_prior.Scale(-1.0);
 
   // we'll have num_frames_consumed >= num_frames_output; num_frames_consumed is
   // the number of feature frames consumed by the nnet computation,
   // num_frames_output is the number of frames of loglikes the nnet computation
   // has produced, which may be less than num_frames_consumed due to the
   // right-context of the network.
   int32 num_frames_consumed = 0, num_frames_output = 0;
 
   while (true) {
     bool last_time = false;
 
     /****** Begin locking of feature pipeline mutex. ******/
     feature_pipeline_mutex_.lock();
     if (!FeatureComputation(num_frames_consumed)) {  // error
       feature_pipeline_mutex_.unlock();
       return false;
     }
     // take care of silence weighting.
     if (silence_weighting_.Active() &&
         feature_pipeline_.IvectorFeature() != NULL) {
       silence_weighting_mutex_.lock();
       std::vector<std::pair<int32, BaseFloat> > delta_weights;
       silence_weighting_.GetDeltaWeights(
           feature_pipeline_.IvectorFeature()->NumFramesReady(),
           &delta_weights);
       silence_weighting_mutex_.unlock();
       feature_pipeline_.IvectorFeature()->UpdateFrameWeights(delta_weights);
     }
 
     int32 num_frames_ready = feature_pipeline_.NumFramesReady(),
         num_frames_usable = num_frames_ready - num_frames_consumed;
     bool features_done = feature_pipeline_.IsLastFrame(num_frames_ready - 1);
 
     int32 num_frames_evaluate = std::min<int32>(num_frames_usable,
                                                 config_.nnet_batch_size);
 
     Matrix<BaseFloat> feats;
     if (num_frames_evaluate > 0) {
       // we have something to do...
       feats.Resize(num_frames_evaluate, feature_pipeline_.Dim());
       for (int32 i = 0; i < num_frames_evaluate; i++) {
         int32 t = num_frames_consumed + i;
         SubVector<BaseFloat> feat(feats, i);
         feature_pipeline_.GetFrame(t, &feat);
       }
     }
     /****** End locking of feature pipeline mutex. ******/
     feature_pipeline_mutex_.unlock();
 
     CuMatrix<BaseFloat> cu_loglikes;
 
     if (feats.NumRows() == 0) {
       if (features_done) {
         // flush out the last few frames.  Note: this is the only place from
         // which we check feature_buffer_finished_, and we'll exit the loop, so
         // if we reach here it must be the first time it was true.
         last_time = true;
         computer.Flush(&cu_loglikes);
         ProcessLoglikes(log_inv_prior, &cu_loglikes);
       }
     } else {
       CuMatrix<BaseFloat> cu_feats;
       cu_feats.Swap(&feats);  // If we don't have a GPU (and not having a GPU is
                               // the normal expected use-case for this code),
                               // this would be a lightweight operation, swapping
                               // pointers.
 
       computer.Compute(cu_feats, &cu_loglikes);
       num_frames_consumed += cu_feats.NumRows();
       ProcessLoglikes(log_inv_prior, &cu_loglikes);
     }
 
     Matrix<BaseFloat> loglikes;
     loglikes.Swap(&cu_loglikes);  // If we don't have a GPU (and not having a
                                   // GPU is the normal expected use-case for
                                   // this code), this would be a lightweight
                                   // operation, swapping pointers.
 
 
     // OK, at this point we may have some newly created log-likes and we want to
     // give them to the decoding thread.
 
     int32 num_loglike_frames = loglikes.NumRows();
 
     if (num_loglike_frames != 0) {  // if we need to output some loglikes...
       while (true) {
         // we may have to grab and release the decodable mutex
         // a few times before it's ready to accept the loglikes.
         if (!decodable_synchronizer_.Lock(ThreadSynchronizer::kProducer))
           return false;
         int32 num_frames_decoded = num_frames_decoded_;
         // we can't have output fewer frames than were decoded.
         KALDI_ASSERT(num_frames_output >= num_frames_decoded);
         if (num_frames_output - num_frames_decoded <= config_.max_loglikes_copy) {
           // If we would have to copy fewer than config_.max_loglikes_copy
           // previously output log-likelihoods inside the decodable object, then
           // we go ahead and copy them to that object.
           int32 frames_to_discard = num_frames_decoded_ -
               decodable_.FirstAvailableFrame();
           KALDI_ASSERT(frames_to_discard >= 0);
           num_frames_output += num_loglike_frames;
           decodable_.AcceptLoglikes(&loglikes, frames_to_discard);
           if (!decodable_synchronizer_.UnlockSuccess(ThreadSynchronizer::kProducer))
             return false;
           break;  // break from the innermost while loop.
         } else {
           // There are too many frames already available to the decoder, that it
           // hasn't processed yet, and we don't want them to have to be copied
           // inside AcceptLoglikes(), so we wait for a bit.
           // we want the next call to Lock to block until the decoder has
           //  processed more frames.
           if (!decodable_synchronizer_.UnlockFailure(ThreadSynchronizer::kProducer))
             return false;
         }
       }
     }
     if (last_time) {
       // Inform the decodable object that there will be no more input.
       if (!decodable_synchronizer_.Lock(ThreadSynchronizer::kProducer))
         return false;
       decodable_.InputIsFinished();
       if (!decodable_synchronizer_.UnlockSuccess(ThreadSynchronizer::kProducer))
         return false;
       KALDI_ASSERT(num_frames_consumed == num_frames_output);
       return true;
     }
   }
 }

◆ TerminateDecoding()

void TerminateDecoding ( )

You can call this if you don't want the decoding to proceed further with this utterance.

It just won't do any more processing, but you can still use the lattice from the decoding that it's already done. Note: it may still continue decoding up to decode_batch_size (default: 2) frames of data before the decoding thread exits. You can call Wait() after calling this, if you want to wait for that.

Definition at line 215 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::AbortAllThreads().

                                                             {
   bool error = false;
   AbortAllThreads(error);
 }

◆ Wait()

void Wait ( )

This call will block until all the data has been decoded; it must only be called after either InputFinished() has been called or TerminateDecoding() has been called; otherwise, to call it is an error.

Definition at line 220 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::abort_, SingleUtteranceNnet2DecoderThreaded::input_finished_, KALDI_ERR, and SingleUtteranceNnet2DecoderThreaded::WaitForAllThreads().

                                                {
   if (!input_finished_ && !abort_) {
     KALDI_ERR << "You cannot call Wait() before calling either InputFinished() "
               << "or TerminateDecoding().";
   }
   WaitForAllThreads();
 }

◆ WaitForAllThreads()

void WaitForAllThreads ( )

private

Definition at line 385 of file online-nnet2-decoding-threaded.cc.

References SingleUtteranceNnet2DecoderThreaded::error_, rnnlm::i, KALDI_ERR, and SingleUtteranceNnet2DecoderThreaded::threads_.

Referenced by SingleUtteranceNnet2DecoderThreaded::Wait(), and SingleUtteranceNnet2DecoderThreaded::~SingleUtteranceNnet2DecoderThreaded().

                                                             {
   for (int32 i = 0; i < 2; i++) {  // there are 2 spawned threads.
     if (threads_[i].joinable())
       threads_[i].join();
   }
   if (error_)
     KALDI_ERR << "Error encountered during decoding.  See above.";
 }