diff options
Diffstat (limited to '3rdparty/luv/tests/manual-test-cluster.lua')
-rw-r--r-- | 3rdparty/luv/tests/manual-test-cluster.lua | 213 |
1 files changed, 213 insertions, 0 deletions
diff --git a/3rdparty/luv/tests/manual-test-cluster.lua b/3rdparty/luv/tests/manual-test-cluster.lua new file mode 100644 index 00000000000..304e4be634d --- /dev/null +++ b/3rdparty/luv/tests/manual-test-cluster.lua @@ -0,0 +1,213 @@ +-- This is quite the involved test. Basically it binds +-- to a tcp port, spawns n children (one per CPU core) +-- who all listen on the same shared port and act as a +-- load balancing cluster. +-- Then N clients are spawned that connect to the cluster +-- The application itself kills the worker upon connection +-- All N workers should accept exactly one request and all close. + +return require('lib/tap')(function (test) + + -- This function will be run in a child process + local worker_code = string.dump(function () + local dump = require('lib/utils').dump + + local function print(...) + local n = select('#', ...) + local arguments = {...} + for i = 1, n do + arguments[i] = tostring(arguments[i]) + end + + local text = table.concat(arguments, "\t") + text = " " .. string.gsub(text, "\n", "\n ") + _G.print(text) + end + + local function p(...) + local n = select('#', ...) + local arguments = { ... } + + for i = 1, n do + arguments[i] = dump(arguments[i]) + end + + print(table.concat(arguments, "\t")) + end + + local uv = require('luv') + local answer = -1 + + -- The parent is going to pass us the server handle over a pipe + -- This will be our local file descriptor at PIPE_FD + local pipe = uv.new_pipe(true) + local pipe_fd = tonumber(os.getenv("PIPE_FD")) + assert(uv.pipe_open(pipe, pipe_fd)) + + -- Configure the server handle + local server = uv.new_tcp() + local done = false + local function onconnection() + print("NOT ACCEPTING, already done") + if done then return end + local client = uv.new_tcp() + assert(uv.accept(server, client)) + p("New TCP", client, "on", server) + p{client=client} + assert(uv.write(client, "BYE!\n")); + assert(uv.shutdown(client, function () + uv.close(client) + uv.unref(server) + done = true + answer = 42 + end)) + end + + -- Read the server handle from the parent + local function onread(err, data) + p("onread", {err=err,data=data}) + assert(not err, err) + if uv.pipe_pending_count(pipe) > 0 then + local pending_type = uv.pipe_pending_type(pipe) + p("pending_type", pending_type) + assert(pending_type == "tcp") + assert(uv.accept(pipe, server)) + assert(uv.listen(server, 0, onconnection)) + p("Received server handle from parent process", server) + elseif data then + p("ondata", data) + else + p("onend", data) + end + end + uv.read_start(pipe, onread) + + -- Start the event loop! + uv.run() + + os.exit(answer) + end) + + local client_code = string.dump(function () + local dump = require('lib/utils').dump + + local function print(...) + local n = select('#', ...) + local arguments = {...} + for i = 1, n do + arguments[i] = tostring(arguments[i]) + end + + local text = table.concat(arguments, "\t") + text = " " .. string.gsub(text, "\n", "\n ") + _G.print(text) + end + + local function p(...) + local n = select('#', ...) + local arguments = { ... } + + for i = 1, n do + arguments[i] = dump(arguments[i]) + end + + print(table.concat(arguments, "\t")) + end + + local uv = require('luv') + + local host = os.getenv("HOST") + local port = tonumber(os.getenv("PORT")) + + local socket = uv.new_tcp() + + assert(uv.tcp_connect(socket, host, port, function (err) + p("client connected", {err=err}) + assert(not err, err) + end)) + + -- Start the event loop! + uv.run() + end) + + test("tcp cluster", function (print, p, expect, uv) + + local exepath = assert(uv.exepath()) + local cpu_count = # assert(uv.cpu_info()) + local left = cpu_count + + local server = uv.new_tcp() + assert(uv.tcp_bind(server, "::1", 0)) + + local address = uv.tcp_getsockname(server) + p{server=server,address=address} + + print("Master process bound to TCP port " .. address.port .. " on " .. address.ip) + + local function spawnWorker() + local pipe = uv.new_pipe(true) + local input = uv.new_pipe(false) + local child, pid + child, pid = assert(uv.spawn(exepath, { + cwd = uv.cwd(), + stdio = {input,1,2,pipe}, + env= {"PIPE_FD=3"} + }, expect(function (status, signal) + p("Worker exited", {status=status,signal=signal}) + assert(status == 42, "worker should return 42") + assert(signal == 0) + left = left - 1 + uv.close(child) + uv.close(input) + uv.close(pipe) + if left == 0 then + p("All workers are now dead") + uv.close(server) + end + end))) + p("Spawned worker", pid, "and sending handle", server) + assert(uv.write(input, worker_code)) + assert(uv.write2(pipe, "123", server)) + assert(uv.shutdown(input)) + assert(uv.shutdown(pipe)) + end + + local function spawnClient() + local input = uv.new_pipe(false) + local child, pid + child, pid = assert(uv.spawn(exepath, { + stdio = {input,1,2}, + cwd = uv.cwd(), + env= { + "HOST=" .. address.ip, + "PORT=" .. address.port, + } + }, expect(function (status, signal) + p("Client exited", {status=status,signal=signal}) + assert(status == 0) + assert(signal == 0) + uv.close(child) + end, left))) + p("Spawned client", pid) + assert(uv.write(input, client_code)) + assert(uv.shutdown(input)) + uv.close(input) + end + + -- Spawn a child process for each CPU core + for _ = 1, cpu_count do + spawnWorker() + end + + -- Spawn the clients after a short delay + local timer = uv.new_timer() + uv.timer_start(timer, 1000, 0, expect(function () + for _ = 1, cpu_count do + spawnClient() + end + uv.close(timer) + end)) + + end) +end) + |