Redis pub/sub issue in conjunction with websockets

bogdan · 2014-10-22T19:07:24+00:00

Thanks a ton! Looking forward to it :-) On Thursday, March 19, 2015 at 12:59:07 PM UTC-7, agentzh wrote:Hello! On Thu, Mar 19, 2015 at 12:53 PM, S wr...

Redis pub/sub issue in conjunction with websockets

bogdan

Hello, all

I am building an application with Redis and using the Pub/Sub feature.
I am publishing on several channels events at each 5 seconds. I also subscribe to these channels from multiple clients (loops) in Lua.
A period of time after starting Redis, the events arrive to the clients in a timely manner. But, after a while, the messages (events) don't arrive one by one to the clients, but bulk, being delayed. They tend to arrive 5-6 events at once, at every 30 seconds or so.
Is this a normal behavior? Do you have any idea why this would happen?
During the execution of this flow, some other Redis operations take place.

Here is the code that handles the websocket feature, which uses the pub/sub connection in a light thread.
Please do comment on the quality of this code (these are some excerpts from a module):

local function redis_connect(timeout)
    local redis = require "resty.redis"
    local red = redis:new()

    red:set_timeout(timeout*1000)

    local ok, err = red:connect("127.0.0.1", 6380)
    if not ok then
        return nil, "Failed to connect to Redis: " .. err
    else
        return red
    end
end

local function subscribe(ws, hosts, user_id, cid, charts)
    local red, err = redis_connect(31556926) -- timeout Redis connection in a year
    if not red then
        return
    end

    for i, host in ipairs(hosts) do
        local channel = "stats:"..host['id']..":*"
        local res, err = red:psubscribe(channel)
        if not res then
            ngx.log(ngx.ERR, "Could not subscribe to channel(s) \""..channel.."\"; err: "..err)
            return
        end
    end

    while true do
        local bytes, err = red:read_reply()
        if err then
            ngx.log(ngx.ERR, "Error reading reply: "..err)
            return
        end
        if bytes then
            -- here we do event processing (Redis-related too, but on a separate connection with lower timeout)
            -- and printing something on the web socket
        end
    end
end

function _M.run_websocket(subscription, cid)
    local server = require "resty.websocket.server"

    local wb, err = server:new{
        timeout = 5000, -- in milliseconds
        max_payload_len = 65535,
    }
    if not wb then
        ngx.log(ngx.ERR, "Failed to create new websocket: ", err)
        return ngx.exit(444)
    end

    local hosts = .... prepare some data....
    local charts = .... prepare some data....
    local thread = ngx.thread.spawn(subscribe, wb, hosts, subscription['user_id'], cid, charts)

    while true do
        local data, typ, err = wb:recv_frame()

        if not data then
            ngx.log(ngx.ERR, "Failed to receive a frame: ", err)
            ngx.thread.kill(thread)
            return ngx.exit(444)
        end

        if typ == "close" then
            -- send a close frame back:
            ngx.thread.kill(thread)
            local bytes, err = wb:send_close(1000, "connection will close")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the close frame: ", err)
                return ngx.exit(444)
            end
            return
        elseif typ == "ping" then
            -- send a pong frame back:

            local bytes, err = wb:send_pong(data)
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the pong frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        elseif typ == "pong" or (typ == "text" and string.lower(data) == 'ping') then
            -- just discard the incoming pong frame

        elseif typ == "text" then
            -- just "echo" back what I received

            bytes, err = wb:send_text("RECEIVED:\""..data.."\"")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send a text frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        end
    end
end

Thank you

Bogdan

manchev

Hello,

The delays and the fact you're receiving multiple messages at once (didn't think it's even possible) seems like a problem with your code, particularly the publisher code. Maybe you can share some more code so people on the list can help you.

As for comments on quality, here are some general pointers:

- bind Redis on a unix socket -- it's both more secure and faster if you're going to run it on the same node

- use plain for loop instead of the ipairs -- http://lua-users.org/lists/lua-l/2011-05/msg00007.html

- depending on your use case you might want to add some ngx.sleep at the end of tight loops in order to let other light threads run in the mean time

