update to master

server/torr/apihelper.go

@@ -2,6 +2,7 @@ package torr
 
 import (
 	"io"
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sort"
@@ -125,22 +126,19 @@ func SetTorrent(hashHex, title, poster, data string) *Torrent {
 
 func RemTorrent(hashHex string) {
 	hash := metainfo.NewHashFromHex(hashHex)
-	bts.RemoveTorrent(hash)
-	RemTorrentDB(hash)
-	if sets.BTsets.UseDisk &&
-		hashHex != "" &&
-		hashHex != "/" &&
-		sets.BTsets.TorrentsSavePath != "" &&
-		sets.BTsets.TorrentsSavePath != "/" {
-
+	if sets.BTsets.UseDisk && hashHex != "" && hashHex != "/" {
 		name := filepath.Join(sets.BTsets.TorrentsSavePath, hashHex)
-		err := os.RemoveAll(name)
+		ff, _ := ioutil.ReadDir(name)
+		for _, f := range ff {
+			os.Remove(filepath.Join(name, f.Name()))
+		}
+		err := os.Remove(name)
 		if err != nil {
 			log.TLogln("Error remove cache:", err)
-		} else {
-			log.TLogln("Remove cache from disk:", hashHex)
 		}
 	}
+	bts.RemoveTorrent(hash)
+	RemTorrentDB(hash)
 }
 
 func ListTorrent() []*Torrent {

server/torr/preload.go

@@ -0,0 +1,154 @@
+package torr
+
+import (
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/anacrolix/torrent"
+
+	"server/log"
+	"server/settings"
+	"server/torr/state"
+	utils2 "server/utils"
+)
+
+func (t *Torrent) Preload(index int, size int64) {
+	if size <= 0 {
+		return
+	}
+	t.PreloadSize = size
+
+	if t.Stat == state.TorrentGettingInfo {
+		if !t.WaitInfo() {
+			return
+		}
+		// wait change status
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	t.muTorrent.Lock()
+	if t.Stat != state.TorrentWorking {
+		t.muTorrent.Unlock()
+		return
+	}
+
+	t.Stat = state.TorrentPreload
+	t.muTorrent.Unlock()
+
+	defer func() {
+		if t.Stat == state.TorrentPreload {
+			t.Stat = state.TorrentWorking
+		}
+	}()
+
+	file := t.findFileIndex(index)
+	if file == nil {
+		file = t.Files()[0]
+	}
+
+	if size > file.Length() {
+		size = file.Length()
+	}
+
+	if t.Info() != nil {
+		// Запуск лога в отдельном потоке
+		go func() {
+			for t.Stat == state.TorrentPreload {
+				stat := fmt.Sprint(file.Torrent().InfoHash().HexString(), " ", utils2.Format(float64(t.PreloadedBytes)), "/", utils2.Format(float64(t.PreloadSize)), " Speed:", utils2.Format(t.DownloadSpeed), " Peers:[", t.Torrent.Stats().ConnectedSeeders, "]", t.Torrent.Stats().ActivePeers, "/", t.Torrent.Stats().TotalPeers)
+				log.TLogln("Preload:", stat)
+				t.AddExpiredTime(time.Second * time.Duration(settings.BTsets.TorrentDisconnectTimeout))
+				time.Sleep(time.Second)
+			}
+		}()
+
+		mb5 := int64(5 * 1024 * 1024)
+
+		readerStart := file.NewReader()
+		defer readerStart.Close()
+		readerStart.SetResponsive()
+		readerStart.SetReadahead(0)
+		readerStartEnd := size - mb5
+
+		if readerStartEnd < 0 {
+			// Если конец начального ридера оказался за началом
+			readerStartEnd = size
+		}
+		if readerStartEnd > file.Length() {
+			// Если конец начального ридера оказался после конца файла
+			readerStartEnd = file.Length()
+		}
+
+		readerEndStart := file.Length() - mb5
+		readerEndEnd := file.Length()
+
+		var wa sync.WaitGroup
+		go func() {
+			offset := int64(0)
+			if readerEndStart > readerStartEnd {
+				// Если конечный ридер не входит в диапозон начального
+				wa.Add(1)
+				defer wa.Done()
+				readerEnd := file.NewReader()
+				readerEnd.SetResponsive()
+				readerEnd.SetReadahead(0)
+				readerEnd.Seek(readerEndStart, io.SeekStart)
+				offset = readerEndStart
+				tmp := make([]byte, 32768, 32768)
+				for offset+int64(len(tmp)) < readerEndEnd {
+					n, err := readerEnd.Read(tmp)
+					if err != nil {
+						break
+					}
+					offset += int64(n)
+				}
+				readerEnd.Close()
+			}
+		}()
+
+		pieceLength := t.Info().PieceLength
+		readahead := pieceLength * 4
+		if readerStartEnd < readahead {
+			readahead = 0
+		}
+		readerStart.SetReadahead(readahead)
+		offset := int64(0)
+		tmp := make([]byte, 32768, 32768)
+		for offset+int64(len(tmp)) < readerStartEnd {
+			n, err := readerStart.Read(tmp)
+			if err != nil {
+				log.TLogln("Error preload:", err)
+				return
+			}
+			offset += int64(n)
+			if readahead > 0 && readerStartEnd-(offset+int64(len(tmp))) < readahead {
+				readahead = 0
+				readerStart.SetReadahead(0)
+			}
+		}
+
+		wa.Wait()
+	}
+	log.TLogln("End preload:", file.Torrent().InfoHash().HexString(), "Peers:[", t.Torrent.Stats().ConnectedSeeders, "]", t.Torrent.Stats().ActivePeers, "/", t.Torrent.Stats().TotalPeers)
+}
+
+func (t *Torrent) findFileIndex(index int) *torrent.File {
+	st := t.Status()
+	var stFile *state.TorrentFileStat
+	for _, f := range st.FileStats {
+		if index == f.Id {
+			stFile = f
+			break
+		}
+	}
+	if stFile == nil {
+		return nil
+	}
+	for _, file := range t.Files() {
+		if file.Path() == stFile.Path {
+			return file
+		}
+	}
+	return nil
+}

server/torr/storage/torrstor/cache.go

@@ -1,11 +1,9 @@
 package torrstor
 
 import (
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sort"
-	"strconv"
 	"sync"
 	"time"
 
@@ -76,21 +74,6 @@ func (c *Cache) Init(info *metainfo.Info, hash metainfo.Hash) {
 	for i := 0; i < c.pieceCount; i++ {
 		c.pieces[i] = NewPiece(i, c)
 	}
-
-	if settings.BTsets.UseDisk {
-		name := filepath.Join(settings.BTsets.TorrentsSavePath, hash.HexString())
-		fs, err := ioutil.ReadDir(name)
-		if err == nil {
-			for _, f := range fs {
-				id, err := strconv.Atoi(f.Name())
-				if err == nil {
-					c.pieces[id].Size = f.Size()
-					c.pieces[id].Complete = f.Size() == c.pieceLength
-					c.pieces[id].Accessed = f.ModTime().Unix()
-				}
-			}
-		}
-	}
 }
 
 func (c *Cache) SetTorrent(torr *torrent.Torrent) {
@@ -107,6 +90,19 @@ func (c *Cache) Piece(m metainfo.Piece) storage.PieceImpl {
 func (c *Cache) Close() error {
 	log.TLogln("Close cache for:", c.hash)
 	delete(c.storage.caches, c.hash)
+
+	if settings.BTsets.RemoveCacheOnDrop {
+		name := filepath.Join(settings.BTsets.TorrentsSavePath, c.hash.HexString())
+		if name != "" && name != "/" {
+			for _, v := range c.pieces {
+				if v.dPiece != nil {
+					os.Remove(v.dPiece.name)
+				}
+			}
+			os.Remove(name)
+		}
+	}
+
 	c.pieces = nil
 
 	c.muReaders.Lock()
@@ -137,28 +133,33 @@ func (c *Cache) GetState() *state.CacheState {
 
 	piecesState := make(map[int]state.ItemState, 0)
 	var fill int64 = 0
-	for _, p := range c.pieces {
-		if p.Size > 0 {
-			fill += p.Size
-			piecesState[p.Id] = state.ItemState{
-				Id:        p.Id,
-				Size:      p.Size,
-				Length:    c.pieceLength,
-				Completed: p.Complete,
+
+	if len(c.pieces) > 0 {
+		for _, p := range c.pieces {
+			if p.Size > 0 {
+				fill += p.Size
+				piecesState[p.Id] = state.ItemState{
+					Id:        p.Id,
+					Size:      p.Size,
+					Length:    c.pieceLength,
+					Completed: p.Complete,
+				}
 			}
 		}
 	}
 
 	readersState := make([]*state.ReaderState, 0)
 	c.muReaders.Lock()
-	for r, _ := range c.readers {
-		rng := r.getPiecesRange()
-		pc := r.getReaderPiece()
-		readersState = append(readersState, &state.ReaderState{
-			Start:  rng.Start,
-			End:    rng.End,
-			Reader: pc,
-		})
+	if len(c.readers) > 0 {
+		for r, _ := range c.readers {
+			rng := r.getPiecesRange()
+			pc := r.getReaderPiece()
+			readersState = append(readersState, &state.ReaderState{
+				Start:  rng.Start,
+				End:    rng.End,
+				Reader: pc,
+			})
+		}
 	}
 	c.muReaders.Unlock()
 
@@ -224,6 +225,12 @@ func (c *Cache) getRemPieces() []*Piece {
 					piecesRemove = append(piecesRemove, p)
 				}
 			}
+		} else {
+			// on preload clean
+			//TODO проверить
+			if p.Size > 0 && !c.isIdInFileBE(ranges, id) {
+				piecesRemove = append(piecesRemove, p)
+			}
 		}
 	}
 
@@ -269,68 +276,6 @@ func (c *Cache) isIdInFileBE(ranges []Range, id int) bool {
 	return false
 }
 
-// run only in cache on disk
-func (c *Cache) LoadPiecesOnDisk() {
-	if c.torrent == nil {
-		return
-	}
-
-	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()
-
-	ranges := make([]Range, 0)
-	c.muReaders.Lock()
-	for r, _ := range c.readers {
-		ranges = append(ranges, r.getPiecesRange())
-	}
-	c.muReaders.Unlock()
-	ranges = mergeRange(ranges)
-
-	for r, _ := range c.readers {
-		pc := r.getReaderPiece()
-		limit := 5
-
-		for limit > 0 {
-			if !c.pieces[pc].Complete {
-				if c.torrent.PieceState(pc).Priority == torrent.PiecePriorityNone {
-					c.torrent.Piece(pc).SetPriority(torrent.PiecePriorityNormal)
-				}
-				limit--
-			}
-			pc++
-		}
-	}
-	if len(c.readers) == 0 {
-		limit := 5
-		pc := 0
-		end := c.pieceCount
-		for pc <= end {
-			if !c.pieces[pc].Complete {
-				break
-			}
-			pc++
-		}
-		for pc <= end && limit > 0 {
-			if !c.pieces[pc].Complete {
-				if c.torrent.PieceState(pc).Priority == torrent.PiecePriorityNone {
-					c.torrent.Piece(pc).SetPriority(torrent.PiecePriorityNormal)
-				}
-				limit--
-			}
-			pc++
-		}
-	}
-}
-
 //////////////////
 // Reader section
 ////////

server/torr/storage/torrstor/diskpiece.go

@@ -1,12 +1,15 @@
 package torrstor
 
 import (
+	"io"
 	"os"
 	"path/filepath"
 	"strconv"
 	"sync"
 	"time"
 
+	"github.com/anacrolix/torrent"
+
 	"server/log"
 	"server/settings"
 )
@@ -14,30 +17,38 @@ import (
 type DiskPiece struct {
 	piece *Piece
 
-	file *os.File
+	name string
 
 	mu sync.RWMutex
 }
 
 func NewDiskPiece(p *Piece) *DiskPiece {
 	name := filepath.Join(settings.BTsets.TorrentsSavePath, p.cache.hash.HexString(), strconv.Itoa(p.Id))
-	ff, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, 0666)
-	if err != nil {
-		log.TLogln("Error open file:", err)
-		return nil
+	ff, err := os.Stat(name)
+	if err == nil {
+		p.Size = ff.Size()
+		p.Complete = ff.Size() == p.cache.pieceLength
+		p.Accessed = ff.ModTime().Unix()
 	}
-	return &DiskPiece{piece: p, file: ff}
+	return &DiskPiece{piece: p, name: name}
 }
 
 func (p *DiskPiece) WriteAt(b []byte, off int64) (n int, err error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	n, err = p.file.WriteAt(b, off)
-
-	go p.piece.cache.LoadPiecesOnDisk()
+	ff, err := os.OpenFile(p.name, os.O_RDWR|os.O_CREATE, 0666)
+	if err != nil {
+		log.TLogln("Error open file:", err)
+		return 0, err
+	}
+	defer ff.Close()
+	n, err = ff.WriteAt(b, off)
 
 	p.piece.Size += int64(n)
+	if p.piece.Size > p.piece.cache.pieceLength {
+		p.piece.Size = p.piece.cache.pieceLength
+	}
 	p.piece.Accessed = time.Now().Unix()
 	return
 }
@@ -46,12 +57,31 @@ func (p *DiskPiece) ReadAt(b []byte, off int64) (n int, err error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
-	n, err = p.file.ReadAt(b, off)
+	ff, err := os.OpenFile(p.name, os.O_RDONLY, 0666)
+	if os.IsNotExist(err) {
+		return 0, io.EOF
+	}
+	if err != nil {
+		log.TLogln("Error open file:", err)
+		return 0, err
+	}
+	defer ff.Close()
+
+	n, err = ff.ReadAt(b, off)
 
 	p.piece.Accessed = time.Now().Unix()
+	if int64(len(b))+off >= p.piece.Size {
+		go p.piece.cache.cleanPieces()
+	}
 	return n, nil
 }
 
 func (p *DiskPiece) Release() {
-	p.file.Close()
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.piece.Size = 0
+	p.piece.Complete = false
+
+	p.piece.cache.torrent.Piece(p.piece.Id).SetPriority(torrent.PiecePriorityNone)
 }

server/torr/storage/torrstor/mempiece.go

@@ -29,6 +29,9 @@ func (p *MemPiece) WriteAt(b []byte, off int64) (n int, err error) {
 	}
 	n = copy(p.buffer[off:], b[:])
 	p.piece.Size += int64(n)
+	if p.piece.Size > p.piece.cache.pieceLength {
+		p.piece.Size = p.piece.cache.pieceLength
+	}
 	p.piece.Accessed = time.Now().Unix()
 	return
 }

server/torr/storage/torrstor/reader.go

@@ -135,7 +135,7 @@ func (r *Reader) getPieceNum(offset int64) int {
 
 func (r *Reader) getOffsetRange() (int64, int64) {
 
-	if time.Now().Unix() > r.lastAccess+60 {
+	if time.Now().Unix() > r.lastAccess+60 && len(r.cache.readers) > 1 {
 		return r.file.Offset(), r.file.Offset()
 	}
 

server/torr/torrent.go

@@ -2,21 +2,19 @@ package torr
 
 import (
 	"errors"
-	"fmt"
 	"sort"
 	"sync"
 	"time"
 
+	"github.com/anacrolix/torrent"
+	"github.com/anacrolix/torrent/metainfo"
+
 	"server/log"
 	"server/settings"
 	"server/torr/state"
 	cacheSt "server/torr/storage/state"
 	"server/torr/storage/torrstor"
 	"server/torr/utils"
-	utils2 "server/utils"
-
-	"github.com/anacrolix/torrent"
-	"github.com/anacrolix/torrent/metainfo"
 )
 
 type Torrent struct {
@@ -109,7 +107,6 @@ func (t *Torrent) WaitInfo() bool {
 	case <-t.Torrent.GotInfo():
 		t.cache = t.bt.storage.GetCache(t.Hash())
 		t.cache.SetTorrent(t.Torrent)
-		go t.cache.LoadPiecesOnDisk()
 		return true
 	case <-t.closed:
 		return false
@@ -248,105 +245,6 @@ func (t *Torrent) GetCache() *torrstor.Cache {
 	return t.cache
 }
 
-func (t *Torrent) Preload(index int, size int64) {
-	if size <= 0 {
-		return
-	}
-	t.PreloadSize = size
-
-	if t.Stat == state.TorrentGettingInfo {
-		if !t.WaitInfo() {
-			return
-		}
-		// wait change status
-		time.Sleep(100 * time.Millisecond)
-	}
-
-	t.muTorrent.Lock()
-	if t.Stat != state.TorrentWorking {
-		t.muTorrent.Unlock()
-		return
-	}
-
-	t.Stat = state.TorrentPreload
-	t.muTorrent.Unlock()
-
-	defer func() {
-		if t.Stat == state.TorrentPreload {
-			t.Stat = state.TorrentWorking
-		}
-	}()
-
-	file := t.findFileIndex(index)
-	if file == nil {
-		file = t.Files()[0]
-	}
-
-	if size > file.Length() {
-		size = file.Length()
-	}
-
-	if t.Info() != nil {
-		pieceLength := t.Info().PieceLength
-		mb5 := int64(5 * 1024 * 1024)
-
-		pieceFileStart := int(file.Offset() / pieceLength)
-		pieceFileEnd := int((file.Offset() + file.Length()) / pieceLength)
-		readerPieceBefore := int((file.Offset() + size - mb5) / pieceLength)
-		readerPieceAfter := int((file.Offset() + file.Length() - mb5) / pieceLength)
-
-		lastStat := time.Now().Add(-time.Second)
-
-		for true {
-			t.muTorrent.Lock()
-			if t.Torrent == nil {
-				return
-			}
-
-			t.PreloadedBytes = t.cache.GetState().Filled
-			t.muTorrent.Unlock()
-
-			stat := fmt.Sprint(file.Torrent().InfoHash().HexString(), " ", utils2.Format(float64(t.PreloadedBytes)), "/", utils2.Format(float64(t.PreloadSize)), " Speed:", utils2.Format(t.DownloadSpeed), " Peers:[", t.Torrent.Stats().ConnectedSeeders, "]", t.Torrent.Stats().ActivePeers, "/", t.Torrent.Stats().TotalPeers)
-			if time.Since(lastStat) > time.Second {
-				log.TLogln("Preload:", stat)
-				lastStat = time.Now()
-			}
-
-			isComplete := true
-			if readerPieceBefore >= pieceFileStart {
-				for i := pieceFileStart; i < readerPieceBefore; i++ {
-					if !t.PieceState(i).Complete {
-						isComplete = false
-						if t.PieceState(i).Priority == torrent.PiecePriorityNone {
-							t.Piece(i).SetPriority(torrent.PiecePriorityNormal)
-						}
-					}
-				}
-			}
-			if readerPieceAfter <= pieceFileEnd {
-				for i := readerPieceAfter; i <= pieceFileEnd; i++ {
-					if !t.PieceState(i).Complete {
-						isComplete = false
-						if t.PieceState(i).Priority == torrent.PiecePriorityNone {
-							t.Piece(i).SetPriority(torrent.PiecePriorityNormal)
-						}
-					}
-				}
-			}
-			if t.PreloadedBytes >= size-pieceLength {
-				isComplete = true
-			}
-
-			t.AddExpiredTime(time.Second * time.Duration(settings.BTsets.TorrentDisconnectTimeout))
-			if isComplete {
-				break
-			}
-			time.Sleep(time.Second)
-		}
-	}
-	log.TLogln("End preload:", file.Torrent().InfoHash().HexString(), "Peers:[", t.Torrent.Stats().ConnectedSeeders, "]", t.Torrent.Stats().ActivePeers, "/", t.Torrent.Stats().TotalPeers)
-}
-
 func (t *Torrent) drop() {
 	t.muTorrent.Lock()
 	if t.Torrent != nil {
@@ -427,6 +325,7 @@ func (t *Torrent) Status() *state.TorrentStatus {
 			}
 		}
 	}
+
 	return st
 }
 
@@ -438,23 +337,3 @@ func (t *Torrent) CacheState() *cacheSt.CacheState {
 	}
 	return nil
 }
-
-func (t *Torrent) findFileIndex(index int) *torrent.File {
-	st := t.Status()
-	var stFile *state.TorrentFileStat
-	for _, f := range st.FileStats {
-		if index == f.Id {
-			stFile = f
-			break
-		}
-	}
-	if stFile == nil {
-		return nil
-	}
-	for _, file := range t.Files() {
-		if file.Path() == stFile.Path {
-			return file
-		}
-	}
-	return nil
-}