Make database asynchronous

This commit is contained in:
Coder12a 2018-11-10 21:40:49 -06:00
parent ab7337ba83
commit 213c3bd3be
6 changed files with 336 additions and 90 deletions

View File

@ -1,7 +1,7 @@
ColdDB ColdDB
=========== ===========
ColdDB is a minetest mod that implements a serverless, NoSQL database engine.<br> ColdDB is a minetest mod that implements a serverless, asynchronous, NoSQL database engine.<br>
It provides a key or key-value storage system using plain Lua tables. It also can iterate through the keys.<br> It provides a key or key-value storage system using plain Lua tables. It also can iterate through the keys.<br>
It is not required to add this mod to secure.trusted_mods this mod will still work. 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 ```lua
colddb.set(coldbase,"MyKeyAndValue","Hello world") 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 ```lua
local key_and_value = colddb.get(coldbase,"MyKeyAndValue") colddb.get(coldbase,"MyKeyAndValue",nil,function(arg)
local key = colddb.get_key(coldbase,"MyKey") 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. 6. delete key(file) this function works on both keys and key-value keys.
```lua ```lua
colddb.remove(coldbase,"MyKeyAndValue") 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 ```lua
coldbase.add_to_mem_pool = true coldbase.add_to_mem_pool = true
``` ```
@ -48,9 +58,9 @@ coldbase.indexes = true
```lua ```lua
colddb.get_count(coldbase) 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 ```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. 11. adds folders which can be used in other functions that have tag_name arg.
```lua ```lua
@ -75,18 +85,14 @@ ip_db = colddb.get_db("watchlist")
colddb.add_global_tag(ip_db,"ips") colddb.add_global_tag(ip_db,"ips")
-- return a recorded ip address from the data base. -- return a recorded ip address from the data base.
function ip_db.find(player) function ip_db.find(player,callback)
local f = colddb.get(ip_db,player) colddb.get(ip_db,player,nil,callback)
if f then
return f
end
return nil
end end
-- Key is the file and file name. Value is the content's within the file. -- Key is the file and file name. Value is the content's within the file.
-- global tag(ips)--->key(Player name)--->value(ip address) -- global tag(ips)--->key(Player name)--->value(ip address)
function ip_db.record_ip(player,ip) function ip_db.record_ip(player,ip)
colddb.set(ip_db,player, ip) colddb.set(ip_db,player,ip)
end end
function ip_db.delete(player) function ip_db.delete(player)
@ -97,6 +103,23 @@ end
minetest.register_on_prejoinplayer(function(name, ip) minetest.register_on_prejoinplayer(function(name, ip)
ip_db.record_ip(name,ip) ip_db.record_ip(name,ip)
end) end)
minetest.register_chatcommand("ip", {
params = "<player>",
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 Quick Look Notes
@ -126,7 +149,7 @@ API
- **colddb.get_count(db,tag_name)** - **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)** - **begin_func(args)**
@ -138,9 +161,9 @@ API
- **colddb.set_key(db,name,tag_name)** - **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)** - **colddb.remove(db,name,tag_name)**

209
async.lua Normal file
View File

@ -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

View File

@ -33,7 +33,9 @@ function colddb.get_db(directory)
iterate_queue = {}, iterate_queue = {},
indexes = false, indexes = false,
add_to_mem_pool = true, 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 -- make tables weak so the garbage-collector will remove unused data
setmetatable(db.tags, {__mode = "kv"}) setmetatable(db.tags, {__mode = "kv"})
setmetatable(db.mem_pool, {__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 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} local args = {db=db,cs=cs,oldfile=oldfile,copyfile=copyfile,do_not_skip_removed_items=true}
db.indexes_pool[cs] = f 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
end end
@ -374,21 +376,15 @@ function colddb.get_count(db,tag_name)
return nil return nil
end 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 f = db.indexes_pool[cs]
local fl = f.file local fl = f.file
for i=1,cycles_per_tick do extended_api.Async.iterate(db.async_pool,1,count,function(i)
if count < 1 then
break
end
local line = fl:read("*l") local line = fl:read("*l")
if args.do_not_skip_removed_items or not db.indexes_pool[cs].deleted_items[line] then if args.do_not_skip_removed_items or not db.indexes_pool[cs].deleted_items[line] then
func_on_iterate(line,i,args) func_on_iterate(line,i,args)
end end
count = count - 1 end,function()
end
db.indexes_pool[cs] = f
if count < 1 then
if end_func then if end_func then
end_func(args) end_func(args)
end end
@ -405,19 +401,18 @@ local function iterate(db,func_on_iterate,end_func,cycles_per_tick,count,cs,args
copy.args = a copy.args = a
end end
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) table.remove(db.iterate_queue[cs],1)
return true return false
else else
db.iterate_queue[cs] = nil db.iterate_queue[cs] = nil
end end
db.indexes_pool[cs].iterating = false db.indexes_pool[cs].iterating = false
return true return false
end end)
minetest.after(0,iterate,db,func_on_iterate,end_func,cycles_per_tick,count,cs,args)
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 t = ""
local name = "æIndex_table" local name = "æIndex_table"
if tag_name then if tag_name then
@ -455,7 +450,7 @@ function colddb.iterate_index_table(db,begin_func,func_on_iterate,end_func,cycle
end end
end end
-- Start iterating the index table -- 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 elseif f and f.file then
-- If its iterating some other function then add this one to the queue list -- If its iterating some other function then add this one to the queue list
fl:seek("set") 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 if not db.iterate_queue[cs] then
db.iterate_queue[cs] = {} db.iterate_queue[cs] = {}
end 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) table.insert(db.iterate_queue[cs],_table)
end end
end end
@ -494,14 +489,18 @@ function colddb.set(db,name,_table,tag_name)
t = colddb.get_tag(db,tag_name) t = colddb.get_tag(db,tag_name)
end end
if db.indexes and not colddb.file_Exists(db,name,tag_name) then if db.indexes and not colddb.file_Exists(db,name,tag_name) then
extended_api.Async.queue_task(db.async_pool,function()
local cs2 = string.format("%s%s",t,"æIndex_table") local cs2 = string.format("%s%s",t,"æIndex_table")
local om = db.indexes_pool[cs2] local om = db.indexes_pool[cs2]
if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then
colddb.open_index_table(db,tag_name) colddb.open_index_table(db,tag_name)
end end
minetest.after(0,function(db,name,tag_name)colddb.append_index_table(db,name,tag_name)end,db,name,tag_name) colddb.append_index_table(db,name,tag_name)
end)
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 if db.add_to_mem_pool then
load_into_mem(db,name,_table,tag_name) load_into_mem(db,name,_table,tag_name)
end end
@ -513,20 +512,25 @@ function colddb.set_key(db,name,tag_name)
t = colddb.get_tag(db,tag_name) t = colddb.get_tag(db,tag_name)
end end
if db.indexes and not colddb.file_Exists(db,name,tag_name) then if db.indexes and not colddb.file_Exists(db,name,tag_name) then
extended_api.Async.queue_task(db.async_pool,function()
local cs2 = string.format("%s%s",t,"æIndex_table") local cs2 = string.format("%s%s",t,"æIndex_table")
local om = db.indexes_pool[cs2] local om = db.indexes_pool[cs2]
if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then if not colddb.file_Exists(db,"æIndex_table",tag_name) or not (om and om.file) then
colddb.open_index_table(db,tag_name) colddb.open_index_table(db,tag_name)
end end
minetest.after(0,function(db,name,tag_name)colddb.append_index_table(db,name,tag_name)end,db,name,tag_name) colddb.append_index_table(db,name,tag_name)
end)
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 if db.add_to_mem_pool then
load_into_mem(db,name,"",tag_name) load_into_mem(db,name,"",tag_name)
end end
end end
function colddb.get(db,name,tag_name) function colddb.get(db,name,tag_name,callback)
extended_api.Async.queue_task(db.async_pool,function()
local t = "" local t = ""
if tag_name then if tag_name then
t = colddb.get_tag(db,tag_name) t = colddb.get_tag(db,tag_name)
@ -543,9 +547,11 @@ function colddb.get(db,name,tag_name)
end end
end end
return nil return nil
end,callback)
end end
function colddb.get_key(db,name,tag_name) function colddb.get_key(db,name,tag_name,callback)
extended_api.Async.queue_task(db.async_pool,function()
local t = "" local t = ""
if tag_name then if tag_name then
t = colddb.get_tag(db,tag_name) t = colddb.get_tag(db,tag_name)
@ -562,6 +568,7 @@ function colddb.get_key(db,name,tag_name)
end end
end end
return nil return nil
end,callback)
end end
function colddb.remove(db,name,tag_name) function colddb.remove(db,name,tag_name)
@ -574,11 +581,15 @@ function colddb.remove(db,name,tag_name)
db.mem_pool[cs] = nil db.mem_pool[cs] = nil
end end
if db.indexes and colddb.file_Exists(db,"æIndex_table",tag_name) then if db.indexes and colddb.file_Exists(db,"æIndex_table",tag_name) then
extended_api.Async.queue_task(db.async_pool,function()
local cs2 = string.format("%s%s",t,"æIndex_table") local cs2 = string.format("%s%s",t,"æIndex_table")
if not (db.indexes_pool[cs2] and db.indexes_pool[cs2].file) then if not (db.indexes_pool[cs2] and db.indexes_pool[cs2].file) then
colddb.open_index_table(db,tag_name) colddb.open_index_table(db,tag_name)
end end
colddb.delete_lines(db,name,tag_name) colddb.delete_lines(db,name,tag_name)
end)
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 end

View File

@ -0,0 +1 @@
extended_api?

View File

@ -1 +1 @@
pure lua No-sql database mod. pure lua No-sql Asynchronous database mod.

View File

@ -1,2 +1,4 @@
modpath = minetest.get_modpath("colddb") modpath = minetest.get_modpath("colddb")
dofile(modpath .. "/colddb.lua")
dofile(string.format("%s/async.lua",modpath))
dofile(string.format("%s/colddb.lua",modpath))