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.

232 lines
6.6 KiB

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