Τι είναι το workshopping;

Το work stealing είναι ένας μηχανισμός εξισορρόπησης φόρτου εργασίας σε συστήματα πολλαπλών νημάτων, όπου ένα αδρανές νήμα "κλέβει" εργασίες από την ουρά ενός άλλου απασχολημένου νήματος.

Χαρακτηριστικά:

• Αποκεντρωμένο: Κάθε νήμα διαχειρίζεται τη δική του τοπική ουρά εργασιών.
• Ενεργό: Το αδρανές νήμα ενεργά αναζητά εργασία από άλλα.
• Υψηλή απόδοση: Αποτελεσματικό για εργασίες με ανισοκατανομή φόρτου.

Στο Golang, χρησιμοποιείται στον scheduler για την κατανομή των goroutines στα συστήματα νημάτων (M). Όταν ένα M ολοκληρώσει την τρέχουσα εργασία του (πακέτο goroutines στο P), αναζητά goroutines σε άλλα P μέσω των τοπικών ουρών τους.

// Ψευδοκώδικας που δείχνει την έννοια του work stealing
package main

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

type Task struct {
	id int
}

// Προσομοίωση ουράς εργασιών
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
}

// Προσπάθεια "κλέψιμο" εργασίας από άλλη ουρά
func (q *TaskQueue) StealTask() (Task, bool) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if len(q.tasks) < 2 { // Δεν κλέβουμε αν υπάρχουν λίγες εργασίες
		return Task{}, false
	}
	// Κλέβουμε από το μέσο ή το τέλος, για να αποφύγουμε συγκρούσεις με τοπική εξαγωγή
	index := len(q.tasks) / 2
	task := q.tasks[index]
	q.tasks = append(q.tasks[:index], q.tasks[index+1:]...)
	return task, true
}

// Προσομοίωση worker νήματος
func worker(id int, localQueue *TaskQueue, otherQueues []*TaskQueue, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		// Προσπάθεια λήψης εργασίας από την τοπική ουρά
		task, ok := localQueue.GetLocalTask()
		if ok {
			fmt.Printf("Εργαζόμενος %d εκτελεί εργασία %d\n", id, task.id)
			time.Sleep(100 * time.Millisecond) // Προσομοίωση εργασίας
			continue
		}

		// Αν η τοπική ουρά είναι άδεια, προσπαθούμε να κλέψουμε
		stolen := false
		for _, queue := range otherQueues {
			if queue == localQueue {
				continue // Δεν κλέβουμε από τον εαυτό μας
			}
			task, ok := queue.StealTask()
			if ok {
				fmt.Printf("Εργαζόμενος %d, κλέβει εργασία %d\n", id, task.id)
				time.Sleep(100 * time.Millisecond) // Προσομοίωση εργασίας
				stolen = true
				break // Επιτυχής κλοπή, έξοδος από την λούπα
			}
		}

		if !stolen {
			// Αν δεν κλέψαμε, πιθανώς να μην υπάρχουν άλλες εργασίες
			// Σε πραγματικό scheduler, υπάρχουν πιο πολύπλοκοι μηχανισμοί τερματισμού
			// Για επίδειξη, απλά βγαίνουμε μετά από κάποιο χρόνο αναμονής
			fmt.Printf("Εργαζόμενος %d, σε αδράνεια, περιμένει...\n", id)
			time.Sleep(50 * time.Millisecond)
			// Σε πραγματικό σενάριο, εδώ θα ήταν μηχανισμός τερματισμού ή παρκάρισμα νήματος
			// return // Για επίδειξη, θα επιτρέψουμε την έξοδο
			break // Απλοποίηση της προσομοίωσης
		}
	}
}

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

	// Κατανομή εργασιών ανομοιόμορφα για την επίδειξη work stealing
	for i := 0; i < numTasks; i++ {
		queueIndex := i % 2 // Περισσότερες εργασίες στους πρώτους δύο εργαζόμενους
		taskQueues[queueIndex].AddTask(Task{id: i})
	}

	var wg sync.WaitGroup
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		// Μεταβιβάζουμε αναφορές σε όλες τις ουρές, ώστε ο worker να μπορεί να "κλέψει"
		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("Όλες οι εργασίες ολοκληρώθηκαν")
}