AsynchronousProgramming

  class background_task
  {
  public:
      void operator()() const
      {
          do_something();
          do_something_else();
      }
  };
  
  // thread 接受任意可调用对象，函数、重载了调用操作符的类、lambda等
  background_task f;
  // f会被copy到属于新线程的存储空间，并被调用
  std::thread my_thread(f);
  
  // Tips：
  // 下面的代码会被解析为声明一个名称为my_thread的函数
  // 函数参数为函数指针
  // 参数函数的参数为空
  // 参数函数的返回值类型为background_task
  std::thread my_thread(background_task());
  
  // 正确的方式如下
  std::thread my_thread((background_task()));
  std::thread my_thread{background_task()};
  
  // 使用 lambda 作为 thread 的执行函数
  std::thread my_thread([]{
      do_something();
      do_something_else();
  });

  class X
  {
  public:
      void do_lengthy_work();
  };
  
  X my_x;
  std::thread t(&X::do_lengthy_work, &my_x);

  void process_big_object(std::unique_ptr<big_object>);
  std::unique_ptr<big_object> p(new big_object);
  p->prepare_data(42);
  std::thread t(process_big_object, std::move(p));

class scoped_thread
{
    std::thread t;
public:
    explicit scoped_thread(std::thread t_): t(std::move(t_))
    {
        if(!t.joinable())
            throw std::logic_error(“No thread”);
    }
    ~scoped_thread()
    {
        t.join();
    }
    scoped_thread(scoped_thread const&)=delete;
    scoped_thread& operator=(scoped_thread const&)=delete;
};
struct func; // 实现了调用操作符的struct
void f()
{
    int some_local_state;
    scoped_thread t{std::thread(func(some_local_state))};
    do_something_in_current_thread();
}

class joining_thread
{
    std::thread t;
public:
    joining_thread() noexcept=default;
    template<typename Callable,typename ... Args>
    explicit joining_thread(Callable&& func,Args&& ... args): t(std::forward<Callable>(func),std::forward<Args>(args)...)
    {}
    explicit joining_thread(std::thread t_) noexcept:
        t(std::move(t_))
    {}
    joining_thread(joining_thread&& other) noexcept:
        t(std::move(other.t))
    {}
    joining_thread& operator=(joining_thread&& other) noexcept
    {
        if(joinable())
            join();
        t=std::move(other.t);
        return *this;
    }
    joining_thread& operator=(std::thread other) noexcept
    {
        if(joinable())
            join();
        t=std::move(other);
        return *this;
    }
    ~joining_thread() noexcept
    {
        if(joinable())
            join();
    }
    void swap(joining_thread& other) noexcept
    {
        t.swap(other.t);
    }
    std::thread::id get_id() const noexcept
    {
        return t.get_id();
    }
    bool joinable() const noexcept
    {
        return t.joinable();
    }
    void join() { t.join(); }
    void detach()
    {
        t.detach();
    }
    std::thread& as_thread() noexcept
    {
        return t;
    }
    const std::thread& as_thread() const noexcept
    {
        return t;
    }
};

void do_work(unsigned id);
void f()
{
    std::vector<std::thread> threads;
    for(unsigned i=0;i<20;++i)
    {
        threads.emplace_back(do_work,i);
    }
    for(auto& entry: threads)
        entry.join();
}

class some_data
{
    int a;
    std::string b;
public:
    void do_something();
};
class data_wrapper
{
private:
    some_data data;
    std::mutex m;
public:
    template<typename Function>
    void process_data(Function func)
    {
        std::lock_guard<std::mutex> l(m);
        // 将保护数据的引用传递给不受控制的func函数，func函数将保护数据的指针保存下来，导致了共享数据指针的逃逸
        func(data);
    }
};
some_data* unprotected;
void malicious_function(some_data& protected_data)
{
    unprotected=&protected_data;
}
data_wrapper x;
void foo()
{
    x.process_data(malicious_function);
    unprotected->do_something();
}

template<typename T,typename Container=std::deque<T> >
class stack
{
public:
    explicit stack(const Container&);
    explicit stack(Container&& = Container());
    template <class Alloc> explicit stack(const Alloc&);
    template <class Alloc> stack(const Container&, const Alloc&);
    template <class Alloc> stack(Container&&, const Alloc&);
    template <class Alloc> stack(stack&&, const Alloc&);
    bool empty() const;
    size_t size() const;
    T& top();
    T const& top() const;
    void push(T const&);
    void push(T&&);
    void pop();
    void swap(stack&&);
    template <class... Args> void emplace(Args&&... args);
};

stack<int> s;
if(!s.empty())                // mark 1
{
    int const value=s.top();  // mark 2
    s.pop();                  // mark 3
    do_something(value);
}

class some_big_object;
void swap(some_big_object& lhs,some_big_object& rhs);
class X
{
private:
    some_big_object some_detail;
    std::mutex m;
public:
    X(some_big_object const& sd):some_detail(sd){}
    friend void swap(X& lhs, X& rhs)
    {
        // 如果两个参数是同一个对象，则返回。避免对同一个互斥量上锁多次
        // 对同一个互斥量上锁多次属于未定义行为
        if(&lhs==&rhs) 
            return;
        std::lock(lhs.m,rhs.m);
        std::lock_guard<std::mutex> lock_a(lhs.m,std::adopt_lock);
        std::lock_guard<std::mutex> lock_b(rhs.m,std::adopt_lock);
        swap(lhs.some_detail,rhs.some_detail);
    }
};

// C++17 引入了 std::scoped_lock<> 可以对上面swap函数进行简化
void swap(X& lhs, X& rhs)
{
    if(&lhs==&rhs)  
        return;
    // std::scoped_lock 是一个模板类，此处利用了C++17中引入的类模板参数类型推断
    std::scoped_lock guard(lhs.m,rhs.m);
    swap(lhs.some_detail,rhs.some_detail);
}

