<commit_before>%%%-------------------------------------------------------------------
%% @doc edht top level supervisor.
%% @end
%%%-------------------------------------------------------------------

-module(edht_sup).
-behaviour(supervisor).

%% API
-export([start_link/0]).
-export([client/5]).

%% Supervisor callbacks
-export([init/1]).

-define(SERVER, ?MODULE).
-define(P, io:format).

%% Node config keys
-define(DATADIR, ".bitcaskdata").
-define(CLIENT_PORT, "client_port").
-define(NODE_PORT, "node_port").
-define(NODE_IPS, "node_ips").
-define(NAME, "name").
-define(REPLICATION, "replication").
-define(NODE_TIMEOUT_MS, "node_timeout_ms").

-record(client_request, {method, key, value=undefined}).
-record(node_request, {type, client_request}).

%%====================================================================
%% API functions
%%====================================================================

start_link() ->
    supervisor:start_link({local, ?SERVER}, ?MODULE, []).

%%====================================================================
%% Supervisor callbacks
%%====================================================================

%% Child :: {Id,StartFunc,Restart,Shutdown,Type,Modules}
init(Args) ->
    io:format('Starting EDHT with: ~p~n', [Args]),
    % load protobuf file for encoding and decoding
    protobuffs_compile:scan_file("src/request.proto"),

    % read config for listening on different ports
    ConfigItemTypeMap = #{?CLIENT_PORT => fun list_to_integer/1, 
                          ?NODE_PORT => fun list_to_integer/1,
                          ?NODE_IPS => fun convert_ips_to_tuple/1,
                          ?REPLICATION => fun list_to_integer/1,
                          ?NODE_TIMEOUT_MS => fun list_to_integer/1,
                          ?NAME => fun(X) -> X end},
    Config = read_config(),
    Map = lists:foldl(fun(Pair, Acc) -> pick(Pair, Acc, ConfigItemTypeMap) end, maps:new(), Config),
    register(config_manager, spawn_link(fun() -> config_manager(Map) end)),

    % Start proc to listen to client
    spawn_link(fun() -> client_listener_loop(get_config(?CLIENT_PORT)) end),

    % Arbiter which handles local io
    register(arbiter, spawn_link(fun() -> arbiter_loop(0) end)),

    % listen to internode communication
    spawn_link(fun() -> listen_to_nodes() end),

    % proc to keep track of nodes
    NodeIPs = get_config(?NODE_IPS),
    io:format("Node IPs: ~p~n", [NodeIPs]),
    NodeRing = concha:new(5, NodeIPs),
    register(node_hash_lookup, spawn_link(fun() -> node_hash_lookup(NodeRing) end)),
    {ok, { {one_for_all, 0, 1}, []} }.

%%====================================================================
%% Internal functions
%%====================================================================

pick(Pair, Acc, ConfigItemTypeMap) ->
    {Key, Val} = Pair,
    Fun = maps:get(Key, ConfigItemTypeMap),
    maps:put(Key, Fun(Val), Acc).

%% Process which return config values to other process
config_manager(Map) ->
    receive
        {From, Key} ->
            From ! maps:get(Key, Map),
            config_manager(Map)
    end.

%% Config value coordinator
get_config(Key) ->
    %% Function internally calls config_manager to pull the config value
    config_manager ! {self(), Key},
    receive
        Val ->
            Val
    end.

%% Convert IP:Port string into {Host, Port}
ip_string_to_tuple(Str) ->
    Token = string:tokens(Str, ":"),
    Octets = string:tokens(hd(Token), "."),
    IPTuple = list_to_tuple(lists:map(fun(Octet) -> list_to_integer(Octet) end, Octets)),
    {IPTuple, list_to_integer(lists:nth(2, Token))}.

convert_ips_to_tuple(IPString) ->
    %% Convert list of ips into [{{Host}, Port}]
    IPList = string:tokens(IPString, ","),
    lists:map(fun(IP) -> ip_string_to_tuple(IP) end, IPList).

%% Erlang process responsible for key node lookup
node_hash_lookup(NodeRing) ->
    %% Function to keep track of ring and return node[s] responsible for the key
    receive
        {From, lookup, Key} ->
            Node = concha:lookup(Key, NodeRing),
            From ! {lookup, Key, Node, self()},
            node_hash_lookup(NodeRing);
        {From, successor, Key, Length} ->
            Nodes = successor(NodeRing, Key, Length),
            From ! {successor, Nodes},
            node_hash_lookup(NodeRing)
        end.

