--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Reducing Response Latency in Go HTTP File Uploads: Implementing an Asynchronous Worker Queue Based on Channels' description: This structure separates temporary file reception and the Incus container transfer operations in the net/http file upload handler into a buffered channel and a goroutine worker pool, thereby reducing endpoint response latency. language: en date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # Overview When first learning Go, developers often build backend servers as a learning exercise. Let us consider an example of creating an implementation that receives a file stream from a RestAPI or similar interface and uploads it to a server. Because the Go net/http server natively handles multiple requests concurrently, concurrent uploads themselves do not present a problem. However, if all operations following stream reception are processed synchronously, the response of the endpoint will be delayed. Let us examine techniques to prevent such situations. # Cause Receiving a stream generally requires a significant amount of time, and for large files, a single request may take several minutes to process. In such cases, it is important to process operations subsequent to reception as swiftly as possible. This example scenario involves receiving a stream, saving it as a temporary file, and pushing it to a container. If the step of pushing the temporary file to the container is handled by a worker pool, response latency can be reduced. ```go package file_upload import ( "fmt" "io" "log" "net/http" "os" "path/filepath" "time" ) const uploadTempDir = "/tmp/incus_uploads" // Host temporary directory // UploadTask holds data for asynchronous file push. type UploadTask struct { HostTempFilePath string ContainerName string HostFilename string ContainerDestinationPath string } // UploadHandler processes file uploads. Saves to temp file, then queues for Incus push. 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 } // Create unique temporary file path on host 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 } // Create and copy request body to temporary file (synchronous) 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) // Enqueue task for asynchronous Incus push 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) // Send immediate 202 Accepted response wr.WriteHeader(http.StatusAccepted) fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName) } ``` Examining the section marked as THIS PART, you will have noticed that tasks are inserted into a queue. Now, let us observe how the work queue operates. ```go package file_upload import ( "log" "sync" ) var taskQueue chan UploadTask var once sync.Once // InitWorkQueue initializes the in-memory task queue. func InitWorkQueue() { once.Do(func() { taskQueue = make(chan UploadTask, 100) log.Println("Upload Info: Work queue initialized.") }) } // EnqueueTask adds an UploadTask to the queue. 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 retrieves an UploadTask from the queue, blocking if empty. 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 returns current queue size. func GetQueueLength() int { if taskQueue == nil { return 0 } return len(taskQueue) } ``` The work queue provided in the example is implemented simply. It possesses a straightforward structure that extracts tasks residing in the queue from the channel. Below is the worker method for pushing to the container after file uploading. To achieve good responsiveness and straightforward implementation, the current method utilizes an infinite loop, though you may add algorithms depending on your specific use case. ```go func StartFilePushWorker() { for { task := DequeueTask() log.Printf("Upload Info: Worker processing task for %s from %s.", task.ContainerName, task.HostFilename) // Defer cleanup of the temporary file 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) // Process task with retries for transient Incus errors for i := 0; i <= MaxRetries; i++ { err := processUploadTask(task) //separate upload task if err == nil { log.Printf("SUCCESS: Worker: Task completed for %s.", task.ContainerName) break } isTransient := true if err.Error() == "incus: container not found" { // Example permanent error 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 } } } } ``` First, this function continuously loops in an attempt to retrieve tasks from the queue. Subsequently, within the retry limits, it attempts the upload phase from the temporary file to the container rather than from the stream to the temporary file. # Benefits The advantages of this processing approach are that, **provided the stream upload is successful**, latency for subsequent operations can be reduced, and resource exhaustion caused by simultaneous container operations can be prevented. As demonstrated in the current code, the number of concurrently executable container operations is limited by the channel capacity. We have thus explored an example where Go's parallel processing can be applied practically. If you wish to view additional examples, please visit the links below. [Module containing examples](https://github.com/gg582/linux_virt_unit) [Project utilizing examples](https://github.com/gg582/incuspeed) Since the project itself contains many auxiliary components, your study of the worker itself only requires a brief look at how the worker initialization function is called in main.go, after which you may move on. Please note that the module also includes other forms of workers.