class some_big_object;
void swap(some_big_object& lhs,some_big_object& rhs);
class X
{
private:
    some_big_object some_detail;
    std::mutex m;
public:
    X(some_big_object const& sd):some_detail(sd){}
    friend void swap(X& lhs, X& rhs)
    {
        if(&lhs==&rhs)
            return;
        // std::defer_lock 表示，暂时不对mutex上锁
        std::unique_lock<std::mutex> lock_a(lhs.m,std::defer_lock);
        std::unique_lock<std::mutex> lock_b(rhs.m,std::defer_lock);
        // 此处使用std::lock对 std::unique_lock 上锁，会调用std::unique_lock的lock成员函数，进而调用mutex的lock成员函数
        std::lock(lock_a,lock_b);
        swap(lhs.some_detail,rhs.some_detail);
    }
};

std::unique_lock<std::mutex> get_lock()
{
    // TODO Why extern?
    extern std::mutex some_mutex;
    std::unique_lock<std::mutex> lk(some_mutex);
    prepare_data();
    // TODO mutex 无法copy，那么lk对象如何传递其关联的mutex？
    return lk;
}
void process_data()
{
    std::unique_lock<std::mutex> lk(get_lock());
    do_something();
}

void get_and_process_data()
{
    std::unique_lock<std::mutex> my_lock(the_mutex);
    some_class data_to_process=get_next_data_chunk();
    my_lock.unlock(); // 获得数据后，释放锁
    result_type result=process(data_to_process);
    my_lock.lock();   // 写数据时，再上锁
    write_result(data_to_process,result);
}

class Y
{
private:
    int some_detail;
    mutable std::mutex m;
    int get_detail() const
    {
        std::lock_guard<std::mutex> lock_a(m);
        return some_detail;
    }
public:
    Y(int sd):some_detail(sd){}
    friend bool operator==(Y const& lhs, Y const& rhs)
    {
        if(&lhs==&rhs)
            return true;
        // 此处分开对两个参数进行上锁，减低了锁的粒度，提高了性能。
        // 但是，也修改了operator==的语意，其不再表示某个时间点两个对象相等，而是表示某两个时间点两个对象相等
        int const lhs_value=lhs.get_detail();
        int const rhs_value=rhs.get_detail();
        return lhs_value==rhs_value;
    }
};

std::shared_ptr<some_resource> resource_ptr;
void foo()
{
    if(!resource_ptr)
    {
        resource_ptr.reset(new some_resource);
    }
    resource_ptr->do_something();
}

#include <map>
#include <string>
#include <mutex>
#include <shared_mutex>
class dns_entry;
class dns_cache
{
    std::map<std::string,dns_entry> entries;
    // reader-writer 互斥量
    mutable std::shared_mutex entry_mutex;
public:
    dns_entry find_entry(std::string const& domain) const
    {
        // 获取共享锁，多个线程可以并行执行find_entry，但会阻塞调用update_or_add_entry的线程
        std::shared_lock<std::shared_mutex> lk(entry_mutex);
        std::map<std::string,dns_entry>::const_iterator const it= entries.find(domain);
        return (it==entries.end())?dns_entry():it->second;
    }
    void update_or_add_entry(std::string const& domain, dns_entry const& dns_details)
    {
        // 获取互斥锁，只有一个线程可以执行 update_or_add_entry。会阻塞其他调用 update_or_add_entry 或 find_entry 的线程
        std::lock_guard<std::shared_mutex> lk(entry_mutex);
        entries[domain]=dns_details;
    }
};

std::mutex mut;
std::queue<data_chunk> data_queue;
std::condition_variable data_cond;
void data_preparation_thread()
{
    while(more_data_to_prepare())
    {
        data_chunk const data=prepare_data();
        {
            // 对互斥量进行上锁
            std::lock_guard<std::mutex> lk(mut);
            data_queue.push(data);
        }
        // 调用condition_variable的notify_one函数通知等待线程
        // 等待解锁后再调用notify_one，避免等待线程立即wake后，由于互斥量未被解锁而又被阻塞。
        // 解锁一个等待该条件变量的线程。如果不存在等待线程，则什么也不发生；如果存在多个等待线程，则无法确定唤醒哪个等待线程
        data_cond.notify_one();
    }
}
void data_processing_thread()
{
    while(true)
    {
        // 对互斥量进行上锁
        // Tips: 此处使用的是 std::unique_lock<std::mutex>，不能使用 std::lock_guard<std::mutex>(该函数没有提供unlock函数)
        std::unique_lock<std::mutex> lk(mut);
        
        // 第一个参数：std::unique_lock<std::mutex>
        // 第二个参数：lambda 函数用于表示等待的条件
        // wait 函数的实现：先调用lambda函数检查条件是否满足，如果满足，则返回；如果不满足，则调用 lk.unlock 释放锁，并且将线程状态转换为阻塞或等待状态。当 data_preparation_thread 线程调用条件变量 data_cond 的 notify_one 函数时，data_processing_thread 从睡眠状态醒来，调用 lk.lock 上锁，然后再次调用 lambda 函数检查条件是否满足。
        data_cond.wait(lk,[]{return !data_queue.empty();});
        data_chunk data=data_queue.front();
        data_queue.pop();
        
        // 当前面 wait 没有阻塞过时，此处 unlock 对应与开始的 lock
        // 当前面 wait 阻塞过时，此处的 unlock 对于与 wait 中的 lock
        lk.unlock();
        process(data);
        if(is_last_chunk(data))
            break;
    }
}

template<typename Predicate>
void minimal_wait(std::unique_lock<std::mutex>& lk,Predicate pred)
{
    while(!pred())
    {
        lk.unlock();
        lk.lock();
    }
}

#include <queue>
#include <memory> // 为了使用std::shared_ptr
#include <mutex>
#include <condition_variable>

template<typename T>
class threadsafe_queue
{
private:
    mutable std::mutex mut;  // bool empty() const成员函数中使用了该变量，因此需要使用mutable修饰
    std::queue<T> data_queue;
    std::condition_variable data_cond;
public:
    threadsafe_queue() {}
    threadsafe_queue(threadsafe_queue const& other)
    {
        std::lock_guard<std::mutex> lk(other.mut);
        data_queue=other.data_queue;
    }
    threadsafe_queue& operator=(const threadsafe_queue&) = delete;  // 为了简单起见，不允许赋值操作符