%% Peristance/lookup from stored file
arbiter_loop(Ref) ->
    receive
        {Caller, Op, Key, Val} ->
            io:format('Received data for lookup ~p, ~p, ~p~n', [Op, Key, Val]),
            Parent = self(),
            spawn_link(fun() -> handle_io(Parent, Op, Key, Val) end),
            arbiter_loop(Caller);
        {Op, Key, Val} ->
            io:format('Sending data after lookup ~p, ~p, ~p~n', [Op, Key, Val]),
            Ref ! {arbiter, Op, Key, Val},
            arbiter_loop(Ref);
        Pat -> io:format("failed the pattern, ~p ~n", [Pat])
        end.

handle_node_get(Parent, DBHandle, Key) ->
    BinaryKey = to_binary(Key),
    case bitcask:get(DBHandle, BinaryKey) of
        {ok, Value} ->
            Parent ! {"GET", Key, Value};
        not_found ->
            Parent ! {"GET", Key, undefined}
    end.

handle_node_put(Parent, DBHandle, Key, Val) ->
    BinaryKey = to_binary(Key),
    BinaryVal = to_binary(Val),
    case bitcask:put(DBHandle, BinaryKey, BinaryVal) of
        ok ->
            Parent ! {"PUT", Key, Val};
        {error, _} ->
            Parent ! {"PUT", Key, error}
    end.

handle_io(Parent, Op, Key, Val) ->
    DBHandle = open_store(),
    if 
        Op =:= "GET" ->
            handle_node_get(Parent, DBHandle, Key);
        Op =:= "PUT" ->
            handle_node_put(Parent, DBHandle, Key, Val);
        true -> Parent ! {Op, Key, Val}
    end,
    bitcask:close(DBHandle).

read_config() ->
    application:ensure_all_started(econfig),<commit_msg>Add sleep for handling concurrent update request<commit_after>%%%-------------------------------------------------------------------
%% @doc edht top level supervisor.
%% @end
%%%-------------------------------------------------------------------

-module(edht_sup).
-behaviour(supervisor).

%% API
-export([start_link/0]).
-export([client/5]).

%% Supervisor callbacks
-export([init/1]).

-define(SERVER, ?MODULE).
-define(P, io:format).

%% Node config keys
-define(DATADIR, ".bitcaskdata").
-define(CLIENT_PORT, "client_port").
-define(NODE_PORT, "node_port").
-define(NODE_IPS, "node_ips").
-define(NAME, "name").
-define(REPLICATION, "replication").
-define(NODE_TIMEOUT_MS, "node_timeout_ms").

-record(client_request, {method, key, value=undefined}).
-record(node_request, {type, client_request}).

%%====================================================================
%% API functions
%%====================================================================

start_link() ->
    supervisor:start_link({local, ?SERVER}, ?MODULE, []).

%%====================================================================
%% Supervisor callbacks
%%====================================================================

%% Child :: {Id,StartFunc,Restart,Shutdown,Type,Modules}
init(Args) ->
    io:format('Starting EDHT with: ~p~n', [Args]),
    % load protobuf file for encoding and decoding
    protobuffs_compile:scan_file("src/request.proto"),

    % read config for listening on different ports
    ConfigItemTypeMap = #{?CLIENT_PORT => fun list_to_integer/1, 
                          ?NODE_PORT => fun list_to_integer/1,
                          ?NODE_IPS => fun convert_ips_to_tuple/1,
                          ?REPLICATION => fun list_to_integer/1,
                          ?NODE_TIMEOUT_MS => fun list_to_integer/1,
                          ?NAME => fun(X) -> X end},
    Config = read_config(),
    Map = lists:foldl(fun(Pair, Acc) -> pick(Pair, Acc, ConfigItemTypeMap) end, maps:new(), Config),
    register(config_manager, spawn_link(fun() -> config_manager(Map) end)),

    % Start proc to listen to client
    spawn_link(fun() -> client_listener_loop(get_config(?CLIENT_PORT)) end),

    % Arbiter which handles local io
    register(arbiter, spawn_link(fun() -> arbiter_loop(0) end)),

    % listen to internode communication
    spawn_link(fun() -> listen_to_nodes() end),

    % proc to keep track of nodes
    NodeIPs = get_config(?NODE_IPS),
    io:format("Node IPs: ~p~n", [NodeIPs]),
    NodeRing = concha:new(5, NodeIPs),
    register(node_hash_lookup, spawn_link(fun() -> node_hash_lookup(NodeRing) end)),
    {ok, { {one_for_all, 0, 1}, []} }.

