我编写了一个简单的并发调度程序，但它似乎在高级并发上存在性能问题。
这是代码（调度程序+并发限速器测试）：
package main
import (
"flag"
"fmt"
"log"
"os"
"runtime"
"runtime/pprof"
"sync"
"time"
"github.com/gomodule/redigo/redis"
)
// a scheduler is composed by load function and process function
type Scheduler struct {
// query channel
reqChan chan interface{}
// max routine
maxRoutine int
// max routine
chanSize int
wg sync.WaitGroup
// query process function
process func(interface{})
}
func NewScheduler(maxRoutine int, chanSize int, process func(interface{})) *Scheduler {
s := &Scheduler{}
if maxRoutine == 0 {
s.maxRoutine = 10
} else {
s.maxRoutine = maxRoutine
}
if chanSize == 0 {
s.chanSize = 100
} else {
s.chanSize = chanSize
}
s.reqChan = make(chan interface{}, s.chanSize)
s.process = process
return s
}
func (s *Scheduler) Start() {
// start process
for i := 0; i < s.maxRoutine; i++ {
go s.processRequest()
}
}
func (s *Scheduler) processRequest() {
for {
select {
case req := <-s.reqChan:
s.process(req)
s.wg.Done()
}
}
}
func (s *Scheduler) Enqueue(req interface{}) {
select {
case s.reqChan <- req:
s.wg.Add(1)
}
}
func (s *Scheduler) Wait() {
s.wg.Wait()
}
const script = `
local required_permits = tonumber(ARGV[2]);
local next_free_micros = redis.call('hget',KEYS[1],'next_free_micros');
if(next_free_micros == false) then
next_free_micros = 0;
else
next_free_micros = tonumber(next_free_micros);
end;
local time = redis.call('time');
local now_micros = tonumber(time[1])*1000000 + tonumber(time[2]);
--[[
try aquire
--]]
if(ARGV[3] ~= nil) then
local micros_to_wait = next_free_micros - now_micros;
if(micros_to_wait > tonumber(ARGV[3])) then
return micros_to_wait;
end
end
local stored_permits = redis.call('hget',KEYS[1],'stored_permits');
if(stored_permits == false) then
stored_permits = 0;
else
stored_permits = tonumber(stored_permits);
end
local stable_interval_micros = 1000000/tonumber(ARGV[1]);
local max_stored_permits = tonumber(ARGV[1]);
if(now_micros > next_free_micros) then
local new_stored_permits = stored_permits + (now_micros - next_free_micros) / stable_interval_micros;
if(max_stored_permits < new_stored_permits) then
stored_permits = max_stored_permits;
else
stored_permits = new_stored_permits;
end
next_free_micros = now_micros;
end
local moment_available = next_free_micros;
local stored_permits_to_spend = 0;
if(stored_permits < required_permits) then
stored_permits_to_spend = stored_permits;
else
stored_permits_to_spend = required_permits;
end
local fresh_permits = required_permits - stored_permits_to_spend;
local wait_micros = fresh_permits * stable_interval_micros;
redis.replicate_commands();
redis.call('hset',KEYS[1],'stored_permits',stored_permits - stored_permits_to_spend);
redis.call('hset',KEYS[1],'next_free_micros',next_free_micros + wait_micros);
redis.call('expire',KEYS[1],10);
return moment_available - now_micros;
`
var (
rlScript *redis.Script
)
func init() {
rlScript = redis.NewScript(1, script)
}
func take(key string, qps, requires int, pool *redis.Pool) (int64, error) {
c := pool.Get()
defer c.Close()
var err error
if err := c.Err(); err != nil {
return 0, err
}
reply, err := rlScript.Do(c, key, qps, requires)
if err != nil {
return 0, err
}
return reply.(int64), nil
}
func NewRedisPool(address, password string) *redis.Pool {
pool := &redis.Pool{
MaxIdle: 50,
IdleTimeout: 240 * time.Second,
TestOnBorrow: func(c redis.Conn, t time.Time) error {
_, err := c.Do("PING")
return err
},
Dial: func() (redis.Conn, error) {
return dial("tcp", address, password)
},
}
return pool
}
func dial(network, address, password string) (redis.Conn, error) {
c, err := redis.Dial(network, address)
if err != nil {
return nil, err
}
if password != "" {
if _, err := c.Do("AUTH", password); err != nil {
c.Close()
return nil, err
}
}
return c, err
}
func main() {
var cpuprofile = flag.String("cpuprofile", "", "write cpu profile `file`")
var memprofile = flag.String("memprofile", "", "write memory profile to `file`")
flag.Parse()
if *cpuprofile != "" {
f, err := os.Create(*cpuprofile)
if err != nil {
log.Fatal("could not create CPU profile: ", err)
}
if err := pprof.StartCPUProfile(f); err != nil {
log.Fatal("could not start CPU profile: ", err)
}
defer pprof.StopCPUProfile()
}
test()
if *memprofile != "" {
f, err := os.Create(*memprofile)
if err != nil {
log.Fatal("could not create memory profile: ", err)
}
runtime.GC() // get up-to-date statistics
if err := pprof.WriteHeapProfile(f); err != nil {
log.Fatal("could not write memory profile: ", err)
}
f.Close()
}
}
func test() {
pool := NewRedisPool("127.0.0.1:6379", "")
s1 := NewScheduler(10000, 1000000, func(r interface{}) {
take("xxx", 1000000, 1, pool)
})
s1.Start()
start := time.Now()
for i := 0; i < 100000; i++ {
s1.Enqueue(i)
}
fmt.Println(time.Since(start))
s1.Wait()
fmt.Println(time.Since(start))
}
问题在于 10000 个例程，有时即使没有向 redis 发送命令，程序也会卡住（使用“redis-cli monitor”检查），并且我的系统最大打开文件设置为 20000。
我做了分析，很多“syscall.Syscall”，有人可以给点建议吗？我的调度程序有问题吗？