    void push(T new_value)
    {
        {
            std::lock_guard<std::mutex> lk(mut);
            data_queue.push(new_value);
        }
        // 通知等待线程 队列中有数据了
        // notify_one 无法保证会通知哪个等待线程
        data_cond.notify_one();
    }

    void wait_and_pop(T& value)
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        value=data_queue.front();
        data_queue.pop();
    }

    std::shared_ptr<T> wait_and_pop()
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return res;
    }

    bool try_pop(T& value)
    {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
            return false;
        value=data_queue.front();
        data_queue.pop();
        return true;
    }

    std::shared_ptr<T> try_pop()
    {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
            return std::shared_ptr<T>();
        std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return res;
    }

    bool empty() const
    {
        // 
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};

threadsafe_queue<data_chunk> data_queue;
void data_preparation_thread()
{
    while(more_data_to_prepare())
    {
        data_chunk const data=prepare_data();
        data_queue.push(data);
    }
}

void data_processing_thread()
{
    while(true)
    {
        data_chunk data;
        data_queue.wait_and_pop(data);
        process(data);
        if(is_last_chunk(data))
            break;
    }
}

#include <future>
#include <iostream>

int find_the_answer_to_ltuae();
void do_other_stuff();
int main()
{
    std::future<int> the_answer=std::async(find_the_answer_to_ltuae);
    do_other_stuff();
    std::cout<<"The answer is "<<the_answer.get()<<std::endl;
}

#include <string>
#include <future>
struct X
{
    void foo(int,std::string const&);
    std::string bar(std::string const&);
};
X x;
auto f1=std::async(&X::foo,&x,42,"hello");  // 调用p->foo(42, "hello")，p是指向x的指针
auto f2=std::async(&X::bar,x,"goodbye");    // 调用tmpx.bar("goodbye")， tmpx是x的拷贝副本
struct Y
{
    double operator()(double);
};
Y y;
auto f3=std::async(Y(),3.141);          // 调用tmpy(3.141)，tmpy通过Y的移动构造函数得到
auto f4=std::async(std::ref(y),2.718);  // 调用y(2.718)
X baz(X&);
std::async(baz,std::ref(x));  // 调用baz(x)
class move_only
{
public:
    move_only();
    move_only(move_only&&)
    move_only(move_only const&) = delete;
    move_only& operator=(move_only&&);
    move_only& operator=(move_only const&) = delete;
  
    void operator()();
};
auto f5=std::async(move_only());  // 调用tmp()，tmp是通过std::move(move_only())构造得到

auto f6=std::async(std::launch::async, Y(), 1.2);             // 在新线程上执行
auto f7=std::async(std::launch::deferred, baz, std::ref(x));  // 在wait()或get()调用时执行
auto f8=std::async(std::launch::deferred | std::launch::async, baz, std::ref(x));  // 异步、延迟调用
auto f9=std::async(baz, std::ref(x));
f7.wait();  //  调用延迟函数

template<>
class packaged_task<std::string(std::vector<char>*,int)>
{
public:
    template<typename Callable>
    explicit packaged_task(Callable&& f);
    std::future<std::string> get_future();
    void operator()(std::vector<char>*,int);
};

#include <future>
void process_connections(connection_set& connections)
{
    while(!done(connections))  // 1
    {
        for(connection_iterator connection=connections.begin(),end=connections.end(); // 2 
            connection!=end;
            ++connection)
        {
            if(connection->has_incoming_data())  // 判断是否收到数据
            {
                data_packet data=connection->incoming();
                // 此处假设收到的数据包含一个id和一个payload数据 每个id都对应一个promise对象
                std::promise<payload_type>& p = connection->get_promise(data.id);
                p.set_value(data.payload); // 将payload数据设置为 promise 的值
            }
            if(connection->has_outgoing_data())  // 判断是否发送数据
            {
                outgoing_packet data = connection->top_of_outgoing_queue();
                connection->send(data.payload);
                // 发送数据完成后，设置 promise 的值为ture
                data.promise.set_value(true);
            }
        }
    }
}

#include <vector>
#include <thread>
#include <future>
#include <numeric>
#include <iostream>
#include <chrono>

void accumulate(std::vector<int>::iterator first,
                std::vector<int>::iterator last,
                std::promise<int> accumulate_promise)
{
    int sum = std::accumulate(first, last, 0);
    accumulate_promise.set_value(sum);  // Notify future
}

void do_work(std::promise<void> barrier)
{
    std::this_thread::sleep_for(std::chrono::seconds(1));
    barrier.set_value();
}

int main()
{
    // Demonstrate using promise<int> to transmit a result between threads.
    std::vector<int> numbers = { 1, 2, 3, 4, 5, 6 };
    std::promise<int> accumulate_promise;
    std::future<int> accumulate_future = accumulate_promise.get_future();
    std::thread work_thread(accumulate, numbers.begin(), numbers.end(),
                            std::move(accumulate_promise));

    // future::get() will wait until the future has a valid result and retrieves it.
    // Calling wait() before get() is not needed
    //accumulate_future.wait();  // wait for result
    std::cout << "result=" << accumulate_future.get() << '\n';
    work_thread.join();  // wait for thread completion

    // Demonstrate using promise<void> to signal state between threads.
    std::promise<void> barrier;
    std::future<void> barrier_future = barrier.get_future();
    std::thread new_work_thread(do_work, std::move(barrier));
    barrier_future.wait();
    new_work_thread.join();
}

extern std::promise<double> some_promise;
try
{
    some_promise.set_value(calculate_value());
}
catch(...)
{
    // 此处使用了std::current_exception() 来获得当前抛出的异常
    some_promise.set_exception(std::current_exception());
}

extern std::promise<double> some_promise;

bool result = process_data();
if (result)
{
    some_promise.set_value(true);
}
else
{
    some_promise.set_exception(std::make_exception_ptr(std::logic_error("process data error")));
}

std::promise<int> p;
std::future<int> f(p.get_future());
assert(f.valid());   // f 是合法的
std::shared_future<int> sf(std::move(f));
assert(!f.valid());  // f 现在是不合法的
assert(sf.valid());  // sf 现在是合法的

