mirrors/owncast
1
0
Fork 0
mirror of https://github.com/owncast/owncast.git synced 2024-11-28 11:09:01 +03:00
Code Issues Projects Releases Packages Wiki Activity

fix: limit the different worker pools to available CPU cores. Should resolve #3189

Browse source
This commit is contained in:
Gabe Kangas 2023-07-18 20:26:44 -07:00
parent b1381170c1
commit d0376cdc75
No known key found for this signature in database
GPG key ID: 4345B2060657F330
4 changed files with 14 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
activitypub/inbox/workerpool.go
View file

@ -1,14 +1,14 @@
package inbox package inbox
import ( import (
"runtime"
"github.com/owncast/owncast/activitypub/apmodels" "github.com/owncast/owncast/activitypub/apmodels"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
const ( // workerPoolSize defines the number of concurrent ActivityPub handlers.
// InboxWorkerPoolSize defines the number of concurrent ActivityPub handlers. var workerPoolSize = runtime.GOMAXPROCS(0)
InboxWorkerPoolSize = 10
)
// Job struct bundling the ActivityPub and the payload in one struct. // Job struct bundling the ActivityPub and the payload in one struct.
type Job struct { type Job struct {
@ -22,7 +22,7 @@ func InitInboxWorkerPool() {
queue = make(chan Job) queue = make(chan Job)
// start workers // start workers
for i := 1; i <= InboxWorkerPoolSize; i++ { for i := 1; i <= workerPoolSize; i++ {
go worker(i, queue) go worker(i, queue)
} }
} }

9
activitypub/workerpool/outbound.go
View file

@ -2,14 +2,13 @@ package workerpool
import ( import (
"net/http" "net/http"
"runtime"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
const ( // workerPoolSize defines the number of concurrent HTTP ActivityPub requests.
// ActivityPubWorkerPoolSize defines the number of concurrent HTTP ActivityPub requests. var workerPoolSize = runtime.GOMAXPROCS(0)
ActivityPubWorkerPoolSize = 10
)
// Job struct bundling the ActivityPub and the payload in one struct. // Job struct bundling the ActivityPub and the payload in one struct.
type Job struct { type Job struct {
@ -23,7 +22,7 @@ func InitOutboundWorkerPool() {
queue = make(chan Job) queue = make(chan Job)
// start workers // start workers
for i := 1; i <= ActivityPubWorkerPoolSize; i++ { for i := 1; i <= workerPoolSize; i++ {
go worker(i, queue) go worker(i, queue)
} }
} }

4
core/webhooks/webhooks_test.go
View file

@ -51,7 +51,7 @@ func TestMain(m *testing.M) {
// this test ensures that `SendToWebhooks` without a `WaitGroup` doesn't panic. // this test ensures that `SendToWebhooks` without a `WaitGroup` doesn't panic.
func TestPublicSend(t *testing.T) { func TestPublicSend(t *testing.T) {
// Send enough events to be sure at least one worker delivers a second event. // Send enough events to be sure at least one worker delivers a second event.
const eventsCount = webhookWorkerPoolSize + 1 eventsCount := webhookWorkerPoolSize + 1
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(eventsCount) wg.Add(eventsCount)
@ -267,7 +267,7 @@ func TestParallel(t *testing.T) {
myId := atomic.AddUint32(&calls, 1) myId := atomic.AddUint32(&calls, 1)
// We made it to the pool size + 1 event, so we're done with the test. // We made it to the pool size + 1 event, so we're done with the test.
if myId == webhookWorkerPoolSize+1 { if myId == uint32(webhookWorkerPoolSize)+1 {
close(finished) close(finished)
return return
} }

7
core/webhooks/workerpool.go
View file

@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"encoding/json" "encoding/json"
"net/http" "net/http"
"runtime"
"sync" "sync"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
@ -12,10 +13,8 @@ import (
"github.com/owncast/owncast/models" "github.com/owncast/owncast/models"
) )
const ( // webhookWorkerPoolSize defines the number of concurrent HTTP webhook requests.
// webhookWorkerPoolSize defines the number of concurrent HTTP webhook requests. var webhookWorkerPoolSize = runtime.GOMAXPROCS(0)
webhookWorkerPoolSize = 10
)
// Job struct bundling the webhook and the payload in one struct. // Job struct bundling the webhook and the payload in one struct.
type Job struct { type Job struct {