relay-1mops.lua

#!/usr/bin/tarantool
-- vim:ts=4 ss=4 sw=4 expandtab
-- Usage: tarantool relay-1mops.lua [nodes]
-- nodes should be a number from 1 to 31

local popen = require('popen')
local clock = require('clock')
local fiber = require('fiber')
local math = require('math')
local yaml = require('yaml')

-- TUNE OPTIONS HERE
-- working dir is cleared/created for every run, set test_dir
-- during the test a subdir will be created for each replica
local test_dir = './relay-test.data'

-- turn true to test the qsync
local test_qsync = false

-- number of operations performed by test
local num_ops = 1000000

-- number of fibers
local num_fibers = 50

-- number of operations per transaction
local ops_per_txn = 100

-- number of nodes - master and replicas
local nodes = 1

-- by default no output from replicas are received
-- redirect it into master's one breaks terminal
local std_redirect = {
    stdin='devnull', 
    stdout='devnull', 
    stderr='devnull'
}

-- END OF TUNABLE OPTIONS

print(string.format('making %d operations, %d operations per txn usnig %d fibers', num_ops, ops_per_txn, num_fibers))

--- transactions per fiber
local trans_per_fiber = num_ops/ops_per_txn/num_fibers

os.execute('rm -rf ' .. test_dir)
os.execute('mkdir ' .. test_dir)

box.cfg{
    log_level = 0,
    listen = 3301,
    read_only = false,
    replication = 'replicator:password@localhost:3301',
    work_dir=test_dir
}

box.schema.user.create('replicator', {password = 'password'})
box.schema.user.grant('replicator', 'replication')

-- number of nodes, storage for popen handles
local nodes_ph = {}

-- run replicas (if needed)
if (arg[1] ~= nil) then
    nodes = tonumber(arg[1])
    if (nodes ~= nil and nodes < 32 and nodes > 0) then
        print('starting ' .. nodes - 1 .. ' replicas')
        for i = 3302, 3300+nodes do
            -- subdir for replica's data
            os.execute('mkdir ' .. i)
            -- command line for replica to start
            local cmd = {
                arg[-1],
                '-e',
                [[
                box.cfg {
                    read_only = false,
                    log_level = 0,
                    replication = 'replicator:password@localhost:3301', ]] ..
                    'listen = ' .. i .. ',' ..
                    'work_dir = \'' .. i .. '\'' ..
                '}'
            }
            local res, err = popen.new(cmd, {stdin='devnull', stdout='devnull', stderr='devnull'})
            if (res) then
                nodes_ph[i] = res
            else
                print('error running replica:', err)
                os.exit()
            end
        end
        -- wait for all replicas to connect
        while #box.info.replication < nodes do
            require('fiber').sleep(0.1)
        end
    else
        print('Incorrect number of nodes \"' .. arg[1] .. '\" must be 1..31')
        os.exit(1)
    end
end

local space = 0

if (test_qsync) then
    box.cfg{replication_synchro_quorum = nodes}
    space = box.schema.create_space('test', {is_sync = true})
else
    space = box.schema.create_space('test')
end

space:create_index('pk',{type = 'HASH'})

-- each fiber runs the fiber_load that has evenly distributed workload
function fiber_load(start, s)
    for i = 1, trans_per_fiber do
        box.begin()
        for j = 1, ops_per_txn do
            if (not s:replace{start, start}) then
                os.exit(1)
            end
            start = start + 1
        end
        box.commit()
    end
end

-- fiber storage to join them
local fibers_storage = {}

-- start timer
local begin = {
    clock.time(),
    clock.proc()
}

for i = 1, num_fibers do
    fibers_storage[i] = fiber.create(fiber_load, i*num_ops, space)
    fibers_storage[i]:wakeup() -- needed for backward compatibility with 1.7
end

-- wait for all fibers to finish
for i=1,num_fibers do
    while fibers_storage[i]:status() ~= 'dead' do
        fiber.yield()
    end -- the loop is needed for backward compatibility with 1.7
end

-- stop timer for master
local len = {
    clock.time()-begin[1],
    clock.proc()-begin[2]
}

local ops_done = box.info.replication[1].lsn

print(string.format('master done %d ops in time: %f, cpu: %f', ops_done, len[1], len[2]))
print('master speed', math.floor(ops_done / len[1]), 'ops/sec')

-- wait for all replicas and kill them
if nodes > 1 then
    local done = false
    while(not done) do
        for i = 2, nodes do
            local r_vclock = box.info.replication[i].downstream.vclock
            if (r_vclock and
                r_vclock[1] < ops_done) then
                done = false
                break
            end
            done = true
        end
        if (not done) then
            fiber.sleep(0.001)
        end
    end
    -- stop timer for replicas
    len = {
        clock.time()-begin[1], 
        clock.proc()-begin[2]
    }
    
    print(string.format('replicas done %d ops in time: %f, cpu: %f', ops_done, len[1], len[2]))
    print('replicas speed', math.floor(ops_done / len[1]), 'ops/sec')
    
    for _, replica in pairs(nodes_ph) do
        replica:kill()
        replica:wait()
    end
end

os.exit(0)