使用atomic来避免lock

在程序中为了互斥,难免要用锁,有些时候可以通过使用atomic来避免锁,
从而更高效.

下面给出一个以太坊中的例子,就是MsgPipeRW,从名字Pipe可以看出,
他实际上就是一个pipe,相比大家对pipe已经比较熟悉了,我就不多解释了.

type MsgPipeRW struct {
	w       chan<- Msg
	r       <-chan Msg
	closing chan struct{}
	closed  *int32
}

//创建一个MsgPipeRw
func MsgPipe() (*MsgPipeRW, *MsgPipeRW) {
	var (
		c1, c2  = make(chan Msg), make(chan Msg)
		closing = make(chan struct{})
		closed  = new(int32)
		rw1     = &MsgPipeRW{c1, c2, closing, closed}
		rw2     = &MsgPipeRW{c2, c1, closing, closed}
	)
	return rw1, rw2
}

pipe就像水管一样,这里MsgPipe创建了两根水管,可以自由双向流动,rw1写,rw2就可以
读到,rw2写,rw1就可以读到.原理也很简单,因为rw1写和rw2操作的是同一个chan Msg,反之亦然.

关键是这里的closed,可以想想rw1,rw2很有可能在不同的goroutine发生读写关闭等操作,
这时候要同时访问closed这个变量,难免会发生冲突,我们看看如何避免.

closed如果为0表示没有关闭,1表示已经关闭,就不应该再进行读写了.

// 从pipe中读取一个msg
func (p *MsgPipeRW) ReadMsg() (Msg, error) {
//这里不能直接*p.closed==0,要使用atomic.LoadInt32来访问
	if atomic.LoadInt32(p.closed) == 0 {
		...
	}
	return Msg{}, ErrPipeClosed
}

// 写的时候也一样
func (p *MsgPipeRW) WriteMsg(msg Msg) error {
	if atomic.LoadInt32(p.closed) == 0 {
		...
	}
	return ErrPipeClosed
}

读写消息只是读取互斥变量,没有发生写入,下面来看看close的时候如何写入

func (p *MsgPipeRW) Close() error {
	if atomic.AddInt32(p.closed, 1) != 1 { //避免锁,
		// someone else is already closing
		atomic.StoreInt32(p.closed, 1) // avoid overflow
		return nil
	}
	close(p.closing)
	return nil
}

atomic.AddInt32能够避免我们一般这样的写法发生的并发访问.

if *p.closed==0 {
    *p.closed+=1
}

感兴趣的可以修改代码试试,采用*p.closed==0这种方式,会不会造成崩溃,测试代码如下

func TestMsgPipeConcurrentClose(t *testing.T) {
	rw1, _ := MsgPipe()
	for i := 0; i < 10; i++ {
		go rw1.Close()
	}
}

atomic看似神奇的避免了锁,实际上这需要处理器的特殊指令支持,尤其是发生在多和处理器上时,atomic指令
会保证对特定地址的锁定.
atomic相对于lock的最大优势就是他只是一条特殊指令,不用发生系统上下文切换,我们都知道系统上下文切换
代价要大得多.

以太坊系列之四: 使用atomic来避免lock

使用atomic来避免lock

推荐阅读