goroutine ParallelGenerate benchmark

generate.go

package main

import (
	"io/ioutil"
	"log"
	"net/http"
	"sync"
	"time"
)

type Generator interface {
	Generate() (interface{}, error)
}

func ParallelGenerate(generators []interface{}) chan []interface{} {
	generated := make(chan interface{})
	finish := make(chan bool)
	results := make(chan []interface{})

	go func() {
		generatedValues := make([]interface{}, 0)

		for {
			select {
			case value := <-generated:
				generatedValues = append(generatedValues, value)
			case <-finish:
				results <- generatedValues
				return
			}
		}
	}()

	go func() {
		var wg sync.WaitGroup
		for _, g := range generators {
			wg.Add(1)
			go func(g interface{}) {
				value, err := g.(Generator).Generate()
				if err != nil {
					log.Printf("Generation failed")
				}
				generated <- value
				wg.Done()
			}(g)
		}
		wg.Wait()
		finish <- true
	}()

	return results
}

type HttpContentGenerator struct {
	url string
}

func (g *HttpContentGenerator) Generate() (interface{}, error) {
	resp, err := http.Get(g.url)
	if err != nil {
		return "", err
	}
	defer resp.Body.Close()
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return "", err
	}

	return string(body), nil
}

func serial(url string, reqTimes int, n int) {
	before := time.Now().UnixNano()
	for i := 0; i < n; i++ {
		contents := make([]string, 0)
		for i := 0; i < reqTimes; i++ {
			c, err := (&HttpContentGenerator{url}).Generate()
			if err != nil {
				log.Printf("Request failed for %s\n", url)
			}
			contents = append(contents, c.(string))
		}
	}
	after := time.Now().UnixNano()
	log.Printf("serial(%s, %2d, %d) %f sec\n", url, reqTimes, n, float64(after - before) / ( 1000 * 1000 * 1000 ) / float64(n))
}

func parallel(url string, reqTimes int, n int) {
	before := time.Now().UnixNano()
	for i := 0; i < n; i++ {
		generators := make([]interface{},0)
		for i := 0; i < reqTimes; i++ {
			generators = append(generators, &HttpContentGenerator{url} )
		}
		results := ParallelGenerate(generators)
		<-results
	}
	after := time.Now().UnixNano()
	log.Printf("parallel(%s, %2d, %d) %f sec\n", url, reqTimes, n, float64(after - before) / ( 1000 * 1000 * 1000 ) / float64(n))
}

func main() {
	serial("http://localhost:5000?sleep=0", 10, 100)
	serial("http://localhost:5000?sleep=1", 10, 100)

	parallel("http://localhost:5000?sleep=0", 10, 100)
	parallel("http://localhost:5000?sleep=1", 10, 100)
}

result.txt

2013/12/23 12:47:01 serial(http://localhost:5000?sleep=0, 10, 100) 0.020681 sec
2013/12/23 13:03:44 serial(http://localhost:5000?sleep=1, 10, 100) 10.030407 sec
2013/12/23 13:03:46 parallel(http://localhost:5000?sleep=0, 10, 100) 0.014724 sec
2013/12/23 13:05:27 parallel(http://localhost:5000?sleep=1, 10, 100) 1.012447 sec

sleep.psgi

use v5.18;
use Plack::Request;
use Time::HiRes qw(sleep);

sub {
    my ($env) = @_;
    my $req = Plack::Request->new($env);
    return [ 404, [], ['404'] ] unless $req->uri->path eq '/';

    my $sleep = $req->parameters->{sleep} || 0;
    warn "start sleeping for $sleep sec";
    sleep $sleep if $sleep;
    warn "end   sleeping for $sleep sec";

    return [ 200, [], ['ok'] ];
}