golang + redis并发调度程序性能问题

Question

我编写了一个简单的并发调度程序，但它在高级别并发上似乎存在性能问题。

下面是代码(调度器+并发速率限制器测试)：

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监视器“检查)，我的系统最大打开文件被设置为20000。

我做了分析，大量的"syscall.Syscall"，谁能给点建议吗？我的调度程序出什么问题了吗？

Answer 1

在表面层面上，我唯一有疑问的是排序递增等待组和请求工作：

func (s *Scheduler) Enqueue(req interface{}) {
    select {
    case s.reqChan <- req:
        s.wg.Add(1)
    }
}

我不认为在实际工作中会造成太多的问题，但我认为这可能是一个合乎逻辑的竞争条件。在较低的并发级别和较小的工作大小时，它可能会将消息排入队列，然后切换到开始处理该消息的goroutine，然后是等待组中的工作。

接下来，您确定process方法是线程安全吗？？因此，根据redis文档，与go run -race一起运行是否有任何输出？

在某种程度上，这是完全合理的，并期望性能下降。我建议启动性能测试，以查看延迟和吞吐量从何处开始下降：

也许一个10,100,500,1000,2500,5000,10000或任何其他的池是有意义的。海事组织似乎有三个重要的变量可调：

• 工人池大小
• 工作队列缓冲区大小
• 红系MaxActive

最重要的是它看起来像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)
        },
    }

//池在给定时间分配的最大连接数。//当为零时，池中的连接数没有限制。MaxActive整型

我个人会试图了解在何时何地性能开始下降与您的员工池的大小有关。这可能使您更容易理解程序受什么限制。

Answer 2

我的测试结果表明，当例程数目增加时，每个例程每次取函数的执行时间几乎呈指数增长。

这应该是redis的一个问题，以下是redis图书馆社区的答复：

问题在于您怀疑池连接锁是什么，如果您的请求是小的/快速的，这将推动您的请求的序列化。 您应该注意，redis是单线程的，所以只需一个连接就可以获得峰值性能。由于从客户端到服务器的往返延迟，这并不完全正确，但是在这种类型的用例中，有限数量的处理器可能是最好的方法。 对于如何改进pool.Get() / conn.Close()，我有一些想法，但在您的示例中，优化例程的数量将是最好的方法。