Merge branch 'dev' of github.com:CortexFoundation/CortexTheseus into dev

Author: ucwong

inference/synapse/local_infer.go

@@ -75,8 +75,8 @@ func (s *Synapse) inferByInfoHash(modelInfoHash, inputInfoHash string) (res []by
 
 	cacheKey := RLPHashString(modelHash + "_" + inputHash)
 
-	if _, ok := CvmFixHashes[cacheKey]; ok {
-		return CvmFixHashes[cacheKey], nil
+	if hash, ok := CvmFixHashes[cacheKey]; ok {
+		return hash, nil
 	}
 
 	if v, ok := s.simpleCache.Load(cacheKey); ok && !s.config.IsNotCache {
@@ -191,8 +191,8 @@ func (s *Synapse) Available(infoHash string, rawSize int64) error {
 	if s.config.IsRemoteInfer {
 		errRes := s.remoteAvailable(
 			infoHash,
-			rawSize,
-			s.config.InferURI)
+			rawSize)
+			//s.config.InferURI)
 		return errRes
 	}
 	if len(infoHash) < 2 || !strings.HasPrefix(infoHash, "0x") {

inference/synapse/remote_infer.go

@@ -33,7 +33,8 @@ func (s *Synapse) remoteGasByModelHash(modelInfoHash string) (uint64, error) {
 	return binary.BigEndian.Uint64(retArray), nil
 }
 
-func (s *Synapse) remoteAvailable(infoHash string, rawSize int64, uri string) error {
+//func (s *Synapse) remoteAvailable(infoHash string, rawSize int64, uri string) error {
+func (s *Synapse) remoteAvailable(infoHash string, rawSize int64) error {
 	inferWork := &inference.AvailableWork{
 		Type:     inference.AVAILABLE_BY_H,
 		InfoHash: infoHash,
@@ -47,7 +48,7 @@ func (s *Synapse) remoteAvailable(infoHash string, rawSize int64, uri string) er
 	}
 	log.Debug("remoteAvailable", "request", string(requestBody))
 
-	_, err := s.sendRequest(string(requestBody), uri)
+	_, err := s.sendRequest(string(requestBody), s.config.InferURI)
 	return err
 }
 

torrentfs/monitor.go

@@ -39,7 +39,7 @@ var (
 )
 
 const (
-	defaultTimerInterval  = 3
+	defaultTimerInterval  = 2
 	connTryTimes          = 300
 	connTryInterval       = 2
 	fetchBlockTryTimes    = 5
@@ -733,19 +733,13 @@ func (m *Monitor) listenLatestBlock() {
 
 func (m *Monitor) listenPeers() {
 	defer m.wg.Done()
-	timer := time.NewTimer(time.Second * 15)
+	timer := time.NewTimer(time.Second * 300)
 
 	for {
 		select {
 		case <-timer.C:
 			m.peers()
-			if healthPeers.Len() == 0 {
-				timer.Reset(time.Second * 5)
-			} else if healthPeers.Len() < 6 {
-				timer.Reset(time.Second * 60)
-			} else {
-				timer.Reset(time.Second * 300)
-			}
+			timer.Reset(time.Second * 300)
 		case <-m.exitCh:
 			log.Info("Peers listener stopped")
 			return
@@ -833,7 +827,7 @@ func (m *Monitor) batch_udp_healthy(ip string, ports []string) ([]string, bool)
 }
 
 const (
-	batch = 2048
+	batch = 4096
 )
 
 func (m *Monitor) syncLastBlock() uint64 {

torrentfs/storage.go

@@ -91,7 +91,7 @@ func NewFileStorage(config *Config) (*FileStorage, error) {
 	if dbErr != nil {
 		return nil, dbErr
 	}
-	//db.NoSync = true
+	db.NoSync = true
 
 	fs := &FileStorage{
 		// filesInfoHash:     make(map[metainfo.Hash]*FileInfo),

torrentfs/torrentClient.go

@@ -5,7 +5,7 @@ import (
 	"crypto/sha1"
 	//"errors"
 	"fmt"
-	"github.com/anacrolix/missinggo/slices"
+	//"github.com/anacrolix/missinggo/slices"
 	"github.com/bradfitz/iter"
 	"github.com/edsrzf/mmap-go"
 	"io"
@@ -114,9 +114,9 @@ func (t *Torrent) ReloadFile(files []string, datas [][]byte, tm *TorrentManager)
 	if err != nil {
 		return
 	}
-	t.Torrent = torrent
 	<-torrent.GotInfo()
 	torrent.VerifyData()
+	t.Torrent = torrent
 	t.Pause()
 }
 
@@ -148,9 +148,9 @@ func (t *Torrent) ReloadTorrent(data []byte, tm *TorrentManager) {
 	if err != nil {
 		return
 	}
-	t.Torrent = torrent
 	<-torrent.GotInfo()
 	torrent.VerifyData()
+	t.Torrent = torrent
 	t.Pause()
 }
 
@@ -354,8 +354,8 @@ func (tm *TorrentManager) Close() error {
 }
 
 func (tm *TorrentManager) dropAll() {
-	tm.lock.Lock()
-	tm.lock.Unlock()
+	//tm.lock.Lock()
+	//tm.lock.Unlock()
 	defer tm.client.Close()
 	for _, t := range tm.torrents {
 		stats := t.Stats()
@@ -495,13 +495,13 @@ func (tm *TorrentManager) AddTorrent(filePath string, BytesRequested int64) *Tor
 		if err != nil {
 			return nil
 		}
-		var ss []string
-		slices.MakeInto(&ss, mi.Nodes)
-		tm.client.AddDHTNodes(ss)
-		torrent := tm.CreateTorrent(t, BytesRequested, torrentPending, ih)
+		//var ss []string
+		//slices.MakeInto(&ss, mi.Nodes)
+		//tm.client.AddDHTNodes(ss)
 		<-t.GotInfo()
 		t.VerifyData()
-		torrent.SeedInQueue()
+		torrent := tm.CreateTorrent(t, BytesRequested, torrentPending, ih)
+		torrent.Pause() //SeedInQueue()
 		return torrent
 	} else {
 		spec.Storage = storage.NewFile(TmpDir)
@@ -512,12 +512,12 @@ func (tm *TorrentManager) AddTorrent(filePath string, BytesRequested int64) *Tor
 		if err != nil {
 			return nil
 		}
-		var ss []string
-		slices.MakeInto(&ss, mi.Nodes)
-		tm.client.AddDHTNodes(ss)
-		torrent := tm.CreateTorrent(t, BytesRequested, torrentPending, ih)
+		//var ss []string
+		//slices.MakeInto(&ss, mi.Nodes)
+		//tm.client.AddDHTNodes(ss)
 		<-t.GotInfo()
 		t.VerifyData()
+		torrent := tm.CreateTorrent(t, BytesRequested, torrentPending, ih)
 		torrent.Pause()
 		return torrent
 	}
@@ -825,16 +825,18 @@ func (tm *TorrentManager) listenTorrentProgress() {
 				t.bytesMissing = t.BytesMissing()
 
 				if t.Finished() {
-					os.Symlink(
+					err := os.Symlink(
 						path.Join(defaultTmpFilePath, t.InfoHash()),
 						path.Join(tm.DataDir, t.InfoHash()),
 					)
-					delete(tm.activeTorrents, ih)
-					tm.seedingTorrents[ih] = t
-					t.Seed()
-					t.loop = defaultSeedInterval / queryTimeInterval
-					total_size += uint64(t.bytesCompleted)
-					current_size += uint64(t.bytesCompleted)
+					if err == nil {
+						delete(tm.activeTorrents, ih)
+						tm.seedingTorrents[ih] = t
+						t.Seed()
+						t.loop = defaultSeedInterval / queryTimeInterval
+						total_size += uint64(t.bytesCompleted)
+						current_size += uint64(t.bytesCompleted)
+					}
 					continue
 				}
 
@@ -926,6 +928,7 @@ func (tm *TorrentManager) listenTorrentProgress() {
 			}
 			timer.Reset(time.Second * queryTimeInterval)
 		case <-tm.closeAll:
+			log.Info("Listen torrent progress closed")
 			return
 		}
 	}