std::promise<std::string> p;
std::shared_future<std::string> sf(p.get_future());  // 隐式转移所有权 使用std::future右值对象构造一个std::shared_future 对象

std::promise< std::map< SomeIndexType, SomeDataType, SomeComparator, SomeAllocator>::iterator> p;
// 推断 sf 的类型为 std::shared_future< std::map<SomeIndexType, SomeDataType, SomeComparator, SomeAllocator>::iterator >
// 如果 Comparator 或Allocator改变了，只需要修改promise的类型，std::shared_future的类型会自动更新
auto sf=p.get_future().share();

struct system_clock // wraps GetSystemTimePreciseAsFileTime/GetSystemTimeAsFileTime
{ 
    using rep                       = long long;
    using period                    = ratio<1, 10'000'000>; // 100 nanoseconds
    using duration                  = _CHRONO duration<rep, period>;
    using time_point                = _CHRONO time_point<system_clock>;
    static constexpr bool is_steady = false;

    _NODISCARD static time_point now() noexcept { // get current time
        return time_point(duration(_Xtime_get_ticks()));
    }

    _NODISCARD static __time64_t to_time_t(const time_point& _Time) noexcept { // convert to __time64_t
        return duration_cast<seconds>(_Time.time_since_epoch()).count();
    }

    _NODISCARD static time_point from_time_t(__time64_t _Tm) noexcept { // convert from __time64_t
        return time_point{seconds{_Tm}};
    }
};

cout << chrono::system_clock::period::num << "/" << chrono::system_clock::period::den << endl;

// system_clock::to_time_t
#include <iostream>
#include <ctime>
#include <ratio>
#include <chrono>

int main ()
{
    using namespace std::chrono;

    duration<int,std::ratio<60*60*24> > one_day (1);

    system_clock::time_point today = system_clock::now();
    system_clock::time_point tomorrow = today + one_day;

    time_t tt;

    tt = system_clock::to_time_t ( today );
    std::cout << "today is: " << ctime(&tt);

    tt = system_clock::to_time_t ( tomorrow );
    std::cout << "tomorrow will be: " << ctime(&tt);

    return 0;
}

//////////
// OUTPUT:
// today is: Wed May 30 12:38:06 2012
// tomorrow will be: Thu May 31 12:38:06 2012

template <class _Rep, class _Period>
class duration { // represents a time duration
public:
    using rep    = _Rep;
    using period = typename _Period::type;

    static_assert(!_Is_duration_v<_Rep>, "duration can't have duration as first template argument");
    static_assert(_Is_ratio_v<_Period>, "period not an instance of std::ratio");
    static_assert(0 < _Period::num, "period negative or zero");
    
    // ......
    
private:
    _Rep _MyRep; // the stored rep
};

using namespace std::chrono_literals;
auto one_day=24h;                     // 相当于 std::chrono::hours(24)
auto half_an_hour=30min;              // 相当于 std::chrono::minutes(30)
auto max_time_between_messages=30ms;  // 相当于 std::chrono::milliseconds(30)
auto float_literal_dur = 2.5min;      // 相当于 std::chrono::duration<float, std::ratio<60,1>>

std::chrono::milliseconds ms(54802);
std::chrono::seconds s = std::chrono::duration_cast<std::chrono::seconds>(ms);  // 使用std::chrono::duration_cast<> 进行显示转化

std::future<int> f=std::async(some_task);
if(f.wait_for(std::chrono::milliseconds(35))==std::future_status::ready)
{
    do_something_with(f.get());
}

template<class Clock, class Duration = typename Clock::duration> class time_point;

// some_time_point 保存了 std::chrono::system_clock 相关的时间点，并且以分为单位来衡量而不是 std::chrono::system_clock 指定的时钟滴答周期。
std::chrono::time_point<std::chrono::system_clock, std::chrono::minutes> some_time_point;

// 创建 time_point
const auto p0 = std::chrono::time_point<std::chrono::system_clock>{};
std::time_t epoch_time = std::chrono::system_clock::to_time_t(p0);
std::cout << "epoch: " << std::ctime(&epoch_time);  // OUTPUT--> epoch: Thu Jan  1 00:00:00 1970

// time_point to unix-timestamp
auto microsecondsUTC = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch()).count();

// time_point + duration
std::chrono::high_resolution_clock::now() + std::chrono::nanoseconds(500)

// 打印代码执行时长
// time_point - time_point --> duration 
auto start=std::chrono::high_resolution_clock::now();
do_something();
auto stop=std::chrono::high_resolution_clock::now();
std::cout<<"do_something() took " << std::chrono::duration<double>(stop-start).count() << " seconds"<<std::endl;

#include <condition_variable>
#include <mutex>
#include <chrono>
std::condition_variable cv;
bool done;
std::mutex m;
bool wait_loop()
{
    auto const timeout= std::chrono::steady_clock::now()+std::chrono::milliseconds(500);
    std::unique_lock<std::mutex> lk(m);
    // 当使用条件变量时，没有传入predicate，就需要使用循环来处理伪唤醒的情况
    while(!done)
    {
        // Tips: 在循环中使用 wait_for，可能在伪唤醒之前就等待超时，下次再进入等待后，又重新开始等待一段时间。这样的情况可能重复任意多次，使得总的等待时长是无界限的。而wait_util则不会出现这种情况
        if(cv.wait_until(lk,timeout)==std::cv_status::timeout)
            break;
    }
    return done;
}

template<typename T>
// 函数式编程风格的接口，函数不会修改任何外部状态，函数的执行结果通过返回值返回
std::list<T> sequential_quick_sort(std::list<T> input)
{
    if(input.empty())
    {
        return input;
    }
    
    // 因为使用了函数式风格的接口，此处需要创建新的list 存放结果
    std::list<T> result;
    // splice(pos, from_list, from_list_pos)
    // 将input中开始的元素移动到result的开始位置
    result.splice(result.begin(),input,input.begin());
    T const& pivot=*result.begin();

    auto divide_point=std::partition(input.begin(),input.end(), [&](T const& t){return t<pivot;});

    // 因为使用了函数式风格的接口，此处需要创建新的list
    std::list<T> lower_part;
    lower_part.splice(lower_part.end(),input,input.begin(), divide_point);
    auto new_lower(sequential_quick_sort(std::move(lower_part)));
    auto new_higher(sequential_quick_sort(std::move(input)));

    result.splice(result.end(),new_higher);
    result.splice(result.begin(),new_lower);
    return result;
}

