[Go] usage of nil channel

func asChan(vs ...int) <-chan int {
	c := make(chan int)
	go func() {
		for _, v := range vs {
			c <- v
			time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
		}
		close(c)
	}()
	return c
}


func merge(a, b <-chan int) <-chan int {
	c := make(chan int)
	go func() {
		defer close(c)
		for a != nil || b != nil {
			select {
			case v, ok := <-a:
				if !ok {
					fmt.Println("a is done")
					a = nil    // set to nil prevent processing
					continue
				}
				c <- v
			case v, ok := <-b:
				if !ok {
					fmt.Println("b is done")
					b = nil    // set to nil prevent processing
					continue
				}
				c <- v
			}
		}
	}()
	return c
}


func main() {
	a := asChan(1, 3, 5, 7)
	b := asChan(2, 4, 6, 8)
	c := merge(a, b)
	for v := range c {
		fmt.Println(v)
	}
}

[Go] hchan data structure

Reference:
https://groups.google.com/g/golang-nuts/c/y_MZqc_WHnw/m/S64o1iPuCQAJ

Quote:

The garbage collector has to always know exactly where pointers are stored in memory.
We could in principle use contiguous allocation for the case where the element type contains pointers, but we would have to build a description that tells the garbage collector exactly where those pointers are. Those descriptions are built by the compiler (on tip, the code is in cmd/compile/internal/reflectdata/reflect.go), so the compiler would have to build a new descriptor for every channel type with an element type that contains pointers.

And the descriptor would have to vary based on the channel size, so it would be based not just on the channel type but also on the argument passed to "make".

Of course the argument passed to "make" can be a variable, so that adds another complication.

code snippet:

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)
}

Quote: 

AFAIK when there's no pointer in elem (the channel's element's type), we can cheat/optimize (call as you wish), and allocate the channel's backing memory AND all it's elements' memory at once, AS A BYTE ARRAY (nil is the second argument of mallocgc).

Then we can play unsafe tricks an treat it as proper channel and its buffer.

When there's pointer somewhere in the element's type, then - as the garbage collector must know each and every pointer's placement - we don't play tricks, and give mallocgc the proper type information, and allocate the element buffer separate from the channel backing. You'd have to repeat mallocgc's functionality to provide the gc with proper information, just to save one allocation. That'd be too much work for less gain.

Quote:

When the code you showed calls mallocgc, it passes the type descriptor

for the channel element type. This is called "elem" in the code.

This type descriptor was created by the compiler.

If the runtime code allocated both the channel data structure and the

buffer in a single memory allocation, it would need to have a type

descriptor that combined the channel data structure with the element

type. Not only that, this new type descriptor would change based on

the argument passed to make. In the existing code, that is not

necessary; when mallocgc is passed a type descriptor to allocate a

size that is larger than the type, it understands that it is

allocating an array. That wouldn't work for an allocation that shares

the channel data structure with the channel buffer.

I don't know what you mean when you say "gc would find the descriptor

later with more time." The compiler would have to create the

descriptor at compile time, so that the runtime code could use that

type descriptor to allocate the memory. The gc can't find the

descriptor later.

[Go] consider the size of a channel

While setting the size of a channel,
should consider how go scheduling mechanism works.

A small size of channel(including an un-buffered channel) causing the producer to block the running goroutine thus forces a re-scheduling (on same physical thread most of the time).

While a larger size of channel can reduce the latency of producing data into the channel but having the consumer delay the processing.

Here's the time slice down into different stages(ref by Vincent Blanchon's article):

Measure the performance before hard-setting the channel size,
i.e make the size configurable.

Reference:
https://medium.com/a-journey-with-go/go-what-does-a-goroutine-switch-actually-involve-394c202dddb7

[Go] signal passing through channel implement differently on each platform

Reference: 
https://groups.google.com/forum/#!msg/golang-nuts/J5SXvinQTFE/FiCJPAV8BAAJ

futex:
http://vsdmars.blogspot.com/2019/03/futex-futex-skim-through.html

Below the program crashes at Darwin while OK under Linux.
This is due to Darwin is using pipe to (which opens up the minimum fd, i.e 3)
and Linux is using futex for the signal event passing. (which is fast)

Do NOT open random fd and manipulate with it unless you know the program logic~

package main

import (
        "os"
        "os/signal"
)

func main() {
        sigs := make(chan os.Signal)
        signal.Notify(sigs, os.Interrupt)

        if f := os.NewFile(3, ""); f != nil {
                f.Close()
        }

        <-sigs
}

[Go] select through slice of channels from reflect.Select

Reference:
https://golang.org/pkg/reflect/#Select

e.g https://play.golang.org/p/Ul-e3DoVv5K

