GoSudaRiduzione del ritardo di risposta nell'upload di file HTTP in Go: implementazione di una coda di worker asincroni basata su canali
Panoramica
Quando si impara Go, capita spesso di farlo implementando un server backend. In questo caso, facciamo come esempio la creazione di un programma che riceve uno stream di file tramite una RestAPI o simili e lo carica sul server. Il server net/http del linguaggio Go gestisce per impostazione predefinita più richieste contemporaneamente, quindi di per sé il caricamento simultaneo e concorrente non presenta problemi. Tuttavia, se tutte le operazioni successive alla ricezione dello stream vengono eseguite in modo sincrono, la risposta dell'endpoint viene ritardata. Vediamo una tecnica per prevenire questa situazione.
Causa
Anche la ricezione di uno stream richiede generalmente molto tempo e, nel caso di file di grandi dimensioni, una singola richiesta può richiedere diversi minuti per essere elaborata. In questi casi è importante elaborare il più rapidamente possibile le operazioni successive alla ricezione. Lo scenario di questo esempio è quello di ricevere uno stream, salvarlo come file temporaneo e inviarlo a un container. A questo punto, se la parte che invia il file temporaneo al container viene gestita tramite un worker pool, è possibile ridurre il ritardo della risposta.
1package file_upload
2
3import (
4 "fmt"
5 "io"
6 "log"
7 "net/http"
8 "os"
9 "path/filepath"
10 "time"
11)
12
13const uploadTempDir = "/tmp/incus_uploads" // Directory temporanea sull'host
14
15// UploadTask contiene i dati per l'invio asincrono del file.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler elabora i caricamenti dei file. Salva in un file temporaneo, poi accoda per l'invio a Incus.
24func UploadHandler(wr http.ResponseWriter, req *http.Request) {
25 if req.Method != http.MethodPost {
26 http.Error(wr, "POST method required.", http.StatusMethodNotAllowed)
27 return
28 }
29 originalFilePath := req.Header.Get("X-File-Path")
30 originalFilename := filepath.Base(req.Header.Get("X-Host-Path"))
31 containerName := req.Header.Get("X-Container-Name")
32 if originalFilePath == "" || containerName == "" {
33 http.Error(wr, "Missing X-File-Path or X-Container-Name header.", http.StatusBadRequest)
34 return
35 }
36
37 cleanContainerDestPath := filepath.Clean(originalFilePath)
38 if !filepath.IsAbs(cleanContainerDestPath) {
39 http.Error(wr, "File path must be absolute.", http.StatusBadRequest)
40 return
41 }
42
43 // Crea un percorso univoco per il file temporaneo sull'host
44 tempFileName := fmt.Sprintf("%d-%s", time.Now().UnixNano(), filepath.Base(originalFilePath))
45 hostTempFilePath := filepath.Join(uploadTempDir, tempFileName)
46
47 if err := os.MkdirAll(uploadTempDir, 0755); err != nil {
48 log.Printf("ERROR: Failed to create temp upload directory: %v", err)
49 http.Error(wr, "Server error.", http.StatusInternalServerError)
50 return
51 }
52
53 // Crea il file temporaneo e copia il corpo della richiesta in esso (sincrono)
54 outFile, err := os.Create(hostTempFilePath)
55 if err != nil {
56 log.Printf("ERROR: Failed to create temporary file: %v", err)
57 http.Error(wr, "Server error.", http.StatusInternalServerError)
58 return
59 }
60 defer outFile.Close()
61
62 bytesWritten, err := io.Copy(outFile, req.Body)
63 if err != nil {
64 outFile.Close()
65 os.Remove(hostTempFilePath)
66 log.Printf("ERROR: Failed to copy request body to temp file: %v", err)
67 http.Error(wr, "File transfer failed.", http.StatusInternalServerError)
68 return
69 }
70 log.Printf("Upload Info: Received %d bytes, saved to %s.", bytesWritten, hostTempFilePath)
71
72 // Accoda l'attività per l'invio asincrono a Incus
73 task := UploadTask{
74 HostTempFilePath: hostTempFilePath,
75 ContainerName: containerName,
76 HostFilename : originalFilename,
77 ContainerDestinationPath: cleanContainerDestPath,
78 }
79 EnqueueTask(task) //THIS PART
80 log.Printf("Upload Info: Task enqueued for %s to %s.", originalFilePath, containerName)
81
82 // Invia immediatamente la risposta 202 Accepted
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Guardando la parte indicata con THIS PART, si sarà notato che qui l'attività viene inserita in una coda.
Vediamo ora come funziona la coda delle attività.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue inizializza la coda delle attività in memoria.
12func InitWorkQueue() {
13 once.Do(func() {
14 taskQueue = make(chan UploadTask, 100)
15 log.Println("Upload Info: Work queue initialized.")
16 })
17}
18
19// EnqueueTask aggiunge un UploadTask alla coda.
20func EnqueueTask(task UploadTask) {
21 if taskQueue == nil {
22 log.Fatal("ERROR: Task queue not initialized.")
23 }
24 taskQueue <- task
25 log.Printf("Upload Info: Queue: Task enqueued. Size: %d", len(taskQueue))
26}
27
28// DequeueTask recupera un UploadTask dalla coda, bloccandosi se è vuota.
29func DequeueTask() UploadTask {
30 if taskQueue == nil {
31 log.Fatal("ERROR: Task queue not initialized.")
32 }
33 task := <-taskQueue
34 log.Printf("Upload Info: Queue: Task dequeued. Size: %d", len(taskQueue))
35 return task
36}
37
38// GetQueueLength restituisce la dimensione attuale della coda.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
La coda delle attività fornita come esempio è implementata in modo semplice. Questa coda ha una struttura elementare che estrae dal canale le attività accodate.
Di seguito è riportato il metodo worker per inviare il file al container dopo il caricamento. Il metodo attuale utilizza un ciclo infinito per garantire una buona reattività e un'implementazione semplice, ma è possibile aggiungere algoritmi in base all'uso previsto.
1func StartFilePushWorker() {
2 for {
3 task := DequeueTask()
4 log.Printf("Upload Info: Worker processing task for %s from %s.", task.ContainerName, task.HostFilename)
5
6 // Pianifica la pulizia del file temporaneo
7 defer func(filePath string) {
8 if err := os.Remove(filePath); err != nil {
9 log.Printf("ERROR: Worker: Failed to remove temp file '%s': %v", filePath, err)
10 } else {
11 log.Printf("Upload Info: Worker: Cleaned up temp file: %s", filePath)
12 }
13 }(task.HostTempFilePath)
14
15 // Elabora l'attività con nuovi tentativi per gli errori transitori di Incus
16 for i := 0; i <= MaxRetries; i++ {
17 err := processUploadTask(task) //separate upload task
18 if err == nil {
19 log.Printf("SUCCESS: Worker: Task completed for %s.", task.ContainerName)
20 break
21 }
22
23 isTransient := true
24 if err.Error() == "incus: container not found" { // Esempio di errore permanente
25 isTransient = false
26 }
27
28 if isTransient && i < MaxRetries {
29 log.Printf("WARNING: Worker: Task failed for %s (attempt %d/%d): %v. Retrying.",
30 task.ContainerName, i+1, MaxRetries, err)
31 time.Sleep(RetryDelay)
32 } else {
33 log.Printf("ERROR: Worker: Task permanently failed for %s after %d attempts: %v.",
34 task.ContainerName, i+1, err)
35 break
36 }
37 }
38 }
39}
Innanzitutto, questa funzione continua a girare in un ciclo cercando di prelevare le attività dalla coda. Successivamente, entro l'intervallo dei tentativi, tenta la fase di caricamento dal file temporaneo al container, non dallo stream al file temporaneo.
Vantaggi
Il vantaggio di questo approccio è che, purché il caricamento dello stream vada a buon fine, è possibile ridurre il ritardo delle operazioni successive e prevenire l'esaurimento delle risorse causato da operazioni concorrenti sui container. Come si vede nel codice attuale, il numero di operazioni sui container che possono essere eseguite contemporaneamente è limitato dal numero dei canali. In questo modo abbiamo esaminato un esempio pratico di come utilizzare il parallelismo di Go. Per vedere più esempi, visitate i link seguenti.Modulo contenente esempi Progetto che utilizza gli esempi Il progetto stesso contiene molti componenti accessori, quindi per studiare il worker in sé è sufficiente vedere brevemente come viene richiamata la funzione init del worker da main.go. Il modulo include anche altri tipi di worker, quindi vi invitiamo a consultarli.