摘要：序面包店算法是解決多個(gè)線程并發(fā)訪問(wèn)一個(gè)共享的單用戶資源的互斥問(wèn)題的算法。面包店一次只能接待一位顧客的采購(gòu)。已知有位顧客要進(jìn)入面包店采購(gòu)，按照次序安排他們?cè)谇芭_(tái)登記一個(gè)簽到號(hào)碼。顧客根據(jù)簽到號(hào)碼的由小到大的順序依次入店購(gòu)貨。

Lamport面包店算法是解決多個(gè)線程并發(fā)訪問(wèn)一個(gè)共享的單用戶資源的互斥問(wèn)題的算法。由萊斯利·蘭波特發(fā)明。

Lamport把這個(gè)并發(fā)控制算法非常直觀地類比為顧客去面包店采購(gòu)。

面包店一次只能接待一位顧客的采購(gòu)。

已知有n位顧客要進(jìn)入面包店采購(gòu)，按照次序安排他們?cè)谇芭_(tái)登記一個(gè)簽到號(hào)碼。該簽到號(hào)碼逐次增加1。

顧客根據(jù)簽到號(hào)碼的由小到大的順序依次入店購(gòu)貨。

完成購(gòu)買的顧客在前臺(tái)把其簽到號(hào)碼歸0。如果完成購(gòu)買的顧客要再次進(jìn)店購(gòu)買，就必須重新排隊(duì)。

這個(gè)類比中的顧客就相當(dāng)于線程，而入店購(gòu)貨就是進(jìn)入臨界區(qū)獨(dú)占訪問(wèn)該共享資源。由于計(jì)算機(jī)實(shí)現(xiàn)的特點(diǎn)，存在兩個(gè)線程獲得相同的簽到號(hào)碼的情況，這是因?yàn)閮蓚€(gè)線程幾乎同時(shí)申請(qǐng)排隊(duì)的簽到號(hào)碼，讀取已經(jīng)發(fā)出去的簽到號(hào)碼情況，這兩個(gè)線程讀到的數(shù)據(jù)是完全一樣的，然后各自在讀到的數(shù)據(jù)上找到最大值，再加1作為自己的排隊(duì)簽到號(hào)碼。

為此，該算法規(guī)定如果兩個(gè)線程的排隊(duì)簽到號(hào)碼相等，則線程id號(hào)較小的具有優(yōu)先權(quán)。

Lamport時(shí)間戳原理如下：

每個(gè)事件對(duì)應(yīng)一個(gè)Lamport時(shí)間戳，初始值為0

如果事件在節(jié)點(diǎn)內(nèi)發(fā)生，時(shí)間戳加1

如果事件屬于發(fā)送事件，時(shí)間戳加1并在消息中帶上該時(shí)間戳

如果事件屬于接收事件，時(shí)間戳 = Max(本地時(shí)間戳，消息中的時(shí)間戳) + 1

為了請(qǐng)求資源，進(jìn)程A發(fā)送消息(Tm:A)給所有的其他進(jìn)程，并且把這個(gè)消息放到進(jìn)程隊(duì)列中，Tm是消息的時(shí)間戳

當(dāng)進(jìn)程B接收到了進(jìn)程A的(Tm:A)請(qǐng)求后，會(huì)把它放到自己的請(qǐng)求隊(duì)列，然后發(fā)送一個(gè)帶時(shí)間戳的確認(rèn)消息給A

為了釋放資源，進(jìn)程A移除所有(Tm:A)的請(qǐng)求消息，然后發(fā)送帶時(shí)間戳的A釋放資源請(qǐng)求消息給其他所有的進(jìn)程

當(dāng)進(jìn)程B接收到進(jìn)程A釋放資源的請(qǐng)求，它會(huì)移除隊(duì)列中任意的(Tm:A)的資源請(qǐng)求

當(dāng)滿足以下兩個(gè)條件時(shí)，進(jìn)程A會(huì)被分配該資源：

a)有一個(gè)(Tm:A)的請(qǐng)求，按照=>關(guān)系排在隊(duì)列第一位；

