MirrorHub/dendrite
0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-11-11 20:31:08 +01:00
Code Issues Releases Wiki Activity

Workers

Browse source
This commit is contained in:
Neil Alexander 2021-01-29 14:31:28 +00:00
parent 972c848730
commit 92ec1b2845
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
1 changed files with 5 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
syncapi/streams/stream_pdu.go
View file

@ -11,8 +11,8 @@ import (
"go.uber.org/atomic" "go.uber.org/atomic"
) )
const COMPLETE_SYNC_QUEUE = 2048 const PDU_STREAM_QUEUESIZE = 2048
const COMPLETE_SYNC_WORKERS = 256 const PDU_STREAM_WORKERS = 256
type PDUStreamProvider struct { type PDUStreamProvider struct {
StreamProvider StreamProvider
@ -34,17 +34,16 @@ func (p *PDUStreamProvider) worker() {
} }
func (p *PDUStreamProvider) queue(f func()) { func (p *PDUStreamProvider) queue(f func()) {
p.tasks <- f if p.workers.Load() < PDU_STREAM_WORKERS {
if p.workers.Load() < COMPLETE_SYNC_WORKERS {
p.workers.Inc() p.workers.Inc()
go p.worker() go p.worker()
} }
p.tasks <- f
} }
func (p *PDUStreamProvider) Setup() { func (p *PDUStreamProvider) Setup() {
p.StreamProvider.Setup() p.StreamProvider.Setup()
p.tasks = make(chan func(), PDU_STREAM_QUEUESIZE)
p.tasks = make(chan func(), COMPLETE_SYNC_QUEUE)
p.latestMutex.Lock() p.latestMutex.Lock()
defer p.latestMutex.Unlock() defer p.latestMutex.Unlock()