Author: meow uncle: the original blog. Betacat. IO/post/golang…

In the Go language standard library, the Sync/Atomic package encapsulates the atomic operations provided by the underlying hardware as Go functions. But these operations only support a few basic data types, so to expand the scope of atomic operations, Go added a new type Value to the Sync/Atomic package in version 1.4. A value of this type acts as a container that can be used atomically to Store and Load any type of value.

Historical origin

I found this discussion in 2014 on the Golang-dev mailing list, and some users reported performance issues with encoding/ GOB packages on multi-core machines (80-core), Think encoding/characteristics of the gob are not fully utilized to multicore because it use a lot of the mutex (mutex), if change the mutex to use atomic LoadPointer/StorePointer for concurrency control, the performance will be able to ascend 20 times.

To address this issue, it has been proposed to encapsulate an atomic.Value type from the existing atomic package so that users can use atomic operations provided by atomic without relying on the Go internal type unsafe.pointer. So all of the atomic packages we see today, except for atomic.Value, were written in assembly earlier, and the underlying implementation of the atomic.Value type is based on existing atomic packages.

So why would atomic perform so much better than Mutex in the above scenario? Author Dmitry Vyukov summarizes a difference between the two:

Mutexes do no scale. Atomic loads do.

Mutex is implemented by the operating system, while atomic operations in the atomic package are directly supported by the underlying hardware. In the CPU implementation instruction set, there are some instructions encapsulated in atomic package, these instructions are not allowed to interrupt during execution, so atomic operation can guarantee concurrency safety in lock-free condition, and its performance can be linearly expanded with the increase of the number of cpus.

Well, having said so much about atomic operations, let’s first look at what kind of operations can be called atomic operations.

atomic

The ability of one or more operations to remain uninterrupted while the CPU is executing is called atomicity. These operations appear to the outside world as an integral whole, and they are either all performed or none performed. The outside world does not see that they are only half-performed. In the real world, cpus cannot perform a sequence of operations without interruption, but if we can make their intermediate states invisible while performing multiple operations, we can claim that they have “indivisible” atomicity.

For those of you who don’t know, a normal assignment statement in Go (or even most languages) is not an atomic operation. For example, writing a variable of type INT64 on a 32-bit machine will have an intermediate state because it will be split into two write operations (MOV) — write the lower 32 bits and write the higher 32 bits, as shown below:

If one thread writes the lower 32 bits and another thread reads the variable before it can write the higher 32 bits, it gets an intermediate variable that doesn’t make any sense, which can cause weird bugs in our program.

This is just a basic type, and if we assign a value to a structure, it’s more likely to have concurrency problems. It is possible that the writer thread reads the variable as soon as it has written only half of the field, and that only part of the value is changed. This obviously breaks the integrity of the variable and the value read is completely wrong.

In this case, the main character, atomic.Value, allows us to write and write variables in multiple threads without having to rely on the uncompatibility unsafe.Pointer type, while encapsulating reads and writes of arbitrary data types into atomic operations (making intermediate states invisible).

Use the pose

The atomic.Value type is exposed to just two methods:

  • v.Store(c)– Write operation will be the original variablecStore it in aatomic.ValueThe type ofvIn the water.
  • c = v.Load()– Read operations from thread-safevRead the contents stored in the previous step.

The compact interface makes it easy to use, simply by substituting Load() and Store() for variable reads and assignments that require concurrent protection.

The following is a common usage scenario: the application periodically retrieves the latest configuration information from the outside world and then changes the configuration variables maintained in its own memory. The worker thread processes the request based on the latest configuration.

package main

import (
  "sync/atomic"
  "time"
)

func loadConfig(a) map[string]string {
  // Read configuration information from the database or file system and store it in memory as a map
  return make(map[string]string)}func requests(a) chan int {
  // Put requests received from the outside world into a channel
  return make(chan int)}func main(a) {
  The config variable is used to store the configuration information of the service
  var config atomic.Value
  // Load the config file from somewhere else during initialization and store it in the config variable
  config.Store(loadConfig())
  go func(a) {
    // Pull the latest configuration information every 10 seconds and update it to the config variable
    for {
      time.Sleep(10 * time.Second)
      // corresponding to assignment operation config = loadConfig()
      config.Store(loadConfig())
    }
  }()
  // Create worker threads. Each worker thread processes requests based on the latest configuration information it reads
  for i := 0; i < 10; i++ {
    go func(a) {
      for r := range requests() {
        // Corresponds to the value operation c := config
        // Since Load() returns an interface{} type, we need to cast it
        c := config.Load().(map[string]string)
        // Here is the logic for processing requests based on configuration information...
        _, _ = r, c
      }
    }()
  }
}
Copy the code

Internal implementation

Luo Yonghao hao once said:

Simplicity is the hidden complexity

Let’s take a look at the hidden complexity behind the simple facade.

The data structure

Atomic.Value is designed to store any type of data, so its internal field is an interface{} type, which is pretty straightforward.

type Value struct {
  v interface{}}Copy the code

In addition to Value, this file defines an ifaceWords type, which is an internal representation of an empty interface (interface{}) (see runtime/runtime2.go for the definition of eface). It splits the interface{} type into two fields.

type ifaceWords struct {
  typ  unsafe.Pointer
  data unsafe.Pointer
}
Copy the code

Write (Store) operation

Before we get to writing, let’s take a look at the safe.pointer type inside the Go language.

unsafe.Pointer

The Go language doesn’t support direct memory manipulation for security reasons, but the library offers an unsafe Pointer type, unsafe.Pointer, that allows applications to manipulate memory flexibly.

