--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Go HTTP-tiedostolatauksen vasteajan lyhentäminen: kanavapohjaisen asynkronisen työntekijäjonon toteutus' description: Tämä rakenne erottaa net/http-tiedostolatauskäsittelijässä väliaikaisten tiedostojen vastaanoton ja Incus-konttiin tapahtuvan siirron omiksi tehtävikseen puskurikanavan ja goroutine-työntekijäpoolin avulla, mikä vähentää päätepisteen vastausviivettä. language: fi date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # 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ää. ```go package file_upload import ( "fmt" "io" "log" "net/http" "os" "path/filepath" "time" ) const uploadTempDir = "/tmp/incus_uploads" // Isäntäkoneen väliaikaishakemisto // UploadTask sisältää tiedot asynkronista tiedostonsiirtoa varten. type UploadTask struct { HostTempFilePath string ContainerName string HostFilename string ContainerDestinationPath string } // UploadHandler käsittelee tiedostolatauksia. Tallentaa väliaikaistiedostoon ja asettaa jonoon Incus-siirtoa varten. func UploadHandler(wr http.ResponseWriter, req *http.Request) { if req.Method != http.MethodPost { http.Error(wr, "POST method required.", http.StatusMethodNotAllowed) return } originalFilePath := req.Header.Get("X-File-Path") originalFilename := filepath.Base(req.Header.Get("X-Host-Path")) containerName := req.Header.Get("X-Container-Name") if originalFilePath == "" || containerName == "" { http.Error(wr, "Missing X-File-Path or X-Container-Name header.", http.StatusBadRequest) return } cleanContainerDestPath := filepath.Clean(originalFilePath) if !filepath.IsAbs(cleanContainerDestPath) { http.Error(wr, "File path must be absolute.", http.StatusBadRequest) return } // Luo isäntäkoneelle yksilöllinen väliaikaistiedoston polku tempFileName := fmt.Sprintf("%d-%s", time.Now().UnixNano(), filepath.Base(originalFilePath)) hostTempFilePath := filepath.Join(uploadTempDir, tempFileName) if err := os.MkdirAll(uploadTempDir, 0755); err != nil { log.Printf("ERROR: Failed to create temp upload directory: %v", err) http.Error(wr, "Server error.", http.StatusInternalServerError) return } // Luo väliaikaistiedosto ja kopioi pyynnön runko siihen (synkronisesti) outFile, err := os.Create(hostTempFilePath) if err != nil { log.Printf("ERROR: Failed to create temporary file: %v", err) http.Error(wr, "Server error.", http.StatusInternalServerError) return } defer outFile.Close() bytesWritten, err := io.Copy(outFile, req.Body) if err != nil { outFile.Close() os.Remove(hostTempFilePath) log.Printf("ERROR: Failed to copy request body to temp file: %v", err) http.Error(wr, "File transfer failed.", http.StatusInternalServerError) return } log.Printf("Upload Info: Received %d bytes, saved to %s.", bytesWritten, hostTempFilePath) // Lisää tehtävä jonoon asynkronista Incus-siirtoa varten task := UploadTask{ HostTempFilePath: hostTempFilePath, ContainerName: containerName, HostFilename : originalFilename, ContainerDestinationPath: cleanContainerDestPath, } EnqueueTask(task) //THIS PART log.Printf("Upload Info: Task enqueued for %s to %s.", originalFilePath, containerName) // Lähetä välittömästi 202 Accepted -vastaus wr.WriteHeader(http.StatusAccepted) fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName) } ``` Huomasit varmasti, että kohdassa, jonka merkitsin THIS PART, tehtävä lisätään jonoon. Tarkastellaan seuraavaksi, miten tehtäväjono toimii. ```go package file_upload import ( "log" "sync" ) var taskQueue chan UploadTask var once sync.Once // InitWorkQueue alustaa muistissa olevan tehtäväjonon. func InitWorkQueue() { once.Do(func() { taskQueue = make(chan UploadTask, 100) log.Println("Upload Info: Work queue initialized.") }) } // EnqueueTask lisää UploadTaskin jonoon. func EnqueueTask(task UploadTask) { if taskQueue == nil { log.Fatal("ERROR: Task queue not initialized.") } taskQueue <- task log.Printf("Upload Info: Queue: Task enqueued. Size: %d", len(taskQueue)) } // DequeueTask hakee UploadTaskin jonosta ja estää suorituksen, jos jono on tyhjä. func DequeueTask() UploadTask { if taskQueue == nil { log.Fatal("ERROR: Task queue not initialized.") } task := <-taskQueue log.Printf("Upload Info: Queue: Task dequeued. Size: %d", len(taskQueue)) return task } // GetQueueLength palauttaa jonon nykyisen koonnin. func GetQueueLength() int { if taskQueue == nil { return 0 } return len(taskQueue) } ``` 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. ```go func StartFilePushWorker() { for { task := DequeueTask() log.Printf("Upload Info: Worker processing task for %s from %s.", task.ContainerName, task.HostFilename) // Siivoa väliaikaistiedosto funktion lopussa (defer) defer func(filePath string) { if err := os.Remove(filePath); err != nil { log.Printf("ERROR: Worker: Failed to remove temp file '%s': %v", filePath, err) } else { log.Printf("Upload Info: Worker: Cleaned up temp file: %s", filePath) } }(task.HostTempFilePath) // Suorita tehtävä uudelleen ohimenevien Incus-virheiden varalta for i := 0; i <= MaxRetries; i++ { err := processUploadTask(task) // erillinen lataustehtävä if err == nil { log.Printf("SUCCESS: Worker: Task completed for %s.", task.ContainerName) break } isTransient := true if err.Error() == "incus: container not found" { // Esimerkki pysyvästä virheestä isTransient = false } if isTransient && i < MaxRetries { log.Printf("WARNING: Worker: Task failed for %s (attempt %d/%d): %v. Retrying.", task.ContainerName, i+1, MaxRetries, err) time.Sleep(RetryDelay) } else { log.Printf("ERROR: Worker: Task permanently failed for %s after %d attempts: %v.", task.ContainerName, i+1, err) break } } } } ``` 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ä](https://github.com/gg582/linux_virt_unit) [Esimerkkejä hyödyntävä projekti](https://github.com/gg582/incuspeed) 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.