initial

src/server/torr/storage/Storage.go

@@ -0,0 +1,15 @@
+package storage
+
+import (
+	"server/torr/storage/state"
+
+	"github.com/anacrolix/torrent/metainfo"
+	"github.com/anacrolix/torrent/storage"
+)
+
+type Storage interface {
+	storage.ClientImpl
+
+	GetStats(hash metainfo.Hash) *state.CacheState
+	CloseHash(hash metainfo.Hash)
+}

src/server/torr/storage/filecache/Cache.go

@@ -0,0 +1 @@
+package filecache

src/server/torr/storage/filecache/Storage.go

@@ -0,0 +1,69 @@
+package filecache
+
+import (
+	"path/filepath"
+	"sync"
+
+	"server/settings"
+	"server/torr/storage"
+	"server/torr/storage/state"
+
+	"github.com/anacrolix/missinggo/filecache"
+	"github.com/anacrolix/torrent/metainfo"
+	storage2 "github.com/anacrolix/torrent/storage"
+)
+
+type Storage struct {
+	storage.Storage
+
+	caches   map[metainfo.Hash]*filecache.Cache
+	capacity int64
+	mu       sync.Mutex
+}
+
+func NewStorage(capacity int64) storage.Storage {
+	stor := new(Storage)
+	stor.capacity = capacity
+	stor.caches = make(map[metainfo.Hash]*filecache.Cache)
+	return stor
+}
+
+func (s *Storage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage2.TorrentImpl, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	path := filepath.Join(settings.Path, "cache", infoHash.String())
+	cache, err := filecache.NewCache(path)
+	if err != nil {
+		return nil, err
+	}
+	cache.SetCapacity(s.capacity)
+	s.caches[infoHash] = cache
+	return storage2.NewResourcePieces(cache.AsResourceProvider()).OpenTorrent(info, infoHash)
+}
+
+func (s *Storage) GetStats(hash metainfo.Hash) *state.CacheState {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return nil
+}
+
+func (s *Storage) Clean() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+}
+
+func (s *Storage) CloseHash(hash metainfo.Hash) {
+	if s.caches == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+}
+
+func (s *Storage) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	return nil
+}

src/server/torr/storage/memcache/Buffer.go

