Golang 自学之路（三）：并发与网络

Go Golang

---

高级 Go

Goroutines

• runtime.Gosched()
• runtime.Goexit()

An experiential rule of thumb seems to be that for n cores setting GOMAXPROCS to n-1 yields the best performance, and the following should also be followed:
number of goroutines > 1 + GOMAXPROCS > 1

var numCores = flag.Int(“n”, 2, “number of CPU cores to use”)
runtime.GOMAXPROCS(*numCores)

• goroutine example

func main() {
    go longWait()
    go shortWait()
    fmt.Println(“About to sleep in main()”)
    time.Sleep(10 * 1e9)  
    fmt.Println(“At the end of main()”)
}
func longWait() {
    fmt.Println(“Beginning longWait()”)
    time.Sleep(5 * 1e9)         // sleep for 5 seconds
    fmt.Println(“End of longWait()”)
}
func shortWait() {
    fmt.Println(“Beginning shortWait()”)
    time.Sleep(2 * 1e9)         // sleep for 2 seconds
    fmt.Println(“End of shortWait()”)
}

using recover with goroutines

func server(workChan <-chan *Work) {
    for work := range workChan {
        go safelyDo(work) 
    }
}
func safelyDo(work *Work) {
    defer func() {
       if err := recover(); err != nil {
        log.Printf(“work failed with %s in %v:”, err, work)
       }
    }()
    do(work)
}

benchmarking goroutines

func main() {
    fmt.Println(“sync”, testing.Benchmark(BenchmarkChannelSync).String())
    fmt.Println(“buffered”,  testing.Benchmark(BenchmarkChannelBuffered).String())
}
func BenchmarkChannelSync(b *testing.B) {
    ch := make(chan int)
    go func() {
        for i := 0; i < b.N; i++ {
                ch <- i
        }
        close(ch)
    }()
    for _ = range ch {
    }
}
func BenchmarkChannelBuffered(b *testing.B) {
    ch := make(chan int, 128)
    go func() {
        for i := 0; i < b.N; i++ {
                ch <- i
        }
        close(ch)
    }()
    for _ = range ch {
    }
}
/* Output:
Windows:    N           Time 1 op               Operations per sec
sync        1000000     2443 ns/op      -->     409 332 / s
buffered    1000000     4850 ns/op      -->     810 477 / s

Channels

var ch1 chan string
ch1 = make(chan string)
ch1 := make(chan string)
buf := 100
ch1 := make(chan string, buf)
chanOfChans := make(chan chan int)
funcChan := chan func()

func main() {
    ch := make(chan string)
    go sendData(ch)
    go getData(ch)
    time.Sleep(1e9)
}
func sendData(ch chan string) {
    ch <- “Washington”
    ch <- “Tripoli”
    ch <- “London”
}
func getData(ch chan string) {
    var input string
    for { input = <-ch; fmt.Printf(“%s “, input) }
}

Semaphore pattern

type Empty interface {}
var empty Empty
...
data := make([]float64, N)
res := make([]float64, N)
sem := make(chan Empty, N)  // semaphore 
...
for i, xi := range data {
        go func (i int, xi float64) {
            res[i] = doSomething(i,xi)
            sem <- empty
        } (i, xi)
}
for i := 0; i < N; i++ {  // wait for goroutines to finish
    <-sem 
}

Channel Factory pattern

func main() {
    stream := pump()
    go suck(stream)     // shortened : go suck( pump() )
    time.Sleep(1e9)
}
func pump() chan int {
    ch := make(chan int)
    go func() {
        for i := 0; ; i++ {
            ch <- i
        }
    }()
    return ch
}
func suck(ch chan int) {
    for {
        fmt.Println(<-ch)
    }
}
func suck(ch chan int) {
    go func() {
        for v := range ch {
            fmt.Println(v)
        }
    }()
}

Channel directionality

• // channel can only receive data and cannot be closed
  var send_only chan<- int
var recv_only <-chan int // channel can only send data

var c = make(chan int) // bidirectional
go source(c)
go sink(c)
func source(ch chan<- int) {
    for { ch <- 1 }
}
func sink(ch <-chan int) {
    for { <-ch }
}

closing a channel

func sendData(ch chan string) {
    ch <- “Washington”
    ch <- “Tripoli”
    ch <- “London”
    ch <- “Beijing”
    ch <- “Tokio”
    close(ch)
}
func getData(ch chan string) {
    for {
        input, open := <-ch
        if !open {
                break
        }
        fmt.Printf(“%s “, input)
    }
}

Switching between goroutines with select

select {
case u:= <- ch1:
    ...
case v:= <- ch2:
    ...
default: // no value ready to be received
    ...
}

• rule:
  
  • if all are blocked, it waits until one can proceed
  • if multiple can proceed, it chooses one at random.
  • when none of the channel operations can proceed and the default clause is present, then this is executed: the default is always runnable (that is: ready to execute). Using a send operation in a select statement with a default case guarantees that the send will be non-blocking!

channels with timeouts and tickers

• func Tick(d Duration) <-chan Time

import “time”
rate_per_sec := 10
var dur Duration = 1e8      // rate_per_sec
chRate := time.Tick(dur)    // every 1/10th of a second
for req := range requests {
    <- chRate  // rate limit our Service.Method RPC calls
    go client.Call(“Service.Method”, req, ...)
}

• func After(d Duration) <-chan Time

func main() {
    tick := time.Tick(1e8)
    boom := time.After(5e8)
    for {
        select {
        case <-tick:
            fmt.Println(“tick.”)
        case <-boom:
            fmt.Println(“BOOM!”)
            return
        default:
            fmt.Println(“    .”)
            time.Sleep(5e7)
        }
    }
}

Tasks and Worker Processes

type Pool struct {
    Mu sync.Mutex
    Tasks []Task
}
func Worker(pool *Pool) { 
    for {
        pool.Mu.Lock()
        // begin critical section:
        task := pool.Tasks[0]           // take the first task
        pool.Tasks = pool.Tasks[1:]     // update the pool
        // end critical section
        pool.Mu.Unlock()
        process(task)
    }
}
func main() {
    pending, done := make(chan *Task), make(chan *Task)
    go sendWork(pending)        // put tasks with work
    for i := 0; i < N; i++ {    // start N goroutines to do
        go Worker(pending, done)
    }
    consumeWork(done)
}
func Worker(in, out chan *Task) {
    for {
        t := <-in
        process(t)
        out <- t
    }
}

• rule - use locking (mutexes) when:
  
  • caching information in a shared data structure
  • holding state information
• rule - use channels when:
  
  • communicating asynchronous results
  • distributing units of work
  • passing ownership of data

lazy generator

var resume chan int
func integers() chan int {
    yield := make (chan int)
    count := 0
    go func () {
        for {
            yield <- count
            count++
    }
    } ()
    return yield
}
func generateInteger() int {
    return <-resume
}
func main() {
    resume = integers()
    fmt.Println(generateInteger()) //=> 0
    fmt.Println(generateInteger()) //=> 1
}

implement a mutex

/* mutexes */
func (s semaphore) Lock() {
    s.P(1)
}
func (s semaphore) Unlock() {
    s.V(1)
}
/* signal-wait */
func (s semaphore) Wait(n int) {    
    s.P(n)
}
func (s semaphore) Signal() {
    s.V(1)
}

