use mutex to make thread safe

Author: berndpfrommer

include/ffmpeg_encoder_decoder/decoder.hpp

@@ -21,8 +21,10 @@
 #include <ffmpeg_encoder_decoder/types.hpp>
 #include <functional>
 #include <memory>
+#include <mutex>
 #include <rclcpp/rclcpp.hpp>
 #include <string>
+#include <thread>
 #include <unordered_map>
 
 extern "C" {
@@ -92,15 +94,19 @@ class Decoder
   Decoder();
 
   /**
-   * \brief Destructor. Calls reset();
+   * \brief Destructor.
    */
   ~Decoder();
 
   /**
    * Test if decoder is initialized.
    * \return true if the decoder is initialized.
    */
-  bool isInitialized() const { return (codecContext_ != NULL); }
+  bool isInitialized() const
+  {
+    Lock lock(mutex_);
+    return (codecContext_ != NULL);
+  }
 
   /**
    * \brief Initializes the decoder for a given codec and libav decoder.
@@ -160,36 +166,21 @@ class Decoder
    * \brief Overrides the default ("Decoder") logger.
    * \param logger the logger to override the default ("Decoder") with.
    */
-  void setLogger(rclcpp::Logger logger) { logger_ = logger; }
-
-  /**
-   * \brief Finds all hardware and software decoders for a given codec.
-   *
-   * Finds the name of all hardware and software decoders that match
-   * a certain codec (or encoder).
-   * \param codec name of the codec, i.e. h264, hevc etc
-   * \param hw_decoders non-null pointer for returning list of hardware decoders
-   * \param sw_decoders non-null pointer for returning list of software decoders
-   */
-  static void findDecoders(
-    const std::string & codec, std::vector<std::string> * hw_decoders,
-    std::vector<std::string> * sw_decoders);
-
-  /**
-   * \brief Finds all decoders that can decode a given codec.
-   *
-   * Finds the name of all hardware and software decoders (combined)
-   * that match a certain codec (or encoder).
-   * \param codec name of the codec, i.e. h264, hevc etc
-   * \return string with comma-separated list of libav decoders
-   */
-  static std::string findDecoders(const std::string & codec);
+  void setLogger(rclcpp::Logger logger)
+  {
+    Lock lock(mutex_);
+    logger_ = logger;
+  }
 
   /**
    * \brief Enables or disables performance measurements. Poorly tested, may be broken.
    * \param p set to true to enable performance debugging.
    */
-  void setMeasurePerformance(bool p) { measurePerformance_ = p; }
+  void setMeasurePerformance(bool p)
+  {
+    Lock lock(mutex_);
+    measurePerformance_ = p;
+  }
 
   /**
    * \brief Prints performance timers. Poorly tested, may be broken.
@@ -201,16 +192,41 @@ class Decoder
    * \brief resets performance debugging timers. Poorly tested, may be broken.
    */
   void resetTimers();
+
   /**
    * \brief adds AVOption setting to list of options to be applied before opening the encoder
    * \param key   name of AVOption to set, e.g. "preset"
    * \param value value of AVOption e.g. "slow"
    */
   void addAVOption(const std::string & key, const std::string & value)
   {
+    Lock lock(mutex_);
     avOptions_.push_back({key, value});
   }
 
+  /**
+   * \brief Finds all hardware and software decoders for a given codec.
+   *
+   * Finds the name of all hardware and software decoders that match
+   * a certain codec (or encoder).
+   * \param codec name of the codec, i.e. h264, hevc etc
+   * \param hw_decoders non-null pointer for returning list of hardware decoders
+   * \param sw_decoders non-null pointer for returning list of software decoders
+   */
+  static void findDecoders(
+    const std::string & codec, std::vector<std::string> * hw_decoders,
+    std::vector<std::string> * sw_decoders);
+
+  /**
+   * \brief Finds all decoders that can decode a given codec.
+   *
+   * Finds the name of all hardware and software decoders (combined)
+   * that match a certain codec (or encoder).
+   * \param codec name of the codec, i.e. h264, hevc etc
+   * \return string with comma-separated list of libav decoders
+   */
+  static std::string findDecoders(const std::string & codec);
+
   // ------------------- deprecated functions ---------------
   /**
    * \deprecated Use findDecoders(codec) instead.
@@ -220,17 +236,20 @@ class Decoder
   getDefaultEncoderToDecoderMap();
 
 private:
+  using Lock = std::unique_lock<std::mutex>;
   bool doInitDecoder(const std::string & encoding, const std::string & decoder);
   bool initDecoder(const std::string & encoding, const std::string & decoders);
   int receiveFrame();
   int convertFrameToMessage(const AVFrame * frame, const ImagePtr & image);
   void setAVOption(const std::string & field, const std::string & value);
   void setEncoding(const std::string & encoding);
+  void resetNoLock();
   // --------------- variables
   rclcpp::Logger logger_;
   Callback callback_;
   PTSMap ptsToStamp_;
   std::vector<std::pair<std::string, std::string>> avOptions_;
+  mutable std::mutex mutex_;
   // --- performance analysis
   bool measurePerformance_{false};
   TDiff tdiffTotal_;

src/decoder.cpp

@@ -27,9 +27,15 @@ namespace ffmpeg_encoder_decoder
 
 Decoder::Decoder() : logger_(rclcpp::get_logger("Decoder")) {}
 
-Decoder::~Decoder() { reset(); }
+Decoder::~Decoder() { resetNoLock(); }
 
 void Decoder::reset()
+{
+  Lock lock(mutex_);
+  resetNoLock();
+}
+
+void Decoder::resetNoLock()
 {
   if (codecContext_) {
     avcodec_free_context(&codecContext_);
@@ -55,6 +61,7 @@ void Decoder::reset()
 
 void Decoder::setOutputMessageEncoding(const std::string & output_encoding)
 {
+  Lock lock(mutex_);
   RCLCPP_INFO_STREAM(logger_, "forcing output encoding: " << output_encoding);
   outputMsgEncoding_ = output_encoding;
 }
@@ -80,6 +87,7 @@ void Decoder::setEncoding(const std::string & encoding)
 bool Decoder::initialize(
   const std::string & encoding, Callback callback, const std::string & decoder)
 {
+  Lock lock(mutex_);
   callback_ = callback;
   return (initDecoder(encoding, decoder));
 }
@@ -282,7 +290,7 @@ bool Decoder::doInitDecoder(const std::string & encoding, const std::string & de
     outputFrame_ = av_frame_alloc();
   } catch (const std::runtime_error & e) {
     RCLCPP_ERROR_STREAM(logger_, e.what());
-    reset();
+    resetNoLock();
     return (false);
   }
   RCLCPP_INFO_STREAM(logger_, "decoding with " << decoder);
@@ -291,6 +299,7 @@ bool Decoder::doInitDecoder(const std::string & encoding, const std::string & de
 
 bool Decoder::flush()
 {
+  Lock lock(mutex_);
   if (!codecContext_) {
     return (false);
   }
@@ -395,7 +404,8 @@ bool Decoder::decodePacket(
   const std::string & encoding, const uint8_t * data, size_t size, uint64_t pts,
   const std::string & frame_id, const rclcpp::Time & stamp)
 {
-  if (!isInitialized()) {
+  Lock lock(mutex_);
+  if (codecContext_ == nullptr) {
     RCLCPP_ERROR_STREAM(logger_, "decoder is not initialized!");
     return (false);
   }
@@ -439,10 +449,15 @@ bool Decoder::decodePacket(
   return (rret == AVERROR(EAGAIN));
 }
 
-void Decoder::resetTimers() { tdiffTotal_.reset(); }
+void Decoder::resetTimers()
+{
+  Lock lock(mutex_);
+  tdiffTotal_.reset();
+}
 
 void Decoder::printTimers(const std::string & prefix) const
 {
+  Lock lock(mutex_);
   RCLCPP_INFO_STREAM(logger_, prefix << " total decode: " << tdiffTotal_);
 }