template<typename T>
std::list<T> parallel_quick_sort(std::list<T> input)
{
    if(input.empty())
    {
        return input;
    }
    std::list<T> result;
    result.splice(result.begin(),input,input.begin());
    T const& pivot=*result.begin();

    auto divide_point=std::partition(input.begin(),input.end(), [&](T const& t){return t<pivot;});
    std::list<T> lower_part;
    lower_part.splice(lower_part.end(),input,input.begin(), divide_point);

    // 使用future保存异步计算的结果
    std::future<std::list<T>> new_lower(std::async(&parallel_quick_sort<T>,std::move(lower_part)));

    auto new_higher(parallel_quick_sort(std::move(input)));

    result.splice(result.end(),new_higher);
    result.splice(result.begin(),new_lower.get());
    return result;
}

class atm
{
    messaging::receiver incoming;
    messaging::sender bank;
    messaging::sender interface_hardware;

    void (atm::*state)();

    std::string account;
    unsigned withdrawal_amount;
    std::string pin;
    
    // 处理取款
    void process_withdrawal()
    {
        incoming.wait()
            .handle<withdraw_ok>(
                [&](withdraw_ok const& msg)
                {
                    interface_hardware.send(
                        issue_money(withdrawal_amount));

                    bank.send(
                        withdrawal_processed(account,withdrawal_amount));

                    state=&atm::done_processing;
                })
            .handle<withdraw_denied>(
                [&](withdraw_denied const& msg)
                {
                    interface_hardware.send(display_insufficient_funds());

                    state=&atm::done_processing;
                })
            .handle<cancel_pressed>(
                [&](cancel_pressed const& msg)
                {
                    bank.send(
                        cancel_withdrawal(account,withdrawal_amount));

                    interface_hardware.send(
                        display_withdrawal_cancelled());

                    state=&atm::done_processing;
                });
    }

    // 处理余额查询
    void process_balance()
    {
        incoming.wait()
            .handle<balance>(
                [&](balance const& msg)
                {
                    interface_hardware.send(display_balance(msg.amount));

                    state=&atm::wait_for_action;
                })
            .handle<cancel_pressed>(
                [&](cancel_pressed const& msg)
                {
                    state=&atm::done_processing;
                });
    }

    // 等待用户取款|查看余额|退出
    void wait_for_action()
    {
        interface_hardware.send(display_withdrawal_options());

        incoming.wait()
            .handle<withdraw_pressed>(
                [&](withdraw_pressed const& msg)
                {
                    withdrawal_amount=msg.amount;
                    bank.send(withdraw(account,msg.amount,incoming));
                    state=&atm::process_withdrawal;
                })
            .handle<balance_pressed>(
                [&](balance_pressed const& msg)
                {
                    bank.send(get_balance(account,incoming));
                    state=&atm::process_balance;
                })
            .handle<cancel_pressed>(
                [&](cancel_pressed const& msg)
                {
                    state=&atm::done_processing;
                });
    }

    // 验证密码
    void verifying_pin()
    {
        incoming.wait()
            .handle<pin_verified>(
                [&](pin_verified const& msg)
                {
                    state=&atm::wait_for_action;
                })
            .handle<pin_incorrect>(
                [&](pin_incorrect const& msg)
                {
                    interface_hardware.send(
                        display_pin_incorrect_message());
                    state=&atm::done_processing;
                })
            .handle<cancel_pressed>(
                [&](cancel_pressed const& msg)
                {
                    state=&atm::done_processing;
                });
    }

    // 等待输入密码
    void getting_pin()
    {
        incoming.wait()
            .handle<digit_pressed>(
                [&](digit_pressed const& msg)
                {
                    unsigned const pin_length=4;
                    pin+=msg.digit;

                    if(pin.length()==pin_length)
                    {
                        bank.send(verify_pin(account,pin,incoming));
                        state=&atm::verifying_pin;
                    }
                })
            .handle<clear_last_pressed>(
                [&](clear_last_pressed const& msg)
                {
                    if(!pin.empty())
                    {
                        pin.pop_back();
                    }
                })
            .handle<cancel_pressed>(
                [&](cancel_pressed const& msg)
                {
                    state=&atm::done_processing;
                });
    }

    // 等待插入磁卡
    void waiting_for_card()
    {
        interface_hardware.send(display_enter_card());

        incoming.wait()
            .handle<card_inserted>(
                [&](card_inserted const& msg)
                {
                    account=msg.account;
                    pin="";
                    interface_hardware.send(display_enter_pin());
                    state=&atm::getting_pin;
                });
    }

    void done_processing()
    {
        interface_hardware.send(eject_card());
        state=&atm::waiting_for_card;
    }

    atm(atm const&)=delete;
    atm& operator=(atm const&)=delete;
public:
    atm(messaging::sender bank_,
        messaging::sender interface_hardware_):
        bank(bank_),interface_hardware(interface_hardware_)
    {}

    void done()
    {
        get_sender().send(messaging::close_queue());
    }

    void run()
    {
        state=&atm::waiting_for_card;
        try
        {
            for(;;)
            {
                (this->*state)();
            }
        }
        catch(messaging::close_queue const&)
        {
        }
    }

    messaging::sender get_sender()
    {
        return incoming;
    }
};

std::experimental::future<int> find_the_answer;
auto fut=find_the_answer();
// 当fut状态变为ready后，find_the_question函数会被调度，从而在某个未指明的线程上执行。
// 此处给了实现者自由来决定函数是在一个线程池上执行还是在库管理的线性上执行。这样做是有意而为之，当continuations被加到c++标准库后，实现者就可以根据他们的经验来更好地指定线程的选择并为用户提供合适的机制来控制线程的选择
auto fut2=fut.then(find_the_question);
assert(!fut.valid());
assert(fut2.valid());

