标签: goroutine

我正在通过Go中的goroutines优化矩阵乘法.

我的基准测试显示,每行或每个元素引入并发性会大大降低性能:

goos: darwin
goarch: amd64
BenchmarkMatrixDotNaive/A.MultNaive-8                            2000000               869 ns/op               0 B/op          0 allocs/op
BenchmarkMatrixDotNaive/A.ParalMultNaivePerRow-8                  100000             14467 ns/op              80 B/op          9 allocs/op
BenchmarkMatrixDotNaive/A.ParalMultNaivePerElem-8                  20000             77299 ns/op             528 B/op         65 allocs/op

我知道缓存局部性的一些基本的先验知识,每个元素并发性能会降低性能.但是,为什么即使在天真的版本中,每行仍然会降低性能？

事实上,我还写了一个块/平铺优化,它的vanilla版本(没有goroutine并发)甚至比天真版本更糟糕(这里不存在,让我们首先关注天真).

我在这做错了什么？为什么？如何在这里优化？

乘法:

package naive

import (
    "errors"
    "sync"
)

// Errors
var (
    ErrNumElements = errors.New("Error number of elements")
    ErrMatrixSize  = errors.New("Error size of matrix")
)

// Matrix is a 2d array
type Matrix struct {
    N    int
    data [][]float64
}

// New a size by size matrix …

我试图理解非缓冲通道，所以我编写了一个小应用程序，它遍历用户输入的数组，做一些工作，将信息放在非缓冲通道上，然后读取它。但是，我无法从频道中读取信息。这是我的代码

toProcess := os.Args[1:]

var wg sync.WaitGroup
results := make(chan string)
errs := make(chan error)

for _, t := range toProcess {
    wg.Add(1)
    go Worker(t, "text", results, errs, &wg)
}


go func() {
    for err := range errs {
        if err != nil {
            fmt.Println(err)
        }
    }
}()


go func() {
    for res := range results {
        fmt.Println(res)
    }
}()

我对非缓冲通道有什么不了解？我想我应该在上面放置信息，并从中读取另一个常规内容。

编辑：使用两个 goroutines 解决了问题，但是当出现错误时它仍然给我以下信息：

open /Users/roosingh/go/src/github.com/nonbuff/files/22.txt: no such file or directory
fatal error: all goroutines are asleep - deadlock!

goroutine …

我有一个grpc基准测试代码，该代码使用一个函数使用for-select子句将数百个goroutine通道合并到一个通道。代码是这样的

     func (b *B) merge(
          ctx context.Context,
          nodes ...<-chan *pb.Node,
        ) chan *pb.Node {
    allNodes := make(chan *pb.Node)
    var wg sync.WaitGroup
    wg.Add(len(nodes))
    for _, n := range nodes {
        go func(n <-chan *pb.Node) {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-n:
                    if ok {
                        allNodes <- val
                    }
                }
            }
        }(n)
    }
    go func() {
        wg.Wait()
        close(allNodes)
    }()
    return allNodes
}

当我在ubuntu 16.04中通过top命令监视代码时，我看到2核服务器变得疯狂，超过了CPU使用率的196％。

然后，我使用pprof包分析了我的代码，它说98％的cpu旋转了此函数，并且top函数生成了这样的结果

    flat  flat%   sum%        cum   cum%
   1640ms  5.78%  5.78%    27700ms …

我有以下代码行：

var a_map = make(map[string] []int)

我使用 a_map 变量的部分代码偶尔会引发以下错误：

fatal error: concurrent map read and map write

为了创建一个更健壮的解决方案，一个没有此类错误的解决方案，我想使用一个 sync.Map 而不是通用映射。对此堆栈溢出问题提供的唯一答案启发了我这样做。但是，我不清楚这样做的正确语法。

对于我的第一次尝试，我使用了以下代码行：

var a_map = make(sync.Map[string] []int)

这导致了以下错误：

...syntax error: unexpected ], expecting expression

然后我尝试：

 sync_map := new(sync.Map)
 var a_map = make(sync_map[string] []int)

这导致了同样的错误：

