kvs.erl 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324
  1. -module(kvs).
  2. -copyright('Synrc Research Center s.r.o.').
  3. -compile(export_all).
  4. -include_lib("stdlib/include/qlc.hrl").
  5. -include("config.hrl").
  6. -include("metainfo.hrl").
  7. -include("kvs.hrl").
  8. -include("api.hrl").
  9. % Public Main Backend is given in sys.config and
  10. % could be obtained with application:get_env(kvs,dba,store_mnesia).
  11. delete(Table,Key) -> delete (Table, Key, #kvs{mod=?DBA}).
  12. remove(Table,Key) -> remove (Table, Key, #kvs{mod=?DBA}).
  13. get(Table,Key) -> get (Table, Key, #kvs{mod=?DBA}).
  14. index(Table,K,V) -> index (Table, K,V, #kvs{mod=?DBA}).
  15. next_id(Table,DX) -> next_id (Table, DX, #kvs{mod=?DBA}).
  16. change_storage(Table,Type) -> change_storage(Table,Type, #kvs{mod=?DBA}).
  17. entries(A,B,C) -> entries (A,B,C, #kvs{mod=?DBA}).
  18. join() -> join ([], #kvs{mod=?DBA}).
  19. join(Node) -> join (Node, #kvs{mod=?DBA}).
  20. count(Table) -> count (Table, #kvs{mod=?DBA}).
  21. add(Table) -> add (Table, #kvs{mod=?DBA}).
  22. all(Table) -> all (Table, #kvs{mod=?DBA}).
  23. put(Table) -> put (Table, #kvs{mod=?DBA}).
  24. link(Table) -> link (Table, #kvs{mod=?DBA}).
  25. traversal(T,S,C,D) -> traversal(T,S,C,D, #kvs{mod=?DBA}).
  26. info(T) -> info (T, #kvs{mod=?DBA}).
  27. start() -> start (#kvs{mod=?DBA}).
  28. stop() -> stop (#kvs{mod=?DBA}).
  29. destroy() -> destroy (#kvs{mod=?DBA}).
  30. version() -> version (#kvs{mod=?DBA}).
  31. dir() -> dir (#kvs{mod=?DBA}).
  32. % Implementation
  33. init(Backend, Module) ->
  34. [ begin
  35. Backend:create_table(T#table.name, [{attributes,T#table.fields},{T#table.copy_type, [node()]}]),
  36. [ Backend:add_table_index(T#table.name, Key) || Key <- T#table.keys ],
  37. T
  38. end || T <- (Module:metainfo())#schema.tables ].
  39. start(#kvs{mod=DBA}) -> DBA:start().
  40. stop(#kvs{mod=DBA}) -> DBA:stop().
  41. change_storage(Type) -> [ change_storage(Name,Type) || #table{name=Name} <- kvs:tables() ].
  42. change_storage(Table,Type,#kvs{mod=DBA}) -> DBA:change_storage(Table,Type).
  43. destroy(#kvs{mod=DBA}) -> DBA:destroy().
  44. join(Node,#kvs{mod=DBA}) -> Tables = nonexistent(), DBA:join(Node), rotate_new(Tables), load_partitions().
  45. version(#kvs{mod=DBA}) -> DBA:version().
  46. tables() -> lists:flatten([ (M:metainfo())#schema.tables || M <- modules() ]).
  47. table(Name) -> lists:keyfind(Name,#table.name,tables()).
  48. dir(#kvs{mod=DBA}) -> DBA:dir().
  49. info(T,#kvs{mod=DBA}) -> DBA:info(T).
  50. modules() -> kvs:config(schema).
  51. containers() ->
  52. lists:flatten([ [ {T#table.name,T#table.fields}
  53. || T=#table{container=true} <- (M:metainfo())#schema.tables ]
  54. || M <- modules() ]).
  55. create(ContainerName) -> create(ContainerName, kvs:next_id(atom_to_list(ContainerName), 1), #kvs{mod=?DBA}).
  56. create(ContainerName, Id, Driver) ->
  57. kvs:info(?MODULE,"Create: ~p",[ContainerName]),
  58. Instance = list_to_tuple([ContainerName|proplists:get_value(ContainerName, kvs:containers())]),
  59. Top = setelement(#container.id,Instance,Id),
  60. Top2 = setelement(#container.top,Top,undefined),
  61. Top3 = setelement(#container.count,Top2,0),
  62. ok = kvs:put(Top3, Driver),
  63. Id.
  64. ensure_link(Record, #kvs{mod=_Store}=Driver) ->
  65. Id = element(2,Record),
  66. Type = table_type(element(1,Record)),
  67. CName = element(#iterator.container, Record),
  68. Cid = table_type(case element(#iterator.feed_id, Record) of
  69. undefined -> table_type(element(1,Record));
  70. Fid -> Fid end),
  71. Container = case kvs:get(CName, Cid, Driver) of
  72. {ok,Res} -> Res;
  73. {error, _} when Cid /= undefined ->
  74. NC = setelement(#container.id,
  75. list_to_tuple([CName|
  76. proplists:get_value(CName, kvs:containers())]), Cid),
  77. NC1 = setelement(#container.count, NC, 0),
  78. NC1;
  79. _Error -> error end,
  80. case Container of
  81. error -> {error, no_container};
  82. _ when element(#container.top,Container) == Id -> {error,just_added};
  83. _ ->
  84. Next = undefined,
  85. Prev = case element(#container.top, Container) of
  86. undefined -> undefined;
  87. Tid -> case kvs:get(Type, Tid, Driver) of
  88. {error, _} -> undefined;
  89. {ok, Top} -> NewTop = setelement(#iterator.next, Top, Id),
  90. kvs:put(NewTop, Driver),
  91. element(#iterator.id, NewTop) end end,
  92. C1 = setelement(#container.top, Container, Id),
  93. C2 = setelement(#container.count, C1,
  94. element(#container.count, Container)+1),
  95. kvs:put(C2, Driver), % Container
  96. R = setelement(#iterator.feeds, Record,
  97. [ case F1 of
  98. {FN, Fd} -> {FN, Fd};
  99. _-> {F1, kvs:create(CName,{F1,element(#iterator.id,Record)},Driver)}
  100. end || F1 <- element(#iterator.feeds, Record)]),
  101. R1 = setelement(#iterator.next, R, Next),
  102. R2 = setelement(#iterator.prev, R1, Prev),
  103. R3 = setelement(#iterator.feed_id, R2, element(#container.id, Container)),
  104. kvs:put(R3, Driver), % Iterator
  105. kvs:info(?MODULE,"Put: ~p~n", [element(#container.id,R3)]),
  106. {ok, R3}
  107. end.
  108. link(Record,#kvs{mod=_Store}=Driver) ->
  109. Id = element(#iterator.id, Record),
  110. case kvs:get(table_type(element(1,Record)), Id, Driver) of
  111. {ok, Exists} -> ensure_link(Exists, Driver);
  112. {error, not_found} -> {error, not_found} end.
  113. add(Record, #kvs{mod=_Store}=Driver) when is_tuple(Record) ->
  114. Id = element(#iterator.id, Record),
  115. case kvs:get(table_type(element(1,Record)), Id, Driver) of
  116. {error, _} -> ensure_link(Record, Driver);
  117. {aborted, Reason} -> {aborted, Reason};
  118. {ok, _} -> {error, exist} end.
  119. reverse(#iterator.prev) -> #iterator.next;
  120. reverse(#iterator.next) -> #iterator.prev.
  121. relink(Container, E, Driver) ->
  122. Id = element(#iterator.id, E),
  123. Next = element(#iterator.next, E),
  124. Prev = element(#iterator.prev, E),
  125. Top = element(#container.top, Container),
  126. case kvs:get(element(1,E), Prev, Driver) of
  127. {ok, PE} -> kvs:put(setelement(#iterator.next, PE, Next), Driver);
  128. _ -> ok end,
  129. case kvs:get(element(1,E), Next, Driver) of
  130. {ok, NE} -> kvs:put(setelement(#iterator.prev, NE, Prev), Driver);
  131. _ -> ok end,
  132. C = case Top of
  133. Id -> setelement(#container.top, Container, Prev);
  134. _ -> Container end,
  135. case element(#container.top,C) of
  136. undefined -> kvs:delete(element(1,C),element(#container.id,C));
  137. _ -> kvs:put(setelement(#container.count,C,element(#container.count,C)-1), Driver) end.
  138. delete(Tab, Key, #kvs{mod=Mod}) ->
  139. case range(Tab,Key) of
  140. [] -> Mod:delete(Tab, Key);
  141. T -> Mod:delete(T, Key) end.
  142. remove(Record,Id,#kvs{mod=Mod}=Driver) ->
  143. case get(Record,Id) of
  144. {error, not_found} -> kvs:error(?MODULE,"Can't remove ~p~n",[{Record,Id}]);
  145. {ok,R} -> do_remove(R,Driver) end.
  146. do_remove(E,#kvs{mod=Mod}=Driver) ->
  147. case get(element(#iterator.container,E),element(#iterator.feed_id,E)) of
  148. {ok, C} -> relink(C,E,Driver);
  149. _ -> skip end,
  150. kvs:info(?MODULE,"Delete: ~p", [E]),
  151. kvs:delete(element(1,E),element(2,E), Driver).
  152. traversal(Table, Start, Count, Direction, Driver)->
  153. fold(fun(A,Acc) -> [A|Acc] end,[],Table,Start,Count,Direction,Driver).
  154. % kvs:fold(fun(X,A)->[X|A]end,[],process,2152,-1,#iterator.next,#kvs{mod=store_mnesia}).
  155. fold(___,___,_,undefined,_,_,_) -> [];
  156. fold(___,Acc,_,_,0,_,_) -> Acc;
  157. fold(Fun,Acc,Table,Start,Count,Direction,Driver) ->
  158. RecordType = table_type(Table),
  159. %io:format("fold: ~p~n",[{RecordType, Start, Driver}]),
  160. case kvs:get(RecordType, Start, Driver) of
  161. {ok, R} -> Prev = element(Direction, R),
  162. Count1 = case Count of C when is_integer(C) -> C - 1; _-> Count end,
  163. fold(Fun, Fun(R,Acc), Table, Prev, Count1, Direction, Driver);
  164. Error -> kvs:error(?MODULE,"Error: ~p~n",[Error]), Acc end.
  165. entries({error,_},_,_,_) -> [];
  166. entries({ok,Container},N,C,Driver) -> entries(Container,N,C,Driver);
  167. entries(T,N,C,Driver) -> traversal(N,element(#container.top,T),C,#iterator.prev,Driver).
  168. entries(N, Start, Count, Direction, Driver) ->
  169. E = traversal(N, Start, Count, Direction, Driver),
  170. case Direction of #iterator.next -> lists:reverse(E);
  171. #iterator.prev -> E end.
  172. add_seq_ids() ->
  173. Init = fun(Key) ->
  174. case kvs:get(id_seq, Key) of
  175. {error, _} -> {Key,kvs:put(#id_seq{thing = Key, id = 0})};
  176. {ok, _} -> {Key,skip} end end,
  177. [ Init(atom_to_list(Name)) || {Name,_Fields} <- containers() ].
  178. put(Record,#kvs{mod=Mod}) ->
  179. case range(element(1,Record),element(2,Record)) of
  180. [] -> Mod:put(Record);
  181. Name -> Mod:put(setelement(1,Record,Name)) end.
  182. table_type(A) -> kvs:rname(A).
  183. range(RecordName,Id) -> Ranges = kvs:config(RecordName), find(Ranges,RecordName,Id).
  184. find([],_,_Id) -> [];
  185. find([Range|T],RecordName,Id) ->
  186. case lookup(Range,Id) of
  187. [] -> find(T,RecordName,Id);
  188. Name -> Name end.
  189. lookup(#interval{left=Left,right=Right,name=Name},Id) when Id =< Right, Id >= Left -> Name;
  190. lookup(#interval{},_Id) -> [].
  191. get(RecordName, Key, #kvs{mod=Mod}) ->
  192. case range(RecordName,Key) of
  193. [] -> Mod:get(RecordName, Key);
  194. Name -> case Mod:get(Name, Key) of
  195. {ok,Record} -> {ok,setelement(1,Record,RecordName)};
  196. Else -> Else end end.
  197. count(RecordName,#kvs{mod=DBA}) -> DBA:count(RecordName).
  198. all(RecordName,#kvs{mod=DBA}) -> DBA:all(RecordName).
  199. index(RecordName, Key, Value,#kvs{mod=DBA}) -> DBA:index(RecordName, Key, Value).
  200. next_id(RecordName, Incr,#kvs{mod=DBA}) -> DBA:next_id(RecordName, Incr).
  201. save_db(Path) ->
  202. Data = lists:append([all(B) || B <- [list_to_atom(Name) || {table,Name} <- kvs:dir()] ]),
  203. kvs:save(Path, Data).
  204. load_db(Path) ->
  205. add_seq_ids(),
  206. AllEntries = kvs:load(Path),
  207. [kvs:put(E) || E <- lists:filter(fun(E) -> is_tuple(E) end ,AllEntries)].
  208. save(Dir, Value) ->
  209. filelib:ensure_dir(Dir),
  210. file:write_file(Dir, term_to_binary(Value)).
  211. load(Key) ->
  212. {ok, Bin} = file:read_file(Key),
  213. binary_to_term(Bin).
  214. notify(_EventPath, _Data) -> skip.
  215. config(Key) -> config(kvs, Key, "").
  216. config(App,Key) -> config(App,Key, "").
  217. config(App, Key, Default) -> case application:get_env(App,Key) of
  218. undefined -> Default;
  219. {ok,V} -> V end.
  220. log_modules() -> [].
  221. -define(ALLOWED, (config(kvs,log_modules,kvs))).
  222. log(Module, String, Args, Fun) ->
  223. case lists:member(Module,?ALLOWED:log_modules()) of
  224. true -> error_logger:Fun("~p:"++String, [Module|Args]);
  225. false -> skip end.
  226. info(Module, String, Args) -> log(Module, String, Args, info_msg).
  227. warning(Module,String, Args) -> log(Module, String, Args, warning_msg).
  228. error(Module, String, Args) -> log(Module, String, Args, error_msg).
  229. dump() ->
  230. io:format("~20w ~20w ~10w ~10w~n",[name,storage_type,memory,size]),
  231. [ io:format("~20w ~20w ~10w ~10w~n",[Name,
  232. mnesia:table_info(Name,storage_type),
  233. mnesia:table_info(Name,memory),
  234. mnesia:table_info(Name,size)]) || #table{name=Name} <- kvs:tables()],
  235. io:format("Snapshot taken: ~p~n",[calendar:now_to_datetime(os:timestamp())]).
  236. % Table Partitions
  237. nonexistent() -> [ T || #table{name=T} <- kvs:tables(), kvs:info(T) == [] ].
  238. rotate_new(Tables) -> kvs:info(?MODULE,"New Tables: ~p~n",[Tables]), [ kvs:rotate1(kvs:table(T)) || T<- Tables].
  239. rotate1(Table) -> update_config(rname(Table#table.name),Table#table.name),
  240. kvs:put(#id_seq{thing=lists:concat([rname(Table#table.name),".tables"]),id=wf:to_integer(nname(Table))}).
  241. load_partitions() -> [ case kvs:get(config,Table) of
  242. {ok,{config,_,List}} -> application:set_env(kvs,Table,List);
  243. Else -> ok end || {table,Table} <- kvs:dir() ].
  244. limit() -> 10000000000000000000.
  245. store(Table,X) -> application:set_env(kvs,Table,X), X.
  246. last_table(Table) -> list_to_atom(lists:concat([Table,id_seq(lists:concat([Table,".tables"]))])).
  247. cname(Table) -> list_to_atom(lists:concat([Table,id_seq(lists:concat([Table,".tables"]))-1])).
  248. rname(Table) -> list_to_atom(lists:filter(fun(X) -> not lists:member(X,"1234567890") end, atom_to_list(Table))).
  249. nname(Table) -> list_to_atom(lists:filter(fun(X) -> lists:member(X,"1234567890") end, atom_to_list(Table#table.name))).
  250. fold(N) -> kvs:fold(fun(X,A)->[X|A]end,[],process,N,-1,#iterator.next,#kvs{mod=store_mnesia}).
  251. top(Table) -> id_seq(Table).
  252. name(T) -> list_to_atom(lists:concat([T,kvs:next_id(lists:concat([T,".tables"]),1)])).
  253. init(T) -> store_mnesia:create_table(T#table.name, [{attributes,T#table.fields},{T#table.copy_type, [node()]}]),
  254. [ store_mnesia:add_table_index(T#table.name, Key) || Key <- T#table.keys ].
  255. id_seq(Tab) -> T = atom_to_list(Tab), case kvs:get(id_seq,T) of {ok,#id_seq{id=Id}} -> Id; _ -> kvs:next_id(T,1) end.
  256. % rotate DETS table
  257. interval(L,R,Name) -> #interval{left=L,right=R,name=Name}.
  258. rotate(Table) -> Name = name(Table), init(setelement(#table.name,kvs:table(Table),Name)), update_config(rname(Table),Name).
  259. update_config(Table,Name) ->
  260. kvs:put(#config{key = Table,
  261. value = store(Table,case kvs:get(config,Table) of
  262. {error,not_found} -> update_list(Table,[],Name);
  263. {ok,#config{value=List}} -> update_list(Table,List,Name) end)}).
  264. update_list(Table,List,Name) ->
  265. io:format("Name ~p Table ~p ~n",[Name,Table]),
  266. [ interval(top(Table)+1,limit(),Name) ] ++
  267. case lists:keyfind(Name,#interval.name,List) of
  268. false -> List;
  269. CI -> lists:keyreplace(Name,#interval.name,List,CI#interval{right=top(Table)}) end.