aboutsummaryrefslogtreecommitdiff
path: root/weed/server
diff options
context:
space:
mode:
Diffstat (limited to 'weed/server')
-rw-r--r--weed/server/common.go42
-rw-r--r--weed/server/filer_server_handlers_read.go32
-rw-r--r--weed/server/volume_server_handlers_read.go19
3 files changed, 68 insertions, 25 deletions
diff --git a/weed/server/common.go b/weed/server/common.go
index d88298402..a7d67fb2e 100644
--- a/weed/server/common.go
+++ b/weed/server/common.go
@@ -6,7 +6,6 @@ import (
"encoding/json"
"errors"
"fmt"
- "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
"io"
"io/fs"
"mime/multipart"
@@ -18,6 +17,9 @@ import (
"sync"
"time"
+ "github.com/seaweedfs/seaweedfs/weed/filer"
+ "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+
"google.golang.org/grpc"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -282,7 +284,7 @@ func adjustHeaderContentDisposition(w http.ResponseWriter, r *http.Request, file
}
}
-func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64, mimeType string, writeFn func(writer io.Writer, offset int64, size int64) error) error {
+func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64, mimeType string, prepareWriteFn func(offset int64, size int64) (filer.DoStreamContent, error)) error {
rangeReq := r.Header.Get("Range")
bufferedWriter := writePool.Get().(*bufio.Writer)
bufferedWriter.Reset(w)
@@ -293,7 +295,13 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
if rangeReq == "" {
w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
- if err := writeFn(bufferedWriter, 0, totalSize); err != nil {
+ writeFn, err := prepareWriteFn(0, totalSize)
+ if err != nil {
+ glog.Errorf("processRangeRequest: %v", err)
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return fmt.Errorf("processRangeRequest: %v", err)
+ }
+ if err = writeFn(bufferedWriter); err != nil {
glog.Errorf("processRangeRequest: %v", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return fmt.Errorf("processRangeRequest: %v", err)
@@ -335,8 +343,14 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
w.Header().Set("Content-Length", strconv.FormatInt(ra.length, 10))
w.Header().Set("Content-Range", ra.contentRange(totalSize))
+ writeFn, err := prepareWriteFn(ra.start, ra.length)
+ if err != nil {
+ glog.Errorf("processRangeRequest range[0]: %+v err: %v", w.Header(), err)
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return fmt.Errorf("processRangeRequest: %v", err)
+ }
w.WriteHeader(http.StatusPartialContent)
- err = writeFn(bufferedWriter, ra.start, ra.length)
+ err = writeFn(bufferedWriter)
if err != nil {
glog.Errorf("processRangeRequest range[0]: %+v err: %v", w.Header(), err)
http.Error(w, err.Error(), http.StatusInternalServerError)
@@ -346,11 +360,20 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
}
// process multiple ranges
- for _, ra := range ranges {
+ writeFnByRange := make(map[int](func(writer io.Writer) error))
+
+ for i, ra := range ranges {
if ra.start > totalSize {
http.Error(w, "Out of Range", http.StatusRequestedRangeNotSatisfiable)
return fmt.Errorf("out of range: %v", err)
}
+ writeFn, err := prepareWriteFn(ra.start, ra.length)
+ if err != nil {
+ glog.Errorf("processRangeRequest range[%d] err: %v", i, err)
+ http.Error(w, "Internal Error", http.StatusInternalServerError)
+ return fmt.Errorf("processRangeRequest range[%d] err: %v", i, err)
+ }
+ writeFnByRange[i] = writeFn
}
sendSize := rangesMIMESize(ranges, mimeType, totalSize)
pr, pw := io.Pipe()
@@ -359,13 +382,18 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
sendContent := pr
defer pr.Close() // cause writing goroutine to fail and exit if CopyN doesn't finish.
go func() {
- for _, ra := range ranges {
+ for i, ra := range ranges {
part, e := mw.CreatePart(ra.mimeHeader(mimeType, totalSize))
if e != nil {
pw.CloseWithError(e)
return
}
- if e = writeFn(part, ra.start, ra.length); e != nil {
+ writeFn := writeFnByRange[i]
+ if writeFn == nil {
+ pw.CloseWithError(e)
+ return
+ }
+ if e = writeFn(part); e != nil {
pw.CloseWithError(e)
return
}
diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go
index f5f652f83..d1cd3beae 100644
--- a/weed/server/filer_server_handlers_read.go
+++ b/weed/server/filer_server_handlers_read.go
@@ -231,14 +231,16 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
}
}
- processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64) error {
+ processRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
if offset+size <= int64(len(entry.Content)) {
- _, err := writer.Write(entry.Content[offset : offset+size])
- if err != nil {
- stats.FilerHandlerCounter.WithLabelValues(stats.ErrorWriteEntry).Inc()
- glog.Errorf("failed to write entry content: %v", err)
- }
- return err
+ return func(writer io.Writer) error {
+ _, err := writer.Write(entry.Content[offset : offset+size])
+ if err != nil {
+ stats.FilerHandlerCounter.WithLabelValues(stats.ErrorWriteEntry).Inc()
+ glog.Errorf("failed to write entry content: %v", err)
+ }
+ return err
+ }, nil
}
chunks := entry.GetChunks()
if entry.IsInRemoteOnly() {
@@ -249,17 +251,25 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
}); err != nil {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadCache).Inc()
glog.Errorf("CacheRemoteObjectToLocalCluster %s: %v", entry.FullPath, err)
- return fmt.Errorf("cache %s: %v", entry.FullPath, err)
+ return nil, fmt.Errorf("cache %s: %v", entry.FullPath, err)
} else {
chunks = resp.Entry.GetChunks()
}
}
- err = filer.StreamContentWithThrottler(fs.filer.MasterClient, writer, chunks, offset, size, fs.option.DownloadMaxBytesPs)
+ streamFn, err := filer.PrepareStreamContentWithThrottler(fs.filer.MasterClient, chunks, offset, size, fs.option.DownloadMaxBytesPs)
if err != nil {
stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
- glog.Errorf("failed to stream content %s: %v", r.URL, err)
+ glog.Errorf("failed to prepare stream content %s: %v", r.URL, err)
+ return nil, err
}
- return err
+ return func(writer io.Writer) error {
+ err := streamFn(writer)
+ if err != nil {
+ stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadStream).Inc()
+ glog.Errorf("failed to stream content %s: %v", r.URL, err)
+ }
+ return err
+ }, nil
})
}
diff --git a/weed/server/volume_server_handlers_read.go b/weed/server/volume_server_handlers_read.go
index 5f4beb77d..08e536811 100644
--- a/weed/server/volume_server_handlers_read.go
+++ b/weed/server/volume_server_handlers_read.go
@@ -15,6 +15,7 @@ import (
"sync/atomic"
"time"
+ "github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/util/mem"
@@ -382,12 +383,14 @@ func writeResponseContent(filename, mimeType string, rs io.ReadSeeker, w http.Re
return nil
}
- return processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64) error {
- if _, e = rs.Seek(offset, 0); e != nil {
+ return processRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
+ return func(writer io.Writer) error {
+ if _, e = rs.Seek(offset, 0); e != nil {
+ return e
+ }
+ _, e = io.CopyN(writer, rs, size)
return e
- }
- _, e = io.CopyN(writer, rs, size)
- return e
+ }, nil
})
}
@@ -409,8 +412,10 @@ func (vs *VolumeServer) streamWriteResponseContent(filename string, mimeType str
return
}
- processRangeRequest(r, w, totalSize, mimeType, func(writer io.Writer, offset int64, size int64) error {
- return vs.store.ReadVolumeNeedleDataInto(volumeId, n, readOption, writer, offset, size)
+ processRangeRequest(r, w, totalSize, mimeType, func(offset int64, size int64) (filer.DoStreamContent, error) {
+ return func(writer io.Writer) error {
+ return vs.store.ReadVolumeNeedleDataInto(volumeId, n, readOption, writer, offset, size)
+ }, nil
})
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 18:04:11 +0000