multithreading - 在Windows Store App中生成线程会在UI线程中的任何定时等待调用中导致死锁

Question

我正在移植一个跨平台代码，主要是用C++编写的，以支持Windows(Phone)8.1。到目前为止，借助C++ 11功能，这是一项非常容易的任务，但是最近我偶然发现了一个非常奇怪的错误。创建std::thread会导致后续任何对UI线程中的定时等待方法(std::condition_variable::wait_for(), std::condition_variable::wait_until(), std::this_thread::sleep_for())的调用都导致死锁:不仅此类调用永不返回，而且调用std::condition_variable::notify_all不会唤醒等待线程。

这是一个快速的代码示例(应从UI线程调用该代码):

// returns normally
std::this_thread::sleep_for(std::chrono::seconds(1));

std::thread trd([] { });
if (trd.joinable()) {
    trd.join();
}

// deadlock
std::this_thread::sleep_for(std::chrono::seconds(1));

其他观察:

此错误不影响后台线程

非定时等待效果很好(即std::condition_variable::wait())。

有任何想法吗？

Answer 1

进一步谷歌搜索之后，我在MSDN论坛上遇到了this thread。事实证明，这是MSFT所理解的“不是错误，而是功能”，即显然是WINAPI实现中的错误。
幸运的是，正如我在问题中提到的那样，后台线程不受影响，因此我决定编写并共享一段代码，该代码通过将等待操作卸载到后台线程来解决了该问题。解决方法基于存储在线程本地存储中的始终在线异步“消息泵”。您可以选择手动注册COM或STD线程，或定义一个回调以自动区分它们。

这是代码(用您喜欢的 namespace 包装):

PlatformConcurrency.hpp

#ifdef _WIN32
    #include <atomic>
    #include <Fibersapi.h>
#endif

