Concurrency and Indeterminate Buffered Asynchronous Channels and time.Sleep()
YardVox
As a novice 'Go' experimenter I can not get Buffered Asynchronous
Channels to
work without using the time.Sleep function.
When I remove the time.sleep function from my code Goroutines/
processes fail to hook up.
Is this the norm or is there a proper design pattern for this.
Robert Sexmith
chanleng.go follows:
package main
import (
"fmt"
"time"
"os"
"strconv"
)
func f(zcnt int, fchan chan<- string) {
var outstring string
print("f_", zcnt, "\n")
outstring = "OutString_" + strconv.Itoa(zcnt)
fchan <- outstring
}
func fout(fchan <-chan string) {
var zflag bool
for {
if len(fchan) > 0 {
fmt.Println(<-fchan)
zflag = true
} else {
time.Sleep(1)
if len(fchan) == 0 && zflag == true {
os.Exit(0)
}
}
}
}
func main() {
var fchan = make(chan string, 25)
go fout(fchan)
for i := 1; i < 26; i++ {
go f(i, fchan)
}
print("forever loop\n")
for {
time.Sleep(1)
}
}
Ian Lance Taylor
> As a novice 'Go' experimenter I can not get Buffered Asynchronous
> Channels to
> work without using the time.Sleep function.
> When I remove the time.sleep function from my code Goroutines/
> processes fail to hook up.
>
> Is this the norm or is there a proper design pattern for this.
You should not have to sleep to make channels work.
> func fout(fchan <-chan string) {
> var zflag bool
> for {
> if len(fchan) > 0 {
> fmt.Println(<-fchan)
> zflag = true
> } else {
> time.Sleep(1)
> if len(fchan) == 0 && zflag == true {
> os.Exit(0)
> }
> }
> }
> }
Calling len on a channel and using that to make a decision tends to
indicate a race condition. The state of the channel can change
between the time you call len and the time you use the result of len.
You should instead just receive a value. Use a nonblocking receive if
appropriate.
Indicate end-of-data on a channel by sending a marker value or by
calling close().
Ian
YardVox
I modified the func fout -- see below:
to receive a value via
the idiom receiveval, ok = <- fchan.
instead of getting the length of the channel, testing length>0 then
action ...
Still requires the time.Sleep(1) in func fout() to run.
Somehow the funct fout -- forever loop is locking/block the fchan
channel
and time.Sleep(1) frees the lock/block.
Ian also points out a technique to handle end of channel data
without using os.Exit(0).
My example is a concept test where I force my program to exit.
What I am really experimenting with is with continuous communications
that has no end of channel data. Example a continuous data feed
from some sensor on some weather station or the like.
Robert Sexmith
Modified func fout() follows:
func fout(fchan <-chan string) {
var ok, zflag bool
var dastring string
for {
time.Sleep(1)
dastring, ok = <-fchan
if ok {
fmt.Println(dastring, ok)
zflag = true
} else {
if zflag {
os.Exit(0)
}
}
}
}
On Dec 22, 12:25 am, Ian Lance Taylor <i...@google.com> wrote:
Russ Cox
> time.Sleep(1)
> dastring, ok = <-fchan
> if ok {
> fmt.Println(dastring, ok)
> zflag = true
> } else {
> if zflag {
> os.Exit(0)
> }
> }
> }
There should be no need to write loops like this:
reading from fchan will block until a new value is ready.
If the code needs to do something else concurrently with
reading from fchan, it can use another goroutine.
If that activity has to coordinate with the read from
fchan, it can talk to the fchan reader using a channel.
(The fchan reader would then select on both channels.)
Russ
YardVox
Per Russ's suggestion that "reading from fchan will block until a new
value is ready"
I rewrote func fout()
func fout(fchan <-chan string) {
for {
fmt.Println(<-fchan)
}
}
This code runs, does not require time.Sleep() inside func fout() and
handles a continuous data feed on a channel
within the role of a server that should run nonstop forever. -- But
it runs forever and does not care when
the 26 indeterminate goroutines finished. For my test code -- To force
an exit I could within func main -- sleep for time
or within func fout -- I could use the same loop count I used to
spawn the goroutines.
Question: So why did I write the loops like I did?
Answer: I wanted to detect when the 26 indeterminate goroutines
completed.
First look at the "Indeterminate" output of this code using
runtime.GOMAXPROCS(8) on an 8 core Intel platform with 16gig memory.
Output follows:
f_ 4
f_ 3
f_ 2
f_ 5
f_ 18
f_ 17
f_ 19
f_ 22
f_ 23
f_ 24
f_ 25
f_ 21
f_ 6
f_ 15
f_ 16
f_ 13
f_ 14
f_ 20
f_ 1
f_ 7
f_ 10
OutString_19
f_ 8
OutString_22
Inside Main -- post launch of all go func()s
OutString_23
f_ 11
OutString_24
f_ 9
f_ 12
OutString_25
OutString_17
OutString_18
OutString_4
OutString_5
OutString_2
OutString_3
OutString_14
OutString_13
OutString_16
OutString_15
OutString_6
OutString_21
OutString_1
OutString_7
OutString_10
OutString_8
OutString_20
OutString_11
OutString_9
OutString_12
So how do I detect when the 26 indeterminate goroutines completed?
If I put an ExitNow string on fchan from my last func f(25)
then I would have halted the output prematurely -- see output above.
I suppose I could use some data structure or a map to track the
completion
of each goroutine.
but instead:
I wanted to try and follow the Effective Go principal of
"Do not communicate by sharing memory. Instead, share memory by
communicating"
I wanted to avoid shared memory variables and only use only the
channel
to detect when all 26 indeterminate process completed.
Thus my loops once: using len(chan) and once: using ok==true on a
channel read.
Go Channels are powerful. Go Channels are different than Erlang.
I am trying to learn how to program using channels or better yet
I am trying to learn how to program "Concurrently" using channels.
Russ --
You should have insights into GO I will never possess.
Thus what insights/techniques supported with example code
could you provide in answering the question
How does one know when X indeterminate goroutines finish?
Robert Sexmith "Old but still willing to learn how to think in GO"
ps I like the language alot -- kinda like a concurrent Perl with a
"type nazi" looking over your shoulder
Ian Lance Taylor
> This code runs, does not require time.Sleep() inside func fout() and
> handles a continuous data feed on a channel
> within the role of a server that should run nonstop forever. -- But
> it runs forever and does not care when
> the 26 indeterminate goroutines finished. For my test code -- To force
> an exit I could within func main -- sleep for time
> or within func fout -- I could use the same loop count I used to
> spawn the goroutines.
>
> Question: So why did I write the loops like I did?
> Answer: I wanted to detect when the 26 indeterminate goroutines
> completed.
The answer to this question depends upon your algorithm: specifically,
how you know that the goroutines are finished. Do you want to wait
until they are all certainly finished? In that case you need to have
a counter somewhere, though there are various different places that it
could be. Otherwise, you won't be able to tell whether one is just
taking a long time to produce an answer. Or, do you want to wait
until no answer is available for a second? In that case, use a select
statement with a heartbeat from another channel.
Ian
YardVox
I would like to see your code sample on using a select
statement with a heartbeat from another channel as a way to detect
when
X indeterminate goroutines are completed.
Regarding an algorithm to detect/warrant when X indeterminate
goroutines are completed --
the only approach that quickly came to mind is the old fashion batch
control model.
A batch control clerk prepares a batch control total of all inputs
then submits
the batch job to the computer. The computer runs and produces a
runtime batch control
of what it processed. Then a second batch control clerk audits the
input batch control total
against the runtime batch control total to assure that they tie out
(ie agree).
The code for this follows; but, I have got to believe that you or Russ
or some computer science major
reading this could come up with a better algorithm.
package main
import (
"fmt"
"time"
"os"
"strconv"
"runtime"
)
const Maxpids int = 100
type P struct {
outstring string
go_id int
}
type A struct {
cmd string
fpid int
}
func audit(cop <-chan A) {
var copin A
var pidin [Maxpids]int
var pidout [Maxpids]int
var sum_in, sum_out int
for {
copin = <-cop
switch copin.cmd {
case "fin":
pidin[copin.fpid] = 1
pidout[copin.fpid] = 0
case "fout":
pidout[copin.fpid] = 1
default:
fmt.Println("BadCaseFallThru")
}
sum_in = 0
sum_out = 0
for _, v := range pidin {
sum_in += v
}
// fmt.Println(sum_in)
for _, k := range pidout {
sum_out += k
}
// fmt.Println(sum_out)
if sum_in == sum_out {
fmt.Println("ControlTotalsMatch")
os.Exit(0)
}
}
}
func f(zpid int, fchan chan<- P, cop chan<- A) {
var pout P
var copout A
fmt.Println("f_", zpid)
pout.outstring = "OutString_" + strconv.Itoa(zpid)
pout.go_id = zpid
copout.cmd = "fin"
copout.fpid = zpid
fchan <- pout
cop <- copout
}
func fout(fchan <-chan P, cop chan<- A) {
var zin P
var copout A
for {
zin = <-fchan
fmt.Println(zin.outstring)
copout.cmd = "fout"
copout.fpid = zin.go_id
cop <- copout
}
}
func main() {
runtime.GOMAXPROCS(8)
var fchan = make(chan P, 25)
var cop = make(chan A, 25)
go audit(cop)
go fout(fchan, cop)
for i := 1; i < 26; i++ {
go f(i, fchan, cop)
}
fmt.Println("Inside Main -- post launch of all go func()s")
for {
time.Sleep(1)
}
}
On Dec 23, 7:58 pm, Ian Lance Taylor <i...@google.com> wrote:
⚛
On Dec 24, 1:41 am, YardVox <yard...@gmail.com> wrote:
> Question: So why did I write the loops like I did?
> Answer: I wanted to detect when the 26 indeterminate goroutines
> completed.
I don't think you understand the programming patterns for use with Go-
routines.
The code "ok = <-chan" is more-or-less semantically equivalent to the
following pseudo-code (modulo locking):
function blocking_read_from_a_channel(Channel c) {
while(c.is_there_a_message()) do {/*nothing*/}
return c.get_and_remove_the_first_message();
}
So, in order to detect something you need NOT put the read into
another loop. It does not make any sense.
Look for articles about "discrete event simulation" and/or
"communicating sequential processes".
⚛
> function blocking_read_from_a_channel(Channel c) {
> while(c.is_there_a_message()) do {/*nothing*/}
> return c.get_and_remove_the_first_message();
> }
Correction:
function blocking_read_from_a_channel(Channel c) {
while(c.there_is_no_message()) do {/*nothing*/}
return c.get_and_remove_the_first_message();
}
SnakE
As a novice 'Go' experimenter I can not get Buffered Asynchronous
Channels to
work without using the time.Sleep function.
If I understand your basic problem correctly, you have many workers writing to a single channel, and you need to know when these workers are done. I think the best you can do is to count responses.
Here's your first example, fixed to use response counting. I'm not using your later examples because they're much more complex and harder to understand.
package main
import (
"fmt"
)
func f(zcnt int, fchan chan<- string) {
var outstring string
print("f_", zcnt, "\n")
outstring = "OutString_" + strconv.Itoa(zcnt)
fchan <- outstring
}
for ; count > 0; count-- {
fmt.Println(<-fchan)
}
echan <- true
}
func main() {
var fchan = make(chan string, 25)
go fout(fchan, echan, 25)
for i := 1; i < 26; i++ {
go f(i, fchan)
}
print("done.\n")
}
Ian Lance Taylor
> I would like to see your code sample on using a select
> statement with a heartbeat from another channel as a way to detect
> when
> X indeterminate goroutines are completed.
Have all the goroutines write into one channel c1. Then do something
like:
c2 := make(chan bool)
go func() { time.Sleep(10 * 1e9); c2 <- true }()
L: for {
select {
case v := <-c1:
// Do something with v.
case <-c2:
break L
}
}
> Regarding an algorithm to detect/warrant when X indeterminate
> goroutines are completed --
> the only approach that quickly came to mind is the old fashion batch
> control model.
> A batch control clerk prepares a batch control total of all inputs
> then submits
> the batch job to the computer. The computer runs and produces a
> runtime batch control
> of what it processed. Then a second batch control clerk audits the
> input batch control total
> against the runtime batch control total to assure that they tie out
> (ie agree).
Another approach would be to increment a count as you start them and
decrement it as you stop them. When the count gets to zero, you are
done.
Ian
SnakE
> Regarding an algorithm to detect/warrant when X indeterminate> goroutines are completed --Another approach would be to increment a count as you start them and
> the only approach that quickly came to mind is the old fashion batch
> control model.
> A batch control clerk prepares a batch control total of all inputs
> then submits
> the batch job to the computer. The computer runs and produces a
> runtime batch control
> of what it processed. Then a second batch control clerk audits the
> input batch control total
> against the runtime batch control total to assure that they tie out
> (ie agree).
decrement it as you stop them. When the count gets to zero, you are
done.
This won't work if they stop faster than you can start them. Or if you check for completeness before any worker is started.
Qtvali
> 2009/12/28 Ian Lance Taylor <i...@google.com>
> This won't work if they stop faster than you can start them. Or if you
> check for completeness before any worker is started.
You can do:
count += 7;
go 7 workers.