加载带有和不带有 go-routines 的地图

【问题标题】：Load a map with and without go-routines加载带有和不带有 go-routines 的地图
【发布时间】：2019-12-24 00:45:47
【问题描述】：

这是我遇到的一个有趣的情况。在使用 go-routines 进行一些数据操作之后，我需要从文件中读取，并根据我们发现的内容填充地图。这是简化的问题陈述和示例：

运行gen_data.sh生成需要的数据

#!/bin/bash 

rm some.dat || : 
for i in `seq 1 10000`; do 
    echo "$i `date` tx: $RANDOM rx:$RANDOM" >> some.dat
done

如果我使用loadtoDict.go 将some.dat 中的这些行读入map[int]string 而不使用go-routines，它会保持对齐。 （因为第一个和第二个词是一样的，见下面的o/p。）

在现实生活中，我确实需要在将线条加载到地图之前对其进行处理（昂贵），使用 go-routines 加快了我的字典创建速度，这是解决实际问题的重要要求。

loadtoDict.go

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
)

var (
    fileName = "some.dat"
)

func checkerr(err error) {
    if err != nil {
        fmt.Println(err)
        log.Fatal(err)
    }
}

func main() {
    ourDict := make(map[int]string)
    f, err := os.Open(fileName)
    checkerr(err)
    defer f.Close()

    fscanner := bufio.NewScanner(f)

    indexPos := 1

    for fscanner.Scan() {
        text := fscanner.Text()
        //fmt.Println("text", text)
        ourDict[indexPos] = text
        indexPos++

    }

    for i, v := range ourDict {
        fmt.Printf("%d: %s\n", i, v)
    }

}

跑步：

$ ./loadtoDict
...
8676: 8676 Mon Dec 23 15:52:24 PST 2019 tx: 17718 rx:1133
2234: 2234 Mon Dec 23 15:52:20 PST 2019 tx: 13170 rx:15962
3436: 3436 Mon Dec 23 15:52:21 PST 2019 tx: 17519 rx:5419
6177: 6177 Mon Dec 23 15:52:23 PST 2019 tx: 5731 rx:5449

注意第一个词和第二个词是如何“对齐”的。但是，如果我使用 go-routines 加载我的地​​图，就会出错：

async_loadtoDict.go 

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
    "sync"
)

var (
    fileName = "some.dat"
    mu       = &sync.RWMutex{}
    MAX = 9000
)

func checkerr(err error) {
    if err != nil {
        fmt.Println(err)
        log.Fatal(err)
    }
}

func main() {
    ourDict := make(map[int]string)
    f, err := os.Open(fileName)
    checkerr(err)
    defer f.Close()

    fscanner := bufio.NewScanner(f)

    indexPos := 1
    var wg sync.WaitGroup
    sem := make(chan int, MAX)
    defer close(sem)

    for fscanner.Scan() {
        text := fscanner.Text()
        wg.Add(1)
        sem <- 1
        go func() {
            mu.Lock()
            defer mu.Unlock()
            ourDict[indexPos] = text
            indexPos++
            <- sem
            wg.Done()
        }()

    }

    wg.Wait()

    for i, v := range ourDict {
        fmt.Printf("%d: %s\n", i, v)
    }

}

输出：

$ ./async_loadtoDict 
...
11: 22 Mon Dec 23 15:52:19 PST 2019 tx: 25688 rx:7602
5716: 6294 Mon Dec 23 15:52:23 PST 2019 tx: 28488 rx:3572
6133: 4303 Mon Dec 23 15:52:21 PST 2019 tx: 24286 rx:1565
7878: 9069 Mon Dec 23 15:52:25 PST 2019 tx: 16863 rx:24234
8398: 7308 Mon Dec 23 15:52:23 PST 2019 tx: 4321 rx:20642
9566: 3489 Mon Dec 23 15:52:21 PST 2019 tx: 14447 rx:12630
2085: 2372 Mon Dec 23 15:52:20 PST 2019 tx: 14375 rx:24151

尽管使用互斥锁保护了摄取 ourDict[indexPos]。我希望我的地图索引与摄取尝试保持一致。

谢谢！

