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.

179 lines
5.3 KiB

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