Questions d'entretien délicates avec Golang - Part Data Race

Voici une autre question d'entretien de révision de code pour vous. Cette question est plus avancée que les précédentes et s’adresse à un public plus senior. Le problème nécessite la connaissance des tranches et le partage de données entre processus parallèles.

Si vous n'êtes pas familier avec les tranches et la façon dont elles sont construites, veuillez consulter mon article précédent sur l'en-tête de tranche

Qu’est-ce qu’une course aux données ?

Une course aux données se produit lorsque deux threads ou plus (ou goroutines, dans le cas de Go) accèdent simultanément à la mémoire partagée, et au moins un de ces accès est une opération d'écriture. S'il n'y a pas de mécanismes de synchronisation appropriés (tels que des verrous ou des canaux) pour gérer l'accès, le résultat peut être un comportement imprévisible, notamment une corruption des données, des états incohérents ou des plantages.

Essentiellement, une course aux données se produit lorsque :

• Deux threads ou plus (ou goroutines) accèdent au même emplacement mémoire en même temps.
• Au moins un des threads (ou goroutines) écrit dans cette mémoire.
• Il n'y a pas de synchronisation pour contrôler l'accès à cette mémoire.

Pour cette raison, l'ordre dans lequel les threads ou les goroutines accèdent ou modifient la mémoire partagée est imprévisible, conduisant à un comportement non déterministe qui peut varier d'une exécution à l'autre.

     +----------------------+      +---------------------+
     | Thread A: Write      |      | Thread B: Read      |
     +----------------------+      +---------------------+
     | 1. Reads x           |      | 1. Reads x          |
     | 2. Adds 1 to x       |      |                     |
     | 3. Writes new value  |      |                     |
     +----------------------+      +---------------------+

                    Shared variable x
                    (Concurrent access without synchronization)

Ici, le fil A modifie x (y écrit), tandis que le fil B le lit en même temps. Si les deux threads s'exécutent simultanément et qu'il n'y a pas de synchronisation, le thread B pourrait lire x avant que le thread A n'ait fini de le mettre à jour. En conséquence, les données pourraient être incorrectes ou incohérentes.

Question : un de vos coéquipiers a soumis le code suivant pour une révision de code. Veuillez lire attentivement le code et identifier tout problème potentiel.
Et voici le code que vous devez revoir :

package main  

import (  
    "bufio"  
    "bytes"
    "io"
    "math/rand"
    "time"
)  

func genData() []byte {  
    r := rand.New(rand.NewSource(time.Now().Unix()))  
    buffer := make([]byte, 512)  
    if _, err := r.Read(buffer); err != nil {  
       return nil  
    }  
    return buffer  
}  

func publish(input []byte, output chan<- []byte) {  
    reader := bytes.NewReader(input)  

    bufferSize := 8  
    buffer := make([]byte, bufferSize)  

    for {  
       n, err := reader.Read(buffer)  
       if err != nil || err == io.EOF {  
          return  
       }  
       if n == 0 {  
          break  
       }  
       output <- buffer[:n]  
    }  
}  

func consume(input []byte) {  
    scanner := bufio.NewScanner(bytes.NewReader(input))  
    for scanner.Scan() {  
       b := scanner.Bytes()  
       _ = b  
       // does the magic  
    }  
}  

func main() {  
    data := genData()  
    workersCount := 4  
    chunkChannel := make(chan []byte, workersCount)  

    for i := 0; i < workersCount; i++ {  
       go func() {  
          for chunk := range chunkChannel {  
             consume(chunk)  
          }  
       }()  
    }  

    publish(data, chunkChannel)  
    close(chunkChannel)  
}

Qu'est-ce qu'on a ici ?

