ch8/ch8-05 #174

package main

import (
	"crypto/rand"
	"fmt"
	"image"
	"image/color"
	"image/png"
	"math"
	"math/cmplx"
	"os"
	"runtime"
	"sync"
	"time"
)

const (
	xmin, ymin, xmax, ymax = -2, -2, +2, +2
	width, height          = 1024, 1024
)

const (
	RUN_TIMES = 1e1
	CAP       = 1024
	RATIO     = 0.9
)

var CORES int
var GOROUTINE_NUM int

type pixel struct {
	x int
	y int
	c color.Color
}

func init() {
	CORES = runtime.NumCPU()
	GOROUTINE_NUM = int(float64(CORES) * RATIO)
	fmt.Printf("Cores: %d, Go routine count: %d\n", CORES, GOROUTINE_NUM)
}

func main() {
	// 朴素版本
	var total int64 = 0
	for i := 1; i <= RUN_TIMES; i++ {
		total += drawPlain()
	}
	fmt.Printf("it takes %.2f milliseconds for drawPlain() to execute\n", float64(total)/float64(RUN_TIMES))

	// 异步版本，很多go routine，破坏局部性
	// total = 0
	// for i := 1; i <= RUN_TIMES; i++ {
	// 	total += drawAsync()
	// }
	// fmt.Printf("it takes %.2f milliseconds for drawAsync() to execute\n", float64(total)/float64(RUN_TIMES))

	// 异步版本，局部性
	// total = 0
	// for i := 1; i <= RUN_TIMES; i++ {
	// 	total += drawAsyncLocality()
	// }
	// fmt.Printf("it takes %.2f milliseconds for drawAsyncLocality() to execute\n", float64(total)/float64(RUN_TIMES))

	// 减少竞争
	total = 0
	for i := 1; i <= RUN_TIMES; i++ {
		total += drawAsyncTender()
	}
	fmt.Printf("it takes %.2f milliseconds for drawAsyncTender() to execute\n", float64(total)/float64(RUN_TIMES))
}

// 朴素版本
func drawPlain() (duration int64) {
	defer func(start time.Time) {
		duration = calcExecTime(start)
	}(time.Now())

	img := image.NewRGBA(image.Rect(0, 0, width, height))
	for py := 0; py < height; py++ {
		y := float64(py)/height*(ymax-ymin) + ymin
		for px := 0; px < width; px++ { // cache locality
			x := float64(px)/width*(xmax-xmin) + xmin
			z := complex(x, y)
			// Image point (px, py) represents complex value z.
			img.Set(px, py, mandelbrot(z))
		}
	}

	file, err := os.OpenFile("./plain.png", os.O_RDWR|os.O_CREATE, 0666)
	if err != nil {
		fmt.Println("open file err:", err)
	}
	defer file.Close()
	png.Encode(file, img) // NOTE: ignoring errors
	return
}

// 异步执行
func drawAsync() (duration int64) {
	defer func(start time.Time) {
		duration = calcExecTime(start)
	}(time.Now())

	var wg sync.WaitGroup

	pixelCh := make(chan pixel, CAP)

	img := image.NewRGBA(image.Rect(0, 0, width, height))
	for py := 0; py < height; py++ {
		y := float64(py)/height*(ymax-ymin) + ymin
		for px := 0; px < width; px++ { // cache locality
			x := float64(px)/width*(xmax-xmin) + xmin
			z := complex(x, y)
			// Image point (px, py) represents complex value z.
			wg.Add(1)
			go func(x, y int, z complex128) {
				defer wg.Done()
				pixelCh <- pixel{x: x, y: y, c: mandelbrot(z)}
			}(px, py, z)
		}
	}

	go func() {
		wg.Wait()
		close(pixelCh)
	}()

	for pixel := range pixelCh {
		img.Set(pixel.x, pixel.y, pixel.c)
	}

	file, err := os.OpenFile("./async.png", os.O_RDWR|os.O_CREATE, 0666)
	if err != nil {
		fmt.Println("open file err:", err)
	}
	defer file.Close()
	png.Encode(file, img) // NOTE: ignoring errors
	return
}

