Ordering

現實生活中時間可以記錄事情發生的時刻、比較事情發生的先后順序。

分布式系統的一些場景也需要記錄和比較不同節點間事件發生的順序。如數據寫入先后順序，事件發生的先后順序等等。

關系

復習下離散數學中關系：

假設A是一個集合 {1,2,3,4} ；R是集合A上的關系，例如{<1,1>,<2,2>,<3,3>,<4,4>,<1,2>,<1,4>,<2,4>,<3,4>}

• 自反性：任取一個A中的元素x，如果都有在R中，那么R是自反的。
  • <1,1>,<2,2>,<3,3>,<4,4>
• 對稱性：任取一個A中的元素x,y，如果 在關系R上,那么 也在關系R上，那么R是對稱的。
• 反對稱性：任取一個A中的元素x,y(x!=y)，如果 在關系R上,那么 不在關系R上，那么R是反對稱的。
  • 對于 <1,2>，有 <2,1> 不在R中；對于<2,4> 有<4,2>不在R中；對于<3,4> 有<4,3> 不在 R中，滿足。
• 傳遞性：任取一個A中的元素x,y,z，如果, 在關系R上，那么 也在關系R上，那么R是對稱的。
  • <1,1><1,2>在R中，并且<1,2>在R中；<1,1><1,4>在R中，并且<1,4>在R中；<2,2><2,4>在R中，并且<2,4>在R中；<3,3><3,4>在R中，并且<3,4>在R中；等等其他，滿足。
• 完全性（全關系）：包含了自反性；對集合A中所有，都有關系x到y或y到x；
  • R中并沒有<1, 3>，所以不滿足完全性

偏序The Partial Ordering

集合內只有部分元素之間是可以比較的。

偏序關系的定義（R為A上的偏序關系）：設R是集合A上的一個二元關系，若R滿足：

• 反對稱性：對任意x,y∈A，若xRy，且yRx，則x=y；
• 傳遞性：對任意x, y,z∈A，若xRy，且yRz，則xRz
• 自反性：對任意x∈A，有xRx；

一個partitial ordering關系滿足的條件是自反的，反對稱的和可傳遞的，因此在partitial ordering中，可能有兩個元素之間是不相關的。

全序The Total Ordering

集合內只有部分元素之間是可以比較的。

比如：比如復數集中并不是所有的數都可以比較大小，那么“大小”就是復數集的一個偏序關系。

全序關系的定義：

• 反對稱性：對任意x,y∈A，若xRy，且yRx，則x=y；
• 傳遞性：對任意x, y,z∈A，若xRy，且yRz，則xRz
• 完全性(total relation全關系)：對任意x,y∈A，由xRy或yRx （包括了自反性）

完全性本身也包括了自反性，所以全序關系是偏序關系。

所以偏序中滿足完全性就是全序了。

一個total ordering關系滿足的條件是反對稱的，可傳遞的和完全性，因此在total ordering中，兩個元素一定是有關系的，要么是a<>b或b<>a。

happens before

在分布式系統中，一個進程包含一系列的事件，對于同一進程內的事件，如果a happens before b，那么a發生在b之前。并且，假定收或發消息都是一個事件。

happens before的定義如下(用->表示)

• 如果a和b在同一進程中，并且a發生在b之前，那么a->b
• 如果a是一個進程發消息的事件，b是另一個進程接收這條消息的事件，則a->b
• 如果a->b且b->c，那么a->c。
• 如果同時不滿足a->b，且b->a，那么說a和b是并發的concurrent

[圖來自Time, Clocks, and the Ordering of Events in a Distributed System]

以一個例子來說明happens before關系，如上圖，垂直線上代表一個進程，從下往上，時間依次增加，水平的距離代表空間的隔離。原點代表一個事件，而曲線代表一條消息。

從圖中很容易地看出，如果一個事件a，能通過進程的線和消息線，到達b，那么a->b。

在圖中，p3和q4是并行的事件，因為，只有到了p4才能確定q4的發生，而q3也只能確定p1發生。