%%====================================================================
%% Internal functions
%%====================================================================

pick(Pair, Acc, ConfigItemTypeMap) ->
    {Key, Val} = Pair,
    Fun = maps:get(Key, ConfigItemTypeMap),
    maps:put(Key, Fun(Val), Acc).

%% Process which return config values to other process
config_manager(Map) ->
    receive
        {From, Key} ->
            Val = maps:get(Key, Map),
            From ! {config_manager, Val},
            config_manager(Map)
    end.

%% Config value coordinator
get_config(Key) ->
    %% Function internally calls config_manager to pull the config value
    config_manager ! {self(), Key},
    receive
        {config_manager, Val} ->
            Val
    end.

%% Convert IP:Port string into {Host, Port}
ip_string_to_tuple(Str) ->
    Token = string:tokens(Str, ":"),
    Octets = string:tokens(hd(Token), "."),
    IPTuple = list_to_tuple(lists:map(fun(Octet) -> list_to_integer(Octet) end, Octets)),
    {IPTuple, list_to_integer(lists:nth(2, Token))}.

convert_ips_to_tuple(IPString) ->
    %% Convert list of ips into [{{Host}, Port}]
    IPList = string:tokens(IPString, ","),
    lists:map(fun(IP) -> ip_string_to_tuple(IP) end, IPList).

%% Erlang process responsible for key node lookup
node_hash_lookup(NodeRing) ->
    %% Function to keep track of ring and return node[s] responsible for the key
    receive
        {From, lookup, Key} ->
            Node = concha:lookup(Key, NodeRing),
            From ! {lookup, Key, Node, self()},
            node_hash_lookup(NodeRing);
        {From, successor, Key, Length} ->
            Nodes = successor(NodeRing, Key, Length),
            From ! {successor, Nodes},
            node_hash_lookup(NodeRing)
        end.

%% Peristance/lookup from stored file
arbiter_loop(Ref) ->
    receive
        {Caller, Op, Key, Val} ->
            io:format('Received data for lookup ~p, ~p, ~p~n', [Op, Key, Val]),
            Parent = self(),
            spawn_link(fun() -> handle_io(Parent, Op, Key, Val) end),
            arbiter_loop(Caller);
        {Op, Key, Val} ->
            io:format('Sending data after lookup ~p, ~p, ~p~n', [Op, Key, Val]),
            Ref ! {arbiter, Op, Key, Val},
            arbiter_loop(Ref);
        Pat -> io:format("failed the pattern, ~p ~n", [Pat])
        end.

handle_node_get(Parent, DBHandle, Key) ->
    BinaryKey = to_binary(Key),
    case bitcask:get(DBHandle, BinaryKey) of
        {ok, Value} ->
            Parent ! {"GET", Key, Value};
        not_found ->
            Parent ! {"GET", Key, undefined}
    end.

handle_node_put(Parent, DBHandle, Key, Val) ->
    BinaryKey = to_binary(Key),
    BinaryVal = to_binary(Val),
    case bitcask:put(DBHandle, BinaryKey, BinaryVal) of
        ok ->
            Parent ! {"PUT", Key, Val};
        {error, Reason} ->
            ?P("Put failed for key: ~p for ~p~n", [BinaryKey, Reason]),
            Parent ! {"PUT", Key, error}
    end.

handle_io(Parent, Op, Key, Val) ->
    case Op of
        "GET" ->
            DBHandle = open_store(),
            handle_node_get(Parent, DBHandle, Key),
            bitcask:close(DBHandle);
        "PUT" ->
            DBHandle = open_store(),
            % bitcask writes data to disk. Every key is mapped to one file.
            % So updating same key across multiple process requires acquiring the lock.
            % Inorder to handle atleast 100 requests or so for same key sleep is used.
            timer:sleep(10),
            handle_node_put(Parent, DBHandle, Key, Val),
            bitcask:close(DBHandle);
        true -> Parent ! {Op, Key, Val}
    end.

read_config() ->
    application:ensure_all_started(econfig),
{'fraction_non_alphanumeric': 0.2220314033366045, 'fraction_numerical': 0.0025351651946352634, 'mean_word_length': 2.810844499844188, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2436029', 'n_tokens_mistral': 3730, 'n_tokens_neox': 3607, 'n_words': 892}