std::string find_the_question(std::experimental::future<int> the_answer);

template<typename Func>
std::experimental::future<decltype(std::declval<Func>()())>
spawn_async(Func&& func){
    std::experimental::promise<decltype(std::declval<Func>()())> p;
    auto res=p.get_future();
    std::thread t(
        [p=std::move(p),f=std::decay_t<Func>(func)]() mutable
        {
            try
            {
                p.set_value_at_thread_exit(f());
            }
            catch(...)
            {
                p.set_exception_at_thread_exit(std::current_exception());
            }
        });
    t.detach();
    return res;
}
// TIPS: 使用p.set_value_at_thread_exit 和 p.set_exception_at_thread_exit 保证 future 状态变为 ready 时，线程的局部变量已经被清除。

std::future<FinalResult> process_data(std::vector<MyData>& vec)
{
    size_t const chunk_size = whatever;
    std::vector<std::future<ChunkResult>> results;
    for (auto begin=vec.begin(), end=vec.end(); beg!=end;)
    {
        size_t const remaining_size = end - begin;
        size_t const this_chunk_size = std::min(remaining_size, chunk_size);
        results.push_back(std::async(process_chunk, begin, begin+this_chunk_size));
        begin += this_chunk_size;
    }
    return std::async([all_results=std::move(results)](){
        std::vector<ChunkResult> v;
        v.reserve(all_results.size());
        for (auto& f : all_results)
        {
            v.push_back(f.get());  // 此处会不断地阻塞唤醒
        }
        return gather_results(v);
    });
}

std::experimental::future<FinalResult> process_data(
    std::vector<MyData>& vec)
{
    size_t const chunk_size = whatever;
    std::vector<std::experimental::future<ChunkResult>> results;
    for (auto begin = vec.begin(), end = vec.end(); beg != end)
    {
        size_t const remaining_size = end - begin;
        size_t const this_chunk_size = std::min(remaining_size, chunk_size);
        results.push_back(spawn_async(process_chunk, begin, begin+this_chunk_size));
        begin += this_chunk_size;
    }
    return std::experimental::when_all(
        results.begin(), results.end()).then(
            [](std::future<std::vector<std::experimental::future<ChunkResult>>> ready_results)
            {
                std::vector<std::experimental::future<ChunkResult>> all_results = ready_results.get();
                std::vector<ChunkResult> v;
                v.reserve(all_results.size());
                for (auto& f: all_results){
                    v.push_back(f.get()); // 此处不会阻塞
                }
                return gather_results(v);
            });
}

std::experimental::future<FinalResult>
find_and_process_value(std::vector<MyData> &data)
{
    unsigned const concurrency = std::thread::hardware_concurrency();
    unsigned const num_tasks = (concurrency > 0)? concurrency : 2;
    std::vector<std::experimental::future<MyData *>> results;
    auto const chunk_size = (data.size() + num_tasks - 1) / num_tasks;
    auto chunk_begin = data.begin();
    std::shared_ptr<std::atomic<bool>> done_flag = std::make_shared<std::atomic<bool>>(false);  // 标记找到满足特定条件的值
    for (unsigned i = 0; i < num_tasks; ++i) // 生成 num_tasks 个线程
    {
        auto chunk_end = (i < (num_tasks - 1)? chunk_begin + chunk_size : data.end());
        results.push_back(spawn_async([=]{   // lambda 使用copy capture，这样传入的变量就不会有 lifetime 问题了
            for (auto entry = chunk_begin; !*done_flag && (entry != chunk_end); ++entry)
            {
                if (matches_find_criteria(*entry))
                {
                    *done_flag = true;
                    return &*entry;
                }
            }
            return (MyData *)nullptr;
        }));
        chunk_begin = chunk_end;
    }
    std::shared_ptr<std::experimental::promise<FinalResult>> final_result = std::make_shared<std::experimental::promise<FinalResult>>();
    struct DoneCheck
    {
        std::shared_ptr<std::experimental::promise<FinalResult>> final_result;

        DoneCheck(std::shared_ptr<std::experimental::promise<FinalResult>> final_result_)
            : final_result(std::move(final_result_)) {}

        void operator()(std::experimental::future<std::experimental::when_any_result<std::vector<std::experimental::future<MyData *>>>> results_param)
        {
            auto results = results_param.get();
            MyData *const ready_result = results.futures[results.index].get();  // 从 future 中获得对应的值
            if (ready_result) // 如果值不为空，则已经找到符合条件的值
            {
                final_result->set_value(process_found_value(*ready_result));
            }
            else
            {
                results.futures.erase(results.futures.begin() + results.index); // 移除处理过的结果
                if (!results.futures.empty()) 
                {
                    std::experimental::when_any(results.futures.begin(), results.futures.end()).then(std::move(*this)); // 继续处理完成的任务
                }
                else
                {
                    final_result->set_exception(std::make_exception_ptr(std::runtime_error("Not found")));
                }
            }
        }
    };
    std::experimental::when_any(results.begin(), results.end()).then(DoneCheck(final_result));  // 当任何一个任务返回后，执行DoneCheck
    return final_result->get_future();
}

std::experimental::future<int> f1=spawn_async(func1);
std::experimental::future<std::string> f2=spawn_async(func2);
std::experimental::future<double> f3=spawn_async(func3);
std::experimental::future<
    std::tuple<
        std::experimental::future<int>,
        std::experimental::future<std::string>,
        std::experimental::future<double>>> result =
    std::experimental::when_all(std::move(f1),std::move(f2),std::move(f3));

struct my_data
{
    int i;
    double d;
    unsigned bf1       :10;
    int      bf2       :23;
    int      /*bf3*/   :0;
    int      bf4       :5;
    int i2;
    char c1,c2;
    std::string s;
};

class spinlock_mutex
{
    std::atomic_flag flag;
public:
    spinlock_mutex(): flag(ATOMIC_FLAG_INIT) {}
    void lock()
    {
        // 检查flag是否被设置（即flag是否为true）
        // 若被设置直接返回true，若没有设置则设置flag为true后再返回false
        while(flag.test_and_set(std::memory_order_acquire));
    }
    void unlock()
    {
        // 清除flag标志即flag=false
        flag.clear(std::memory_order_release);
    }
};