clock時鐘

物理時鐘

晶振和時鐘偏移

計算機有固定頻率晶體的震蕩次數，晶體的振蕩周期決定了單機的時鐘精度。

時鐘頻率也可能因為溫度等外部因素導致時鐘偏移，普通的石英晶體的漂移大10 ^-6^ 。 原子鐘的漂移約為 10^-13^ ，所以原子鐘精度遠遠高于石英晶體。

分布式下帶來的問題

不同機器上的物理時鐘難以同步，導致無法區分在分布式系統中多個節點的事件時序。即使設置了 NTP 時間同步節點間也存在毫秒級別的偏差，因而分布式系統需要有另外的方法記錄事件順序關系。

1978年Lamport在《Time, Clocks and the Ordering of Events in a Distributed System》中提出了邏輯時鐘的概念，來解決分布式系統中區分事件發生的時序問題。

邏輯時鐘Logical clocks

邏輯時鐘指的是分布式系統中用于區分事件的發生順序的時間機制。 從某種意義上講，現實世界中的物理時間其實是邏輯時鐘的特例。

Logical Clock解決的問題是找到一種方法，給分布式系統中所有時間定一個序，這個序能夠正確地排列出具有因果關系的事件（注意，是不能保證并發事件的真實順序的），使得分布式系統在邏輯上不會發生因果倒置的錯誤。因果一致性

Lamport timestamps

論文

Time, Clocks, and the Ordering of Events in a Distributed System

Lamport timestamps

Leslie Lamport 在1978年提出邏輯時鐘的概念，并描述了一種邏輯時鐘的表示方法，這個方法被稱為Lamport時間戳(Lamport timestamps)。

分布式系統中按是否存在節點交互可分為三類事件:

• 發生在節點內部
• 發送事件
• 接收事件

時鐘的定義如下

• 對于一個進程i，Ci(a)表示進程i中事件a的發生時間
• 對于整個系統來講，對于任意的事件b，其發生時間為C(b)，當b為進程j的事件時，則C(b) = Cj(b)為了使得事件按照正確的排序，需要使得如果事件a發生在事件b之前，那么a發生的時間要小于b，如下

for any events a, b

 if a->b then C(a) < C(b)

根據關系->的定義，我們可以得出

• 如果a和b都是進程i中的事件，且a發生在b之前，那么Ci(a) < Ci(b)
• 如果事件a發送消息給事件b，a屬于進程i，b屬于進程j，那么Ci(a) < Cj(b)

為了讓系統滿足上述條件，在實現中，需要滿足以下原則

• 對于每個進程，相鄰的事件的時鐘要增加1
• (a) 如果事件a是進程i發送消息m的事件，發送時帶時間戳Tm = Ci(a)，(b)事件b是進程j接受消息m的事件，那么事件b的取值為max(進程b的當前時鐘，Tm+1)

假設有事件a、b，C(a)、C(b)分別表示事件a、b對應的Lamport時間戳，如果a->b,則C(a) < C(b)，a發生在b之前(happened before)。

所以Lamport timestamps原理如下：

• 每個事件對應一個Lamport時間戳，初始值為0
• 如果事件在節點內發生，時間戳加1
• 如果事件屬于發送事件，時間戳加1并在消息中帶上該時間戳
• 如果事件屬于接收事件，時間戳 = Max(本地時間戳，消息中的時間戳) + 1

通過該定義，事件集中Lamport時間戳不等的事件可進行比較，我們獲得事件的偏序關系(partial order)。

上圖更形象的解釋了事件之間的關系。

以B4事件為基準：

• B4左邊深灰色的區域的事件，都發生在B4前，和B4具有因果關系，這些事件屬于與B4因果關系中的因(cause)
• B4右邊的深紅色區域的事件，都發生在B4后，和B4具有因果關系，這些事件屬于與B4因果關系中的果(effect)
• B4上下的白色區域是跟B4無關的事件，可以認為是并發關系(concurrent)
• 在淺灰色和淺紅色區域中的事件，C2、A3兩個事件與B4是并行關系，根據Lamport timestamps的定義，將他們判定為與B4具前后關系。（所以Lamport timestamps并不能嚴格的表示并行關系）

