diff options
Diffstat (limited to 'util/datamanager.lua')
| -rw-r--r-- | util/datamanager.lua | 324 |
1 files changed, 320 insertions, 4 deletions
diff --git a/util/datamanager.lua b/util/datamanager.lua index c57f4a0e..4269e5d1 100644 --- a/util/datamanager.lua +++ b/util/datamanager.lua @@ -7,35 +7,42 @@ -- +local string = string; local format = string.format; local setmetatable = setmetatable; local ipairs = ipairs; local char = string.char; local pcall = pcall; -local log = require "util.logger".init("datamanager"); +local log = require "prosody.util.logger".init("datamanager"); local io_open = io.open; local os_remove = os.remove; local os_rename = os.rename; local tonumber = tonumber; +local floor = math.floor; local next = next; local type = type; local t_insert = table.insert; local t_concat = table.concat; -local envloadfile = require"util.envload".envloadfile; -local serialize = require "util.serialization".serialize; +local envloadfile = require"prosody.util.envload".envloadfile; +local envload = require"prosody.util.envload".envload; +local serialize = require "prosody.util.serialization".serialize; local lfs = require "lfs"; -- Extract directory separator from package.config (an undocumented string that comes with lua) local path_separator = assert ( package.config:match ( "^([^\n]+)" ) , "package.config not in standard form" ) local prosody = prosody; +--luacheck: ignore 211/blocksize 211/remove_blocks +local blocksize = 0x1000; local raw_mkdir = lfs.mkdir; local atomic_append; +local remove_blocks; local ENOENT = 2; pcall(function() - local pposix = require "util.pposix"; + local pposix = require "prosody.util.pposix"; raw_mkdir = pposix.mkdir or raw_mkdir; -- Doesn't trample on umask atomic_append = pposix.atomic_append; + -- remove_blocks = pposix.remove_blocks; ENOENT = pposix.ENOENT or ENOENT; end); @@ -239,6 +246,14 @@ local function append(username, host, datastore, ext, data) end local pos = f:seek("end"); + --[[ TODO needs tests + if (blocksize-(pos%blocksize)) < (#data%blocksize) then + -- pad to blocksize with newlines so that the next item is both on a new + -- block and a new line + atomic_append(f, ("\n"):rep(blocksize-(pos%blocksize))); + pos = f:seek("end"); + end + --]] local ok, msg = atomic_append(f, data); @@ -255,6 +270,13 @@ local function append(username, host, datastore, ext, data) return true, pos; end +local index_fmt, index_item_size, index_magic; +if string.packsize then + index_fmt = "T"; -- offset to the end of the item, length can be derived from two index items + index_item_size = string.packsize(index_fmt); + index_magic = string.pack(index_fmt, 7767639 + 1); -- Magic string: T9 for "prosody", version number +end + local function list_append(username, host, datastore, data) if not data then return; end if callback(username, host, datastore) == false then return true; end @@ -267,6 +289,22 @@ local function list_append(username, host, datastore, data) datastore, msg, where, username or "nil", host or "nil"); return ok, msg; end + if string.packsize then + local offset = type(msg) == "number" and msg or 0; + local index_entry = string.pack(index_fmt, offset + #data); + if offset == 0 then + index_entry = index_magic .. index_entry; + end + local ok, off = append(username, host, datastore, "lidx", index_entry); + off = off or 0; + -- If this was the first item, then both the data and index offsets should + -- be zero, otherwise there's some kind of mismatch and we should drop the + -- index and recreate it from scratch + -- TODO Actually rebuild the index in this case? + if not ok or (off == 0 and offset ~= 0) or (off ~= 0 and offset == 0) then + os_remove(getpath(username, host, datastore, "lidx")); + end + end return true; end @@ -280,6 +318,7 @@ local function list_store(username, host, datastore, data) for i, item in ipairs(data) do d[i] = "item(" .. serialize(item) .. ");\n"; end + os_remove(getpath(username, host, datastore, "lidx")); local ok, msg = atomic_store(getpath(username, host, datastore, "list", true), t_concat(d)); if not ok then log("error", "Unable to write to %s storage ('%s') for user: %s@%s", datastore, msg, username or "nil", host or "nil"); @@ -294,6 +333,160 @@ local function list_store(username, host, datastore, data) return true; end +local function build_list_index(username, host, datastore, items) + log("debug", "Building index for (%s@%s/%s)", username, host, datastore); + local filename = getpath(username, host, datastore, "list"); + local fh, err, errno = io_open(filename); + if not fh then + return fh, err, errno; + end + local prev_pos = 0; -- position before reading + local last_item_start = nil; + + if items and items[1] then + local last_item = items[#items]; + last_item_start = fh:seek("set", last_item.start + last_item.length); + else + items = {}; + end + + for line in fh:lines() do + if line:sub(1, 4) == "item" then + if prev_pos ~= 0 and last_item_start then + t_insert(items, { start = last_item_start; length = prev_pos - last_item_start }); + end + last_item_start = prev_pos + end + -- seek position is at the start of the next line within each loop iteration + -- so we need to collect the "current" position at the end of the previous + prev_pos = fh:seek() + end + fh:close(); + if prev_pos ~= 0 then + t_insert(items, { start = last_item_start; length = prev_pos - last_item_start }); + end + return items; +end + +local function store_list_index(username, host, datastore, index) + local data = { index_magic }; + for i, v in ipairs(index) do + data[i + 1] = string.pack(index_fmt, v.start + v.length); + end + local filename = getpath(username, host, datastore, "lidx"); + return atomic_store(filename, t_concat(data)); +end + +local index_mt = { + __index = function(t, i) + if type(i) ~= "number" or i % 1 ~= 0 or i < 0 then + return + end + if i <= 0 then + return 0 + end + local fh = t.file; + local pos = (i - 1) * index_item_size; + if fh:seek("set", pos) ~= pos then + return nil + end + local data = fh:read(index_item_size * 2); + if not data or #data ~= index_item_size * 2 then + return nil + end + local start, next_pos = string.unpack(index_fmt .. index_fmt, data); + if pos == 0 then + start = 0 + end + local length = next_pos - start; + local v = { start = start; length = length }; + t[i] = v; + return v; + end; + __len = function(t) + -- Account for both the header and the fence post error + return floor(t.file:seek("end") / index_item_size) - 1; + end; +} + +local function get_list_index(username, host, datastore) + log("debug", "Loading index for (%s@%s/%s)", username, host, datastore); + local index_filename = getpath(username, host, datastore, "lidx"); + local ih = io_open(index_filename); + if ih then + local magic = ih:read(#index_magic); + if magic ~= index_magic then + log("debug", "Index %q has wrong version number (got %q, expected %q), rebuilding...", index_filename, magic, index_magic); + -- wrong version or something + ih:close(); + ih = nil; + end + end + + if ih then + local first_length = string.unpack(index_fmt, ih:read(index_item_size)); + return setmetatable({ file = ih; { start = 0; length = first_length } }, index_mt); + end + + local index, err = build_list_index(username, host, datastore); + if not index then + return index, err + end + + -- TODO How to handle failure to store the index? + local dontcare = store_list_index(username, host, datastore, index); -- luacheck: ignore 211/dontcare + return index; +end + +local function list_load_one(fh, start, length) + if fh:seek("set", start) ~= start then + return nil + end + local raw_data = fh:read(length) + if not raw_data or #raw_data ~= length then + return + end + local item; + local data, err, errno = envload(raw_data, "@list", { + item = function(i) + item = i; + end; + }); + if not data then + return data, err, errno + end + local success, ret = pcall(data); + if not success then + return success, ret; + end + return item; +end + +local function list_close(list) + if list.index and list.index.file then + list.index.file:close(); + end + return list.file:close(); +end + +local indexed_list_mt = { + __index = function(t, i) + if type(i) ~= "number" or i % 1 ~= 0 or i < 1 then + return + end + local ix = t.index[i]; + if not ix then + return + end + local item = list_load_one(t.file, ix.start, ix.length); + return item; + end; + __len = function(t) + return #t.index; + end; + __close = list_close; +} + local function list_load(username, host, datastore) local items = {}; local data, err, errno = envloadfile(getpath(username, host, datastore, "list"), {item = function(i) t_insert(items, i); end}); @@ -314,6 +507,123 @@ local function list_load(username, host, datastore) return items; end +local function list_open(username, host, datastore) + if not index_magic then + log("debug", "Falling back from lazy loading to to loading full list for %s storage for user: %s@%s", datastore, username or "nil", host or "nil"); + return list_load(username, host, datastore); + end + local filename = getpath(username, host, datastore, "list"); + local file, err, errno = io_open(filename); + if not file then + if errno == ENOENT then + return nil; + end + return file, err, errno; + end + local index, err = get_list_index(username, host, datastore); + if not index then + file:close() + return index, err; + end + return setmetatable({ file = file; index = index; close = list_close }, indexed_list_mt); +end + +local function shift_index(index_filename, index, trim_to, offset) -- luacheck: ignore 212 + os_remove(index_filename); + return "deleted"; + -- TODO move and recalculate remaining items +end + +local function list_shift(username, host, datastore, trim_to) + if trim_to == 1 then + return true + end + if type(trim_to) ~= "number" or trim_to < 1 then + return nil, "invalid-argument"; + end + local list_filename = getpath(username, host, datastore, "list"); + local index_filename = getpath(username, host, datastore, "lidx"); + local index, err = get_list_index(username, host, datastore); + if not index then + return nil, err; + end + + local new_first = index[trim_to]; + if not new_first then + os_remove(index_filename); + return os_remove(list_filename); + end + + local offset = new_first.start; + if offset == 0 then + return true; + end + + --[[ + if remove_blocks then + local f, err = io_open(list_filename, "r+"); + if not f then + return f, err; + end + + local diff = 0; + local block_offset = 0; + if offset % 0x1000 ~= 0 then + -- Not an even block boundary, we will have to overwrite + diff = offset % 0x1000; + block_offset = offset - diff; + end + + if block_offset == 0 then + log("debug", "") + else + local ok, err = remove_blocks(f, 0, block_offset); + log("debug", "remove_blocks(%s, 0, %d)", f, block_offset); + if not ok then + log("warn", "Could not remove blocks from %q[%d, %d]: %s", list_filename, 0, block_offset, err); + else + if diff ~= 0 then + -- overwrite unaligned leftovers + if f:seek("set", 0) then + local wrote, err = f:write(string.rep("\n", diff)); + if not wrote then + log("error", "Could not blank out %q[%d, %d]: %s", list_filename, 0, diff, err); + end + end + end + local ok, err = f:close(); + shift_index(index_filename, index, trim_to, offset); -- Shift or delete the index + return ok, err; + end + end + end + --]] + + local r, err = io_open(list_filename, "r"); + if not r then + return nil, err; + end + local w, err = io_open(list_filename .. "~", "w"); + if not w then + return nil, err; + end + r:seek("set", offset); + for block in r:lines(0x1000) do + local ok, err = w:write(block); + if not ok then + return nil, err; + end + end + r:close(); + local ok, err = w:close(); + if not ok then + return nil, err; + end + shift_index(index_filename, index, trim_to, offset) + return os_rename(list_filename .. "~", list_filename); +end + + local type_map = { keyval = "dat"; list = "list"; @@ -392,6 +702,8 @@ local function purge(username, host) local ok, err = do_remove(getpath(username, host, store_name, "list")); if not ok then errs[#errs+1] = err; end + local ok, err = do_remove(getpath(username, host, store_name, "lidx")); + if not ok then errs[#errs+1] = err; end end end return #errs == 0, t_concat(errs, ", "); @@ -414,4 +726,8 @@ return { purge = purge; path_decode = decode; path_encode = encode; + + build_list_index = build_list_index; + list_open = list_open; + list_shift = list_shift; }; |