fixed evoyager

-- stress test base class

require "std"
require "oop"
require "etc"
require "net"



local c,p = class:TcpStress {
    host = "localhost",
    port = 7000,
    count = 1000,
    threads = 10,
    objID_from = 0,
    password = "12345678",
    version = "VER 07.503.503",
    complex_timeout_sec = 0,
    complex_timeout_usec = 100000,
    epolls = {},
    clients = {},
    cgroups = {},
    timer = 0,
    sent = 0,
    received = 0,
    total_timer = 0,
    total_sent = 0,
    total_received = 0,
    eop = "\r\nOK\r\n",
}

c.stages = {
    READY   = 1,
    VERSION = 2,
    COMPLEX = 3,
    SENDOK  = 4,
}

c.packets = {
    READY   = "\r\nREADY\r\n",
    VERSION = "\r\n+v\r\n",
    COMPLEX = "\r\n+complex\r\n",
    SENDOK  = "\r\nsend ok\r\n",
}

c.history = {
    0x30, 0x30, 0x33, 0x43, 0x3e, 0x1e, 0x3e, 0x25,
    0x25, 0x1c, 0x81, 0x81, 0x1c, 0x14, 0x16, 0x18,
    0x1a, 0x1c, 0x1e, 0x20, 0x22, 0x24, 0x26, 0x28,
    0x2a, 0x2c, 0x2e, 0x30, 0x32, 0x33, 0x35, 0x37,
    0x39, 0x3b, 0x3d, 0x3f, 0x41, 0x43, 0x45, 0x47,
    0x49, 0x4b, 0x4d, 0xb8, 0xac, 0xa0, 0x94, 0x88,
    0x7c, 0x70, 0x64, 0x58, 0x4c, 0x40, 0x34, 0x28,
    0x1c, 0x10, 0x04, 0xf8, 0xec, 0xe0, 0xd4, 0xc8,
    0xbc, 0xb0, 0xa4, 0x98, 0x8c, 0x80, 0x74, 0x68,
    0x5c, 0x2d, 0x2d, 0x1c, 0xda, 0xda, 0x1c, 0x2d,
    0x31, 0x35, 0x39, 0x3d, 0x41, 0x45, 0x49, 0x4d,
    0x51, 0x54, 0x58, 0x5c, 0x60, 0x64, 0x68, 0x6c,
    0x70, 0x74, 0x78, 0x7c, 0x7f, 0x83, 0x87, 0x8b,
    0x8f, 0x93, 0x97, 0x9b, 0x9f, 0xec, 0xd4, 0xbc,
    0xa4, 0x8c, 0x74, 0x5c, 0x44, 0x2c, 0x14, 0xfc,
    0xe4, 0xcc, 0xb4, 0x9c, 0x84, 0x6c, 0x54, 0x3c,
    0x24, 0x0c, 0xf4, 0xdc, 0xc4, 0xac, 0x94, 0x7c,
    0x64, 0x4c, 0x34, 0x01, 0x01, 0x1c, 0x64, 0x64,
    0x1c, 0x4b, 0x4b, 0x1c, 0x01, 0x01, 0x1c, 0xb7,
    0xb7, 0x1c, 0x7b, 0x7b, 0x1c, 0x00, 0x00, 0x1c,
    0x00, 0x00, 0x1c, 0x00, 0x00, 0x1c, 0x00, 0x00,
    0x1c, 0x00, 0x00, 0x1c, 0x00, 0x00, 0x1c, 0x00,
    0x00, 0x1c, 0x00, 0x00, 0x1c, 0x00, 0x00, 0x1c,
    0x00, 0x00, 0x1c, 0x3b, 0x3b, 0x1c, 0x01, 0x01,
    0x1c, 0x13, 0x30, 0x4d, 0x6a, 0x87, 0xa4, 0xc1,
    0xde, 0xfb, 0x18, 0x36, 0x53, 0x70, 0x8d, 0xaa,
    0xc7, 0xe5, 0x02, 0x1f, 0x3c, 0x59, 0x77, 0x94,
    0xb1, 0xce, 0xeb, 0x08, 0x25, 0x42, 0x5f }

c.historychanged = false



function p:init()
    self:prepare_history()
    self:init_clients()
    self:init_epolls()
    thread.newthread(self.printer, {self})
    self.timer = etc.microtime()
    self.total_timer = self.timer