Lamport timestamps與偏序關系

Lamport timestamps只保證因果關系（偏序）的正確性，不保證絕對時序的正確性。

Lamport logical clock

由于Lamport timestamps只能得到偏序關系，如果要得到全序關系，就需要給Ci(a) = Cj(b)的事件定一個先后順序。

total order的事件關系=>定義如下：

如果事件a發生在進程Pi，事件b發生在進程Pj，那么當滿足下列兩者條件之一時，a=>b

• Ci(a) < Cj(b)
• Ci(a) = Cj(b) 且 Pi < Pj

根據以上條件，對于任意的兩個事件，都能判斷出它們之間的關系，因此是total ordering的。

當Lamport timestamp一致時，通過義Pi < Pj來定義順序，確保分布式場景下各個進程間發生的事件的全序定義。至于Pj < Pj：可采用不同的方式，Lamport Logical Clock提到的 arbitrary total ordering。

vector clock

Lamport timestamp得到的是全序關系，但無法嚴格表示對于沒有因果關系、存在同時發生關系(concurrent)的事件。

Vector clock是在Lamport timestamp基礎上改進的一種邏輯時鐘方法，它構不但記錄本節點的Lamport timestamp，同時也記錄了其他節點的Lamport timestamp。

原理如下:

• 本地vector clock的clock數組中每一個邏輯時間(clock)對應一個進程的clock
• 初始化vector clock中每一個邏輯時間為0；
• 每一次處理內完內部事件，將vector clock中自己的邏輯時間戳+1；
• 每發送一個消息的時候，將vector clock中自己的邏輯時間+1，且將其和消息一起發送出去
• 每接收到一個消息的時候，需要將本地的vector clock中自己的邏輯時間戳+1，且將自己vector clock中的邏輯時間和消息中攜帶的進行比較，取最大的更新本地vector clock中的邏輯時間。

圖來源于wikipedia

vector clock判定并發關系：

• 事件i、事件j對應的vector clock中，每一個進程Pk的邏輯時間戳都滿足Vi[Pk]
• vector clock中，存在P1、P2，使得Vi[P1]Vj[P2]，我們稱事件i和事件j是并發關系(沒有因果關系)；

和之前lamport timestamp的一樣，以B4事件為基準(vector clock為[A:2,B:4,C:1])，根據vector clock的判定，可以判斷出

• 灰色區域的事件happens before B4事件，B4事件happens before紅色區域的事件
• 白色區域與B4事件沒有因果關系。

特性：

• vector clock不需要在節點之間同步時鐘，不需要在所有節點上維護一段數據的版本數；
• 缺點是時鐘值的大小隨著節點增多和時間不斷增長

version vector

分布式系統多個副本被同時更新時，會導致副本之間數據的不一致。version vector用于來發現這些不一致的沖突。

version vector只能發現沖突，無法解決沖突；當然也可以通過再添加一個維度信息timestamp，發生沖突時進行比較，但是又回到了物理時鐘不同步的問題。

下圖展示了數據由不同副本處理后導致的不同版本沖突。

D5時發現了數據的沖突，這時會將不同版本數據都存儲下來，一般由客戶端來解決沖突。

version vector與vector clock的差異

• vector clocks 使用 receive和send 方法來更新clock,而version vector使用sync方法來更新。
• vector clocks是給事件定序的，確定事件的因果關系；而version vector是確定同一個數據不同版本的因果關系。

分布式與時鐘

分布式系統中，每個節點的物理時鐘是不同步的，都有一定的差異。

這樣就帶來了一些分布式系統實現的難題，如基于MVCC實現的事務，基于MVCC實現事務會要求版本之間能判斷先后順序，只有確定先后才知道應該用哪一個版本的數據，確定先后順序就涉及到時間，而不同機器之間的本地時鐘是無法保證一致的，所以這就需要確保時鐘的同步。