- make sure you're doing proper error handling in such tight loops or otherwise you're risking your app to eat all of your CPU(s)

Never actually used pattern subscriptions in production due to time complexity so some of the side effects you're seeing might be coming from that -- would be great if someone else can chime in.

Best,
Vladislav

On Wed, Oct 22, 2014 at 2:07 PM, Bogdan Irimia <bo...@digitair.ro> wrote:

Hello, all

I am building an application with Redis and using the Pub/Sub feature.
I am publishing on several channels events at each 5 seconds. I also subscribe to these channels from multiple clients (loops) in Lua.
A period of time after starting Redis, the events arrive to the clients in a timely manner. But, after a while, the messages (events) don't arrive one by one to the clients, but bulk, being delayed. They tend to arrive 5-6 events at once, at every 30 seconds or so.
Is this a normal behavior? Do you have any idea why this would happen?
During the execution of this flow, some other Redis operations take place.

Here is the code that handles the websocket feature, which uses the pub/sub connection in a light thread.
Please do comment on the quality of this code (these are some excerpts from a module):

local function redis_connect(timeout)
    local redis = require "resty.redis"
    local red = redis:new()

    red:set_timeout(timeout*1000)

    local ok, err = red:connect("127.0.0.1", 6380)
    if not ok then
        return nil, "Failed to connect to Redis: " .. err
    else
        return red
    end
end

local function subscribe(ws, hosts, user_id, cid, charts)
    local red, err = redis_connect(31556926) -- timeout Redis connection in a year
    if not red then
        return
    end

    for i, host in ipairs(hosts) do
        local channel = "stats:"..host['id']..":*"
        local res, err = red:psubscribe(channel)
        if not res then
            ngx.log(ngx.ERR, "Could not subscribe to channel(s) \""..channel.."\"; err: "..err)
            return
        end
    end

    while true do
        local bytes, err = red:read_reply()
        if err then
            ngx.log(ngx.ERR, "Error reading reply: "..err)
            return
        end
        if bytes then
            -- here we do event processing (Redis-related too, but on a separate connection with lower timeout)
            -- and printing something on the web socket
        end
    end
end

function _M.run_websocket(subscription, cid)
    local server = require "resty.websocket.server"

    local wb, err = server:new{
        timeout = 5000, -- in milliseconds
        max_payload_len = 65535,
    }
    if not wb then
        ngx.log(ngx.ERR, "Failed to create new websocket: ", err)
        return ngx.exit(444)
    end

    local hosts = .... prepare some data....
    local charts = .... prepare some data....
    local thread = ngx.thread.spawn(subscribe, wb, hosts, subscription['user_id'], cid, charts)

    while true do
        local data, typ, err = wb:recv_frame()

        if not data then
            ngx.log(ngx.ERR, "Failed to receive a frame: ", err)
            ngx.thread.kill(thread)
            return ngx.exit(444)
        end

        if typ == "close" then
            -- send a close frame back:
            ngx.thread.kill(thread)
            local bytes, err = wb:send_close(1000, "connection will close")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the close frame: ", err)
                return ngx.exit(444)
            end
            return
        elseif typ == "ping" then
            -- send a pong frame back:

            local bytes, err = wb:send_pong(data)
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the pong frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        elseif typ == "pong" or (typ == "text" and string.lower(data) == 'ping') then
            -- just discard the incoming pong frame

        elseif typ == "text" then
            -- just "echo" back what I received

            bytes, err = wb:send_text("RECEIVED:\""..data.."\"")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send a text frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        end
    end
end

Thank you

Bogdan

.

bogdan

Thank you for your feedback.
Actually, the messages don't come really "at once", but at very small intervals one after another, so that the user perceives them as "one".
Thank you for the advices. We moved to unix sockets indeed. The advice with the ipairs is one to remember :) I will keep that in mind.
Regarding the ngx.sleep, it's a thing I'm missing both in my websocket loop, as well as in my pub/sub redis light thread loop. I will try to use sleeps in both loops and see what happens.

