tarantool remote multiplexing call example

remote-multi-call-example.lua

box.cfg {
  username = nil;
  work_dir = nil;
  wal_dir = ".";
  snap_dir = ".";
  sophia_dir = ".";
  listen = 3307;
  pid_file = "example.pid";
  background = true;
  slab_alloc_arena = 1.0;
  slab_alloc_minimal = 64;
  slab_alloc_maximal = 10485760;
  slab_alloc_factor = 1.06;
  snapshot_period = 0;
  snapshot_count = 6;
  panic_on_snap_error = true;
  panic_on_wal_error = true;
  rows_per_wal = 500000;
  snap_io_rate_limit = nil;
  wal_mode = "none";
  wal_dir_rescan_delay = 2.0;
  io_collect_interval = nil;
  readahead = 16320;
  log_level = 5;
  logger_nonblock = true;
  too_long_threshold = 0.5;
}

local function bootstrap()
  local space = box.schema.create_space('example')
  space:create_index('primary')
end

box.once('example-1.0', bootstrap)


fiber = require('fiber')
net_box = require('net.box')
log = require('log')
yaml = require('yaml')

log.error('start')

local conn={}

-- ping  > 7ms
conn[1] = net_box.new('192.168.0.104', 3406)


local cdefault = 100000
local c = cdefault
local datacall = '1234567890'
datacall = datacall .. datacall.. datacall.. datacall
local st = 0


local function ff(id, channel, datacall)
  local res = channel:get()
  local x = conn[id]:call('echo', datacall)
  -- log.error(x[1][1])
  c=c-1
  if c <= 0 then
    log.error('finish simple call')
    log.error(fiber.time()-st)
  end  
end


local function t1()
  local channel = fiber.channel(1)
  for i=1,c do
    fiber.create(ff, (i%1)+1, channel, datacall..i..'_')
  end
  st = fiber.time()
  channel:close()
end







local qq={}
local channelqq = fiber.channel(0)



local function worker(channelqq)
  local q={}
  local c={}
  while true do
    
    local qq = channelqq:get()
    
    --log.error('worker wakeup')
    
    q = {}
    c= {}
    for i,data in ipairs(qq) do 
      q[i]= data
      c[i] = q[i][3]
      q[i][3] = nil
    end
    
    --  local res = conn[math.random(9)]:call('mulcall', q)
    local res = conn[1]:call('mulcall', q)
    --        log.error('mulcall res ')
    --        log.error(yaml.encode(res))
    for i,data in ipairs(res[1]) do
      c[i]:put(data)
    end  
    
  end
end 




--  --  on server 


--function mulcall(req)
--  local res={}
--  for i,data in ipairs(req) do 
--    res[i] = _G[data[1]](data[2])
--  end
--  --log.error('mulcall')
--  return res
--end


local function remotecall(fname, arg1)
  
  local channel = fiber.channel(1)
  table.insert(qq, {fname, arg1,channel})
  
  
  if #qq >= 1000  then  -- orevery 3 ms
    local post = qq
    qq = {}
    
    local p = channelqq:put(post, 0.05)
    --local p = channelqq:put(post)
    if not p then
      -- log.error('fiber create worker')
      fiber.create(worker,channelqq)
      channelqq:put(post)
    end
    
  end
  
  return channel
end





local function t2()
  
  
  for i=1,100 do
    fiber.create(worker,channelqq)
  end
  
  
  
  
  
  --    local cc = remotecall('echo', 'b')
  --    local rr = cc:get()1
  --    log.error('remotecall res')
  --    
  --    log.error(yaml.encode(rr))
  
  
  с=cdefault
  local st = 0
  local channel = fiber.channel(1)
  
  local function fiber_remote_call(channel, datacall)
    
    local res = channel:get()
    local cc = remotecall('echo', datacall)
    local rr = cc:get()
    --        log.error('remotecall res')
    --        log.error(rr)
    c=c-1
    if c <= 0 then
      log.error('finish remotecall')
      log.error(fiber.time()-st)
    end  
  end
  
  for i=1,c do
    fiber.create(fiber_remote_call, channel, datacall..i..'_')
  end
  st = fiber.time()
  channel:close()
  
end


t1()  --simple call

--t2()  -- multiplexing call

server

box.cfg {
    username = nil;
    work_dir = nil;
    wal_dir = ".";
    snap_dir = ".";
    sophia_dir = ".";
    listen = 3306;
    pid_file = "s.pid";
    background = true;
    slab_alloc_arena = 1.0;
    slab_alloc_minimal = 64;
    slab_alloc_maximal = 10485760;
    slab_alloc_factor = 1.06;
    snapshot_period = 0;
    snapshot_count = 6;
    panic_on_snap_error = true;
    panic_on_wal_error = true;
    rows_per_wal = 500000;
    snap_io_rate_limit = nil;
    wal_mode = "none";
    wal_dir_rescan_delay = 2.0;
    io_collect_interval = nil;
    readahead = 16320;
    log_level = 5;
    logger_nonblock = true;
    too_long_threshold = 0.5;
}

local function bootstrap()
    local space = box.schema.create_space('example')
    space:create_index('primary')
     box.schema.user.grant('guest', 'read,write,execute', 'universe')
end
box.once('example-1.0', bootstrap)
log = require('log')

function mulcall(req)
  local res={}
  for i,data in ipairs(req) do 
    res[i] = _G[data[1]](data[2])
  end
   --log.info('mulcall')
  return res
end

function echo(e)
  return e .. e
end