package main

import (
 "context"
 "fmt"
 "math/rand"
 "reflect"
 "time"
)

func main() {
 const numberOfTest = 10
 soc := make([]reflect.SelectCase, numberOfTest)
 channels := make([]chan int, numberOfTest)

 for idx := range channels {
  channels[idx] = make(chan int, 1)
 }

 for i := range soc {
  soc[i] = reflect.SelectCase{
   Dir:  reflect.SelectRecv,
   Chan: reflect.ValueOf(channels[i]),
  }
 }

 ctx, cancel := context.WithCancel(context.Background())

 go func() {
  send := make(chan struct {
   int
   reflect.Value
   bool
  }, 1)

  receive := send

  go func() {
   for {
    // block call, could use a circuit breaker,
    // e.g https://github.com/sony/gobreaker
    idx, val, ok := reflect.Select(soc)

    send <- struct {
     int
     reflect.Value
     bool
    }{idx, val, ok}
   }
  }()

  for {
   select {
   case <-ctx.Done():
    fmt.Println("bye~")
    return
   case payload := <-receive:
    fmt.Printf("idx: %d, value: %v, bool: %t\n",
     payload.int,
     payload.Value,
     payload.bool)
   }
  }
 }()

 go func() {
  for {
   for idx := range channels {
    channels[idx] <- rand.Intn(100)
   }

   time.Sleep(3 * time.Second)
  }

 }()

 time.Sleep(10 * time.Second)
 cancel()
 time.Sleep(3 * time.Second)
}

[Go] channel in detail

Channel traits:

• goroutine is safe
• store and pass values between goroutines, Golang is a value based language same as C and C++
• provides FIFO semantic
• can cause goroutines to block and unblock

How would you design a channel?

Ring buffer!

With ring buffer, we need:

• the type that this ring buffer holds, thus compiler knows the size of the ring buffer which could allocate it's memory.
• start idx(hchan.recvx), end idx(hchan.sendx) (thus we know the current data size)
  while hchan.recvx == hchan.sendx means either 0 data or ring is full.
• ring buffer size
• how many senders/receivers are consuming the ring
  thus we need mutex lock

Let's take a look at hchan struct in golang chan.go implement:
https://github.com/golang/go/blob/master/src/runtime/chan.go#L32


hchan instance usually allocated on the heap, unless this channel is only been used in one single thread(stack, to be precise).


ch <- 42 will copy data into the ring buffer with mutex locks hchan, increase the hchan.sendx.
<- ch will copy the value from ring buffer with mutex locks hchan, increase the hchan.recvx.


If the ring buffer is full, send blocks, and golang scheduler suspends the send goroutine (aka. user space thread, not the OS thread) until there's room in the ring buffer.

Please refer to previous post about scheduler:
http://vsdmars.blogspot.com/2018/06/golang-go-scheduler.html
Golang scheduler design doc:
https://docs.google.com/document/d/1TTj4T2JO42uD5ID9e89oa0sLKhJYD0Y_kqxDv3I3XMw


Each (G)oroutine is managed by (P)rocessor, and each (P)rocessor can only be run by one (M)thread.
(P)rocessor has the runQ (list) holds (G)oroutines.

If the ring buffer is full, the sender pauses, scheduler calls 'gopark' to remove the caller (G)oroutine from (M)thread into (P)rocessor's runQ(of (G)oroutines).
https://github.com/golang/go/blob/master/src/runtime/proc.go#L284:6


hchan has the information of senders and receivers.
https://github.com/golang/go/blob/master/src/runtime/chan.go#L53:6

hchan.sendq // waitq
hchan.recvq  // waitq

type waitq struct {
    first *sudog
    last  *sudog
}

sudog:
https://github.com/golang/go/blob/master/src/runtime/runtime2.go#L276

sudog.elem represends the data which is about to send(copy)/receive(copy) to/from the ring buffer.



Here's the interesting part, the implementation optimized sending/receiving data by skip copy data into/out from ring buffer iff there's a waiting (G)oroutine, which directly store the sending/receiving pointer to data into the waiting (G)oroutine's sudog.elem
Receive optimized:
https://github.com/golang/go/blob/master/src/runtime/chan.go#L190
Send optimized:
https://github.com/golang/go/blob/master/src/runtime/chan.go#L473



Fast path implement:
Sender:
https://github.com/golang/go/blob/master/src/runtime/chan.go#L159
Receiver: (atomic.Load involved)
https://github.com/golang/go/blob/master/src/runtime/chan.go#L437


Quote from Ian Lance Taylor:
https://groups.google.com/forum/#!topic/golang-nuts/L-bGvB52UrU
The runtime package is always compiled by a known compiler, and is permitted to know how that specific compiler behaves. If the compiler changes, while still staying within spec, the runtime package will change too. You can't argue about the language's memory model based on the code in the runtime package.


