[mlir:async] Fix deadlock in async runtime await-and-execute functions

[ezhulenev]

Summary

emplace??? functions running concurrently can set the ready flag and then pending awaiter will never be executed

mlir/lib/ExecutionEngine/AsyncRuntime.cpp

extern "C" void mlirAsyncRuntimeExecute(CoroHandle handle, CoroResume resume) {
   auto *runtime = getDefaultAsyncRuntime();
   runtime->getThreadPool().async([handle, resume]() { (*resume)(handle); });
 }
 
 extern "C" void mlirAsyncRuntimeAwaitTokenAndExecute(AsyncToken *token,
                                                      CoroHandle handle,
                                                      CoroResume resume) {
   auto execute = [handle, resume]() { (*resume)(handle); };
+  std::unique_lock<std::mutex> lock(token->mu);
   if (token->ready) {
+    lock.unlock();
     execute();
   } else {
-    std::unique_lock<std::mutex> lock(token->mu);
     token->awaiters.push_back([execute]() { execute(); });
   }
 }
 
 extern "C" void mlirAsyncRuntimeAwaitValueAndExecute(AsyncValue *value,
                                                      CoroHandle handle,
                                                      CoroResume resume) {
   auto execute = [handle, resume]() { (*resume)(handle); };
+  std::unique_lock<std::mutex> lock(value->mu);
   if (value->ready) {
+    lock.unlock();
     execute();
   } else {
-    std::unique_lock<std::mutex> lock(value->mu);
     value->awaiters.push_back([execute]() { execute(); });
   }
 }
 
 extern "C" void mlirAsyncRuntimeAwaitAllInGroupAndExecute(AsyncGroup *group,
                                                           CoroHandle handle,
                                                           CoroResume resume) {
   auto execute = [handle, resume]() { (*resume)(handle); };
+  std::unique_lock<std::mutex> lock(group->mu);
   if (group->pendingTokens == 0) {
+    lock.unlock();
     execute();
   } else {
-    std::unique_lock<std::mutex> lock(group->mu);
     group->awaiters.push_back([execute]() { execute(); });
   }
 }
 
 //===----------------------------------------------------------------------===//
 // Small async runtime support library for testing.
 //===----------------------------------------------------------------------===//