Merge pull request #10 from pakozm/devel

Added heap and improved efficiency of merge in reduce step

Author: pakozm

.travis.yml

@@ -6,9 +6,9 @@ before_install:
   - sudo apt-get install -y screen lua5.2 liblua5.2-dev libboost-filesystem-dev libboost-filesystem-dev libboost-thread-dev libssl-dev
   - git clone https://github.com/mongodb/mongo-cxx-driver.git
   - cd mongo-cxx-driver && sudo scons --prefix=/usr install-mongoclient && cd -
-  - wget https://github.com/moai/luamongo/archive/v0.4.1.tar.gz
-  - tar zxvf v0.4.1.tar.gz
-  - cd luamongo-0.4.1 && make && sudo mkdir -p /usr/lib/lua/5.2 && sudo cp mongo.so /usr/lib/lua/5.2 && cd -
+  - wget https://github.com/moai/luamongo/archive/v0.4.2.tar.gz
+  - tar zxvf v0.4.2.tar.gz
+  - cd luamongo-0.4.2 && make && sudo mkdir -p /usr/lib/lua/5.2 && sudo cp mongo.so /usr/lib/lua/5.2 && cd -
   - ssh-keygen -b 2048 -f /home/travis/.ssh/id_rsa -t rsa -q -N ""
   - ssh-keyscan -t rsa localhost > ~/.ssh/known_hosts
   - ssh-keyscan -t rsa $(hostname) >> ~/.ssh/known_hosts

mapreduce/heap.lua

