func Action(ctx context.Context, db *Databus) (*Databus, error)

type Data struct {
	input 		*Input
  collector *ChekerCollector
}

type CheckerCollector struct {
  lock 			sync.Lock
	data map[string]*CheckRes
}

func (cc *CheckerCollector) Report(key string, res *CheckRes) {
  cc.lock.Lock()
  map[key] = res
  cc.lock.Unlock()
}

ChannelType = ( "chan" | "chan" "<-" | "<-" "chan" ) ElementType .

	var ch1 chan int   // 声明一个传递整型的通道
    var ch2 chan bool  // 声明一个传递布尔型的通道
    var ch3 chan []int // 声明一个传递int切片的通道

func main() {
	ch := make(chan int)
	ch <- 9527
	fmt.Println("OK")
}

func recv(c chan int) {
    ret := <-c
    fmt.Println("recv ok", ret)
}
func main() {
    ch := make(chan int)
    go recv(ch) // 启用goroutine从通道接收值
    ch <- 10
    fmt.Println("ok")
}

func main() {
    ch := make(chan int, 1) // 创建一个容量为1的有缓冲区通道
    ch <- 10
    fmt.Println("OK")
}

var ch chan int
fmt.Println(ch) // <nil>

ch4 := make(chan int)
ch5 := make(chan bool 10)
ch6 := make(chan []int 1)

ch <- 10 //将10发送到channel中

select {
    case ch <- 10:
    	xxxxx
    default:
    	xxxx
}

x, ok := <- ch //comma ok风格，判断ch是否已经关闭，x为ch元素类型零值
x := <- ch //从chan中接受值并赋值给x
<- ch //从ch中接收值并忽略，常用于消息通知

select {
	case <-ch:
		xxxx
	default:
		xxxx
}

select {
    case a := <- ch1:
		xxxx
	case ch2 <- b:
		xxx
    //default: 可选
}

close(ch)	

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

	go func() {
		for i := 0; i < 100; i++ {
			ch <- i
		}
	}()

	<-time.After(2 * time.Second)

	go func() {
		close(ch)
	}()

	<-time.After(1 * time.Second)
}

//% go run main.go
//panic: send on closed channel
//
//goroutine 4 [running]:
//main.main.func1(0x1400006e000)
//	/Users/didi/work/src/my/test/test/main.go:15 +0x40
//created by main.main
//	/Users/didi/work/src/my/test/test/main.go:13 +0x54
//exit status 2



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

	go func() {
		v := <-ch
		fmt.Println(v)
	}()

	<-time.After(2 * time.Second)

	go func() {
		close(ch)
	}()

	<-time.After(1 * time.Second)
}

//$ go run main.go
//0

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)
    // 开启goroutine将0~100的数发送到ch1中
    go func() {
        for i := 0; i < 100; i++ {
            ch1 <- i
        }
        close(ch1)
    }()
    // 开启goroutine从ch1中接收值，并将该值的平方发送到ch2中
    go func() {
        for {
            i, ok := <-ch1 // 通道关闭后再取值ok=false
            if !ok {
                break
            }
            ch2 <- i * i
        }
        close(ch2)
    }()
    // 在主goroutine中从ch2中接收值打印
    for i := range ch2 { // 通道关闭后会退出for range循环
        fmt.Println(i)
    }
}

//使用
func main() {
	ch := make(chan int)
    
    //sender
	go func() {
		for i := 0; i < 100; i++ {
			ch <- i
		}
		close(ch)
	}()

    //reciver
	go func() {
		for {
			i, ok := <-ch
			if !ok { //如果已经关闭了,则直接中断
				break
			}
			fmt.Println(i)
		}
	}()

    //reciver
	go func() {
		for {
			i, ok := <-ch
			if !ok { //如果已经关闭了,则直接中断
				break
			}
			fmt.Println(i)
		}
	}()

	select {
	case <-time.After(time.Hour):
	}
}