b)A接收到了一個(gè)時(shí)間戳大于Tm的來(lái)自所有其他進(jìn)程的消息

private void processRevcMsg(Message m) throws InterruptedException {
        // 原理4 如果事件屬于接收事件，時(shí)間戳 = Max(本地時(shí)間戳，消息中的時(shí)間戳) + 1
        clock.update(m.getTimestamp());
        lastSendMap.put(m.getFrom(), m);
        switch (m.getMsgType()) {
            case REQUEST_RES:
                // rule 2 當(dāng)進(jìn)程B接收到了進(jìn)程A的(Tm:A)請(qǐng)求后，會(huì)把它放到自己的請(qǐng)求隊(duì)列，然后發(fā)送一個(gè)帶時(shí)間戳的確認(rèn)消息給A
                addMessageToReqMap(m);
                Message ackMsg = new Message(pid, m.getMsgId(), MessageType.REQUEST_ACK, clock.time());
                // send ack to sender
                sendToTargetProcess(ackMsg,m.getFrom());
                break;
            case REQUEST_ACK:
                break;
            case RELEASE_RES:
                // rule 4 當(dāng)進(jìn)程B接收到進(jìn)程A釋放資源的請(qǐng)求，它會(huì)移除隊(duì)列中任意的(Tm:A)的資源請(qǐng)求
                dropMessageFromReqMap(m);
                break;
            default:
                break;
        }
        tryToAcquireResource();
    }

    private void tryToAcquireResource() {
        synchronized (reqMap) {
            if(!reqMap.containsKey(pid) || reqMap.get(pid).isEmpty()){
                return ;
            }

            Message myMessage = reqMap.get(pid).get(0);
            int acceptCount = 1;

            // rule 5 當(dāng)滿足以下兩個(gè)條件時(shí)，進(jìn)程A會(huì)被分配該資源：a)有一個(gè)(Tm:A)的請(qǐng)求，按照=>關(guān)系排在隊(duì)列第一位；b)A接收到了一個(gè)時(shí)間戳大于Tm的來(lái)自所有其他進(jìn)程的消息

            // condition (ii) of rule 5
            // A接收到了一個(gè)來(lái)自所有其他進(jìn)程的消息,而且時(shí)間戳大于Tm
            for (Map.Entry entry : lastSendMap.entrySet()) {
                if (entry.getKey() == pid) {
                    continue;
                }
                if (isFirstEarlier(myMessage, entry.getValue())) {
                    acceptCount++;
                }else{
                    return ;
                }
            }
            if (!coordinator.hasAcceptedAll(acceptCount)){
                return;
            }

            // condition (i) of rule 5
            // 有一個(gè)Tm:A的請(qǐng)求，按照=>關(guān)系排在隊(duì)列第一位
            for (Map.Entry> entry : reqMap.entrySet()) {
                if (entry.getKey() != pid && !entry.getValue().isEmpty()) {
                    if (!isFirstEarlier(myMessage, entry.getValue().get(0))) {
                        return;
                    }
                }
            }

            // remove this request message
            final Message firstMsg = reqMap.get(pid).remove(0);
            workingPool.execute(new Runnable() {
                public void run() {
                    coordinator.acquire(firstMsg.getMsgId(), pid, firstMsg.getTimestamp());
                    // emulate owning resources for a long time
                    try {
                        Thread.sleep(50L);
                        // rule 3 為了釋放資源，進(jìn)程A移除所有(Tm:A)的請(qǐng)求消息，然后發(fā)送帶時(shí)間戳的A釋放資源請(qǐng)求消息給其他所有的進(jìn)程程
                        coordinator.release(firstMsg.getMsgId(), pid, firstMsg.getTimestamp());
                        Message releaseMsg = new Message(pid, firstMsg.getMsgId(),MessageType.RELEASE_RES, clock.time());
                        sendToOtherProcesses(releaseMsg);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }
            });
        }
    }

Lamport面包店算法

分布式系統(tǒng)理論基礎(chǔ) - 時(shí)間、時(shí)鐘和事件順序

分布式系統(tǒng)時(shí)序基礎(chǔ)

lamport