如何跨任务继续块序列化线程执行? [英] How can I serialize thread execution across a task continuation block?
问题描述
在下面的示例中，对 HandleChangesAsync 的调用是从异步任务中通过事件处理程序进行的.
In the below example, a call to HandleChangesAsync is made from within an asynchronous task, and via an event handler.
问题 - 有没有办法确保一次只有一个线程可以执行 HandleChangesAsync create_task + 任务延续块(即使任务延续块调用其他异步函数)?
Question - Is there a way to ensure that only one thread can execute the HandleChangesAsync create_task + task continuation blocks at a time (even if the task continuation blocks invoke other async functions)?
请注意，我不能只使用同步原语，因为 HandleChangesAsync 可以在异步操作完成之前返回.
Note that I can't just use a synchronization primitive because HandleChangesAsync can return before the async operations complete.
void MyClass::DoStuffAsync()
{
WeakReference weakThis(this);
create_task(DoMoreStuffAsync())
.then([weakThis]())
{
auto strongThis = weakThis.Resolve<HomePromotionTemplateViewModel>();
if (strongThis)
{
strongThis->RegisterForChanges();
strongThis->HandleChangesAsync();
}
});
}
void MyClass::RegisterForChanges()
{
// Attach event handler to &MyClass::OnSomethingChanged
}
void MyClass::OnSomethingChanged()
{
HandleChangesAsync();
}
void MyClass::HandleChangesAsync()
{
WeakReference weakThis(this);
create_task(DoMoreCoolStuffAsync(m_needsProtection))
.then([weakThis]()
{
// do async stuff
// update m_needsProtection
})
.then([weakThis]()
{
// do more async stuff
// update m_needsProtection
});
}
推荐答案
假设您只想忽略重叠的请求(而不是将它们排队)，这很容易使用原子布尔值来实现.可以将以下内容粘贴到新的 XAML 页面中，然后只需检查输出窗口即可.有两个线程每个都竞相调用 DoStuff() 但一次只会执行其中一个——value_ 的值总是恰好 0 或 16 在工作完成时；从来没有别的.如果多个线程同时执行工作，您可能会得到其他数字.
Assuming you just want to ignore overlapping requests (vs. queue them up) this is pretty easily accomplished with an atomic Boolean. The following can be pasted into a new XAML page and then just check the output window. There are two threads each racing to call DoStuff() but only one of them will ever execute at a time -- the value of value_ will always be exactly 0 or 16 at the completion of the work; never anything else. If multiple threads executed the work at the same time, you'd potentially get other numbers.
有一堆愚蠢的代码可以做工作"，但基本在 compare_exchange_strong() 调用.它基本上是说我希望 busy_ 的值为 false，在这种情况下更新 busy_ 为 true并返回 true(在这种情况下，我将开始工作).但是如果 busy_ 不是 的值已经 false 然后返回 false (我不会做任何工作)".注意 if 中的逻辑否定 ! :-)
There's a bunch of silly code to "do work" but the basic is in the compare_exchange_strong() call. It basically says "I expect the value of busy_ to be false, in which case update busy_ to be true and return true (in which case, I'll start doing work). But if the value of busy_ wasn't already false then return false (and I won't do any work)". Note the logical negation ! inside the if :-)
我不是内存排序方面的专家，所以如果您在一个紧凑的循环中运行，可能有一种更有效的方法来做到这一点(即，传递一个显式的 memory_order 值)，但它应该是正确的，应该足以满足正常的用户体验工作:
I'm not an expert in memory ordering so it's possible there's a more efficient way to do this (ie, passing an explicit memory_order value) if you were running in a tight loop, but it should be correct and should suffice for normal UX work:
#include <atomic>
#include <ppltasks.h>
#include <string>
using namespace concurrency;
class Test
{
std::atomic<bool> busy_;
int value_;
task<int> AddNumbersAsync(int x, int y)
{
return task<int>{[x, y]
{
Sleep(20);
return x + y;
}};
}
void CompleteStuff()
{
OutputDebugStringA("** Done with work; value is ");
OutputDebugStringA(std::to_string(value_).c_str());
OutputDebugStringA("
");
busy_ = false;
}
public:
Test()
: busy_{ false }, value_{ 0 }
{}
void DoStuff()
{
// ---
// This is where the magic happens...
// ---
bool expected{ false };
if (!busy_.compare_exchange_strong(expected, true))
{
OutputDebugStringA("Work already in progress; bailing.
");
return;
}
OutputDebugStringA("Doing work...
");
value_ = 2;
try
{
AddNumbersAsync(value_, value_).then([this](int i)
{
value_ = i;
return AddNumbersAsync(value_, value_);
}).then([this](int i)
{
value_ = i;
return AddNumbersAsync(value_, value_);
}).then([this](task<int> i)
{
// ---
// Handle any async exceptions
// ---
try
{
value_ = i.get();
}
catch (...)
{
OutputDebugStringA("Oops, an async exception! Resetting value_
");
value_ = 0;
}
CompleteStuff();
});
}
// ---
// Handle any sync exceptions
// ---
catch (...)
{
OutputDebugStringA("Oops, an exception! Resetting value_
");
value_ = 0;
CompleteStuff();
}
}
};
Test t;
task<void> TestTask1;
task<void> TestTask2;
MainPage::MainPage()
{
InitializeComponent();
TestTask1 = task<void>([]
{
for (int i = 0; i < 100; i++)
{
t.DoStuff();
Sleep(20);
}
});
TestTask1 = task<void>([]
{
for (int i = 0; i < 100; i++)
{
t.DoStuff();
Sleep(30);
}
});
}
请注意，我们需要捕获同步异常(那些可能由任务链的第一部分引起的异常)和异步异常(由其中一个任务或其延续引起的异常).我们通过使用 task 类型的延续来捕获异步异常，然后通过在任务上显式调用 get() 来检查异常.在这两种情况下，我们都会重置 value_ 和 busy_ 标志.
Note that we need to catch both synchronous exceptions (those potentially caused by the first part of the task chain) and asynchronous exceptions (those caused by one of the tasks or their continuations). We catch asynchronous exceptions by using a continuation of type task<int> and then check for exceptions by explicitly calling get() on the task. In both cases we reset the value_ and the busy_ flag.
您应该会看到如下输出；您可以看出多个线程正在相互竞争，因为有时 OutputDebugString 调用会交错:
You should see some output such as the following; you can tell that multiple threads are competing with each other since sometimes the OutputDebugString calls get interleaved:
Doing work...
Work already in progress; bailing.
Work already in progress; bailing.
Work already in progress; bailing.
Work already in progress; bailing.
** Done with work; value is Work already in progress; bailing.
16
这篇关于如何跨任务继续块序列化线程执行?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
