Не в состоянии очистить данные при попытке использовать стандартные процедуры

Question

Я пытаюсь очистить связанные слова для данного слова, для которого я использую BFS, начиная с данного слова и перебирая каждое связанное слово на dictionary.com

Я пробовал этот код без параллелизма, и онработает просто отлично, но занимает много времени, поэтому попробовал использовать процедуры go, но мой код застревает после первой итерации.Первый уровень BFS работает просто отлично, но затем на втором уровне он зависает!

package main

import (
    "fmt"
    "github.com/gocolly/colly"
    "sync"
)

var wg sync.WaitGroup


func buildURL(word string) string {
    return "https://www.dictionary.com/browse/" + string(word)
}

func get(url string) []string {
    c := colly.NewCollector()
    c.IgnoreRobotsTxt = true
    var ret []string
    c.OnHTML("a.css-cilpq1.e15p0a5t2", func(e *colly.HTMLElement) {
        ret = append(ret, string(e.Text))
    })
    c.Visit(url)
    c.Wait()

    return ret
}

func threading(c chan []string, word string) {
    defer wg.Done()
    var words []string
    for _, w := range get(buildURL(word)) {
        words = append(words, w)
    }
    c <- words
}

func main() {
    fmt.Println("START")
    word := "jump"
    maxDepth := 2

    //bfs
    var q map[string]int
    nq := map[string]int {
        word: 0,
    }

    vis := make(map[string]bool)
    queue := make(chan []string, 5000)

    for i := 1; i <= maxDepth; i++ {
        fmt.Println(i)
        q, nq = nq, make(map[string]int)
        for word := range q {
            if _, ok := vis[word]; !ok {
                wg.Add(1)
                vis[word] = true
                go threading(queue, word)

                for v := range queue {
                    fmt.Println(v)
                    for _, w := range v {
                        nq[w] = i
                    }
                }
            }
        }
    }
    wg.Wait()
    close(queue)
    fmt.Println("END")
}

ВЫХОД:

START
1
[plunge dive rise upsurge bounce hurdle fall vault drop advance upturn inflation increment spurt boost plummet skip bound surge take]

висит здесь навсегда, counter = 2 не печатается!

Можете проверить здесь https://www.dictionary.com/browse/jump для связанных слов.

Answer 1

Согласно Tour of Go

Отправка в буферный блок канала только при заполнении буфера.Получает блок, когда буфер пуст.

Итак, в этом случае вы создаете буферизованный канал, используя длину 5000.

for i := 1; i <= maxDepth; i++ {
    fmt.Println(i)
    q, nq = nq, make(map[string]int)
    for word := range q { // for each word
        if _, ok := vis[word]; !ok { // if not visited visit
            wg.Add(1) // add a worker
            vis[word] = true
            go threading(queue, word) // fetch in concurrent manner
            for v := range queue { // <<< blocks here when queue is empty
                fmt.Println(v)
                for _, w := range v {
                    nq[w] = i
                }
            }
        }
    }
}

Как видите, япрокомментировал в коде, после 1-й итерации цикл for будет блокироваться, пока канал не станет пустым.В этом случае после извлечения jump Он отправляет массив, соответствующий подобным словам, но после этого, поскольку цикл for блокируется, поскольку zerkems объясняет, что вы не перейдете к следующей итерации (i = 2).В конечном итоге вы можете закрыть канал, чтобы завершить блокировку цикла for.Но так как вы используете один и тот же канал для записи нескольких процедур, он будет паниковать, если вы закроете его из нескольких программ.

Чтобы преодолеть это, мы можем найти хороший обходной путь.

1. Мы точно знаем, сколько невидимых элементов мы выбираем.
2. Теперь мы знаем, где находится блок

Сначала нам нужно посчитать не посещенные слова, а затем мы можем повторить стольковремени

    vis := make(map[string]bool)
    queue := make(chan []string, 5000)
    for i := 1; i <= maxDepth; i++ {
        fmt.Println(i)
        q, nq = nq, make(map[string]int)
        unvisited := 0
        for word := range q {
            if _, ok := vis[word]; !ok {
                vis[word] = true
                unvisited++
                wg.Add(1)
                go threading(queue, word)
            }
        }

        wg.Wait() // wait until jobs are done
        for j := 0; j < unvisited; j++ { // << does not block as we know how much
            v := <-queue // we exactly try to get unvisited amount
            fmt.Println(v)
            for _, w := range v {
                nq[w] = i
            }
        }
    }

В этой ситуации мы просто подсчитываем минимальные итерации, которые нам нужно пройти, чтобы получить результаты.Кроме того, вы можете видеть, что я переместился вниз по внешнему циклу for и использовал оригинальный, чтобы просто добавить слова к работникам.Он будет запрашивать выборку всех слов и будет ждать в следующем цикле неблокируемого завершения задач.

Последний цикл ждет, пока все рабочие не будут выполнены.После этой следующей итерации могут быть достигнуты работы и следующий уровень BFS.

Сводка

1. Распределение рабочей нагрузки
2. Ожидание результатов
3. Не делайте оба одновременно

Надеюсь, это поможет.