<commit_before>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2017 ACK CYFRONET AGH
%%% This software is released under the MIT license
%%% cited in 'LICENSE.txt'.
%%% @end
%%%-------------------------------------------------------------------
%%% @doc
%%% This module provides an API for datastore cache. Datastore cache consists
%%% of two components: memory and disc store. Operations on memory take
%%% precedence over disc, as long as memory limit is not reached. For each
%%% successful operation along with result a durability level is returned.
%%% It denotes which layer, memory or disc, has been used. An exception
%%% constitute {@link get/2} and {@link flush/2} functions, which are meant
%%% to operate in memory and on disc accordingly and therefore do not return
%%% durability level.
%%% @end
%%%-------------------------------------------------------------------
-module(datastore_cache).
-author("<NAME>").

-include("modules/datastore/datastore_models_def.hrl").

%% API
-export([get/2, fetch/2, save/2, update/2, update/3, flush/2, flush/1]).
-export([inactivate/1]).

-type ctx() :: #{prefix => binary(),
                 mutator_pid => pid(),
                 volatile => boolean(),
                 memory_driver => datastore:memory_driver(),
                 memory_driver_ctx => datastore:memory_driver_ctx(),
                 disc_driver => datastore:disc_driver(),
                 disc_driver_ctx => datastore:disc_driver_ctx(),
                % TODO - remove when driver and datastore ctx are separated
                 atom() => term()}.
-type key() :: datastore:key().
-type value() :: datastore:document().
-type diff() :: fun((value()) -> {ok, value()} | {error, term()}).
-type durability() :: memory | disc.

-export_type([ctx/0, diff/0, durability/0]).

-record(future, {
    durability :: undefined | durability(),
    driver :: undefined | datastore:driver(),
    value :: {ok, value()} | {error, term()} | couchbase_pool:future()
}).

-type future() :: #future{}.

-define(FUTURE(Value), ?FUTURE(undefined, undefined, Value)).
-define(FUTURE(Durability, Driver, Value), #future{
    durability = Durability,
    driver = Driver,
    value = Value
}).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Retrieves values from memory only.
%% @end
%%--------------------------------------------------------------------
-spec get(ctx(), key()) -> {ok, value()} | {error, term()};
    (ctx(), [datastore:key()]) -> [{ok, value()} | {error, term()}].
get(Ctx, <<_/binary>> = Key) ->
    hd(get(Ctx, [Key]));
get(Ctx, Keys) when is_list(Keys) ->
    lists:map(fun
        ({ok, memory, Value}) -> {ok, Value};
        ({error, Reason}) -> {error, Reason}
    end, wait([get_async(Ctx, Key, false) || Key <- Keys])).

%%--------------------------------------------------------------------
%% @doc
%% Retrieves values from memory or if missing from disc. For values read from
%% disc, an attempt is made to store them in memory. If cache update fails with
%% out of memory error, the returned durability level equals to 'disc',
%% otherwise it is set to 'memory'.
%% @end
%%--------------------------------------------------------------------
-spec fetch(ctx(), key()) -> {ok, durability(), value()} | {error, term()};
    (ctx(), [key()]) -> [{ok, durability(), value()} | {error, term()}].
fetch(Ctx, <<_/binary>> = Key) ->
    hd(fetch(Ctx, [Key]));
