chunked_file.go 5.4 KB

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