end



function p:prepare_history()
    if c.historychanged then return end
    c.historychanged = true

    for i,chr in ipairs(c.history) do
        c.history[i] = string.char(chr)
    end

    c.history = table.concat(c.history, "")
end



-- nonblock client socket creation + binding to epolls
function p:init_clients()
    local spt = math.ceil(self.count / self.threads) -- sockets per thread/epoll
    local n = 1 -- socket number
    local cid, tid -- client id, thread id

    for tid=1,self.threads do
        self.epolls[tid] = assert(net.epoll())
        self.cgroups[tid] = {}

        for i=1,spt do
            cid = i + spt*(tid-1)
            if cid > self.count then return end

            local s,fd = assert(net.tcp.client(self.host, self.port, 0))
            assert(self.epolls[tid]:watch(s,1))

            local objID = tostring(self.objID_from + cid)
            while #objID < 4 do objID = "0"..objID end

            local client = {
                socket = s,
                connected = false,
                stage = c.stages.READY,
                objID = objID,
                buffer = "",
            }

            self.clients[fd] = client
            self.cgroups[tid][fd] = client
        end
    end
end



-- start all epolls
function p:init_epolls()
    local void = function() end
    local onread, onwrite, onclose, ontimeout, onerror = void, void, void, void, void

    for tid=1,self.threads do
        local epoll = self.epolls[tid]
        if not epoll then return end

        onerror = function( err )
            if type(err) == "number" then
                print("Fatal error: cant connect")
            else
                print("Fatal error: "..err)
            end
            os.exit(1)
        end

        onread = function( fd )
            local client = self.clients[fd]
            if not client.connected then client.connected = true end

            while true do
                local msg,len = client.socket:read()
                if not msg then break end

                if len == 0 then
                    error("client was dropped")
                end

                client.buffer = client.buffer .. msg
                self:client_process(client)
            end
        end

        thread.newthread(epoll.start, {epoll, -1, onread, onwrite, onclose, ontimeout, onerror})
    end
end



function p:client_process( client )
    if client.stage == c.stages.READY then
        if client.buffer:sub(1,#c.packets.READY) == c.packets.READY then
            client.buffer = client.buffer:sub(#c.packets.READY+1)
            client.stage = c.stages.VERSION
            self:client_process(client)
        else
            trace(client.buffer)
            error("wrong buf")
        end
    elseif client.stage == c.stages.VERSION then
        if client.buffer == c.packets.VERSION then
            client.buffer = ""
            client.socket:write(self.version..self.eop)
            client.stage = c.stages.COMPLEX
        else
            trace(client.buffer)
            error("wrong buf")
        end
    elseif client.stage == c.stages.COMPLEX then
        if client.buffer == c.packets.COMPLEX then
            client.buffer = ""
            client.socket:write("complex>"..client.objID..self.password.."000000023F"..self.eop)
            etc.sleep(self.complex_timeout_sec,self.complex_timeout_usec)
            client.socket:write(c.history..self.eop)
            self.sent = self.sent + 1
            self.total_sent = self.total_sent + 1
            client.stage = c.stages.SENDOK
        else
            trace(client.buffer)
            error("wrong buf")
        end
    elseif client.stage == c.stages.SENDOK then
        if client.buffer == c.packets.SENDOK then
            client.buffer = ""
            client.socket:write(c.history..self.eop)
            self.sent = self.sent + 1
            self.received = self.received + 1
            self.total_sent = self.total_sent + 1
            self.total_received = self.total_received + 1
        else
            trace(client.buffer)
            error("wrong buf")
        end
    else
        error("unknown stage")
    end
end



--
function p:printer()
    while true do
        etc.sleep(1)

        local time = etc.microtime() - self.timer
        local total_time = etc.microtime() - self.total_timer

        io.write("\r", string.timer(total_time),
            "; t-recvd:", self.total_received,
            "; t-rps:", math.precision(self.total_received/total_time),
            "; recvd:", self.received,
            "; rps:", math.precision(self.received/time),
            "; t-sent:", self.total_sent,
            "; t-sps:", math.precision(self.total_sent/total_time),
        "  ")
        io.flush()

        if time > 5 then
            self.timer = etc.microtime()
            self.received = 0
            self.sent = 0
        end
    end
end