Channel in Go小结

func main() {
    // Q: 有缓冲和无缓冲在使用上有什么区别？
    // ch := make(chan int) // 无缓冲
    ch := make(chan int, 1) // 有缓冲，大小为1

    // 发送
    ch <- 1

    fmt.Println(<-ch)
}

// A: 代码上从发送端控制channel的关闭，同时为了避免重复关闭，使用sync.Once来协助。
// create
ch := make(chan int, 233)
// sending
ch <- 1
// close from sender
once.Do(close(ch))

// runtime/chan.go#makechan
func makechan(t *chantype, size int) *hchan {
	elem := t.elem
    // ...

    // elem.size * size 是否溢出
	mem, overflow := math.MulUintptr(elem.size, uintptr(size))
	if overflow || mem > maxAlloc-hchanSize || size < 0 {
		panic(plainError("makechan: size out of range"))
	}

	// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
	// buf points into the same allocation, elemtype is persistent.
	// SudoG's are referenced from their owning thread so they can't be collected.
    // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
    // 大意是是说，当存储在buf的元素中不包含指针时，gc对此不感兴趣？（留坑）因此才会将buf和hchan分配在同一片内存空间。
	var c *hchan
	switch {
	case mem == 0:
		// 无缓冲（只分配hchan）
		c = (*hchan)(mallocgc(hchanSize, nil, true))
		c.buf = c.raceaddr()
	case elem.ptrdata == 0:
		// 不含指针（hchan和缓冲区在一起）
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        // buf的起始地址 = 基地址 + hchan结构体大小偏移量
		c.buf = add(unsafe.Pointer(c), hchanSize)
	default:
		// 有缓冲，含指针（hchan和缓冲区不在一起）
		c = new(hchan)
		c.buf = mallocgc(mem, elem, true)
	}

	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	c.dataqsiz = uint(size)
    // ...
	return c
}

// runtime/chan.go#closechan
func closechan(c *hchan) {
	if c == nil {
		// 如果channel为空，则会触发panic
		panic(plainError("close of nil channel"))
	}

	lock(&c.lock)
	if c.closed != 0 {
		// 关闭已经关闭的channel，也会触发panic
		unlock(&c.lock)
		panic(plainError("close of closed channel"))
	}

	// ...
	// 设置已关闭标志位
	c.closed = 1

	var glist gList

	// 释放所有的recvq中的g
	for {
		sg := c.recvq.dequeue()
		if sg == nil {
			break
		}
		// 清空数据
		if sg.elem != nil {
			typedmemclr(c.elemtype, sg.elem)
			sg.elem = nil
		}
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = nil
		// ...
		glist.push(gp)
	}

	// 释放所有的sendq中的g
	for {
		// ... 与接受者处理类似
	}
	unlock(&c.lock)

	// 让所有的读/写g就绪
	for !glist.empty() {
		gp := glist.pop()
		gp.schedlink = 0
		goready(gp, 3)
	}
}

// case 1
ch <- val

// case 2 
select {
    case ch <- 2:
        // foo
    default:
        // bar
}

// case 1
func chansend1(c *hchan, elem unsafe.Pointer) {
	chansend(c, elem, true, getcallerpc())
}

// case 2
func selectnbsend(c *hchan, elem unsafe.Pointer) (selected bool) {
	return chansend(c, elem, false, getcallerpc())
}

// case 1
val := <- ch

// case 2 
select {
    case val := <- ch:
        // foo
    default:
        // bar
}

// 此外接收，还有另外一个参数，ok表示已接收
val, ok := <- ch

// case 1
func chanrecv1(c *hchan, elem unsafe.Pointer) {
	chanrecv(c, elem, true)
}
// 多返回值
func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
	_, received = chanrecv(c, elem, true)
	return
}

// case 2
func selectnbrecv(elem unsafe.Pointer, c *hchan) (selected bool) {
	selected, _ = chanrecv(c, elem, false)
	return
}
// 多返回值
func selectnbrecv2(elem unsafe.Pointer, received *bool, c *hchan) (selected bool) {
	// TODO(khr): just return 2 values from this function, now that it is in Go.
	selected, *received = chanrecv(c, elem, false)
	return
}

func main() {
    done := make(chan int, 1)

    go func(){
        println("i'm running")
        done <- 1
    }()

    // 如果有多个goroutine, 可以对done接收计数，或者检测goroutine关闭来判断同步完成
    <-done
    println("main goroutine done")
}

func Producer(ch chan int) {
    for {
        ch <- 1
    }
}

func Consumer(ch chan int) {
    for {
        v <- ch
        println(v)
    }
}

func main() {
    ch := make(chan int, 2)
	go Producer(ch)
	go Producer(ch)
	go Consumer(ch)
	go Consumer(ch)

    select{}
}

type topic struct{
	chPub chan int
	chSub []chan int
	bufsize int
}

func newTopic(bufsize int) *topic {
	return &topic{
		bufsize: bufsize,
		chPub:   make(chan int, bufsize),
		chSub:   make([]chan int, 0, 5),
	}
}

// 这里不是并发安全的！！！
func (t *topic) Sub() chan int  {
	ch := make(chan int, t.bufsize)
	t.chSub = append(t.chSub, ch)
	return ch
}

func (t *topic) Pub(v int) {
	t.chPub <- v
	// 分发
	go func() {
		v2 := <- t.chPub
		for idx := range t.chSub {
			t.chSub[idx] <- v2
		}
	}()
}

// Q：如何实现topic的关闭方法
func main() {
	topic := newTopic(topic)
	// 开启5个订阅
	for i := 0; i < 5; i++ {
		go func() {
			ch := topic.Sub()
			for v := range ch {
				println("recv topic", v)
			}
		}()
	}
	// 发布消息
	for v := 10; v < 100; v++{
		topic.Pub(v)
		time.Sleep(100 *time.Millisecond)
	}
}

type obj int

func factory() obj {
	return "newobj"
}

type pool struct {
	ch chan obj
}

func newpool() pool {
	p := pool{
		ch: make(chan obj, 10),
	}
	for i := 0; i < 10; i++ {
		p.put(obj(i))
	} 
}

// 需要增加并发安全！！！
func (p pool) put(obj obj)  {
	p.ch <- obj
}

// 需要增加并发安全！！！
func (p pool) get() obj {
	v := <- p.ch
	return v
}

func main() {
	pool := newpool()

	go func() {
		for {
			println("num of running g:", runtime.NumGoroutine())
			time.Sleep(100 * time.Millisecond)
		}
	}()

	for i := 0; i < 100; i++ {
		obj := pool.get()
		go func() {
			defer pool.put(obj)
			// do some work
			time.Sleep(100 *time.Millisecond)			
		}()
	}
}

// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12
// num of running g: 12