func main() {
	ch := make(chan int, 100)

	ctx, cancel := context.WithCancel(context.TODO())
	ctx1 := context.WithValue(ctx, struct{}{}, struct{}{})
    ctx2 := context.WithValue(ctx, struct{}{}, struct{}{})
	wg := sync.WaitGroup{}
	wg.Add(2)

	//sender
	go func(ctx context.Context) {
		for i := 0; i < 100; i++ {
			select {
			case <-ctx.Done():
				wg.Done()
				return
			case ch <- i:
			}
		}
	}(ctx1)

	//sender
	go func(ctx context.Context) {
		for i := 0; i < 100; i++ {
			select {
			case <-ctx.Done():
				wg.Done()
				return
			case ch <- i:
			}
		}
	}(ctx2)

	//closer
	go func() {
		//满足一定条件以后
		select {
		case <-time.After(2 * time.Second):
		}

		//关闭所有的sender
		cancel()
		wg.Wait()
		close(ch)
	}()

	//revicer
	go func() {
		nums := 0
		for {
			i, ok := <-ch
			if !ok { //如果已经关闭了,则直接中断
				break
			}
			nums++
			fmt.Println(i)
		}
	}()

	select {
	case <-time.After(time.Hour):
	}
}

len(ch) //查看ch缓冲区内的元素
cap(ch) //查看ch缓冲区的最大容量

type hchan struct {
	qcount   uint           // 队列中的所有元素个数
	dataqsiz uint           // ring buffer的大小
	buf      unsafe.Pointer // ring buffer 数组实现
	elemsize uint16					// 元素大小
	closed   uint32					// 是否关闭
	elemtype *_type // 元素类型
	sendx    uint   // 发送的索引
	recvx    uint   // 接受索引
  	recvq    waitq  // recv 等待队列 (<-chane)
  	sendq    waitq  // send 等待列表 (chan<-)

	lock mutex //锁
}

package ring_buffer

//模拟假设只存入int类型
type hchan struct {
	qcount   uint  // total data in the queue
	dataqsiz uint  // size of the circular queue
	buf      []int // points to an array of dataqsiz elements
	sendx    uint  // send index
	recvx    uint  // receive index
}

//创建
func makechan(size int) *hchan {
	c := &hchan{}
	c.dataqsiz = uint(size)
	c.buf = make([]int, size, size)
	return c
}

func full(c *hchan) bool {
	return c.qcount == c.dataqsiz
}

func empty(c *hchan) bool {
	return c.qcount == 0
}

func send(c *hchan, v int) bool {
	//已满
	if c.qcount == c.dataqsiz {
		return false
	}

	c.buf[c.sendx] = v
	c.sendx++
	if c.sendx == c.dataqsiz {
		c.sendx = 0
	}
	c.qcount++
	return true
}

func recv(c *hchan) int {
	if c.qcount == 0 {
		return -1
	}

	v := c.buf[c.recvx]
	c.recvx++
	if c.recvx == c.dataqsiz {
		c.recvx = 0
	}
	c.qcount--
	return v
}

type waitq struct {
	first *sudog
	last  *sudog
}

type sudog struct {
    // The following fields are protected by the hchan.lock of the
    // channel this sudog is blocking on. shrinkstack depends on
    // this for sudogs involved in channel ops.
  	//下面的字段，由hchan.lock保护
    g *g

    // isSelect indicates g is participating in a select, so
    // g.selectDone must be CAS'd to win the wake-up race.
  	// isSelect 标识 g 是否正在参与一个select，因此g.selectDone必须以 CAS 的方式来避免唤醒时的race
    isSelect bool
    next     *sudog
    prev     *sudog
    elem     unsafe.Pointer // data element (may point to stack) 指向要发送或者接受的元素数据，可能指向栈

    // The following fields are never accessed concurrently.
    // For channels, waitlink is only accessed by g.
    // For semaphores, all fields (including the ones above)
    // are only accessed when holding a semaRoot lock.
  	//以下字段永远不会被并发访问，

    acquiretime int64
    releasetime int64
    ticket      uint32
    parent      *sudog // semaRoot binary tree
    waitlink    *sudog // g.waiting list or semaRoot
    waittail    *sudog // semaRoot
    c           *hchan // channel 反向索引
}

