lanes_example.lua for parallel processing

lanes_example.lua

-- Specify path to Lua Lanes module
-- Install via: 'luarocks install lanes'
local lanes_path = "/usr/local/share/lua/5.1/lanes.lua"

-- Load Lanes lib
local lanes = assert(loadfile(lanes_path))().configure()

-- Lua function that will be executed as separate threads
local function process_log(timestamp, record)
    -- Add your CPU intensive code here
    print("Timestamp:", timestamp.sec, timestamp.nsec)
    print("Record:", record.message)

    record.message = "Modified"
    return timestamp, record
end


-- Entry function
function threads(tag, records)
  print("LUA ")
  local thread_handles = {}
  local results = {}
  if records and type(records) == "table" then

    print("Number of incoming records:", #records)

    for i, log_event in ipairs(records) do
        local timestamp = log_event.timestamp
        local record = log_event.record

          -- Use lanes.gen to create a new thread
          local thread = lanes.gen("*", process_log)(timestamp, record)

          -- Store the thread handle
          table.insert(thread_handles, thread)

    end
    -- Wait for all threads to finish
    for _, thread in ipairs(thread_handles) do
        -- Get the result returned by each thread
        local modified_record = thread[2]
        local modified_timestamp = thread[1]
        local result = {timestamp = modified_timestamp, record = modified_record}
        table.insert(results, result)
    end
    print("All threads returned")
  else
    print("Error: Invalid or nil 'records' table.")
  end

  return results
end