Recording thread deadlock fix

AirLib/include/common/WorkerThread.hpp

@@ -102,6 +102,34 @@ class WorkerThreadSignal
         return true;
     }
 
+    void wait()
+    {
+        // wait for signal or timeout or cancel predicate
+        std::unique_lock<std::mutex> lock(mutex_);
+        while (!signaled_) {
+            cv_.wait(lock);
+        }
+        lock.unlock();
+        signaled_ = false;
+    }
+
+    bool waitForRetry(double timeout_sec, int n_times)
+    {
+        std::unique_lock<std::mutex> lock(mutex_);
+        while (!signaled_ && n_times > 0) {
+            cv_.wait_for(lock, std::chrono::milliseconds(static_cast<long long>(timeout_sec * 1000)));
+            --n_times;
+        }
+        lock.unlock();
+        if (n_times == 0 && !signaled_) {
+            return false;
+        }
+        else {
+            signaled_ = false;
+            return true;
+        }
+    }
+
 };
 
 // This class provides a synchronized worker thread that guarantees to execute

Unreal/Plugins/AirSim/Source/Recording/RecordingThread.cpp

@@ -8,7 +8,10 @@
 #include "PIPCamera.h"
 
 
-std::unique_ptr<FRecordingThread> FRecordingThread::instance_;
+std::unique_ptr<FRecordingThread> FRecordingThread::running_instance_;
+std::unique_ptr<FRecordingThread> FRecordingThread::finishing_instance_;
+msr::airlib::WorkerThreadSignal FRecordingThread::finishing_signal_;
+bool FRecordingThread::first_ = true;
 
 
 FRecordingThread::FRecordingThread()
@@ -18,51 +21,77 @@ FRecordingThread::FRecordingThread()
 }
 
 
