What's the issue with this code


(Jameswang2015) #1

Hi, I’m learning goroutine with channel and go http.get, in the followoing code when it runs it stucks without any error or warning. what’s the issue here?

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"sync"
)

var wg sync.WaitGroup

func worker(url <-chan string){
	defer wg.Done()
	resp, err := http.Get(<-url)
	if err != nil {
		panic(err)
	}
	fmt.Println("response status:", resp.Status)
	scanner := bufio.NewScanner(resp.Body)
	for i := 0; scanner.Scan() && i < 5; i++ {
		fmt.Println(scanner.Text())
	}
	if err := scanner.Err(); err != nil {
		panic(err)
	}
}

func main() {
	wg.Add(3)
	url := make(chan string)
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	for i := 0; i < 3; i++ {
		url <- urls[i]
		go worker(url)
	}

	wg.Wait()
}

(Norbert Melzer) #2

url is unbuffered, therefore it won’t get beyond url <- urls[I] as there is no receiver on the channel.


(Jameswang2015) #3

Thanks @NobbZ!! I’m learning goroutine with channel, and I noticed that for unbuffered channel, it has to be used (either sending to channel or receiving from channel) before goroutine in the main function. (agreed?) but when I code I just forgot this rule - thanks for pointing this out! I’m just wondering why this code doesn’t report “dead lock” error as usual instead just freezing - is this because I’m using http.get staff?

I also updated the code as below and now it’s working - let me know if there is other way to do this. Thanks!

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"sync"
)

var wg sync.WaitGroup

func worker(url <-chan string){
	defer wg.Done()
	resp, err := http.Get(<-url)
	if err != nil {
		panic(err)
	}
	fmt.Println("response status:", resp.Status)
	scanner := bufio.NewScanner(resp.Body)
	for i := 0; scanner.Scan() && i < 5; i++ {
		fmt.Println(scanner.Text())
	}
	if err := scanner.Err(); err != nil {
		panic(err)
	}
}

func main() {
	wg.Add(3)
	url := make(chan string)
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	for i := 0; i < 3; i++ {
		go worker(url)
	}
	for i := 0; i < 3; i++ {
		url <- urls[i]
	}
	wg.Wait()
}

(Holloway) #4

