hpz937/zurg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Use ants pool instead

Browse Source
This commit is contained in:
Ben Sarmiento
2023-11-18 20:50:14 +01:00
parent 44ec4a0b00
commit 6ef5b147c0
4 changed files with 44 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
cmd/zurg/main.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/debridmediamanager.com/zurg/internal/net"
"github.com/debridmediamanager.com/zurg/internal/torrent"
"github.com/debridmediamanager.com/zurg/internal/version"
"github.com/panjf2000/ants/v2"
// zurghttp "github.com/debridmediamanager.com/zurg/pkg/http"
"github.com/debridmediamanager.com/zurg/pkg/logutil"
@@ -44,7 +45,13 @@ func main() {
rd := realdebrid.NewRealDebrid(config.GetToken(), logutil.NewLogger().Named("realdebrid"))
torrentMgr := torrent.NewTorrentManager(config, rd)
p, err := ants.NewPool(config.GetNumOfWorkers())
if err != nil {
log.Panicf("Failed to create ants pool: %v", err)
}
defer p.Release()
torrentMgr := torrent.NewTorrentManager(config, rd, p)
mux := http.NewServeMux()
net.Router(mux, config, torrentMgr, cache)

2
go.mod
View File

@@ -11,6 +11,8 @@ require (
require github.com/orcaman/concurrent-map/v2 v2.0.1
require github.com/panjf2000/ants/v2 v2.8.2
require (
github.com/winfsp/cgofuse v1.5.0
go.uber.org/multierr v1.10.0 // indirect

15
go.sum
View File

@@ -1,13 +1,21 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/orcaman/concurrent-map/v2 v2.0.1 h1:jOJ5Pg2w1oeB6PeDurIYf6k9PQ+aTITr/6lP/L/zp6c=
github.com/orcaman/concurrent-map/v2 v2.0.1/go.mod h1:9Eq3TG2oBe5FirmYWQfYO5iH1q0Jv47PLaNK++uCdOM=
github.com/panjf2000/ants/v2 v2.8.2 h1:D1wfANttg8uXhC9149gRt1PDQ+dLVFjNXkCEycMcvQQ=
github.com/panjf2000/ants/v2 v2.8.2/go.mod h1:7ZxyxsqE4vvW0M7LSD8aI3cKwgFhBHbxnlN8mDqHa1I=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/winfsp/cgofuse v1.5.0 h1:MsBP7Mi/LiJf/7/F3O/7HjjR009ds6KCdqXzKpZSWxI=
github.com/winfsp/cgofuse v1.5.0/go.mod h1:h3awhoUOcn2VYVKCwDaYxSLlZwnyK+A8KaDoLUp2lbU=
go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk=
@@ -16,9 +24,12 @@ go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=
go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so=
golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q=
golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

41
internal/torrent/manager.go
View File

@@ -16,6 +16,7 @@ import (
"github.com/debridmediamanager.com/zurg/pkg/logutil"
"github.com/debridmediamanager.com/zurg/pkg/realdebrid"
cmap "github.com/orcaman/concurrent-map/v2"
"github.com/panjf2000/ants/v2"
"go.uber.org/zap"
)
@@ -25,20 +26,20 @@ type TorrentManager struct {
requiredVersion string
checksum string
api *realdebrid.RealDebrid
workerPool chan bool
antsPool *ants.Pool
log *zap.SugaredLogger
}
// NewTorrentManager creates a new torrent manager
// it will fetch all torrents and their info in the background
// and store them in-memory and cached in files
func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid) *TorrentManager {
func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, p *ants.Pool) *TorrentManager {
t := &TorrentManager{
cfg: cfg,
DirectoryMap: cmap.New[cmap.ConcurrentMap[string, *Torrent]](),
requiredVersion: "10.11.2023",
requiredVersion: "18.11.2023",
api: api,
workerPool: make(chan bool, cfg.GetNumOfWorkers()),
antsPool: p,
log: logutil.NewLogger().Named("manager"),
}
@@ -59,11 +60,10 @@ func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid) *
for i := range newTorrents {
wg.Add(1)
go func(idx int) {
defer wg.Done()
t.workerPool <- true
// TODO wrap getMoreInfo and limit the execution time!
torrentsChan <- t.getMoreInfo(newTorrents[idx])
<-t.workerPool
_ = t.antsPool.Submit(func() {
defer wg.Done()
torrentsChan <- t.getMoreInfo(newTorrents[idx])
})
}(i)
}
wg.Wait()
@@ -164,9 +164,7 @@ func (t *TorrentManager) mergeToMain(t1, t2 *Torrent) *Torrent {
// proxy
func (t *TorrentManager) UnrestrictUntilOk(link string) *realdebrid.UnrestrictResponse {
t.workerPool <- true
ret := t.api.UnrestrictUntilOk(link)
<-t.workerPool
return ret
}
@@ -248,10 +246,10 @@ func (t *TorrentManager) startRefreshJob() {
for i := range newTorrents {
wg.Add(1)
go func(idx int) {
defer wg.Done()
t.workerPool <- true
torrentsChan <- t.getMoreInfo(newTorrents[idx])
<-t.workerPool
_ = t.antsPool.Submit(func() {
defer wg.Done()
torrentsChan <- t.getMoreInfo(newTorrents[idx])
})
}(i)
}
wg.Wait()
@@ -484,13 +482,16 @@ func (t *TorrentManager) organizeChaos(links []string, selectedFiles []*File) ([
for _, link := range links {
wg.Add(1)
go func(lnk string) {
link := link // redeclare to avoid closure on loop variable
// Use the existing workerPool to submit tasks
err := t.antsPool.Submit(func() {
defer wg.Done()
t.workerPool <- true
resp := t.api.UnrestrictUntilOk(lnk)
<-t.workerPool
resp := t.api.UnrestrictUntilOk(link)
resultsChan <- Result{Response: resp}
}(link)
})
if err != nil {
t.log.Errorf("Error submitting task to workerPool: %v", err)
}
}
go func() {