fetch(#{memory_driver := MemoryDriver} = Ctx, Keys) when is_list(Keys) ->
    Futures = lists:map(fun
        ({ok, memory, Value}) -> ?FUTURE(memory, MemoryDriver, {ok, Value});
        ({ok, disc, Value}) -> save_async(Ctx, Value, false);
        ({error, Reason}) -> ?FUTURE({error, Reason})
    end, wait([get_async(Ctx, Key, true) || Key <- Keys])),

    lists:map(fun
        ({ok, memory, Value}) -> {ok, memory, Value};
        ({error, {enomem, Value}}) -> {ok, disc, Value};
        ({error, Reason}) -> {error, Reason}
    end, wait(Futures)).
<commit_msg>VFS-3567 Store missing documents in datastore cache<commit_after>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2017 ACK CYFRONET AGH
%%% This software is released under the MIT license
%%% cited in 'LICENSE.txt'.
%%% @end
%%%-------------------------------------------------------------------
%%% @doc
%%% This module provides an API for datastore cache. Datastore cache consists
%%% of two components: memory and disc store. Operations on memory take
%%% precedence over disc, as long as memory limit is not reached. For each
%%% successful operation along with result a durability level is returned.
%%% It denotes which layer, memory or disc, has been used. An exception
%%% constitute {@link get/2} and {@link flush/2} functions, which are meant
%%% to operate in memory and on disc accordingly and therefore do not return
%%% durability level.
%%% @end
%%%-------------------------------------------------------------------
-module(datastore_cache).
-author("<NAME>").

-include("modules/datastore/datastore_models_def.hrl").

%% API
-export([get/2, fetch/2, save/2, update/2, update/3, flush/2, flush/1]).
-export([inactivate/1]).

-type ctx() :: #{prefix => binary(),
                 mutator_pid => pid(),
                 volatile => boolean(),
                 memory_driver => datastore:memory_driver(),
                 memory_driver_ctx => datastore:memory_driver_ctx(),
                 disc_driver => datastore:disc_driver(),
                 disc_driver_ctx => datastore:disc_driver_ctx(),
                % TODO - remove when driver and datastore ctx are separated
                 atom() => term()}.
-type key() :: datastore:key().
-type value() :: datastore:document().
-type diff() :: fun((value()) -> {ok, value()} | {error, term()}).
-type durability() :: memory | disc.

-export_type([ctx/0, diff/0, durability/0]).

-record(future, {
    durability :: undefined | durability(),
    driver :: undefined | datastore:driver(),
    value :: {ok, value()} | {error, term()} | couchbase_pool:future()
}).

-type future() :: #future{}.

-define(FUTURE(Value), ?FUTURE(undefined, undefined, Value)).
-define(FUTURE(Durability, Driver, Value), #future{
    durability = Durability,
    driver = Driver,
    value = Value
}).

%%%===================================================================
%%% API
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Retrieves values from memory only.
%% @end
%%--------------------------------------------------------------------
-spec get(ctx(), key()) -> {ok, value()} | {error, term()};
    (ctx(), [datastore:key()]) -> [{ok, value()} | {error, term()}].
get(Ctx, <<_/binary>> = Key) ->
    hd(get(Ctx, [Key]));
get(Ctx, Keys) when is_list(Keys) ->
    lists:map(fun
        ({ok, memory, Value}) -> {ok, Value};
        ({error, Reason}) -> {error, Reason}
    end, wait([get_async(Ctx, Key, false) || Key <- Keys])).

%%--------------------------------------------------------------------
%% @doc
%% Retrieves values from memory or if missing from disc. For values read from
%% disc, an attempt is made to store them in memory. If cache update fails with
%% out of memory error, the returned durability level equals to 'disc',
%% otherwise it is set to 'memory'.
%% @end
%%--------------------------------------------------------------------
-spec fetch(ctx(), key()) -> {ok, durability(), value()} | {error, term()};
    (ctx(), [key()]) -> [{ok, durability(), value()} | {error, term()}].
fetch(Ctx, <<_/binary>> = Key) ->
    hd(fetch(Ctx, [Key]));
fetch(#{memory_driver := MemoryDriver} = Ctx, Keys) when is_list(Keys) ->
    Futures = lists:map(fun
        ({_Key, {ok, memory, Value}}) ->
            ?FUTURE(memory, MemoryDriver, {ok, Value});
        ({_Key, {ok, disc, Value}}) ->
            save_async(Ctx, Value, false);
        ({Key, {error, key_enoent}}) ->
            Value = #document{key = Key, value = undefined, deleted = true},
            save_async(Ctx, Value, false),
            ?FUTURE({error, key_enoent});
        ({_Key, {error, Reason}}) ->
            ?FUTURE({error, Reason})
    end, lists:zip(Keys, wait([get_async(Ctx, Key, true) || Key <- Keys]))),

    lists:map(fun
        ({ok, memory, Value}) -> {ok, memory, Value};
        ({error, {enomem, Value}}) -> {ok, disc, Value};
        ({error, Reason}) -> {error, Reason}
    end, wait(Futures)).

{'fraction_non_alphanumeric': 0.3017568033069239, 'fraction_numerical': 0.004363302330922035, 'mean_word_length': 2.2463660081997765, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 66, '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': '2223634', 'n_tokens_mistral': 2444, 'n_tokens_neox': 2394, 'n_words': 739}