mmap_span/mmap_span.go

   1 package mmap_span
   2
   3 import (
   4         "fmt"
   5         "io"
   6         "sync"
   7
   8         "github.com/anacrolix/torrent/segments"
   9 )
  10
  11 type Mmap interface {
  12         Flush() error
  13         Unmap() error
  14         Bytes() []byte
  15 }
  16
  17 type MMapSpan struct {
  18         mu             sync.RWMutex
  19         mMaps          []Mmap
  20         segmentLocater segments.Index
  21 }
  22
  23 func (ms *MMapSpan) Append(mMap Mmap) {
  24         ms.mMaps = append(ms.mMaps, mMap)
  25 }
  26
  27 func (ms *MMapSpan) Flush() (errs []error) {
  28         ms.mu.RLock()
  29         defer ms.mu.RUnlock()
  30         for _, mMap := range ms.mMaps {
  31                 err := mMap.Flush()
  32                 if err != nil {
  33                         errs = append(errs, err)
  34                 }
  35         }
  36         return
  37 }
  38
  39 func (ms *MMapSpan) Close() (errs []error) {
  40         ms.mu.Lock()
  41         defer ms.mu.Unlock()
  42         for _, mMap := range ms.mMaps {
  43                 err := mMap.Unmap()
  44                 if err != nil {
  45                         errs = append(errs, err)
  46                 }
  47         }
  48         // This is for issue 211.
  49         ms.mMaps = nil
  50         ms.InitIndex()
  51         return
  52 }
  53
  54 func (me *MMapSpan) InitIndex() {
  55         i := 0
  56         me.segmentLocater = segments.NewIndex(func() (segments.Length, bool) {
  57                 if i == len(me.mMaps) {
  58                         return -1, false
  59                 }
  60                 l := int64(len(me.mMaps[i].Bytes()))
  61                 i++
  62                 return l, true
  63         })
  64         // log.Printf("made mmapspan index: %v", me.segmentLocater)
  65 }
  66
  67 func (ms *MMapSpan) ReadAt(p []byte, off int64) (n int, err error) {
  68         // log.Printf("reading %v bytes at %v", len(p), off)
  69         ms.mu.RLock()
  70         defer ms.mu.RUnlock()
  71         n = ms.locateCopy(func(a, b []byte) (_, _ []byte) { return a, b }, p, off)
  72         if n != len(p) {
  73                 err = io.EOF
  74         }
  75         return
  76 }
  77
  78 func copyBytes(dst, src []byte) int {
  79         return copy(dst, src)
  80 }
  81
  82 func (ms *MMapSpan) locateCopy(copyArgs func(remainingArgument, mmapped []byte) (dst, src []byte), p []byte, off int64) (n int) {
  83         ms.segmentLocater.Locate(segments.Extent{off, int64(len(p))}, func(i int, e segments.Extent) bool {
  84                 mMapBytes := ms.mMaps[i].Bytes()[e.Start:]
  85                 // log.Printf("got segment %v: %v, copying %v, %v", i, e, len(p), len(mMapBytes))
  86                 _n := copyBytes(copyArgs(p, mMapBytes))
  87                 p = p[_n:]
  88                 n += _n
  89
  90                 if segments.Int(_n) != e.Length {
  91                         panic(fmt.Sprintf("did %d bytes, expected to do %d", _n, e.Length))
  92                 }
  93                 return true
  94         })
  95         return
  96 }
  97
  98 func (ms *MMapSpan) WriteAt(p []byte, off int64) (n int, err error) {
  99         // log.Printf("writing %v bytes at %v", len(p), off)
 100         ms.mu.RLock()
 101         defer ms.mu.RUnlock()
 102         n = ms.locateCopy(func(a, b []byte) (_, _ []byte) { return b, a }, p, off)
 103         if n != len(p) {
 104                 err = io.ErrShortWrite
 105         }
 106         return
 107 }