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.

224 lines
6.4 KiB

4 years ago
5 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
5 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 filer
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math/rand"
  7. "sync"
  8. "github.com/chrislusf/seaweedfs/weed/util/log"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. "github.com/chrislusf/seaweedfs/weed/util/chunk_cache"
  12. "github.com/chrislusf/seaweedfs/weed/wdclient"
  13. "github.com/golang/groupcache/singleflight"
  14. )
  15. type ChunkReadAt struct {
  16. masterClient *wdclient.MasterClient
  17. chunkViews []*ChunkView
  18. lookupFileId LookupFileIdFunctionType
  19. readerLock sync.Mutex
  20. fileSize int64
  21. fetchGroup singleflight.Group
  22. lastChunkFileId string
  23. lastChunkData []byte
  24. chunkCache chunk_cache.ChunkCache
  25. }
  26. // var _ = io.ReaderAt(&ChunkReadAt{})
  27. type LookupFileIdFunctionType func(fileId string) (targetUrls []string, err error)
  28. func LookupFn(filerClient filer_pb.FilerClient) LookupFileIdFunctionType {
  29. vidCache := make(map[string]*filer_pb.Locations)
  30. var vicCacheLock sync.RWMutex
  31. return func(fileId string) (targetUrls []string, err error) {
  32. vid := VolumeId(fileId)
  33. vicCacheLock.RLock()
  34. locations, found := vidCache[vid]
  35. vicCacheLock.RUnlock()
  36. if !found {
  37. util.Retry("lookup volume "+vid, func() error {
  38. err = filerClient.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  39. resp, err := client.LookupVolume(context.Background(), &filer_pb.LookupVolumeRequest{
  40. VolumeIds: []string{vid},
  41. })
  42. if err != nil {
  43. return err
  44. }
  45. locations = resp.LocationsMap[vid]
  46. if locations == nil || len(locations.Locations) == 0 {
  47. log.Infof("failed to locate %s", fileId)
  48. return fmt.Errorf("failed to locate %s", fileId)
  49. }
  50. vicCacheLock.Lock()
  51. vidCache[vid] = locations
  52. vicCacheLock.Unlock()
  53. return nil
  54. })
  55. return err
  56. })
  57. }
  58. if err != nil {
  59. return nil, err
  60. }
  61. for _, loc := range locations.Locations {
  62. volumeServerAddress := filerClient.AdjustedUrl(loc)
  63. targetUrl := fmt.Sprintf("http://%s/%s", volumeServerAddress, fileId)
  64. targetUrls = append(targetUrls, targetUrl)
  65. }
  66. for i := len(targetUrls) - 1; i > 0; i-- {
  67. j := rand.Intn(i + 1)
  68. targetUrls[i], targetUrls[j] = targetUrls[j], targetUrls[i]
  69. }
  70. return
  71. }
  72. }
  73. func NewChunkReaderAtFromClient(filerClient filer_pb.FilerClient, chunkViews []*ChunkView, chunkCache chunk_cache.ChunkCache, fileSize int64) *ChunkReadAt {
  74. return &ChunkReadAt{
  75. chunkViews: chunkViews,
  76. lookupFileId: LookupFn(filerClient),
  77. chunkCache: chunkCache,
  78. fileSize: fileSize,
  79. }
  80. }
  81. func (c *ChunkReadAt) ReadAt(p []byte, offset int64) (n int, err error) {
  82. c.readerLock.Lock()
  83. defer c.readerLock.Unlock()
  84. log.Tracef("ReadAt [%d,%d) of total file size %d bytes %d chunk views", offset, offset+int64(len(p)), c.fileSize, len(c.chunkViews))
  85. return c.doReadAt(p[n:], offset+int64(n))
  86. }
  87. func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
  88. var buffer []byte
  89. startOffset, remaining := offset, int64(len(p))
  90. var nextChunk *ChunkView
  91. for i, chunk := range c.chunkViews {
  92. if remaining <= 0 {
  93. break
  94. }
  95. if i+1 < len(c.chunkViews) {
  96. nextChunk = c.chunkViews[i+1]
  97. } else {
  98. nextChunk = nil
  99. }
  100. if startOffset < chunk.LogicOffset {
  101. gap := int(chunk.LogicOffset - startOffset)
  102. log.Tracef("zero [%d,%d)", startOffset, startOffset+int64(gap))
  103. n += int(min(int64(gap), remaining))
  104. startOffset, remaining = chunk.LogicOffset, remaining-int64(gap)
  105. if remaining <= 0 {
  106. break
  107. }
  108. }
  109. // fmt.Printf(">>> doReadAt [%d,%d), chunk[%d,%d)\n", offset, offset+int64(len(p)), chunk.LogicOffset, chunk.LogicOffset+int64(chunk.Size))
  110. chunkStart, chunkStop := max(chunk.LogicOffset, startOffset), min(chunk.LogicOffset+int64(chunk.Size), startOffset+remaining)
  111. if chunkStart >= chunkStop {
  112. continue
  113. }
  114. log.Tracef("read [%d,%d), %d/%d chunk %s [%d,%d)", chunkStart, chunkStop, i, len(c.chunkViews), chunk.FileId, chunk.LogicOffset-chunk.Offset, chunk.LogicOffset-chunk.Offset+int64(chunk.Size))
  115. buffer, err = c.readFromWholeChunkData(chunk, nextChunk)
  116. if err != nil {
  117. log.Errorf("fetching chunk %+v: %v\n", chunk, err)
  118. return
  119. }
  120. bufferOffset := chunkStart - chunk.LogicOffset + chunk.Offset
  121. copied := copy(p[startOffset-offset:chunkStop-chunkStart+startOffset-offset], buffer[bufferOffset:bufferOffset+chunkStop-chunkStart])
  122. n += copied
  123. startOffset, remaining = startOffset+int64(copied), remaining-int64(copied)
  124. }
  125. log.Tracef("doReadAt [%d,%d), n:%v, err:%v", offset, offset+int64(len(p)), n, err)
  126. if err == nil && remaining > 0 && c.fileSize > startOffset {
  127. delta := int(min(remaining, c.fileSize-startOffset))
  128. log.Tracef("zero2 [%d,%d) of file size %d bytes", startOffset, startOffset+int64(delta), c.fileSize)
  129. n += delta
  130. }
  131. if err == nil && offset+int64(len(p)) >= c.fileSize {
  132. err = io.EOF
  133. }
  134. // fmt.Printf("~~~ filled %d, err: %v\n\n", n, err)
  135. return
  136. }
  137. func (c *ChunkReadAt) readFromWholeChunkData(chunkView *ChunkView, nextChunkViews ...*ChunkView) (chunkData []byte, err error) {
  138. if c.lastChunkFileId == chunkView.FileId {
  139. return c.lastChunkData, nil
  140. }
  141. v, doErr := c.readOneWholeChunk(chunkView)
  142. if doErr != nil {
  143. return nil, doErr
  144. }
  145. chunkData = v.([]byte)
  146. c.lastChunkData = chunkData
  147. c.lastChunkFileId = chunkView.FileId
  148. for _, nextChunkView := range nextChunkViews {
  149. if c.chunkCache != nil && nextChunkView != nil {
  150. go c.readOneWholeChunk(nextChunkView)
  151. }
  152. }
  153. return
  154. }
  155. func (c *ChunkReadAt) readOneWholeChunk(chunkView *ChunkView) (interface{}, error) {
  156. var err error
  157. return c.fetchGroup.Do(chunkView.FileId, func() (interface{}, error) {
  158. log.Tracef("readFromWholeChunkData %s offset %d [%d,%d) size at least %d", chunkView.FileId, chunkView.Offset, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.ChunkSize)
  159. data := c.chunkCache.GetChunk(chunkView.FileId, chunkView.ChunkSize)
  160. if data != nil {
  161. log.Tracef("cache hit %s [%d,%d)", chunkView.FileId, chunkView.LogicOffset-chunkView.Offset, chunkView.LogicOffset-chunkView.Offset+int64(len(data)))
  162. } else {
  163. var err error
  164. data, err = c.doFetchFullChunkData(chunkView)
  165. if err != nil {
  166. return data, err
  167. }
  168. c.chunkCache.SetChunk(chunkView.FileId, data)
  169. }
  170. return data, err
  171. })
  172. }
  173. func (c *ChunkReadAt) doFetchFullChunkData(chunkView *ChunkView) ([]byte, error) {
  174. log.Tracef("+ doFetchFullChunkData %s", chunkView.FileId)
  175. data, err := fetchChunk(c.lookupFileId, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped)
  176. log.Tracef("- doFetchFullChunkData %s", chunkView.FileId)
  177. return data, err
  178. }