simple multi goroutine dispatcher/broadcaster

README.md

multi-goroutine-control

A simple example to test synchronizing multiple goroutines.

This application has 4 goroutines:

• a ticker goroutine, sending data every 2 seconds to a monitor goroutine
• a monitoring goroutine that waits on a data channel, displaying its content when there is new data
• a user-input goroutine, waiting on stdin for user input, which is then sent to the control goroutine
• a control goroutine that waits on a command channel, taking actions according to the received command ("d": display data, "t" display elapsed time", "w": quit)

Example

$> go run main.go
2016/03/03 17:53:15 mon: 2016-03-03 16:53:15.124840889 +0000 UTC
d
2016/03/03 17:53:16 cmd: "d"
2016/03/03 17:53:16 today is 2016-03-03 17:53:16.865216802 +0100 CET
2016/03/03 17:53:17 mon: 2016-03-03 16:53:17.124858925 +0000 UTC
d
2016/03/03 17:53:17 cmd: "d"
2016/03/03 17:53:17 today is 2016-03-03 17:53:17.409236028 +0100 CET
t
2016/03/03 17:53:18 cmd: "t"
2016/03/03 17:53:18 elapsed: 5.05254336s
2016/03/03 17:53:19 mon: 2016-03-03 16:53:19.124846239 +0000 UTC
d
2016/03/03 17:53:19 cmd: "d"
2016/03/03 17:53:19 today is 2016-03-03 17:53:19.713227 +0100 CET
2016/03/03 17:53:21 mon: 2016-03-03 16:53:21.124851038 +0000 UTC
t
2016/03/03 17:53:21 cmd: "t"
2016/03/03 17:53:21 elapsed: 8.284506582s
2016/03/03 17:53:23 mon: 2016-03-03 16:53:23.124853101 +0000 UTC
2016/03/03 17:53:25 mon: 2016-03-03 16:53:25.124852105 +0000 UTC
2016/03/03 17:53:27 mon: 2016-03-03 16:53:27.124854667 +0000 UTC
q
2016/03/03 17:53:28 cmd: "q"
2016/03/03 17:53:28 goroutine-control exiting...
2016/03/03 17:53:28 goroutine-monitor exiting...
2016/03/03 17:53:28 goroutine-usr-input exiting...
2016/03/03 17:53:28 scanner-goroutine exiting...
2016/03/03 17:53:28 ticker exiting...
bye.

multi-goroutine-control.go

package main

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"os"
	"sync"
	"time"
)

const N = 4 // number of goroutines

var wg sync.WaitGroup

func main() {
	usrc := make(chan string)
	monc := make(chan string)
	done := make(chan int)

	wg.Add(N)

	go userInput(done, usrc)
	go monitor(done, monc)

	go func() {
		tck := time.NewTicker(2 * time.Second)
		defer tck.Stop()
		defer wg.Done()
		for {
			select {
			case t := <-tck.C:
				monc <- fmt.Sprintf("%v", t.UTC())
			case <-done:
				log.Printf("ticker exiting...\n")
				return
			}
		}
	}()

	go control(done, usrc)

	wg.Wait()
	fmt.Printf("bye.\n")
}

func userInput(done chan int, usrc chan string) {
	pr, pw := io.Pipe()
	defer pw.Close()
	defer wg.Done()

	go func() {
		for {
			_, err := io.Copy(pw, os.Stdin)
			if err != nil {
				log.Printf("i/o error: %v\n", err)
				break
			}
		}
		log.Printf("pipe-goroutine exiting...\n")
	}()

	go func() {
		scan := bufio.NewScanner(pr)
		for scan.Scan() {
			err := scan.Err()
			if err != nil {
				log.Printf("error reading from stdin: %v\n", err)
				break
			}

			usrc <- scan.Text()
		}
		log.Printf("scanner-goroutine exiting...\n")
	}()

	for {
		select {
		case <-done:
			log.Printf("goroutine-usr-input exiting...\n")
			var err error
			err = pw.CloseWithError(io.EOF)
			if err != nil {
				log.Printf("error closing pipe-writer: %v\n", err)
			}
			err = pr.CloseWithError(io.EOF)
			if err != nil {
				log.Printf("error closing pipe-reader: %v\n", err)
			}
			return
		}
	}
}

func control(done chan int, usrc chan string) {
	defer wg.Done()
	start := time.Now()
	for {
		select {
		case <-done:
			log.Printf("goroutine-control exiting...\n")
			return
		case cmd := <-usrc:
			log.Printf("cmd: %q\n", cmd)
			switch cmd {
			case "quit", "q":
				close(done)
			case "date", "d":
				log.Printf("today is %v\n", time.Now())
			case "time", "t":
				log.Printf("elapsed: %v\n", time.Since(start))

			default:
				log.Printf("unknown command %q\n", cmd)
			}
		}
	}
}

func monitor(done chan int, monc chan string) {
	defer wg.Done()
	for {
		select {
		case <-done:
			log.Printf("goroutine-monitor exiting...\n")
			return
		case mon := <-monc:
			log.Printf("mon: %v\n", mon)
		}
	}
}