mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-22 02:53:09 +00:00
云风
1
Makefile
1
Makefile
@@ -63,6 +63,7 @@ LUA_CLIB_SKYNET = \
|
||||
lua-stm.c \
|
||||
lua-mysqlaux.c \
|
||||
lua-debugchannel.c \
|
||||
lua-datasheet.c \
|
||||
\
|
||||
|
||||
SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
|
||||
|
||||
372
lualib-src/lua-datasheet.c
Normal file
372
lualib-src/lua-datasheet.c
Normal file
@@ -0,0 +1,372 @@
|
||||
#include <lua.h>
|
||||
#include <lauxlib.h>
|
||||
#include <stdint.h>
|
||||
|
||||
#define NODECACHE "_ctable"
|
||||
#define PROXYCACHE "_proxy"
|
||||
#define TABLES "_ctables"
|
||||
|
||||
#define VALUE_NIL 0
|
||||
#define VALUE_INTEGER 1
|
||||
#define VALUE_REAL 2
|
||||
#define VALUE_BOOLEAN 3
|
||||
#define VALUE_TABLE 4
|
||||
#define VALUE_STRING 5
|
||||
#define VALUE_INVALID 6
|
||||
|
||||
#define INVALID_OFFSET 0xffffffff
|
||||
|
||||
struct proxy {
|
||||
const char * data;
|
||||
int index;
|
||||
};
|
||||
|
||||
struct document {
|
||||
uint32_t strtbl;
|
||||
uint32_t n;
|
||||
uint32_t index[1];
|
||||
// table[n]
|
||||
// strings
|
||||
};
|
||||
|
||||
struct table {
|
||||
uint32_t array;
|
||||
uint32_t dict;
|
||||
uint8_t type[1];
|
||||
// value[array]
|
||||
// kvpair[dict]
|
||||
};
|
||||
|
||||
static inline const struct table *
|
||||
gettable(const struct document *doc, int index) {
|
||||
if (doc->index[index] == INVALID_OFFSET) {
|
||||
return NULL;
|
||||
}
|
||||
return (const struct table *)((const char *)doc + sizeof(uint32_t) + sizeof(uint32_t) + doc->n * sizeof(uint32_t) + doc->index[index]);
|
||||
}
|
||||
|
||||
static void
|
||||
create_proxy(lua_State *L, const void *data, int index) {
|
||||
const struct table * t = gettable(data, index);
|
||||
if (t == NULL) {
|
||||
luaL_error(L, "Invalid index %d", index);
|
||||
}
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, NODECACHE);
|
||||
if (lua_rawgetp(L, -1, t) == LUA_TTABLE) {
|
||||
lua_replace(L, -2);
|
||||
return;
|
||||
}
|
||||
lua_pop(L, 1);
|
||||
lua_newtable(L);
|
||||
lua_pushvalue(L, lua_upvalueindex(1));
|
||||
lua_setmetatable(L, -2);
|
||||
lua_pushvalue(L, -1);
|
||||
// NODECACHE, table, table
|
||||
lua_rawsetp(L, -3, t);
|
||||
// NODECACHE, table
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, PROXYCACHE);
|
||||
// NODECACHE, table, PROXYCACHE
|
||||
lua_pushvalue(L, -2);
|
||||
// NODECACHE, table, PROXYCACHE, table
|
||||
struct proxy * p = lua_newuserdata(L, sizeof(struct proxy));
|
||||
// NODECACHE, table, PROXYCACHE, table, proxy
|
||||
p->data = data;
|
||||
p->index = index;
|
||||
lua_rawset(L, -3);
|
||||
// NODECACHE, table, PROXYCACHE
|
||||
lua_pop(L, 1);
|
||||
// NODECACHE, table
|
||||
lua_replace(L, -2);
|
||||
// table
|
||||
}
|
||||
|
||||
static void
|
||||
clear_table(lua_State *L) {
|
||||
int t = lua_gettop(L); // clear top table
|
||||
if (lua_type(L, t) != LUA_TTABLE) {
|
||||
luaL_error(L, "Invalid cache");
|
||||
}
|
||||
lua_pushnil(L);
|
||||
while (lua_next(L, t) != 0) {
|
||||
// key value
|
||||
lua_pop(L, 1);
|
||||
lua_pushvalue(L, -1);
|
||||
lua_pushnil(L);
|
||||
// key key nil
|
||||
lua_rawset(L, t);
|
||||
// key
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
update_cache(lua_State *L, const void *data, const void * newdata) {
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, NODECACHE);
|
||||
int t = lua_gettop(L);
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, PROXYCACHE);
|
||||
int pt = t + 1;
|
||||
lua_newtable(L); // temp table
|
||||
int nt = pt + 1;
|
||||
lua_pushnil(L);
|
||||
while (lua_next(L, t) != 0) {
|
||||
// pointer (-2) -> table (-1)
|
||||
lua_pushvalue(L, -1);
|
||||
if (lua_rawget(L, pt) == LUA_TUSERDATA) {
|
||||
// pointer, table, proxy
|
||||
struct proxy * p = lua_touserdata(L, -1);
|
||||
if (p->data == data) {
|
||||
// update to newdata
|
||||
p->data = newdata;
|
||||
const struct table * newt = gettable(newdata, p->index);
|
||||
lua_pop(L, 1);
|
||||
// pointer, table
|
||||
clear_table(L);
|
||||
lua_pushvalue(L, lua_upvalueindex(1));
|
||||
// pointer, table, meta
|
||||
lua_setmetatable(L, -2);
|
||||
// pointer, table
|
||||
if (newt) {
|
||||
lua_rawsetp(L, nt, newt);
|
||||
} else {
|
||||
lua_pop(L, 1);
|
||||
}
|
||||
// pointer
|
||||
lua_pushvalue(L, -1);
|
||||
lua_pushnil(L);
|
||||
lua_rawset(L, t);
|
||||
} else {
|
||||
lua_pop(L, 2);
|
||||
}
|
||||
} else {
|
||||
lua_pop(L, 2);
|
||||
// pointer
|
||||
}
|
||||
}
|
||||
// copy nt to t
|
||||
lua_pushnil(L);
|
||||
while (lua_next(L, nt) != 0) {
|
||||
lua_pushvalue(L, -2);
|
||||
lua_insert(L, -2);
|
||||
// key key value
|
||||
lua_rawset(L, t);
|
||||
}
|
||||
// NODECACHE PROXYCACHE TEMP
|
||||
lua_pop(L, 3);
|
||||
}
|
||||
|
||||
static int
|
||||
lupdate(lua_State *L) {
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, PROXYCACHE);
|
||||
lua_pushvalue(L, 1);
|
||||
// PROXYCACHE, table
|
||||
if (lua_rawget(L, -2) != LUA_TUSERDATA) {
|
||||
luaL_error(L, "Invalid proxy table %p", lua_topointer(L, 1));
|
||||
}
|
||||
struct proxy * p = lua_touserdata(L, -1);
|
||||
luaL_checktype(L, 2, LUA_TLIGHTUSERDATA);
|
||||
const char * newdata = lua_touserdata(L, 2);
|
||||
update_cache(L, p->data, newdata);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static inline uint32_t
|
||||
getuint32(const void *v) {
|
||||
union {
|
||||
uint32_t d;
|
||||
uint8_t t[4];
|
||||
} test = { 1 };
|
||||
if (test.t[0] == 0) {
|
||||
// big endian
|
||||
test.d = *(const uint32_t *)v;
|
||||
return test.t[0] | test.t[1] << 4 | test.t[2] << 8 | test.t[3] << 12;
|
||||
} else {
|
||||
return *(const uint32_t *)v;
|
||||
}
|
||||
}
|
||||
|
||||
static inline float
|
||||
getfloat(const void *v) {
|
||||
union {
|
||||
uint32_t d;
|
||||
float f;
|
||||
uint8_t t[4];
|
||||
} test = { 1 };
|
||||
if (test.t[0] == 0) {
|
||||
// big endian
|
||||
test.d = *(const uint32_t *)v;
|
||||
test.d = test.t[0] | test.t[1] << 4 | test.t[2] << 8 | test.t[3] << 12;
|
||||
return test.f;
|
||||
} else {
|
||||
return *(const float *)v;
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
pushvalue(lua_State *L, const void *v, int type, const struct document * doc) {
|
||||
switch (type) {
|
||||
case VALUE_NIL:
|
||||
lua_pushnil(L);
|
||||
break;
|
||||
case VALUE_INTEGER:
|
||||
lua_pushinteger(L, (int32_t)getuint32(v));
|
||||
break;
|
||||
case VALUE_REAL:
|
||||
lua_pushnumber(L, getfloat(v));
|
||||
break;
|
||||
case VALUE_BOOLEAN:
|
||||
lua_pushboolean(L, getuint32(v));
|
||||
break;
|
||||
case VALUE_TABLE:
|
||||
create_proxy(L, doc, getuint32(v));
|
||||
break;
|
||||
case VALUE_STRING:
|
||||
lua_pushstring(L, (const char *)doc + doc->strtbl + getuint32(v));
|
||||
break;
|
||||
default:
|
||||
luaL_error(L, "Invalid type %d at %p", type, v);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
copytable(lua_State *L, int tbl, struct proxy *p) {
|
||||
const struct document * doc = (const struct document *)p->data;
|
||||
if (p->index < 0 || p->index >= doc->n) {
|
||||
luaL_error(L, "Invalid proxy (index = %d, total = %d)", p->index, (int)doc->n);
|
||||
}
|
||||
const struct table * t = gettable(doc, p->index);
|
||||
if (t == NULL) {
|
||||
luaL_error(L, "Invalid proxy (index = %d)", p->index);
|
||||
}
|
||||
const uint32_t * v = (const uint32_t *)((const char *)t + sizeof(uint32_t) + sizeof(uint32_t) + ((t->array + t->dict + 3) & ~3));
|
||||
int i;
|
||||
for (i=0;i<t->array;i++) {
|
||||
pushvalue(L, v++, t->type[i], doc);
|
||||
lua_rawseti(L, tbl, i+1);
|
||||
}
|
||||
for (i=0;i<t->dict;i++) {
|
||||
pushvalue(L, v++, VALUE_STRING, doc);
|
||||
pushvalue(L, v++, t->type[t->array+i], doc);
|
||||
lua_rawset(L, tbl);
|
||||
}
|
||||
}
|
||||
|
||||
static int
|
||||
lnew(lua_State *L) {
|
||||
luaL_checktype(L, 1, LUA_TLIGHTUSERDATA);
|
||||
const char * data = lua_touserdata(L, 1);
|
||||
// hold ref to data
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, TABLES);
|
||||
lua_pushvalue(L, 1);
|
||||
lua_rawsetp(L, -2, data);
|
||||
|
||||
create_proxy(L, data, 0);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void
|
||||
copyfromdata(lua_State *L) {
|
||||
lua_getfield(L, LUA_REGISTRYINDEX, PROXYCACHE);
|
||||
lua_pushvalue(L, 1);
|
||||
// PROXYCACHE, table
|
||||
if (lua_rawget(L, -2) != LUA_TUSERDATA) {
|
||||
luaL_error(L, "Invalid proxy table %p", lua_topointer(L, 1));
|
||||
}
|
||||
struct proxy * p = lua_touserdata(L, -1);
|
||||
lua_pop(L, 2);
|
||||
copytable(L, 1, p);
|
||||
lua_pushnil(L);
|
||||
lua_setmetatable(L, 1); // remove metatable
|
||||
}
|
||||
|
||||
static int
|
||||
lindex(lua_State *L) {
|
||||
copyfromdata(L);
|
||||
lua_rawget(L, 1);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lnext(lua_State *L) {
|
||||
luaL_checktype(L, 1, LUA_TTABLE);
|
||||
lua_settop(L, 2); /* create a 2nd argument if there isn't one */
|
||||
if (lua_next(L, 1))
|
||||
return 2;
|
||||
else {
|
||||
lua_pushnil(L);
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
static int
|
||||
lpairs(lua_State *L) {
|
||||
copyfromdata(L);
|
||||
lua_pushcfunction(L, lnext);
|
||||
lua_pushvalue(L, 1);
|
||||
lua_pushnil(L);
|
||||
return 3;
|
||||
}
|
||||
|
||||
static int
|
||||
llen(lua_State *L) {
|
||||
copyfromdata(L);
|
||||
lua_pushinteger(L, lua_rawlen(L, 1));
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void
|
||||
new_weak_table(lua_State *L, const char *mode) {
|
||||
lua_newtable(L); // NODECACHE { pointer:table }
|
||||
|
||||
lua_createtable(L, 0, 1); // weak meta table
|
||||
lua_pushstring(L, mode);
|
||||
lua_setfield(L, -2, "__mode");
|
||||
|
||||
lua_setmetatable(L, -2); // make NODECACHE weak
|
||||
}
|
||||
|
||||
static void
|
||||
gen_metatable(lua_State *L) {
|
||||
new_weak_table(L, "kv"); // NODECACHE { pointer:table }
|
||||
lua_setfield(L, LUA_REGISTRYINDEX, NODECACHE);
|
||||
|
||||
new_weak_table(L, "k"); // PROXYCACHE { table:userdata }
|
||||
lua_setfield(L, LUA_REGISTRYINDEX, PROXYCACHE);
|
||||
|
||||
lua_newtable(L);
|
||||
lua_setfield(L, LUA_REGISTRYINDEX, TABLES);
|
||||
|
||||
lua_createtable(L, 0, 1); // mod table
|
||||
|
||||
lua_createtable(L, 0, 2); // metatable
|
||||
luaL_Reg l[] = {
|
||||
{ "__index", lindex },
|
||||
{ "__pairs", lpairs },
|
||||
{ "__len", llen },
|
||||
{ NULL, NULL },
|
||||
};
|
||||
lua_pushvalue(L, -1);
|
||||
luaL_setfuncs(L, l, 1);
|
||||
}
|
||||
|
||||
static int
|
||||
lstringpointer(lua_State *L) {
|
||||
const char * str = luaL_checkstring(L, 1);
|
||||
lua_pushlightuserdata(L, (void *)str);
|
||||
return 1;
|
||||
}
|
||||
|
||||
LUAMOD_API int
|
||||
luaopen_skynet_datasheet_core(lua_State *L) {
|
||||
luaL_checkversion(L);
|
||||
luaL_Reg l[] = {
|
||||
{ "new", lnew },
|
||||
{ "update", lupdate },
|
||||
{ NULL, NULL },
|
||||
};
|
||||
|
||||
luaL_newlibtable(L,l);
|
||||
gen_metatable(L);
|
||||
luaL_setfuncs(L, l, 1);
|
||||
lua_pushcfunction(L, lstringpointer);
|
||||
lua_setfield(L, -2, "stringpointer");
|
||||
return 1;
|
||||
}
|
||||
167
lualib/skynet/datasheet/builder.lua
Normal file
167
lualib/skynet/datasheet/builder.lua
Normal file
@@ -0,0 +1,167 @@
|
||||
local skynet = require "skynet"
|
||||
local dump = require "skynet.datasheet.dump"
|
||||
local core = require "skynet.datasheet.core"
|
||||
local service = require "skynet.service"
|
||||
|
||||
local builder = {}
|
||||
|
||||
local cache = {}
|
||||
local dataset = {}
|
||||
|
||||
local function monitor(pointer)
|
||||
skynet.fork(function()
|
||||
skynet.call(service.address, "lua", "collect", pointer)
|
||||
for k,v in pairs(cache) do
|
||||
if v == pointer then
|
||||
cache[k] = nil
|
||||
return
|
||||
end
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
local function dumpsheet(v)
|
||||
if type(v) == "string" then
|
||||
return v
|
||||
else
|
||||
return dump.dump(v)
|
||||
end
|
||||
end
|
||||
|
||||
function builder.new(name, v)
|
||||
assert(dataset[name] == nil)
|
||||
local datastring = dumpsheet(v)
|
||||
local pointer = core.stringpointer(datastring)
|
||||
skynet.call(service.address, "lua", "update", name, pointer)
|
||||
cache[datastring] = pointer
|
||||
dataset[name] = datastring
|
||||
monitor(pointer)
|
||||
end
|
||||
|
||||
function builder.update(name, v)
|
||||
local lastversion = assert(dataset[name])
|
||||
local newversion = dumpsheet(v)
|
||||
local diff = dump.diff(lastversion, newversion)
|
||||
local pointer = core.stringpointer(diff)
|
||||
skynet.call(service.address, "lua", "update", name, pointer)
|
||||
cache[diff] = pointer
|
||||
local lp = assert(cache[lastversion])
|
||||
skynet.send(service.address, "lua", "release", lp)
|
||||
dataset[name] = diff
|
||||
monitor(pointer)
|
||||
end
|
||||
|
||||
function builder.compile(v)
|
||||
return dump.dump(v)
|
||||
end
|
||||
|
||||
local function datasheet_service()
|
||||
|
||||
local skynet = require "skynet"
|
||||
|
||||
local datasheet = {}
|
||||
local handles = {} -- handle:{ ref:count , name:name , collect:resp }
|
||||
local dataset = {} -- name:{ handle:handle, monitor:{monitors queue} }
|
||||
|
||||
local function releasehandle(handle)
|
||||
local h = handles[handle]
|
||||
h.ref = h.ref - 1
|
||||
if h.ref == 0 and h.collect then
|
||||
h.collect(true)
|
||||
h.collect = nil
|
||||
handles[handle] = nil
|
||||
end
|
||||
end
|
||||
|
||||
-- from builder, create or update handle
|
||||
function datasheet.update(name, handle)
|
||||
local t = dataset[name]
|
||||
if not t then
|
||||
-- new datasheet
|
||||
t = { handle = handle, monitor = {} }
|
||||
dataset[name] = t
|
||||
handles[handle] = { ref = 1, name = name }
|
||||
else
|
||||
t.handle = handle
|
||||
-- report update to customers
|
||||
handles[handle] = { ref = 1 + #t.monitor, name = name }
|
||||
|
||||
for k,v in ipairs(t.monitor) do
|
||||
v(true, handle)
|
||||
t.monitor[k] = nil
|
||||
end
|
||||
end
|
||||
skynet.ret()
|
||||
end
|
||||
|
||||
-- from customers
|
||||
function datasheet.query(name)
|
||||
local t = assert(dataset[name], "create data first")
|
||||
local handle = t.handle
|
||||
local h = handles[handle]
|
||||
h.ref = h.ref + 1
|
||||
skynet.ret(skynet.pack(handle))
|
||||
end
|
||||
|
||||
-- from customers, monitor handle change
|
||||
function datasheet.monitor(handle)
|
||||
local h = assert(handles[handle], "Invalid data handle")
|
||||
local t = dataset[h.name]
|
||||
if t.handle ~= handle then -- already changes
|
||||
skynet.ret(skynet.pack(t.handle))
|
||||
else
|
||||
h.ref = h.ref + 1
|
||||
table.insert(t.monitor, skynet.response())
|
||||
end
|
||||
end
|
||||
|
||||
-- from customers, release handle , ref count - 1
|
||||
function datasheet.release(handle)
|
||||
-- send message, don't ret
|
||||
releasehandle(handle)
|
||||
end
|
||||
|
||||
-- from builder, monitor handle release
|
||||
function datasheet.collect(handle)
|
||||
local h = assert(handles[handle], "Invalid data handle")
|
||||
if h.ref == 0 then
|
||||
handles[handle] = nil
|
||||
skynet.ret()
|
||||
else
|
||||
assert(h.collect == nil, "Only one collect allows")
|
||||
h.collect = skynet.response()
|
||||
end
|
||||
end
|
||||
|
||||
skynet.dispatch("lua", function(_,_,cmd,...)
|
||||
datasheet[cmd](...)
|
||||
end)
|
||||
|
||||
skynet.info_func(function()
|
||||
local info = {}
|
||||
local tmp = {}
|
||||
for k,v in pairs(handles) do
|
||||
tmp[k] = v
|
||||
end
|
||||
for k,v in pairs(dataset) do
|
||||
local h = handles[v.handle]
|
||||
tmp[v.handle] = nil
|
||||
info[k] = {
|
||||
handle = v.handle,
|
||||
monitors = #v.monitor,
|
||||
}
|
||||
end
|
||||
for k,v in pairs(tmp) do
|
||||
info[k] = v.ref
|
||||
end
|
||||
|
||||
return info
|
||||
end)
|
||||
|
||||
end
|
||||
|
||||
skynet.init(function()
|
||||
service.new("datasheet", datasheet_service)
|
||||
end)
|
||||
|
||||
return builder
|
||||
270
lualib/skynet/datasheet/dump.lua
Normal file
270
lualib/skynet/datasheet/dump.lua
Normal file
@@ -0,0 +1,270 @@
|
||||
--[[ file format
|
||||
document :
|
||||
int32 strtbloffset
|
||||
int32 n
|
||||
int32*n index table
|
||||
table*n
|
||||
strings
|
||||
|
||||
table:
|
||||
int32 array
|
||||
int32 dict
|
||||
int8*(array+dict) type (align 4)
|
||||
value*array
|
||||
kvpair*dict
|
||||
|
||||
kvpair:
|
||||
string k
|
||||
value v
|
||||
|
||||
value: (union)
|
||||
int32 integer
|
||||
float real
|
||||
int32 boolean
|
||||
int32 table index
|
||||
int32 string offset
|
||||
|
||||
type: (enum)
|
||||
0 nil
|
||||
1 integer
|
||||
2 real
|
||||
3 boolean
|
||||
4 table
|
||||
5 string
|
||||
]]
|
||||
|
||||
local ctd = {}
|
||||
local math = math
|
||||
local table = table
|
||||
local string = string
|
||||
|
||||
function ctd.dump(root)
|
||||
local doc = {
|
||||
table_n = 0,
|
||||
table = {},
|
||||
strings = {},
|
||||
offset = 0,
|
||||
}
|
||||
local function dump_table(t)
|
||||
local index = doc.table_n + 1
|
||||
doc.table_n = index
|
||||
doc.table[index] = false -- place holder
|
||||
local array_n = 0
|
||||
local array = {}
|
||||
local kvs = {}
|
||||
local types = {}
|
||||
local function encode(v)
|
||||
local t = type(v)
|
||||
if t == "table" then
|
||||
local index = dump_table(v)
|
||||
return '\4', string.pack("<i4", index-1)
|
||||
elseif t == "number" then
|
||||
if math.tointeger(v) then
|
||||
return '\1', string.pack("<i4", v)
|
||||
else
|
||||
return '\2', string.pack("<f",v)
|
||||
end
|
||||
elseif t == "boolean" then
|
||||
if v then
|
||||
return '\3', "\0\0\0\1"
|
||||
else
|
||||
return '\3', "\0\0\0\0"
|
||||
end
|
||||
elseif t == "string" then
|
||||
local offset = doc.strings[v]
|
||||
if not offset then
|
||||
offset = doc.offset
|
||||
doc.offset = offset + #v + 1
|
||||
doc.strings[v] = offset
|
||||
table.insert(doc.strings, v)
|
||||
end
|
||||
return '\5', string.pack("<I4", offset)
|
||||
else
|
||||
error ("Unsupport value " .. tostring(v))
|
||||
end
|
||||
end
|
||||
for i,v in ipairs(t) do
|
||||
types[i], array[i] = encode(v)
|
||||
array_n = i
|
||||
end
|
||||
for k,v in pairs(t) do
|
||||
if type(k) == "string" then
|
||||
local _, kv = encode(k)
|
||||
local tv, ev = encode(v)
|
||||
table.insert(types, tv)
|
||||
table.insert(kvs, kv .. ev)
|
||||
else
|
||||
local ik = math.tointeger(k)
|
||||
assert(ik and ik > 0 and ik <= array_n)
|
||||
end
|
||||
end
|
||||
-- encode table
|
||||
local typeset = table.concat(types)
|
||||
local align = string.rep("\0", (4 - #typeset & 3) & 3)
|
||||
local tmp = {
|
||||
string.pack("<i4i4", array_n, #kvs),
|
||||
typeset,
|
||||
align,
|
||||
table.concat(array),
|
||||
table.concat(kvs),
|
||||
}
|
||||
doc.table[index] = table.concat(tmp)
|
||||
return index
|
||||
end
|
||||
dump_table(root)
|
||||
-- encode document
|
||||
local index = {}
|
||||
local offset = 0
|
||||
for i, v in ipairs(doc.table) do
|
||||
index[i] = string.pack("<I4", offset)
|
||||
offset = offset + #v
|
||||
end
|
||||
local tmp = {
|
||||
string.pack("<I4", 4 + 4 + 4 * doc.table_n + offset),
|
||||
string.pack("<I4", doc.table_n),
|
||||
table.concat(index),
|
||||
table.concat(doc.table),
|
||||
table.concat(doc.strings, "\0")
|
||||
}
|
||||
return table.concat(tmp)
|
||||
end
|
||||
|
||||
function ctd.undump(v)
|
||||
local stringtbl, n = string.unpack("<I4I4",v)
|
||||
local index = { string.unpack("<" .. string.rep("I4", n), v, 9) }
|
||||
local header = 4 + 4 + 4 * n + 1
|
||||
stringtbl = stringtbl + 1
|
||||
local tblidx = {}
|
||||
local function decode(n)
|
||||
local toffset = index[n+1] + header
|
||||
local array, dict = string.unpack("<I4I4", v, toffset)
|
||||
local types = { string.unpack(string.rep("B", (array+dict)), v, toffset + 8) }
|
||||
local offset = ((array + dict + 8 + 3) & ~3) + toffset
|
||||
local result = {}
|
||||
local function value(t)
|
||||
local off = offset
|
||||
offset = offset + 4
|
||||
if t == 1 then -- integer
|
||||
return (string.unpack("<i4", v, off))
|
||||
elseif t == 2 then -- float
|
||||
return (string.unpack("<f", v, off))
|
||||
elseif t == 3 then -- boolean
|
||||
return string.unpack("<i4", v, off) ~= 0
|
||||
elseif t == 4 then -- table
|
||||
local tindex = (string.unpack("<I4", v, off))
|
||||
return decode(tindex)
|
||||
elseif t == 5 then -- string
|
||||
local sindex = string.unpack("<I4", v, off)
|
||||
return (string.unpack("z", v, stringtbl + sindex))
|
||||
else
|
||||
error (string.format("Invalid data at %d (%d)", off, t))
|
||||
end
|
||||
end
|
||||
for i=1,array do
|
||||
table.insert(result, value(types[i]))
|
||||
end
|
||||
for i=1,dict do
|
||||
local sindex = string.unpack("<I4", v, offset)
|
||||
offset = offset + 4
|
||||
local key = string.unpack("z", v, stringtbl + sindex)
|
||||
result[key] = value(types[array + i])
|
||||
end
|
||||
tblidx[result] = n
|
||||
return result
|
||||
end
|
||||
return decode(0), tblidx
|
||||
end
|
||||
|
||||
local function diffmap(last, current)
|
||||
local lastv, lasti = ctd.undump(last)
|
||||
local curv, curi = ctd.undump(current)
|
||||
local map = {} -- new(current index):old(last index)
|
||||
local function comp(lastr, curr)
|
||||
local old = lasti[lastr]
|
||||
local new = curi[curr]
|
||||
map[new] = old
|
||||
for k,v in pairs(lastr) do
|
||||
if type(v) == "table" then
|
||||
local newv = curr[k]
|
||||
if type(newv) == "table" then
|
||||
comp(v, newv)
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
comp(lastv, curv)
|
||||
return map
|
||||
end
|
||||
|
||||
function ctd.diff(last, current)
|
||||
local map = diffmap(last, current)
|
||||
local stringtbl, n = string.unpack("<I4I4",current)
|
||||
local _, lastn = string.unpack("<I4I4",last)
|
||||
local existn = 0
|
||||
for k,v in pairs(map) do
|
||||
existn = existn + 1
|
||||
end
|
||||
local newn = lastn
|
||||
for i = 0, n-1 do
|
||||
if not map[i] then
|
||||
map[i] = newn
|
||||
newn = newn + 1
|
||||
end
|
||||
end
|
||||
-- remap current
|
||||
local index = { string.unpack("<" .. string.rep("I4", n), current, 9) }
|
||||
local header = 4 + 4 + 4 * n + 1
|
||||
local function remap(n)
|
||||
local toffset = index[n+1] + header
|
||||
local array, dict = string.unpack("<I4I4", current, toffset)
|
||||
local types = { string.unpack(string.rep("B", (array+dict)), current, toffset + 8) }
|
||||
local hlen = (array + dict + 8 + 3) & ~3
|
||||
local hastable = false
|
||||
for _, v in ipairs(types) do
|
||||
if v == 4 then -- table
|
||||
hastable = true
|
||||
break
|
||||
end
|
||||
end
|
||||
if not hastable then
|
||||
return string.sub(current, toffset, toffset + hlen + (array + dict * 2) * 4 - 1)
|
||||
end
|
||||
local offset = hlen + toffset
|
||||
local pat = "<" .. string.rep("I4", array + dict * 2)
|
||||
local values = { string.unpack(pat, current, offset) }
|
||||
for i = 1, array do
|
||||
if types[i] == 4 then -- table
|
||||
values[i] = map[values[i]]
|
||||
end
|
||||
end
|
||||
for i = 1, dict do
|
||||
if types[i + array] == 4 then -- table
|
||||
values[array + i * 2] = map[values[array + i * 2]]
|
||||
end
|
||||
end
|
||||
return string.sub(current, toffset, toffset + hlen - 1) ..
|
||||
string.pack(pat, table.unpack(values))
|
||||
end
|
||||
-- rebuild
|
||||
local oldindex = { string.unpack("<" .. string.rep("I4", n), current, 9) }
|
||||
local index = {}
|
||||
for i = 1, newn do
|
||||
index[i] = 0xffffffff
|
||||
end
|
||||
for i = 0, #map do
|
||||
index[map[i]+1] = oldindex[i+1]
|
||||
end
|
||||
|
||||
local tmp = {
|
||||
string.pack("<I4I4", stringtbl + (newn - n) * 4, newn), -- expand index table
|
||||
string.pack("<" .. string.rep("I4", newn), table.unpack(index)),
|
||||
}
|
||||
for i = 0, n - 1 do
|
||||
table.insert(tmp, remap(i))
|
||||
end
|
||||
table.insert(tmp, string.sub(current, stringtbl+1))
|
||||
|
||||
return table.concat(tmp)
|
||||
end
|
||||
|
||||
return ctd
|
||||
77
lualib/skynet/datasheet/init.lua
Normal file
77
lualib/skynet/datasheet/init.lua
Normal file
@@ -0,0 +1,77 @@
|
||||
local skynet = require "skynet"
|
||||
local service = require "skynet.service"
|
||||
local core = require "skynet.datasheet.core"
|
||||
|
||||
local datasheet_svr
|
||||
|
||||
skynet.init(function()
|
||||
datasheet_svr = service.query "datasheet"
|
||||
end)
|
||||
|
||||
local datasheet = {}
|
||||
local sheets = setmetatable({}, {
|
||||
__gc = function(t)
|
||||
for _,v in pairs(t) do
|
||||
skynet.send(datasheet_svr, "lua", "release", v.handle)
|
||||
end
|
||||
end,
|
||||
})
|
||||
|
||||
local function querysheet(name)
|
||||
return skynet.call(datasheet_svr, "lua", "query", name)
|
||||
end
|
||||
|
||||
local function updateobject(name)
|
||||
local t = sheets[name]
|
||||
if not t.object then
|
||||
t.object = core.new(t.handle)
|
||||
end
|
||||
local function monitor()
|
||||
local handle = t.handle
|
||||
local newhandle = skynet.call(datasheet_svr, "lua", "monitor", handle)
|
||||
core.update(t.object, newhandle)
|
||||
t.handle = newhandle
|
||||
skynet.send(datasheet_svr, "lua", "release", handle)
|
||||
skynet.fork(monitor)
|
||||
end
|
||||
skynet.fork(monitor)
|
||||
end
|
||||
|
||||
function datasheet.query(name)
|
||||
local t = sheets[name]
|
||||
if not t then
|
||||
t = {}
|
||||
sheets[name] = t
|
||||
end
|
||||
if t.error then
|
||||
error(t.error)
|
||||
end
|
||||
if t.object then
|
||||
return t.object
|
||||
end
|
||||
if t.queue then
|
||||
local co = coroutine.running()
|
||||
table.insert(t.queue, co)
|
||||
skynet.wait(co)
|
||||
else
|
||||
t.queue = {} -- create wait queue for other query
|
||||
local ok, handle = pcall(querysheet, name)
|
||||
if ok then
|
||||
t.handle = handle
|
||||
updateobject(name)
|
||||
else
|
||||
t.error = handle
|
||||
end
|
||||
local q = t.queue
|
||||
t.queue = nil
|
||||
for _, co in ipairs(q) do
|
||||
skynet.wakeup(co)
|
||||
end
|
||||
end
|
||||
if t.error then
|
||||
error(t.error)
|
||||
end
|
||||
return t.object
|
||||
end
|
||||
|
||||
return datasheet
|
||||
28
test/testdatasheet.lua
Normal file
28
test/testdatasheet.lua
Normal file
@@ -0,0 +1,28 @@
|
||||
local skynet = require "skynet"
|
||||
local builder = require "skynet.datasheet.builder"
|
||||
local datasheet = require "skynet.datasheet"
|
||||
|
||||
local function dump(t, prefix)
|
||||
for k,v in pairs(t) do
|
||||
print(prefix, k, v)
|
||||
if type(v) == "table" then
|
||||
dump(v, prefix .. "." .. k)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
skynet.start(function()
|
||||
builder.new("foobar", {a = 1, b = 2 , c = {3} })
|
||||
local t = datasheet.query "foobar"
|
||||
local c = t.c
|
||||
dump(t, "[1]")
|
||||
builder.update("foobar", { b = 4, c = { 5 } })
|
||||
print("sleep")
|
||||
skynet.sleep(100)
|
||||
dump(t, "[2]")
|
||||
dump(c, "[2.c]")
|
||||
builder.update("foobar", { a = 6, c = 7, d = 8 })
|
||||
print("sleep")
|
||||
skynet.sleep(100)
|
||||
dump(t, "[3]")
|
||||
end)
|
||||
Reference in New Issue
Block a user