mgo中的atomic.Value

閑話

最近在刷MIT的分布式課程,線程和鎖一定是最基本的元素啦.
由於GO本身提倡的 Share memory by communicating; don't communicate by sharing memory.,所以在實現的時候試圖不用sharing memory + lock而多用channel來實現,這時候就帶來了一些小小的不方便,比如一個字符串s可以被多個goroutine讀而被一個goroutine寫,顯然我們要將其加鎖.
因為不想加鎖於是我試圖用atomic來解決這個問題,其中有一個Value類型我發現很有趣,於是決定寫下來.

源代碼分析

• atomic.Value分為兩個操作,通過Store()存儲Value,通過Load()來讀取Value的值.

• 源碼我就不貼了,貼一個關鍵的struct:

  type ifaceWords struct {
  	typ  unsafe.Pointer
  	data unsafe.Pointer
  }
  

ifaceWords結構體是實際存儲我們的Value值的地方,可以看到,我們存儲的實際是指向Value的type和data的指針.

• Store操作有兩種行為模式:

  • First Store : 當我們第一次調用Store的時候,Store函數會初始化typ指針(需要注意的是,每一個Value在第一次Store之后typ就被確定而不能更改了,否則會panic).
    如果typ==nil則函數會調用runtime_procPin(沒找到實現,但注釋中說是active spin wait)
    隨后調用原子操作函數CompareAndSwapPointer(typ, nil, unsafe.Pointer(^uintptr(0))),如果此時結果返回false說明typ已經不等於nil(被其他goroutine修改過),於是調用runtime_procUnpin解鎖並重新進行Store過程.
    如果原子操作函數返回了true,即typ == nil,那么存儲typ以及data的指針.

  • 后面的每次Store調用都是直接替換掉data指針

• Load函數檢測typ的值,如果為nil或者正在進行首次調用Store則會返回nil.否則返回一個interface{}(實際存儲的是ifaceWords值)

用例

• 在MIT的課程中,我們設計的每一個Server都需要訪問viewService來獲得最新的Primary/Backup數據庫服務器視圖.
  這時候其實對這個View的操作就是周期寫,不定時讀.這時候就是一個使用Value的合適場景(也可以使用原子函數CompareAndSwapPointer).

• Go官網上給出的例子

  type Map map[string]string
  var m Value
  m.Store(make(Map))
  var mu sync.Mutex // used only by writers
  // read function can be used to read the data without further synchronization
  read := func(key string) (val string) {
          m1 := m.Load().(Map)
          return m1[key]
  }
  // insert function can be used to update the data without further synchronization
  insert := func(key, val string) {
          mu.Lock() // synchronize with other potential writers
          defer mu.Unlock()
          m1 := m.Load().(Map) // load current value of the data structure
          m2 := make(Map)      // create a new value
          for k, v := range m1 {
                  m2[k] = v // copy all data from the current object to the new one
          }
          m2[key] = val // do the update that we need
          m.Store(m2)   // atomically replace the current object with the new one
          // At this point all new readers start working with the new version.
          // The old version will be garbage collected once the existing readers
          // (if any) are done with it.
  }