This is one weird way of doing things. When passing a channel, use `func worker(url chan string) instead.

You should use internal timeout when you perform a network request. This is important because you can never know what kind of error you will bump into and your application should be independent enough to call it off. One good example:

client := http.Client{
    Timeout: 5 * time.Second,
}
resp, err := client.Get(url)
...

Your concurrency is unclear and unplanned. Otherwise, you won’t be using global variable. You can try a few things:

  1. Perform the http.Client.Get without concurrency. Do everything on main first.
  2. Still maintaining only in main, abstract the http.Client.Get into its own function.
  3. Plan your concurrency (Why I receive an intermittent deadlock error?).

Remember, if you can’t handle it in 1 process, don’t bother multiplying in concurrency.


(Jameswang2015) #5

func worker(url <-chan string) is to define a channel that only receive value from, it’s like defining a more strickly type of channel.

I’m not very clear of using the internal timeout (why we need it in this case?) and planning concurrency. can you make a example for this?


(Holloway) #6

Say the server serving a request of 120 seconds (1 minute) and your application is only allowed maximum (60 seconds) per request, you are going to use up all your time budget just for waiting.

Which is why I would not recommend for now (unless you mastered your concurrency and have a strong control over your logic). It’s best to pass in a channel so that you can have full control over waiting/reading.

Have you check out the link? There has an clear example on how to plan concurrency. If you plan your concurrency contexts clearly, you won’t be using a global variable at all, especially mutex lock.


(Holloway) #7

In case you got lost, try to do this one step at a time and post your codes. This time, you’ll be the one coding. :grin:


(Jameswang2015) #8

here is the step 1

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"time"
)

func main() {
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	client := http.Client{
		Timeout: 5 * time.Second,
	}
	for i := 0; i < len(urls); i++ {
		received, err := client.Get(urls[i])
		if err != nil {
			panic(err)
		}
		fmt.Println("status: ", received.Status)
		scanner := bufio.NewScanner(received.Body)
		for i := 0; scanner.Scan() && i < 5; i++ {
			fmt.Println(scanner.Text())
		}
		if err := scanner.Err(); err != nil {
			panic(err)
		}
	}
}

(Jameswang2015) #9

here is the step 2:

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"time"
)

func httpCall(url string) {
	client := http.Client{
		Timeout: 5 * time.Second,
	}
	received, err := client.Get(url)
	if err != nil {
		panic(err)
	}
	defer received.Body.Close()

	fmt.Println("status: ", received.StatusCode)
	scanner := bufio.NewScanner(received.Body)
	for i := 0; scanner.Scan() && i < 5; i++ {
		fmt.Println(scanner.Text())
	}
	if err := scanner.Err(); err != nil {
		panic(err)
	}
}
func main() {
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	for i := 0; i < len(urls); i++ {
		httpCall(urls[i])
	}
}

(Holloway) #10

Had you run it? I’m getting:

main.go:22:30: bodyclose: response body must be closed (bodyclose)             
                received, err := client.Get(urls[i])  

You need to close the Body. Otherwise, there will be a resource leak and client may not be able to re-use a persistent TCP connection to the server for a subsequent “keep-alive”. To close it, use the following after the get:

defer received.Body.Close()

NOTE:
You’re already at Step 2.


Before moving to step 3, make sure you do all the refactoring as clean as possible.


(Jameswang2015) #11

I did run step 1 and it run successfully. my go verison is go1.12.9


(Holloway) #12

Likely due to my golang-linters. Heed the warning: https://stackoverflow.com/questions/33238518/what-could-happen-if-i-dont-close-response-body-in-golang


(Jameswang2015) #13

here is updated step 1:

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"time"
)

func main() {
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	client := http.Client{
		Timeout: 5 * time.Second,
	}
	for i := 0; i < len(urls); i++ {
		received, err := client.Get(urls[i])
		if err != nil {
			panic(err)
		}
		defer received.Body.Close()
		fmt.Println("status: ", received.Status)
		scanner := bufio.NewScanner(received.Body)
		for i := 0; scanner.Scan() && i < 5; i++ {
			fmt.Println(scanner.Text())
		}
		if err := scanner.Err(); err != nil {
			panic(err)
		}
	}
}

is that right to put the defer in the for loop? I get a warning “possible resource leak”


(Holloway) #14

Makes no sense to use defer in for loop. You need to close it manually using received.Body.Close() without defer keyword. E.g.:

  1. before all panic(err)
  2. right at the end of the first loop (for i := 0; i < len(urls); i++), do another one.

How to find out?

Look for all the loop’s exit points (e.g. panic is an exit point), end of loop is another exit point.


(Jameswang2015) #15

here is updated step 1 again:

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"time"
)

func main() {
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	client := http.Client{
		Timeout: 5 * time.Second,
	}
	for i := 0; i < len(urls); i++ {
		received, err := client.Get(urls[i])
		if err != nil {
			received.Body.Close()
			panic(err)
		}
		fmt.Println("status: ", received.Status)
		scanner := bufio.NewScanner(received.Body)
		for i := 0; scanner.Scan() && i < 5; i++ {
			fmt.Println(scanner.Text())
		}
		if err := scanner.Err(); err != nil {
			received.Body.Close()
			panic(err)
		}
		received.Body.Close()
	}
}

(Holloway) #16

Excellent.

Now are you confident with your step 2? You can proceed to step 3. Start with a plan. We can review it before writing into codes.

NOTE:
If you need a little bit more challenge, you can try to abstract all the fmt printings into a function, before step 3. This complicates the existing step 2 in exchange some practical debugging experience and takes longer time to learn. :grin:


(Jameswang2015) #17

for step 2, I think the defer received.Body.Close() is ok since it’s not in a for loop.

it’s very late here I will do the step 3 tomorrow.


(Holloway) #18

Noted.


(Jameswang2015) #19

my plan is as bellow.

  • There are 5 jobs, each is a http.get call to one website and get the first 5 rows of that website(body html). Each job is assigned to a goroutine with a receiving-only channel as input and output is to print the first 5 rows of html.
    • the input channel is string of url
    • use unbuffered channel
  • There are 3 workers, each worker is a goroutine
  • work flow:
    • in main, define channel url, then three works, followed by 5 jobs, and then close channel
    • in worker goroutine, for receivedURL := range url to keep receiving job via channel, and then do the http.Client.Get job. if channel closes, then go routine stop working and notify main by defer wg.Done()
    • back to main, sync.WaitGroup is added with 3 - the number of workers, and wg.Wait at the end to block the main until all the workers get all jobs done and notify main.

here is my code in this plan:

package main

import (
	"bufio"
	"fmt"
	"net/http"
	"sync"
	"time"
)

const (
	numberOfWorkers int = 3
	numberOfTasks int = 5
)

var wg sync.WaitGroup

func worker(id int, url <-chan string){
	defer wg.Done()
	client := http.Client{
		Timeout: 5 * time.Second,
	}
	for receivedURL := range url{
		fmt.Printf("worker %d start on %s\n", id, receivedURL)
		resp, err := client.Get(receivedURL)
		if err != nil {
			resp.Body.Close()
			panic(err)
		}
		fmt.Println("response status:", resp.Status)
		scanner := bufio.NewScanner(resp.Body)
		for i := 0; scanner.Scan() && i < 5; i++ {
			fmt.Println(scanner.Text())
		}
		if err := scanner.Err(); err != nil {
			resp.Body.Close()
			panic(err)
		}
		fmt.Printf("worker %d Done with %s\n\n\n\n", id, receivedURL)
		resp.Body.Close()
	}

}

func main() {
	wg.Add(numberOfWorkers)
	url := make(chan string)
	urls := []string{
		"http://gobyexample.com",
		"http://att.com",
		"http://domaintools.com",
		"http://microsoft.com",
		"http://google.com",
	}
	for i := 0; i < 3; i++ {
		go worker(i, url)
	}
	for i := 0; i < numberOfTasks; i++ {
		url <- urls[i]
	}
	close(url)
	wg.Wait()
}

It looks like this code is not stable - sometimes it runs smoothly but sometimes I get this error

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x40 pc=0x122eaa2]

goroutine 5 [running]:
main.worker(0x1, 0xc0000c00c0)
        /Users/zwang/GolandProjects/learngo/cmd/cmd1/httpRoutine2-6.go:28 +0x5c2
created by main.main
        /Users/zwang/GolandProjects/learngo/cmd/cmd1/httpRoutine2-6.go:57 +0xb5
exit status 2

(Holloway) #20

In planning, you do not charge yourself into codes. This will bump into non-meaningful problems that makes others hard to answer for you.

A proper plan has very clear contact switching, and explanation on WHO is doing WHAT at WHEN.
It look something like this in the end:

It can be easily translated into codes without much alterations.

One coding Example
package main

import (
	"fmt"
	"sync"
)

const (
	maxEating = 2
	fullnessLimit = 2
)

// Chopsticks is an object structure that serves as a tool to eat
type Chopsticks struct {
	mutex  sync.RWMutex
	ID     int
	isUsed bool
}

// NewChopsticks creates a Chopsticks object with the given id label.
func NewChopsticks(id int) *Chopsticks {
	return &Chopsticks{
		mutex:  sync.RWMutex{},
		ID:     id,
		isUsed: false,
	}
}

// Use is to set the Chopsticks status to "In-Use"
func (c *Chopsticks) Use() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.isUsed = true
}

// Free is to set the Chopsticks status to "Free"
func (c *Chopsticks) Free() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.isUsed = false
}

// IsInUse is to check the Chopsticks status, currently "In-Use" or "Free".
func (c *Chopsticks) IsInUse() bool {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	return c.isUsed
}

// Table is the structure serving foods and chopsticks
type Table struct {
	mutex      sync.RWMutex
	isEating   uint
	chopsticks []*Chopsticks
}

// NewTable creates the table object with person quantity.
func NewTable(quantity uint) *Table {
	t := &Table{
		mutex:      sync.RWMutex{},
		isEating:   0,
		chopsticks: []*Chopsticks{},
	}
	for i := 0; i < int(quantity); i++ {
		t.chopsticks = append(t.chopsticks, NewChopsticks(i))
	}

	return t
}

// RequestChopsticks is to allows a customer to eat using an available
// Chopsticks.
func (t *Table) RequestChopsticks() *Chopsticks {
	t.mutex.Lock()
	defer t.mutex.Unlock()

	// table is full
	if t.isEating >= maxEating {
		return nil
	}

	// permit to eat. Scan for available chopsticks
	c := t.seekChopsticks()
	c.Use()
	t.isEating++
	return c
}

func (t *Table) seekChopsticks() *Chopsticks {
	// NOTE: here, you can use random. I will use FIFO instead.
	for _, c := range t.chopsticks {
		if !c.IsInUse() {
			return c
		}
	}
	return nil
}

// ReturnChopsticks is to allow a customer to place back chopsticks when
// he/she is done eating
func (t *Table) ReturnChopsticks(c *Chopsticks) {
	t.mutex.Lock()
	defer t.mutex.Unlock()
	t.isEating--
	c.Free()
}

func philosopher(id int, fullness chan int, table *Table) {
	eatCount := fullnessLimit

	for {
		chopsticks := table.RequestChopsticks()
		if chopsticks == nil {
			continue
		}

		// start eating
		fmt.Printf("P%d: START eating with chopstick %d.\n",
			id,
			chopsticks.ID)

		// eating
		eatCount--

		// stop eating
		fmt.Printf("P%d: FINISH eating with chopstick %d.\n",
			id,
			chopsticks.ID)
		table.ReturnChopsticks(chopsticks)

		// check fullness
		if eatCount == 0 {
			fullness <- id
			return
		}

	}
}

func main() {
	headcount := 5

	t := NewTable(uint(headcount))
	c1 := make(chan int)
	fullness := 0

	for i := 0; i < headcount; i++ {
		go philosopher(i, c1, t)
	}

	// Wait for fullness
	for {
		select {
		case person := <-c1:
			fullness++
			fmt.Printf("Philosopher %d is full\n", person)
			if fullness == headcount {
				fmt.Printf("All are full.\n[ ENDED ]\n")
				return
			}
		}
	}
}

// Output:
// P4: START eating with chopstick 0.
// P4: FINISH eating with chopstick 0.
// P4: START eating with chopstick 0.
// P4: FINISH eating with chopstick 0.
// Philosopher 4 is full
// P0: START eating with chopstick 0.
// P0: FINISH eating with chopstick 0.
// P0: START eating with chopstick 0.
// P0: FINISH eating with chopstick 0.
// Philosopher 0 is full
// P3: START eating with chopstick 0.
// P2: START eating with chopstick 1.
// P2: FINISH eating with chopstick 1.
// P3: FINISH eating with chopstick 0.
// P3: START eating with chopstick 0.
// P3: FINISH eating with chopstick 0.
// Philosopher 3 is full
// P2: START eating with chopstick 1.
// P2: FINISH eating with chopstick 1.
// Philosopher 2 is full
// P1: START eating with chopstick 0.
// P1: FINISH eating with chopstick 0.
// P1: START eating with chopstick 0.
// P1: FINISH eating with chopstick 0.
// Philosopher 1 is full
// All are full.
// [ ENDED ]


Right now, it is your plan 1. It’s time to refactor the plan until you reach the clarity as the proper plan.

Do not worry, even we do this on daily basis. See this link on how I transform a 5 philosophers chopstick solution plan from Plan 1 to Plan 4: Planning Example.