// 提高局部性
func drawAsyncLocality() (duration int64) {
	defer func(start time.Time) {
		duration = calcExecTime(start)
	}(time.Now())

	var wg sync.WaitGroup

	pixelCh := make(chan pixel, CAP)

	img := image.NewRGBA(image.Rect(0, 0, width, height))
	for py := 0; py < height; py++ {
		y := float64(py)/height*(ymax-ymin) + ymin
		wg.Add(1)
		go func(py int, y float64) {
			defer wg.Done()
			for px := 0; px < width; px++ {
				x := float64(px)/width*(xmax-xmin) + xmin
				z := complex(x, y)
				pixelCh <- pixel{x: px, y: py, c: mandelbrot(z)}
			}
		}(py, y)
	}

	go func() {
		wg.Wait()
		close(pixelCh)
	}()

	for pixel := range pixelCh {
		// fmt.Printf("%v\n", pixel)
		img.Set(pixel.x, pixel.y, pixel.c)
	}

	file, err := os.OpenFile("./asyncLocality.png", os.O_RDWR|os.O_CREATE, 0666)
	if err != nil {
		fmt.Println("open file err:", err)
	}
	defer file.Close()
	png.Encode(file, img) // NOTE: ignoring errors
	return
}

// 限制协程数目，减少交流和同步的时间
func drawAsyncTender() (duration int64) {
	defer func(start time.Time) {
		duration = calcExecTime(start)
	}(time.Now())

	pixelCh := make(chan pixel, CAP)
	done := make(chan int)

	wg := sync.WaitGroup{}

	routinePool := make(chan struct{}, GOROUTINE_NUM)
	for i := 0; i < GOROUTINE_NUM; i++ {
		routinePool <- struct{}{}
	}

	img := image.NewRGBA(image.Rect(0, 0, width, height))

	go func() {
		for pixel := range pixelCh {
			// fmt.Printf("%v\n", pixel)
			img.Set(pixel.x, pixel.y, pixel.c)
		}
		done <- 1
	}()

	for py := 0; py < height; py++ {
		y := float64(py)/height*(ymax-ymin) + ymin

		// 取出一个routine
		<-routinePool
		wg.Add(1)

		go func(py int, y float64) {
			defer func() {
				// 添加一个routine
				routinePool <- struct{}{}
				wg.Done()
			}()

			for px := 0; px < width; px++ {
				x := float64(px)/width*(xmax-xmin) + xmin
				z := complex(x, y)
				pixelCh <- pixel{x: px, y: py, c: mandelbrot(z)}
			}
		}(py, y)
	}

	go func() {
		wg.Wait()
		close(pixelCh)
	}()

	<-done

	// fmt.Println("hello")

	file, err := os.OpenFile("./asyncTender.png", os.O_RDWR|os.O_CREATE, 0666)
	if err != nil {
		fmt.Println("open file err:", err)
	}
	defer file.Close()
	png.Encode(file, img) // NOTE: ignoring errors
	return
}

func mandelbrot(z complex128) color.Color {
	const iterations = 200
	const contrast = 15

	// writeLargeFile("./temp.txt")
	const gn = 10
	alpha := [][]float64{
		{3.6, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 0.2, 0.1},
		{0.3, 3.5, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.2},
		{0.4, 0.2, 3.4, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.3},
		{0.5, 0.3, 0.1, 3.3, 0.1, 0.2, 0.3, 0.4, 0.5, 0.4},
		{0.6, 0.4, 0.2, 0.1, 3.2, 0.1, 0.2, 0.3, 0.4, 0.5},
		{0.7, 0.5, 0.3, 0.2, 0.1, 3.1, 0.1, 0.2, 0.3, 0.6},
		{0.8, 0.6, 0.4, 0.3, 0.2, 0.1, 3.0, 0.1, 0.2, 0.7},
		{0.9, 0.7, 0.5, 0.4, 0.3, 0.2, 0.1, 2.9, 0.1, 0.8},
		{0.2, 0.8, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1, 2.8, 0.9},
		{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 2.7},
	}
	beta := []float64{1.1, 1.5, 2.4, 4.1, 6.4, 9.0, 12.9, 18.0, 24.3, 32.8}

	gaussJordan(gn, alpha, beta)

	// fmt.Println(x)

	var v complex128
	for n := uint8(0); n < iterations; n++ {
		v = v*v + z
		if cmplx.Abs(v) > 2 {
			return color.Gray{255 - contrast*n}
		}
	}
	return color.Black
}

func calcExecTime(start time.Time) (duration int64) {
	return time.Since(start).Milliseconds()
}

