mirror of
https://github.com/hibiken/asynq.git
synced 2026-04-14 22:55:58 +08:00
Compare commits
6 Commits
v0.26.0
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5ae1ae1299 | ||
|
|
23905a286f | ||
|
|
5586efeae7 | ||
|
|
dd3c923f44 | ||
|
|
f81c78e68d | ||
|
|
2fd155e31d |
2
.github/workflows/build.yml
vendored
2
.github/workflows/build.yml
vendored
@@ -79,6 +79,6 @@ jobs:
|
|||||||
go-version: stable
|
go-version: stable
|
||||||
|
|
||||||
- name: golangci-lint
|
- name: golangci-lint
|
||||||
uses: golangci/golangci-lint-action@v6
|
uses: golangci/golangci-lint-action@v9
|
||||||
with:
|
with:
|
||||||
version: v1.61
|
version: v1.61
|
||||||
|
|||||||
@@ -264,8 +264,10 @@ for i=1,2 do
|
|||||||
if (table.getn(ids) > 0) then
|
if (table.getn(ids) > 0) then
|
||||||
for _, id in ipairs(ids) do
|
for _, id in ipairs(ids) do
|
||||||
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
||||||
|
if bytes then
|
||||||
sample_total = sample_total + bytes
|
sample_total = sample_total + bytes
|
||||||
end
|
end
|
||||||
|
end
|
||||||
local n = redis.call("LLEN", KEYS[i])
|
local n = redis.call("LLEN", KEYS[i])
|
||||||
local avg = sample_total / table.getn(ids)
|
local avg = sample_total / table.getn(ids)
|
||||||
memusg = memusg + (avg * n)
|
memusg = memusg + (avg * n)
|
||||||
@@ -281,8 +283,10 @@ for i=3,6 do
|
|||||||
if (table.getn(ids) > 0) then
|
if (table.getn(ids) > 0) then
|
||||||
for _, id in ipairs(ids) do
|
for _, id in ipairs(ids) do
|
||||||
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
||||||
|
if bytes then
|
||||||
sample_total = sample_total + bytes
|
sample_total = sample_total + bytes
|
||||||
end
|
end
|
||||||
|
end
|
||||||
local n = redis.call("ZCARD", KEYS[i])
|
local n = redis.call("ZCARD", KEYS[i])
|
||||||
local avg = sample_total / table.getn(ids)
|
local avg = sample_total / table.getn(ids)
|
||||||
memusg = memusg + (avg * n)
|
memusg = memusg + (avg * n)
|
||||||
@@ -304,13 +308,17 @@ if table.getn(groups) > 0 then
|
|||||||
local ids = redis.call("ZRANGE", group_key, 0, sample_size - 1)
|
local ids = redis.call("ZRANGE", group_key, 0, sample_size - 1)
|
||||||
for _, id in ipairs(ids) do
|
for _, id in ipairs(ids) do
|
||||||
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
local bytes = redis.call("MEMORY", "USAGE", ARGV[1] .. id)
|
||||||
|
if bytes then
|
||||||
agg_task_sample_total = agg_task_sample_total + bytes
|
agg_task_sample_total = agg_task_sample_total + bytes
|
||||||
agg_task_sample_size = agg_task_sample_size + 1
|
agg_task_sample_size = agg_task_sample_size + 1
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
end
|
||||||
|
if agg_task_sample_size > 0 then
|
||||||
local avg = agg_task_sample_total / agg_task_sample_size
|
local avg = agg_task_sample_total / agg_task_sample_size
|
||||||
memusg = memusg + (avg * agg_task_count)
|
memusg = memusg + (avg * agg_task_count)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
return memusg
|
return memusg
|
||||||
`)
|
`)
|
||||||
|
|||||||
@@ -1488,6 +1488,7 @@ func (r *RDB) CancelationPubSub() (*redis.PubSub, error) {
|
|||||||
pubsub := r.client.Subscribe(ctx, base.CancelChannel)
|
pubsub := r.client.Subscribe(ctx, base.CancelChannel)
|
||||||
_, err := pubsub.Receive(ctx)
|
_, err := pubsub.Receive(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
pubsub.Close()
|
||||||
return nil, errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub receive error: %v", err))
|
return nil, errors.E(op, errors.Unknown, fmt.Sprintf("redis pubsub receive error: %v", err))
|
||||||
}
|
}
|
||||||
return pubsub, nil
|
return pubsub, nil
|
||||||
|
|||||||
@@ -3274,6 +3274,29 @@ func TestCancelationPubSub(t *testing.T) {
|
|||||||
mu.Unlock()
|
mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestCancelationPubSubReceiveError(t *testing.T) {
|
||||||
|
// Use a client connected to a non-existent Redis server to trigger
|
||||||
|
// a Receive() error. This verifies that the pubsub connection is
|
||||||
|
// closed on error, preventing connection leaks.
|
||||||
|
client := redis.NewClient(&redis.Options{
|
||||||
|
Addr: "localhost:0", // invalid port — connection will fail
|
||||||
|
})
|
||||||
|
r := NewRDB(client)
|
||||||
|
defer r.Close()
|
||||||
|
|
||||||
|
pubsub, err := r.CancelationPubSub()
|
||||||
|
if err == nil {
|
||||||
|
// If no error, we must clean up the pubsub.
|
||||||
|
if pubsub != nil {
|
||||||
|
pubsub.Close()
|
||||||
|
}
|
||||||
|
t.Fatal("(*RDB).CancelationPubSub() expected to return an error when redis is unreachable")
|
||||||
|
}
|
||||||
|
if pubsub != nil {
|
||||||
|
t.Error("(*RDB).CancelationPubSub() expected nil pubsub on error")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestWriteResult(t *testing.T) {
|
func TestWriteResult(t *testing.T) {
|
||||||
r := setup(t)
|
r := setup(t)
|
||||||
defer r.Close()
|
defer r.Close()
|
||||||
|
|||||||
Reference in New Issue
Block a user