La fonction publier() est responsable de la lecture des données d'entrée morceau par morceau et de l'envoi de chaque morceau au canal de sortie. Cela commence par utiliser bytes.NewReader(input) pour créer un lecteur à partir des données d'entrée, ce qui permet de lire les données de manière séquentielle. Un tampon de taille 8 est créé pour contenir chaque bloc de données lors de sa lecture à partir de l'entrée. Au cours de chaque itération, reader.Read(buffer) lit jusqu'à 8 octets de l'entrée, et la fonction envoie ensuite une tranche de ce tampon (buffer[:n]) contenant jusqu'à 8 octets au canal de sortie. La boucle continue jusqu'à ce que reader.Read(buffer) rencontre une erreur ou atteigne la fin des données d'entrée.

La fonction consume() gère les morceaux de données reçus du canal. Il traite ces morceaux à l'aide d'un bufio.Scanner, qui analyse chaque morceau de données, le divisant potentiellement en lignes ou en jetons selon la façon dont il est configuré. La variable b := scanner.Bytes() récupère le jeton en cours d'analyse. Cette fonction représente un traitement d'entrée de base.

Main() crée un chunkChannel de canal tamponné avec une capacité égale à WorkersCount, qui est définie sur 4 dans ce cas. La fonction lance ensuite 4 goroutines de travail, dont chacune lira simultanément les données du chunkChannel. Chaque fois qu'un travailleur reçoit un morceau de données, il traite ce morceau en appelant la fonction consume(). La fonction publier() lit les données générées, les divise en morceaux allant jusqu'à 8 octets et les envoie au canal.

Le programme utilise des goroutines pour créer plusieurs consommateurs, permettant un traitement simultané des données. Chaque consommateur s'exécute dans une goroutine distincte, traitant des morceaux de données indépendamment.

Si vous exécutez ce code, une observation suspecte se produira :

[Running] go run "main.go"

[Done] exited with code=0 in 0.94 seconds

Mais il y a un problème. Nous avons un Risque de course aux données. Dans ce code, il existe une course potentielle aux données car la fonction publier() réutilise la même tranche de tampon pour chaque morceau. Les consommateurs lisent simultanément dans ce tampon, et comme les tranches partagent la mémoire sous-jacente, plusieurs consommateurs pourraient lire la même mémoire, conduisant à une course aux données. Essayons d'utiliser une détection de race. Go fournit un outil intégré pour détecter les courses de données : le détecteur de courses. Vous pouvez l'activer en exécutant votre programme avec le drapeau -race :

go run -race main.go

Si nous ajoutons l'indicateur -race à la commande run, nous recevrons le résultat suivant :

[Running] go run -race "main.go"

==================
WARNING: DATA RACE
Read at 0x00c00011e018 by goroutine 6:
  runtime.slicecopy()
      /GOROOT/go1.22.0/src/runtime/slice.go:325 +0x0
  bytes.(*Reader).Read()
      /GOROOT/go1.22.0/src/bytes/reader.go:44 +0xcc
  bufio.(*Scanner).Scan()
      /GOROOT/go1.22.0/src/bufio/scan.go:219 +0xef4
  main.consume()
      /GOPATH/example/main.go:40 +0x140
  main.main.func1()
      /GOPATH/example/main.go:55 +0x48

Previous write at 0x00c00011e018 by main goroutine:
  runtime.slicecopy()
      /GOROOT/go1.22.0/src/runtime/slice.go:325 +0x0
  bytes.(*Reader).Read()
      /GOROOT/go1.22.0/src/bytes/reader.go:44 +0x168
  main.publish()
      /GOPATH/example/main.go:27 +0xe4
  main.main()
      /GOPATH/example/main.go:60 +0xdc

Goroutine 6 (running) created at:
  main.main()
      /GOPATH/example/main.go:53 +0x50
==================
Found 1 data race(s)
exit status 66

[Done] exited with code=0 in 0.94 seconds

The warning you’re seeing is a classic data race detected by Go’s race detector. The warning message indicates that two goroutines are accessing the same memory location (0x00c00011e018) concurrently. One goroutine is reading from this memory, while another goroutine is writing to it at the same time, without proper synchronization.

The first part of the warning tells us that Goroutine 6 (which is one of the worker goroutines in your program) is reading from the memory address 0x00c00011e018 during a call to bufio.Scanner.Scan() inside the consume() function.