而通常解決方案有兩種：

• 中心化的時鐘方案，如Timestamp oracle(TSO)
• 無中心化的時鐘方案，如google True Time，Hybrid Logic Time

Timestamp oracle

如果我們整個系統不復雜，而且沒有跨全球的需求，這時用一臺中心授時服務就可以了。

如TiDB使用的就是TSO方案，tipb作為一個TSO集群，來提供授時服務。

使用TSO的好處在于因為只有一個中心授時，所以我們一定能確定所有時間的時間，但TSO需要關注幾個問題：

• 網絡延時：因為所有的事件都需要從TSO獲取時間，所以TSO只適合小集群部署，不能是那種全球級別的數據庫
• 性能：每個事件都需要從TSO獲取時間，所以TSO需要非常高的性能
• 容錯：TSO是一個單點，需要考慮節點的failover

True Time

由于節點間NTP是有偏差的，且可能出現時間回退的情況，所以NTP無法準確的判定事件的全序關系。在Google Spanner里面，通過引入True Time來解決了分布式時間問題。

True Time實現

Spanner通過使用GPS + 原子鐘atomic clock來對集群的機器時間進行校對，保證了集群機器的時間戳差距不會超過一個上限值(ε)。

用兩種技術來處理，是因為導致這兩種技術的失敗的原因是不同的。

• GPS會有一個天線，電波干擾會導致其失靈。原子鐘很穩定。
• 當GPS失靈的時候，原子鐘仍然能保證在相當長的時間內，不會出現偏差。

API

• TT.now() : 返回一個當前時間，其位于范圍區間[earliest,latest]
• TT.after(t) : 當前時間是否在t之后
• TT.before(t) : 當前時間是否在t之前

雖然spanner引入了TrueTime可以得到全球范圍的時序一致性，但由于TrueTime返回的時間仍然有一定的偏差，如果要給兩個事件定序，就需要等待2個偏差的時間間隔，來確保其先后順序。

• 事件a：[Tai, Taj], Taj-Tai=ε
• 事件b：[Tbi, Tbj], Tbj-Tbi=ε
• 所以要確定b>a, 那么就要確保Tbi > Taj, 就需要在事件b進行等待，以確保：事件b時間 - 事件a時間 > 2ε

Hybrid logical clock

HLC

Logical Physical Clocks and Consistent Snapshots in Globally Distributed Databases

TrueTime 需要硬件的支持，所以有一定的成本，而HLC無需硬件支持也能解決分布式下時間問題。

HLC同時使用了物理時鐘和邏輯時鐘（physical clock + logical clock），能夠保證單點的時間發生器是單調遞增的，同時能夠盡量控制不同節點之間的時鐘偏差在規定的偏差范圍內。

判斷兩個事件的先后順序：先判斷物理時間，再判斷邏輯時間。

HLC的算法

l.j維護的是節點j當前已知的最大的物理時間(wall time)，c.j則是當前的邏輯時間。

// 在節點j上面：初始化: l.j = 0，c.j = 0。
initially l.j :=0; c.j := 0