---

A tcp server

import (
    “fmt”
    “net”
)
func main() {
    fmt.Println(“Starting the server ...”)
    listener, err := net.Listen(“tcp”, “localhost:50000”)
    if err != nil {
        fmt.Println(“Error listening”, err.Error())
        return              // terminate program
    }
    for {
        conn, err := listener.Accept()
        if err != nil {
            fmt.Println(“Error accepting”, err.Error())
            return
        }
        go doServerStuff(conn)
    }
}
func doServerStuff(conn net.Conn) {
    for {
        buf := make([]byte, 512)
        _, err := conn.Read(buf)
        if err != nil {
            fmt.Println(“Error reading”, err.Error())
            return
        }
        fmt.Printf(“Received data: %v”, string(buf))
    }
}

func main() {
    conn, err := net.Dial(“tcp”, “localhost:50000”)
    if err != nil {
        fmt.Println(“Error dialing”, err.Error())
        return 
    }
    inputReader := bufio.NewReader(os.Stdin)
    fmt.Println(“First, what is your name?”)
    clientName, _ := inputReader.ReadString(‘\n’)
    // fmt.Printf(“CLIENTNAME %s”,clientName)
    trimmedClient := strings.Trim(clientName, “\r\n”) 
    for {
        fmt.Println(“What to send to the server? Type Q to quit.”)
        input, _ := inputReader.ReadString(‘\n’)
        trimmedInput := strings.Trim(input, “\r\n”)
        if trimmedInput == “Q” {
            return
        }
        _, err = conn.Write([]byte(trimmedClient + “ says: ” + trimmedInput))
    }
}

func main() {
    var (
        host = “www.apache.org”
        port = “80”
        remote = host + “:” + port
        msg string = “GET / \n”
        data = make([]uint8, 4096)  
        read = true  
        count = 0  
    )  
    con, err := net.Dial(“tcp”, remote)   
    io.WriteString(con, msg)  
    for read {  
        count, err = con.Read(data)  
        read = (err == nil)  
        fmt.Printf(string(data[0:count]))  
    }  
    con.Close()  
}  

