diff --git a/cmd/zurg/main.go b/cmd/zurg/main.go index a9c7353..3002151 100644 --- a/cmd/zurg/main.go +++ b/cmd/zurg/main.go @@ -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) diff --git a/go.mod b/go.mod index 1056e14..b938258 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 1896074..febb751 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/internal/torrent/manager.go b/internal/torrent/manager.go index 5980a36..9bc1c0d 100644 --- a/internal/torrent/manager.go +++ b/internal/torrent/manager.go @@ -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() {