You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

204 lines
6.2 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
3 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
3 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
  1. package weed_server
  2. import (
  3. "bytes"
  4. "crypto/md5"
  5. "fmt"
  6. "golang.org/x/exp/slices"
  7. "hash"
  8. "io"
  9. "net/http"
  10. "strconv"
  11. "strings"
  12. "sync"
  13. "sync/atomic"
  14. "time"
  15. "github.com/seaweedfs/seaweedfs/weed/filer"
  16. "github.com/seaweedfs/seaweedfs/weed/glog"
  17. "github.com/seaweedfs/seaweedfs/weed/operation"
  18. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  19. "github.com/seaweedfs/seaweedfs/weed/security"
  20. "github.com/seaweedfs/seaweedfs/weed/stats"
  21. "github.com/seaweedfs/seaweedfs/weed/util"
  22. )
  23. var bufPool = sync.Pool{
  24. New: func() interface{} {
  25. return new(bytes.Buffer)
  26. },
  27. }
  28. func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
  29. query := r.URL.Query()
  30. isAppend := isAppend(r)
  31. if query.Has("offset") {
  32. offset := query.Get("offset")
  33. offsetInt, err := strconv.ParseInt(offset, 10, 64)
  34. if err != nil || offsetInt < 0 {
  35. err = fmt.Errorf("invalid 'offset': '%s'", offset)
  36. return nil, nil, 0, err, nil
  37. }
  38. if isAppend && offsetInt > 0 {
  39. err = fmt.Errorf("cannot set offset when op=append")
  40. return nil, nil, 0, err, nil
  41. }
  42. chunkOffset = offsetInt
  43. }
  44. md5Hash = md5.New()
  45. var partReader = io.NopCloser(io.TeeReader(reader, md5Hash))
  46. var wg sync.WaitGroup
  47. var bytesBufferCounter int64
  48. bytesBufferLimitCond := sync.NewCond(new(sync.Mutex))
  49. var fileChunksLock sync.Mutex
  50. for {
  51. // need to throttle used byte buffer
  52. bytesBufferLimitCond.L.Lock()
  53. for atomic.LoadInt64(&bytesBufferCounter) >= 4 {
  54. glog.V(4).Infof("waiting for byte buffer %d", bytesBufferCounter)
  55. bytesBufferLimitCond.Wait()
  56. }
  57. atomic.AddInt64(&bytesBufferCounter, 1)
  58. bytesBufferLimitCond.L.Unlock()
  59. bytesBuffer := bufPool.Get().(*bytes.Buffer)
  60. glog.V(4).Infof("received byte buffer %d", bytesBufferCounter)
  61. limitedReader := io.LimitReader(partReader, int64(chunkSize))
  62. bytesBuffer.Reset()
  63. dataSize, err := bytesBuffer.ReadFrom(limitedReader)
  64. // data, err := io.ReadAll(limitedReader)
  65. if err != nil || dataSize == 0 {
  66. bufPool.Put(bytesBuffer)
  67. atomic.AddInt64(&bytesBufferCounter, -1)
  68. bytesBufferLimitCond.Signal()
  69. uploadErr = err
  70. break
  71. }
  72. if chunkOffset == 0 && !isAppend {
  73. if dataSize < fs.option.SaveToFilerLimit || strings.HasPrefix(r.URL.Path, filer.DirectoryEtcRoot) {
  74. chunkOffset += dataSize
  75. smallContent = make([]byte, dataSize)
  76. bytesBuffer.Read(smallContent)
  77. bufPool.Put(bytesBuffer)
  78. atomic.AddInt64(&bytesBufferCounter, -1)
  79. bytesBufferLimitCond.Signal()
  80. stats.FilerRequestCounter.WithLabelValues(stats.ContentSaveToFiler).Inc()
  81. break
  82. }
  83. } else {
  84. stats.FilerRequestCounter.WithLabelValues(stats.AutoChunk).Inc()
  85. }
  86. wg.Add(1)
  87. go func(offset int64) {
  88. defer func() {
  89. bufPool.Put(bytesBuffer)
  90. atomic.AddInt64(&bytesBufferCounter, -1)
  91. bytesBufferLimitCond.Signal()
  92. wg.Done()
  93. }()
  94. chunk, toChunkErr := fs.dataToChunk(fileName, contentType, bytesBuffer.Bytes(), offset, so)
  95. if toChunkErr != nil {
  96. uploadErr = toChunkErr
  97. }
  98. if chunk != nil {
  99. fileChunksLock.Lock()
  100. fileChunks = append(fileChunks, chunk)
  101. fileChunksLock.Unlock()
  102. glog.V(4).Infof("uploaded %s chunk %d to %s [%d,%d)", fileName, len(fileChunks), chunk.FileId, offset, offset+int64(chunk.Size))
  103. }
  104. }(chunkOffset)
  105. // reset variables for the next chunk
  106. chunkOffset = chunkOffset + dataSize
  107. // if last chunk was not at full chunk size, but already exhausted the reader
  108. if dataSize < int64(chunkSize) {
  109. break
  110. }
  111. }
  112. wg.Wait()
  113. if uploadErr != nil {
  114. fs.filer.DeleteChunks(fileChunks)
  115. return nil, md5Hash, 0, uploadErr, nil
  116. }
  117. slices.SortFunc(fileChunks, func(a, b *filer_pb.FileChunk) bool {
  118. return a.Offset < b.Offset
  119. })
  120. return fileChunks, md5Hash, chunkOffset, nil, smallContent
  121. }
  122. func (fs *FilerServer) doUpload(urlLocation string, limitedReader io.Reader, fileName string, contentType string, pairMap map[string]string, auth security.EncodedJwt) (*operation.UploadResult, error, []byte) {
  123. stats.FilerRequestCounter.WithLabelValues(stats.ChunkUpload).Inc()
  124. start := time.Now()
  125. defer func() {
  126. stats.FilerRequestHistogram.WithLabelValues(stats.ChunkUpload).Observe(time.Since(start).Seconds())
  127. }()
  128. uploadOption := &operation.UploadOption{
  129. UploadUrl: urlLocation,
  130. Filename: fileName,
  131. Cipher: fs.option.Cipher,
  132. IsInputCompressed: false,
  133. MimeType: contentType,
  134. PairMap: pairMap,
  135. Jwt: auth,
  136. }
  137. uploadResult, err, data := operation.Upload(limitedReader, uploadOption)
  138. if uploadResult != nil && uploadResult.RetryCount > 0 {
  139. stats.FilerRequestCounter.WithLabelValues(stats.ChunkUploadRetry).Add(float64(uploadResult.RetryCount))
  140. }
  141. return uploadResult, err, data
  142. }
  143. func (fs *FilerServer) dataToChunk(fileName, contentType string, data []byte, chunkOffset int64, so *operation.StorageOption) (*filer_pb.FileChunk, error) {
  144. dataReader := util.NewBytesReader(data)
  145. // retry to assign a different file id
  146. var fileId, urlLocation string
  147. var auth security.EncodedJwt
  148. var uploadErr error
  149. var uploadResult *operation.UploadResult
  150. for i := 0; i < 3; i++ {
  151. // assign one file id for one chunk
  152. fileId, urlLocation, auth, uploadErr = fs.assignNewFileInfo(so)
  153. if uploadErr != nil {
  154. glog.V(4).Infof("retry later due to assign error: %v", uploadErr)
  155. stats.FilerRequestCounter.WithLabelValues(stats.ChunkAssignRetry).Inc()
  156. time.Sleep(time.Duration(i+1) * 251 * time.Millisecond)
  157. continue
  158. }
  159. // upload the chunk to the volume server
  160. uploadResult, uploadErr, _ = fs.doUpload(urlLocation, dataReader, fileName, contentType, nil, auth)
  161. if uploadErr != nil {
  162. glog.V(4).Infof("retry later due to upload error: %v", uploadErr)
  163. stats.FilerRequestCounter.WithLabelValues(stats.ChunkDoUploadRetry).Inc()
  164. time.Sleep(time.Duration(i+1) * 251 * time.Millisecond)
  165. continue
  166. }
  167. break
  168. }
  169. if uploadErr != nil {
  170. glog.Errorf("upload error: %v", uploadErr)
  171. return nil, uploadErr
  172. }
  173. // if last chunk exhausted the reader exactly at the border
  174. if uploadResult.Size == 0 {
  175. return nil, nil
  176. }
  177. return uploadResult.ToPbFileChunk(fileId, chunkOffset), nil
  178. }