class Foo{};
Foo some_array[5];
std::atomic<Foo*> p(some_array);
Foo* x=p.fetch_add(2);  // p加2，并返回原始值
assert(x==some_array);
assert(p.load()==&some_array[2]);

x=(p-=1);  // p减1，并返回新值
assert(x==&some_array[1]);
assert(p.load()==&some_array[1]);

std::atomic_store(&atomic_var,new_value);
std::atomic_store_explicit(&atomic_var,new_value,std::memory_order_release);

// 下面代码返回值相同
std::atomic_is_lock_free(&a);
a.is_lock_free();

// atomic_flag 对应的 free function如下，名称中包含了flag
std::atomic_flag_test_and_set();
std::atomic_flag_test_and_set_explicit()

std::shared_ptr<my_data> p;
void process_global_data()
{
    // 还可以使用 std::atomic_load_explicit 版本来指定 memory-ordering
    // 同样可以使用 std::atomic_is_lock_free(&p) 来判断，实现是否是lock-free的
    std::shared_ptr<my_data> local=std::atomic_load(&p);
    process_data(local);
}
void update_global_data()
{
    std::shared_ptr<my_data> local(new my_data);
    // 还可以使用 std::atomic_store_explicit 版本来指定 memory-ordering
    std::atomic_store(&p,local);
}

#include <iostream>
void foo(int a,int b)
{
    std::cout<<a<<","<<b<<std::endl;
}
int get_num()
{
    static int i=0;
    return ++i;
}
int main()
{
    // 可能输出 2, 1 也可能输出 1, 2
    foo(get_num(),get_num());
}

#include <atomic>
#include <thread>

std::vector<int> queue_data;
std::atomic<int> count;

void populate_queue()
{
    unsigned const number_of_items=20;
    queue_data.clear();
    for(unsigned i=0;i<number_of_items;++i)
    {
        queue_data.push_back(i);
    }

    count.store(number_of_items, std::memory_order_release);  // 1 初始化存储
}

void consume_queue_items()
{
    while(true)
    {
        int item_index;
        if((item_index=count.fetch_sub(1,std::memory_order_acquire))<=0)  // 2 一个“读-改-写”操作
        {
            wait_for_more_items();  // 3 等待更多元素
            continue;
        }
        process(queue_data[item_index-1]);  // 4 安全读取queue_data
    }
}

int main()
{
    std::thread a(populate_queue);
    std::thread b(consume_queue_items);
    std::thread c(consume_queue_items);
    a.join();
    b.join();
    c.join();
}

#include <atomic>
#include <thread>
#include <assert.h>

std::atomic<bool> x,y;
std::atomic<int> z;

void write_x_then_y()
{
    x.store(true,std::memory_order_relaxed);  // 1
    std::atomic_thread_fence(std::memory_order_release);  // 2
    y.store(true,std::memory_order_relaxed);  // 3
}

void read_y_then_x()
{
    while(!y.load(std::memory_order_relaxed));  // 4
    std::atomic_thread_fence(std::memory_order_acquire);  // 5
    if(x.load(std::memory_order_relaxed))  // 6
        ++z;
}

int main()
{
    x=false;
    y=false;
    z=0;
    std::thread a(write_x_then_y);
    std::thread b(read_y_then_x);
    a.join();
    b.join();
    assert(z.load()!=0);  // 7
}

#include <atomic>
#include <thread>
#include <assert.h>

bool x=false;  // x现在是一个非原子变量
std::atomic<bool> y;
std::atomic<int> z;

void write_x_then_y()
{
    x=true;  // 1 在栅栏前存储x
    std::atomic_thread_fence(std::memory_order_release);
    y.store(true,std::memory_order_relaxed);  // 2 在栅栏后存储y
}

void read_y_then_x()
{
    while(!y.load(std::memory_order_relaxed));  // 3 在#2写入前，持续等待
    std::atomic_thread_fence(std::memory_order_acquire);
    if(x)  // 4 这里读取到的值，是#1中写入
        ++z;
}
int main()
{
    x=false;
    y=false;
    z=0;
    std::thread a(write_x_then_y);
    std::thread b(read_y_then_x);
    a.join();
    b.join();
    assert(z.load()!=0);  // 5 断言将不会触发
}

class spinlock_mutex
{
    std::atomic_flag flag;
public:
    spinlock_mutex(): flag(ATOMIC_FLAG_INIT) {}
    void lock()
    {
        // 检查flag是否被设置（即flag是否为true）
        // 若被设置直接返回true，若没有设置则设置flag为true后再返回false
        while(flag.test_and_set(std::memory_order_acquire));
    }
    void unlock()
    {
        // 清除flag标志即flag=false
        flag.clear(std::memory_order_release);
    }
};

#include <exception>

struct empty_stack: std::exception
{
    const char* what() const throw();
};

template<typename T>
class threadsafe_stack
{
private:
    std::stack<T> data;
    mutable std::mutex m;
public:
    threadsafe_stack(){}
    threadsafe_stack(const threadsafe_stack& other)
    {
        std::lock_guard<std::mutex> lock(other.m);
        data=other.data;
    }

    threadsafe_stack& operator=(const threadsafe_stack&) = delete;

    void push(T new_value)
    {
        std::lock_guard<std::mutex> lock(m);
        data.push(std::move(new_value));  // 1
    }
    std::shared_ptr<T> pop()
    {
        std::lock_guard<std::mutex> lock(m);
        if(data.empty()) throw empty_stack();  // 2
        std::shared_ptr<T> const res(std::make_shared<T>(std::move(data.top())));  // 3
        data.pop();  // 4
        return res;
    }
    void pop(T& value)
    {
        std::lock_guard<std::mutex> lock(m);
        if(data.empty()) throw empty_stack();
        value=std::move(data.top());  // 5
        data.pop();  // 6
    }
    bool empty() const
    {
        std::lock_guard<std::mutex> lock(m);
        return data.empty();
    }
};

