From a67f0ae4adf8dcee5f069a41f3faeaf5acbac26b Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Mon, 28 Jan 2019 22:01:37 +0100 Subject: [PATCH] les: new serving queue --- les/handler.go | 340 +++++++++++++++----------------------------- les/servingqueue.go | 283 +++++++++++++++++++++--------------- 2 files changed, 283 insertions(+), 340 deletions(-) diff --git a/les/handler.go b/les/handler.go index c61d07b899..801853636f 100644 --- a/les/handler.go +++ b/les/handler.go @@ -335,16 +335,16 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { p.responseCount++ responseCount := p.responseCount var ( - maxCost uint64 - priority int64 + maxCost uint64 + task *servingTask ) - reject := func(reqID, reqCnt, maxCnt uint64) bool { + accept := func(reqID, reqCnt, maxCnt uint64) bool { if reqCnt == 0 { - return true + return false } if p.fcClient == nil || reqCnt > maxCnt { - return true + return false } maxCost = p.fcCosts.getCost(msg.Code, reqCnt) @@ -352,11 +352,11 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if bufShort > 0 { p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/p.fcParams.MinRecharge))) } - return true + return false } else { - priority = servingPriority + task = pm.servingQueue.newTask(servingPriority) } - return false + return task.start() } if msg.Size > ProtocolMaxMsgSize { @@ -366,12 +366,6 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { var deliverMsg *Msg - errorFn := func(err error) { - if err != nil { - p.errCh <- err - } - } - sendReq := func(reqID, amount uint64, reply *reply, servingTime uint64) { p.responseLock.Lock() defer p.responseLock.Unlock() @@ -448,24 +442,21 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } query := req.Query - if reject(req.ReqID, query.Amount, MaxHeaderFetch) { + if !accept(req.ReqID, query.Amount, MaxHeaderFetch) { return errResp(ErrRequestRejected, "") } + go func() { + hashMode := query.Origin.Hash != (common.Hash{}) + first := true + maxNonCanonical := uint64(100) - hashMode := query.Origin.Hash != (common.Hash{}) - first := true - maxNonCanonical := uint64(100) - - // Gather headers until the fetch or network limits is reached - var ( - bytes common.StorageSize - headers []*types.Header - unknown bool - ) - - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { + // Gather headers until the fetch or network limits is reached + var ( + bytes common.StorageSize + headers []*types.Header + unknown bool + ) + for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit { // Retrieve the next header satisfying the query var origin *types.Header if hashMode { @@ -482,7 +473,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number) } if origin == nil { - return true, nil + break } headers = append(headers, origin) bytes += estHeaderRlpSize @@ -533,14 +524,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { // Number based traversal towards the leaf block query.Origin.Number += query.Skip + 1 } - return unknown || len(headers) >= int(query.Amount) || bytes >= softResponseLimit, nil - }, - // after: sendFunc(query.Amount, func(bv uint64) error { return p.SendBlockHeaders(req.ReqID, bv, headers) }), - send: func(servingTime uint64) { - sendReq(req.ReqID, query.Amount, p.ReplyBlockHeaders(req.ReqID, headers), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, query.Amount, p.ReplyBlockHeaders(req.ReqID, headers), task.done()) + }() case BlockHeadersMsg: if pm.downloader == nil { @@ -582,18 +568,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { bodies []rlp.RawValue ) reqCnt := len(req.Hashes) - if reject(req.ReqID, uint64(reqCnt), MaxBodyFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxBodyFetch) { return errResp(ErrRequestRejected, "") } - - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - hash := req.Hashes[index] - index++ + go func() { + for _, hash := range req.Hashes { if bytes >= softResponseLimit { - return true, nil + break } // Retrieve the requested block body, stopping if enough was found if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil { @@ -602,13 +583,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { bytes += len(data) } } - return index == reqCnt, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyBlockBodiesRLP(req.ReqID, bodies), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyBlockBodiesRLP(req.ReqID, bodies), task.done()) + }() case BlockBodiesMsg: if pm.odr == nil { @@ -647,42 +624,33 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { data [][]byte ) reqCnt := len(req.Reqs) - if reject(req.ReqID, uint64(reqCnt), MaxCodeFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxCodeFetch) { return errResp(ErrRequestRejected, "") } - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - req := req.Reqs[index] - index++ - done := index == reqCnt + go func() { + for _, req := range req.Reqs { // Retrieve the requested state entry, stopping if enough was found if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil { if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil { statedb, err := pm.blockchain.State() if err != nil { - return done, nil + continue } account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey)) if err != nil { - return done, nil + continue } code, _ := statedb.Database().TrieDB().Node(common.BytesToHash(account.CodeHash)) data = append(data, code) if bytes += len(code); bytes >= softResponseLimit { - return true, nil + break } } } - return done, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyCode(req.ReqID, data), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyCode(req.ReqID, data), task.done()) + }() case CodeMsg: if pm.odr == nil { @@ -721,19 +689,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { receipts []rlp.RawValue ) reqCnt := len(req.Hashes) - if reject(req.ReqID, uint64(reqCnt), MaxReceiptFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxReceiptFetch) { return errResp(ErrRequestRejected, "") } - - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - hash := req.Hashes[index] - index++ - done := index == reqCnt + go func() { + for _, hash := range req.Hashes { if bytes >= softResponseLimit { - return true, nil + break } // Retrieve the requested block's receipts, skipping if unknown to us var results types.Receipts @@ -742,7 +704,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } if results == nil { if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash { - return done, nil + continue } } // If known, encode and queue for response packet @@ -752,13 +714,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { receipts = append(receipts, encoded) bytes += len(encoded) } - return done, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyReceiptsRLP(req.ReqID, receipts), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyReceiptsRLP(req.ReqID, receipts), task.done()) + }() case ReceiptsMsg: if pm.odr == nil { @@ -797,29 +755,23 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { proofs proofsData ) reqCnt := len(req.Reqs) - if reject(req.ReqID, uint64(reqCnt), MaxProofsFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxProofsFetch) { return errResp(ErrRequestRejected, "") } - - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - req := req.Reqs[index] - index++ - done := index == reqCnt + go func() { + for _, req := range req.Reqs { // Retrieve the requested state entry, stopping if enough was found if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil { if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil { statedb, err := pm.blockchain.State() if err != nil { - return done, nil + continue } var trie state.Trie if len(req.AccKey) > 0 { account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey)) if err != nil { - return done, nil + continue } trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root) } else { @@ -831,18 +783,14 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { proofs = append(proofs, proof) if bytes += proof.DataSize(); bytes >= softResponseLimit { - return true, nil + break } } } } - return done, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyProofs(req.ReqID, proofs), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyProofs(req.ReqID, proofs), task.done()) + }() case GetProofsV2Msg: p.Log().Trace("Received les/2 proofs request") @@ -861,19 +809,14 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { root common.Hash ) reqCnt := len(req.Reqs) - if reject(req.ReqID, uint64(reqCnt), MaxProofsFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxProofsFetch) { return errResp(ErrRequestRejected, "") } + go func() { - nodes := light.NewNodeSet() + nodes := light.NewNodeSet() - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - req := req.Reqs[index] - index++ - done := index == reqCnt + for _, req := range req.Reqs { // Look up the state belonging to the request if statedb == nil || req.BHash != lastBHash { statedb, root, lastBHash = nil, common.Hash{}, req.BHash @@ -886,34 +829,30 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } if statedb == nil { - return done, nil + continue } // Pull the account or storage trie of the request var trie state.Trie if len(req.AccKey) > 0 { account, err := pm.getAccount(statedb, root, common.BytesToHash(req.AccKey)) if err != nil { - return done, nil + continue } trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root) } else { trie, _ = statedb.Database().OpenTrie(root) } if trie == nil { - return done, nil + continue } // Prove the user's request from the account or stroage trie trie.Prove(req.Key, req.FromLevel, nodes) if nodes.DataSize() >= softResponseLimit { - return true, nil + break } - return done, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyProofsV2(req.ReqID, nodes.NodeList()), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyProofsV2(req.ReqID, nodes.NodeList()), task.done()) + }() case ProofsV1Msg: if pm.odr == nil { @@ -973,24 +912,18 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { proofs []ChtResp ) reqCnt := len(req.Reqs) - if reject(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) { return errResp(ErrRequestRejected, "") } - trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix)) - - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - req := req.Reqs[index] - index++ - done := index == reqCnt + go func() { + trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix)) + for _, req := range req.Reqs { if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil { sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*pm.iConfig.ChtSize-1) if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) { trie, err := trie.New(root, trieDb) if err != nil { - return done, nil + continue } var encNumber [8]byte binary.BigEndian.PutUint64(encNumber[:], req.BlockNum) @@ -1000,17 +933,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { proofs = append(proofs, ChtResp{Header: header, Proof: proof}) if bytes += proof.DataSize() + estHeaderRlpSize; bytes >= softResponseLimit { - return true, nil + break } } } - return done, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyHeaderProofs(req.ReqID, proofs), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyHeaderProofs(req.ReqID, proofs), task.done()) + }() case GetHelperTrieProofsMsg: p.Log().Trace("Received helper trie proof request") @@ -1028,24 +957,19 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { auxData [][]byte ) reqCnt := len(req.Reqs) - if reject(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) { + if !accept(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) { return errResp(ErrRequestRejected, "") } + go func() { - var ( - lastIdx uint64 - lastType uint - root common.Hash - auxTrie *trie.Trie - ) - nodes := light.NewNodeSet() - - index := 0 - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - req := req.Reqs[index] - index++ + var ( + lastIdx uint64 + lastType uint + root common.Hash + auxTrie *trie.Trie + ) + nodes := light.NewNodeSet() + for _, req := range req.Reqs { if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx { auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx @@ -1072,15 +996,11 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } if nodes.DataSize()+auxBytes >= softResponseLimit { - return true, nil + break } - return index == reqCnt, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyHelperTrieProofs(req.ReqID, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData}), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyHelperTrieProofs(req.ReqID, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData}), task.done()) + }() case HeaderProofsMsg: if pm.odr == nil { @@ -1133,21 +1053,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } reqCnt := len(txs) - if reject(0, uint64(reqCnt), MaxTxSend) { + if !accept(0, uint64(reqCnt), MaxTxSend) { return errResp(ErrRequestRejected, "") } - - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - pm.txpool.AddRemotes(txs) - return true, nil - }, - send: func(servingTime uint64) { - sendReq(0, uint64(reqCnt), nil, servingTime) - }, - fail: errorFn, - }) + go func() { + pm.txpool.AddRemotes(txs) + sendReq(0, uint64(reqCnt), nil, task.done()) + }() case SendTxV2Msg: if pm.txpool == nil { @@ -1162,35 +1074,26 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } reqCnt := len(req.Txs) - if reject(req.ReqID, uint64(reqCnt), MaxTxSend) { + if !accept(req.ReqID, uint64(reqCnt), MaxTxSend) { return errResp(ErrRequestRejected, "") } - - var stats []txStatus - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - hashes := make([]common.Hash, len(req.Txs)) - for i, tx := range req.Txs { - hashes[i] = tx.Hash() - } - stats = pm.txStatus(hashes) - for i, stat := range stats { - if stat.Status == core.TxStatusUnknown { - if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil { - stats[i].Error = errs[0].Error() - continue - } - stats[i] = pm.txStatus([]common.Hash{hashes[i]})[0] + go func() { + hashes := make([]common.Hash, len(req.Txs)) + for i, tx := range req.Txs { + hashes[i] = tx.Hash() + } + stats := pm.txStatus(hashes) + for i, stat := range stats { + if stat.Status == core.TxStatusUnknown { + if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil { + stats[i].Error = errs[0].Error() + continue } + stats[i] = pm.txStatus([]common.Hash{hashes[i]})[0] } - return true, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), servingTime) - }, - fail: errorFn, - }) + } + sendReq(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), task.done()) + }() case GetTxStatusMsg: if pm.txpool == nil { @@ -1205,22 +1108,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } reqCnt := len(req.Hashes) - if reject(req.ReqID, uint64(reqCnt), MaxTxStatus) { + if !accept(req.ReqID, uint64(reqCnt), MaxTxStatus) { return errResp(ErrRequestRejected, "") } - - var stats []txStatus - pm.servingQueue.addTask(&servingTask{ - priority: priority, - run: func() (bool, error) { - stats = pm.txStatus(req.Hashes) - return true, nil - }, - send: func(servingTime uint64) { - sendReq(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), servingTime) - }, - fail: errorFn, - }) + go func() { + stats := pm.txStatus(req.Hashes) + sendReq(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), task.done()) + }() case TxStatusMsg: if pm.odr == nil { diff --git a/les/servingqueue.go b/les/servingqueue.go index a99b452824..6b4b277aa9 100644 --- a/les/servingqueue.go +++ b/les/servingqueue.go @@ -26,13 +26,16 @@ import ( // servingQueue runs serving tasks in a limited number of threads and puts the // waiting tasks in a priority queue type servingQueue struct { - lock sync.Mutex - threadCount int // number of currently running threads - stopCount int // number of threads to be stopped after they finish their current task - queue *prque.Prque // priority queue for waiting or suspended tasks - best *servingTask // either best == nil (queue empty) or waitingForTask is empty - waiting []chan *servingTask // threads waiting for a task - suspendBias int64 // priority bias against suspending an already running task + tokenCh chan runToken + queueAddCh, queueBestCh chan *servingTask + stopThreadCh, quit chan struct{} + setThreadsCh chan int + + wg sync.WaitGroup + threadCount int // number of currently running threads + queue *prque.Prque // priority queue for waiting or suspended tasks + best *servingTask // either best == nil (queue empty) or waitingForTask is empty + suspendBias int64 // priority bias against suspending an already running task } // servingTask represents a request serving task. Tasks can be implemented to @@ -44,145 +47,191 @@ type servingQueue struct { // - run: execute a single step; return true if finished // - after: executed after run finishes or returns an error, receives the total serving time type servingTask struct { + sq *servingQueue servingTime uint64 - done bool - err error priority int64 - run func() (finished bool, err error) - send func(servingTime uint64) - fail func(err error) + biasAdded bool + token runToken + tokenCh chan runToken +} + +type runToken chan struct{} + +func (t *servingTask) start() bool { + select { + case t.token = <-t.sq.tokenCh: + default: + t.tokenCh = make(chan runToken, 1) + select { + case t.sq.queueAddCh <- t: + case <-t.sq.quit: + return false + } + select { + case t.token = <-t.tokenCh: + case <-t.sq.quit: + return false + } + } + if t.token == nil { + return false + } + t.servingTime -= uint64(mclock.Now()) + return true +} + +func (t *servingTask) done() uint64 { + t.servingTime += uint64(mclock.Now()) + close(t.token) + return t.servingTime +} + +func (t *servingTask) waitOrStop() bool { + t.done() + if !t.biasAdded { + t.priority += t.sq.suspendBias + t.biasAdded = true + } + return t.start() } // newServingQueue returns a new servingQueue -func newServingQueue(_suspendBias int64) *servingQueue { - return &servingQueue{ - queue: prque.New(nil), - suspendBias: _suspendBias, +func newServingQueue(suspendBias int64) *servingQueue { + sq := &servingQueue{ + queue: prque.New(nil), + suspendBias: suspendBias, + tokenCh: make(chan runToken), + queueAddCh: make(chan *servingTask, 100), + queueBestCh: make(chan *servingTask), + stopThreadCh: make(chan struct{}), + quit: make(chan struct{}), + setThreadsCh: make(chan int, 10), + } + sq.wg.Add(2) + go sq.queueLoop() + go sq.threadCountLoop() + return sq +} + +func (sq *servingQueue) newTask(priority int64) *servingTask { + return &servingTask{ + sq: sq, + priority: priority, } } -// addTask adds a new task, either starting it immediately or queueing it -func (sq *servingQueue) addTask(task *servingTask) { - sq.lock.Lock() - defer sq.lock.Unlock() - - if l := len(sq.waiting); l != 0 { - l-- - sq.waiting[l] <- task - sq.waiting = sq.waiting[:l] - return +func (sq *servingQueue) threadController() { + for { + token := make(runToken) + select { + case best := <-sq.queueBestCh: + best.tokenCh <- token + default: + select { + case best := <-sq.queueBestCh: + best.tokenCh <- token + case sq.tokenCh <- token: + case <-sq.stopThreadCh: + sq.wg.Done() + return + case <-sq.quit: + sq.wg.Done() + return + } + } + <-token + select { + case <-sq.stopThreadCh: + sq.wg.Done() + return + case <-sq.quit: + sq.wg.Done() + return + default: + } } +} +func (sq *servingQueue) addTask(task *servingTask) { if sq.best == nil { sq.best = task - return - } - if task.priority < sq.best.priority { + } else if task.priority > sq.best.priority { sq.queue.Push(sq.best, sq.best.priority) sq.best = task return + } else { + sq.queue.Push(task, task.priority) } - sq.queue.Push(task, task.priority) } -// getNewTask selects a new task to be processed. If blocking == true then it waits -// until a runnable task arrives or returns nil if the thread should be stopped. -// if currentTask != nil then it returns immediately and only returns a new task -// if the current one should be suspended. -// Note: either blocking should be false or currentTask should be nil. -func (sq *servingQueue) getNewTask(currentTask *servingTask, blocking bool) *servingTask { - sq.lock.Lock() - if sq.stopCount == 0 { - if sq.best != nil && (currentTask == nil || sq.best.priority <= currentTask.priority-sq.suspendBias) { - best := sq.best - if sq.queue.Size() == 0 { - sq.best = nil - } else { - sq.best, _ = sq.queue.PopItem().(*servingTask) +func (sq *servingQueue) queueLoop() { + for { + if sq.best != nil { + select { + case task := <-sq.queueAddCh: + sq.addTask(task) + case sq.queueBestCh <- sq.best: + if sq.queue.Size() == 0 { + sq.best = nil + } else { + sq.best, _ = sq.queue.PopItem().(*servingTask) + } + case <-sq.quit: + sq.wg.Done() + return + } + } else { + select { + case task := <-sq.queueAddCh: + sq.addTask(task) + case <-sq.quit: + sq.wg.Done() + return + } + } + } +} + +func (sq *servingQueue) threadCountLoop() { + var threadCountTarget int + for { + for threadCountTarget > sq.threadCount { + sq.wg.Add(1) + go sq.threadController() + sq.threadCount++ + } + if threadCountTarget < sq.threadCount { + select { + case threadCountTarget = <-sq.setThreadsCh: + case sq.stopThreadCh <- struct{}{}: + sq.threadCount-- + case <-sq.quit: + sq.wg.Done() + return + } + } else { + select { + case threadCountTarget = <-sq.setThreadsCh: + case <-sq.quit: + sq.wg.Done() + return } - sq.lock.Unlock() - return best } - if blocking { - ch := make(chan *servingTask) - sq.waiting = append(sq.waiting, ch) - sq.lock.Unlock() - return <-ch - } - } else { - sq.stopCount-- - sq.threadCount-- } - sq.lock.Unlock() - return nil } // setThreads sets the processing thread count, suspending tasks as soon as // possible if necessary. func (sq *servingQueue) setThreads(threadCount int) { - sq.lock.Lock() - defer sq.lock.Unlock() - - diff := threadCount - sq.threadCount + sq.stopCount - if diff > 0 { - // start more threads - if sq.stopCount >= diff { - sq.stopCount -= diff - } else { - diff -= sq.stopCount - sq.stopCount = 0 - sq.threadCount += diff - for ; diff > 0; diff-- { - go sq.servingThread() - } - } - } - if diff < 0 { - // stop some threads - lw := len(sq.waiting) - sq.stopCount -= diff - for diff < 0 && lw > 0 { - diff++ - lw-- - sq.waiting[lw] <- nil - sq.stopCount-- - sq.threadCount-- - } - sq.waiting = sq.waiting[:lw] + select { + case sq.setThreadsCh <- threadCount: + case <-sq.quit: + return } } // stop stops task processing as soon as possible func (sq *servingQueue) stop() { - sq.setThreads(0) -} - -// servingThread implements a single serving thread -func (sq *servingQueue) servingThread() { - for { - task := sq.getNewTask(nil, true) - if task == nil { - return - } - task.servingTime -= uint64(mclock.Now()) - for { - task.done, task.err = task.run() - if task.done || task.err != nil { - task.servingTime += uint64(mclock.Now()) - if task.err == nil { - task.send(task.servingTime) - } else { - task.fail(task.err) - } - break - } - if newTask := sq.getNewTask(task, false); newTask != nil { - now := uint64(mclock.Now()) - task.servingTime += now - sq.addTask(task) - task = newTask - task.servingTime -= now - } - } - } + close(sq.quit) + sq.wg.Wait() }