#include <chrono>
#include <thread>
#include <mutex>
#include <condition_variable>

    class PlatformConcurrency {

#ifdef _WIN32
        #define PLATFORM_CONCURRENCY_POLICY_AUTO 0
        #define PLATFORM_CONCURRENCY_POLICY_REGISTER_COM_THREADS 1
        #define PLATFORM_CONCURRENCY_POLICY_REGISTER_STD_THREADS 2

        private:
            enum class ThreadNature {
                STD, COM
            };

            class IOperation {
                public:
                    virtual ~IOperation() { };
                    virtual void preexecute() = 0;
                    virtual void execute() = 0;
            };
            template <typename T> class SleepUntilOperation : public virtual IOperation {
                private:
                    T _timePoint;

                public:
                    SleepUntilOperation(T timePoint) : _timePoint(timePoint) { }
                    void preexecute() override { }
                    void execute() override {
                        std::this_thread::sleep_until(_timePoint);
                    }
            };
            template <typename T> class WaitUntilOperation : public virtual IOperation {
                private:
                    enum class State {
                        INITIAL,
                        PRE_WAIT, WAITING,
                        AWAKE, TIMED_AWAKE, FINISHED
                    };

                public:
                    std::cv_status result;

                private:
                    std::condition_variable* _condition;
                    std::unique_lock<std::mutex>* _lock;
                    T _timePoint;

                    std::atomic<State> _state = State::INITIAL;
                    std::mutex _ownMutex;
                    std::condition_variable _ownCondition;

                public:
                    WaitUntilOperation(std::condition_variable* condition, std::unique_lock<std::mutex>* lock, T timePoint) :
                            _condition(condition), _lock(lock), _timePoint(timePoint) {
                    }
                    void preexecute() override {
                        std::unique_lock<std::mutex> ownLock(_ownMutex);

                        while(_state == State::INITIAL) {
                            _ownCondition.wait(ownLock);
                        }

                        _state = State::WAITING;
                        _ownCondition.notify_one();
                        ownLock.unlock();
                        _condition->wait(*_lock);
                        ownLock.lock();

                        switch(_state) {
                            case State::TIMED_AWAKE:
                                _state = State::FINISHED;
                                break;
                            default:
                                _state = State::AWAKE;
                                result = std::cv_status::no_timeout;
                                _ownCondition.notify_one();
                                break;
                        }

                    }
                    void execute() override {
                        std::unique_lock<std::mutex> ownLock(_ownMutex);

                        _state = State::PRE_WAIT;
                        _ownCondition.notify_one();
                        while (_state == State::PRE_WAIT) {
                            _ownCondition.wait(ownLock);
                        }

                        std::cv_status status = _ownCondition.wait_until(ownLock, _timePoint);

                        switch (_state) {
                            case State::AWAKE:
                                break;
                            default:
                                _state = State::TIMED_AWAKE;
                                result = status;
                                ownLock.unlock();
                                do {
                                    _condition->notify_all(); // we can't notify our specific thread, so count this one as spurious wakeup :)
                                } while (_state != State::FINISHED);
                                break;
                        }

                    }
            };

            class IImplementation {
                public:
                    const ThreadNature nature;

                public:
                    IImplementation(ThreadNature nature) : nature(nature) { };
                    virtual ~IImplementation() { };
                    virtual void post(IOperation* operation) = 0;
            };
            class STDImplementation : public virtual IImplementation {
                public:
                    STDImplementation() : IImplementation(ThreadNature::STD) { };
                    void post(IOperation* operation) override {
                        operation->preexecute();
                        operation->execute();
                    }
            };
            class COMImplementation : public virtual IImplementation {
                private:
                    enum class MessageID {
                        OPERATION, EXIT
                    };
                    class Message {
                        public:
                            const MessageID id;
                            IOperation* const operation;

                        private:
                            std::atomic<bool> _handled = false;
                            std::mutex _mutex;
                            std::condition_variable _condition;

                        public:
                            Message(MessageID id) : Message(id, 0) { };
                            Message(MessageID id, IOperation* operation) : id(id), operation(operation) { };

                            void handled() {
                                {
                                    std::unique_lock<std::mutex> lock(_mutex);
                                    _handled = true;
                                }
                                _condition.notify_one();
                            }
                            void waitForHandled() {
                                std::unique_lock<std::mutex> lock(_mutex);
                                while (!_handled) {
                                    _condition.wait(lock);
                                }
                            }
                    };

                private:
                    std::atomic<Message*> _message = 0;
                    std::mutex _mutex;
                    std::condition_variable _condition;
                    std::thread _thread;

                public:
                    COMImplementation() : IImplementation(ThreadNature::COM), _thread(&COMImplementation::run, this) { }
                    ~COMImplementation() {
                        Message msg(MessageID::EXIT);
                        post(&msg);
                        if (_thread.joinable()) {
                            _thread.join();
                        }
                    }
                    void post(IOperation* operation) override {
                        Message msg(MessageID::OPERATION, operation);
                        post(&msg);
                    }

                private:
                    void post(Message* message) {
                        {
                            std::unique_lock<std::mutex> lock(_mutex);
                            _message = message;
                        }
                        _condition.notify_one();
                        if (message->id == MessageID::OPERATION) {
                            message->operation->preexecute();
                        }
                        message->waitForHandled();
                    }
                    void run() {
                        do {
                            Message* msg = pull();
                            switch(msg->id) {
                                case MessageID::OPERATION:
                                    msg->operation->execute();
                                    msg->handled();
                                    break;
                                default:
                                    msg->handled();
                                    return;
                            }
                        } while(true);
                    }
                    Message* pull() {
                        std::unique_lock<std::mutex> lock(_mutex);
                        Message* msg = _message;
                        if (msg == 0) {
                            do {
                                _condition.wait(lock);
                                msg = _message;
                            } while (msg == 0);
                        }
                        _message = 0;
                        return msg;
                    }
            };

        private:
            static DWORD IMPLEMENTATION;
#endif

        public:
#ifdef _WIN32
    #if (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_REGISTER_COM_THREADS)
            static void registerCOMThread() {
                IImplementation* implementation = reinterpret_cast<IImplementation*>(FlsGetValue(IMPLEMENTATION));
                if (implementation == 0) {
                    implementation = new COMImplementation();
                    FlsSetValue(IMPLEMENTATION, implementation);
                }
            }
    #elif (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_REGISTER_STD_THREADS)
            static void registerSTDThread() {
                IImplementation* implementation = reinterpret_cast<IImplementation*>(FlsGetValue(IMPLEMENTATION));
                if (implementation == 0) {
                    implementation = new STDImplementation();
                    FlsSetValue(IMPLEMENTATION, implementation);
                }
            }
    #endif
#endif
            template <typename T> static void sleepFor(T duration) {
                sleepUntil(std::chrono::steady_clock::now() + duration);
            }
            template <typename T> static void sleepUntil(T timePoint) {
#ifdef _WIN32
                IImplementation* implementation = getImplementation();
                if ((implementation != 0) && (implementation->nature != ThreadNature::STD)) {
                    SleepUntilOperation<T> op(timePoint);
                    implementation->post(&op);
                } else
#endif
                {
                    std::this_thread::sleep_until(timePoint);
                }
            }
            template <typename T> static std::cv_status waitFor(std::condition_variable& condition, std::unique_lock<std::mutex>& lock, T duration) {
                return waitUntil(condition, lock, std::chrono::steady_clock::now() + duration);
            }
            template <typename T> static std::cv_status waitUntil(std::condition_variable& condition, std::unique_lock<std::mutex>& lock, T timePoint) {
#ifdef _WIN32
            IImplementation* implementation = getImplementation();
            if ((implementation != 0) && (implementation->nature != ThreadNature::STD)) {
                WaitUntilOperation<T> op(&condition, &lock, timePoint);
                implementation->post(&op);
                return op.result;
            }
#endif
            return condition.wait_until(lock, timePoint);
        }

        private:
#ifdef _WIN32
            static IImplementation* getImplementation() {
                IImplementation* implementation = reinterpret_cast<IImplementation*>(FlsGetValue(IMPLEMENTATION));
#if (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_AUTO) || (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_REGISTER_STD_THREADS)
                if (implementation == 0) {
#if (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_AUTO)
                    if (getThreadNature() == ThreadNature::COM) {
                        implementation = new COMImplementation();
                    } else {
                        implementation = new STDImplementation();
                    }
#else
                    implementation = new COMImplementation();
#endif
                    FlsSetValue(IMPLEMENTATION, implementation);
                }
#endif
                return implementation;
            }
#if (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_AUTO)
            static ThreadNature getThreadNature();
#endif
            static void NTAPI destroyImplementation(void* context) {
                delete reinterpret_cast<IImplementation*>(context);
            }
#endif
    };

PlatformConcurrency.cpp

#include "PlatformConcurrency.hpp"

#ifdef _WIN32
    DWORD PlatformConcurrency::IMPLEMENTATION = FlsAlloc(&PlatformConcurrency::destroyImplementation);

#if (PLATFORM_CONCURRENCY_POLICY == PLATFORM_CONCURRENCY_POLICY_AUTO)
    // define PlatformConcurrency::ThreadNature PlatformConcurrency::getThreadNature() here
#endif
#endif

如果您有任何疑问和/或更正-请不要犹豫发布它们!