[onert] Propagate async buffer fill errors to waiters (#16365)

It adds error handling mechanism for asynchronous buffer fill operations
in BulkPipelineModel to prevent indefinite blocking when async buffer fill
encounters exceptions.

ONE-DCO-1.0-Signed-off-by: Jonghwa Lee <jonghwa3.lee@samsung.com>

Author: batcheu

runtime/onert/backend/trix/ops/BulkPipelineModel.cc

@@ -184,7 +184,12 @@ void BulkPipelineModel::setNextModel(std::shared_ptr<BulkPipelineModel> next)
 void BulkPipelineModel::waitForBufferReady()
 {
   std::unique_lock<std::mutex> lock(_buffer_mutex);
-  _buffer_cv.wait(lock, [this] { return _buffer_ready.load(); });
+  _buffer_cv.wait(lock, [this] { return _buffer_ready.load() || _buffer_error.load(); });
+
+  if (_buffer_error.load())
+  {
+    throw std::runtime_error("Buffer fill failed");
+  }
 }
 
 void BulkPipelineModel::markBufferReady()
@@ -199,6 +204,7 @@ void BulkPipelineModel::markBufferReady()
 void BulkPipelineModel::startAsyncBufferFill()
 {
   _buffer_ready = false;
+  _buffer_error = false;
   _async_fill_future = std::async(std::launch::async, [this] {
     try
     {
@@ -207,6 +213,12 @@ void BulkPipelineModel::startAsyncBufferFill()
     }
     catch (const std::exception &e)
     {
+      {
+        std::lock_guard<std::mutex> lock(_buffer_mutex);
+        _buffer_ready = false;
+        _buffer_error = true;
+      }
+      _buffer_cv.notify_all();
       std::cerr << "Failed to fill buffers asynchronously: " << e.what() << std::endl;
     }
   });

runtime/onert/backend/trix/ops/BulkPipelineModel.h

@@ -106,6 +106,7 @@ class BulkPipelineModel
   std::mutex _buffer_mutex;
   std::condition_variable _buffer_cv;
   std::atomic<bool> _buffer_ready{false};
+  std::atomic<bool> _buffer_error{false};
   std::future<void> _async_fill_future;
 };
 

runtime/onert/backend/trix/ops/test/BulkPipelineModel.test.cc

@@ -114,3 +114,26 @@ TEST_F(BulkPipelineModelTest, test_model_release)
   EXPECT_EQ(model->modelId(), 0);
   EXPECT_EQ(model->metadata(), nullptr);
 }
+
+TEST_F(BulkPipelineModelTest, test_async_fill)
+{
+  model->initialize();
+  model->prepare();
+
+  std::shared_ptr<BulkPipelineModel> next_model;
+  next_model = std::make_shared<BulkPipelineModel>("next_model_path", 0);
+  next_model->initialize();
+  next_model->prepare();
+  next_model->setBufferOwnership(BulkPipelineModel::BufferOwnership::SHARED);
+
+  model->setNextModel(next_model);
+
+  EXPECT_NO_THROW(next_model->startAsyncBufferFill());
+  EXPECT_NO_THROW(next_model->waitForBufferReady());
+
+  // Release model for negative test
+  next_model->release();
+  // Exception will be thrown to different thread
+  next_model->startAsyncBufferFill();
+  EXPECT_ANY_THROW(next_model->waitForBufferReady());
+}