func writeLargeFile(filepath string) error {
	// 创建文件
	f, err := os.Create(filepath)
	if err != nil {
		fmt.Println("create file err:", err)
		return err
	}
	defer f.Close()

	// 写入 10kB 数据
	var data [1024]byte // 1MB
	for i := 0; i < 10; i++ {
		_, err := rand.Read(data[:])
		if err != nil {
			fmt.Println("read rand err:", err)
			return err
		}
		_, err = f.Write(data[:])
		if err != nil {
			fmt.Println("write data err:", err)
			return err
		}
	}

	return nil
}

func gaussJordan(n int, A [][]float64, b []float64) []float64 {
	//高斯-约旦方法的实现
	for k := 0; k < n; k++ {
		p := k
		for i := k + 1; i < n; i++ {
			if math.Abs(A[i][k]) > math.Abs(A[p][k]) {
				p = i
			}
		}

		A[k], A[p] = A[p], A[k]
		b[k], b[p] = b[p], b[k]

		for i := k + 1; i < n; i++ {
			alpha := A[i][k] / A[k][k]
			for j := k; j < n; j++ {
				A[i][j] -= alpha * A[k][j]
			}
			b[i] -= alpha * b[k]
		}
	}

	//回带解方程
	x := make([]float64, n)
	for i := n - 1; i >= 0; i-- {
		sum := b[i]
		for j := i + 1; j < n; j++ {
			sum -= A[i][j] * x[j]
		}
		x[i] = sum / A[i][i]
	}

	return x
}

0 replies

KrxkGit · 2023-10-27T17:44:09Z

KrxkGit
Oct 27, 2023 — with giscus

这是一个典型的生产者-消费者模型，文中处理的方式的特别之处是采用了阻塞的channel，文中的方式实际上是构建了一种类似semaphore的模型。
还有就是文中的需求：需要等待所有routine完成后再统计，所以总结一下需求：统计前要进入串行等待状态，且串行等待状态下不能阻塞前面异步的完成，否则将造成死锁。而文中的方案是：用异步的单routine的func去完成这个串行等待（由于是单routine且前期wg已经准备好，所以可以保证routine内部是串行的）

1 reply

shanhuhai Jan 9, 2024 — with giscus

估计有人不知道怎么调 makeThumbnails6 , 提供一个调用的例子：

func main() {
	images := []string{
		"/Users/xxx/Pictures/avatar/1.jpeg",
		"/Users/xxx/Pictures/avatar/2.jpeg",
		"/Users/xxx/Pictures/avatar/3.jpeg",
	}
	
	ch := make(chan string)
	go func() {
		defer close(ch)
		for _, filename := range images {
			ch <- filename
			log.Println(filename + " sent .")
		}
	}()
	result := makeThumbnails6(ch)
	log.Printf("Total size of thumbnails: %d bytes\n", result)

}

func makeThumbnails6(filenames <-chan string) int64 {
	sizes := make(chan int64)
	var wg sync.WaitGroup
	for f := range filenames {

		log.Printf("%v receive .", f)
		wg.Add(1)
		go func(f string) {
			defer wg.Done()
			thumb, err := thumbnail.ImageFile(f)
			if err != nil {
				log.Println(err)
				return
			}
			info, _ := os.Stat(thumb)
			sizes <- info.Size()
		}(f)
	}
	go func() {
		wg.Wait()
		close(sizes)
	}()
	var total int64
	for size := range sizes {
		total += size
	}
	return total
}

profikiller-zzy · 2024-05-29T13:11:55Z

profikiller-zzy
May 29, 2024 — with giscus

我靠这个wg.add(1)的位置还有closer goroutine写得真是既优雅又巧妙！

0 replies

ice-jade · 2024-07-10T13:34:33Z

ice-jade
Jul 10, 2024 — with giscus

有一个点很好奇：在makeThumbnails6中如果两次for循环间隔太大，是否导致该函数提前退出。
比如说filenames chan每隔1分钟发送一个消息。但 for f := range filenames 中goroutine只需要半分钟就执行完，看起来会在第二次循环前导致wg.Wait()解除，sizes chan关闭，从而makeThumbnails6未等待filenames chan关闭就返回了

1 reply

ReReniao Jul 10, 2024 — with giscus