@@ -0,0 +1,86 @@
+package memcache
+
+import (
+	"fmt"
+	"sync"
+
+	"server/utils"
+)
+
+type buffer struct {
+	pieceId int
+	buf     []byte
+	used    bool
+}
+
+type BufferPool struct {
+	buffs map[int]*buffer
+	frees int
+	size  int64
+	mu    sync.Mutex
+}
+
+func NewBufferPool(bufferLength int64, capacity int64) *BufferPool {
+	bp := new(BufferPool)
+	buffsSize := int(capacity/bufferLength) + 3
+	bp.frees = buffsSize
+	bp.size = bufferLength
+	return bp
+}
+
+func (b *BufferPool) mkBuffs() {
+	if b.buffs != nil {
+		return
+	}
+	b.buffs = make(map[int]*buffer, b.frees)
+	fmt.Println("Create", b.frees, "buffers")
+	for i := 0; i < b.frees; i++ {
+		buf := buffer{
+			-1,
+			make([]byte, b.size),
+			false,
+		}
+		b.buffs[i] = &buf
+	}
+}
+
+func (b *BufferPool) GetBuffer(p *Piece) (buff []byte, index int) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.mkBuffs()
+	for id, buf := range b.buffs {
+		if !buf.used {
+			buf.used = true
+			buf.pieceId = p.Id
+			buff = buf.buf
+			index = id
+			b.frees--
+			//fmt.Printf("Get buffer: %v %v %v %p\n", id, p.Id, b.frees, buff)
+			return
+		}
+	}
+	fmt.Println("Create slow buffer")
+	return make([]byte, b.size), -1
+}
+
+func (b *BufferPool) ReleaseBuffer(index int) {
+	if index == -1 {
+		utils.FreeOSMem()
+		return
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.mkBuffs()
+	if buff, ok := b.buffs[index]; ok {
+		buff.used = false
+		buff.pieceId = -1
+		b.frees++
+		//fmt.Println("Release buffer:", index, b.frees)
+	} else {
+		utils.FreeOSMem()
+	}
+}
+
+func (b *BufferPool) Len() int {
+	return b.frees
+}

src/server/torr/storage/memcache/Cache.go

@@ -0,0 +1,196 @@
+package memcache
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+
+	"server/torr/storage/state"
+	"server/utils"
+
+	"github.com/anacrolix/torrent/metainfo"
+	"github.com/anacrolix/torrent/storage"
+)
+
+type Cache struct {
+	storage.TorrentImpl
+
+	s *Storage
+
+	capacity int64
+	filled   int64
+	hash     metainfo.Hash
+
+	pieceLength int64
+	pieceCount  int
+	piecesBuff  int
+
+	muPiece  sync.Mutex
+	muRemove sync.Mutex
+	isRemove bool
+
+	pieces     map[int]*Piece
+	bufferPull *BufferPool
+
+	prcLoaded int
+}
+
+func NewCache(capacity int64, storage *Storage) *Cache {
+	ret := &Cache{
+		capacity: capacity,
+		filled:   0,
+		pieces:   make(map[int]*Piece),
+		s:        storage,
+	}
+
+	return ret
+}
+
+func (c *Cache) Init(info *metainfo.Info, hash metainfo.Hash) {
+	fmt.Println("Create cache for:", info.Name)
+	//Min capacity of 2 pieces length
+	cap := info.PieceLength * 2
+	if c.capacity < cap {
+		c.capacity = cap
+	}
+	c.pieceLength = info.PieceLength
+	c.pieceCount = info.NumPieces()
+	c.piecesBuff = int(c.capacity / c.pieceLength)
+	c.hash = hash
+	c.bufferPull = NewBufferPool(c.pieceLength, c.capacity)
+
+	for i := 0; i < c.pieceCount; i++ {
+		c.pieces[i] = &Piece{
+			Id:     i,
+			Length: info.Piece(i).Length(),
+			Hash:   info.Piece(i).Hash().HexString(),
+			cache:  c,
+		}
+	}
+}
+
+func (c *Cache) Piece(m metainfo.Piece) storage.PieceImpl {
+	c.muPiece.Lock()
+	defer c.muPiece.Unlock()
+	if val, ok := c.pieces[m.Index()]; ok {
+		return val
+	}
+	return nil
+}
+
+func (c *Cache) Close() error {
+	c.isRemove = false
+	fmt.Println("Close cache for:", c.hash)
+	if _, ok := c.s.caches[c.hash]; ok {
+		delete(c.s.caches, c.hash)
+	}
+	c.pieces = nil
+	c.bufferPull = nil
+	utils.FreeOSMemGC()
+	return nil
+}
+
+func (c *Cache) GetState() state.CacheState {
+	cState := state.CacheState{}
+	cState.Capacity = c.capacity
+	cState.PiecesLength = c.pieceLength
+	cState.PiecesCount = c.pieceCount
+	cState.Hash = c.hash.HexString()
+
+	stats := make(map[int]state.ItemState, 0)
+	c.muPiece.Lock()
+	var fill int64 = 0
+	for _, value := range c.pieces {
+		stat := value.Stat()
+		if stat.BufferSize > 0 {
+			fill += stat.BufferSize
+			stats[stat.Id] = stat
+		}
+	}
+	c.filled = fill
+	c.muPiece.Unlock()
+	cState.Filled = c.filled
+	cState.Pieces = stats
+	return cState
+}
+
+func (c *Cache) cleanPieces() {
+	if c.isRemove {
+		return
+	}
+	c.muRemove.Lock()
+	if c.isRemove {
+		c.muRemove.Unlock()
+		return
+	}
+	c.isRemove = true
+	defer func() { c.isRemove = false }()
+	c.muRemove.Unlock()
+
+	remPieces := c.getRemPieces()
+	if len(remPieces) > 0 && (c.capacity < c.filled || c.bufferPull.Len() <= 1) {
+		remCount := int((c.filled - c.capacity) / c.pieceLength)
+		if remCount < 1 {
+			remCount = 1
+		}
+		if remCount > len(remPieces) {
+			remCount = len(remPieces)
+		}
+
+		remPieces = remPieces[:remCount]
+
+		for _, p := range remPieces {
+			c.removePiece(p)
+		}
+	}
+}
+
+func (c *Cache) getRemPieces() []*Piece {
+	pieces := make([]*Piece, 0)
+	fill := int64(0)
+	loading := 0
+	used := make(map[int]struct{})
+	for _, b := range c.bufferPull.buffs {
+		if b.used {
+			used[b.pieceId] = struct{}{}
+		}
+	}
+	for u := range used {
+		v := c.pieces[u]
+		if v.Size > 0 {
+			if v.Id > 0 {
+				pieces = append(pieces, v)
+			}
+			fill += v.Size
+			if !v.complete {
+				loading++
+			}
+		}
+	}
+	c.filled = fill
+	sort.Slice(pieces, func(i, j int) bool {
+		return pieces[i].accessed.Before(pieces[j].accessed)
+	})
+
+	c.prcLoaded = prc(c.piecesBuff-loading, c.piecesBuff)
+	return pieces
+}
+
+func (c *Cache) removePiece(piece *Piece) {
+	c.muPiece.Lock()
+	defer c.muPiece.Unlock()
+	piece.Release()
+
+	st := fmt.Sprintf("%v%% %v\t%s\t%s", c.prcLoaded, piece.Id, piece.accessed.Format("15:04:05.000"), piece.Hash)
+	if c.prcLoaded >= 95 {
+		fmt.Println("Clean memory GC:", st)
+		utils.FreeOSMemGC()
+	} else {
+		fmt.Println("Clean memory:", st)
+		utils.FreeOSMem()
+	}
+}
+
+func prc(val, of int) int {
+	return int(float64(val) * 100.0 / float64(of))
+}

