Go Channels freigeschaltet: Wie sie funktionieren

Channel: Eine wichtige Funktion in Golang und eine wichtige Verkörperung des Golang CSP-Concurrency-Modells

Channel ist eine sehr wichtige Funktion in Golang und auch eine wichtige Manifestation des Golang CSP-Concurrency-Modells. Vereinfacht gesagt, kann die Kommunikation zwischen Goroutinen über Channels erfolgen.

Channel ist in Golang so wichtig und wird im Code so häufig verwendet, dass man nicht umhin kann, neugierig auf seine interne Implementierung zu sein. Dieser Artikel analysiert die internen Implementierungsprinzipien von Channels auf der Grundlage des Quellcodes von Go 1.13.

Grundlegende Verwendung von Channel

Bevor wir die Implementierung von Channels formal analysieren, werfen wir zunächst einen Blick auf die grundlegendste Verwendung von Channels. Der Code lautet wie folgt:

package main
import "fmt"

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

    go func() {
        c <- 1 // send to channel
    }()

    x := <-c // recv from channel

    fmt.Println(x)
}

Im obigen Code erstellen wir einen Channel vom Typ int über make(chan int). In einer Goroutine verwenden wir c <- 1, um Daten an den Channel zu senden. In der Main-Goroutine lesen wir Daten aus dem Channel über x := <- c und weisen sie x zu.

Der obige Code entspricht zwei grundlegenden Operationen von Channels:

Die send-Operation c <- 1, die das Senden von Daten an den Channel bedeutet.
Die recv-Operation x := <- c, die das Empfangen von Daten aus dem Channel bedeutet.

Darüber hinaus werden Channels in gepufferte und ungepufferte Channels unterteilt. Im obigen Code verwenden wir einen ungepufferten Channel. Bei einem ungepufferten Channel blockiert der Sender an der Sendeanweisung, wenn derzeit keine andere Goroutine Daten aus dem Channel empfängt.

Wir können die Puffergröße bei der Initialisierung des Channels angeben. Beispielsweise gibt make(chan int, 2) eine Puffergröße von 2 an. Bevor der Puffer voll ist, kann der Sender Daten an den Channel senden, ohne zu blockieren, und muss nicht warten, bis der Empfänger bereit ist. Wenn der Puffer jedoch voll ist, blockiert der Sender weiterhin.

Zugrunde liegende Implementierungsfunktionen von Channel

Bevor wir den Quellcode von Channels untersuchen, müssen wir zunächst herausfinden, wo sich die spezifische Implementierung von Channels in Golang befindet. Denn wenn wir Channels verwenden, verwenden wir das Symbol <- und können seine Implementierung nicht direkt im Go-Quellcode finden. Der Golang-Compiler übersetzt das Symbol <- jedoch sicherlich in die entsprechende zugrunde liegende Implementierung.

Wir können den integrierten Go-Befehl verwenden: go tool compile -N -l -S hello.go, um den Code in entsprechende Assembly-Anweisungen zu übersetzen.

Alternativ können wir direkt das Online-Tool Compiler Explorer verwenden. Für den obigen Beispielcode können Sie seine Assembly-Ergebnisse direkt unter diesem Link anzeigen: go.godbolt.org/z/3xw5Cj. Wie in der folgenden Abbildung dargestellt:

Channel-Assembly-Anweisungen

Durch sorgfältiges Untersuchen der Assembly-Anweisungen, die dem obigen Beispielcode entsprechen, können die folgenden Entsprechungen festgestellt werden:

Die Channel-Konstruktionsanweisung make(chan int) entspricht der Funktion runtime.makechan.
Die Sendeanweisung c <- 1 entspricht der Funktion runtime.chansend1.
Die Empfangsanweisung x := <- c entspricht der Funktion runtime.chanrecv1.

Die Implementierungen der obigen Funktionen befinden sich alle in der Codedatei runtime/chan.go im Go-Quellcode. Als Nächstes werden wir die Implementierung von Channels untersuchen, indem wir diese Funktionen anvisieren.

Channel-Konstruktion

Die Channel-Konstruktionsanweisung make(chan int) wird vom Golang-Compiler in die Funktion runtime.makechan übersetzt, und ihre Funktionssignatur lautet wie folgt:

func makechan(t *chantype, size int) *hchan

Hier ist t *chantype der Elementtyp, der beim Erstellen des Channels übergeben wird. size int ist die vom Benutzer angegebene Puffergröße des Channels, die 0 ist, falls nicht angegeben. Der Rückgabewert dieser Funktion ist *hchan. hchan ist die interne Implementierung von Channels in Golang. Seine Definition lautet wie folgt:

type hchan struct {
        qcount   uint           // Die Anzahl der bereits im Puffer platzierten Elemente
        dataqsiz uint           // Die beim Erstellen des Channels vom Benutzer angegebene Puffergröße
        buf      unsafe.Pointer // Puffer
        elemsize uint16         // Die Größe jedes Elements im Puffer
        closed   uint32         // Gibt an, ob der Channel geschlossen ist, == 0 bedeutet nicht geschlossen
        elemtype *_type         // Typinformationen von Channel-Elementen
        sendx    uint           // Die Indexposition der gesendeten Elemente im Puffer Sendeindex
        recvx    uint           // Die Indexposition der empfangenen Elemente im Puffer Empfangsindex
        recvq    waitq          // Liste der Goroutinen, die auf den Empfang warten Empfangs-Worker
        sendq    waitq          // Liste der Goroutinen, die auf das Senden warten Sende-Worker

        lock mutex
}

Alle Attribute in hchan lassen sich grob in drei Kategorien einteilen:

Pufferbezogene Attribute: wie z. B. buf, dataqsiz, qcount usw. Wenn die Puffergröße des Channels nicht 0 ist, speichert der Puffer die zu empfangenden Daten. Er wird mithilfe eines Ringpuffers implementiert.
waitq-bezogene Attribute: Er kann als Standard-FIFO-Warteschlange verstanden werden. Unter diesen enthält recvq Goroutinen, die auf den Empfang von Daten warten, und sendq Goroutinen, die auf das Senden von Daten warten. waitq wird mithilfe einer doppelt verketteten Liste implementiert.
Andere Attribute: wie z. B. lock, elemtype, closed usw.

Der gesamte Prozess von makechan besteht im Wesentlichen aus einigen Legalitätsprüfungen und Speicherzuweisung für Puffer, hchan und andere Attribute, und wir werden dies hier nicht eingehend erörtern. Interessierte können sich den Quellcode hier direkt ansehen.

Durch einfaches Analysieren der Attribute von hchan können wir erkennen, dass es zwei wichtige Komponenten gibt, buffer und waitq. Alle Verhaltensweisen und Implementierungen von hchan drehen sich um diese beiden Komponenten.

Senden von Daten in den Channel

Die Sende- und Empfangsprozesse von Channels sind sehr ähnlich. Analysieren wir zunächst den Sendeprozess von Channels (z. B. c <- 1), der der Implementierung der Funktion runtime.chansend entspricht.

Beim Versuch, Daten an einen Channel zu senden, wird zunächst eine Goroutine, die auf den Empfang von Daten wartet, aus dem Kopf von recvq entnommen, wenn die Warteschlange recvq nicht leer ist. Und die Daten werden direkt an diese Goroutine gesendet. Der Code lautet wie folgt:

if sg := c.recvq.dequeue(); sg!= nil {
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
}

recvq enthält Goroutinen, die auf den Empfang von Daten warten. Wenn eine Goroutine die Operation recv verwendet (z. B. x := <- c), wird diese Goroutine und die Adresse der zu empfangenden Daten in ein sudog-Objekt verpackt und in recvq eingefügt, wenn sich zu diesem Zeitpunkt keine Daten im Cache des Channels befinden und keine andere Goroutine auf das Senden von Daten wartet (d. h. sendq ist leer).

Fahren wir mit dem obigen Code fort: Wenn recvq zu diesem Zeitpunkt nicht leer ist, wird die Funktion send aufgerufen, um die Daten auf den Stack der entsprechenden Goroutine zu kopieren.

Die Implementierung der Funktion send umfasst hauptsächlich zwei Punkte:

memmove(dst, src, t.size) führt Datenübertragung durch, was im Wesentlichen eine Speicherkopie ist.
goready(gp, skip+1) Die Funktion von goready besteht darin, die entsprechende Goroutine zu reaktivieren.

Wenn die Warteschlange recvq leer ist, bedeutet dies, dass zu diesem Zeitpunkt keine Goroutine auf den Empfang von Daten wartet. Anschließend versucht der Channel, die Daten in den Puffer zu legen. Der Code lautet wie folgt:

if c.qcount < c.dataqsiz {
        // Äquivalent zu c.buf[c.sendx]
        qp := chanbuf(c, c.sendx)
        // Kopieren der Daten in den Puffer
        typedmemmove(c.elemtype, qp, ep)
        c.sendx++
        if c.sendx == c.dataqsiz {
                c.sendx = 0
        }
        c.qcount++
        unlock(&c.lock)
        return true
}