不会；for f := range filenames 中启动的 goroutine 向 sizes 中发送消息会阻塞（无缓存channel），这样只有循环结束才会开始接收size信息（主goroutine），wg.done之后才会执行，即wg等待的goroutine数量只会一直增加到最大，然后开始减少,直到wg.wait

ffgan · 2024-07-23T02:57:57Z

ffgan
Jul 23, 2024 — with giscus

···
sizes channel携带了每一个文件的大小到main goroutine，在main goroutine中使用了range loop来计算总和。观察一下我们是怎样创建一个closer goroutine，并让其在所有worker goroutine们结束之后再关闭sizes channel的。两步操作：wait和close，必须是基于sizes的循环的并发。考虑一下另一种方案：如果等待操作被放在了main goroutine中，在循环之前，这样的话就永远都不会结束了，如果在循环之后，那么又变成了不可达的部分，因为没有任何东西去关闭这个channel，这个循环就永远都不会终止。
···

对于另一种方案，这里分两种情况
1： wait和close放在统计size之前，就会导致wg阻塞，因为wg的计数没到0，注意到wg的计数是在worker goroutine处理完之后的 defer语句，但是worker goroutine的最后有一句sizes <- info.Size()，而sizes是无缓存的channel，在wait之前都没有语句来接收这个sizes的数据所以会一直阻塞worker goroutine，也就是无法执行wg.Done()，所以会永远无法结束
2：放在统计sizes之后，for range没法结束，如上所述，是因为没人去关闭sizes

0 replies

Capricornsong · 2024-09-29T02:52:35Z

Capricornsong
Sep 29, 2024 — with giscus

sync.WaitGroup 内部维护了一个计数器，用于记录仍需等待的 Goroutine 数量。这个计数器具有以下特性：
当计数器为0时，WaitGroup 的 Wait 方法会立即返回。当计数器不为0时，Wait 方法将会阻塞，直到计数器归零。
Add(delta int)：将计数器的值增加 delta。delta 可以为负数，用于减少计数器的值。特别注意，Add 的调用应当以非阻塞的方式进行，且不能在 Goroutine 中执行。
Done()：将计数器的值减一，实际上是 Add(-1) 的简写形式。
Wait()：阻塞当前的 Goroutine，直到计数器的值为0。

0 replies

mefly2012 · 2024-10-09T02:49:52Z

mefly2012
Oct 9, 2024 — with giscus

如下写法也并没有全部输出5呢，
package main

import (
"fmt"
"time"
)
func main() {
a := []string{"1", "2", "3", "4", "5"}
ch := make(chan int, 5)
for _, i := range a {
go func() {
time.Sleep(1 * time.Second)
fmt.Println(i)
ch <- 1
}()
}
for range ch {
<-ch
}
}

0 replies

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ch8/ch8-05 #174

{{title}}

Replies: 10 comments 3 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

ch8/ch8-05 #174

giscus[bot] bot Jan 23, 2023

ch8/ch8-05

Replies: 10 comments · 3 replies

Fu-LianPeng Jan 23, 2023 — with giscus

Mahoo12138 Mar 5, 2023 — with giscus

ReReniao Oct 29, 2023 — with giscus

whaliendev Mar 30, 2023 — with giscus

whaliendev Mar 30, 2023 — with giscus

8.5 改写mandelbrot程序

KrxkGit Oct 27, 2023 — with giscus

shanhuhai Jan 9, 2024 — with giscus

profikiller-zzy May 29, 2024 — with giscus

ice-jade Jul 10, 2024 — with giscus

ReReniao Jul 10, 2024 — with giscus

ffgan Jul 23, 2024 — with giscus

Capricornsong Sep 29, 2024 — with giscus

mefly2012 Oct 9, 2024 — with giscus

giscus[bot]
bot Jan 23, 2023

Replies: 10 comments 3 replies

Fu-LianPeng
Jan 23, 2023 — with giscus

Mahoo12138
Mar 5, 2023 — with giscus

whaliendev
Mar 30, 2023 — with giscus

whaliendev
Mar 30, 2023 — with giscus

KrxkGit
Oct 27, 2023 — with giscus

profikiller-zzy
May 29, 2024 — with giscus

ice-jade
Jul 10, 2024 — with giscus

ffgan
Jul 23, 2024 — with giscus

Capricornsong
Sep 29, 2024 — with giscus

mefly2012
Oct 9, 2024 — with giscus