Optimize streaming from Real-Debrid

config.example.yml

@@ -53,9 +53,6 @@ on_library_update: |
       echo "detected update on: $arg"
   done
 
-# buffer size when zurg is streaming files
-network_buffer_size: 1048576 # 1 MiB
-
 # true = send link to rclone and rclone will stream the file
 # false = zurg will stream the file
 serve_from_rclone: false

internal/config/types.go

@@ -17,7 +17,6 @@ type ConfigInterface interface {
 	GetDownloadTimeoutSecs() int
 	GetDumpTorrentsEveryMins() int
 	GetHost() string
-	GetNetworkBufferSize() int
 	GetNumberOfHosts() int
 	GetNumberOfWorkers() int
 	GetOnLibraryUpdate() string
@@ -56,7 +55,6 @@ type ZurgConfig struct {
 	Host                      string   `yaml:"host" json:"host"`
 	IgnoreRenames             bool     `yaml:"ignore_renames" json:"ignore_renames"`
 	LogRequests               bool     `yaml:"log_requests" json:"log_requests"`
-	NetworkBufferSize         int      `yaml:"network_buffer_size" json:"network_buffer_size"`
 	NumberOfHosts             int      `yaml:"number_of_hosts" json:"number_of_hosts"`
 	NumOfWorkers              int      `yaml:"concurrent_workers" json:"concurrent_workers"`
 	OnLibraryUpdate           string   `yaml:"on_library_update" json:"on_library_update"`
@@ -150,13 +148,6 @@ func (z *ZurgConfig) GetOnLibraryUpdate() string {
 	return z.OnLibraryUpdate
 }
 
-func (z *ZurgConfig) GetNetworkBufferSize() int {
-	if z.NetworkBufferSize == 0 {
-		return 32 * 1024 // 32kb
-	}
-	return z.NetworkBufferSize
-}
-
 func (z *ZurgConfig) EnableRetainFolderNameExtension() bool {
 	return z.RetainFolderNameExtension
 }

internal/handlers/home.go

@@ -315,7 +315,7 @@ func (zr *Handlers) handleHome(resp http.ResponseWriter, req *http.Request) {
 
 	out += fmt.Sprintf(`
         <tr>
-            <td rowspan="23">Config</td>
+            <td rowspan="22">Config</td>
             <td>Version</td>
             <td>%s</td>
         </tr>
@@ -391,10 +391,6 @@ func (zr *Handlers) handleHome(resp http.ResponseWriter, req *http.Request) {
             <td>Additional Playable Extensions</td>
             <td>%s</td>
         </tr>
-        <tr>
-            <td>Network Buffer Size</td>
-            <td>%d bytes</td>
-        </tr>
         <tr>
             <td>Serve From Rclone</td>
             <td>%t</td>
@@ -428,7 +424,6 @@ func (zr *Handlers) handleHome(resp http.ResponseWriter, req *http.Request) {
 		response.Config.ShouldAutoAnalyzeNewTorrents(),
 		response.Config.ShouldCacheNetworkTestResults(),
 		response.Config.GetPlayableExtensions(),
-		response.Config.GetNetworkBufferSize(),
 		response.Config.ShouldServeFromRclone(),
 		response.Config.ShouldForceIPv6(),
 		response.Config.GetOnLibraryUpdate(),

internal/universal/downloader.go

@@ -68,28 +68,28 @@ func (dl *Downloader) DownloadFile(
 ) {
 	torrents, ok := torMgr.DirectoryMap.Get(directory)
 	if !ok {
-		log.Warnf("Cannot find directory %s", directory)
+		log.Errorf("Cannot find directory %s", directory)
 		http.Error(resp, "File not found", http.StatusNotFound)
 		return
 	}
 
 	torrent, ok := torrents.Get(torrentName)
 	if !ok {
-		log.Warnf("Cannot find torrent %sfrom path %s", torrentName, req.URL.Path)
+		log.Errorf("Cannot find torrent %s from path %s", torrentName, req.URL.Path)
 		http.Error(resp, "File not found", http.StatusNotFound)
 		return
 	}
 
 	file, ok := torrent.SelectedFiles.Get(fileName)
 	if !ok || !file.State.Is("ok_file") {
-		// log.Warnf("Cannot find file %s from path %s", fileName, req.URL.Path)
+		log.Errorf("Cannot find file %s from path %s", fileName, req.URL.Path)
 		http.Error(resp, "File not found", http.StatusNotFound)
 		return
 	}
 
 	unrestrict, err := torMgr.UnrestrictFile(file, cfg.ShouldServeFromRclone())
 	if dlErr, ok := err.(*zurghttp.DownloadErrorResponse); ok && dlErr.Message == "bytes_limit_reached" {
-		// log.Warnf("Your account has reached the bandwidth limit, please try again after 12AM CET")
+		// log.Errorf("Your account has reached the bandwidth limit, please try again after 12AM CET")
 		http.Error(resp, "File is not available (bandwidth limit reached)", http.StatusBadRequest)
 		return
 	}
@@ -133,7 +133,7 @@ func (dl *Downloader) DownloadLink(
 	// log.Debugf("Opening file %s (%s)", fileName, link)
 	unrestrict, err := torMgr.UnrestrictLink(link, cfg.ShouldServeFromRclone())
 	if dlErr, ok := err.(*zurghttp.DownloadErrorResponse); ok && dlErr.Message == "bytes_limit_reached" {
-		// log.Warnf("Your account has reached the bandwidth limit, please try again after 12AM CET")
+		// log.Errorf("Your account has reached the bandwidth limit, please try again after 12AM CET")
 		http.Error(resp, "Link is not available (bandwidth limit reached)", http.StatusBadRequest)
 		return
 	}
@@ -172,18 +172,16 @@ func (dl *Downloader) streamFileToResponse(
 	// Add the range header if it exists
 	if req.Header.Get("Range") != "" {
 		dlReq.Header.Add("Range", req.Header.Get("Range"))
-		// log.Debugf("Serving file %s: %s", unrestrict.Filename, req.Header.Get("Range"))
 	}
 
-	// Perform the request
 	downloadResp, err := dl.client.Do(dlReq)
 	if dlErr, ok := err.(*zurghttp.DownloadErrorResponse); ok && dlErr.Message == "bytes_limit_reached" {
-		// log.Warnf("Your account has reached the bandwidth limit, please try again after 12AM CET")
+		// log.Errorf("Your account has reached the bandwidth limit, please try again after 12AM CET")
 		http.Error(resp, "File is not available (bandwidth limit reached)", http.StatusBadRequest)
 		return
 	}
 	if err != nil {
-		log.Warnf("Cannot download file %s: %v", unrestrict.Download, err)
+		log.Errorf("Cannot download file %s: %v", unrestrict.Download, err)
 		if file != nil && file.State.Event(context.Background(), "break_file") == nil {
 			torMgr.EnqueueForRepair(torrent)
 		}
@@ -194,7 +192,7 @@ func (dl *Downloader) streamFileToResponse(
 
 	// Check if the download was not successful
 	if downloadResp.StatusCode != http.StatusOK && downloadResp.StatusCode != http.StatusPartialContent {
-		log.Warnf("Received a %s status code for file %s", downloadResp.Status, unrestrict.Filename)
+		log.Errorf("Received a %s status code for file %s", downloadResp.Status, unrestrict.Filename)
 		if file != nil && file.State.Event(context.Background(), "break_file") == nil {
 			torMgr.EnqueueForRepair(torrent)
 		}
@@ -202,15 +200,43 @@ func (dl *Downloader) streamFileToResponse(
 		return
 	}
 
-	// Copy the content-range header from the download response to the response
 	if cr := downloadResp.Header.Get("Content-Range"); cr != "" {
 		resp.Header().Set("Content-Range", cr)
 	}
 
-	buf := make([]byte, cfg.GetNetworkBufferSize())
+	var n int64
-	n, _ := io.CopyBuffer(resp, downloadResp.Body, buf)
+
+	if cfg.ShouldLogRequests() {
+		var totalBytes int64
+		ticker := time.NewTicker(60 * time.Second)
+		done := make(chan bool)
+		go func() {
+			for {
+				select {
+				case <-ticker.C:
+					mbps := float64(totalBytes*8) / 60_000_000
+					if mbps > 0 {
+						log.Debugf("%s | %.2f MB/s", dlReq.URL, mbps)
+					}
+					totalBytes = 0
+				case <-done:
+					ticker.Stop()
+					return
+				}
+			}
+		}()
+		defer func() {
+			done <- true
+		}()
+		n, _ = io.Copy(io.MultiWriter(resp, &byteCounter{&totalBytes}), downloadResp.Body)
+	} else {
+		n, _ = io.Copy(resp, downloadResp.Body)
+	}
 
 	dl.workerPool.Submit(func() {
+		if n == 0 {
+			return
+		}
 		// Update the download statistics
 		reqBytes, _ := parseRangeHeader(req.Header.Get("Range"))
 		if reqBytes == 0 && unrestrict != nil {
@@ -272,6 +298,16 @@ func parseRangeHeader(rangeHeader string) (uint64, error) {
 	return bytesToRead, nil
 }
 
+type byteCounter struct {
+	totalBytes *int64
+}
+
+func (bc *byteCounter) Write(p []byte) (int, error) {
+	n := len(p)
+	*bc.totalBytes += int64(n)
+	return n, nil
+}
+
 func bToMb(b uint64) uint64 {
 	return b / 1024 / 1024
 }

pkg/http/client.go

@@ -91,11 +91,9 @@ func NewHTTPClient(
 
 	client.client.Transport = &http.Transport{
 		ResponseHeaderTimeout: time.Duration(timeoutSecs) * time.Second,
-		MaxIdleConns:          0,
+		MaxIdleConnsPerHost:   32,
 		MaxConnsPerHost:       32,
-		DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
+		IdleConnTimeout:       90 * time.Second,
-			return dialer.Dial(network, address)
-		},
 	}
 
 	if forceIPv6 {

sample.yml

@@ -19,7 +19,6 @@ token:
 # dump_torrents_every_mins: 1440 # 24 hrs
 enable_repair: true
 
-# network_buffer_size: 32768 # 32kb
 # force_ipv6: false
 # serve_from_rclone: false