luvi thread example

inproc.lua

local uv = require('uv')
local nn = require 'nanomsg-ffi'
local ffi = require('ffi')

local ADDRESS = "inproc://abcd"
--ADDRESS = "tcp://127.0.0.1:5556"

local rep = nn.socket( nn.REP )
local sndfd = rep:getsockopt(nn.SOL_SOCKET, nn.SNDFD)
local rcvfd = rep:getsockopt(nn.SOL_SOCKET, nn.RCVFD)

print('SND',sndfd,'RCV',rcvfd)

local snd = assert(uv.new_socket_poll(sndfd))
local rcv = assert(uv.new_socket_poll(rcvfd))

--[[
uv.poll_start(snd, 'rw', function(err,event)
    print('SND',err,event)
end)
--]]
local i = ffi.typeof('double')(0)
uv.poll_start(rcv, 'r', function(err,event)
        local msg = rep:recv_zc(nn.DONTWAIT)
        i = i + #msg
        rep:send_zc( msg,nn.DONTWAIT)
end)

local cid, err = rep:bind( ADDRESS )
assert( cid, nn.strerror(err) )

--[[
    repeat

        echo = msg:tostring()
        --print( "GOT:", '"' .. echo .. '"' )
        --io.write('.')
        sz, err = rep:send( echo, #echo ) -- TODO msg.ptr, msg.size )
        assert( sz > 0, nn.strerror(err) )
    until msg==nil

    rep:shutdown( cid )
end
--]]
print("...done...")
  local function setInterval(interval, callback)
    local timer = uv.new_timer()
    timer:start(interval, interval, function ()
      --timer:stop()
      --timer:close()
      callback()
    end)
    return timer
  end
  setInterval(1000,function()
    io.write(string.format('SPEED_S(MBps)\t%s\n', i/(1024*1024) ));
    i=0
  end)

local a = function()
  local uv = require('uv')
  local nn = require('nanomsg-ffi')
  local ffi = require'ffi'

  local ADDRESS = "inproc://abcd"
  --ADDRESS = "tcp://127.0.0.1:5556"

  local i = ffi.typeof('double')(0)
  --server:join()
  --[[
  local client = uv.new_thread(function()
    _G.arg = {'ABCD'}
    dofile('client.lua')
  end)
  --client:join()
  --]]
  local unm = {}
  local req, err = nn.socket( nn.REQ )
  assert( req, nn.strerror(err) )

  unm.req = req
  local sndfd = req:getsockopt(nn.SOL_SOCKET, nn.SNDFD)
  local rcvfd = req:getsockopt(nn.SOL_SOCKET, nn.RCVFD)

  local snd = assert(uv.new_socket_poll(sndfd))
  local rcv = assert(uv.new_socket_poll(rcvfd))
  uv.poll_start(snd, 'w', function(err,event)
    assert(not err,err)
    if not unm.connected then
      unm.connected = true
      uv.poll_stop(snd)

      --local msg
      uv.poll_start(rcv, 'r', function(err,status)
        --assert(status, err)
        local msg = req:recv_zc(nn.DONTWAIT)
        i = i+#msg
        req:send_zc( msg, nn.DONTWAIT)
      end)
      local msg = string.rep('0123456789abcdef',1024*1024)
      req:send(msg,#msg)
    end
  end)

  local eid, err = req:connect( ADDRESS )
  assert( eid, nn.strerror(err) )

  print('OK')
  local function setInterval(interval, callback)
    local timer = uv.new_timer()
    timer:start(interval, interval, function ()
      --timer:stop()
      --timer:close()
      callback()
    end)
    return timer
  end
  setInterval(1000,function()
    io.write(string.format('SPEED_C(MBps)\t%s\n', i/(1024*1024) ));
    i=0
  end)
  uv.run()
end

assert(uv.new_thread(pcall(a)))
--assert(uv.new_thread(pcall(a)))
--assert(uv.new_thread(pcall(a)))
--assert(uv.new_thread(pcall(a)))

uv.run()