GoSudaGo HTTP Dosya Yükleme Yanıt Gecikmesinin Azaltılması: Kanal Tabanlı Asenkron İşçi Kuyruğunun Uygulanması
Genel Bakış
Go dilini ilk öğrenirken, arka uç sunucusunu uygulayarak öğrenmek sık görülen bir durumdur. Bu noktada RestAPI gibi ortamlarda dosya akışını alıp sunucuya yükleyen bir örnek oluşturmayı ele alalım. Go dilinin net/http sunucusu varsayılan olarak birden fazla isteği eşzamanlı işlediğinden, eşzamanlı ve yoğun yükleme işlemlerinin kendisinde bir sorun yoktur. Ancak akış alındıktan sonraki tüm işlemler eşzamanlı olarak gerçekleştirilirse, uç noktanın yanıtı gecikir. Bu durumu önlemek için kullanılabilecek tekniği birlikte inceleyelim.
Neden
Akışın alınması da genellikle uzun sürer ve büyük dosyalarda tek bir istek birkaç dakika boyunca işlenebilir. Bu nedenle akışın alınmasından sonraki işlemleri mümkün olduğunca hızlı gerçekleştirmek önemlidir. Bu örnek senaryo, akışı alıp geçici bir dosyaya kaydetmeyi ve ardından bir konteynere göndermeyi içermektedir. Bu aşamada geçici dosyanın konteynere gönderilmesi kısmını bir çalışan havuzu (worker pool) ile işlersek, yanıt gecikmesini azaltabiliriz.
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" // Ana makinedeki geçici dizin
14
15// UploadTask, eşzamanlı olmayan dosya gönderimi için verileri tutar.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler dosya yüklemelerini işler. Geçici dosyaya kaydeder, ardından Incus gönderimi için kuyruğa ekler.
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 // Ana makinede benzersiz geçici dosya yolu oluştur
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 // İstek gövdesini geçici dosyaya oluştur ve kopyala (eşzamanlı)
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 // Eşzamanlı Incus gönderimi için görevi kuyruğa ekle
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 // Hemen 202 Accepted yanıtı gönder
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Burada THIS PART olarak işaretlediğimiz kısma baktığınızda, görevin kuyruğa eklendiğini fark etmişsinizdir.
Şimdi iş kuyruğunun nasıl çalıştığına bakalım.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue bellek içi görev kuyruğunu başlatır.
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 kuyruğa bir UploadTask ekler.
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 kuyruktan bir UploadTask alır, kuyruk boşsa bloke eder.
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 mevcut kuyruk boyutunu döndürür.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
Örnek olarak sunulan iş kuyruğu basit biçimde uygulanmıştır. Bu iş kuyruğu, kuyruğa eklenmiş görevleri kanaldan çıkaran basit bir yapıya sahiptir.
Aşağıda, dosya yüklemesinden sonra konteynere gönderim yapmak için kullanılan worker metodu yer almaktadır. Mevcut metot, iyi bir yanıt verebilirlik ve kolay uygulama için sonsuz bir döngü kullanmaktadır; ancak kullanım amacına göre algoritmalar eklenebilir.
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 // Geçici dosyanın temizliğini ertele
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 // Geçici Incus hataları için yeniden denemelerle görevi işle
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" { // Kalıcı hataya örnek
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}
Öncelikle bu fonksiyon sürekli bir döngüde dönerek kuyruktan görevleri almaya çalışır. Ardından yeniden deneme kapsamı içinde, akıştan geçici dosyaya değil, geçici dosyadan konteynere yükleme aşamasını dener.
Avantajlar
Bu işleme yönteminin avantajı, yalnızca akış yüklemesi başarılı olduğu sürece sonraki işlenen görevlerin gecikme süresinin azaltılabilmesi ve eşzamanlı çok sayıda konteyner işleminden kaynaklanan kaynak tükenmesinin önlenebilmesidir. Mevcut kodda görüldüğü gibi, aynı anda ilerleyebilen konteyner işlemleri kanalın kapasitesiyle sınırlandırılmaktadır. Bu şekilde Go'nun paralel işleme özelliğini pratik olarak kullanabileceğimiz bir örneği incelemiş olduk. Daha fazla örnek görmek isterseniz, aşağıdaki bağlantıları ziyaret edebilirsiniz.Örnekleri içeren modül Örneği kullanan proje Proje kendisinde birçok yardımcı bileşen bulunduğundan, worker'ın kendisi hakkında öğrenmek için main.go dosyasında worker init fonksiyonunun nasıl çağrıldığına kısaca bakıp geçmeniz yeterlidir. Modül, başka türde worker'lar da içermektedir; lütfen bunlara da göz atın.