Unsafe.Pointer is special because it can be converted to and from any Pointer type, bypassing the Go language’s type system. In other words, if two types have the same layout, we can use broadening.Pointer as a bridge between the two types to interpret the same memory in two different ways.

For example, []byte and String actually have the same internal storage structure, but Go’s type system forbids them from interchanging. Unbroadening.Pointer allows us to convert the []byte array directly to string without copying anything.

bytes := []byte{104.101.108.108.111}

p := unsafe.Pointer(&bytes) // Cast to unsafe.Pointer, the compiler will not report an error
str := *(*string)(p) // Cast to a pointer to string and fetch the value as string
fmt.Println(str) / / output "hello"
Copy the code

Given what unsafe.Pointer does, we can jump right into the code:

func (v *Value) Store(x interface{}) {
  if x == nil {
    panic("sync/atomic: store of nil value into Value")
  }
  vp := (*ifaceWords)(unsafe.Pointer(v))  // Old value
  xp := (*ifaceWords)(unsafe.Pointer(&x)) // New value
  for {
    typ := LoadPointer(&vp.typ)
    if typ == nil {
      // Attempt to start first store.
      // Disable preemption so that other goroutines can use
      // active spin wait to wait for completion; and so that
      // GC does not see the fake type accidentally.
      runtime_procPin()
      if! CompareAndSwapPointer(&vp.typ,nil, unsafe.Pointer(^uintptr(0))) {
        runtime_procUnpin()
        continue
      }
      // Complete first store.
      StorePointer(&vp.data, xp.data)
      StorePointer(&vp.typ, xp.typ)
      runtime_procUnpin()
      return
    }
    if uintptr(typ) == ^uintptr(0) {
      // First store in progress. Wait.
      // Since we disable preemption around the first store,
      // we can wait with active spinning.
      continue
    }
    // First store completed. Check type and overwrite data.
    iftyp ! = xp.typ {panic("sync/atomic: store of inconsistently typed value into Value")
    }
    StorePointer(&vp.data, xp.data)
    return}}Copy the code

General logic:

  • Lines 5 to 6 – Passunsafe.PointerwillThe existingandWant to writeThe values are respectively converted toifaceWordsType, so we can get these two nextinterface{}The primitive type (TYP) and the real value (data) of.
  • An infinite for loop begins at line 7. Cooperate withCompareAndSwapEdible, can achieve the effect of optimistic lock.
  • Line 8, we can passLoadPointerThe atomic operation is brought to the presentValueThe type of the storage in. The following three cases are handled according to the different types.
  1. First write (lines 9 to 24) – oneValueAfter the instance is initialized, itstypThe field is set to nil for the pointer, so line 9 determines iftypNil so let’s prove thisValueNo data has been written to it yet. That is followed by an initial write:
    • runtime_procPin()This is a function in The Runtime, the specific function is not very clear to me, and I have not found relevant documents. On the one hand, it prevents the scheduler from preempting the current goroutine, allowing it to execute the current logic without being interrupted, so that it can complete its work as quickly as possible because someone else is waiting for it. On the other hand, the GC thread cannot be enabled during preemption, which prevents the GC thread from seeing an unexplained pointer^uintptr(0)This is the intermediate state in the assignment process.
    • useCASOperation, first try willtypSet to^uintptr(0)This intermediate state. If that fails, proving that another thread has completed the assignment first, it unlocks the preemption lock and returns to the first step of the for loop.
    • If this is set successfully, it proves that the current thread has grabbed the “optimistic lock” and it can be safely lockedvSet to the new value passed in (lines 19-23). Notice, I’m writing it firstdataField, and then writetypField. Because we are based ontypThe value of the field is used to determine whether the write is complete.
  2. The first write is not complete (lines 25-30) – if you seetypField or^uintptr(0)This intermediate type, it proves that the first write that we just did hasn’t finished yet, so it will continue to loop, “busy” until the first write is done.
  3. First write completed (line 31 and after) – First check to see if the type of the last write is the same as the type of the next write, and throw an exception if it is not. Otherwise, the value to be written directly todataField.

The main idea of this logic is that in order to write atomicity to multiple fields, we can grab one of the fields and use its state to indicate the state of the entire atomic write. This is an idea I have seen similar in TiDB’s transaction implementation, which is called the Percolator model. The main idea is to first select a primaryRow and then all operations are marked by the success of primaryRow. Well, there’s nothing new under the sun.

If you don’t have the patience to look at the code, here’s a simplified version of the flowchart:

Load operation

Code first:

func (v *Value) Load(a) (x interface{}) {
  vp := (*ifaceWords)(unsafe.Pointer(v))
  typ := LoadPointer(&vp.typ)
  if typ == nil || uintptr(typ) == ^uintptr(0) {
    // First store not yet completed.
    return nil
  }
  data := LoadPointer(&vp.data)
  xp := (*ifaceWords)(unsafe.Pointer(&x))
  xp.typ = typ
  xp.data = data
  return
}
Copy the code

Reading is much simpler and has two branches:

  1. If the currenttypIs nil or^uintptr(0)That proves the first write hasn’t started or hasn’t finished yet, and returns nil (no intermediate state exposed).
  2. Otherwise, based on what you currently seetypanddataConstruct a new oneinterface{}Go back out.

conclusion

This article starts with a discussion on the mailing list and explains the historical reasons for the presence of atomic.value. Then it introduces its use posture and internal implementation from shallow to deep. So that you know not only what it is, but why it is.

Also, again, atomic operations are supported by the underlying hardware, while locks are implemented by apis provided by the operating system. The former are often more efficient when they perform the same functions, and take advantage of multiple computer cores. Therefore, in the future, when we want to update some variables concurrently and safely, we should prefer to use atomic.Value.