func initServer(hostAndPort string) *net.TCPListener {
    serverAddr, err := net.ResolveTCPAddr(“tcp”, hostAndPort)
    checkError(err, “Resolving address:port failed: `” + hostAndPort + “’”)
    listener, err := net.ListenTCP(“tcp”, serverAddr)
    checkError(err, “ListenTCP: “)
    println(“Listening to: “, listener.Addr().String())
    return listener
}
func connectionHandler(conn net.Conn) {
    connFrom := conn.RemoteAddr().String()
    println(“Connection from: “, connFrom)
    sayHello(conn)
    for {
        var ibuf []byte = make([]byte, maxRead + 1)
        length, err := conn.Read(ibuf[0:maxRead])
        ibuf[maxRead] = 0 // to prevent overflow
        switch err {
        case nil:
            handleMsg(length, err, ibuf)
        case os.EAGAIN: // try again
            continue
        default:
            goto DISCONNECT
    }
}
DISCONNECT:
    err := conn.Close()
    println(“Closed connection: “, connFrom)
    checkError(err, “Close: “)
}

a simple webserver

http.URL	resp, err := http.Head(url)
http.Request	resp.Status
request.ParseForm();	req.FormValue(“var1”)
var1, found := request.Form[“var1”]
http.Response	http.ResponseWriter
http.StatusContinue = 100	http.StatusUnauthorized = 401
http.StatusOK = 200	http.StatusForbidden = 403
http.StatusFound = 302	http.StatusNotFound = 404
http.StatusBadRequest = 400	http.StatusInternalServerError=500

- http.Redirect(w ResponseWriter, r *Request, url string, code int)
- http.NotFound(w ResponseWriter, r *Request)
- `http.Error(w ResponseWriter, error string, code int)

package main
import (
    “fmt”
    “net/http”
    “log”
)
func HelloServer(w http.ResponseWriter, req *http.Request) {
    fmt.Println(“Inside HelloServer handler”)
    fmt.Fprint(w, “Hello,” + req.URL.Path[1:])
    // fmt.Fprintf(w, “<h1>%s</h1><div>%s</div>”, title, body)
}   // w.Header().Set(“Content-Type”, “../..”)
func main() {
    http.HandleFunc(“/”,HelloServer)
    err := http.ListenAndServe(“localhost:8080”, nil)
    // http.ListenAndServe(“:8080”, http.HandlerFunc(HelloServer)
    // http.ListenAndServeTLS()
    if err != nil {
        log.Fatal(“ListenAndServe: “, err.Error())
    }
}
func main() {                                                                              
    xlsUrl := "http://market.finance.sina.com.cn/downxls.php?"
    xlsUrl += "date=2014-04-25&symbol=sz000002"
    res, err := http.Get(xlsUrl)                                                             
    CheckError(err)                                                                          
    data, err := ioutil.ReadAll(res.Body)                                                    
    CheckError(err)                                                                          
    fmt.Printf("%s", string(data))                                                           
}                                                                                          
func CheckError(err error) {                                                               
    if err != nil {                                                                          
            log.Fatalf("Get: %v", err)                                                             
    }                                                                                        
}                                                                                          

twitter_status.go

type Status struct {
    Text string
}
type User struct {
    XMLName xml.Name
    Status  Status
}
func main() {
    response, _ := http.Get(“http://twitter.com/users/Googland.xml”)
    user := User{xml.Name{“”, “user”}, Status{“”}}
    xml.Unmarshal(response.Body, &user)
    fmt.Printf(“status: %s”, user.Status.Text)
}

Making a web application robust

type HandleFnc func(http.ResponseWriter, *http.Request)
// ...
func main() {
     http.HandleFunc(“/test1”, logPanics(SimpleServer))
     http.HandleFunc(“/test2”, logPanics(FormServer))
     if err := http.ListenAndServe(“:8088”, nil); err != nil {
        panic(err)
    }
}
func logPanics(function HandleFnc) HandleFnc {
    return func(writer http.ResponseWriter, request *http.Request) {
        defer func() {
         if x := recover(); x != nil {
            log.Printf(“[%v] caught panic: %v”, request.RemoteAddr, x)
         }
        }()
        function(writer, request)
    }
}

Sending mails with smtp

package main
import (
    “log”
    “smtp”
)
func main() {
    auth := smtp.PlainAuth(
        “”,
        “user@example.com”,
        “password”,
        “mail.example.com”,
    )
    err := smtp.SendMail(
        “mail.example.com:25”,
        auth,
        “sender@example.org”,
        []string{“recipient@example.net”},
        []byte(“This is the email body.”),
    )
    if err != nil {
        log.Fatal(err)
    }
}