GoSudaSkrátenie oneskorenia odpovede pri nahrávaní súborov cez Go HTTP: implementácia asynchrónnej worker fronty založenej na kanáloch
Prehľad
Pri prvom učení jazyka Go sa často stáva, že sa jazyk učíme implementáciou backendového servera. Uveďme si príklad, v ktorom sa v RestAPI alebo podobnom rozhraní prijíma prúd súboru (file stream) a nahráva sa na server. Server net/http v jazyku Go v predvolenom nastavení spracúva viacero požiadaviek súbežne, takže samotné súbežné nahrávanie viacerých súborov nepredstavuje problém. Ak však všetky operácie po prijatí prúdu spracúvame synchrónne, odpoveď endpointu sa oneskorí. Pozrime sa na techniku, ktorá tomuto stavu zabraňuje.
Príčina
Prijatie prúdu zvyčajne trvá dlho a pri veľkých súboroch môže spracovanie jednej požiadavky trvať niekoľko minút. V takomto prípade je dôležité spracovať operácie nasledujúce po prijatí čo najrýchlejšie. Scenár tohto príkladu je: prijať prúd, uložiť ho ako dočasný súbor a odoslať ho do kontajnera. Ak časť odosielania dočasného súboru do kontajnera spracujeme pomocou worker pool, môžeme skrátiť oneskorenie odpovede.
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" // Dočasný adresár hostiteľa
14
15// UploadTask uchováva dáta pre asynchrónne odoslanie súboru.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler spracúva nahrávanie súborov. Uloží ich do dočasného súboru a potom ich zaradí do fronty na odoslanie do 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 // Vytvorí jedinečnú cestu k dočasnému súboru na hostiteľovi
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 // Vytvorí dočasný súbor a skopíruje doň telo požiadavky (synchrónne)
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 // Zaradí úlohu do fronty na asynchrónne odoslanie do 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 // Okamžite odošle odpoveď 202 Accepted
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Ak sa pozriete na časť označenú THIS PART, zrejme ste si všimli, že sa tu úloha vkladá do fronty.
Pozrime sa teraz na to, ako pracuje fronta úloh.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue inicializuje frontu úloh v pamäti.
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 pridá UploadTask do fronty.
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 získa UploadTask z fronty; ak je fronta prázdna, blokuje.
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 vráti aktuálnu veľkosť fronty.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
Fronta úloh uvedená ako príklad je implementovaná jednoducho. Má jednoduchú štruktúru, ktorá z channelu vyberá úlohy čakajúce vo fronte.
Nižšie je worker metóda, ktorá po nahratí súboru odosiela súbor do kontajnera. Aktuálna metóda používa nekonečnú slučku, čo zabezpečuje dobrú odozvu a jednoduchú implementáciu; podľa účelu však možno pridať aj ďalšie algoritmy.
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 // Odložené odstránenie dočasného súboru
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 // Spracuje úlohu s opakovaniami pri dočasných chybách Incus
16 for i := 0; i <= MaxRetries; i++ {
17 err := processUploadTask(task) //samostatná úloha nahrávania
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" { // Príklad trvalej chyby
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}
Najprv táto funkcia v nekonečnej slučke neustále sa pokúša prijímať úlohy z fronty. Potom v rámci rozsahu opakovaní vykoná krok nahrávania, ktorý nie je prenosom prúd-dočasný súbor, ale prenosom dočasný súbor-kontajner.
Výhody
Výhodou tohto spôsobu spracovania je, že ak je nahrávanie prúdu úspešné, môžeme skrátiť oneskorenie operácií, ktoré nasledujú, a zabrániť vyčerpaniu zdrojov spôsobenému súbežnými operáciami s kontajnermi. Ako je vidieť v aktuálnom kóde, počet kontajnerových operácií, ktoré môžu prebiehať súčasne, je obmedzený počtom channelov. Takto sme si prakticky ukázali príklad, ako používať paralelné spracovanie v jazyku Go. Ak chcete vidieť viac príkladov, navštívte prosím odkazy nižšie.modul obsahujúci príklady projekt využívajúci príklady Projekt sám obsahuje veľa pomocných komponentov, preto pri štúdiu samotného workera stačí iba stručne pozrieť, ako sa init funkcia workera volá v main.go. Modul obsahuje aj iné typy workerov, pozrite si ich prosím.