摘要：本文基于指令完成一個無等待并發(fā)算法。并且導(dǎo)致它失敗的那一方必定取得了進展。通過將包裹的，從更新為來更新狀態(tài)的同時傳遞對應(yīng)線程通過判定操作已完成。，代表這個已經(jīng)被對應(yīng)的線程預(yù)定了，剩余線程達成共識。

本文基于compareandswap指令完成一個無等待并發(fā)算法。
根據(jù)維基百科，它的定義如下：

An algorithm is wait-free if every operation has a bound on the number of steps the algorithm will take before the operation completes.

本文的方法參考了Wait-free queues with multiple enqueuers and dequeuers，A methodology for creating fast wait-free data structures，A practical wait-free simulation for lock-free data structures。
它的目標非常簡單，就是把一個值通過原子指令CMPXCHG加N，然后返回原來的值。
lock-free版本如下：

    public final int getAndIncrement(int add) {
        for (;;) {
            int current = get();
            int next = current + add;
            if (compareAndSet(current, next))
                return current;
        }
    }

那么，這里的方式是，compareAndSet失敗了重試。并且導(dǎo)致它失敗的那一方必定取得了進展。
這里的問題在于，究竟需要重試多少次compareAndSet才能成功呢？
假如一直處于競爭環(huán)境下，這個重試次數(shù)是沒有上限的。這里也就是饑餓的問題。
有沒有辦法給出一個上限，哪怕這個上限比較大？
這種并發(fā)的性質(zhì)被稱為wait-freedom。
先給出一個我實現(xiàn)好了的代碼，之后再來講思路：

    public static int getAndIncrement(int index, int add) {
        //fast-path， 最多MAX次。
        int count = MAX;
        for(;;) {
            ValueObj valueObj_ = valueObj;
            if(valueObj_.threadObj == null) {
                ValueObj valueObjNext = new ValueObj(valueObj_.value + add, null);
                if(casValueObj(valueObj_, valueObjNext)) {
                    StateObj myState = states[index];
                    //前進一步，每assistStep，嘗試一個幫助。
                    if(((++myState.steps) & myState.assistStep) == 0){
                        long helpThread = myState.index;
                        help(helpThread);
                        //下一個協(xié)助的對象。
                        ++myState.index;
                    }
                    return valueObj_.value;
                }
                Thread.yield();Thread.yield();Thread.yield();Thread.yield();
            } else {
                helpTransfer(valueObj_);
            }
            
            if(--count == 0)
                break;
        }
//        System.out.println("here " + inter.incrementAndGet());
        for(int j = 0; j < bigYields; ++j)
            Thread.yield();
        
        //slow-path，將自己列為被幫助對象。
        ThreadObj myselfObj = new ThreadObj(new ThreadObj.WrapperObj(null, false), add);
        setThreadObj(index, myselfObj);
        //開始幫助自己
        ValueObj result = help(index);
        setThreadObj(index, null);
        return result.value;
    }
    
    // valueObj->threadObj->wrapperObj->valueObj。
    // step 1-3，每一個步驟都不會阻塞其他步驟。
    // 嚴格遵守以下順序: 
    // step 1: 通過將ValueObj指向ThreadObj:
    //         atomic: (value, null)->(value, ThreadObj)來錨定該值                      //確定該value歸ThreadObj對應(yīng)線程所有。
    // step 2: 通過將ThreadObj包裹的WrapperObj，
    //         atomic: 從(null, false)更新為(valueObj, true)來更新狀態(tài)的同時傳遞value    //對應(yīng)線程通過isFinish判定操作已完成。
    // step 3: 更新ValueObj，提升value，同時設(shè)置ThreadObj為null：
    //         atomic: (value, ThreadObj)->(value+1, null)完成收尾動作                 //此時value值回到了沒有被線程錨定的狀態(tài)，也可以看做step1之前的狀態(tài)。
    private static ValueObj help(long helpIndex) {
        helpIndex = helpIndex % N;
        ThreadObj helpObj = getThreadObj(helpIndex);
        ThreadObj.WrapperObj wrapperObj;
        if(helpObj == null || helpObj.wrapperObj == null)
            return null;
        //判定句，是否該線程對應(yīng)的操作未完成，(先取valueObj，再取isFinish，這很重要)。
        ValueObj valueObj_ = valueObj;
        while(!(wrapperObj = helpObj.wrapperObj).isFinish) {
            /*ValueObj valueObj_ = valueObj;*/
            if(valueObj_.threadObj == null) {
                ValueObj intermediateObj = new ValueObj(valueObj_.value, helpObj);
                //step1
                if(!casValueObj(valueObj_, intermediateObj)) {
                    valueObj_ = valueObj;
                    continue;
                }
                //step1: 錨定該ValueObj，接下來所有看到該valueObj的線程，都會一致地完成一系列操作.
                valueObj_ = intermediateObj;
            }
            //完成ValueObj、ThreadObj中的WrapperObj的狀態(tài)遷移。
            helpTransfer(valueObj_);
            valueObj_ = valueObj;
        }
        valueObj_ = wrapperObj.value;
        helpValueTransfer(valueObj_);
        //返回錨定的valueObj。
        return valueObj_;
    }
    
    private static void helpTransfer(ValueObj valueObj_) {
        ThreadObj.WrapperObj wrapperObj = valueObj_.threadObj.wrapperObj;
        //step2: 先完成ThreadObj的狀態(tài)遷移，WrapperObj(valueObj，true)分別表示(值，完成)，原子地將這兩個值喂給threadObj。
        if(!wrapperObj.isFinish) {
            ThreadObj.WrapperObj wrapValueFiniash = new ThreadObj.WrapperObj(valueObj_, true);
            valueObj_.threadObj.casWrapValue(wrapperObj, wrapValueFiniash);
        }
        //step3: 最后完成ValueObj上的狀態(tài)遷移
        helpValueTransfer(valueObj_);
    }
    
    private static ValueObj helpValueTransfer(ValueObj valueObj_) {
        if(valueObj_ == valueObj) {
            ValueObj valueObjNext = new ValueObj(valueObj_.value + valueObj_.threadObj.add, null);
            casValueObj(valueObj_, valueObjNext);
        }
        return valueObj_;
    }

