chunked_file.go 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239
  1. package operation
  2. import (
  3. "context"
  4. "encoding/json"
  5. "errors"
  6. "fmt"
  7. "io"
  8. "net/http"
  9. "sort"
  10. "sync"
  11. "google.golang.org/grpc"
  12. "github.com/seaweedfs/seaweedfs/weed/glog"
  13. "github.com/seaweedfs/seaweedfs/weed/pb"
  14. "github.com/seaweedfs/seaweedfs/weed/util"
  15. )
  16. var (
  17. // when the remote server does not allow range requests (Accept-Ranges was not set)
  18. ErrRangeRequestsNotSupported = errors.New("Range requests are not supported by the remote server")
  19. // ErrInvalidRange is returned by Read when trying to read past the end of the file
  20. ErrInvalidRange = errors.New("Invalid range")
  21. )
  22. type ChunkInfo struct {
  23. Fid string `json:"fid"`
  24. Offset int64 `json:"offset"`
  25. Size int64 `json:"size"`
  26. }
  27. type ChunkList []*ChunkInfo
  28. type ChunkManifest struct {
  29. Name string `json:"name,omitempty"`
  30. Mime string `json:"mime,omitempty"`
  31. Size int64 `json:"size,omitempty"`
  32. Chunks ChunkList `json:"chunks,omitempty"`
  33. }
  34. // seekable chunked file reader
  35. type ChunkedFileReader struct {
  36. totalSize int64
  37. chunkList []*ChunkInfo
  38. master pb.ServerAddress
  39. pos int64
  40. pr *io.PipeReader
  41. pw *io.PipeWriter
  42. mutex sync.Mutex
  43. grpcDialOption grpc.DialOption
  44. }
  45. func (s ChunkList) Len() int { return len(s) }
  46. func (s ChunkList) Less(i, j int) bool { return s[i].Offset < s[j].Offset }
  47. func (s ChunkList) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
  48. func LoadChunkManifest(buffer []byte, isCompressed bool) (*ChunkManifest, error) {
  49. if isCompressed {
  50. var err error
  51. if buffer, err = util.DecompressData(buffer); err != nil {
  52. glog.V(0).Infof("fail to decompress chunk manifest: %v", err)
  53. }
  54. }
  55. cm := ChunkManifest{}
  56. if e := json.Unmarshal(buffer, &cm); e != nil {
  57. return nil, e
  58. }
  59. sort.Sort(cm.Chunks)
  60. return &cm, nil
  61. }
  62. func (cm *ChunkManifest) Marshal() ([]byte, error) {
  63. return json.Marshal(cm)
  64. }
  65. func (cm *ChunkManifest) DeleteChunks(masterFn GetMasterFn, usePublicUrl bool, grpcDialOption grpc.DialOption) error {
  66. var fileIds []string
  67. for _, ci := range cm.Chunks {
  68. fileIds = append(fileIds, ci.Fid)
  69. }
  70. results, err := DeleteFiles(masterFn, usePublicUrl, grpcDialOption, fileIds)
  71. if err != nil {
  72. glog.V(0).Infof("delete %+v: %v", fileIds, err)
  73. return fmt.Errorf("chunk delete: %v", err)
  74. }
  75. for _, result := range results {
  76. if result.Error != "" {
  77. glog.V(0).Infof("delete file %+v: %v", result.FileId, result.Error)
  78. return fmt.Errorf("chunk delete %v: %v", result.FileId, result.Error)
  79. }
  80. }
  81. return nil
  82. }
  83. func readChunkNeedle(fileUrl string, w io.Writer, offset int64, jwt string) (written int64, e error) {
  84. req, err := http.NewRequest("GET", fileUrl, nil)
  85. if err != nil {
  86. return written, err
  87. }
  88. if offset > 0 {
  89. req.Header.Set("Range", fmt.Sprintf("bytes=%d-", offset))
  90. }
  91. resp, err := util.Do(req)
  92. if err != nil {
  93. return written, err
  94. }
  95. defer util.CloseResponse(resp)
  96. switch resp.StatusCode {
  97. case http.StatusRequestedRangeNotSatisfiable:
  98. return written, ErrInvalidRange
  99. case http.StatusOK:
  100. if offset > 0 {
  101. return written, ErrRangeRequestsNotSupported
  102. }
  103. case http.StatusPartialContent:
  104. break
  105. default:
  106. return written, fmt.Errorf("Read chunk needle error: [%d] %s", resp.StatusCode, fileUrl)
  107. }
  108. return io.Copy(w, resp.Body)
  109. }
  110. func NewChunkedFileReader(chunkList []*ChunkInfo, master pb.ServerAddress, grpcDialOption grpc.DialOption) *ChunkedFileReader {
  111. var totalSize int64
  112. for _, chunk := range chunkList {
  113. totalSize += chunk.Size
  114. }
  115. sort.Sort(ChunkList(chunkList))
  116. return &ChunkedFileReader{
  117. totalSize: totalSize,
  118. chunkList: chunkList,
  119. master: master,
  120. grpcDialOption: grpcDialOption,
  121. }
  122. }
  123. func (cf *ChunkedFileReader) Seek(offset int64, whence int) (int64, error) {
  124. var err error
  125. switch whence {
  126. case io.SeekStart:
  127. case io.SeekCurrent:
  128. offset += cf.pos
  129. case io.SeekEnd:
  130. offset = cf.totalSize + offset
  131. }
  132. if offset > cf.totalSize {
  133. err = ErrInvalidRange
  134. }
  135. if cf.pos != offset {
  136. cf.Close()
  137. }
  138. cf.pos = offset
  139. return cf.pos, err
  140. }
  141. func (cf *ChunkedFileReader) WriteTo(w io.Writer) (n int64, err error) {
  142. chunkIndex := -1
  143. chunkStartOffset := int64(0)
  144. for i, ci := range cf.chunkList {
  145. if cf.pos >= ci.Offset && cf.pos < ci.Offset+ci.Size {
  146. chunkIndex = i
  147. chunkStartOffset = cf.pos - ci.Offset
  148. break
  149. }
  150. }
  151. if chunkIndex < 0 {
  152. return n, ErrInvalidRange
  153. }
  154. for ; chunkIndex < len(cf.chunkList); chunkIndex++ {
  155. ci := cf.chunkList[chunkIndex]
  156. // if we need read date from local volume server first?
  157. fileUrl, jwt, lookupError := LookupFileId(func(_ context.Context) pb.ServerAddress {
  158. return cf.master
  159. }, cf.grpcDialOption, ci.Fid)
  160. if lookupError != nil {
  161. return n, lookupError
  162. }
  163. if wn, e := readChunkNeedle(fileUrl, w, chunkStartOffset, jwt); e != nil {
  164. return n, e
  165. } else {
  166. n += wn
  167. cf.pos += wn
  168. }
  169. chunkStartOffset = 0
  170. }
  171. return n, nil
  172. }
  173. func (cf *ChunkedFileReader) ReadAt(p []byte, off int64) (n int, err error) {
  174. cf.Seek(off, 0)
  175. return cf.Read(p)
  176. }
  177. func (cf *ChunkedFileReader) Read(p []byte) (int, error) {
  178. return cf.getPipeReader().Read(p)
  179. }
  180. func (cf *ChunkedFileReader) Close() (e error) {
  181. cf.mutex.Lock()
  182. defer cf.mutex.Unlock()
  183. return cf.closePipe()
  184. }
  185. func (cf *ChunkedFileReader) closePipe() (e error) {
  186. if cf.pr != nil {
  187. if err := cf.pr.Close(); err != nil {
  188. e = err
  189. }
  190. }
  191. cf.pr = nil
  192. if cf.pw != nil {
  193. if err := cf.pw.Close(); err != nil {
  194. e = err
  195. }
  196. }
  197. cf.pw = nil
  198. return e
  199. }
  200. func (cf *ChunkedFileReader) getPipeReader() io.Reader {
  201. cf.mutex.Lock()
  202. defer cf.mutex.Unlock()
  203. if cf.pr != nil && cf.pw != nil {
  204. return cf.pr
  205. }
  206. cf.closePipe()
  207. cf.pr, cf.pw = io.Pipe()
  208. go func(pw *io.PipeWriter) {
  209. _, e := cf.WriteTo(pw)
  210. pw.CloseWithError(e)
  211. }(cf.pw)
  212. return cf.pr
  213. }