//编译器翻译映射
make(chan interface{}, size) ⇒ runtime.makechan(interface{}, size)
make(chan interface{})       ⇒ runtime.makechan(interface{}, 0)

func makechan(t *chantype, size int) *hchan {
	//无聊的检查
    ...
    
    //可恶的GC，我们先不用关心这块逻辑
	// 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.
	var c *hchan
	switch {
	case mem == 0:
		// Queue or element size is zero.
		c = (*hchan)(mallocgc(hchanSize, nil, true))
		// Race detector uses this location for synchronization.
		c.buf = c.raceaddr()
	case elem.ptrdata == 0:
		// Elements do not contain pointers.
		// Allocate hchan and buf in one call.
		c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
		c.buf = add(unsafe.Pointer(c), hchanSize)
	default:
		// Elements contain pointers.
		c = new(hchan)
		c.buf = mallocgc(mem, elem, true)
	}

	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	c.dataqsiz = uint(size)
	lockInit(&c.lock, lockRankHchan)

	if debugChan {
		print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
	}
	return c
}

ch <- v                      ⇒ runtime.chansend1(ch, &v)

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
	//先进行一堆繁琐的检查
    ...
    
    //加锁
	lock(&c.lock)
	
    // 不允许向已经 close 的 channel 发送数据 否则panic
	if c.closed != 0 { 
		unlock(&c.lock)
		panic(plainError("send on closed channel"))
	}

	// 1. 找到了阻塞在 channel 上的 reciver，发送
	if sg := c.recvq.dequeue(); sg != nil {
		// Found a waiting receiver. We pass the value we want to send
		// directly to the receiver, bypassing the channel buffer (if any).
		send(c, sg, ep, func() { unlock(&c.lock) }, 3) //其中第四个入参是执行成功的回调函数
		return true
	}

	// 2. 判断 channel 中缓存是否仍然有空间剩余
	if c.qcount < c.dataqsiz {
		// Space is available in the channel buffer. Enqueue the element to send.
		// 有空间剩余，入队 因为是内存操作，可以简单的理解为buf[sendx] = elem
		qp := chanbuf(c, c.sendx)
		if raceenabled {
			racereleaseacquire(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
	}

	// Block on the channel. Some receiver will complete our operation for us.
	// 3. 阻塞在 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
	mysg.g = gp
	mysg.isSelect = false
	mysg.c = c
	gp.waiting = mysg
	gp.param = nil
	c.sendq.enqueue(mysg)
	// Signal to anyone trying to shrink our stack that we're about
	// to park on a channel. The window between when this G's status
	// changes and when we set gp.activeStackChans is not safe for
	// stack shrinking.
	atomic.Store8(&gp.parkingOnChan, 1)
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2) // 将当前的 g 从调度队列移出
	// 因为调度器在停止当前 g 的时候会记录运行现场，当恢复阻塞的发送操作时候，会从此处继续开始执行
	// Ensure the value being sent is kept alive until the
	// receiver copies it out. The sudog has a pointer to the
	// stack object, but sudogs aren't considered as roots of the
	// stack tracer.
	KeepAlive(ep)

	// someone woke us up.
	// 被唤醒
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	closed := !mysg.success
	gp.param = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	mysg.c = nil
	releaseSudog(mysg)
	if closed {
		// 正常唤醒状态，goroutine 应该包含需要传递的参数，但如果没有唤醒时的参数，且 channel 没有被关闭，则为虚假唤醒
		if c.closed == 0 {
			throw("chansend: spurious wakeup")
		}
		panic(plainError("send on closed channel"))
	}
	return true
}

func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {
	//竞争检测
    ...
    
	if sg.elem != nil {
		sendDirect(c.elemtype, sg, ep)
		sg.elem = nil
	}
	gp := sg.g
	unlockf()
	gp.param = unsafe.Pointer(sg)
	sg.success = true
	if sg.releasetime != 0 {
		sg.releasetime = cputicks()
	}
	// 放入调度队列等待被后续调度
	// 第二个参数用于 trace 追踪 ip 寄存器的位置，go runtime 又不希望暴露太多内部的调用，因此记录需要跳过多少 ip
	goready(gp, skip+1)
}

