Plusieurs goroutines écoutant sur un canal

Question

J'ai plusieurs goroutines qui essaient de recevoir simultanément sur le même canal. Il semble que la dernière goroutine qui commence à recevoir sur le canal obtienne la valeur. Est-ce quelque part dans les spécifications du langage ou est-ce un comportement non défini ?

c := make(chan string)
for i := 0; i < 5; i++ {
    go func(i int) {
        <-c
        c <- fmt.Sprintf("goroutine %d", i)
    }(i)
}
c <- "hi"
fmt.Println(<-c)

Sortie :

goroutine 4

Exemple sur le terrain de jeu

EDITAR:

Je viens de réaliser que c'est plus compliqué que je ne le pensais. Le message est transmis à toutes les goroutines.

c := make(chan string)
for i := 0; i < 5; i++ {
    go func(i int) {
        msg := <-c
        c <- fmt.Sprintf("%s, hi from %d", msg, i)
    }(i)
}
c <- "original"
fmt.Println(<-c)

Sortie :

original, hi from 0, hi from 1, hi from 2, hi from 3, hi from 4

NOTE : la sortie ci-dessus est périmée dans les versions plus récentes de Go (voir commentaires)

Exemple sur le terrain de jeu

Answer 1

Oui, c'est compliqué, mais il existe quelques règles empiriques qui devraient rendre les choses beaucoup plus simples.

• préfèrent utiliser des arguments formels pour les canaux que vous passez aux go-routines au lieu d'accéder aux canaux dans la portée globale. Vous pouvez obtenir plus de vérification du compilateur de cette façon, et une meilleure modularité aussi.
• éviter la lecture et l'écriture sur le même canal dans un go-routine particulier (y compris le "principal"). Sinon, le risque de blocage est beaucoup plus grand.

Voici une version alternative de votre programme, en appliquant ces deux directives. Ce cas démontre plusieurs rédacteurs et un lecteur sur une chaîne :

c := make(chan string)

for i := 1; i <= 5; i++ {
    go func(i int, co chan<- string) {
        for j := 1; j <= 5; j++ {
            co <- fmt.Sprintf("hi from %d.%d", i, j)
        }
    }(i, c)
}

for i := 1; i <= 25; i++ {
    fmt.Println(<-c)
}

http://play.golang.org/p/quQn7xePLw

