channel and csp;

1
2
3
4
5
6
7

if channe.receiveQueue.lengt==0 &&buffer.full ; 
	block:
		sleep(current);channel.sendqueue.push(g)

if channel.reqceveiqueue.length>0;
	g = receiveq.first;
	wakeup g;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

 channel struct{
	 recevieQueue
	 sendQueue
	 buffer
 }


sendtoChannel(data):
	if buffer.isFull:
	    g.status = waiting
		channel.pushtoSendqueue(currentG);
		reScheule()

	if buffer.isFull == NO:
		if channel.receiveQ notEmpty:
			g = receiveQ.pop()
			g.status = runable;
			g.receive(data) 
			put in procesoor local queue 
	   else:
			 buffer.add(data)
				   
			

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

type hchan struct {
    qcount   uint           // total data in the queue
    dataqsiz uint           // size of the circular queue
    buf      unsafe.Pointer // points to an array of dataqsiz elements
    elemsize uint16
    closed   uint32
    elemtype *_type // element type
    sendx    uint   // send index
    recvx    uint   // receive index
    recvq    waitq  // list of recv waiters
    sendq    waitq  // list of send waiters

    lock mutex
}

type sudog struct{
   g *g
   isSelect bool
   next *sudog
   prev *sudog
   elem unsafe.Pointer //data element
   ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

func gopark(unlockf func(*g, unsafe.Pointer) bool, lock unsafe.Pointer, reason waitReason, traceEv byte, traceskip int) {
	if reason != waitReasonSleep {
		checkTimeouts() // timeouts may expire while two goroutines keep the scheduler busy
	}
	mp := acquirem()
	gp := mp.curg
	status := readgstatus(gp)
	if status != _Grunning && status != _Gscanrunning {
		throw("gopark: bad g status")
	}
	mp.waitlock = lock
	mp.waitunlockf = *(*unsafe.Pointer)(unsafe.Pointer(&unlockf))
	gp.waitreason = reason
	mp.waittraceev = traceEv
	mp.waittraceskip = traceskip
	releasem(mp)
	// can't do anything that might move the G between Ms here.
	mcall(park_m)
}
func park_m(gp *g) {
	// g0
	_g_ := getg()

	if trace.enabled {
		traceGoPark(_g_.m.waittraceev, _g_.m.waittraceskip)
	}
	//线程安全更新gp的状态，置为_Gwaiting
	casgstatus(gp, _Grunning, _Gwaiting)
	// 移除gp与m的绑定关系
	dropg()

	if _g_.m.waitunlockf != nil {
		fn := *(*func(*g, unsafe.Pointer) bool)(unsafe.Pointer(&_g_.m.waitunlockf))
		ok := fn(gp, _g_.m.waitlock)
		_g_.m.waitunlockf = nil
		_g_.m.waitlock = nil
		if !ok {
			if trace.enabled {
				traceGoUnpark(gp, 2)
			}
			casgstatus(gp, _Gwaiting, _Grunnable)
			execute(gp, true) // Schedule it back, never returns.
		}
	}
	// 重新做一次调度
	schedule()
}

1
2
3
4
5
6
7
8
9

data, open :=<-c
if !open print("closed")



data range channel{
	...
}
fmt.Println("now is cloese")

1
2

g1  handle data; Locker update data UnLocker  
g2  ...

1
2

g1 handle data    ->   g3: update data
g2 handle data    ->   g3: update data

1
2
3

	 for data range datas: go run data

	 for data range channel: updateby data

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

	producer:
   		go:  queryMysql(data1) ;
   		go:  queryMysql(data2)

	consumer and new producer:
	 	
		for data range c{
			go: queryES(data)
		}
		// or 
		for data range c{
			list.append(data)
		}
		go ES(data[0]+ data[1])
		go ES(data[1]+ data[2])

	consumer:
		for range channel

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

 func search(c chan, ctx context):
 	var  innerChannel = make(chan int, 1)
 	
 	go:	
 		data = doSearch(innerChannel);
 		innerChannel <- data;
 	select:
 		case <-ctx.done(): //
 			return;
 		case data := <-innerChannel:
 			 	outChannel <-data

1
2
3

go func(){
	<-c // 没有接受者
}()

1
2
3
4
5

go func1: defer wg.done
go func2:  defer wg.done;

go func: wg.wait; close(chanel)

1
2
3
4
5
6

go func: buffer<-data

go func: 
	select 
		<-bufer 
		 <-ctx

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func sendLog(logC chan string, w *sync.WaitGroup) {
	defer w.Done()
	select {
	case logC <- "hello":
		fmt.Println("log file is empty")
	case <-time.After(time.Millisecond * 100):
		fmt.Println("abort log data")
	}
}
func mockWriteToDisk(logC <-chan string) {
	for v := range logC {
		// mock write to disk operation
		time.Sleep(time.Second)
		fmt.Println("write to disk: ", v)
	}
}

1
2
3
4
5
6
7

cotrol = make(chan struct{},3)
for _,value :=  range []int{1,2,3}:
	 control <- struct{}
	 go func:
		xxx
		<-control

1
2

go sender:
	buffer<- data

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

c1 := make(chan int,1)
c2 : = make(chan int,1)



g1: // c1.full, block.
	// 
	c1<-1
	fmt.Println(1)
	c2<-1  // unblock c2
	c1<-1 // block me 
	fmt.Println(3)
	c2<-1 // unblock c2
	c1<-1 // block me 
	...

g2: // c2.empty, bloock
	<-c2 // block  
	fmt.Println(2)
	<-c1 // unbock g1
	<-c2 //block   
	fmt.Println(4)
	<-c1 // unblock c1
	<-c2 // block me 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

go func1():
	defer signalChan<-
	do sth...

go func2(): 
	....


for{
	<-signalChann<-
	finish++;
	if finish == num:
		close(c)
}

1
2
3
4
5
6
7

w.Add(N);

go fun1:
	defer w.Done
	....

go: w.wait()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func longJob: ctx,cha // producer
	select:
		case <-ctx.Done():
			return
		case data <-innerChan:
			cha<-data
			
	go:
		data = query()
		innerChan<-dat
	

1
2
3
4
5

controlChannel = make(chan struct{},poolSize);
controlChannel <-  struct{}{};
go func(){

}()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// generator
for data := range datas:
	dataC<-data



// consumer1
go for:
	defer  wg.done;
	v,ok := <-dataC
	if !ok return ;
	excute task
	resultC <- result 

// consumer2
go for:
	....

// --- -----
go countG:
	wg.wait
	close(resultC)


for result  range  resultG: