Update multi-uv example.

Author: moteus

examples/cURLv3/multi-uv.lua

@@ -1,12 +1,15 @@
 local curl = require "cURL"
 local uv   = require "lluv"
+local ut   = require "lluv.utils"
 
 local fprintf = function(f, ...) f:write((string.format(...))) end
 local printf  = function(...) fprintf(io.stdout, ...) end
 
 local stderr = io.stderr
 
-local timeout, curl_handle
+local trace = false
+
+trace = trace and print or function() end
 
 local ACTION_NAMES = {
   [curl.POLL_IN     ] = "POLL_IN";
@@ -15,141 +18,218 @@ local ACTION_NAMES = {
   [curl.POLL_NONE   ] = "POLL_NONE";
   [curl.POLL_REMOVE ] = "POLL_REMOVE";
 }
+
 local EVENT_NAMES = {
   [ uv.READABLE ] = "READABLE";
   [ uv.WRITABLE ] = "WRITABLE";
   [ uv.READABLE + uv.WRITABLE ] = "READABLE + WRITABLE";
 }
+
 local FLAGS = {
-  [ uv.READABLE ]               = curl.CSELECT_IN;
-  [ uv.WRITABLE ]               = curl.CSELECT_OUT;
+  [ uv.READABLE               ] = curl.CSELECT_IN;
+  [ uv.WRITABLE               ] = curl.CSELECT_OUT;
   [ uv.READABLE + uv.WRITABLE ] = curl.CSELECT_IN + curl.CSELECT_OUT;
-
 }
 
-local trace = true
+local Context = ut.class() do
 
-trace = trace and print or function() end
+function Context:__init(fd)
+  self._fd   = assert(fd)
+  self._poll = uv.poll_socket(fd)
+  self._poll.data = {context = self}
 
-local CONTEXT = {}
+  assert(self._poll:fileno() == fd)
 
-function create_curl_context(sockfd)
-  local context = {
-    sockfd      = sockfd;
-    poll_handle = uv.poll_socket(sockfd);
-  }
-  context.poll_handle.data = context
-  
-  return context
+  return self
+end
+
+function Context:close()
+  if not self._poll then return end
+  self._poll.data = nil
+  self._poll:close()
+  self._poll, self._fd = nil
+end
+
+function Context:poll(...)
+  self._poll:start(...)
+end
+
+function Context:fileno()
+  return self._fd
+end
+
+end
+
+-- Number of parallel request
+local MAX_REQUESTS
+local timer, multi
+local qtask = ut.Queue.new() -- wait tasks
+local qfree = ut.Queue.new() -- avaliable easy handles
+local qeasy = {} -- all easy handles
+
+local function cleanup()
+  timer:close()
+
+  for i, easy in ipairs(qeasy) do
+    multi:remove_handle(easy)
+    easy:close()
+  end
+
+  multi:close()
 end
 
-function destroy_curl_context(context)
-  context.poll_handle:close()
-end 
+local proceed_queue, add_download do
+
+proceed_queue = function()
+  if qtask:empty() then return end
+
+  if qfree:empty() then
+    if #qeasy < MAX_REQUESTS then
+      local easy = assert(curl.easy())
+      qeasy[#qeasy + 1] = easy
+      qfree:push(easy)
+    else
+      return
+    end
+  end
+
+  local task = assert(qtask:pop())
+  local url, num = task[1], task[2]
 
-function add_download(url, num)
   local filename = tostring(num) ..  ".download"
   local file = io.open(filename, "w")
   if not file then
     fprintf(stderr, "Error opening %s\n", filename)
     return
   end
 
-  local handle = curl.easy{
+  local handle = assert(qfree:pop())
+
+  handle:setopt{
     url           = url;
     writefunction = file;
   }
 
-  handle.data = file
+  handle.data = { file = file }
+
+  multi:add_handle(handle)
 
-  curl_handle:add_handle(handle)
   fprintf(stderr, "Added download %s -> %s\n", url, filename);
 end
 
-function check_multi_info()
+add_download = function(url, num)
+  qtask:push{url, num}
+
+  proceed_queue()
+end
+
+end
+
+local on_libuv_poll, on_libuv_timeout
+
+local on_curl_timeout, on_curl_action do
+
+on_curl_timeout = function(ms)
+  -- calls by curl --
+  trace("CURL::TIMEOUT", ms)
+
+  if ms <= 0 then ms = 1 end
+
+  timer:start(ms, 0, on_libuv_timeout)
+end
+
+on_curl_action = function(easy, fd, action)
+  local ok, err = pcall(function()
+    trace("CURL::SOCKET", easy, s, ACTION_NAMES[action] or action)
+
+    local context = easy.data.context
+    if (action == curl.POLL_IN) or (action == curl.POLL_OUT) then
+      if not context then
+        context = Context.new(fd)
+        easy.data.context = context
+      end
+    end
+
+    assert(context:fileno() == fd)
+
+    if     action == curl.POLL_IN  then  context:poll(uv.READABLE, on_libuv_poll)
+    elseif action == curl.POLL_OUT then  context:poll(uv.WRITABLE, on_libuv_poll)
+    elseif action == curl.POLL_REMOVE then
+      if context then
+        easy.data.context = nil
+        context:close()
+      end
+    end
+  end)
+
+  if not ok then uv.defer(function() error(err) end) end
+end
+
+end
+
+-- on_libuv_poll, on_libuv_timeout 
+do
+
+local curl_check_multi_info = function()
   while true do
-    local easy, ok, err = curl_handle:info_read(true)
-    if not easy then curl_handle:close() error(err) end
+    local easy, ok, err = multi:info_read(true)
+
+    if not easy then
+      multi:close()
+      error(err)
+    end
+
     if easy == 0 then break end
 
-    local context = CONTEXT[e]
-    if context then destroy_curl_context(context) end
-    local file = assert(easy.data)
-    file:close()
     local done_url = easy:getinfo_effective_url()
-    easy:close()
+
+    local context = easy.data.context
+    if context then context:close() end
+    local file    = assert(easy.data.file)
+    file:close()
+
+    easy.data = nil
+    qfree:push(easy)
+
     if ok then
       printf("%s DONE\n", done_url);
     elseif data == "error" then
       printf("%s ERROR - %s\n", done_url, tostring(err));
     end
+
+    proceed_queue()
   end
 end
 
-function curl_perform(handle, err, events)
-  -- calls by libuv --
+on_libuv_poll = function(handle, err, events)
   trace("UV::POLL", handle, err, EVENT_NAMES[events] or events)
 
   local flags = assert(FLAGS[events], ("unknown event:" .. events))
 
-  context = handle.data
+  context = handle.data.context
 
-  curl_handle:socket_action(context.sockfd, flags)
+  multi:socket_action(context:fileno(), flags)
 
-  check_multi_info()
+  curl_check_multi_info()
 end
 
-function on_timeout(timer)
-  -- calls by libuv --
+on_libuv_timeout = function(timer)
   trace("UV::TIMEOUT", timer)
 
-  local running_handles, err = curl_handle:socket_action()
-
-  check_multi_info()
-end
-
-function start_timeout(timeout_ms)
-  -- calls by curl --
-  trace("CURL::TIMEOUT", timeout_ms)
+  local running_handles, err = multi:socket_action()
 
-  -- 0 means directly call socket_action, but we'll do it in a bit
-  if timeout_ms <= 0 then timeout_ms = 1 end
-
-  timeout:stop():start(timeout_ms, 0, on_timeout)
+  curl_check_multi_info()
 end
 
-function handle_socket(easy, s, action)
-  local ok, err = pcall(function()
-    -- calls by curl --
-    trace("CURL::SOCKET", easy, s, ACTION_NAMES[action] or action)
-
-    local curl_context = CONTEXT[easy] or create_curl_context(s)
-    CONTEXT[easy] = curl_context
-
-    assert(curl_context.sockfd == s)
-
-    if action == curl.POLL_IN then
-      curl_context.poll_handle:start(uv.READABLE, curl_perform)
-    elseif action == curl.POLL_OUT then
-      curl_context.poll_handle:start(uv.WRITABLE, curl_perform)
-    elseif action == curl.POLL_REMOVE then
-      CONTEXT[easy] = nil
-      destroy_curl_context(curl_context)
-    end
-  end)
-  if not ok then uv.defer(function() error(err) end) end
 end
 
-timeout = uv.timer()
+MAX_REQUESTS = 64
 
-curl_handle = curl.multi{
-  socketfunction = handle_socket;
-  timerfunction  = start_timeout;
-}
+timer = uv.timer()
 
-curl_handle = curl.multi{
-  socketfunction = handle_socket;
-  timerfunction  = start_timeout;
+multi = curl.multi{
+  timerfunction = on_curl_timeout;
+  socketfunction = on_curl_action;
 }
 
 for i = 1, math.huge do
@@ -158,4 +238,6 @@ for i = 1, math.huge do
   add_download(url, i)
 end
 
-uv.run(loop, UV_RUN_DEFAULT)
+uv.run()
+
+cleanup()