Read at 0x00c00011e018 by goroutine 6:
  runtime.slicecopy()
  /GOROOT/go1.22.0/src/runtime/slice.go:325 +0x0
  bytes.(*Reader).Read()
  /GOROOT/go1.22.0/src/bytes/reader.go:44 +0xcc
  bufio.(*Scanner).Scan()
  /GOROOT/go1.22.0/src/bufio/scan.go:219 +0xef4
  main.consume()
  /GOPATH/example/main.go:40 +0x140
  main.main.func1()
  /GOPATH/example/main.go:55 +0x48

The second part of the warning shows that the main goroutine previously wrote to the same memory location (0x00c00011e018) during a call to bytes.Reader.Read() inside the publish() function.

Previous write at 0x00c00011e018 by main goroutine:
  runtime.slicecopy()
  /GOROOT/go1.22.0/src/runtime/slice.go:325 +0x0
  bytes.(*Reader).Read()
  /GOROOT/go1.22.0/src/bytes/reader.go:44 +0x168
  main.publish()
  /GOPATH/example/main.go:27 +0xe4
  main.main()
  /GOPATH/example/main.go:60 +0xdc

The final part of the warning explains that Goroutine 6 was created in the main function.

Goroutine 6 (running) created at:
  main.main()
  /GOPATH/example/main.go:53 +0x50

In this case, while one goroutine (Goroutine 6) is reading from the buffer in consume(), the publish() function in the main goroutine is simultaneously writing to the same buffer, leading to the data race.

+-------------------+               +--------------------+
|     Publisher     |               |      Consumer      |
+-------------------+               +--------------------+
        |                                   |
        v                                   |
1. Read data into buffer                    |
        |                                   |
        v                                   |
2. Send slice of buffer to chunkChannel     |
        |                                   |
        v                                   |
 +----------------+                         |
 |  chunkChannel  |                         |
 +----------------+                         |
        |                                   |
        v                                   |
3. Consume reads from slice                 |
                                            v
                                    4. Concurrent access
                                    (Data Race occurs)

Why the Data Race Occurs

The data race in this code arises because of how Go slices work and how memory is shared between goroutines when a slice is reused. To fully understand this, let’s break it down into two parts: the behavior of the buffer slice and the mechanics of how the race occurs. When you pass a slice like buffer[:n] to a function or channel, what you are really passing is the slice header which contains a reference to the slice’s underlying array. Any modifications to the slice or the underlying array will affect all other references to that slice.

buffer = [ a, b, c, d, e, f, g, h ]  <- Underlying array
           ↑
          Slice: buffer[:n]

func publish(input []byte, output chan<- []byte) {  
    reader := bytes.NewReader(input)  

    bufferSize := 8  
    buffer := make([]byte, bufferSize)  

    for {  
       // ....
       output <- buffer[:n] // <-- passing is a reference to the underlying array
    }  
}

If you send buffer[:n] to a channel, both the publish() function and any consumer goroutines will be accessing the same memory. During each iteration, the reader.Read(buffer) function reads up to 8 bytes from the input data into this buffer slice. After reading, the publisher sends buffer[:n] to the output channel, where n is the number of bytes read in the current iteration.

The problem here is that buffer is reused across iterations. Every time reader.Read() is called, it overwrites the data stored in buffer.

• Iteration 1: publish() function reads the first 8 bytes into buffer and sends buffer[:n] (say, [a, b, c, d, e, f, g, h]) to the channel.
• Iteration 2: The publish() function overwrites the buffer with the next 8 bytes, let’s say [i, j, k, l, m, n, o, p], and sends buffer[:n] again.

At this point, if one of the worker goroutines is still processing the first chunk, it is now reading stale or corrupted data because the buffer has been overwritten by the second chunk. Reusing a slice neans sharing the same memory.

How to fix the Data Race?

To avoid the race condition, we must ensure that each chunk of data sent to the channel has its own independent memory. This can be achieved by creating a new slice for each chunk and copying the data from the buffer to this new slice. The key fix is to copy the contents of the buffer into a new slice before sending it to the chunkChannel:

chunk := make([]byte, n)    // Step 1: Create a new slice with its own memory
copy(chunk, buffer[:n])     // Step 2: Copy data from buffer to the new slice
output <- chunk             // Step 3: Send the new chunk to the channel

Why this fix works? By creating a new slice (chunk) for each iteration, you ensure that each chunk has its own memory. This prevents the consumers from reading from the buffer that the publisher is still modifying. copy() function copies the contents of the buffer into the newly allocated slice (chunk). This decouples the memory used by each chunk from the buffer. Now, when the publisher reads new data into the buffer, it doesn’t affect the chunks that have already been sent to the channel.

+-------------------------+           +------------------------+
|  Publisher (New Memory) |           | Consumers (Read Copy)  |
|  [ a, b, c ] --> chunk1 |           |  Reading: chunk1       |
|  [ d, e, f ] --> chunk2 |           |  Reading: chunk2       |
+-------------------------+           +------------------------+
         ↑                                    ↑
        (1)                                  (2)
   Publisher Creates New Chunk          Consumers Read Safely

This solution works is that it breaks the connection between the publisher and the consumers by eliminating shared memory. Each consumer now works on its own copy of the data, which the publisher does not modify. Here’s how the modified publish() function looks:

func publish(input []byte, output chan<- []byte) {
    reader := bytes.NewReader(input)
    bufferSize := 8
    buffer := make([]byte, bufferSize)

    for {
        n, err := reader.Read(buffer)
        if err != nil || err == io.EOF {
            return
        }
        if n == 0 {
            break
        }

        // Create a new slice for each chunk and copy the data from the buffer
        chunk := make([]byte, n)
        copy(chunk, buffer[:n])

        // Send the newly created chunk to the channel
        output <- chunk
    }
}

Summary

Slices Are Reference Types:
As mentioned earlier, Go slices are reference types, meaning they point to an underlying array. When you pass a slice to a channel or a function, you’re passing a reference to that array, not the data itself. This is why reusing a slice leads to a data race: multiple goroutines end up referencing and modifying the same memory.

Allocation de mémoire :
Lorsque nous créons une nouvelle tranche avec make([]byte, n), Go alloue un bloc de mémoire distinct pour cette tranche. Cela signifie que la nouvelle tranche (morceau) possède son propre tableau de sauvegarde, indépendant du tampon. En copiant les données du buffer[:n] dans le chunk, nous garantissons que chaque chunk dispose de son propre espace mémoire privé.

Découplage de la mémoire :
En découplant la mémoire de chaque fragment du tampon, l'éditeur peut continuer à lire de nouvelles données dans le tampon sans affecter les fragments déjà envoyés au canal. Chaque morceau possède désormais sa propre copie indépendante des données, afin que les consommateurs puissent traiter les morceaux sans interférence de l'éditeur.

Prévenir les courses aux données :
La principale source de la course aux données était l’accès simultané au tampon partagé. En créant de nouvelles tranches et en copiant les données, on élimine la mémoire partagée, et chaque goroutine fonctionne sur ses propres données. Cela supprime la possibilité d'une condition de concurrence car il n'y a plus de conflit sur la même mémoire.

Conclusion

Le cœur du correctif est simple mais puissant : en garantissant que chaque bloc de données dispose de sa propre mémoire, nous éliminons la ressource partagée (le tampon) qui était à l'origine de la course aux données. Ceci est réalisé en copiant les données du tampon dans une nouvelle tranche avant de les envoyer au canal. Avec cette approche, chaque consommateur travaille sur sa propre copie des données, indépendamment des actions de l’éditeur, garantissant un traitement simultané sécurisé sans conditions de concurrence. Cette méthode de découplage de la mémoire partagée est une stratégie fondamentale en programmation concurrente. Il empêche les comportements imprévisibles provoqués par les conditions de concurrence et garantit que vos programmes Go restent sûrs, prévisibles et corrects, même lorsque plusieurs goroutines accèdent aux données simultanément.

C'est aussi simple que cela !

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!