What about the code for the websocket? Is that conceptually correct?

Thank you

Vladislav Manchev

Wednesday, October 22, 2014 17:21

Hello,

The delays and the fact you're receiving multiple messages at once (didn't think it's even possible) seems like a problem with your code, particularly the publisher code. Maybe you can share some more code so people on the list can help you.

As for comments on quality, here are some general pointers:
- bind Redis on a unix socket -- it's both more secure and faster if you're going to run it on the same node
- use plain for loop instead of the ipairs -- http://lua-users.org/lists/lua-l/2011-05/msg00007.html
- depending on your use case you might want to add some ngx.sleep at the end of tight loops in order to let other light threads run in the mean time
- make sure you're doing proper error handling in such tight loops or otherwise you're risking your app to eat all of your CPU(s)

Never actually used pattern subscriptions in production due to time complexity so some of the side effects you're seeing might be coming from that -- would be great if someone else can chime in.

Best,
Vladislav

        max_payload_len = 65535,
    }
    if not wb then
        ngx.log(ngx.ERR, "Failed to create new websocket: ", err)
        return ngx.exit(444)
    end

    local hosts = .... prepare some data....
    local charts = .... prepare some data....
    local thread = ngx.thread.spawn(subscribe, wb, hosts, subscription['user_id'], cid, charts)

    while true do
        local data, typ, err = wb:recv_frame()

        if not data then
            ngx.log(ngx.ERR, "Failed to receive a frame: ", err)
            ngx.thread.kill(thread)
            return ngx.exit(444)
        end

        if typ == "close" then
            -- send a close frame back:
            ngx.thread.kill(thread)
            local bytes, err = wb:send_close(1000, "connection will close")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the close frame: ", err)
                return ngx.exit(444)
            end
            return
        elseif typ == "ping" then
            -- send a pong frame back:

            local bytes, err = wb:send_pong(data)
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the pong frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        elseif typ == "pong" or (typ == "text" and string.lower(data) == 'ping') then
            -- just discard the incoming pong frame

        elseif typ == "text" then
            -- just "echo" back what I received

            bytes, err = wb:send_text("RECEIVED:\""..data.."\"")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send a text frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        end
    end
end

Thank you

Bogdan

manchev

Yes, websocket code looks good to me, we're using something similar in production and it's been running smoothly for quite some time already.

The problem with the way messages are received is interesting though -- I was wondering if you're somehow saturating your NIC or TCP stack?

One more thing to keep in mind -- make sure you're using the connection pool feature of lua-resty-redis, i.e. call set_keepalive [1] on the Redis connection when you're done.

[1]: https://github.com/openresty/lua-resty-redis#set_keepalive

Best,

Vladislav

On Thu, Oct 23, 2014 at 11:13 AM, Bogdan Irimia <bo...@digitair.ro> wrote:

Thank you for your feedback.
Actually, the messages don't come really "at once", but at very small intervals one after another, so that the user perceives them as "one".
Thank you for the advices. We moved to unix sockets indeed. The advice with the ipairs is one to remember :) I will keep that in mind.
Regarding the ngx.sleep, it's a thing I'm missing both in my websocket loop, as well as in my pub/sub redis light thread loop. I will try to use sleeps in both loops and see what happens.

What about the code for the websocket? Is that conceptually correct?

Thank you

Vladislav Manchev

Wednesday, October 22, 2014 17:21

Hello,

The delays and the fact you're receiving multiple messages at once (didn't think it's even possible) seems like a problem with your code, particularly the publisher code. Maybe you can share some more code so people on the list can help you.

As for comments on quality, here are some general pointers:
- bind Redis on a unix socket -- it's both more secure and faster if you're going to run it on the same node
- use plain for loop instead of the ipairs -- http://lua-users.org/lists/lua-l/2011-05/msg00007.html
- depending on your use case you might want to add some ngx.sleep at the end of tight loops in order to let other light threads run in the mean time
- make sure you're doing proper error handling in such tight loops or otherwise you're risking your app to eat all of your CPU(s)