src/server/torr/storage/memcache/Piece.go

@@ -0,0 +1,115 @@
+package memcache
+
+import (
+	"errors"
+	"io"
+	"sync"
+	"time"
+
+	"server/torr/storage/state"
+
+	"github.com/anacrolix/torrent/storage"
+)
+
+type Piece struct {
+	storage.PieceImpl
+
+	Id     int
+	Hash   string
+	Length int64
+	Size   int64
+
+	complete bool
+	readed   bool
+	accessed time.Time
+	buffer   []byte
+	bufIndex int
+
+	mu    sync.RWMutex
+	cache *Cache
+}
+
+func (p *Piece) WriteAt(b []byte, off int64) (n int, err error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.buffer == nil {
+		go p.cache.cleanPieces()
+		p.buffer, p.bufIndex = p.cache.bufferPull.GetBuffer(p)
+		if p.buffer == nil {
+			return 0, errors.New("Can't get buffer write")
+		}
+	}
+	n = copy(p.buffer[off:], b[:])
+	p.Size += int64(n)
+	p.accessed = time.Now()
+	return
+}
+
+func (p *Piece) ReadAt(b []byte, off int64) (n int, err error) {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+
+	size := len(b)
+	if size+int(off) > len(p.buffer) {
+		size = len(p.buffer) - int(off)
+		if size < 0 {
+			size = 0
+		}
+	}
+	if len(p.buffer) < int(off) || len(p.buffer) < int(off)+size {
+		return 0, io.ErrUnexpectedEOF
+	}
+	n = copy(b, p.buffer[int(off) : int(off)+size][:])
+	p.accessed = time.Now()
+	if int(off)+size >= len(p.buffer) {
+		p.readed = true
+	}
+	if int64(len(b))+off >= p.Size {
+		go p.cache.cleanPieces()
+	}
+	return n, nil
+}
+
+func (p *Piece) MarkComplete() error {
+	if len(p.buffer) == 0 {
+		return errors.New("piece is not complete")
+	}
+	p.complete = true
+	return nil
+}
+
+func (p *Piece) MarkNotComplete() error {
+	p.complete = false
+	return nil
+}
+
+func (p *Piece) Completion() storage.Completion {
+	return storage.Completion{
+		Complete: p.complete && len(p.buffer) > 0,
+		Ok:       true,
+	}
+}
+
+func (p *Piece) Release() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.buffer != nil {
+		p.buffer = nil
+		p.cache.bufferPull.ReleaseBuffer(p.bufIndex)
+		p.bufIndex = -1
+	}
+	p.Size = 0
+	p.complete = false
+}
+
+func (p *Piece) Stat() state.ItemState {
+	itm := state.ItemState{
+		Id:         p.Id,
+		Hash:       p.Hash,
+		Accessed:   p.accessed,
+		Completed:  p.complete,
+		BufferSize: p.Size,
+	}
+	return itm
+}

src/server/torr/storage/memcache/Storage.go

@@ -0,0 +1,66 @@
+package memcache
+
+import (
+	"sync"
+
+	"server/torr/storage"
+	"server/torr/storage/state"
+
+	"github.com/anacrolix/torrent/metainfo"
+	storage2 "github.com/anacrolix/torrent/storage"
+)
+
+type Storage struct {
+	storage.Storage
+
+	caches   map[metainfo.Hash]*Cache
+	capacity int64
+	mu       sync.Mutex
+}
+
+func NewStorage(capacity int64) storage.Storage {
+	stor := new(Storage)
+	stor.capacity = capacity
+	stor.caches = make(map[metainfo.Hash]*Cache)
+	return stor
+}
+
+func (s *Storage) OpenTorrent(info *metainfo.Info, infoHash metainfo.Hash) (storage2.TorrentImpl, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	ch := NewCache(s.capacity, s)
+	ch.Init(info, infoHash)
+	s.caches[infoHash] = ch
+	return ch, nil
+}
+
+func (s *Storage) GetStats(hash metainfo.Hash) *state.CacheState {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if c, ok := s.caches[hash]; ok {
+		st := c.GetState()
+		return &st
+	}
+	return nil
+}
+
+func (s *Storage) CloseHash(hash metainfo.Hash) {
+	if s.caches == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if ch, ok := s.caches[hash]; ok {
+		ch.Close()
+		delete(s.caches, hash)
+	}
+}
+
+func (s *Storage) Close() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, ch := range s.caches {
+		ch.Close()
+	}
+	return nil
+}

src/server/torr/storage/state/state.go

@@ -0,0 +1,22 @@
+package state
+
+import (
+	"time"
+)
+
+type CacheState struct {
+	Hash         string
+	Capacity     int64
+	Filled       int64
+	PiecesLength int64
+	PiecesCount  int
+	Pieces       map[int]ItemState
+}
+
+type ItemState struct {
+	Id         int
+	Accessed   time.Time
+	BufferSize int64
+	Completed  bool
+	Hash       string
+}