...syntax error: unexpected ], expecting expression

我有两个 goroutine，如下面的代码片段所示。我想同步它们，这样当一个返回时，另一个也应该退出。实现这一目标的最佳方法是什么？

func main() {

  go func() {
    ...
    if err != nil {
      return
    }
  }()

  go func() {
    ...
    if err != nil {
      return
    }
  }()


}

我在这里模拟了这个场景https://play.golang.org/p/IqawStXt7rt并试图用一个通道来解决它来表示例程已经完成。这看起来可能是对关闭通道的写入导致恐慌。解决这个问题的最佳方法是什么？

拿这段代码：

func main() {
    var x int
    go func() {
        for {
            x++
        }
    }()
    time.Sleep(time.Second)
    fmt.Println("x =", x)
}

为什么最后x相等0？我知道 Go 的调度程序需要time.Sleep()调用来获取 goroutine，但为什么不这样做？

提示：在 for 循环内放置 atime.Sleep()或调用runtime.Gosched()可修复此代码。但为什么？

更新：检查相同代码的以下版本：

func main() {
    var x int
    go func() {
        for i := 0; i < 10000; i++ {
            x++
        }
    }()
    time.Sleep(time.Second)
    fmt.Println("x =", x)
}

奇怪的是 goroutine 里面的代码现在被执行了，x不再是 0 了。 编译器在这里做了什么优化吗？

我有这个示例代码，我正面临这个同步问题，任何人都可以帮助我如何实现这一点。

package main

import "fmt"

func main() {

baseChan := make(chan int)
go func(bCh chan int){
for {
select{
    case stats, _ := <- bCh:
    fmt.Println("base stats", stats)
}}
}(baseChan)

second := make(chan int)
go func (sCh chan int) {
fmt.Println("second channel")
for {
select {
case stats, _ := <- sCh:
    fmt.Println("seconds stats", stats)
    baseChan <- stats
}
}
}(second)
runLoop(second)
}

func runLoop(second chan int) {
 for i := 0; i < 5; i++ {
fmt.Println("writing i", i) …

这是 Go 代码 https://www.intervue.io/sandbox-ILSCXZ6RR

func worker() chan int {
   ch := make(chan int)

   go func() {
      time.Sleep(3 * time.Second)
      ch <- 42
   }()

   return ch
}

func main() {
   timeStart := time.Now()

   _, _ = <-worker(), <-worker()

   println(int(time.Since(timeStart).Seconds())) // 3 or 6 ?
}

如何在 3 秒内执行而不是在 6 秒内执行？

我有这个循环，它的作用是试图重复轮询另一台服务器。我使用 Ticker 来实现这一点，但是程序反复显示 100% 的 CPU 使用率。

这个代码运行在一个 goroutine 中。并且 HTTP 服务器在另一个 goroutine 中运行。

func() Monitor() {

  abort := make(chan bool)

  log.Info("Monitor started.")

  // start the monitor goroutine
  go func() {
      defer log.Info("Stopped monitor")
        
      ticker := time.NewTicker(time.Duration(35.0) * time.Second)
      defer ticker.Stop()
        
      log.Info("Monitor started! \n")
      for {
        select {
        case t := <-ticker.C:
            log.Infof("Subscribe to service at time %v\n", t)
            if err := selfConn.SubscribeToService(); err != nil {
                log.Errorf("Failed to subscribe to primary connector: %v", err)
            } 
        case <-abort:
            log.Info("Finished routine!") …

我有这个基本的 go 程序，它打印到控制台并调用 2 个 goroutines

package main

import (
    "fmt"
    "time"
)

func f(from string) {
    for i := 0; i < 3; i++ {
        fmt.Println(from, ":", i)
    }
}

func main() {
    f("hello")
    go f("foo")
    go f("bar")
    time.Sleep(time.Second)
}

输出如下——我想知道为什么在“foo”之前打印“bar”——是什么决定了 goroutines 的执行顺序？

hello : 0
hello : 1
hello : 2
bar : 0
bar : 1
bar : 2
foo : 0
foo : 1
foo : 2