v := <- ch                   ⇒ runtime.chanrecv1(ch, &v)
v, ok := <- ch               ⇒ ok := runtime.chanrecv2(ch, &v)

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	//复杂枯燥的检查
	...
    
	lock(&c.lock)
	// 1. channel 已经 close，且 channel 中没有数据，则直接返回
	if c.closed != 0 && c.qcount == 0 {
		if raceenabled {
			raceacquire(c.raceaddr())
		}
		unlock(&c.lock)
		if ep != nil {
			typedmemclr(c.elemtype, ep)
		}
		return true, false
	}
	// 2. 找到发送方，直接接收
	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) }, 3)
		return true, true
	}
    
	// 3. channel 的 buf 不空
	if c.qcount > 0 {
		// Receive directly from queue
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			racereleaseacquire(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.
	// 4. 没有更多的发送方，阻塞 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.isSelect = false
	mysg.c = c
	gp.param = nil
	c.recvq.enqueue(mysg)
	// Signal to anyone trying to shrink our stack that we're about
	// to park on a channel. The window between when this G's status
	// changes and when we set gp.activeStackChans is not safe for
	// stack shrinking.
	atomic.Store8(&gp.parkingOnChan, 1)
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)

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

func closechan(c *hchan) {
	//关闭空chan会panic
    if c == nil {
		panic(plainError("close of nil channel"))
	}

    //加锁
	lock(&c.lock)
    
    //多次close panic
	if c.closed != 0 {
		unlock(&c.lock)
		panic(plainError("close of closed channel"))
	}

	c.closed = 1

	var glist gList

	// release all readers
	// 释放所有的读者
	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 = unsafe.Pointer(sg)
		sg.success = false
		if raceenabled {
			raceacquireg(gp, c.raceaddr())
		}
		glist.push(gp)
	}

	// release all writers (they will panic)
	// 释放所有的写者 (panic)
	for {
		sg := c.sendq.dequeue()
		if sg == nil {
			break
		}
		sg.elem = nil
		if sg.releasetime != 0 {
			sg.releasetime = cputicks()
		}
		gp := sg.g
		gp.param = unsafe.Pointer(sg)
		sg.success = false
		if raceenabled {
			raceacquireg(gp, c.raceaddr())
		}
		glist.push(gp)
	}
	unlock(&c.lock)

	// Ready all Gs now that we've dropped the channel lock.
	// 就绪所有的 G 即可释放 channel 锁
	for !glist.empty() {
		gp := glist.pop()
		gp.schedlink = 0
		goready(gp, 3)
	}
}

//sender
	....
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2) // 将当前的 g 从调度队列移出
	// 因为调度器在停止当前 g 的时候会记录运行现场，当恢复阻塞的发送操作时候，会从此处继续开始执行
	// Ensure the value being sent is kept alive until the
	// receiver copies it out. The sudog has a pointer to the
	// stack object, but sudogs aren't considered as roots of the
	// stack tracer.
	KeepAlive(ep)

	// someone woke us up.
	// 被唤醒
	if mysg != gp.waiting {
		throw("G waiting list is corrupted")
	}
	gp.waiting = nil
	gp.activeStackChans = false
	closed := !mysg.success
	gp.param = nil
	if mysg.releasetime > 0 {
		blockevent(mysg.releasetime-t0, 2)
	}
	mysg.c = nil
	releaseSudog(mysg)
	if closed {
		// 正常唤醒状态，goroutine 应该包含需要传递的参数，但如果没有唤醒时的参数，且 channel 没有被关闭，则为虚假唤醒
		if c.closed == 0 {
			throw("chansend: spurious wakeup")
		}
		panic(plainError("send on closed channel")) //直接报panic
	}
	return true
}

//reciver
	...
	gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)

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