template<typename T>
class threadsafe_queue
{
private:
    mutable std::mutex mut;
    std::queue<T> data_queue;
    std::condition_variable data_cond;

public:
    threadsafe_queue()
    {}

    void push(T data)
    {
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(std::move(data));
        data_cond.notify_one();  // 1
    }

    void wait_and_pop(T& value)  // 2
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        value=std::move(data_queue.front());
        data_queue.pop();
    }

    std::shared_ptr<T> wait_and_pop()  // 3
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});  // 4
        std::shared_ptr<T> res(std::make_shared<T>(std::move(data_queue.front())));
        data_queue.pop();
        return res;
    }

    bool try_pop(T& value)
    {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
            return false;
        value=std::move(data_queue.front());
        data_queue.pop();
        return true;
    }

    std::shared_ptr<T> try_pop()
    {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty()) return std::shared_ptr<T>();  // 5
        std::shared_ptr<T> res(std::make_shared<T>(std::move(data_queue.front())));
        data_queue.pop();
        return res;
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};

template<typename T>
class threadsafe_queue
{
private:
    mutable std::mutex mut;
    std::queue<std::shared_ptr<T> > data_queue;
    std::condition_variable data_cond;
public:
    threadsafe_queue()
    {}

    void wait_and_pop(T& value)
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        value=std::move(*data_queue.front());  // 1
        data_queue.pop();
    }

    bool try_pop(T& value)
    {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty()) return false;
        value=std::move(*data_queue.front());  // 2
        data_queue.pop();
        return true;
    }

    std::shared_ptr<T> wait_and_pop()
    {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk,[this]{return !data_queue.empty();});
        std::shared_ptr<T> res=data_queue.front();  // 3
        data_queue.pop();
        return res;
    }

    std::shared_ptr<T> try_pop()
    {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty()) return std::shared_ptr<T>();
        std::shared_ptr<T> res=data_queue.front();  // 4
        data_queue.pop();
        return res;
    }

    void push(T new_value)
    {
        std::shared_ptr<T> data(std::make_shared<T>(std::move(new_value)));  // 5
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(data);
        data_cond.notify_one();
    }

    bool empty() const
    {
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};

template<typename T>
class queue
{
private:
    struct node
    {
        T data;
        // 使用std::unique_ptr 确保不再使用时，数据能被delete
        std::unique_ptr<node> next;

        node(T data_): data(std::move(data_)) {}
    };

    std::unique_ptr<node> head;  // 1
    // tail使用了原始指针，因为其所指node已经被std::unique_ptr<node> 所拥有
    node* tail;  // 2

public:
    queue()
    {}
    queue(const queue& other)=delete;
    queue& operator=(const queue& other)=delete;
    std::shared_ptr<T> try_pop()
    {
        if(!head)
        {
            return std::shared_ptr<T>();
        }
        std::shared_ptr<T> const res(std::make_shared<T>(std::move(head->data)));
        std::unique_ptr<node> const old_head=std::move(head);
        head=std::move(old_head->next);  // 3
        if(!head) tail=nullptr;
        return res;
    }

    void push(T new_value)
    {
        std::unique_ptr<node> p(new node(std::move(new_value)));
        node* const new_tail=p.get();
        if(tail)
        {
            tail->next=std::move(p);  // 4
        }
        else
        {
            head=std::move(p);  // 5
        }
        tail=new_tail;  // 6
    }
};

mapped_type get_value(key_type const& key, mapped_type default_value);

template<typename T>
class threadsafe_list
{
    struct node  // 1
    {
        std::mutex m;
        std::shared_ptr<T> data;
        std::unique_ptr<node> next;
        //  2
        node(): next() {}
        // 3
        node(T const& value): data(std::make_shared<T>(value)) {}
    };

    node head;

public:
    threadsafe_list() {}
    ~threadsafe_list() { remove_if([](node const&){return true;}); }

    threadsafe_list(threadsafe_list const& other)=delete;
    threadsafe_list& operator=(threadsafe_list const& other)=delete;

    void push_front(T const& value)
    {
        std::unique_ptr<node> new_node(new node(value));  // 4
        std::lock_guard<std::mutex> lk(head.m);
        new_node->next=std::move(head.next);  // 5
        head.next=std::move(new_node);  // 6
    }

    template<typename Function>
    void for_each(Function f)  // 7
    {
        node* current=&head;
        std::unique_lock<std::mutex> lk(head.m);  // 8
        while(node* const next=current->next.get())  // 9
        {
            std::unique_lock<std::mutex> next_lk(next->m);  // 10
            lk.unlock();  // 11
            f(*next->data);  // 12
            current=next;
            lk=std::move(next_lk);  // 13
        }
    }

    template<typename Predicate>
    std::shared_ptr<T> find_first_if(Predicate p)  // 14
    {
        node* current=&head;
        std::unique_lock<std::mutex> lk(head.m);
        while(node* const next=current->next.get())
        {
            std::unique_lock<std::mutex> next_lk(next->m);
            lk.unlock();
            if(p(*next->data))  // 15
            {
                return next->data;  // 16
            }
            current=next;
            lk=std::move(next_lk);
        }
        return std::shared_ptr<T>();
    }

    template<typename Predicate>
    void remove_if(Predicate p)  // 17
    {
        node* current=&head;
        std::unique_lock<std::mutex> lk(head.m);
        while(node* const next=current->next.get())
        {
            std::unique_lock<std::mutex> next_lk(next->m);
            if(p(*next->data))  // 18
            {
                std::unique_ptr<node> old_next=std::move(current->next);
                current->next=std::move(next->next);
                next_lk.unlock();
            }  // 20
            else
            {
                lk.unlock();  // 21
                current=next;
                lk=std::move(next_lk);
            }
        }
    }
};

class spinlock_mutex
{
    std::atomic_flag flag;
public:
    spinlock_mutex(): flag(ATOMIC_FLAG_INIT) {}
    void lock()
    {
        while(flag.test_and_set(std::memory_order_acquire));
    }
    void unlock()
    {
        flag.clear(std::memory_order_release);
    }
};