Reference:
Ring buffer in C:
https://embeddedartistry.com/blog/2017/4/6/circular-buffers-in-cc
chan.go:
https://github.com/golang/go/blob/master/src/runtime/chan.go
gopark:
https://github.com/golang/go/blob/master/src/runtime/proc.go#L284:6
goready:
https://github.com/golang/go/blob/master/src/runtime/proc.go#L310:6
Benign Data Races: What Could Possibly Go Wrong?
https://software.intel.com/en-us/blogs/2013/01/06/benign-data-races-what-could-possibly-go-wrong
The Go Memory Model:
https://golang.org/ref/mem

[Actor model] Some real-world implementation details from AKKA

Found this talk on YT recently, my comments + notes below:

Introduction to the Actor Model for Concurrent Computation - John Murray

Actor is:

• persistent
• has internal state
• asyc
• Independent event-loop + memory
• Mailbox (receive message)
• React in FIFO order

Well,
actor model is concurrent with regard to a system of actions.
actor model is Not concurrent with regard to data.
(surely we can do implement concurrent thread inside one actor.)

Using channel as mutex to single variable

So, how to make a mutable variable access by multiple threads?

In Golang, we have channel.

• Create a channel with the length of 1.
• Put the mutable variable in, and mutiple goroutines have access to that channel.
• However, only one goroutine can successful retrieve the mutable varible inside the length of 1 channel, other goroutines since the channel has no value, it will block.
• This acts as a MUTEX for goroutines.
• Once the goroutine which having the mutable variable changed,
  it puts it back to the channel thus other goroutines can access.

Actor can:

• create more Actors
• receive messages and response
• make local decisions
• perform arbitrary, side-effecting action
• send messages
• respond to the sender 0 or more times
  (to clearfy here, by "respond" means
  send message to the sender Actor
  Not duplex channel respond)
• Process exactly one message at a time

Actor communication is:

• No channels or internediaries (such as in CSP, e.g. golang)
• "best effor" delivery
• at-most-once delivery
• Messages can take arbitrary long to be delivered (has no concept of time)
• No message ordering guarantees

Actor address:

• identify the actor
• may also represent a proxy / forwarder to an Actor
• contains location and transport information
• don't care where the Actor lives
  (can be inside the same process, different nodes, different containers, etc)
• one address may represent many Actors(pool)

Error handling:

• Supervision
• the running state of an Actor is monitored and managed by another Actor(the Supervisor)
• Supervision has:
• constantly monitors running state of actor
• can perform actions based on the state of the Actor( e.g unhandled error, restart Actor)
  (In Golang, we take advantage of context.Context)
• transparent life-cycle management
• addresses do not change during restarts
  (we implemented with Actor's hash(name + uuid + hostname) as 'address')
  BE WARE.
  This only has meaning iff Actor is not pure, which is STATEFUL.
• Persist state into local sqlite, loaded with (name + uuid)
• mailboxes are persisted outside the Actor instances (Auh, K.I.S.S)
  I doubt the use of Supervisor idea.
  Think about this, does the Sender care how the Receiver act when it receives the message? ;-P

 

Implement Address contains these as a group:

• mailbox
• Actor

Anti use-case:

• Working on a non-concurrent system
• performance critical applications
• non-concurrent communication is involved
• no mutable state

Draw backs:

• too many Actors
• testing
• debugging

Extra material:
Don't use Actors for concurrency


CRDT:
https://medium.com/@istanbul_techie/a-look-at-conflict-free-replicated-data-types-crdt-221a5f629e7e

[Go][book] Concurrency in Go - Katherine Cox-Buday

Summary of reading:
Concurrency in Go - Katherine Cox-Buday

I agree with Eli Bendersky's point of view about this book i.e
if coming from the arena of C++, which is that, the book is relatively
can be considered as a 'reference book' for some golang tricks.

Before delve into golang concurrency, the understanding of
golang scheduler is a must.
Reference:
http://vsdmars.blogspot.com/2018/06/golang-go-scheduler.html
http://vsdmars.blogspot.com/2018/07/golangnote-analysis-of-go-runtime.html
HPX lib (C++)
Intel TBB (C++)

Understanding memory model is also essential to concurrency, although in golang, this is mostly hidden by it's run-time and channel.
Reference:
http://vsdmars.blogspot.com/2018/09/concurrency-c-wrap-up-2018.html


For advanced concurrency programming, take a look at
The Art of Multiprocessor Programming - Maurice Herlihy & Nir Shavit
Reference:
http://vsdmars.blogspot.com/2016/01/multiprocessor-programming-types-of.html

Jotting down reading notes for those are fun/useful :-)