Never actually used pattern subscriptions in production due to time complexity so some of the side effects you're seeing might be coming from that -- would be great if someone else can chime in.

Best,
Vladislav

.

bogdan

Thank you

How can I saturate the NIC or TCP stack if we're using unix sockets on the same machine?

I am using set_keepalive wherever I can, but in the code below there's no point in doing that, because the websocket connection is persistent, right? In the light thread (where I connect to Redis) I don't know when the execution will end (thread is killed) so I can't set the redis connection to keepalive. But the rest of the code uses keepalive for redis wherever possible.

I'm still digging into this with Redis' latency monitoring tool, but I can't find anything abnormal.

Vladislav Manchev

Thursday, October 23, 2014 16:01

Yes, websocket code looks good to me, we're using something similar in production and it's been running smoothly for quite some time already.

The problem with the way messages are received is interesting though -- I was wondering if you're somehow saturating your NIC or TCP stack?

One more thing to keep in mind -- make sure you're using the connection pool feature of lua-resty-redis, i.e. call set_keepalive [1] on the Redis connection when you're done.

[1]: https://github.com/openresty/lua-resty-redis#set_keepalive

Best,
Vladislav

        max_payload_len = 65535,
    }
    if not wb then
        ngx.log(ngx.ERR, "Failed to create new websocket: ", err)
        return ngx.exit(444)
    end

    local hosts = .... prepare some data....
    local charts = .... prepare some data....
    local thread = ngx.thread.spawn(subscribe, wb, hosts, subscription['user_id'], cid, charts)

    while true do
        local data, typ, err = wb:recv_frame()

        if not data then
            ngx.log(ngx.ERR, "Failed to receive a frame: ", err)
            ngx.thread.kill(thread)
            return ngx.exit(444)
        end

        if typ == "close" then
            -- send a close frame back:
            ngx.thread.kill(thread)
            local bytes, err = wb:send_close(1000, "connection will close")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the close frame: ", err)
                return ngx.exit(444)
            end
            return
        elseif typ == "ping" then
            -- send a pong frame back:

            local bytes, err = wb:send_pong(data)
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send the pong frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        elseif typ == "pong" or (typ == "text" and string.lower(data) == 'ping') then
            -- just discard the incoming pong frame

        elseif typ == "text" then
            -- just "echo" back what I received

            bytes, err = wb:send_text("RECEIVED:\""..data.."\"")
            if not bytes then
                ngx.log(ngx.ERR, "Failed to send a text frame: ", err)
                ngx.thread.kill(thread)
                return ngx.exit(444)
            end
        end
    end
end

Thank you

Bogdan

manchev

Sorry, didn't realize you're still getting the same problem with unix sockets.

You can always try the nginx-systemtap-toolkit [1] tools in order to get a better picture of what's going on in your code. It can be an eye-opener in such cases.

Do keep us posted as this might be useful for others experiencing problems like yours.

[1]: https://github.com/openresty/nginx-systemtap-toolkit

Best,
Vladislav

On Thu, Oct 23, 2014 at 4:09 PM, Bogdan Irimia <bo...@digitair.ro> wrote:

Thank you

How can I saturate the NIC or TCP stack if we're using unix sockets on the same machine?

I am using set_keepalive wherever I can, but in the code below there's no point in doing that, because the websocket connection is persistent, right? In the light thread (where I connect to Redis) I don't know when the execution will end (thread is killed) so I can't set the redis connection to keepalive. But the rest of the code uses keepalive for redis wherever possible.

I'm still digging into this with Redis' latency monitoring tool, but I can't find anything abnormal.

Vladislav Manchev

Thursday, October 23, 2014 16:01

Yes, websocket code looks good to me, we're using something similar in production and it's been running smoothly for quite some time already.

The problem with the way messages are received is interesting though -- I was wondering if you're somehow saturating your NIC or TCP stack?

One more thing to keep in mind -- make sure you're using the connection pool feature of lua-resty-redis, i.e. call set_keepalive [1] on the Redis connection when you're done.

[1]: https://github.com/openresty/lua-resty-redis#set_keepalive

Best,
Vladislav

.