// 本地事件或者發送消息時，
// 如果本地時鐘pt大于當前的混合邏輯時鐘的l，
// 則將l更新成本地時鐘，將c清零。
// 否則，l保持不變，將c加1。
Send or local event
{
    l'.j := l.j;
    l.j := max(l'.j, pt.j);    // 本地物理時間pt
    if (l.j = l'.j) then 
        c.j := c.j+1
    else 
        c.j := 0;
    Timestamp with l.j, c.j
}


// 收到消息時
// l在 當前的邏輯時鐘的l、機器的本地時鐘pt、收到消息里面帶的l，三者中取最大的。
// 如果l部分是更新為本地時鐘了，則將c清零。否則，c取較大的那個l對應到的c加1。
Receive event of message m
{
    l'.j := l.j;
    l.j := max(l'.j, l.m, pt.j);
    if (l.j = l'.j = l.m) then 
        c.j := max(c.j, c.m) + 1
    elseif (l.j=l'.j) then 
        c.j := c.j + 1
    elseif (l.j=l.m) then 
        c.j := c.m + 1
    else 
        c.j := 0
    Timestamp with l.j, c.j
}

特性

HLC算法保證了HLC時間有如下特性：

• 事件e發生在事件f之前，那么事件e的HLC時間一定小于事件f的HLC時間：(l.e, c.e) < (l.f, c.f)
• 本地WallTime大于等于本地物理時間(l.e ≥ pt.e)：HLC時間總是不斷遞增，不會隨著物理時間發生回退。
• 對事件e，l.e是事件e能感知的到的最大物理時間值：如果l.e > pt.e，那么一定存在著一個發生在e之前的事件g，有pt.g=l.e。簡單來說是如果出現l.e > pt.e肯定是因為有一個HLC時間更大的的節點把當前節點的HLC時間往后推了。
• WallTime和物理時鐘的偏差是有界的(ε ≥ |pt.e - l.e| )：因為節點之間通過NTP服務校時，那么節點之間的物理時鐘偏差一定小于某個值ε。那么對于任一事件b和e，如果b hb e，那么事件b的物理時間pt.b一定滿足pt.e + ε ≥ pt.b。結合特性3存在一個事件g滿足，l.e = pt.g。那么 pt.e + ε ≥ l.e=pt.g > pt.e。

開源實現

CockroachDB采用基于NTP時鐘同步的HLC去中心化方案。

時鐘同步

所有節點間的RPC消息都會把時間戳帶入到消息中，接收到消息的節點會通過消息中的時間戳更新自己的時間, 從而達到節點間時間同步的效果。

代碼分析

HLC定義

// Timestamp represents a state of the hybrid logical clock.
type Timestamp struct {
    // Holds a wall time, typically a unix epoch time
    // expressed in nanoseconds.
    WallTime int64 `protobuf:"varint,1,opt,name=wall_time,json=wallTime" json:"wall_time"`
    // The logical component captures causality for events whose wall
    // times are equal. It is effectively bounded by (maximum clock
    // skew)/(minimal ns between events) and nearly impossible to
    // overflow.
    Logical int32 `protobuf:"varint,2,opt,name=logical" json:"logical"`
}

• WallTime：本地已知物理時鐘
• Logical：邏輯時鐘
• Timestamp：HLC，單調遞增

獲取物理時鐘

// PhysicalNow returns the local wall time. It corresponds to the physicalClock
// provided at instantiation. For a timestamp value, use Now() instead.
func (c *Clock) PhysicalNow() int64 {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.getPhysicalClockLocked()
}


// getPhysicalClockLocked returns the current physical clock and checks for
// time jumps.
func (c *Clock) getPhysicalClockLocked() int64 {
    // physicalClock 就是 UnixNano
    newTime := c.physicalClock()


    if c.mu.lastPhysicalTime != 0 {
        interval := c.mu.lastPhysicalTime - newTime
        // 檢查時鐘是否回退
        if interval > int64(c.maxOffset/10) {
            c.mu.monotonicityErrorsCount++
            log.Warningf(context.TODO(), "backward time jump detected (%f seconds)", float64(-interval)/1e9)
        }
    }


    c.mu.lastPhysicalTime = newTime
    return newTime
}




// UnixNano returns the local machine's physical nanosecond
// unix epoch timestamp as a convenience to create a HLC via
// c := hlc.NewClock(hlc.UnixNano, ...).
func UnixNano() int64 {
    return timeutil.Now().UnixNano()
}

獲取當前HLC時鐘

// Now returns a timestamp associated with an event from
// the local machine that may be sent to other members
// of the distributed network. This is the counterpart
// of Update, which is passed a timestamp received from
// another member of the distributed network.
func (c *Clock) Now() Timestamp {
    c.mu.Lock()
    defer c.mu.Unlock()
    if physicalClock := c.getPhysicalClockLocked(); c.mu.timestamp.WallTime >= physicalClock {
        // The wall time is ahead, so the logical clock ticks.
        c.mu.timestamp.Logical++
    } else {
        // Use the physical clock, and reset the logical one.
        c.mu.timestamp.WallTime = physicalClock
        c.mu.timestamp.Logical = 0
    }
    return c.mu.timestamp
}

• 如果當前物理時鐘小于WallTime，則將邏輯時鐘+1
• 如果當前物理時鐘大于WallTime，則更新WallTime為當前物理時鐘，且將邏輯時鐘設置為0

節點時鐘同步

節點之間通過在RPC請求中攜帶HLC時間來進行時鐘同步。

// sendSingleRange gathers and rearranges the replicas, and makes an RPC call.
func (ds *DistSender) sendSingleRange(
    ctx context.Context, ba roachpb.BatchRequest, desc *roachpb.RangeDescriptor,
) (*roachpb.BatchResponse, *roachpb.Error) {
    ......

    br, err := ds.sendRPC(ctx, desc.RangeID, replicas, ba)
    if err != nil {
        log.ErrEvent(ctx, err.Error())
        return nil, roachpb.NewError(err)
    }


    // If the reply contains a timestamp, update the local HLC with it.
    if br.Error != nil && br.Error.Now != (hlc.Timestamp{}) {
        ds.clock.Update(br.Error.Now)
    } else if br.Now != (hlc.Timestamp{}) {
        ds.clock.Update(br.Now)
    }

    ......
}




// Update takes a hybrid timestamp, usually originating from
// an event received from another member of a distributed
// system. The clock is updated and the hybrid timestamp
// associated to the receipt of the event returned.
// An error may only occur if offset checking is active and
// the remote timestamp was rejected due to clock offset,
// in which case the timestamp of the clock will not have been
// altered.
// To timestamp events of local origin, use Now instead.
func (c *Clock) Update(rt Timestamp) Timestamp {
    c.mu.Lock()
    defer c.mu.Unlock()

    // 如果本地物理時間pt
    physicalClock := c.getPhysicalClockLocked()


    // 大于本地WallTime且大于rt.WallTime：
    //      更新本地WallTime=pt，且logical=0
    if physicalClock > c.mu.timestamp.WallTime && physicalClock > rt.WallTime {
        // Our physical clock is ahead of both wall times. It is used
        // as the new wall time and the logical clock is reset.
        c.mu.timestamp.WallTime = physicalClock
        c.mu.timestamp.Logical = 0
        return c.mu.timestamp
    }


    // In the remaining cases, our physical clock plays no role
    // as it is behind the local or remote wall times. Instead,
    // the logical clock comes into play.

    // 如果rt.WallTime > 本地WallTime：
    //   檢查rt.WallTime與pt是否大于時鐘偏差；
    //   本地WallTime=rt.WallTime，logical++
    if rt.WallTime > c.mu.timestamp.WallTime {
        offset := time.Duration(rt.WallTime-physicalClock) * time.Nanosecond
        if c.maxOffset > 0 && offset > c.maxOffset {
            log.Warningf(context.TODO(), "remote wall time is too far ahead (%s) to be trustworthy - updating anyway", offset)
        }
        // The remote clock is ahead of ours, and we update
        // our own logical clock with theirs.
        c.mu.timestamp.WallTime = rt.WallTime
        c.mu.timestamp.Logical = rt.Logical + 1
    } else if c.mu.timestamp.WallTime > rt.WallTime {
       // 如果本地WallTime>rt.WallTime：logical++
        // Our wall time is larger, so it remains but we tick
        // the logical clock.
        c.mu.timestamp.Logical++
    } else {
        // Both wall times are equal, and the larger logical
        // clock is used for the update.
        if rt.Logical > c.mu.timestamp.Logical {
            c.mu.timestamp.Logical = rt.Logical
        }
        c.mu.timestamp.Logical++
    }
    return c.mu.timestamp
}