Il crée les cinq go-routines écrivant sur un seul canal, chacune écrivant cinq fois. La go-routine principale lit les vingt-cinq messages - vous remarquerez que l'ordre dans lequel ils apparaissent n'est souvent pas séquentiel (c'est-à-dire que la concurrence est évidente).

Cet exemple démontre une caractéristique des canaux de Go : il est possible d'avoir plusieurs rédacteurs partageant un canal ; Go intercalera les messages automatiquement.

Il en va de même pour un écrivain et plusieurs lecteurs sur un même canal, comme le montre le deuxième exemple ici :

c := make(chan int)
var w sync.WaitGroup
w.Add(5)

for i := 1; i <= 5; i++ {
    go func(i int, ci <-chan int) {
        j := 1
        for v := range ci {
            time.Sleep(time.Millisecond)
            fmt.Printf("%d.%d got %d\n", i, j, v)
            j += 1
        }
        w.Done()
    }(i, c)
}

for i := 1; i <= 25; i++ {
    c <- i
}
close(c)
w.Wait()

Ce site deuxième exemple comprend une attente imposée à la goroutine principale, qui sinon se terminerait rapidement et entraînerait la fin anticipée des cinq autres goroutines. (merci à olov pour cette correction) .

Dans les deux exemples, aucune mise en mémoire tampon n'a été nécessaire. En règle générale, il est bon de considérer la mise en mémoire tampon uniquement comme un moyen d'améliorer les performances. Si votre programme ne se bloque pas sans les tampons, il n'y aura pas de blocage avec tampons non plus (mais l'inverse n'est pas toujours vrai). Ainsi, comme une autre règle de base, commencez sans mise en mémoire tampon, puis ajoutez-la plus tard si nécessaire. .

Answer 2

Réponse tardive, mais j'espère que cela aidera d'autres personnes à l'avenir. Sondage long, bouton "global", diffusion à tous ?

Go efficace explique le problème :

Les récepteurs bloquent toujours jusqu'à ce qu'il y ait des données à recevoir.

Cela signifie que vous ne pouvez pas avoir plus d'une goroutine qui écoute un canal et vous attendre à ce que TOUTES les goroutines reçoivent la même valeur.

Exécutez ceci Exemple de code .

package main

import "fmt"

func main() {
    c := make(chan int)

    for i := 1; i <= 5; i++ {
        go func(i int) {
        for v := range c {
                fmt.Printf("count %d from goroutine #%d\n", v, i)
            }
        }(i)
    }

    for i := 1; i <= 25; i++ {
        c<-i
    }

    close(c)
}

Vous ne verrez pas "count 1" plus d'une fois même s'il y a 5 goroutines qui écoutent le canal. Ceci est dû au fait que lorsque la première goroutine bloque le canal, toutes les autres goroutines doivent attendre en ligne. Lorsque le canal est débloqué, le compte a déjà été reçu et retiré du canal, de sorte que la goroutine suivante dans la file d'attente reçoit la valeur suivante du compte.

Answer 3

J'ai étudié les solutions existantes et créé une bibliothèque de diffusion simple. https://github.com/grafov/bcast .

    group := bcast.NewGroup() // you created the broadcast group
    go bcast.Broadcasting(0) // the group accepts messages and broadcast it to all members

    member := group.Join() // then you join member(s) from other goroutine(s)
    member.Send("test message") // or send messages of any type to the group 

    member1 := group.Join() // then you join member(s) from other goroutine(s)
    val := member1.Recv() // and for example listen for messages

Answer 4

C'est compliqué.

Aussi, voyez ce qui se passe avec GOMAXPROCS = NumCPU+1 . Par exemple,

package main

import (
    "fmt"
    "runtime"
)

func main() {
    runtime.GOMAXPROCS(runtime.NumCPU() + 1)
    fmt.Print(runtime.GOMAXPROCS(0))
    c := make(chan string)
    for i := 0; i < 5; i++ {
        go func(i int) {
            msg := <-c
            c <- fmt.Sprintf("%s, hi from %d", msg, i)
        }(i)
    }
    c <- ", original"
    fmt.Println(<-c)
}

Sortie :

5, original, hi from 4

Et, voyez ce qui se passe avec les canaux tamponnés. Par exemple,

package main

import "fmt"

func main() {
    c := make(chan string, 5+1)
    for i := 0; i < 5; i++ {
        go func(i int) {
            msg := <-c
            c <- fmt.Sprintf("%s, hi from %d", msg, i)
        }(i)
    }
    c <- "original"
    fmt.Println(<-c)
}

Sortie :

original

Vous devriez être en mesure d'expliquer ces cas également.

Answer 5

Pour l'écoute de plusieurs goroutines sur un canal, oui, c'est possible. Le point clé est le message lui-même, vous pouvez définir un message comme ça :

package main

import (
    "fmt"
    "sync"
)

type obj struct {
    msg string
    receiver int
}

func main() {
    ch := make(chan *obj) // both block or non-block are ok
    var wg sync.WaitGroup
    receiver := 25 // specify receiver count

    sender := func() {
        o := &obj {
            msg: "hello everyone!",
            receiver: receiver,
        }
        ch <- o
    }
    recv := func(idx int) {
        defer wg.Done()
        o := <-ch
        fmt.Printf("%d received at %d\n", idx, o.receiver)
        o.receiver--
        if o.receiver > 0 {
            ch <- o // forward to others
        } else {
            fmt.Printf("last receiver: %d\n", idx)
        }
    }

    go sender()
    for i:=0; i<reciever; i++ {
        wg.Add(1)
        go recv(i)
    }

    wg.Wait()
}

La sortie est aléatoire :

5 received at 25
24 received at 24
6 received at 23
7 received at 22
8 received at 21
9 received at 20
10 received at 19
11 received at 18
12 received at 17
13 received at 16
14 received at 15
15 received at 14
16 received at 13
17 received at 12
18 received at 11
19 received at 10
20 received at 9
21 received at 8
22 received at 7
23 received at 6
2 received at 5
0 received at 4
1 received at 3
3 received at 2
4 received at 1
last receiver 4