RWLock:
[C++]
std::shared_mutex

[Golang]
sync.RWMutex

[Python]
None, and i coined one for fun:
https://github.com/verbalsaintmars/python_util/tree/master/rwlock

Cond Variable:
[C++]
std::condition_variable

[Golang]
sync.NewCond

sync.Once dead lock example:

var onceA, onceB sync.Once
var initB func()
initA := func() { onceB.Do(initB) }
initB = func() { onceA.Do(initA) }
onceA.Do(initA)

sync.Pool
sync.Pool can be the panacea for creating new type instance
inside a 'loop', i.e for loop.

Experienced engineer could spot the performance hit 'bad design'
by seeing new/make_shared/NewXXX etc. inside a for/while loop.

Reusing the type instance from being GCed boosts the performance;
however, just that, the pool stored type instance should be stateless
to avoid side-effect.

GC can still drain the Pool, thus a .Get() with drained Pool will create a new type's instance.

Further intriguing read:
Purpose of sync.Pool 
Learn High Performance Go 

Close(Channel):
close(channel)  acts as a signal for those blocking channels to continue.
e.g
https://play.golang.org/p/k3143TbiqO4

package main

import (
 "fmt"
 "time"
)

var ch = make(chan struct{})

func run(cnt int) {
 <-ch
 fmt.Println(cnt)
}

func main() {
 go run(1)
 go run(2)
 go run(3)
 go run(4)

 time.Sleep(10 * time.Second)
 close(ch)
 time.Sleep(10 * time.Second)
}

Select as normalized distribution seed:
Using select to act as a normalized distribution seed.
e.g:
https://play.golang.org/p/2m4VB9rhfcU

package main
import "fmt"

func main() {
 c1 := make(chan interface{})
 close(c1)
 c2 := make(chan interface{})
 close(c2)

 var c1Count, c2Count int

 for i := 1000; i >= 0; i-- {
  select {
  case <-c1:
   c1Count++
  case <-c2:
   c2Count++
  }
 }
 fmt.Println(c1Count)
 fmt.Println(c2Count)
}