【问题讨论】：

  • 多么不必要的复杂...索引不匹配的原因是即使您以相同的顺序创建 goroutine 并防止并发您有 MAX (9000) goroutines 等待，并且您无法控制它们恢复的顺序，索引代表执行顺序，而不是创建顺序
  • 顺便说一下，您的代码是完全顺序的，只是不确定性。
  • 除非我保留MAX = 1，否则我会观察我上面报告的内容——这比让 go-routines 准备和填充我的map 失败。我确实需要在将线条加载到地图之前对其进行处理，使用 go-routines 加快了我的字典创建速度，这是解决实际问题的重要要求。
  • 在单个 goroutine 中添加对地图的访问，其他的只是准备数据（我想有一些数据操作，因为 if 只是像示例 goroutines 这样的传递实际上会使其变慢开销）。我将一个例子作为答案
  • 正如我在回答中所说，您的信号量 sem 不起作用，因为您对其进行了深度缓冲。当您设置 MAX = 1 时，您将其设置为一个条目，然后它就会起作用：它会强制您的每个衍生的 goroutine 等到前一个完成后才能开始。

标签： go


【解决方案1】：

您的信号量 sem 无法正常工作，因为您对其进行了深度缓冲。

一般来说，为此类任务设置地图的方法是错误的，因为读取文件会很慢。如果你有一个更复杂的任务——例如，读一行，想很多，设置一些东西——你会想要这个作为你的伪代码结构：

type workType struct {
    index int
    line  string
}

var wg sync.WaitGroup
wg.Add(nWorkers)
// I made this buffered originally but there's no real point, so
// fixing that in an edit
work := make(chan workType)
for i := 0; i < nWorkers; i++ {
    go readAndDoWork(work, &wg)
}

for i := 1; fscanner.Scan(); i++ {
    work <- workType{index: i, line: fscanner.Text()}
}
close(work)
wg.Wait()

... now your dictionary is ready ...

工人这样做：

func readAndDoWork(ch chan workType, wg *sync.WorkGroup) {
    for item := range ch {
        ... do computation ...
        insertIntoDict(item.index, result)
    }
    wg.Done()
}

insertIntoDict 抓取互斥体（以保护映射从索引到结果）并写入字典。 （如果您愿意，可以直接内联它。）

这里的想法是设置一定数量的工作人员（可能基于可用 CPU 的数量），每个工作人员都抓取下一个工作项并处理它。主 goroutine 只是打包工作，然后关闭工作通道——这将导致所有工作人员看到输入结束——然后等待他们发​​出信号表明他们已经完成了计算。

（如果您愿意，您可以再创建一个 goroutine 来读取工作人员计算的结果并将它们放入映射中。这样您就不需要映射本身的互斥体。）

【讨论】：

  • @mh-cbon：我不确定自己，但我认为 OP 不喜欢关于为什么 sem 没有序列化计数器的非常简短的描述。
【解决方案2】：

正如我在 cmets 中提到的，您无法控制 goroutine 的执行顺序，因此不应从它们内部更改索引。

这是一个示例，其中与地图的交互在单个 goroutine 中，而您在其他 goroutine 中进行处理：

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
    "sync"
)

var (
    fileName = "some.dat"
    MAX      = 9000
)

func checkerr(err error) {
    if err != nil {
        fmt.Println(err)
        log.Fatal(err)
    }
}

type result struct {
    index int
    data string
}

func main() {
    ourDict := make(map[int]string)
    f, err := os.Open(fileName)
    checkerr(err)
    defer f.Close()

    fscanner := bufio.NewScanner(f)

    var wg sync.WaitGroup
    sem := make(chan struct{}, MAX) // Use empty structs for semaphores as they have no allocation
    defer close(sem)
    out := make(chan result)
    defer close(out)
    indexPos := 1

    for fscanner.Scan() {
        text := fscanner.Text()
        wg.Add(1)
        sem <- struct{}{}

        go func(index int, data string) {
            // Defer the release of your resources, otherwise if any error occur in your goroutine
            // you'll have a deadlock
            defer func() {
                wg.Done()
                <-sem
            }()
            // Process your data
            out <- result{index, data}
        }(indexPos, text) // Pass in the data that will change on the iteration, go optimizer will move it around better

        indexPos++
    }

    // The goroutine is the only one to write to the dict, so no race condition
    go func() {
        for {
            if entry, ok := <-out; ok {
                ourDict[entry.index] = entry.data
            } else {
                return // Exit goroutine when channel closes
            }
        }
    }()

    wg.Wait()

    for i, v := range ourDict {
        fmt.Printf("%d: %s\n", i, v)
    }

}

【讨论】：

  • 在最后一个go func中，简单地越过通道，一旦通道关闭，循环就会退出。
  • 等待条件不正确。您不应该等待输入处理，而是等待输出处理。 defer close(out) 应该在 wg.Wait 发布时发生。超出的范围可以内联到 main 中，如果您只打印结果，则无需预先保存到内存中。这段代码做了很多wg.Add/wg.Done，效率很低。提前声明工人有助于减少争用。
【解决方案3】：

好的，我已经想通了。通过复制给goroutine一个值来挂起，似乎有效。

改变：

for fscanner.Scan() {
    text := fscanner.Text()
    wg.Add(1)
    sem <- 1
    go func() {
        mu.Lock()
        defer mu.Unlock()
        ourDict[indexPos] = text
        indexPos++
        <- sem
        wg.Done()
    }()

}

到

for fscanner.Scan() {
        text := fscanner.Text()
        wg.Add(1)
        sem <- 1
        go func(mypos int) {
                mu.Lock()
                defer mu.Unlock()
                ourDict[mypos] = text
                <-sem
                wg.Done()
        }(indexPos)
        indexPos++
}

完整代码：https://play.golang.org/p/dkHaisPHyHz

使用工人池，

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
    "sync"
)

const (
    MAX      = 10
    fileName = "some.dat"
)

type gunk struct {
    line string
    id   int
}

func main() {
    ourDict := make(map[int]string)
    wg := sync.WaitGroup{}
    mu := sync.RWMutex{}

    cha := make(chan gunk)

    for i := 0; i < MAX; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for {
                textin, ok := <-cha
                if !ok {
                    return
                }
                mu.Lock()
                ourDict[textin.id] = textin.line
                mu.Unlock()
            }
        }(i)
    }

    f, err := os.Open(fileName)
    checkerr(err)
    defer f.Close()
    fscanner := bufio.NewScanner(f)
    indexPos := 1

    for fscanner.Scan() {
        text := fscanner.Text()
        thisgunk := gunk{line: text, id: indexPos}
        cha <- thisgunk
        indexPos++
    }

    close(cha)
    wg.Wait()
    for i, v := range ourDict {
        fmt.Printf("%d: %s\n", i, v)
    }

}

func checkerr(err error) {
    if err != nil {
        fmt.Println(err)
        log.Fatal(err)
    }
}

【讨论】：

  • 传递索引会有所帮助，是的，因为您曾经使用互斥锁来保护它（仅在使用和增量期间，意味着其他 goroutine 在您之前增加了它）但是现在您有一个每个 goroutine 的索引。但是您认为变量sem 为您做了什么？
  • sem 是限制并发到 MAX goroutines。 stackoverflow.com/a/25306439/9488865
  • 正确 - 所以当你将它设置为 9000 时，你会衍生出多达 9000 个并行 goroutine。这并不是真正的高效：您可用的 CPU 数量将限制您可以做的实际工作量。当您将其设置为 1 时，您将自己限制为 1 个 goroutine，然后 indexPos 本身在 所有 goroutines 之间共享这一事实，只有 1 个 gorouting 使用它。使用更新后的代码，您可以在每个 goroutine 中复制 indexPos。
  • 请注意，创建一个新的 goroutine 需要花费（少量），而在通道上发送和接收需要花费（少量）。通常最好启动一次 n-cpus-available 工作人员，然后通过通道为每个工作人员提供工作，而不是为每个工作启动一个一次性工作人员，然后让它通过信号量通道说话以限制如何许多人实际上可以同时运行。
  • 参考torek的答案，它会给你比这个硬而复杂的代码更好的结果。
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2021-12-30
  • 2014-12-21
  • 1970-01-01
  • 2019-10-28
相关资源
最近更新 更多
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode