channel数据结构

诺唯 · · 4966 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

channel是go语言的一大特色,使用原子函数还是使用互斥锁都不如使用channel来的简单,go语言中的channel可以作为函数参数传递和返回值返回,通过发送和接受数据在goroutine之间同步(在学习和使用go语言的时候,我们应该牢记,go语言中所有的结构都是值拷贝的)

本文不对channel使用作讲解,直接上酸(dai)菜(ma):

type hchan struct {
	qcount   uint           //队列数据总的数据数量
	dataqsiz uint           //环形队列的数据大小 
	buf      unsafe.Pointer //指向dataqsiz元素类型大小的数组
	elemsize uint16
	closed   uint32
	elemtype *_type // 元素类型
	sendx    uint   // 发送数据时的游标
	recvx    uint   // 接收数据时的游标
	recvq    waitq  // 接收而阻塞的等待队列
	sendq    waitq  // 发送而阻塞的等待队列
        lock mutex      // 保护hchan所有字段的锁
}
hchan 是chan的结构体,在hchan结构中qcount和elemsize指定队列的容量和使用量,dataqsiz队列的大小,整个hchan结构体只记录了队列大小相关的值,带有缓冲区的chan需要make的时候指定,我们简单的看一下chan的make方法是如何分配缓冲区的
func makechan(t *chantype, size int64) *hchan {
	elem := t.elem
	
        ...
        ...

	var c *hchan
	if elem.kind&kindNoPointers != 0 || size == 0 {
		c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
		if size > 0 && elem.size != 0 {
			c.buf = add(unsafe.Pointer(c), hchanSize)
		} else {
			c.buf = unsafe.Pointer(c)
		}
	} else {
		c = new(hchan)
		c.buf = newarray(elem, int(size))
	}
	...
        ...
}

makechan 将hchan初始化0值之后并判断size如果是有缓冲区的chan则紧挨着hchan结构体中分配size大小的 “_type” 类型的数组。

type waitq struct {
	first *sudog
	last  *sudog
}
type sudog struct {
	g          *g
	selectdone *uint32 
	next       *sudog
	prev       *sudog
	elem       unsafe.Pointer 

	acquiretime int64
	releasetime int64
	ticket      uint32
	waitlink    *sudog // g.waiting list
	c           *hchan // channel
}

g和elem分别存储goroutine的数据

发送channel

向channel中写数据时在runtime包中对应的是,以下方法:

func chansend(t *chantype, c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
	if raceenabled {
		raceReadObjectPC(t.elem, ep, callerpc, funcPC(chansend))
	}
	if msanenabled {
		msanread(ep, t.elem.size)
	}

	if c == nil {
		if !block {
			return false
		}
		gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
		throw("unreachable")
	}

	if debugChan {
		print("chansend: chan=", c, "\n")
	}

	if raceenabled {
		racereadpc(unsafe.Pointer(c), callerpc, funcPC(chansend))
	}

	
	if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
		(c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
		return false
	}

	var t0 int64
	if blockprofilerate > 0 {
		t0 = cputicks()
	}

	lock(&c.lock)

	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}

	if sg := c.recvq.dequeue(); sg != nil {
		send(c, sg, ep, func() { unlock(&c.lock) })
		return true
	}

	if c.qcount < c.dataqsiz {
		qp := chanbuf(c, c.sendx)
		if raceenabled {
			raceacquire(qp)
			racerelease(qp)
		}
		typedmemmove(c.elemtype, qp, ep)
		c.sendx++
		if c.sendx == c.dataqsiz {
			c.sendx = 0
		}
		c.qcount++
		unlock(&c.lock)
		return true
	}

	if !block {
		unlock(&c.lock)
		return false
	}

	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	
	mysg.elem = ep
	mysg.waitlink = nil
	mysg.g = gp
	mysg.selectdone = nil
	mysg.c = c
	gp.waiting = mysg
	gp.param = nil
	c.sendq.enqueue(mysg)
	goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)

	
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	if gp.param == nil {
		if c.closed == 0 {
			throw("chansend: spurious wakeup")
		}
		panic(plainError("send on closed channel"))
	}
	gp.param = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	mysg.c = nil
	releaseSudog(mysg)
	return true
}

发送数据时先判断channel类型,如果有缓冲区,判断channel是否还有空间,然后从等待channel中获取等待channel中的接受者,如果取到接收者,则将对象直接传递给接受者,然后将接受者所在的go放入P所在的可运行G队列,发送过程完成,如果未取到接收者,则将发送者enqueue到发送channel,发送者进入阻塞状态,有缓冲的channel需要先判断channel缓冲是否还有空间,如果缓冲空间已满,则将发送者enqueue到发送channel,发送者进入阻塞状态如果缓冲空间未满,则将元素copy到缓冲中,这时发送者就不会进入阻塞状态,最后尝试唤醒等待队列中的一个接受者。



接收channel

向channel中接收数据时在runtime包中对应的是,以下方法:

func chanrecv(t *chantype, c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	// raceenabled: don't need to check ep, as it is always on the stack
	// or is new memory allocated by reflect.

	if debugChan {
		print("chanrecv: chan=", c, "\n")
	}

	if c == nil {
		if !block {
			return
		}
		gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
		throw("unreachable")
	}

	// Fast path: check for failed non-blocking operation without acquiring the lock.
	//
	// After observing that the channel is not ready for receiving, we observe that the
	// channel is not closed. Each of these observations is a single word-sized read
	// (first c.sendq.first or c.qcount, and second c.closed).
	// Because a channel cannot be reopened, the later observation of the channel
	// being not closed implies that it was also not closed at the moment of the
	// first observation. We behave as if we observed the channel at that moment
	// and report that the receive cannot proceed.
	//
	// The order of operations is important here: reversing the operations can lead to
	// incorrect behavior when racing with a close.
	if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||
		c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&
		atomic.Load(&c.closed) == 0 {
		return
	}

	var t0 int64
	if blockprofilerate > 0 {
		t0 = cputicks()
	}

	lock(&c.lock)

	if c.closed != 0 && c.qcount == 0 {
		if raceenabled {
			raceacquire(unsafe.Pointer(c))
		}
		unlock(&c.lock)
		if ep != nil {
			typedmemclr(c.elemtype, ep)
		}
		return true, false
	}

	if sg := c.sendq.dequeue(); sg != nil {
		// Found a waiting sender. If buffer is size 0, receive value
		// directly from sender. Otherwise, receive from head of queue
		// and add sender's value to the tail of the queue (both map to
		// the same buffer slot because the queue is full).
		recv(c, sg, ep, func() { unlock(&c.lock) })
		return true, true
	}

	if c.qcount > 0 {
		// Receive directly from queue
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			raceacquire(qp)
			racerelease(qp)
		}
		if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
		typedmemclr(c.elemtype, qp)
		c.recvx++
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.qcount--
		unlock(&c.lock)
		return true, true
	}

	if !block {
		unlock(&c.lock)
		return false, false
	}

	// no sender available: block on this channel.
	gp := getg()
	mysg := acquireSudog()
	mysg.releasetime = 0
	if t0 != 0 {
		mysg.releasetime = -1
	}
	// No stack splits between assigning elem and enqueuing mysg
	// on gp.waiting where copystack can find it.
	mysg.elem = ep
	mysg.waitlink = nil
	gp.waiting = mysg
	mysg.g = gp
	mysg.selectdone = nil
	mysg.c = c
	gp.param = nil
	c.recvq.enqueue(mysg)
	goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)

	// someone woke us up
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	closed := gp.param == nil
	gp.param = nil
	mysg.c = nil
	releaseSudog(mysg)
	return true, !closed
}

接收channel与发送类似首先也是判断channel的类型,然后如果是有缓冲的channel就判断缓冲中是否有元素,接着从channel中获取接受者,如果取到,则直接从接收者获取元素,并唤醒发送者,本次接收过程完成,如果没有取到接收者,阻塞当前的goroutine并等待发送者唤醒,如果是拥有缓冲的channel需要先判断缓冲中是否有元素,缓冲为空时,阻塞当前goroutine并等待发送者唤醒,缓冲如果不为空,则取出缓冲中的第一个元素,然后尝试唤醒channel中的一个发送者(这篇文章暂属临时版本,有些话需要斟酌,不久会更新。。。)

接下来我会发表select的结构先说个预告。。。

select {
       case c <- v:
              ... foo
       default:
              ... bar
}

//select的case和default 编译器最终会编译成if else

if selectnbsend(c, v) {
       ... foo
} else {
       ... bar
}

未来几天我会完成select的具体实现。。。


有疑问加站长微信联系(非本文作者)

本文来自:知乎专栏

感谢作者:诺唯

查看原文:channel数据结构

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

4966 次点击  
加入收藏 微博
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传