Daisy-chain (it's simply elegant):
from slide:https://talks.golang.org/2012/concurrency.slide

func f(left, right chan int) {
    left <- 1 + <-right
}

func main() {
    const n = 10000
    leftmost := make(chan int)
    right := leftmost
    left := leftmost
    for i := 0; i < n; i++ {
        right = make(chan int)
        go f(left, right)
        left = right
    }
    go func(c chan int) { c <- 1 }(right)
    fmt.Println(<-leftmost)
}

Legacy golang code alert (prior Go1.5):

runtime.GOMAXPROCS(runtime.NumCPU())

i.e Go 1.5 is set to make the default value of GOMAXPROCS
the same as the number of CPUs on your machine,
so above code isn't necessary anymore.


Channel error/result back to caller:
https://play.golang.org/p/aCowW5mNEp1

package main

import "fmt"

func main() {
 type result struct {
  stat int
 }

 CheckErr := func(input chan int) <-chan result {
  r := make(chan result)

  go func() {
   defer close(r)
   for i := range input {
    r <- result{i}
   }
  }()

  return r
 }

 input := make(chan int)

 go func() {
  defer close(input)
  for i := range [10]struct{}{} {
   input <- i
  }
 }()

 for result := range CheckErr(input) {
  fmt.Println(result)
 }

}

The or-channel: 
https://play.golang.org/p/XsDYdU8ks5D

package main

import (
 "time"
)

func main() {

 var or func(channels ...chan interface{}) <-chan interface{}

 or = func(channels ...chan interface{}) <-chan interface{} {
  switch len(channels) {
  case 0:
   return nil
  case 1:
   return channels[0]
  }

  orDone := make(chan interface{})

  go func() {
   defer close(orDone) // Trick is here :-)

   switch len(channels) {
   case 2:
    select {
    case <-channels[0]:
    case <-channels[1]:
    }
   default:
    select {
    case <-channels[0]:
    case <-channels[1]:
    case <-channels[2]:
    case <-or(append(channels[3:], orDone)...):
    }
   }
  }()
  return orDone
 }

 chs := make([]chan interface{}, 7)
 for i, _ := range chs {
  chs[i] = make(chan interface{}, 1)
 }

 done := make(chan struct{})

 go func() {
  <-or(chs...)
  done <- struct{}{}
 }()
 time.Sleep(3 * time.Second)
 chs[0] <- struct{}{}
 <-done
}

Best Practices for Constructing Pipelines:
https://play.golang.org/p/4efPkGXUSoM

package main

import "fmt"

func main() {
 pipe := func(done chan struct{}, i ...int) chan int {
  retch := make(chan int)

  go func() {
   defer close(retch)
   for fi := range i {
    select {
    case <-done:
    case retch <- fi:
    }
   }
  }()

  return retch
 }

 done := make(chan struct{})
 for c := range pipe(done, 47, 38, 29) {
  fmt.Println(c)
 }
}

Fan-Out, Fan-In (auh, map-reduce):
https://play.golang.org/p/fmGAGFCuzDy
Fan-out:
Use for-range as pipeline channel to fan-out.

Take advantage of
var wg sync.WaitGroup
defer wg.done()
wg.Wait()
to fan-in.

package main

import "sync"

func main() {

 fanIn := func(done <-chan interface{}, channels ...<-chan interface{},
 ) <-chan interface{} {

  var wg sync.WaitGroup

  multiplexedStream := make(chan interface{})

  multiplex := func(c <-chan interface{}) {
   defer wg.Done()
   for i := range c {
    select {
    case <-done:
     return
    case multiplexedStream <- i:
    }
   }
  }

  wg.Add(len(channels))

  for _, c := range channels {
   go multiplex(c)
  }

  go func() {
   wg.Wait()
   close(multiplexedStream)
  }()

  return multiplexedStream
 }
 done := make(chan interface{})
 chan1 := make(chan interface{}, 1)
 chan2 := make(chan interface{}, 1)
 chan1 <- struct{}{}
 chan2 <- struct{}{}
 <-fanIn(done, chan1, chan2)
}

The or-done-channel:
https://play.golang.org/p/HsgZe8Ke9xU
Acted as a building block for other services.

package main

import (
 "fmt"
 "time"
)

func main() {

 orDone := func(done, c <-chan interface{}) <-chan interface{} {
  valStream := make(chan interface{})
  go func() {
   defer close(valStream)
   for {
    select {
    case <-done:
     return
    case v, ok := <-c:
     if ok == false {
      return
     }
     select {
     case valStream <- v:
     case <-done:
     }
    }
   }
  }()
  return valStream
 }
 ch := make(chan interface{}, 3)
 ch <- 1
 ch <- 2
 ch <- 3

 go func() {
  time.Sleep(3 * time.Second)
  close(ch)
 }()

 for v := range orDone(make(chan interface{}), ch) {
  fmt.Println(v)
 }
}

The tee-channel:
https://play.golang.org/p/bcbVnfmUdBT
interleave output.

package main

import (
 "fmt"
 "time"
)

func main() {

 orDone := func(done, c <-chan interface{}) <-chan interface{} {
  valStream := make(chan interface{})
  go func() {
   defer close(valStream)
   for {
    select {
    case <-done:
     return
    case v, ok := <-c:
     if ok == false {
      return
     }
     select {
     case valStream <- v:
     case <-done:
     }
    }
   }
  }()
  return valStream
 }

 done := make(chan interface{})
 ch := make(chan interface{}, 3)
 ch <- 1
 ch <- 2
 ch <- 3

 tee := func(done <-chan interface{}, in <-chan interface{}) (_, _ <-chan interface{}) {
  out1 := make(chan interface{})
  out2 := make(chan interface{})

  go func() {
   defer close(out1)
   defer close(out2)

   for val := range orDone(done, in) {
    var out1, out2 = out1, out2
    for i := 0; i < 2; i++ {
     select {
     case <-done:
      return
     case out1 <- val:
      out1 = nil // next loop out1 will be ignored.
     case out2 <- val:
      out2 = nil // ditto
     }
    }
   }
  }()
  return out1, out2
 }

 go func() {
  time.Sleep(3 * time.Second)
  done <- struct{}{}
 }()

 a, b := tee(done, ch)
 for v := range a {
  fmt.Printf("a val: %v\n", v)
  fmt.Printf("b val: %v\n", <-b)
 }
}

The bridge-channel:
https://play.golang.org/p/jvTDGZbFeZO
Channel contains channels.
Fan-out by assigning each sub-channels into
fan-in channel for caller to consume.

Till now you can see most of the implements from the book
use 0 buffer channel.

It's the lazy-binding technique which only processes data until
the caller consumes the processed data.

Like 'yield' (i.e coroutine, light weight thread/green thread)
in major languages, e.g C#, Python
Reference:
[C++]
http://c9x.me/articles/gthreads/intro.html

package main

import (
 "fmt"
 "time"
)

func main() {
 orDone := func(done, c <-chan interface{}) <-chan interface{} {
  valStream := make(chan interface{})
  go func() {
   defer close(valStream)
   for {
    select {
    case <-done:
     return
    case v, ok := <-c:
     if ok == false {
      return
     }
     select {
     case valStream <- v:
     case <-done:
     }
    }
   }
  }()
  return valStream
 }

 bridge := func(
  done <-chan interface{},
  chanStream <-chan <-chan interface{},
 ) <-chan interface{} {

  valStream := make(chan interface{})

  go func() {
   defer close(valStream)

   for {
    var stream <-chan interface{}

    select {
    case maybeStream, ok := <-chanStream:
     if ok == false {
      return
     }
     stream = maybeStream
    case <-done:
     return
    }

    for val := range orDone(done, stream) {
     select {
     case valStream <- val:
     case <-done:
     }
    }
   }
  }()
  return valStream
 }

 done := make(chan interface{})
 base := make(chan interface{}, 1)
 next := make(chan (<-chan interface{}), 1)
 base <- 42
 next <- base

 go func() {
  time.Sleep(3 * time.Second)
  close(base)
  close(next)
 }()

 result := bridge(done, next)
 for i := range result {
  fmt.Println(i)
 }
}

Queuing:
Buffered channels piped together then run in sequence.
The receiving type and returning type could be the same thus forms a Monad..
Reference:
http://vsdmars.blogspot.com/2017/09/fp-functor-applicative-monoid-monad.html
http://vsdmars.blogspot.com/2016/08/fp-related-readstudy-links.html



The context Package:
Familiar with the usage, skim over this section.

Timeouts and Cancellation:
Use context timeout mechanism, passing down context instance
through the calling stack.


Heartbeat:
https://play.golang.org/p/4RHoiIjnLLL
Use time.Tick(interval) https://golang.org/pkg/time/#Tick  as returning channel.
Can be used as interval probing check.

time.After can be used to check if a select statement is in healthy state within the timeout.

package main

import (
 "fmt"
 "time"
)

func main() {
 doWork := func(done <-chan interface{},
  pulseInterval time.Duration) (<-chan interface{}, <-chan 
time.Time) {

  heartbeat := make(chan interface{})
  results := make(chan time.Time)

  go func() {
   defer close(heartbeat)
   defer close(results)

   pulse := time.Tick(pulseInterval)
   workGen := time.Tick(2 * pulseInterval)

   sendPulse := func() {
    select {
    case heartbeat <- struct{}{}:
    default:
    }
   }

   sendResult := func(r time.Time) {
    for {
     select {
     case <-done:
      return
     case <-pulse:
      sendPulse()
     case results <- r:
      return
     }
    }
   }

   for {
    select {
    case <-done:
     return
    case <-pulse:
     sendPulse()
    case r := <-workGen:
     sendResult(r)
    }
   }
  }()

  return heartbeat, results
 }

 done := make(chan interface{})

 go func() {
  defer close(done)
  time.Sleep(3 * time.Second)
 }()

 h, r := doWork(done, 120*time.Second)

 for {
  select {
  case _, ok := <-h:
   if ok == false {
    return
   }
  case _, ok := <-r:
   if ok == false {
    return
   }
  case <-time.After(1 * time.Second):
   fmt.Println("worker goroutine is not healthy!")
   return
  }
 }
}

Purpose of nil channel:
Avoid busy loop.

func merge(a, b <-chan int) <-chan int {
 c := make(chan int)
 go func() {
  defer close(c)
  for a != nil || b != nil {
   select {
   case v, ok := <-a:
    if !ok {
     fmt.Println("a is done")
     a = nil
     continue
    }
    c <- v
   case v, ok := <-b:
    if !ok {
     fmt.Println("b is done")
     b = nil
     continue
    }
    c <- v
   }
  }
 }()
 return c
}

[Concurrency] [C++][Go] Wrap up - 2018

Modified

• Thie Core's cache line has the modified data.
• Data in memory won't be in other Core's cache line.

Exclusive

• Data aren't modified.
• Data in memory is the latest.
• If the Core has to evict the data inside the cache line,
  nothing need to be write back to memory.

Shared

• Data are shared between Cores' cache line.
• If this core has to modify the data, it needs to ask for data from other cores first.

Invalid

• The data in the cache line is null.

Operations:

Read

• read data from cache lines. Ask for other cores' for data.  

Read response

• Data for read request. Either from Memory or from Cores' cache.

Invalidate

• Invalidate the particular data inside Cores' cache lines.

Invalidate ack

• Response to invalidate request that the request is in the queue.      

Read invalidate

• Read + Invalidate.
• Will get read's data response and invalidate ack.

Writeback

• Write the data from cache line to memory.

Store buffer:

• Read invalidate request to other Cores' cache line for the to be modified data  is not necessary since the data response isn't needed because this Core is going to modify the data anyway.
• Thus, this core will write data to Store buffer first.
• Thus, Core will read data inside the Store Buffer with highest priority if the data exist in Store buffer and Cache line.
• However, there's an issue for Read invalidate request to other cores.
  Consider that data A in Core 1 cache, data B in Core 2 cache.
  And data A, data B has this happen before relationship.
  i.e
  In Core 1, update data A(Core 1), then update data B(Core 2).
  In Core 2, read data A(Core 1), verify data B(Core 2).

 
You see the problem. Between operation on A/B there's interleaves.
 
Core 1 updates data B, write into store buffer, send read invalidate to Core 2 on data B, at the same time, Core 2 reads data A, receives data A from Core 1, and verify data B as well, at the time, data B isn't invalidated yet.

Thus we need some wait mechanism.
i.e memory barrier.

In the above case, we need

• WMB(write memory barrier) on Core 1,
• RMB(read memory barrier) on Core 2. 

For Core 1, we do:
Update B, then WMB, then update data A.

For Core 2, we do:
Read data A, if A is old data, verify B won't happen, and if A is new data.
i.e
Core 1 updated data A, data B in Core 2 SHOULD use new one from Core 1, since it's invalidate queued.
But how to trigger Core 2 to verify invalidate queue? Before verify data B in Core 2, call RMB.
So the sequence for Core 2 will become:
Read Data A, RMB, verify data B.
     
 
WMB:

• Send read invalidate to Core 2 and till Core 2 send back invalidate ack will Core 1 flush data A from store buffer to it's cache line.

RMW:

• Verify invalidate queue, make data that is in the invalidate queue invalid the Core 2's cache line.
• While Core 2 tries to read the data, it will send a 'read' request to Core 1 for data B.

Take away:

• WMB should be issued AFTER a write to the shared data.
• RMW should be issued BEFORE a read to the shared data.

     
Reference:

1. Memory Barriers: a Hardware View for Software Hackers
2. https://en.cppreference.com/w/cpp/atomic/memory_order
3. https://en.cppreference.com/w/cpp/atomic/atomic_thread_fence

Code:

#include <atomic>
#include <iostream>
#include <thread>

using namespace std;
atomic<int> A{0};
atomic<int> B{0};

void t1()
{
    this_thread::sleep_for(1s);
    B.store(38, memory_order_relaxed);

    int a = 42;
    // WMB, prevents 'a' passing A.store.
    A.store(a, memory_order_release);
}

void t2()
{
    // RMB
    while (A.load(memory_order_acquire) != 42) {
        cout << "in while" << endl;
        // Prints 0 or 38 if B.store hasn't process yet.
        cout << B.load(memory_order_relaxed) << endl;
    }
    cout << "out while" << endl;
    // Always print 38.
    cout << B << endl;
}
int main()
{
    thread T1{t1};
    thread T2{t2};
    T1.join();
    T2.join();
}

Fence:

#include <atomic>
#include <iostream>
#include <thread>
using namespace std;
atomic<int> A{0};
atomic<int> B{0};

void t1()
{
    A.store(42, memory_order_relaxed);
    atomic_thread_fence(memory_order_release);
    // B.store can NEVER before A.store.
    B.store(38, memory_order_relaxed);
}

void t2()
{
    while (B.load(memory_order_relaxed) == 38) {
        cout << "in while loop\n";
        cout << A << endl; // Must be 42
        break;
    }
    cout << "out while loop\n";
}

int main()
{
    thread T1{t1};
    thread T2{t2};
    T1.join();
    T2.join();
}

Fence:

Release/Store

• Prevents all preceding memory operations from being reordered past subsequent writes.
• Prevents all following memory operations from being memory reordered before the write.

Acquire/Load

• Prevents all following memory operations from being reordered before this fence.
• Prevents all preceding memory operations from being reordered pass this fence.

Memory fences are NOT an acquire or release operation.

Operation:

Release operation:  store

• Cannot be reordered by compiler.
• Prevents preceding memory operations from being reordered past itself.
  i.e Any operations after a store can be reordered before the store operation.
• Any read or write operation that precedes it in program order.
  i.e those in memory_order_relaxed mode can't be reordered.

     
Acquire operation:  load

• https://en.cppreference.com/w/cpp/atomic/atomic_load
• Cannot be reordered by compiler.
• Any read or write operation that follows it in program order.
  i.e those in memory_order_relaxed mode can't be reordered.
• Those before the load can be memory ordered pass the load. Not as strong as fence.

BUT with different variable(object):
A release operation followed by a acquire operation CAN be reordered.
A acquire operation followed by a release operation CAN be reordered.
i.e
    A.store(1, std::memory_order_release);
    int b = B.load(std::memory_order_acquire);
=>
    int b = B.load(std::memory_order_acquire);
    A.store(1, std::memory_order_release);

However, keep in mind that (different variable/object) even reorder is OK for release/acquire operation, standard also depicts:
http://eel.is/c++draft/intro.multithread#intro.progress-18
An implementation should ensure that the last value (in modification order) assigned by an atomic or synchronization operation will become visible to all other threads in a finite period of time.

Reference:

https://stackoverflow.com/questions/14687703/what-c11-atomic-operations-memory-orders-guarantees-freshness

https://stackoverflow.com/questions/8819095/concurrency-atomic-and-volatile-in-c11-memory-model/8833218#8833218

i.e
If an store operation follows a for/while loop and a load operation,
the compiler shouldn't reorder load before store due to it can't
reasoning that the for/while loop is finite thus other thread can see thread store result in a finite time.
   
Reference:

1. [Preshing] Can Reordering of Release/Acquire Operations Introduce Deadlock?
2. [Bruce Dawson] In Praise of Idleness
3. [Golang bug list] cmd/compile: go1.8 regression: sync/atomic loop elided #19182

Hardware Memory Model:

Weak memory model

• ARM/Power PC

Strong memory model

• x86/64
• Sequential Consistence (software)

Reordering:

Sequentially Consistent types:

• Java: volatile variables
• C++11: atomic (default)

Explicit Compiler Barriers:
gcc:
    asm volatile("" ::: "memory");
Macro:
    #define COMPILER_BARRIER() asm volatile("" ::: "memory")

source:
https://elixir.bootlin.com/linux/latest/source/arch/x86/include/asm/barrier.h#L22

figure source: (https://www.youtube.com/watch?v=ZQFzMfHIxng)

Don't consider Sequential Consistency is SLOW; correctness is the highest priority.

Implied Compiler Barriers:
C++11:

• Every non­-relaxed atomic operation acts as a compiler barrier.
  這裡是compiler reordering, 不是memory reordering.
• Every function containing a compiler barrier must act as a compiler barrier itself, even the function is inlined.
• The majority of function calls act as compiler barriers, whether they contain their own compiler barrier or not due to it's coming from different TU. 
• But does not include inline functions, functions declared with the pure attribute, and cases where link ­time code generation is used.          

Reference:

1. [Implications of pure and constant functions] https://lwn.net/Articles/285332/

What about Golang?

goroutine act as 'user space thread', 

i.e it's runtime memory location is allocated on the heap.

Inside a single goroutine, the compiler is allowed to re-arrange

expressions as long as the reordering does not change the behavior within that goroutine as defined by the language specification. 

Within a single goroutine, the happens-before order is the order expressed by the program.

i.e

a := 42
if a == 42 {
    // always true.
}

// b will always be 42
b := a
// within other goroutines, the observe of a == 42 and c == 38 sequence is not guaranteed.
c := 38  

Reads and writes of values larger than a single machine word behave as multiple machine-word-sized operations in an unspecified order.

Initialization:

• If a package p imports package q, the completion of q's init functions happens before the start of any of p's.
• The start of the function main.main happens after all init functions have finished.

Goroutine creation:

• The go statement that starts a new goroutine happens before the goroutine's execution begins.
• i.e go statement act as a barrier function call in C/C++ which won't be reordered.

Goroutine destruction:

• Can happen any time.
• Be ware that if a goroutine that updates a global value and do nothing (i.e using channel etc.), an aggressive compiler could delete the goroutine entirely for optimization.
  i.e just modify the global variable without creating a goroutine.
  https://github.com/golang/go/issues/19182

Channel communication:

• A send on a channel happens before the corresponding receive from that channel completes.
• The closing of a channel happens before a receive that returns a zero value because the channel is closed.
• A receive from an un-buffered channel happens before the send on that channel completes.
• The kth receive on a channel with capacity C happens before the k+Cth send from that channel completes.

i.e the famous:
Do not communicate by sharing memory; instead, share memory by communicating.

No more memory store/load operation from programmer's point of view.

The Channel acted as the sequential consistence/memory fence contract.

Reference:

https://golang.org/ref/mem

https://blog.golang.org/share-memory-by-communicating

https://github.com/8l/go-learn

https://dougrichardson.org/2016/01/23/go-memory-allocations.html

[Preventing stack guard-page hopping] https://lwn.net/Articles/725832/

[Go] when'll close channel panic

Channel close lookout:

• closing a nil channel panics
• closing a closed channel panics

[Go] golang memory model - with unbuffered channel

reference:

https://golang.org/ref/mem

There is a difference between:

make(chan int) // unbuffered equals to make(chan int, 0)

with

make(chan int, 1) // buffered