Knifflige Golang-Interviewfragen – Teil Datenrennen

Hier ist eine weitere Frage zum Code-Review-Interview für Sie. Diese Frage ist anspruchsvoller als die vorherigen und richtet sich an ein älteres Publikum. Das Problem erfordert Kenntnisse über Slices und den Datenaustausch zwischen parallelen Prozessen.

Wenn Sie mit den Slices und ihrem Aufbau nicht vertraut sind, schauen Sie sich bitte meinen vorherigen Artikel über den Slice-Header an

Was ist ein Datenrennen?

Ein Datenwettlauf tritt auf, wenn zwei oder mehr Threads (oder Goroutinen im Fall von Go) gleichzeitig auf den gemeinsamen Speicher zugreifen und mindestens einer dieser Zugriffe ein Schreibvorgang ist. Wenn keine geeigneten Synchronisierungsmechanismen (z. B. Sperren oder Kanäle) zur Zugriffsverwaltung vorhanden sind, kann dies zu unvorhersehbarem Verhalten führen, einschließlich Datenbeschädigung, inkonsistenten Zuständen oder Abstürzen.

Im Wesentlichen findet ein Datenwettlauf statt, wenn:

• Zwei oder mehr Threads (oder Goroutinen) greifen gleichzeitig auf denselben Speicherort zu.
• Mindestens einer der Threads (oder Goroutinen) schreibt in diesen Speicher.
• Es gibt keine Synchronisierung, um den Zugriff auf diesen Speicher zu steuern.

Aus diesem Grund ist die Reihenfolge, in der die Threads oder Goroutinen auf den gemeinsamen Speicher zugreifen oder ihn ändern, unvorhersehbar, was zu nicht deterministischem Verhalten führt, das zwischen den Läufen variieren kann.

     +----------------------+      +---------------------+
     | 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)

Hier modifiziert Thread A x (schreibt darauf), während Thread B es gleichzeitig liest. Wenn beide Threads gleichzeitig ausgeführt werden und keine Synchronisierung erfolgt, könnte Thread B x lesen, bevor Thread A die Aktualisierung abgeschlossen hat. Infolgedessen könnten die Daten falsch oder inkonsistent sein.

Frage: Einer Ihrer Teamkollegen hat den folgenden Code zur Codeüberprüfung eingereicht. Bitte lesen Sie den Code sorgfältig durch und identifizieren Sie mögliche Probleme.
Und hier der Code, den Sie überprüfen müssen:

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)  
}

Was haben wir hier?

Die Funktion „publish()“ ist dafür verantwortlich, die Eingabedaten Block für Block zu lesen und jeden Block an den Ausgabekanal zu senden. Es beginnt mit der Verwendung von bytes.NewReader(input), um aus den Eingabedaten einen Reader zu erstellen, der das sequentielle Lesen der Daten ermöglicht. Es wird ein Puffer der Größe 8 erstellt, um jeden Datenblock zu speichern, während er aus der Eingabe gelesen wird. Während jeder Iteration liest „reader.Read(buffer)“ bis zu 8 Bytes aus der Eingabe, und die Funktion sendet dann einen Abschnitt dieses Puffers (buffer[:n]) mit bis zu 8 Bytes an den Ausgabekanal. Die Schleife wird fortgesetzt, bis „reader.Read(buffer)“ entweder auf einen Fehler stößt oder das Ende der Eingabedaten erreicht.

Die Funktion „consume()“ verarbeitet die vom Kanal empfangenen Datenblöcke. Es verarbeitet diese Datenblöcke mit einem bufio.Scanner, der jeden Datenblock scannt und ihn je nach Konfiguration möglicherweise in Zeilen oder Token aufteilt. Die Variable b := scanner.Bytes() ruft das aktuell gescannte Token ab. Diese Funktion stellt eine grundlegende Eingabeverarbeitung dar.

main() erstellt einen gepufferten Kanal chunkChannel mit einer Kapazität gleich WorkersCount, die in diesem Fall auf 4 gesetzt ist. Die Funktion startet dann vier Worker-Goroutinen, von denen jede gleichzeitig Daten aus dem chunkChannel liest. Jedes Mal, wenn ein Worker einen Datenblock empfängt, verarbeitet er den Block, indem er die Funktion „consume()“ aufruft. Die Funktion „publish()“ liest die generierten Daten, zerlegt sie in Blöcke von bis zu 8 Bytes und sendet sie an den Kanal.

