Concurrent Clock Server#

Let's implement a concurrent clock server.

|--client
   |--client.go
|--server
   |--server.go
go.mod
main.go

Write a simple TCP server listening on port 8080.

// haimtran 19/05/2024
// concurrent clock server


package main


import (
	"fmt"
	"io"
	"log"
	"net"
	"time"
)


func main() {
	fmt.Println("Hello World")


	// tcp listener
	listener, err := net.Listen("tcp", ":8080")


	if err != nil {
    log.Print(err)
	}


	// infinite loop to handle request
	for {
		conn, err := listener.Accept()
		if err != nil {
      log.Print(err)
		}


    // each request handled by a goroutine
		go handle(conn)
	}
}


func handle(conn net.Conn) {


	defer conn.Close()


	for {
		_, error := io.WriteString(conn, time.Now().Format("15:04:05\n"))
		if error != nil {
      log.Print(error)
      return
		}
		time.Sleep(2 * time.Second)
	}
}

Write a simple client dial on port 8080.

package main


import (
	"io"
	"log"
	"net"
	"os"
)


func main() {


	conn, err := net.Dial("tcp", "localhost:8080")


	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()


	mustCopy(os.Stdout, conn)


}


func mustCopy(dst io.Writer, src io.Reader) {


	if _, err := io.Copy(dst, src); err != nil {
		log.Fatal(err)
	}
}

Alternatively, we can use netcat.

nc localhost 8080

Unbuffered Channel#

Quote from the book the go programming.

A send operation on an unbuffered channel blocks the sending goroutine until another goroutine executes a corresponding receive on the same channel, at which point the value is transmitted and both goroutines is blocked until another goroutine performs a send on the same channel.

func helloUnbufferChannel() {
	// create a channel of integer
	ch := make(chan int)
	// goroutine write to ch
	go func() {
		ch <- 42
		ch <- 43
		ch <- 44
	}()
	// wait until the write channel complete
	fmt.Println(<-ch)
	fmt.Println(<-ch)
	fmt.Println(<-ch)
	// close channel
}

Use close to close a channel.

func helloChannel() {
	// create an unbuffered channel
	c := make(chan int)
	go func() {
		// time add delay of 2 seconds
		time.Sleep(2 * time.Second)
		fmt.Println("Hello from goroutine")
		// write to channel
		c <- 42
		//
		c <- 100
		// close channel
		close(c)
	}()
	// wait until the write channel complete
	for v := range c {
		fmt.Println(v)
	}
}

Buffered Channel#

Quoted from the go programming.

A buffered channel has a queue of elements. The queue's maximum size is determined when it is created, by the capacity argument to make... The channel buffer decouples the sending and receiving goroutines.

func ImageFile(infile string) (string, error) {


	time.Sleep(2 * time.Second)
	fmt.Println("infile", infile)
	return infile, nil
}


// goroutine leak and buffered channel
// page 237 the go programming
func makeThumbnails5(filenames []string) (thumbfiles []string, err error) {


	type item struct {
		thumbfile string
		err       error
	}


	ch := make(chan item, len(filenames))


	for _, f := range filenames {
		go func(f string) {
			var it item
			it.thumbfile, it.err = ImageFile(f)
			ch <- it
		}(f)
	}


	for range filenames {
		it := <-ch
		if it.err != nil {
			return nil, it.err
		}
		thumbfiles = append(thumbfiles, it.thumbfile)
	}


	return thumbfiles, nil
}

Wait Group#

// unpredictable number of filnames
// use channel for filenames
func makeThumbnails6(filenames <-chan string) int64 {


	sizes := make(chan int64)


	var wg sync.WaitGroup // number of goroutines to wait for


	for f := range filenames {
		wg.Add(1)
		// worker goroutine
		go func(f string) {
			defer wg.Done()
			thumb, err := ImageFile(f)
			if err != nil {
				return
			}
			info, _ := os.Stat(thumb)
			sizes <- info.Size()
		}(f)
	}


	// closer
	go func() {
		wg.Wait()
		close(sizes)
	}()


	//
	var total int64


	for size := range sizes {
		total += size
	}
	return total
}
main.go
// haimtran 19/05/2024
// channel and goroutines


package main


import (
	"fmt"
	"os"
	"sync"
	"time"
)


// unbuffered channel and synch
func helloChannel() {
	// create an unbuffered channel
	c := make(chan int)
	go func() {
		// time add delay of 2 seconds
		time.Sleep(2 * time.Second)
		fmt.Println("Hello from goroutine")
		// write to channel
		c <- 42
		//
		c <- 100
		// close channel
		close(c)
	}()
	// wait until the write channel complete
	for v := range c {
		fmt.Println(v)
	}
}


//


func helloUnbufferChannel() {
	// create a channel of integer
	ch := make(chan int)
	// goroutine write to ch
	go func() {
		ch <- 42
		ch <- 43
		ch <- 44
	}()
	// wait until the write channel complete
	fmt.Println(<-ch)
	fmt.Println(<-ch)
	fmt.Println(<-ch)
	// close channel
}


func ImageFile(infile string) (string, error) {


	time.Sleep(2 * time.Second)
	fmt.Println("infile", infile)
	return infile, nil
}


// goroutine leak and buffered channel
// page 237 the go programming
func makeThumbnails5(filenames []string) (thumbfiles []string, err error) {


	type item struct {
		thumbfile string
		err       error
	}


	ch := make(chan item, len(filenames))


	for _, f := range filenames {
		go func(f string) {
			var it item
			it.thumbfile, it.err = ImageFile(f)
			ch <- it
		}(f)
	}


	for range filenames {
		it := <-ch
		if it.err != nil {
			return nil, it.err
		}
		thumbfiles = append(thumbfiles, it.thumbfile)
	}


	return thumbfiles, nil
}




// unpredictable number of filnames
// use channel for filenames
func makeThumbnails6(filenames <-chan string) int64 {


	sizes := make(chan int64)


	var wg sync.WaitGroup // number of goroutines to wait for


	for f := range filenames {
		wg.Add(1)
		// worker goroutine
		go func(f string) {
			defer wg.Done()
			thumb, err := ImageFile(f)
			if err != nil {
				return
			}
			info, _ := os.Stat(thumb)
			sizes <- info.Size()
		}(f)
	}


	// closer
	go func() {
		wg.Wait()
		close(sizes)
	}()


	//
	var total int64


	for size := range sizes {
		total += size
	}
	return total
}


func main() {


	filenames := []string{"a.jpg", "b.jpg", "c.jpg"}


	fmt.Println("Hello")
	// hello channel
	// helloChannel()
	// helloUnbufferChannel()


	makeThumbnails5(filenames)
}