Die Funktion des obigen Codes ist eigentlich sehr einfach, d. h. die Daten einfach in den Puffer zu legen. Dieser Prozess umfasst die Operation des Ringpuffers, wobei dataqsiz die vom Benutzer angegebene Puffergröße des Channels darstellt, die standardmäßig 0 ist, falls nicht angegeben. Andere spezifische detaillierte Operationen werden später im Abschnitt zu Ringpuffern ausführlich beschrieben.

Wenn der Benutzer einen ungepufferten Channel verwendet oder der Puffer zu diesem Zeitpunkt voll ist, wird die Bedingung c.qcount < c.dataqsiz nicht erfüllt, und der obige Prozess wird nicht ausgeführt. Zu diesem Zeitpunkt werden die aktuelle Goroutine und die zu sendenden Daten in die Warteschlange sendq eingefügt, und diese Goroutine wird gleichzeitig herausgeschnitten. Der gesamte Prozess entspricht dem folgenden Code:

gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0!= 0 {
        mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.waiting = mysg
gp.param = nil
c.sendq.enqueue(mysg)
// Wechseln der Goroutine in den Wartezustand und Entsperren
goparkunlock(&c.lock, waitReasonChanSend, traceEvGoBlockSend, 3)

Im obigen Code entsperrt goparkunlock den eingegebenen Mutex und schneidet diese Goroutine heraus, wodurch diese Goroutine in den Wartezustand versetzt wird. gopark und goready oben entsprechen einander und sind inverse Operationen. gopark und goready werden häufig im Quellcode der Laufzeitumgebung angetroffen und umfassen den Scheduling-Prozess von Goroutinen, der hier nicht eingehend erörtert wird, und es wird später ein separater Artikel darüber geschrieben.

Nach dem Aufrufen von gopark blockiert aus Benutzersicht die Codeanweisung zum Senden von Daten an den Channel.

Der obige Prozess ist der interne Workflow der Sendeanweisung des Channels (z. B. c <- 1), und der gesamte Sendeprozess verwendet c.lock zum Sperren, um die gleichzeitige Sicherheit zu gewährleisten.

In einfachen Worten ist der gesamte Prozess wie folgt:

Überprüfen Sie, ob recvq leer ist. Wenn sie nicht leer ist, nehmen Sie eine Goroutine aus dem Kopf von recvq, senden Sie Daten an sie und aktivieren Sie die entsprechende Goroutine.
Wenn recvq leer ist, legen Sie die Daten in den Puffer.
Wenn der Puffer voll ist, verpacken Sie die zu sendenden Daten und die aktuelle Goroutine in ein sudog-Objekt und legen Sie es in sendq. Und setzen Sie die aktuelle Goroutine in den Wartezustand.

Der Prozess des Empfangens von Daten aus dem Channel

Der Prozess des Empfangens von Daten aus einem Channel ist im Wesentlichen dem Sendeprozess ähnlich und wird hier nicht wiederholt. Die spezifischen pufferbezogenen Operationen, die am Empfangsprozess beteiligt sind, werden später ausführlich beschrieben.

Hier ist zu beachten, dass die gesamten Sende- und Empfangsprozesse von Channels runtime.mutex zum Sperren verwenden. runtime.mutex ist eine Lightweight-Sperre, die häufig im Quellcode der Laufzeitumgebung verwendet wird. Der gesamte Prozess ist nicht der effizienteste sperrenfreie Ansatz. In Golang gibt es ein Problem: go/issues#8899, das eine sperrenfreie Channel-Lösung bietet.

Implementierung des Ringpuffers des Channels

Channels verwenden Ringpuffer, um geschriebene Daten zwischenzuspeichern. Ringpuffer haben viele Vorteile und eignen sich sehr gut für die Implementierung von FIFO-Warteschlangen fester Länge.

In Channels sieht die Implementierung des Ringpuffers wie folgt aus:

Implementierung des Ringpuffers im Channel

Es gibt zwei Variablen, die sich auf den Puffer in hchan beziehen: recvx und sendx. Unter diesen stellt sendx den beschreibbaren Index im Puffer dar, und recvx stellt den lesbaren Index im Puffer dar. Die Elemente zwischen recvx und sendx stellen die Daten dar, die normalerweise im Puffer platziert wurden.

Wir können buf[recvx] direkt verwenden, um das erste Element der Warteschlange zu lesen, und buf[sendx] = x verwenden, um Elemente am Ende der Warteschlange zu platzieren.

Pufferschreiben

Wenn der Puffer nicht voll ist, sieht die Operation zum Einfügen von Daten in den Puffer wie folgt aus:

qp := chanbuf(c, c.sendx)
// Kopieren der Daten in den Puffer
typedmemmove(c.elemtype, qp, ep)
c.sendx++
if c.sendx == c.dataqsiz {
        c.sendx = 0
}
c.qcount++

Hier ist chanbuf(c, c.sendx) äquivalent zu c.buf[c.sendx]. Der obige Prozess ist sehr einfach, d. h. die Daten an die Position von sendx im Puffer zu kopieren.

Bewegen Sie dann sendx an die nächste Position. Wenn sendx die letzte Position erreicht hat, setzen Sie es auf 0, was eine typische Head-to-Tail-Verbindungsmethode ist.

Pufferlesen

Wenn der Puffer nicht voll ist, muss sendq zu diesem Zeitpunkt auch leer sein (denn wenn der Puffer nicht voll ist, reiht sich die zum Senden von Daten verwendete Goroutine nicht ein, sondern legt die Daten direkt in den Puffer. Spezifische Logik finden Sie im obigen Abschnitt zum Senden von Daten an den Channel). Zu diesem Zeitpunkt ist der Leseprozess chanrecv des Channels relativ einfach, und Daten können direkt aus dem Puffer gelesen werden, was auch ein Prozess des Verschiebens von recvx ist. Sie ist im Wesentlichen die gleiche wie das obige Pufferschreiben.

Wenn sich wartende Goroutinen in sendq befinden, muss der Puffer zu diesem Zeitpunkt voll sein. Die Leselogik des Channels sieht zu diesem Zeitpunkt wie folgt aus:

// Äquivalent zu c.buf[c.recvx]
qp := chanbuf(c, c.recvx)
// Kopieren von Daten aus der Warteschlange an den Empfänger
if ep!= nil {
        typedmemmove(c.elemtype, ep, qp)
}
// Kopieren von Daten vom Sender in die Warteschlange
typedmemmove(c.elemtype, qp, sg.elem)
c.recvx++
if c.recvx == c.dataqsiz {
        c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz

Im obigen Code ist ep die Adresse, die der Variablen entspricht, die Daten empfängt. Beispielsweise stellt in x := <- c ep die Adresse der Variablen x dar. Und sg stellt das erste sudog dar, das aus sendq entnommen wurde. Und:

typedmemmove(c.elemtype, ep, qp) bedeutet, das aktuell lesbare Element im Puffer an die Adresse der empfangenden Variablen zu kopieren.
typedmemmove(c.elemtype, qp, sg.elem) bedeutet, die Daten, die von der Goroutine in sendq gesendet werden sollen, in den Puffer zu kopieren. Da recv++ später ausgeführt wird, entspricht dies dem Einfügen der Daten in sendq am Ende der Warteschlange.

In einfachen Worten: Hier kopiert der Channel die ersten Daten im Puffer in die entsprechende empfangende Variable und kopiert gleichzeitig die Elemente in sendq an das Ende der Warteschlange, sodass Daten in FIFO (First In First Out) verarbeitet werden können.

Zusammenfassung

Als eine der am häufigsten verwendeten Einrichtungen in Golang kann das Verständnis des Quellcodes von Channels uns helfen, diese besser zu verstehen und zu verwenden. Gleichzeitig werden wir nicht übermäßiggläubisch und abhängig von der Leistung von Channels sein. Es gibt noch viel Raum für Optimierung im aktuellen Design von Channels.

Optimierungshinweise:

Titel (mit # und ## usw.) werden verwendet, um den Artikelinhalt zu strukturieren und die Struktur übersichtlicher zu gestalten.
Codeblöcke sind deutlich gekennzeichnet (mit go), wodurch die Lesbarkeit des Codes verbessert wird.
Die Kommentare in den Codeblöcken werden separat aufgeführt, was die Erläuterung der Code-Logik verdeutlicht und den Einfluss von Kommentaren in den Codeblöcken auf das Leseerlebnis vermeidet.
Einige wichtige Teile werden in Punkten dargestellt, wodurch komplexe Logik leichter verständlich wird, beispielsweise der Sendeprozess von Channels.
Zu einigen Inhalten werden Hyperlinks hinzugefügt, um den Lesern das Abrufen relevanter Materialien zu erleichtern.