@@ -0,0 +1,120 @@
+--[[
+  This file is part of Lua-MapReduce
+  
+  Copyright 2014, Francisco Zamora-Martinez
+  
+  The Lua-MapReduce toolkit is free software; you can redistribute it and/or modify it
+  under the terms of the GNU General Public License version 3 as
+  published by the Free Software Foundation
+  
+  This library is distributed in the hope that it will be useful, but WITHOUT
+  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
+  for more details.
+  
+  You should have received a copy of the GNU General Public License
+  along with this library; if not, write to the Free Software Foundation,
+  Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+]]
+
+local heap = {
+  _VERSION = "0.1",
+  _NAME = "mapreduce.heap",
+}
+
+local function parent(p) return math.floor(p/2) end
+local function left(p) return 2*p end
+local function right(p) return 2*p+1 end
+
+function heap:top()
+  return self.data[1]
+end
+
+function heap:pop()
+  local data, cmp = self.data, self.cmp
+  local v = data[#data]
+  data[#data] = nil
+  if #data == 0 then return end
+  -- heapify(1)
+  local pos = 1
+  while true do
+    local l,r = left(pos), right(pos)
+    local lv, rv = data[l], data[r]
+    if not lv then break end
+    local child = (rv and cmp(rv,lv) and r) or l
+    if cmp(data[child],v) then
+      data[pos] = data[child]
+      pos = child
+    else
+      break
+    end
+  end
+  data[pos] = v
+end
+
+function heap:push(v)
+  local data, cmp = self.data, self.cmp
+  local pos = #data + 1
+  -- bubble up
+  while pos > 1 do
+    local p  = parent(pos)
+    local pv = data[p]
+    if cmp(v,pv) then
+      data[pos] = data[p]
+      pos = p
+    else
+      break
+    end
+  end
+  data[pos] = v
+end
+
+function heap:clear()
+  self.data = {}
+end
+
+function heap:size()
+  return #self.data
+end
+
+function heap:empty()
+  return #self.data == 0
+end
+
+function heap:__call(cmp)
+  local obj = {
+    cmp  = cmp or function(a,b) return a < b end,
+    data = {},
+  }
+  setmetatable(obj, { __index = self,
+                      __len = function(self) return #self.data end })
+  return obj
+end
+setmetatable(heap, heap)
+
+----------------------------------------------------------------------------
+------------------------------ UNIT TEST -----------------------------------
+----------------------------------------------------------------------------
+
+heap.utest = function()
+  local t = { 20, 10, 15, 1 }
+  local h = heap()
+  assert(h:empty())
+  assert(h:top() == nil)
+  assert(h:size() == 0)
+  h:push(30)
+  assert(h:top() == 30)
+  h:clear()
+  assert(h:empty())
+  assert(h:top() == nil)
+  assert(h:size() == 0)
+  for i=1,#t do h:push(t[i]) end
+  table.sort(t)
+  for i=1,#t do
+    assert(h:top() == t[i])
+    h:pop()
+  end
+  assert(h:top() == nil)
+end
+
+return heap

mapreduce/test.lua

@@ -24,6 +24,7 @@ local task   = require "mapreduce.task"
 local server = require "mapreduce.server"
 local worker = require "mapreduce.worker"
 local persistent_table = require "mapreduce.persistent_table"
+local heap = require "mapreduce.heap"
 
 utils.utest()
 cnn.utest()
@@ -33,5 +34,6 @@ task.utest()
 server.utest()
 worker.utest()
 persistent_table.utest()
+heap.utest()
 
 print("Ok")

mapreduce/utils.lua

@@ -17,6 +17,7 @@
   Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 ]]
 local mongo = require "mongo"
+local heap  = require "mapreduce.heap"
 
 assert(mongo._VERSION == "0.4" or tonumber(mongo._VERSION > 0.4))
 
@@ -203,48 +204,40 @@ local function merge_iterator(fs, filenames, make_lines_iterator)
     line_iterators[ #line_iterators+1 ] = make_lines_iterator(name)
   end
   local finished = false
-  local data = {}
+  -- heap for efficient merge operation
+  local cmp   = function(a,b) return a[1] < b[1] end
+  local queue = heap(cmp)
+  local data  = {}
   -- take the next data of a given file number
   local take_next = function(which)
     if line_iterators[which] then
       local line = line_iterators[which]()
       if line then
-        data[which] = data[which] or {}
-        data[which][3],data[which][1],data[which][2] =
-          line,assert(load(line),
-                      string.format("Impossible to load line '%s' from '%s'",
-                                    line, filenames[which]))()
+        local l,k,v = line,assert(load(line),
+                                  string.format("Impossible to load line '%s' from '%s'",
+                                                line, filenames[which]))()
+        queue:push({ k, v, l, which })
       else
-        data[which] = nil
         line_iterators[which] = nil
       end
-    else
-      data[which] = nil
     end
   end
-  -- we finished when all the data is nil
-  local finished = function()
-    local ret = true
-    for i=1,#filenames do
-      if data[i] ~= nil then ret = false break end
+  -- merge all the data which shares min key
+  local merge_min_keys = function()
+    local top = queue:top()
+    queue:pop()
+    local key    = top[1] -- key
+    local result = top[2] -- value
+    take_next(top[4])
+    while not queue:empty() and key == queue:top()[1] do
+      local aux   = queue:top()
+      local v     = aux[2]
+      local which = aux[4]
+      queue:pop()
+      take_next(which)
+      for j=1,#v do result[ #result+1 ] = v[j] end
     end
-    return ret
-  end
-  -- look for all the data which has the min key
-  local search_min = function()
-    local key
-    local list = {}
-    for i=1,#filenames do
-      if data[i] then
-        local current = data[i][1]
-        if not key or current <= key then
-          if not key or current < key then list = {} end
-          table.insert(list,i)
-          key = current
-        end
-      end
-    end
-    return list
+    return key,result
   end
   -- initialize data with first line over all files
   for i=1,#filenames do take_next(i) end
@@ -256,30 +249,12 @@ local function merge_iterator(fs, filenames, make_lines_iterator)
     local assert       = assert
     local data         = data
     local take_next    = take_next
-    local finished     = finished
-    -- merge all the files until finished
-    while not finished() do
+    local queue        = queue
+    -- merge all the files until finished (empty queue)
+    while not queue:empty() do
       counter = counter + 1
       --
-      local mins_list = search_min()
-      assert(#mins_list > 0)
-      local key = data[mins_list[1]][1]
-      local result
-      if #mins_list == 1 then
-        -- only one secuence of values, nothing to merge
-        result = data[mins_list[1]][2]
-        take_next(mins_list[1])
-      else -- if #mins_list == 1 then ... else
-        result = {}
-        for i=1,#mins_list do
-          local which = mins_list[i]
-          -- sanity check
-          assert(data[which][1] == key)
-          local v = data[which][2]
-          for j=1,#v do result[ #result+1 ] = v[j] end
-          take_next(which)
-        end
-      end -- if #mins_list == 1 then ... else ... end
+      local key,result = merge_min_keys()
       -- verbose output
       if counter % MAX_IT_WO_CGARBAGE == 0 then
         collectgarbage("collect")