Was ist Workshopping?

Das Work Stealing (Arbeitsdiebstahl) ist ein Load-Balancing-Mechanismus in Multithread-Systemen, bei dem ein inaktiver Thread (Worker) Aufgaben aus der Warteschlange eines anderen, beschäftigten Threads "stiehlt".

Wichtige Merkmale:

• Dezentralisiert: Jeder Thread verwaltet seine eigene Aufgabenwarteschlange.
• Aktiv: Der inaktive Thread sucht aktiv nach Arbeit bei anderen.
• Hohe Leistung: Effektiv bei ungleichmäßiger Last.

In Golang wird es im Scheduler verwendet, um Goroutines auf die Betriebssystem-Threads (M) zu verteilen. Wenn ein M seine aktuelle Arbeit (die Goroutine-Pakete auf P) beendet, sucht es Goroutines bei anderen P über deren lokale Warteschlangen.

// Pseudocode, der das Konzept des Work Stealing demonstriert
package main

import "fmt"
import "sync"
import "time"

type Task struct {
	id int
}

// Simulation einer Aufgabenwarteschlange
type TaskQueue struct {
	tasks []Task
	mu    sync.Mutex
}

func (q *TaskQueue) AddTask(task Task) {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.tasks = append(q.tasks, task)
}

func (q *TaskQueue) GetLocalTask() (Task, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if len(q.tasks) == 0 {
		return Task{}, false
	}
	task := q.tasks[0]
	q.tasks = q.tasks[1:]
	return task, true
}

// Versuch, eine Aufgabe aus einer anderen Warteschlange "zu stehlen"
func (q *TaskQueue) StealTask() (Task, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if len(q.tasks) < 2 { // Nicht stehlen, wenn wenige Aufgaben
		return Task{}, false
	}
	// Stehlen aus der Mitte oder dem Ende, um Konflikte mit lokalem Zugriff zu vermeiden
	index := len(q.tasks) / 2
	task := q.tasks[index]
	q.tasks = append(q.tasks[:index], q.tasks[index+1:]...)
	return task, true
}

// Simulation eines Workers
func worker(id int, localQueue *TaskQueue, otherQueues []*TaskQueue, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		// Versuch, eine Aufgabe aus der lokalen Warteschlange zu holen
		task, ok := localQueue.GetLocalTask()
		if ok {
			fmt.Printf("Worker %d führt Aufgabe %d lokal aus\n", id, task.id)
			time.Sleep(100 * time.Millisecond) // Arbeitssimulation
			continue
		}

		// Wenn die lokale Warteschlange leer ist, versuchen wir zu stehlen
		stolen := false
		for _, queue := range otherQueues {
			if queue == localQueue {
				continue // Nicht bei sich selbst stehlen
			}
			task, ok := queue.StealTask()
			if ok {
				fmt.Printf("Worker %d hat Aufgabe %d gestohlen\n", id, task.id)
				time.Sleep(100 * time.Millisecond) // Arbeitssimulation
				stolen = true
				break // Nach erfolgreichem Steal beenden
			}
		}

		if !stolen {
			// Wenn kein Steal möglich, vielleicht keine Aufgaben mehr
			// In einem echten Scheduler gäbe es komplexere Mechanismen zum Beenden
			// Für die Demo einfach nach einer Weile beenden
			fmt.Printf("Worker %d ist inaktiv, wartet...\n", id)
			time.Sleep(50 * time.Millisecond)
			// In einem echten Szenario würde hier ein Beendigungs- oder Parking-Mechanismus stehen
			// return // Für die Demo erlauben wir das Beenden
			break // Zur Vereinfachung der Simulation
		}
	}
}

func main() {
	numWorkers := 4
	numTasks := 20
	taskQueues := make([]*TaskQueue, numWorkers)
	for i := range taskQueues {
		taskQueues[i] = &TaskQueue{}
	}

	// Aufgaben ungleichmäßig verteilen, um das Work Stealing zu demonstrieren
	for i := 0; i < numTasks; i++ {
		queueIndex := i % 2 // Mehr Aufgaben bei den ersten beiden
		taskQueues[queueIndex].AddTask(Task{id: i})
	}

	var wg sync.WaitGroup
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		// Referenzen an alle Warteschlangen übergeben, damit der Worker "stehlen" kann
		otherQueues := make([]*TaskQueue, 0, numWorkers-1)
		for j := range taskQueues {
			if i != j {
				otherQueues = append(otherQueues, taskQueues[j])
			}
		}
		go worker(i, taskQueues[i], otherQueues, &wg)
	}

	wg.Wait()
	fmt.Println("Alle Aufgaben sind erledigt")
}

```}}}}}]}},{