GoSudaGo HTTP-tiedostolatauksen vasteajan lyhentäminen: kanavapohjaisen asynkronisen työntekijäjonon toteutus
Yleiskatsaus
Kun opit Go-kieltä ensimmäistä kertaa, opit usein samalla toteuttamaan backend-palvelinta. Tarkastellaan esimerkkinä tapausta, jossa RestAPI:n kautta vastaanotetaan tiedostovirta ja se ladataan palvelimelle. Go-kielen net/http-palvelin käsittelee oletuksena useita pyyntöjä samanaikaisesti, joten samanaikaiset latausten rinnakkaiset lataukset sinänsä eivät aiheuta ongelmia. Jos kuitenkin kaikki virran vastaanoton jälkeiset toiminnot suoritetaan synkronisesti, päätepisteen vastaus viivästyy. Tutustutaan tekniikkaan, jolla tällainen tilanne voidaan välttää.
Syy
Virran vastaanottamiseen kuluu yleensä myös pitkä aika, ja suurilla tiedostoilla yksittäinen pyyntö voi kestää useita minuutteja. Tällaisissa tapauksissa on tärkeää käsitellä vastaanoton jälkeiset toiminnot mahdollisimman nopeasti. Tämän esimerkkiskenaarion mukaan virta vastaanotetaan ja tallennetaan väliaikaistiedostoksi, minkä jälkeen tiedosto siirretään konttiin. Jos väliaikaistiedoston siirto konttiin hoidetaan työntekijäpoolilla (worker pool), vastauksen viivettä voidaan lyhentää.
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" // Isäntäkoneen väliaikaishakemisto
14
15// UploadTask sisältää tiedot asynkronista tiedostonsiirtoa varten.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler käsittelee tiedostolatauksia. Tallentaa väliaikaistiedostoon ja asettaa jonoon Incus-siirtoa varten.
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 // Luo isäntäkoneelle yksilöllinen väliaikaistiedoston polku
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 // Luo väliaikaistiedosto ja kopioi pyynnön runko siihen (synkronisesti)
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 // Lisää tehtävä jonoon asynkronista Incus-siirtoa varten
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 // Lähetä välittömästi 202 Accepted -vastaus
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Huomasit varmasti, että kohdassa, jonka merkitsin THIS PART, tehtävä lisätään jonoon.
Tarkastellaan seuraavaksi, miten tehtäväjono toimii.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue alustaa muistissa olevan tehtäväjonon.
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 lisää UploadTaskin jonoon.
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 hakee UploadTaskin jonosta ja estää suorituksen, jos jono on tyhjä.
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 palauttaa jonon nykyisen koonnin.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
Esimerkkinä esitetty tehtäväjono on toteutettu yksinkertaisesti. Tämä tehtäväjono on rakenteeltaan yksinkertainen: se ottaa jonossa olevat tehtävät ulos kanavasta.
Alla on työntekijämetodi, joka siirtää tiedoston konttiin latauksen jälkeen. Nykyinen metodi käyttää äärettömää silmukkaa hyvän reagoivuuden ja yksinkertaisen toteutuksen vuoksi, mutta käyttötarkoituksesta riippuen algoritmia voi lisätä vapaasti.
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 // Siivoa väliaikaistiedosto funktion lopussa (defer)
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 // Suorita tehtävä uudelleen ohimenevien Incus-virheiden varalta
16 for i := 0; i <= MaxRetries; i++ {
17 err := processUploadTask(task) // erillinen lataustehtävä
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" { // Esimerkki pysyvästä virheestä
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}
Ensin tämä funktio kiertää silmukassa ja yrittää hakea tehtäviä jonosta. Sen jälkeen se yrittää uudelleenyritysten rajoissa vaihetta, jossa ei siirretä virtaa väliaikaistiedostoon, vaan siirretään väliaikaistiedosto konttiin.
Hyödyt
Tämän käsittelytavan etuna on, että jos tiedoston lataus virrasta onnistuu, jälkeenpäin käsiteltävien tehtävien viivettä voidaan vähentää, ja samanaikaisista konttitoimista johtuvaa resurssien ehtymistä voidaan ehkäistä. Kuten nykyisestä koodista näkyy, samanaikaisesti etenevien konttitoimien määrää rajoittaa kanavan koko. Näin olemme tutustuneet esimerkkiin, jossa Go:n rinnakkaisuutta voidaan hyödyntää käytännössä. Jos haluat nähdä lisää esimerkkejä, käy alla olevissa linkeissä.Moduuli, jossa on esimerkkejä Esimerkkejä hyödyntävä projekti Projektissa on itsessään paljon apukomponentteja, joten workerin itsensä opiskeluun riittää, että tarkastelet lyhyesti, miten worker-init-funktiota kutsutaan main.go-tiedostossa, ja jatkat eteenpäin. Moduulissa on myös muunlaisia workereita, joihin kannattaa tutustua.