# C++ thread
###### tags: `cpp`

要在C++使用thread，要記得加編譯器參數
`-std=c++11 -pthread`

## 參考
[thispointer.com](https://thispointer.com/c11-multithreading-tutorial-series/)
[cplusplus.com](http://www.cplusplus.com/reference/multithreading/)
[cppreference.com](https://en.cppreference.com/w/cpp/thread)
[jyt0532 multithread](https://www.jyt0532.com/2016/12/23/c++-multi-thread-p1/)

以下都是std::的，不是boost::

---

## thread
就像一個指向執行緒的指標，在建構時就會順便建構該執行緒，可以用該指標間接控制該執行緒，若detach()該執行緒也只是無法繼續控制該執行緒，該執行緒還會繼續存在，但和指標不一樣的是，new出來的物件如果沒有delete就會一直存在，而創造出來的執行緒只要執行完畢就會自動消失。
> 上面說的指標只是一個概念

// 建立一個執行緒，此執行緒會執行 Fn(Args...)
// 如果要傳參考的話需要加ref()
thread(Fn, Args...)

// 主執行緒會在此一直等待，直到此執行緒執行完
join()

// 檢查此執行緒是否可以 join (是否還和主執行緒連接)
joinable()

// 把此執行緒和主執行緒的連接切斷
detach()

// 如果此執行緒存在，會返回唯一的識別碼
// 回傳型態為 thread::id
get_id()

> 創造者(主執行緒)和被創造者(此執行緒)的關係就像function call一樣，被創造者執行完會回到創造者的執行緒，但是可以detach讓被創造者獨立出去。

---

使用範例

// 此執行緒(thr)會執行一個跑n次的迴圈，n=100
auto thr = thread([](int n){
    for(int i=0; i<n; ++i);
}, 100);

// 切斷連結
thr.detach();

// 因為thr已經和主執行緒分開了，所以回傳false
if(thr.joinable())
    thr.join();

> 主執行緒 : 呼叫 thread() 的執行緒
> 此執行緒 : thread() 建構出的執行緒

---

### this_thread::

// 回傳此執行緒的唯一識別碼
get_id()

// 暫時中斷此執行緒，讓他去外面排隊等著再次被執行
yield()

// 暫時中斷此執行緒，讓他去外面等一段時間再去排隊等著被執行
sleep_for(chrono::duration<>)

// 暫時中斷此執行緒，讓他去外面等到某個時間點再去排隊等著被執行
sleep_until(chrono::time_point<>)

> CPU同時能跑的執行緒是有上限的，如果你創造很多條執行緒，實際上只會有幾條再跑，其他的都在排隊等著被執行，但執行緒可能執行一段時間就會自動yield出去換別人跑(分時多工)，並不會等先到的執行緒整個結束才讓後面的執行。

---

#### chrono::duration
duration<>可以用這些定義的型別取代，比較直覺

std::chrono::nanoseconds()
std::chrono::microseconds()
std::chrono::milliseconds()
std::chrono::seconds()
std::chrono::minutes()
std::chrono::hours()

---

使用範例

// main
cout << "this thread is " << this_thread::get_id() << endl;

for (const char c : "Hello world"){
    cout << c;
    this_thread::sleep_for(chrono::seconds(1));
}

> 先印出主執行緒的id，在每隔一秒輸出一個字。

---

### call_once
如果有多個執行緒一起執行一個function，但只想讓這個function被執行一次，可以用call_once()。
`std::call_once(once_flag, Fn, Args...)`
- once_flag : 用來判斷是否已被執行，所以需要共用同一個flag
- Fn(Args...) : 與一般thread呼叫方式一樣

範例

void test(){
    cout << "once";
}

void tester(){
    static once_flag f;
    call_once(f, test); // 由他呼叫的test()只會被執行一次
}

int main()
{
    for (int i = 0; i < 10; ++i){
        thread(tester).detach();
    }
    return 0;
}

> call_once執行時，會先判斷flag，如果還沒執行過就設定flag並執行，原理就像macro的引入保護一樣。

---

### ! race condition !
因為執行緒會共用資料，所以可能發生同一份資料同時被多條執行緒寫入，結果有可能會和預想的不一樣，要解決這個問題就要用到lock。

---

## lock
C++11的lock有分為:
- 基本型(較不安全):
- mutex
- timed_mutex
- recursive_mutex
- recursive_timed_mutex
- 包裝型(把基本型包起來使用，解構時會自動解鎖，較安全):
- lock_guard<>
- unique_lock<>

---

### mutex

// 互斥鎖鎖定，當其他執行緒也執行到該互斥鎖的lock()時，會等待直到該互斥鎖unlock()
lock()

// 解除鎖定
unlock()

// 若該互斥鎖被鎖定，會回傳false並繼續執行，沒被鎖定就回傳true繼續執行，最好搭配if使用
try_lock()

> 當一執行緒成功lock()，取得該mutex所有權，則只有該執行緒能對該mutex進行unlock()。

---

範例一 lock

void test(){
    static mutex m;
    m.lock(); // maybe waiting
    /*
     *  atomic section
     */
    m.unlock();
    /*
     *  maybe occur race condition
     */
}

---

範例二 try_lock

void test(){
    static mutex m;
    if(m.try_lock()){ // no waiting
        /*
         *  atomic section
         */
        m.unlock();
    }
    /*
     *  maybe occur data condition
     */
}

---

### special mutex
除了基本的mutex，還有幾個特殊功能的mutex

- `timed_mutex` 比基本mutex多了兩個時間功能。
- `try_lock_for(chrono::duration<>)`
- 在一段時間內會一直try_lock(阻塞一段時間)，時間內成功lock回傳true並繼續，超時還沒lock回傳false並繼續。
- `try_lock_until(chrono::time_point<>)`
- 同上，但會一直持續到指定的時間點。
- `recursive_mutex` 可用在遞迴的mutex。
- 最好搭配unique_lock使用，不要自行呼叫unlock()
- `recursive_timed_mutex` 結合以上兩種mutex

---

### *C++14 up : SharedMutex*
為了更方便的製作讀寫鎖，而多了可以共用所有權的mutex
- `shared_timed_mutex` (C++14)
- `shared_mutex` (C++17)

他們的使用方式分成兩種級別，unique、shared
- unique : lock後，只有自己擁有所有權
- lock()
- try_lock()
- unlock()
- shared : lock後，其他人只能用shared方式取得所有權，但所有權會共享
- lock_shared()
- try_lock_shared()
- unlock_shared()

---

### lock_guard
mutex::lock() 有一個問題是**若還沒執行 unlock() 就發生例外跳出去，導致沒人執行 unlock() ，導致一直都是 lock 狀態**。
解決這個問題的就是 lock_guard，原理是當發生例外往外跳時，因為離開scope，所以會解構在這個scope內創造的東西。

mutex m;
{
    lock_guard<mutex> lg(m); // 建構子會執行m.lock()
    /*
     * atomic
     */ 
} // 解構子會執行m.unlock()

---

### unique_lock
lock_guard 保護的範圍是整個 scope ，如果要縮小範圍的話，就用 unique_lock 解決。
和 lock_guard一樣有保護功能，還能使用mutex的功能。
他就像一個指向mutex的智慧指標，需要指向一個mutex，可以使用該mutex的功能，也可用release取消指向該mutex，還會自動使用unlock。

除了可使用mutex的功能以外，還有以下功能

owns_lock() // 是否擁有該mutex的所有權
release()   // 釋放所指向的mutex，並回傳指向他的指標
mutex()     // 回傳指向mutex的指標，但不釋放

---

### constructor flag
unique_lock<>還有另外三種建構模式

// 可用在 unique_lock的建構子上
// 在建構時執行try_lock()
std::try_to_lock

// 可用在 unique_lock的建構子上
// 在建構時"先"不執行lock()
std::defer_lock

// 可用在 unique_lock、lock_guard的建構子上
// 假設在建構前，已經把mutex lock住了
// 所以在建構時"不會再"執行lock()，之後也不會執行lock()
std::adopt_lock

---

範例

mutex m;
{
    unique_lock<mutex> ul(m, try_to_lock); // 建構子會執行m.try_lock()
    if(ul.owns_lock()){
        /*
         * atomic section.
         */
        ul.unlock();
    }
    /*
     * may occur race condition.
     */
     
} // 若還沒unlock()會執行unlock()

---

### *C++14 up : shared_lock<>*
可以使用指向ShareMutex，並用shared方式使用他，配合unipue_lock<>可實作讀寫鎖。

shared_mutex m;

void read(){
    shared_lock<share_mutex> lock(m);
    // 同時擁多個執行緒擁有所有權
}

void write(){
    unique_lock<shared_mutex> lock(m);
    // 同時只有一個執行緒擁有所有權
}

---

### ! dead lock !
當有一執行緒tA 占用資源rA 並等待資源rB
另有一執行緒tB 占用資源rB 並等待資源rA
雙方都沒辦法執行，沒執行完就不會釋放自己的資源，於是一直互相等待。
當這情況發生時，就只能結束整個程式，所以要盡量避免。

---

### multiple lock
為解決這個問題，有以下兩個function，可以一次拿到所有需要的資源。
不過當你要求的資源太多，可能一直沒機會全部拿到，然後就一直等待到所有人都執行結束釋放資源。

std::lock(mutex1, mutex2, ...)
std::try_lock(mutex1, mutex2, ...)
// 這裡參數的mutex可以是一般的mutex或各種特殊功能的mutex，像lock_guard、unique_lock...

---

### *C++17 up : scoped_lock*
如上述multiple lock，在C++17以後可以使用scoped_lock，有同樣的效果，並且和lock_guard<>一樣，會自動unlock。

scoped_lock lock(mutex1, mutex2, ...)
scoped_lock lock(adopt_lock, mutex1, mutex2, ...)

---

## condition_variable
若要讓一個thread準備好資料後通知另一個thread使用，可以用condition_variable
可以依照條件從當前的thread跳出到 condition_variable 內部的一個 queue等待，之後需要時再喚醒他，這樣就不用浪費一個thread一直等待。

---

### wait()

// 會跳出當前執行緒
// 並把此執行緒加入condition_varible內部的queue
// 再把該互斥鎖unlock()
wait(unique_lock<mutex>)

// 若f回傳false: 繼續wait
// 若f回傳true: 不會做事，繼續執行
wait(unique_lock<mutex>, function<bool()>f)

---

### notify

// 喚醒queue內第一個執行緒繼續執行
notify_one()

// 喚醒queue內所有執行緒繼續執行
notify_all()

---

範例

condition_variable cv;
mutex m;
bool ready = false;

void output_data(string* message){
    unique_lock<mutex> ul(m);

    while(!ready)
        cv.wait(ul);

    cout << message->c_str() << endl;
}

void prepare_data(string* message){
    cout << "prepare message ...\n";
    this_thread::sleep_for(chrono::seconds(5));
    *message = "hello world";
    ready = true;
    cout << "message ready, continue thread.\n";
    cv.notify_one();
}

// main
string message;
thread(prepare_data, &message).detach();
thread(output_data, &message).detach();

---

### condition_variable_any
condition_variable<>只能使用unique_lock<>，而多了any就是可以使用任何lockable的東西。

---

## future
若要讓一個thread準備好資料後通知另一個thread使用，也可以用future和promise。

// promise<dataType>
get_future() // 回傳一個與之對應的future物件
set_value()  // 設定共用資料，並通知future物件繼續執行

// future<dataType>
get() // 會等待直到promise設定好資料，並回傳該資料
wait()// 會等待直到promise設定好資料，但不會回傳該資料

wait_for(chrono::duration) 
// 會等待一段時間
// 若在時間內得到資料 回傳future_status::ready 並結束等待
// 時間超過了還沒拿到資料 回傳future_status::timeout 並結束等待

---

範例

void output_data(future<string>& message){
    cout << message.get().c_str() << endl;
}

void prepare_data(promise<string>& message){
    cout << "prepare message ...";
    this_thread::sleep_for(chrono::seconds(5));

    message.set_value("hello world");
    cout << "message ready, continue thread. \n";
}

// main
    promise<string> message;
    future<string> futureObj = message.get_future();

    thread(prepare_data, ref(message)).detach();
    thread(output_data, ref(futureObj)).detach();

---

### async()
建立一個有回傳值的thread，將它的回傳值保存再future物件中，需要時再拿出來。

future<T> async(launch_policy, function, args...)
// 創造一個thread執行function(args...)
// 當function執行完，回傳值會被的future接收
// launch_policy有兩種
    // launch::async 馬上執行function
    // launch::deferred 先不執行function，在future::get時才會執行
    // 如果不指定launch_policy，系統會自動選擇

---

範例

string prepareMessage(){
    this_thread::sleep_for(chrono::seconds(3));
    return "Hello world";
}

// main
future<string> message = async(prepareMessage);
/*
 *  do something ...
 */
cout << message.get();

> 如果function還沒執行完就要get他的值，那會等待他執行完。(阻塞)

---

### packaged_task<>
可以算是async()的底層，把function和promise包成一個物件，此物件可像function object一樣使用，若要get回傳值一定要等到function執行完，不然會一直阻塞，可用move()傳給thread()來達成async()的功能。

packaged_task<returnType(Args...)>(function)
// 建構式和function class差不多
// 也像function一樣可以用()運算子

.get_future()
// 跟promise一樣，回傳一個future<returnType>物件

---

範例一(單執行緒)

string prepareMessage(){
    this_thread::sleep_for(chrono::seconds(3));
    cout << "prepare complete\n";
    return "Hello world";
}

// main
    packaged_task<string()> task(prepareMessage);
    auto result = task.get_future();
    
    task();

    cout << result.get().c_str();

---

範例二(多執行緒)

string prepareMessage(){
    this_thread::sleep_for(chrono::seconds(3));
    cout << "prepare complete\n";
    return "Hello world";
}

// main
    packaged_task<string()> task(prepareMessage);
    auto result = task.get_future();
    
    thread(move(task)).detach();
    /*
     * 可以做其他事，不用一直等
     */
    cout << result.get().c_str();

---

# multi thread programs

## 1
100個資料，101人同時要求資源
比較有lock沒lock的差別
> 共有4種要求資料方式，每按一次 Enter 換下一個，4個後重複。
> 會輸出每種方式所花的時間、更改後的資料。

---

### include

// stdafx.h
#include "targetver.h"
#include <stdio.h>
#include <tchar.h>
#include <iostream>
#include <array>
#include <thread>
#include <mutex>
#include <ctime>
#include <functional>

---

### Data

class Data{
    static int delayTime; // 增加 data race 的機率.
    mutex m;
    int val;

public:
    Data(int val=0) :val(val){}

    int Get(){
        return val;
    }

    void Reset(){
        val = 0;
    }

    void Increase(int v=1){ // non atomic
        int tmp = val;
        for (int i = 0; i < delayTime; ++i); // delay
        tmp += v;
        val = tmp;
    }

    void IncreaseWithLock(int v=1){
        unique_lock<mutex> lock(m);

        int tmp = val;
        for (int i = 0; i < delayTime; ++i); // delay
        tmp += v;
        val = tmp;
    }
};
int Data::delayTime = 50000;

---

### Tester

struct Tester{
    array<Data, 100> data;
    array<thread, 101> people;
    clock_t clockTime; // to account consume time
    
    typedef function<void(array<Data, 100>&)> Fn;

    void test(const char* message, Fn action){
        cout << message << endl;
        ResetData();
        clockTime = clock();

        for (auto&& th : people){
            th = thread(action, ref(data));
        }

        for (auto&& th : people){
            th.join();
        }

        cout << "clock:" << clock() - clockTime << endl;
        OutputData();

        cin.get();
    }

    void ResetData(){
        for (auto& e : data)
            e.Reset();
    }

    void OutputData(){
        for (auto& e : data)
            cout << e.Get() << " ";
    }
};

---

### functions

void requireDataWithLock(array<Data, 100>& data){
    static mutex m;
    unique_lock<mutex> lock(m);
    for (auto&& e : data){
        e.Increase();
    }
}

void requireDataWithoutLock(array<Data, 100>& data){
    for (auto&& e : data){
        e.Increase();
    }
}

void requireAtomicDataWithLock(array<Data, 100>& data){
    static mutex m;
    unique_lock<mutex> lock(m);
    for (auto&& e : data){
        e.IncreaseWithLock();
    }
}

void requireAtomicDataWithoutLock(array<Data, 100>& data){
    for (auto&& e : data){
        e.IncreaseWithLock();
    }
}

---

### main

int main(){
    Tester tester;

    while (true){
        tester.test("Require Data With Lock", requireDataWithLock);
        tester.test("Require Data Without Lock", requireDataWithoutLock);
        tester.test("Require Atomic Data With Lock", requireAtomicDataWithLock);
        tester.test("Require Atomic Data Without Lock", requireAtomicDataWithoutLock);
    }

    return 0;
}

---

## 2
10個攤位，每個攤位都有一個服務客人的時間，一次服務一個客人
多組客人去吃飯，每個客人都有不同的等待耐心當耐心耗完時就會離開不繼續排隊
固定時間會增加客人去排隊

印出每個客人的等待時間, 與是否離開等待
> 這是單執行緒程式，用來模擬多執行緒發生的事
> 因為是一步一步執行的，可以看到每一步發生什麼事
> 按 Enter 下一步

---

### include

// stdafx.h
#include "targetver.h"
#include <stdio.h>
#include <tchar.h>
#include <iostream>
#include <array>
#include <list>
#include <memory>

---

### Client

class Client{
    static int count;
    static int minPatience;
    static int rangePatience;

    int waitTime;
    bool waiting;

public:
    const int id;
    const int patienceTime;

    Client(int patienceTime)
        :patienceTime(patienceTime), id(count++), waitTime(0), waiting(true){
        cout << "client[" << id << "] join , patience : " << patienceTime << endl;
    }

    Client() :Client(rand() % rangePatience + minPatience){}


    void Clock(){
        if (waiting)
            ++waitTime;
    }

    void BeServe(int storeId){
        waiting = false;
        cout << "client[" << id << "] be served by store [" << storeId << "], wait for : " << waitTime << endl;
    }

    void NoPatienceExit(){
        cout << "client[" << id << "], no patience exit, wait for : " << waitTime  << endl;
    }

    bool IsNoPatience()const{
        return waitTime == patienceTime;
    }

};
int Client::count = 0;
int Client::minPatience = 3;
int Client::rangePatience = 2;

---

### Store

class Store{
    static int count;
    static int minServiceTime;
    static int rangeServiceTime;

    unique_ptr<Client> client;
    int passTime;

public:
    const int id;
    const int serviceTime;

    Store(int serviceTime)
        :serviceTime(serviceTime), id(count++), client(nullptr), passTime(0){
        cout << "store [" << id << "]: service time : " << serviceTime << endl;
    }

    Store() :Store(rand() % rangeServiceTime + minServiceTime){}

    void Clock(){
        if (client)
            ++passTime;
    }
    
    void Serve(unique_ptr<Client>&& client){
        client->BeServe(id);
        this->client = move(client);
        passTime = 0;
    }

    void Pop(){
        if (client){
            cout << "store[" << id << "] service finish, client[" << client->id << "] exit" << endl;
            client.release();
        }
    }
    
    bool IsFinish()const{
        return passTime == serviceTime;
    }
    
    bool IsEmpty()const{
        return !(bool)(client);
    }
};
int Store::count = 0;
int Store::minServiceTime = 30;
int Store::rangeServiceTime = 20;

---

### main (scheduler)

int main()
{
    array<Store, 10> stores;
    list< unique_ptr<Client> > clients;

    const int addTime = 2;
    int passTime = 0;

    while (true){

        // add client
        if (passTime%addTime == 0){
            clients.push_back(make_unique<Client>());
        }
        
        // store clock
        for (auto& store : stores)
        {
            if (store.IsFinish())
            	store.Pop();

            if (store.IsEmpty() && !clients.empty()){
            	store.Serve(move(clients.front()));
            	clients.pop_front();
            }

            store.Clock();
        }

        // client clock
        for (auto it = clients.begin(); it != clients.end(); ++it)
        {
            unique_ptr<Client>& client = *it;
            client->Clock();

            if (client->IsNoPatience()){
            	client->NoPatienceExit();
            	it = clients.erase(it);
            }
        }

        ++passTime;
        cout << "--Clock-- "<<passTime;
        cin.get();
    }

    return 0;
}

---

## 3
哲學家用餐問題
把Philosopher::Action裡的 TryToEat() 改成 TryToEat_MultipleLock() 就可解決 deadlock
Philosopher::delay 越大越容易發生deadlock

---

### include

// stdafx.h
#include "targetver.h"
#include <stdio.h>
#include <tchar.h>

#include <ctime>
#include <iostream>
#include <thread>
#include <mutex>
#include <chrono>
#include <vector>

---

### Philosopher

class Philosopher{
    enum Status{think, hungry, eat};

    static int count;
    static int delay;
    static int minEatTime;
    static int minThinkTime;
    static int rangeEatTime;
    static int rangeThinkTime;

    mutex& leftFork;
    mutex& rightFork;
    Status status;
    int forkNum;
public:
    const int id;

    Philosopher(mutex& left, mutex& right)
    	:id(count++), leftFork(left), rightFork(right), status(Status::think), forkNum(0)
    {
        thread(&Philosopher::Action,this).detach();
    }

    void ShowStatus(){
        cout << "Philosopher[" << id << "] : " << StatusStr() << "\thas forks : " << forkNum << endl;
    }

private:
    void Action(){
        while (true){
            Think();
            TryToEat();
            forkNum = 0;
        }
    }

    void TryToEat(){ // may occur deadlock
        status = Status::hungry;

        unique_lock<mutex> l_fork(leftFork);
        ++forkNum;

        for (int i = 0; i < delay; ++i); // 增加發生機率.

        unique_lock<mutex> r_fork(rightFork);
        ++forkNum;

        status = Status::eat;
        this_thread::sleep_for(chrono::milliseconds(rand() % rangeEatTime + minEatTime));
        cout << "eat finish" << endl;
    }

    void TryToEat_multipleLock(){ // can solve deadlock
        status = Status::hungry;

        unique_lock<mutex> l_fork(leftFork, std::defer_lock);
        unique_lock<mutex> r_fork(rightFork, std::defer_lock);
        lock(l_fork, r_fork);
        forkNum = 2;

        status = Status::eat;
        this_thread::sleep_for(chrono::milliseconds(rand() % rangeEatTime + minEatTime));
        cout << "eat finish" << endl;
    }

    void Think(){
        status = Status::think;
        this_thread::sleep_for(chrono::milliseconds(rand() % rangeThinkTime + minThinkTime));
    }

    const char* StatusStr(){
        switch (status)
        {
        case Philosopher::think:
            return "think";
        case Philosopher::hungry:
            return "hungry";
        case Philosopher::eat:
            return "eat";
        default:
            break;
        }
    }
};
int Philosopher::count = 0;
int Philosopher::delay = 200000;
int Philosopher::minEatTime = 1000;
int Philosopher::minThinkTime = 10000;
int Philosopher::rangeEatTime = 4000;
int Philosopher::rangeThinkTime = 2000;

---

### Table

class Table{
    vector<mutex> forks;
    vector<Philosopher> philosophers;

public:
    Table(int number):forks(number){
        philosophers.reserve(number);
        
        for (int i = 0; i < number-1; ++i){
            philosophers.emplace_back(forks[i], forks[i + 1]);
        }
        philosophers.emplace_back(forks[number-1], forks[0]);
    }

    void Action(){
        while (true){
            for (auto&& p : philosophers){
            	p.ShowStatus();
            }
            system("cls");
        }
    }
};

---

### main

int main()
{
    srand(time(nullptr));
    Table table(8);
    table.Action();
    return 0;
}

---