Das Programm verwendet Goroutinen, um mehrere Verbraucher zu erstellen und so eine gleichzeitige Datenverarbeitung zu ermöglichen. Jeder Verbraucher läuft in einer separaten Goroutine und verarbeitet Datenblöcke unabhängig.

Wenn Sie diesen Code ausführen, wird Folgendes verdächtig festgestellt:

[Running] go run "main.go"

[Done] exited with code=0 in 0.94 seconds

Aber es gibt ein Problem. Wir haben ein Datenwettlaufrisiko. In diesem Code besteht ein potenzieller Datenwettlauf, da die Funktion „publish()“ für jeden Block denselben Pufferabschnitt wiederverwendet. Die Verbraucher lesen gleichzeitig aus diesem Puffer, und da Slices den zugrunde liegenden Speicher gemeinsam nutzen, könnten mehrere Verbraucher denselben Speicher lesen, was zu einem Datenwettlauf führt. Versuchen wir, eine Rassenerkennung zu verwenden. Go bietet ein integriertes Tool zur Erkennung von Datenrennen: den Rassendetektor. Sie können es aktivieren, indem Sie Ihr Programm mit der Flagge -race ausführen:

go run -race main.go

Wenn wir das Flag -race zum Ausführungsbefehl hinzufügen, erhalten wir die folgende Ausgabe:

[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.

Speicherzuweisung:
Wenn wir mit make([]byte, n) ein neues Slice erstellen, weist Go diesem Slice einen separaten Speicherblock zu. Das bedeutet, dass der neue Slice (Chunk) unabhängig vom Puffer über ein eigenes Backing-Array verfügt. Indem wir die Daten von buffer[:n] in den Chunk kopieren, stellen wir sicher, dass jeder Chunk seinen eigenen privaten Speicherplatz hat.

Speicher entkoppeln:
Durch die Entkopplung des Speichers jedes Blocks vom Puffer kann der Herausgeber weiterhin neue Daten in den Puffer einlesen, ohne dass dies Auswirkungen auf die Blöcke hat, die bereits an den Kanal gesendet wurden. Jeder Block verfügt nun über eine eigene unabhängige Kopie der Daten, sodass die Verbraucher die Blöcke ohne Einmischung des Herausgebers verarbeiten können.

Datenrennen verhindern:
Die Hauptursache des Datenwettlaufs war der gleichzeitige Zugriff auf den gemeinsam genutzten Puffer. Indem wir neue Slices erstellen und die Daten kopieren, eliminieren wir den gemeinsamen Speicher und jede Goroutine arbeitet mit ihren eigenen Daten. Dadurch wird die Möglichkeit einer Race-Bedingung ausgeschlossen, da kein Konflikt mehr um denselben Speicher besteht.

Abschluss

Der Kern des Fixes ist einfach, aber wirkungsvoll: Indem wir sicherstellen, dass jeder Datenblock über einen eigenen Speicher verfügt, eliminieren wir die gemeinsame Ressource (den Puffer), die den Datenwettlauf verursacht hat. Dies wird erreicht, indem die Daten aus dem Puffer in ein neues Slice kopiert werden, bevor sie an den Kanal gesendet werden. Bei diesem Ansatz arbeitet jeder Verbraucher unabhängig von den Aktionen des Herausgebers an seiner eigenen Kopie der Daten und gewährleistet so eine sichere gleichzeitige Verarbeitung ohne Race Conditions. Diese Methode zur Entkopplung des gemeinsam genutzten Speichers ist eine grundlegende Strategie bei der gleichzeitigen Programmierung. Es verhindert unvorhersehbares Verhalten, das durch Rennbedingungen verursacht wird, und stellt sicher, dass Ihre Go-Programme sicher, vorhersehbar und korrekt bleiben, selbst wenn mehrere Goroutinen gleichzeitig auf Daten zugreifen.

So einfach ist das!

Das obige ist der detaillierte Inhalt vonKnifflige Golang-Interviewfragen – Teil Datenrennen. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!