Packages

kvs

9.4.8
13.5.22-aleph 13.4.16 13.4.15 13.4.14 13.4.13 13.3.1 13.2.28 11.9.1 10.8.3 10.8.2 10.3.0 9.9.2 9.9.1 9.9.0 9.8.0 9.7.0 9.4.8 9.4.7 9.4.6 9.4.5 9.4.4 9.4.3 9.4.2 9.4.1 9.4.0 8.12.0 8.11.2 8.11.1 8.10.4 8.10.3 8.10.2 8.10.1 8.10.0 8.5.2 8.5.1 8.5.0 8.4.1 8.4.0 8.3.1 8.3.0 7.11.5 7.9.1 7.7.0 7.1.3 7.1.2 7.1.1 6.12.11 6.12.10 6.12.9 6.12.8 6.12.7 6.12.6 6.12.5 6.12.4 6.12.3 6.12.2 6.12.1 6.12.0 6.11.2 6.11.1 6.11.0 6.10.2 6.10.1 6.10.0 6.9.2 6.9.1 6.9.0 6.7.7 6.7.6 6.7.5 6.7.4 6.7.3 6.7.2 6.7.1 6.7.0 6.6.0 2.1.0 0.12.1 retired

KVS Key-Value Store Abstraction Layer

HexDocs

Current section

Files

Jump to
Readme
78 Versions
0 Dependencies
Dependants
Files
Activity
kvs src stores kvs_mnesia.erl
Raw

src/stores/kvs_mnesia.erl 

-module(kvs_mnesia).

-include("backend.hrl").

-include("kvs.hrl").

-include("metainfo.hrl").

-include_lib("mnesia/src/mnesia.hrl").

-include_lib("stdlib/include/qlc.hrl").

-export(?BACKEND).

-export([info/1,exec/1,dump/1]).



db()       -> "".

seq_pad()  -> application:get_env(kvs, seq_pad, []).

start()    -> mnesia:start().

stop()     -> mnesia:stop().

destroy()  -> [mnesia:delete_table(T)||{_,T}<-kvs:dir()], mnesia:delete_schema([node()]), ok.

destroy(_) -> [mnesia:delete_table(T)||{_,T}<-kvs:dir()], mnesia:delete_schema([node()]), ok.

leave()    -> ok.

leave(_)   -> ok.

version()  -> {version,"KVS MNESIA"}.

dir()      -> [{table,T}||T<-mnesia:system_info(local_tables)].

join([], _)   -> mnesia:start(), mnesia:change_table_copy_type(schema, node(), disc_copies), initialize();



join(Node, _) ->

    mnesia:start(),

    mnesia:change_config(extra_db_nodes, [Node]),

    mnesia:change_table_copy_type(schema, node(), disc_copies),

    [{Tb, mnesia:add_table_copy(Tb, node(), Type)}

     || {Tb, [{N, Type}]} <- [{T, mnesia:table_info(T, where_to_commit)}

                               || T <- mnesia:system_info(tables)], Node==N].



initialize() ->

    mnesia:create_schema([node()]),

    Res = [ kvs:initialize(kvs_mnesia,Module) || Module <- kvs:modules() ],

    mnesia:wait_for_tables([ T#table.name || T <- kvs:tables()],infinity),

    Res.



index(Tab,Key,Value) ->

    lists:flatten(many(fun() -> mnesia:index_read(Tab,Value,Key) end)).



get(RecordName, Key, _) -> just_one(fun() -> mnesia:read(RecordName, Key) end).

put(R)                  -> put(R,db()).

put(Records, _) when is_list(Records) -> void(fun() -> lists:foreach(fun mnesia:write/1, Records) end);

put(Record, X) -> put([Record], X).

delete(Tab, Key, _) ->

    case mnesia:activity(context(),fun()-> mnesia:delete({Tab, Key}) end) of

        {aborted,Reason} -> {error,Reason};

        {atomic,_Result} -> ok;

        _ -> ok end.

match(Record) -> lists:flatten(many(fun() -> mnesia:match_object(Record) end)).

index_match(Record, Index) -> lists:flatten(many(fun() -> mnesia:index_match_object(Record, Index) end)).

count(RecordName) -> mnesia:table_info(RecordName, size).

all(R, _) -> lists:flatten(many(fun() -> L= mnesia:all_keys(R), [ mnesia:read({R, G}) || G <- L ] end)).

seq(RecordName, []) -> seq(RecordName, 1);

seq(RecordName, Incr) ->

  Val = integer_to_list(mnesia:dirty_update_counter({id_seq, RecordName}, Incr)),

  case (20 - length(Val) > 0) and lists:member(RecordName, seq_pad()) of true -> string:copies("0", 20 - length(Val)); _ -> "" end ++ Val.



many(Fun) -> case mnesia:activity(context(),Fun) of {atomic, [R]} -> R; {aborted, Error} -> {error, Error}; X -> X end.

void(Fun) -> case mnesia:activity(context(),Fun) of {atomic, ok} -> ok; {aborted, Error} -> {error, Error}; X -> X end.

info(T) -> try mnesia:table_info(T,all) catch _:_ -> [] end.

create_table(Name,Options) -> mnesia:create_table(Name, Options).

add_table_index(Record, Field) -> mnesia:add_table_index(Record, Field).

exec(Q) -> F = fun() -> qlc:e(Q) end, {atomic, Val} = mnesia:activity(context(),F), Val.

just_one(Fun) ->

    case mnesia:activity(context(),Fun) of

        {atomic, []} -> {error, not_found};

        {atomic, [R]} -> {ok, R};

        [] -> {error, not_found};

        [R] -> {ok,R};

        R when is_list(R) -> {ok,R};

        Error -> Error end.



context() -> application:get_env(kvs,mnesia_context,async_dirty).



dump() -> dump([ N || #table{name=N} <- kvs:tables() ]), ok.

dump(short) ->

    Gen = fun(T) ->

        {S,M,C}=lists:unzip3([ dump_info(R) || R <- T ]),

        {lists:usort(S),lists:sum(M),lists:sum(C)}

    end,

    dump_format([ {T,Gen(T)} || T <- [ N || #table{name=N} <- kvs:tables() ] ]);

dump(Table) when is_atom(Table) -> dump(Table);

dump(Tables) ->

    dump_format([{T,dump_info(T)} || T <- lists:flatten(Tables) ]).

dump_info(T) ->

    {mnesia:table_info(T,storage_type),

    mnesia:table_info(T,memory) * erlang:system_info(wordsize) / 1024 / 1024,

    mnesia:table_info(T,size)}.

dump_format(List) ->

    io:format("~20s ~32s ~14s ~10s~n~n",["NAME","STORAGE TYPE","MEMORY (MB)","ELEMENTS"]),

    [ io:format("~20s ~32w ~14.2f ~10b~n",[T,S,M,C]) || {T,{S,M,C}} <- List ],

    io:format("~nSnapshot taken: ~p~n",[calendar:now_to_datetime(os:timestamp())]).