From 213c3bd3bee41fd296c7752d077a42a633b241a0 Mon Sep 17 00:00:00 2001 From: Coder12a <38924418+Coder12a@users.noreply.github.com> Date: Sat, 10 Nov 2018 21:40:49 -0600 Subject: [PATCH] Make database asynchronous --- README.md | 57 +++++++++---- async.lua | 209 ++++++++++++++++++++++++++++++++++++++++++++++++ colddb.lua | 153 +++++++++++++++++++---------------- depends.txt | 1 + description.txt | 2 +- init.lua | 4 +- 6 files changed, 336 insertions(+), 90 deletions(-) create mode 100644 async.lua diff --git a/README.md b/README.md index 74f723c..2535620 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ ColdDB =========== -ColdDB is a minetest mod that implements a serverless, NoSQL database engine.
+ColdDB is a minetest mod that implements a serverless, asynchronous, NoSQL database engine.
It provides a key or key-value storage system using plain Lua tables. It also can iterate through the keys.
It is not required to add this mod to secure.trusted_mods this mod will still work. @@ -27,16 +27,26 @@ colddb.set_key(coldbase,"MyKey") ```lua colddb.set(coldbase,"MyKeyAndValue","Hello world") ``` -5. retrieve items (get_key will return true, false, or nil) +5. retrieve items (get_key's callback(arg) will return true, false, or nil) ```lua -local key_and_value = colddb.get(coldbase,"MyKeyAndValue") -local key = colddb.get_key(coldbase,"MyKey") +colddb.get(coldbase,"MyKeyAndValue",nil,function(arg) + if arg then + minetest.log(string.format("value:%s",arg)) + end +end) +colddb.get_key(coldbase,"MyKey",nil,function(arg) + if arg then + minetest.log("Found key") + else + minetest.log("Key not found") + end +end) ``` 6. delete key(file) this function works on both keys and key-value keys. ```lua colddb.remove(coldbase,"MyKeyAndValue") ``` -7. if add_to_mem_pool is true(true by default). keys are stored in a lua table(memory) for 30 seconds or more depending on its use. This is to prevent the database from constantly loading up the data file. +7. if add_to_mem_pool is true(true by default). keys are stored in a weak lua table(memory) it will be removed by the gc if its not in-use. Storing data in memory is to prevent the database from constantly loading up data from files. ```lua coldbase.add_to_mem_pool = true ``` @@ -48,9 +58,9 @@ coldbase.indexes = true ```lua colddb.get_count(coldbase) ``` -10. only if coldbase.indexes == true. iterates through the indexing file 50 times per game tick(breaks and ends if it reached the end of the file). +10. only if coldbase.indexes == true. iterates through the indexing file(breaks and ends if it reached the end of the file). ```lua -colddb.iterate_index_table(coldbase,nil,func_list_keys,nil,50) +colddb.iterate_index_table(coldbase,nil,func_list_keys,nil) ``` 11. adds folders which can be used in other functions that have tag_name arg. ```lua @@ -75,18 +85,14 @@ ip_db = colddb.get_db("watchlist") colddb.add_global_tag(ip_db,"ips") -- return a recorded ip address from the data base. -function ip_db.find(player) - local f = colddb.get(ip_db,player) - if f then - return f - end - return nil +function ip_db.find(player,callback) + colddb.get(ip_db,player,nil,callback) end -- Key is the file and file name. Value is the content's within the file. -- global tag(ips)--->key(Player name)--->value(ip address) function ip_db.record_ip(player,ip) - colddb.set(ip_db,player, ip) + colddb.set(ip_db,player,ip) end function ip_db.delete(player) @@ -97,6 +103,23 @@ end minetest.register_on_prejoinplayer(function(name, ip) ip_db.record_ip(name,ip) end) + +minetest.register_chatcommand("ip", { + params = "", + description = "Get an player's ip address.", + func = function(name, param) + -- Get the ip record asynchronously. + colddb.get(ip_db,param,nil,function(record) + -- If record is contains data send it to the player. + if record then + minetest.chat_send_player(name,string.format("%s:%s",param,record)) + else + -- No record was found. + minetest.chat_send_player(name,"Can not find ip record.") + end + end) + end +}) ``` Quick Look Notes @@ -126,7 +149,7 @@ API - **colddb.get_count(db,tag_name)** - - **colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,cycles_per_tick,args,tag_name)** + - **colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,args,tag_name)** - **begin_func(args)** @@ -138,9 +161,9 @@ API - **colddb.set_key(db,name,tag_name)** - - **colddb.get(db,name,tag_name)** + - **colddb.get(db,name,tag_name,callback(arg))** - - **colddb.get_key(db,name,tag_name)** + - **colddb.get_key(db,name,tag_name,callback(arg))** - **colddb.remove(db,name,tag_name)** diff --git a/async.lua b/async.lua new file mode 100644 index 0000000..bb5682b --- /dev/null +++ b/async.lua @@ -0,0 +1,209 @@ +if not extended_api then + extended_api = {} +end +if not extended_api.Async then + extended_api.Async = {} +end + +function extended_api.Async.create_async_pool() + local pool = {threads = {},globalstep_threads = {},task_queue = {},resting = 200,maxtime = 200,queue_threads = 8,state = "suspended"} + return pool +end + +function extended_api.Async.create_worker(pool,func) + local thread = coroutine.create(func) + table.insert(pool.threads, thread) +end + +function extended_api.Async.create_globalstep_worker(pool,func) + local thread = coroutine.create(func) + table.insert(pool.globalstep_threads, thread) +end + +function extended_api.Async.run_worker(pool,index) + local thread = pool.threads[index] + if thread == nil or coroutine.status(thread) == "dead" then + table.remove(pool.threads, index) + minetest.after(0,extended_api.Async.schedule_worker,pool) + return false + else + coroutine.resume(thread) + minetest.after(0,extended_api.Async.schedule_worker,pool) + return true + end +end + +function extended_api.Async.run_globalstep_worker(pool,index) + local thread = pool.globalstep_threads[index] + if thread == nil or coroutine.status(thread) == "dead" then + table.remove(pool.globalstep_threads, index) + minetest.after(0,extended_api.Async.schedule_globalstep_worker,pool) + return false + else + coroutine.resume(thread) + minetest.after(0,extended_api.Async.schedule_globalstep_worker,pool) + return true + end +end + +function extended_api.Async.schedule_worker(pool) + pool.state = "running" + for index,value in ipairs(pool.threads) do + minetest.after(pool.resting / 1000,extended_api.Async.run_worker,pool,index) + return true + end + pool.state = "suspended" + return false +end + +function extended_api.Async.schedule_globalstep_worker(pool) + for index,value in ipairs(pool.globalstep_threads) do + minetest.after(0,extended_api.Async.run_globalstep_worker,pool,index) + return true + end + return false +end + +function extended_api.Async.priority(pool,resting,maxtime) + pool.resting = resting + pool.maxtime = maxtime +end + +function extended_api.Async.iterate(pool,from,to,func,callback) + extended_api.Async.create_worker(pool,function() + local last_time = minetest.get_us_time() * 1000 + local maxtime = pool.maxtime + for i = from, to do + local b = func(i) + if b and b == false then + break + end + if minetest.get_us_time() * 1000 > last_time + maxtime then + coroutine.yield() + last_time = minetest.get_us_time() * 1000 + end + end + if callback then + callback() + end + end) + extended_api.Async.schedule_worker(pool) +end + +function extended_api.Async.foreach(pool,array, func, callback) + extended_api.Async.create_worker(pool,function() + local last_time = minetest.get_us_time() * 1000 + local maxtime = pool.maxtime + for k,v in ipairs(array) do + local b = func(k,v) + if b and b == false then + break + end + if minetest.get_us_time() * 1000 > last_time + maxtime then + coroutine.yield() + last_time = minetest.get_us_time() * 1000 + end + end + if callback then + callback() + end + end) + extended_api.Async.schedule_worker(pool) +end + +function extended_api.Async.do_while(pool,condition_func, func, callback) + extended_api.Async.create_worker(pool,function() + local last_time = minetest.get_us_time() * 1000 + local maxtime = pool.maxtime + while(condition_func()) do + local c = func() + if c and c ~= condition_func() then + break + end + if minetest.get_us_time() * 1000 > last_time + maxtime then + coroutine.yield() + last_time = minetest.get_us_time() * 1000 + end + end + if callback then + callback() + end + end) + extended_api.Async.schedule_worker(pool) +end + +function extended_api.Async.register_globalstep(pool,func) + extended_api.Async.create_globalstep_worker(pool,function() + local last_time = minetest.get_us_time() * 1000 + local dtime = last_time + while(true) do + local c = func(dtime) + if c and c == false then + break + end + dtime = minetest.get_us_time() * 1000 + -- 0.05 seconds + if minetest.get_us_time() * 1000 > last_time + 50 then + coroutine.yield() + local last_time = minetest.get_us_time() * 1000 + end + end + end) + extended_api.Async.schedule_globalstep_worker(pool) +end + +function extended_api.Async.chain_task(pool,tasks,callback) + extended_api.Async.create_worker(pool,function() + local pass_arg = nil + local last_time = minetest.get_us_time() * 1000 + local maxtime = pool.maxtime + for index, task_func in pairs(tasks) do + local p = task_func(pass_arg) + if p then + pass_arg = p + end + if minetest.get_us_time() * 1000 > last_time + maxtime then + coroutine.yield() + last_time = minetest.get_us_time() * 1000 + end + end + if callback then + callback(pass_arg) + end + end) + extended_api.Async.schedule_worker(pool) +end + +function extended_api.Async.queue_task(pool,func,callback) + table.insert(pool.task_queue,{func = func,callback = callback}) + if pool.queue_threads > 0 then + pool.queue_threads = pool.queue_threads - 1 + extended_api.Async.create_worker(pool,function() + local pass_arg = nil + local last_time = minetest.get_us_time() * 1000 + local maxtime = pool.maxtime + while(true) do + local task_func = pool.task_queue[1] + table.remove(pool.task_queue,1) + if task_func and task_func.func then + pass_arg = nil + local p = task_func.func(pass_arg) + if p then + pass_arg = p + end + if task_func.callback then + task_func.callback(pass_arg) + end + if minetest.get_us_time() * 1000 > last_time + maxtime then + coroutine.yield() + last_time = minetest.get_us_time() * 1000 + end + else + pool.queue_threads = pool.queue_threads + 1 + break + end + end + end) + extended_api.Async.schedule_worker(pool) + end +end \ No newline at end of file diff --git a/colddb.lua b/colddb.lua index d842d3a..07d92a8 100644 --- a/colddb.lua +++ b/colddb.lua @@ -33,7 +33,9 @@ function colddb.get_db(directory) iterate_queue = {}, indexes = false, add_to_mem_pool = true, + async_pool = extended_api.Async.create_async_pool(), } + extended_api.Async.priority(db.async_pool,150,250) -- make tables weak so the garbage-collector will remove unused data setmetatable(db.tags, {__mode = "kv"}) setmetatable(db.mem_pool, {__mode = "kv"}) @@ -250,7 +252,7 @@ function colddb.delete_lines(db,_lines,tag_name) local copyfile = string.format("%s%sæIndex_table.cold.replacer",db.directory,t) local args = {db=db,cs=cs,oldfile=oldfile,copyfile=copyfile,do_not_skip_removed_items=true} db.indexes_pool[cs] = f - colddb.iterate_index_table(db,delete_lines_func_begin,delete_lines_func_i,delete_lines_func_end,150,args,tag_name) + colddb.iterate_index_table(db,delete_lines_func_begin,delete_lines_func_i,delete_lines_func_end,args,tag_name) end end @@ -374,21 +376,15 @@ function colddb.get_count(db,tag_name) return nil end -local function iterate(db,func_on_iterate,end_func,cycles_per_tick,count,cs,args) +local function iterate(db,func_on_iterate,end_func,count,cs,args) local f = db.indexes_pool[cs] local fl = f.file - for i=1,cycles_per_tick do - if count < 1 then - break - end + extended_api.Async.iterate(db.async_pool,1,count,function(i) local line = fl:read("*l") if args.do_not_skip_removed_items or not db.indexes_pool[cs].deleted_items[line] then func_on_iterate(line,i,args) end - count = count - 1 - end - db.indexes_pool[cs] = f - if count < 1 then + end,function() if end_func then end_func(args) end @@ -405,19 +401,18 @@ local function iterate(db,func_on_iterate,end_func,cycles_per_tick,count,cs,args copy.args = a end end - minetest.after(0,iterate,copy.db,copy.func_on_iterate,copy.end_func,copy.cycles_per_tick,copy.count,copy.cs,copy.args) + minetest.after(0,iterate,copy.db,copy.func_on_iterate,copy.end_func,copy.count,copy.cs,copy.args) table.remove(db.iterate_queue[cs],1) - return true + return false else db.iterate_queue[cs] = nil end - db.indexes_pool[cs].iterating = false - return true - end - minetest.after(0,iterate,db,func_on_iterate,end_func,cycles_per_tick,count,cs,args) + db.indexes_pool[cs].iterating = false + return false + end) end -function colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,cycles_per_tick,args,tag_name) +function colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,args,tag_name) local t = "" local name = "æIndex_table" if tag_name then @@ -455,7 +450,7 @@ function colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,cycle end end -- Start iterating the index table - iterate(db,func_on_iterate,end_func,cycles_per_tick,c,cs,args) + iterate(db,func_on_iterate,end_func,c,cs,args) elseif f and f.file then -- If its iterating some other function then add this one to the queue list fl:seek("set") @@ -467,7 +462,7 @@ function colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,cycle if not db.iterate_queue[cs] then db.iterate_queue[cs] = {} end - local _table = {db=db,begin_func=begin_func,func_on_iterate=func_on_iterate,end_func=end_func,cycles_per_tick=cycles_per_tick,count=c,cs=cs,tag_name=tag_name,args=args} + local _table = {db=db,begin_func=begin_func,func_on_iterate=func_on_iterate,end_func=end_func,count=c,cs=cs,tag_name=tag_name,args=args} table.insert(db.iterate_queue[cs],_table) end end @@ -494,14 +489,18 @@ function colddb.set(db,name,_table,tag_name) t = colddb.get_tag(db,tag_name) end if db.indexes and not colddb.file_Exists(db,name,tag_name) then - local cs2 = string.format("%s%s",t,"æIndex_table") - local om = db.indexes_pool[cs2] - if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then - colddb.open_index_table(db,tag_name) - end - minetest.after(0,function(db,name,tag_name)colddb.append_index_table(db,name,tag_name)end,db,name,tag_name) + extended_api.Async.queue_task(db.async_pool,function() + local cs2 = string.format("%s%s",t,"æIndex_table") + local om = db.indexes_pool[cs2] + if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then + colddb.open_index_table(db,tag_name) + end + colddb.append_index_table(db,name,tag_name) + end) end - minetest.after(1,function(db,name, _table,tag_name)colddb.save_table(db,name, _table,tag_name)end,db,name, _table,tag_name) + extended_api.Async.queue_task(db.async_pool,function() + colddb.save_table(db,name, _table,tag_name) + end) if db.add_to_mem_pool then load_into_mem(db,name,_table,tag_name) end @@ -513,55 +512,63 @@ function colddb.set_key(db,name,tag_name) t = colddb.get_tag(db,tag_name) end if db.indexes and not colddb.file_Exists(db,name,tag_name) then - local cs2 = string.format("%s%s",t,"æIndex_table") - local om = db.indexes_pool[cs2] - if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then - colddb.open_index_table(db,tag_name) - end - minetest.after(0,function(db,name,tag_name)colddb.append_index_table(db,name,tag_name)end,db,name,tag_name) + extended_api.Async.queue_task(db.async_pool,function() + local cs2 = string.format("%s%s",t,"æIndex_table") + local om = db.indexes_pool[cs2] + if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then + colddb.open_index_table(db,tag_name) + end + colddb.append_index_table(db,name,tag_name) + end) end - minetest.after(1,function(db,name, tag_name)colddb.save_key(db,name, tag_name)end,db,name, tag_name) + extended_api.Async.queue_task(db.async_pool,function() + colddb.save_key(db,name, tag_name) + end) if db.add_to_mem_pool then load_into_mem(db,name,"",tag_name) end end -function colddb.get(db,name,tag_name) - local t = "" - if tag_name then - t = colddb.get_tag(db,tag_name) - end - local cs = string.format("%s%s",t,name) - local pm = db.mem_pool[cs] - if pm then - return pm.mem - else - local _table = colddb.load_table(db,name,tag_name) - if _table then - load_into_mem(db,name,_table,tag_name) - return _table +function colddb.get(db,name,tag_name,callback) + extended_api.Async.queue_task(db.async_pool,function() + local t = "" + if tag_name then + t = colddb.get_tag(db,tag_name) end - end - return nil + local cs = string.format("%s%s",t,name) + local pm = db.mem_pool[cs] + if pm then + return pm.mem + else + local _table = colddb.load_table(db,name,tag_name) + if _table then + load_into_mem(db,name,_table,tag_name) + return _table + end + end + return nil + end,callback) end -function colddb.get_key(db,name,tag_name) - local t = "" - if tag_name then - t = colddb.get_tag(db,tag_name) - end - local cs = string.format("%s%s",t,name) - local pm = db.mem_pool[cs] - if pm then - return true - else - local bool = colddb.load_key(db,name,tag_name) - if bool then - load_into_mem(db,name,bool,tag_name) - return bool +function colddb.get_key(db,name,tag_name,callback) + extended_api.Async.queue_task(db.async_pool,function() + local t = "" + if tag_name then + t = colddb.get_tag(db,tag_name) end - end - return nil + local cs = string.format("%s%s",t,name) + local pm = db.mem_pool[cs] + if pm then + return true + else + local bool = colddb.load_key(db,name,tag_name) + if bool then + load_into_mem(db,name,bool,tag_name) + return bool + end + end + return nil + end,callback) end function colddb.remove(db,name,tag_name) @@ -574,11 +581,15 @@ function colddb.remove(db,name,tag_name) db.mem_pool[cs] = nil end if db.indexes and colddb.file_Exists(db,"æIndex_table",tag_name) then - local cs2 = string.format("%s%s",t,"æIndex_table") - if not (db.indexes_pool[cs2] and db.indexes_pool[cs2].file) then - colddb.open_index_table(db,tag_name) - end - colddb.delete_lines(db,name,tag_name) + extended_api.Async.queue_task(db.async_pool,function() + local cs2 = string.format("%s%s",t,"æIndex_table") + if not (db.indexes_pool[cs2] and db.indexes_pool[cs2].file) then + colddb.open_index_table(db,tag_name) + end + colddb.delete_lines(db,name,tag_name) + end) end - minetest.after(0,function(db,name,tag_name)colddb.delete_file(db,name,tag_name)end,db,name,tag_name) + extended_api.Async.queue_task(db.async_pool,function() + colddb.delete_file(db,name,tag_name) + end) end \ No newline at end of file diff --git a/depends.txt b/depends.txt index e69de29..7221d40 100644 --- a/depends.txt +++ b/depends.txt @@ -0,0 +1 @@ +extended_api? \ No newline at end of file diff --git a/description.txt b/description.txt index 9934151..542d095 100644 --- a/description.txt +++ b/description.txt @@ -1 +1 @@ -pure lua No-sql database mod. \ No newline at end of file +pure lua No-sql Asynchronous database mod. \ No newline at end of file diff --git a/init.lua b/init.lua index f9cee77..e795c5e 100644 --- a/init.lua +++ b/init.lua @@ -1,2 +1,4 @@ modpath = minetest.get_modpath("colddb") -dofile(modpath .. "/colddb.lua") + +dofile(string.format("%s/async.lua",modpath)) +dofile(string.format("%s/colddb.lua",modpath)) \ No newline at end of file