reader_at.go 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227
  1. package filer
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math/rand"
  7. "sync"
  8. "github.com/seaweedfs/seaweedfs/weed/glog"
  9. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  10. "github.com/seaweedfs/seaweedfs/weed/util"
  11. "github.com/seaweedfs/seaweedfs/weed/wdclient"
  12. )
  13. type ChunkReadAt struct {
  14. masterClient *wdclient.MasterClient
  15. chunkViews *IntervalList[*ChunkView]
  16. fileSize int64
  17. readerCache *ReaderCache
  18. readerPattern *ReaderPattern
  19. lastChunkFid string
  20. }
  21. var _ = io.ReaderAt(&ChunkReadAt{})
  22. var _ = io.Closer(&ChunkReadAt{})
  23. func LookupFn(filerClient filer_pb.FilerClient) wdclient.LookupFileIdFunctionType {
  24. vidCache := make(map[string]*filer_pb.Locations)
  25. var vicCacheLock sync.RWMutex
  26. return func(fileId string) (targetUrls []string, err error) {
  27. vid := VolumeId(fileId)
  28. vicCacheLock.RLock()
  29. locations, found := vidCache[vid]
  30. vicCacheLock.RUnlock()
  31. if !found {
  32. util.Retry("lookup volume "+vid, func() error {
  33. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
  34. resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
  35. VolumeIds: []string{vid},
  36. })
  37. if err != nil {
  38. return err
  39. }
  40. locations = resp.LocationsMap[vid]
  41. if locations == nil || len(locations.Locations) == 0 {
  42. glog.V(0).Infof("failed to locate %s", fileId)
  43. return fmt.Errorf("failed to locate %s", fileId)
  44. }
  45. vicCacheLock.Lock()
  46. vidCache[vid] = locations
  47. vicCacheLock.Unlock()
  48. return nil
  49. })
  50. return err
  51. })
  52. }
  53. if err != nil {
  54. return nil, err
  55. }
  56. fcDataCenter := filerClient.GetDataCenter()
  57. var sameDcTargetUrls, otherTargetUrls []string
  58. for _, loc := range locations.Locations {
  59. volumeServerAddress := filerClient.AdjustedUrl(loc)
  60. targetUrl := fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
  61. if fcDataCenter == "" || fcDataCenter != loc.DataCenter {
  62. otherTargetUrls = append(otherTargetUrls, targetUrl)
  63. } else {
  64. sameDcTargetUrls = append(sameDcTargetUrls, targetUrl)
  65. }
  66. }
  67. rand.Shuffle(len(sameDcTargetUrls), func(i, j int) {
  68. sameDcTargetUrls[i], sameDcTargetUrls[j] = sameDcTargetUrls[j], sameDcTargetUrls[i]
  69. })
  70. rand.Shuffle(len(otherTargetUrls), func(i, j int) {
  71. otherTargetUrls[i], otherTargetUrls[j] = otherTargetUrls[j], otherTargetUrls[i]
  72. })
  73. // Prefer same data center
  74. targetUrls = append(sameDcTargetUrls, otherTargetUrls...)
  75. return
  76. }
  77. }
  78. func NewChunkReaderAtFromClient(readerCache *ReaderCache, chunkViews *IntervalList[*ChunkView], fileSize int64) *ChunkReadAt {
  79. return &ChunkReadAt{
  80. chunkViews: chunkViews,
  81. fileSize: fileSize,
  82. readerCache: readerCache,
  83. readerPattern: NewReaderPattern(),
  84. }
  85. }
  86. func (c *ChunkReadAt) Close() error {
  87. c.readerCache.destroy()
  88. return nil
  89. }
  90. func (c *ChunkReadAt) ReadAt(p []byte, offset int64) (n int, err error) {
  91. c.readerPattern.MonitorReadAt(offset, len(p))
  92. c.chunkViews.Lock.RLock()
  93. defer c.chunkViews.Lock.RUnlock()
  94. // glog.V(4).Infof("ReadAt [%d,%d) of total file size %d bytes %d chunk views", offset, offset+int64(len(p)), c.fileSize, len(c.chunkViews))
  95. n, _, err = c.doReadAt(p, offset)
  96. return
  97. }
  98. func (c *ChunkReadAt) ReadAtWithTime(p []byte, offset int64) (n int, ts int64, err error) {
  99. c.readerPattern.MonitorReadAt(offset, len(p))
  100. c.chunkViews.Lock.RLock()
  101. defer c.chunkViews.Lock.RUnlock()
  102. // glog.V(4).Infof("ReadAt [%d,%d) of total file size %d bytes %d chunk views", offset, offset+int64(len(p)), c.fileSize, len(c.chunkViews))
  103. return c.doReadAt(p, offset)
  104. }
  105. func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, ts int64, err error) {
  106. startOffset, remaining := offset, int64(len(p))
  107. var nextChunks *Interval[*ChunkView]
  108. for x := c.chunkViews.Front(); x != nil; x = x.Next {
  109. chunk := x.Value
  110. if remaining <= 0 {
  111. break
  112. }
  113. if x.Next != nil {
  114. nextChunks = x.Next
  115. }
  116. if startOffset < chunk.ViewOffset {
  117. gap := chunk.ViewOffset - startOffset
  118. glog.V(4).Infof("zero [%d,%d)", startOffset, chunk.ViewOffset)
  119. n += zero(p, startOffset-offset, gap)
  120. startOffset, remaining = chunk.ViewOffset, remaining-gap
  121. if remaining <= 0 {
  122. break
  123. }
  124. }
  125. // fmt.Printf(">>> doReadAt [%d,%d), chunk[%d,%d)\n", offset, offset+int64(len(p)), chunk.ViewOffset, chunk.ViewOffset+int64(chunk.ViewSize))
  126. chunkStart, chunkStop := max(chunk.ViewOffset, startOffset), min(chunk.ViewOffset+int64(chunk.ViewSize), startOffset+remaining)
  127. if chunkStart >= chunkStop {
  128. continue
  129. }
  130. // glog.V(4).Infof("read [%d,%d), %d/%d chunk %s [%d,%d)", chunkStart, chunkStop, i, len(c.chunkViews), chunk.FileId, chunk.ViewOffset-chunk.Offset, chunk.ViewOffset-chunk.Offset+int64(chunk.ViewSize))
  131. bufferOffset := chunkStart - chunk.ViewOffset + chunk.OffsetInChunk
  132. ts = chunk.ModifiedTsNs
  133. copied, err := c.readChunkSliceAt(p[startOffset-offset:chunkStop-chunkStart+startOffset-offset], chunk, nextChunks, uint64(bufferOffset))
  134. if err != nil {
  135. glog.Errorf("fetching chunk %+v: %v\n", chunk, err)
  136. return copied, ts, err
  137. }
  138. n += copied
  139. startOffset, remaining = startOffset+int64(copied), remaining-int64(copied)
  140. }
  141. // glog.V(4).Infof("doReadAt [%d,%d), n:%v, err:%v", offset, offset+int64(len(p)), n, err)
  142. // zero the remaining bytes if a gap exists at the end of the last chunk (or a fully sparse file)
  143. if err == nil && remaining > 0 {
  144. var delta int64
  145. if c.fileSize > startOffset {
  146. delta = min(remaining, c.fileSize-startOffset)
  147. startOffset -= offset
  148. } else {
  149. delta = remaining
  150. startOffset = max(startOffset-offset, startOffset-remaining-offset)
  151. }
  152. glog.V(4).Infof("zero2 [%d,%d) of file size %d bytes", startOffset, startOffset+delta, c.fileSize)
  153. n += zero(p, startOffset, delta)
  154. }
  155. if err == nil && offset+int64(len(p)) >= c.fileSize {
  156. err = io.EOF
  157. }
  158. // fmt.Printf("~~~ filled %d, err: %v\n\n", n, err)
  159. return
  160. }
  161. func (c *ChunkReadAt) readChunkSliceAt(buffer []byte, chunkView *ChunkView, nextChunkViews *Interval[*ChunkView], offset uint64) (n int, err error) {
  162. if c.readerPattern.IsRandomMode() {
  163. n, err := c.readerCache.chunkCache.ReadChunkAt(buffer, chunkView.FileId, offset)
  164. if n > 0 {
  165. return n, err
  166. }
  167. return fetchChunkRange(buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset))
  168. }
  169. shouldCache := (uint64(chunkView.ViewOffset) + chunkView.ChunkSize) <= c.readerCache.chunkCache.GetMaxFilePartSizeInCache()
  170. n, err = c.readerCache.ReadChunkAt(buffer, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset), int(chunkView.ChunkSize), shouldCache)
  171. if c.lastChunkFid != chunkView.FileId {
  172. if chunkView.OffsetInChunk == 0 { // start of a new chunk
  173. if c.lastChunkFid != "" {
  174. c.readerCache.UnCache(c.lastChunkFid)
  175. }
  176. if nextChunkViews != nil {
  177. c.readerCache.MaybeCache(nextChunkViews) // just read the next chunk if at the very beginning
  178. }
  179. }
  180. }
  181. c.lastChunkFid = chunkView.FileId
  182. return
  183. }
  184. func zero(buffer []byte, start, length int64) int {
  185. end := min(start+length, int64(len(buffer)))
  186. start = max(start, 0)
  187. // zero the bytes
  188. for o := start; o < end; o++ {
  189. buffer[o] = 0
  190. }
  191. return int(end - start)
  192. }