首先全局作用域有個值ValueObj,一種自然的想法是ValueObj應(yīng)該值包含一個int值value，圖[1]:

如圖所示，無鎖并發(fā)的版本無非就是從一個值原子地變化為另一個+1的值。
但是我們要的是wait-free，勢必涉及到線程之間的互助。如何在線程之間傳遞信息從而實現(xiàn)協(xié)助呢？
我們給ValueObj增加一個字段threadObj，那么上圖的版本現(xiàn)在可以視為，圖[2]：

那么現(xiàn)在增加了一個數(shù)據(jù)ThreadObj(后面會解釋如何使用)，ValueObj會有兩種狀態(tài)：

（Value, null）

（Value, threadObj）

那么好，現(xiàn)在是兩種情況：

(value, null)，代表這個value我可以去競爭。

(value, threadObj)，代表這個value已經(jīng)被threadObj對應(yīng)的線程預(yù)定了，剩余線程達成共識。

所以情況2我們能做什么呢？
這里就涉及到無等待算法的核心技術(shù)，help，通俗地說，就是多個線程間的相互協(xié)助。
我們會先將value，原子地喂給threadObj，再將(value, threadObj)原子地遷移到(value+1, null)。
這里的順序很重要，因為假如線程A先遷移了(value, threadObj)->（value+1, nulll），那么剩下的其他所有線程都無法再看到threadObj，也就是說盡管value已經(jīng)增加了，但是threadObj對應(yīng)線程被鎖住了。只能等待線程A將value喂給threadObj。
ThreadObj中包裹著WrapperObj，WrapperObj中初始化為(null, false)，將會原子地轉(zhuǎn)變?yōu)?ValueObj, ture)，從而ThreadObj拿到了ValueObj。
所以這里一共是三個steps：
step 1: (value, null) atomic-> (value, threadObj)
step 2: threadObj.WrapperObj(null, false) atomic-> WrapperObj(valueObj, true)
stpe 3: (value, trheadObj) atomic-> (value + 1, null)
如圖所示[3]：

其中灰框黑字標識了關(guān)鍵的3個step.
圖[3]所示是一種通用的無等待構(gòu)造技術(shù)。
對于其中的ThreadObj以及每個線程協(xié)助其他線程的策略states，可以通過數(shù)組來構(gòu)造，具體可在代碼里看到。
最后我們的程序結(jié)構(gòu)，如圖[4]：

最后給出完整的可執(zhí)行代碼 WaitFreeGetAndIncrement:

https://github.com/Pslydhh/Ps...

or WaitFreeAtomic:

https://github.com/Pslydhh/Wa...

，每個loop，100個線程每個操作100000次。
https://github.com/Pslydhh/Ps...
參考文獻：
0：Wait-free synchronization
1：Wait-free queues with multiple enqueuers and dequeuers
2：A methodology for creating fast wait-free data structures
3：A practical wait-free simulation for lock-free data structures