-void FRecordingThread::startRecording(const msr::airlib::ImageCaptureBase* image_capture, const msr::airlib::Kinematics::State* kinematics, 
+void FRecordingThread::startRecording(const msr::airlib::ImageCaptureBase* image_capture, const msr::airlib::Kinematics::State* kinematics,
     const RecordingSetting& settings, msr::airlib::VehicleSimApiBase* vehicle_sim_api)
 {
     stopRecording();
 
     //TODO: check FPlatformProcess::SupportsMultithreading()?
+    assert(!isRecording());
 
-    instance_.reset(new FRecordingThread());
-    instance_->image_capture_ = image_capture;
-    instance_->kinematics_ = kinematics;
-    instance_->settings_ = settings;
-    instance_->vehicle_sim_api_ = vehicle_sim_api;
+    running_instance_.reset(new FRecordingThread());
+    running_instance_->image_capture_ = image_capture;
+    running_instance_->kinematics_ = kinematics;
+    running_instance_->settings_ = settings;
+    running_instance_->vehicle_sim_api_ = vehicle_sim_api;
 
-    instance_->last_screenshot_on_ = 0;
-    instance_->last_pose_ = msr::airlib::Pose();
+    running_instance_->last_screenshot_on_ = 0;
+    running_instance_->last_pose_ = msr::airlib::Pose();
 
-    instance_->is_ready_ = true;
+    running_instance_->is_ready_ = true;
 
-    instance_->recording_file_.reset(new RecordingFile());
-    instance_->recording_file_->startRecording(vehicle_sim_api);
+    running_instance_->recording_file_.reset(new RecordingFile());
+    running_instance_->recording_file_->startRecording(vehicle_sim_api);
 }
 
 FRecordingThread::~FRecordingThread()
 {
-    stopRecording();
+    if (this == running_instance_.get()) stopRecording();
+}
+
+void FRecordingThread::init()
+{
+    first_ = true;
 }
 
 bool FRecordingThread::isRecording()
 {
-    return instance_ != nullptr;
+    return running_instance_ != nullptr;
 }
 
 void FRecordingThread::stopRecording()
 {
-    if (instance_)
+    if (running_instance_)
     {
-        instance_->EnsureCompletion();
-        instance_.reset();
+        assert(finishing_instance_ == nullptr);
+        finishing_instance_ = std::move(running_instance_);
+        assert(!isRecording());
+        finishing_instance_->Stop();
+    }
+}
+
+void FRecordingThread::killRecording()
+{
+    if (first_) return;
+
+    stopRecording();
+    bool finished = finishing_signal_.waitForRetry(1, 5);
+    if (!finished) {
+        UE_LOG(LogTemp, Log, TEXT("killing thread"));
+        finishing_instance_->thread_->Kill(false);
     }
 }
 
 /*********************** methods for instance **************************************/
 
 bool FRecordingThread::Init()
 {
+    if (first_) {
+        first_ = false;
+    }
+    else {
+        finishing_signal_.wait();
+    }
     if (image_capture_ && recording_file_)
     {
         UAirBlueprintLib::LogMessage(TEXT("Initiated recording thread"), TEXT(""), LogDebugLevel::Success);
@@ -85,8 +114,12 @@ uint32 FRecordingThread::Run()
 
                 //TODO: should we go as fast as possible, or should we limit this to a particular number of
                 // frames per second?
-
+
+                //BG: Workaround to get sync ground truth. See https://github.com/Microsoft/AirSim/issues/1494 for details
+                uint64_t timestamp_millis = static_cast<uint64_t>(msr::airlib::ClockFactory::get()->nowNanos() / 1.0E6);
+                std::string gt = vehicle_sim_api_->getRecordFileLine(false);
                 std::vector<msr::airlib::ImageCaptureBase::ImageResponse> responses;
+
                 image_capture_->getImages(settings_.requests, responses);
                 recording_file_->appendRecord(responses, vehicle_sim_api_);
             }
@@ -107,13 +140,8 @@ void FRecordingThread::Stop()
 
 void FRecordingThread::Exit()
 {
+    assert(this == finishing_instance_.get());
     if (recording_file_)
         recording_file_.reset();
-}
-
-void FRecordingThread::EnsureCompletion()
-{
-    Stop();
-    thread_->WaitForCompletion();
-    //UAirBlueprintLib::LogMessage(TEXT("Stopped recording thread"), TEXT(""), LogDebugLevel::Success);
+    finishing_signal_.signal();
 }

Unreal/Plugins/AirSim/Source/Recording/RecordingThread.h

@@ -10,6 +10,7 @@
 #include <memory>
 #include "common/ClockFactory.hpp"
 #include "common/AirSimSettings.hpp"
+#include "common/WorkerThread.hpp"
 
 class FRecordingThread : public FRunnable
 {
@@ -19,9 +20,12 @@ class FRecordingThread : public FRunnable
 public:
     FRecordingThread();
     virtual ~FRecordingThread();
+
+    static void init();
     static void startRecording(const msr::airlib::ImageCaptureBase* camera, const msr::airlib::Kinematics::State* kinematics, 
         const RecordingSetting& settings, msr::airlib::VehicleSimApiBase* vehicle_sim_api);
-    static void stopRecording(); 
+    static void stopRecording();
+    static void killRecording();
     static bool isRecording();
 
 protected:
@@ -30,12 +34,14 @@ class FRecordingThread : public FRunnable
     virtual void Stop() override;
     virtual void Exit() override;
 
-private:
-    void EnsureCompletion();
-
 private:
     FThreadSafeCounter stop_task_counter_;
     FRenderCommandFence read_pixel_fence_;
+
+    static std::unique_ptr<FRecordingThread> running_instance_;
+    static std::unique_ptr<FRecordingThread> finishing_instance_;
+    static msr::airlib::WorkerThreadSignal finishing_signal_;
+    static bool first_;
 
     static std::unique_ptr<FRecordingThread> instance_;
 

Unreal/Plugins/AirSim/Source/SimMode/SimModeBase.cpp

@@ -82,6 +82,7 @@ void ASimModeBase::BeginPlay()
     UAirBlueprintLib::LogMessage(TEXT("Press F1 to see help"), TEXT(""), LogDebugLevel::Informational);
 
     setupVehiclesAndCamera();
+    FRecordingThread::init();
 
     UWorld* World = GetWorld();
     if (World)
@@ -126,6 +127,7 @@ void ASimModeBase::setStencilIDs()
 void ASimModeBase::EndPlay(const EEndPlayReason::Type EndPlayReason)
 {
     FRecordingThread::stopRecording();
+    FRecordingThread::killRecording();
     world_sim_api_.reset();
     api_provider_.reset();
     api_server_.reset();