You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2022/11/04 18:33:01 UTC
[couchdb] branch optimize-smoosh created (now d7feed4ac)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch optimize-smoosh
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at d7feed4ac Improve smoosh_priority_queue
This branch includes the following new commits:
new b19794c5a Add a new smoosh persistence module
new 2ff470279 Improve smoosh_util
new d7feed4ac Improve smoosh_priority_queue
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[couchdb] 02/03: Improve smoosh_util
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch optimize-smoosh
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 2ff47027903f3f28a4e53a1070ed846cbccf9005
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Fri Nov 4 00:59:47 2022 -0400
Improve smoosh_util
Add functions related to dealing with new index cleanup channel.
Add function which fetch the channels list from config. Fix an unfortunate
corner case where operators when creating custom smoosh channel may omit by
mistake some of the built-in channel like view upgrades. To improve the
behavior, make built-in channel always available. Channels configured by the
user are always appended to the list of built-in ones.
The new validate_arg/1 function is in charge of validation smoosh's input args.
It centralizes ?b2l, ?l2b and other tranforms and checks in one place so they
are not sprinkeled throughout the internal functions. After the validation
steps all the db and view names are guranteed to be binaries.
The most significant changes is probably addition of extra tests to get to 100%
test coverage. Since we cover all of the cases in time window checking, remove
the more heavyweight elixir test, since we now get a nice coverage report in
Erlang and the test is overall smaller as well.
---
src/smoosh/src/smoosh_utils.erl | 277 +++++++++++++++++-----
src/smoosh/test/exunit/scheduling_window_test.exs | 79 ------
src/smoosh/test/exunit/test_helper.exs | 2 -
3 files changed, 214 insertions(+), 144 deletions(-)
diff --git a/src/smoosh/src/smoosh_utils.erl b/src/smoosh/src/smoosh_utils.erl
index 1942aebf4..376cc4d08 100644
--- a/src/smoosh/src/smoosh_utils.erl
+++ b/src/smoosh/src/smoosh_utils.erl
@@ -14,8 +14,17 @@
-include_lib("couch/include/couch_db.hrl").
-export([get/2, get/3, split/1, stringify/1, ignore_db/1]).
--export([in_allowed_window/1, is_view_channel/1, write_to_file/3]).
+-export([db_channels/0, view_channels/0, cleanup_channels/0]).
+-export([in_allowed_window/1]).
+-export([concurrency/1, capacity/1]).
-export([log_level/2]).
+-export([validate_arg/1]).
+
+-define(BUILT_IN_DB_CHANNELS, "upgrade_dbs,ratio_dbs,slack_dbs").
+-define(BUILT_IN_VIEW_CHANNELS, "upgrade_views,ratio_views,slack_views").
+-define(BUILT_IN_CLEANUP_CHANNELS, "index_cleanup").
+
+-define(INDEX_CLEANUP, index_cleanup).
get(Channel, Key) ->
?MODULE:get(Channel, Key, undefined).
@@ -23,14 +32,22 @@ get(Channel, Key) ->
get(Channel, Key, Default) ->
config:get("smoosh." ++ Channel, Key, Default).
+split(undefined) ->
+ [];
+split("") ->
+ [];
split(CSV) ->
re:split(CSV, "\\s*,\\s*", [{return, list}, trim]).
+stringify({?INDEX_CLEANUP, DbName}) ->
+ io_lib:format("~s index_cleanup", [DbName]);
stringify({DbName, GroupId}) ->
io_lib:format("~s ~s", [DbName, GroupId]);
stringify(DbName) ->
io_lib:format("~s", [DbName]).
+ignore_db({?INDEX_CLEANUP, DbName}) ->
+ ignore_db(DbName);
ignore_db({DbName, _GroupName}) ->
ignore_db(DbName);
ignore_db(DbName) when is_binary(DbName) ->
@@ -41,13 +58,11 @@ ignore_db(DbName) when is_list(DbName) ->
true;
_ ->
false
- end;
-ignore_db(Db) ->
- ignore_db(couch_db:name(Db)).
+ end.
in_allowed_window(Channel) ->
- From = parse_time(get(Channel, "from"), {00, 00}),
- To = parse_time(get(Channel, "to"), {24, 00}),
+ From = parse_time(?MODULE:get(Channel, "from"), {00, 00}),
+ To = parse_time(?MODULE:get(Channel, "to"), {24, 00}),
in_allowed_window(From, To).
in_allowed_window(From, To) ->
@@ -59,63 +74,6 @@ in_allowed_window(From, To) ->
({HH, MM} >= From) orelse ({HH, MM} < To)
end.
-is_view_channel(ChannelName) ->
- ViewChannels = smoosh_utils:split(
- config:get("smoosh", "view_channels", "upgrade_views,ratio_views,slack_views")
- ),
- lists:member(ChannelName, ViewChannels).
-
-file_delete(Path) ->
- case file:delete(Path) of
- % When deleting a state file, we do not care if it does not exist.
- Ret when Ret =:= ok; Ret =:= {error, enoent} ->
- ok;
- {error, Reason} ->
- {error, Reason}
- end.
-
-write_to_file(Content, FileName, VSN) ->
- Level = log_level("compaction_log_level", "debug"),
- couch_log:Level("~p Writing state ~s", [?MODULE, FileName]),
- OnDisk = <<VSN, (erlang:term_to_binary(Content, [compressed, {minor_version, 1}]))/binary>>,
- TmpFileName = FileName ++ ".tmp",
- case file_delete(TmpFileName) of
- ok ->
- case file:write_file(TmpFileName, OnDisk, [sync]) of
- ok ->
- case file_delete(FileName) of
- ok ->
- case file:rename(TmpFileName, FileName) of
- ok ->
- ok;
- {error, Reason} ->
- couch_log:error(
- "~p (~p) Error renaming temporary state file ~s to ~s", [
- ?MODULE, Reason, TmpFileName, FileName
- ]
- ),
- ok
- end;
- {error, Reason} ->
- couch_log:error("~p (~p) Error deleting state file ~s", [
- ?MODULE, Reason, FileName
- ]),
- ok
- end;
- {error, Reason} ->
- couch_log:error("~p (~p) Error writing state to temporary file ~s", [
- ?MODULE, Reason, TmpFileName
- ]),
- ok
- end;
- {error, Reason} ->
- % Failing to persist the queue should not crash smoosh. Return ok.
- couch_log:error("~p (~p) Error deleting temporary state file ~s", [
- ?MODULE, Reason, TmpFileName
- ]),
- ok
- end.
-
parse_time(undefined, Default) ->
Default;
parse_time(String, Default) ->
@@ -135,3 +93,196 @@ parse_time(String, Default) ->
log_level(Key, Default) when is_list(Key), is_list(Default) ->
list_to_existing_atom(config:get("smoosh", Key, Default)).
+
+db_channels() ->
+ ConfStr = config:get("smoosh", "db_channels"),
+ channel_list(ConfStr, ?BUILT_IN_DB_CHANNELS).
+
+view_channels() ->
+ Conf = config:get("smoosh", "view_channels"),
+ channel_list(Conf, ?BUILT_IN_VIEW_CHANNELS).
+
+cleanup_channels() ->
+ Conf = config:get("smoosh", "cleanup_channels"),
+ channel_list(Conf, ?BUILT_IN_CLEANUP_CHANNELS).
+
+channel_list(ConfStr, Default) ->
+ DefaultList = split(Default),
+ ConfList = split(ConfStr),
+ lists:usort(DefaultList ++ ConfList).
+
+concurrency(ChannelName) ->
+ list_to_integer(?MODULE:get(ChannelName, "concurrency", "1")).
+
+capacity(ChannelName) ->
+ list_to_integer(?MODULE:get(ChannelName, "capacity", "9999")).
+
+% Validate enqueue arg at the front of the API instead of adding ?l2b and ?b2l
+% everywhere internally
+%
+validate_arg({?INDEX_CLEANUP, DbName}) when is_list(DbName) ->
+ validate_arg({?INDEX_CLEANUP, ?l2b(DbName)});
+validate_arg(DbName) when is_list(DbName) ->
+ validate_arg(?l2b(DbName));
+validate_arg({DbName, GroupId}) when is_list(DbName) ->
+ validate_arg({?l2b(DbName), GroupId});
+validate_arg({DbName, GroupId}) when is_list(GroupId) ->
+ validate_arg({DbName, ?l2b(GroupId)});
+validate_arg(DbName) when is_binary(DbName) ->
+ DbName;
+validate_arg({DbName, GroupId}) when is_binary(DbName), is_binary(GroupId) ->
+ {DbName, GroupId};
+validate_arg({?INDEX_CLEANUP, DbName}) when is_binary(DbName) ->
+ {?INDEX_CLEANUP, DbName};
+validate_arg(_) ->
+ error(invalid_smoosh_arg).
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+smoosh_util_validate_test() ->
+ ?assertEqual(<<"x">>, validate_arg(<<"x">>)),
+ ?assertEqual(<<"x">>, validate_arg("x")),
+ ?assertEqual({<<"x">>, <<"y">>}, validate_arg({"x", "y"})),
+ ?assertEqual({<<"x">>, <<"y">>}, validate_arg({<<"x">>, "y"})),
+ ?assertEqual({<<"x">>, <<"y">>}, validate_arg({"x", <<"y">>})),
+ ?assertEqual({<<"x">>, <<"y">>}, validate_arg({<<"x">>, <<"y">>})),
+ ?assertEqual({?INDEX_CLEANUP, <<"x">>}, validate_arg({?INDEX_CLEANUP, "x"})),
+ ?assertEqual({?INDEX_CLEANUP, <<"x">>}, validate_arg({?INDEX_CLEANUP, <<"x">>})),
+ ?assertError(invalid_smoosh_arg, validate_arg(foo)),
+ ?assertError(invalid_smoosh_arg, validate_arg({foo, bar})),
+ ?assertError(invalid_smoosh_arg, validate_arg({?INDEX_CLEANUP, foo})).
+
+smoosh_utils_test_() ->
+ {
+ foreach,
+ fun() ->
+ meck:new(calendar, [passthrough, unstick]),
+ meck:expect(config, get, fun(_, _, Default) -> Default end)
+ end,
+ fun(_) ->
+ meck:unload()
+ end,
+ [
+ ?TDEF_FE(t_channel_list_default),
+ ?TDEF_FE(t_channel_list_configured),
+ ?TDEF_FE(t_capacity),
+ ?TDEF_FE(t_concurrency),
+ ?TDEF_FE(t_stringify),
+ ?TDEF_FE(t_ignore_db),
+ ?TDEF_FE(t_log_level),
+ ?TDEF_FE(t_allowed_window)
+ ]
+ }.
+
+t_channel_list_default(_) ->
+ ?assertEqual(
+ ["ratio_dbs", "slack_dbs", "upgrade_dbs"],
+ db_channels()
+ ),
+ ?assertEqual(
+ ["ratio_views", "slack_views", "upgrade_views"],
+ view_channels()
+ ),
+ ?assertEqual(
+ ["index_cleanup"],
+ cleanup_channels()
+ ).
+
+t_channel_list_configured(_) ->
+ meck_chans("db_channels", "foo_dbs"),
+ ?assertEqual(
+ ["foo_dbs", "ratio_dbs", "slack_dbs", "upgrade_dbs"],
+ db_channels()
+ ),
+ meck_chans("view_channels", "foo_views"),
+ ?assertEqual(
+ ["foo_views", "ratio_views", "slack_views", "upgrade_views"],
+ view_channels()
+ ),
+ meck_chans("cleanup_channels", "foo_cleanup"),
+ ?assertEqual(
+ ["foo_cleanup", "index_cleanup"],
+ cleanup_channels()
+ ),
+ meck_chans("db_channels", undefined),
+ ?assertEqual(
+ ["ratio_dbs", "slack_dbs", "upgrade_dbs"],
+ db_channels()
+ ),
+ meck_chans("db_channels", ""),
+ ?assertEqual(
+ ["ratio_dbs", "slack_dbs", "upgrade_dbs"],
+ db_channels()
+ ).
+
+t_capacity(_) ->
+ ?assertEqual(9999, capacity("foo")),
+ meck:expect(config, get, fun("smoosh.foo", "capacity", _) -> "2" end),
+ ?assertEqual(2, capacity("foo")).
+
+t_concurrency(_) ->
+ ?assertEqual(1, concurrency("foo")),
+ meck:expect(config, get, fun("smoosh.foo", "concurrency", _) -> "2" end),
+ ?assertEqual(2, concurrency("foo")).
+
+t_stringify(_) ->
+ Flat = fun(L) -> lists:flatten(L) end,
+ ?assertEqual("x index_cleanup", Flat(stringify({?INDEX_CLEANUP, <<"x">>}))),
+ ?assertEqual("x y", Flat(stringify({<<"x">>, <<"y">>}))),
+ ?assertEqual("x", Flat(stringify(<<"x">>))).
+
+t_ignore_db(_) ->
+ ?assertNot(ignore_db(<<"x">>)),
+ ?assertNot(ignore_db({<<"x">>, <<"z">>})),
+ ?assertNot(ignore_db({?INDEX_CLEANUP, <<"x">>})),
+ meck:expect(config, get, fun("smoosh.ignore", "x", _) -> "true" end),
+ ?assert(ignore_db(<<"x">>)),
+ ?assert(ignore_db({<<"x">>, <<"z">>})),
+ ?assert(ignore_db({?INDEX_CLEANUP, <<"x">>})).
+
+t_log_level(_) ->
+ ?assertEqual(notice, log_level("compaction_log", "notice")),
+ meck:expect(config, get, fun("smoosh", "compaction_log", _) -> "error" end),
+ ?assertEqual(error, log_level("compaction_log", "notice")).
+
+t_allowed_window(_) ->
+ ?assert(in_allowed_window("baz")),
+
+ meck_from_to("blah", undefined),
+ ?assert(in_allowed_window("foo")),
+
+ meck_from_to("X:Y", "9.5:Z"),
+ ?assert(in_allowed_window("foo")),
+
+ meck_time(3, 30),
+
+ meck_from_to("02:30", "04:30"),
+ ?assert(in_allowed_window("foo")),
+
+ meck_from_to("3:35", "3:50"),
+ ?assertNot(in_allowed_window("foo")),
+
+ meck_from_to("1:20", "3:29"),
+ ?assertNot(in_allowed_window("foo")),
+
+ meck_from_to("23:30", "3:52"),
+ ?assert(in_allowed_window("foo")),
+
+ meck_from_to("23:30", "3:29"),
+ ?assertNot(in_allowed_window("foo")).
+
+meck_time(H, M) ->
+ meck:expect(calendar, universal_time, 0, {{2000, 10, 10}, {H, M, 42}}).
+
+meck_from_to(From, To) ->
+ meck:expect(config, get, fun
+ ("smoosh.foo", "from", _) -> From;
+ ("smoosh.foo", "to", _) -> To
+ end).
+
+meck_chans(Type, Channels) ->
+ meck:expect(config, get, fun("smoosh", T, _) when T =:= Type -> Channels end).
+
+-endif.
diff --git a/src/smoosh/test/exunit/scheduling_window_test.exs b/src/smoosh/test/exunit/scheduling_window_test.exs
deleted file mode 100644
index 9da4a3150..000000000
--- a/src/smoosh/test/exunit/scheduling_window_test.exs
+++ /dev/null
@@ -1,79 +0,0 @@
-defmodule SmooshSchedulingWindowTest do
- use Couch.Test.ExUnit.Case
-
- setup_all(context) do
- test_ctx = :test_util.start_couch([])
-
- on_exit(fn ->
- :config.delete('smoosh.test_channel', 'from')
- :config.delete('smoosh.test_channel', 'to')
- :test_util.stop_couch(test_ctx)
- end)
-
- context
- end
-
- test "in_allowed_window returns true by default", _context do
- assert :smoosh_utils.in_allowed_window('nonexistent_channel') == true
- end
-
- test "in_allowed_window ignores bad input", _context do
- :config.set('smoosh.test_channel', 'from', 'midnight', false)
- :config.set('smoosh.test_channel', 'to', 'infinity', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == true
- end
-
- test "in_allowed_window returns false when now < from < to", _context do
- now = DateTime.utc_now()
- from = DateTime.add(now, 18_000)
- to = DateTime.add(now, 36_000)
- :config.set('smoosh.test_channel', 'from', '#{from.hour}:#{from.minute}', false)
- :config.set('smoosh.test_channel', 'to', '#{to.hour}:#{to.minute}', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == false
- end
-
- test "in_allowed_window returns true when from < now < to", _context do
- now = DateTime.utc_now()
- from = DateTime.add(now, -18_000)
- to = DateTime.add(now, 18_000)
- :config.set('smoosh.test_channel', 'from', '#{from.hour}:#{from.minute}', false)
- :config.set('smoosh.test_channel', 'to', '#{to.hour}:#{to.minute}', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == true
- end
-
- test "in_allowed_window returns false when from < to < now", _context do
- now = DateTime.utc_now()
- from = DateTime.add(now, -36_000)
- to = DateTime.add(now, -18_000)
- :config.set('smoosh.test_channel', 'from', '#{from.hour}:#{from.minute}', false)
- :config.set('smoosh.test_channel', 'to', '#{to.hour}:#{to.minute}', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == false
- end
-
- test "in_allowed_window returns true when to < from < now", _context do
- now = DateTime.utc_now()
- from = DateTime.add(now, -18_000)
- to = DateTime.add(now, -36_000)
- :config.set('smoosh.test_channel', 'from', '#{from.hour}:#{from.minute}', false)
- :config.set('smoosh.test_channel', 'to', '#{to.hour}:#{to.minute}', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == true
- end
-
- test "in_allowed_window returns false when to < now < from", _context do
- now = DateTime.utc_now()
- from = DateTime.add(now, 18_000)
- to = DateTime.add(now, -18_000)
- :config.set('smoosh.test_channel', 'from', '#{from.hour}:#{from.minute}', false)
- :config.set('smoosh.test_channel', 'to', '#{to.hour}:#{to.minute}', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == false
- end
-
- test "in_allowed_window returns true when now < to < from", _context do
- now = DateTime.utc_now()
- from = DateTime.add(now, 36_000)
- to = DateTime.add(now, 18_000)
- :config.set('smoosh.test_channel', 'from', '#{from.hour}:#{from.minute}', false)
- :config.set('smoosh.test_channel', 'to', '#{to.hour}:#{to.minute}', false)
- assert :smoosh_utils.in_allowed_window('test_channel') == true
- end
-end
diff --git a/src/smoosh/test/exunit/test_helper.exs b/src/smoosh/test/exunit/test_helper.exs
deleted file mode 100644
index 314050085..000000000
--- a/src/smoosh/test/exunit/test_helper.exs
+++ /dev/null
@@ -1,2 +0,0 @@
-ExUnit.configure(formatters: [JUnitFormatter, ExUnit.CLIFormatter])
-ExUnit.start()
[couchdb] 03/03: Improve smoosh_priority_queue
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch optimize-smoosh
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit d7feed4ac9742ce047d92d395f25b099276c6cfb
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Fri Nov 4 14:23:47 2022 -0400
Improve smoosh_priority_queue
* Remove `Value` parameter as it was never used.
* Use `make_ref()` instead of `unique_integer([monotonic])` for performance [1].
* Rewrite tests to get 100% coverage. Previous tests didn't actually run due
to a setup error.
* Switch from `size/1` to `qsize/1` as `size/1` is a built-in and we have to
write `?MODULE:size/1` everywhere.
* Do not remove elements from the gb_tree just for inspecting min and max
elements.
* Remove `last_updated/2` logic, as that will be published in an ETS table.
* Replace low level file serialization operations with `to_map/1`, `from_map/3`.
[1]
```
4> timer:tc(fun() -> [make_ref() || _ <- lists:seq(1, 1000000)], ok end ).
{488923,ok}
6> timer:tc(fun() -> [{erlang:monotonic_time(), erlang:unique_integer([monotonic])} || _ <- lists:seq(1, 1000000)], ok end ).
{1178409,ok}
```
---
src/smoosh/src/smoosh_priority_queue.erl | 281 +++++++++++++-----------
src/smoosh/test/smoosh_priority_queue_tests.erl | 167 --------------
2 files changed, 154 insertions(+), 294 deletions(-)
diff --git a/src/smoosh/src/smoosh_priority_queue.erl b/src/smoosh/src/smoosh_priority_queue.erl
index 16deb0f76..b2ef4393d 100644
--- a/src/smoosh/src/smoosh_priority_queue.erl
+++ b/src/smoosh/src/smoosh_priority_queue.erl
@@ -12,19 +12,16 @@
-module(smoosh_priority_queue).
--export([new/1, recover/1]).
-
--export([last_updated/2, is_key/2, in/4, in/5, out/1, size/1, info/1]).
-
--export([flush/1]).
-
--export([from_list/2, to_list/1]).
-
--export([is_empty/1]).
-
--export([write_to_file/1]).
-
--define(VSN, 1).
+-export([
+ new/1,
+ name/1,
+ in/4,
+ out/1,
+ info/1,
+ flush/1,
+ to_map/1,
+ from_map/3
+]).
-record(priority_queue, {
name,
@@ -33,145 +30,175 @@
}).
new(Name) ->
- #priority_queue{name = Name, map = maps:new(), tree = gb_trees:empty()}.
-
-recover(#priority_queue{name = Name, map = Map0} = Q) ->
- case do_recover(file_name(Q)) of
- {ok, Terms} ->
- Map = maps:merge(Map0, Terms),
- Tree = maps:fold(
- fun(Key, {TreeKey, Value}, TreeAcc) ->
- gb_trees:enter(TreeKey, {Key, Value}, TreeAcc)
- end,
- gb_trees:empty(),
- Map
- ),
- #priority_queue{name = Name, map = Map, tree = Tree};
- error ->
- Q
- end.
+ #priority_queue{name = Name, map = #{}, tree = gb_trees:empty()}.
-write_to_file(#priority_queue{map = Map} = Q) ->
- smoosh_utils:write_to_file(Map, file_name(Q), ?VSN).
+name(#priority_queue{name = Name}) ->
+ Name.
-flush(#priority_queue{name = Name} = Q) ->
- Q#priority_queue{name = Name, map = maps:new(), tree = gb_trees:empty()}.
+flush(#priority_queue{} = Q) ->
+ Q#priority_queue{map = #{}, tree = gb_trees:empty()}.
-last_updated(Key, #priority_queue{map = Map}) ->
+in(Key, Priority, Capacity, #priority_queue{map = Map, tree = Tree} = Q) ->
case maps:find(Key, Map) of
- {ok, {_Priority, {LastUpdatedMTime, _MInt}}} ->
- LastUpdatedMTime;
+ {ok, {Priority, _}} ->
+ % Priority matches, keep everything as is. This might be the case
+ % for upgrade channels, for instance, where priority is 1.
+ Q;
+ {ok, TreeKey} ->
+ Tree1 = gb_trees:delete(TreeKey, Tree),
+ insert(Key, Priority, Capacity, Q#priority_queue{tree = Tree1});
error ->
- false
+ insert(Key, Priority, Capacity, Q)
end.
-is_key(Key, #priority_queue{map = Map}) ->
- maps:is_key(Key, Map).
-
-in(Key, Value, Priority, Q) ->
- in(Key, Value, Priority, infinity, Q).
-
-in(Key, Value, Priority, Capacity, #priority_queue{name = Name, map = Map, tree = Tree}) ->
- Tree1 =
- case maps:find(Key, Map) of
- {ok, TreeKey} ->
- gb_trees:delete_any(TreeKey, Tree);
- error ->
- Tree
- end,
- Now = {erlang:monotonic_time(), erlang:unique_integer([monotonic])},
- TreeKey1 = {Priority, Now},
- Tree2 = gb_trees:enter(TreeKey1, {Key, Value}, Tree1),
- Map1 = maps:put(Key, TreeKey1, Map),
- truncate(Capacity, #priority_queue{name = Name, map = Map1, tree = Tree2}).
-
-out(#priority_queue{name = Name, map = Map, tree = Tree}) ->
+out(#priority_queue{map = Map, tree = Tree} = Q) ->
case gb_trees:is_empty(Tree) of
true ->
false;
false ->
- {_, {Key, Value}, Tree1} = gb_trees:take_largest(Tree),
- Map1 = maps:remove(Key, Map),
- Q = #priority_queue{name = Name, map = Map1, tree = Tree1},
- {Key, Value, Q}
+ {_, Key, Tree1} = gb_trees:take_largest(Tree),
+ {Key, Q#priority_queue{map = maps:remove(Key, Map), tree = Tree1}}
end.
-size(#priority_queue{tree = Tree}) ->
+qsize(#priority_queue{tree = Tree}) ->
gb_trees:size(Tree).
info(#priority_queue{tree = Tree} = Q) ->
[
- {size, ?MODULE:size(Q)}
+ {size, qsize(Q)}
| case gb_trees:is_empty(Tree) of
true ->
[];
false ->
- {Min, _, _} = gb_trees:take_smallest(Tree),
- {Max, _, _} = gb_trees:take_largest(Tree),
+ {{Min, _}, _} = gb_trees:smallest(Tree),
+ {{Max, _}, _} = gb_trees:largest(Tree),
[{min, Min}, {max, Max}]
end
].
-from_list(Orddict, #priority_queue{name = Name}) ->
- Map = maps:from_list(Orddict),
- Tree = gb_trees:from_orddict(Orddict),
- #priority_queue{name = Name, map = Map, tree = Tree}.
-
-to_list(#priority_queue{tree = Tree}) ->
- gb_trees:to_list(Tree).
-
-is_empty(#priority_queue{tree = Tree}) ->
- gb_trees:is_empty(Tree).
-
-file_name(#priority_queue{name = Name}) ->
- filename:join(config:get("smoosh", "state_dir", "."), Name ++ ".waiting").
+insert(Key, Priority, Capacity, #priority_queue{tree = Tree, map = Map} = Q) ->
+ TreeKey = {Priority, make_ref()},
+ Tree1 = gb_trees:insert(TreeKey, Key, Tree),
+ truncate(Capacity, Q#priority_queue{map = Map#{Key => TreeKey}, tree = Tree1}).
-truncate(infinity, Q) ->
- Q;
-truncate(Capacity, Q) when Capacity > 0 ->
- truncate(Capacity, ?MODULE:size(Q), Q).
+truncate(Capacity, Q) when is_integer(Capacity), Capacity > 0 ->
+ truncate(Capacity, qsize(Q), Q).
-truncate(Capacity, Size, Q) when Size =< Capacity ->
+truncate(Capacity, Size, Q) when is_integer(Capacity), Size =< Capacity ->
Q;
-truncate(Capacity, Size, #priority_queue{name = Name, map = Map, tree = Tree}) when Size > 0 ->
- {_, {Key, _}, Tree1} = gb_trees:take_smallest(Tree),
- Q1 = #priority_queue{name = Name, map = maps:remove(Key, Map), tree = Tree1},
- truncate(Capacity, ?MODULE:size(Q1), Q1).
-
-do_recover(FilePath) ->
- case file:read_file(FilePath) of
- {ok, Content} ->
- <<Vsn, Binary/binary>> = Content,
- try parse_queue(Vsn, ?VSN, Binary) of
- Bin ->
- Level = smoosh_utils:log_level("compaction_log_level", "debug"),
- couch_log:Level(
- "~p Successfully restored state file ~s", [?MODULE, FilePath]
- ),
- {ok, Bin}
- catch
- error:Reason ->
- couch_log:error(
- "~p Invalid queue file (~p). Deleting ~s", [?MODULE, Reason, FilePath]
- ),
- file:delete(FilePath),
- error
- end;
- {error, enoent} ->
- Level = smoosh_utils:log_level("compaction_log_level", "debug"),
- couch_log:Level(
- "~p (~p) Queue file ~s does not exist. Not restoring.", [?MODULE, enoent, FilePath]
- ),
- error;
- {error, Reason} ->
- couch_log:error(
- "~p Cannot read the queue file (~p). Deleting ~s", [?MODULE, Reason, FilePath]
- ),
- file:delete(FilePath),
- error
- end.
-
-parse_queue(1, ?VSN, Binary) ->
- erlang:binary_to_term(Binary, [safe]);
-parse_queue(Vsn, ?VSN, _) ->
- error({unsupported_version, Vsn}).
+truncate(Capacity, Size, Q) when is_integer(Capacity), Size > 0 ->
+ #priority_queue{map = Map, tree = Tree} = Q,
+ {_, Key, Tree1} = gb_trees:take_smallest(Tree),
+ Q1 = Q#priority_queue{map = maps:remove(Key, Map), tree = Tree1},
+ truncate(Capacity, qsize(Q1), Q1).
+
+% Serialize the queue to/from simple maps which look like #{Key => Priority}.
+% The intent is for these to be used by the smoosh persistence facility.
+%
+to_map(#priority_queue{map = Map}) ->
+ Fun = fun(_Key, {Priority, _Ref}) ->
+ Priority
+ end,
+ maps:map(Fun, Map).
+
+from_map(Name, Capacity, #{} = SerializedMap) ->
+ Fun = fun(Key, Priority, Acc) ->
+ insert(Key, Priority, Capacity, Acc)
+ end,
+ maps:fold(Fun, new(Name), SerializedMap).
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+-define(K1, <<"db1">>).
+-define(K2, {<<"db1">>, <<"design/_doc1">>}).
+-define(K3, {index_cleanup, <<"db1">>}).
+-define(P1, 1).
+-define(P2, 2.4).
+-define(P3, infinity).
+
+basics_test() ->
+ Q = new("foo"),
+ ?assertMatch(#priority_queue{}, Q),
+ ?assertEqual("foo", name(Q)),
+ ?assertEqual([{size, 0}], info(Q)).
+
+empty_test() ->
+ Q = new("foo"),
+ ?assertEqual(false, out(Q)),
+ ?assertEqual(Q, truncate(1, Q)),
+ ?assertEqual(Q, flush(Q)),
+ ?assertEqual(#{}, to_map(Q)),
+ ?assertEqual(Q, from_map("foo", 1, #{})).
+
+one_element_test() ->
+ Q0 = new("foo"),
+ Q = in(?K1, ?P1, 1, Q0),
+ ?assertMatch(#priority_queue{}, Q),
+ ?assertEqual([{size, 1}, {min, 1}, {max, 1}], info(Q)),
+ ?assertEqual(Q, truncate(1, Q)),
+ ?assertMatch({?K1, #priority_queue{}}, out(Q)),
+ {?K1, Q2} = out(Q),
+ ?assertEqual(Q2, Q0),
+ ?assertEqual(#{?K1 => ?P1}, to_map(Q)),
+ Q3 = from_map("foo", 1, to_map(Q)),
+ ?assertEqual("foo", name(Q3)),
+ ?assertEqual([{size, 1}, {min, ?P1}, {max, ?P1}], info(Q3)),
+ ?assertEqual(to_map(Q), to_map(Q3)),
+ ?assertEqual(Q0, flush(Q)).
+
+multiple_elements_basics_test() ->
+ Q0 = new("foo"),
+ Q1 = in(?K1, ?P1, 10, Q0),
+ Q2 = in(?K2, ?P2, 10, Q1),
+ Q3 = in(?K3, ?P3, 10, Q2),
+ ?assertEqual([{size, 3}, {min, ?P1}, {max, ?P3}], info(Q3)),
+ ?assertEqual([?K3, ?K2, ?K1], drain(Q3)).
+
+update_element_same_priority_test() ->
+ Q0 = new("foo"),
+ Q1 = in(?K1, ?P1, 10, Q0),
+ ?assertEqual(Q1, in(?K1, ?P1, 10, Q1)).
+
+update_element_new_priority_test() ->
+ Q0 = new("foo"),
+ Q1 = in(?K1, ?P1, 10, Q0),
+ Q2 = in(?K2, ?P2, 10, Q1),
+ Q3 = in(?K1, ?P3, 10, Q2),
+ ?assertEqual([{size, 2}, {min, ?P2}, {max, ?P3}], info(Q3)),
+ ?assertEqual([?K1, ?K2], drain(Q3)).
+
+capacity_test() ->
+ Q0 = new("foo"),
+ Q1 = in(?K1, ?P1, 2, Q0),
+ % Capacity = 1, one element only remains
+ ?assertEqual([?K2], drain(in(?K2, ?P2, 1, Q1))),
+ % Capacity = 2, only top two elements remain
+ Q2 = in(?K2, ?P2, 2, Q1),
+ Q3 = in(?K3, ?P3, 2, Q2),
+ ?assertEqual([?K3, ?K2], drain(Q3)).
+
+a_lot_of_elements_test() ->
+ N = 100000,
+ KVs = lists:map(
+ fun(I) ->
+ P = rand:uniform(100),
+ {{I, P}, P}
+ end,
+ lists:seq(1, N)
+ ),
+ Q = from_map("foo", N, maps:from_list(KVs)),
+ ?assertMatch([{size, N} | _], info(Q)),
+ {_, Priorities} = lists:unzip(drain(Q)),
+ ?assertEqual(lists:reverse(lists:sort(Priorities)), Priorities).
+
+drain(Q) ->
+ lists:reverse(drain(out(Q), [])).
+
+drain(false, Acc) ->
+ Acc;
+drain({Key, Q}, Acc) ->
+ drain(out(Q), [Key | Acc]).
+
+-endif.
diff --git a/src/smoosh/test/smoosh_priority_queue_tests.erl b/src/smoosh/test/smoosh_priority_queue_tests.erl
deleted file mode 100644
index 289804ca5..000000000
--- a/src/smoosh/test/smoosh_priority_queue_tests.erl
+++ /dev/null
@@ -1,167 +0,0 @@
--module(smoosh_priority_queue_tests).
-
--include_lib("proper/include/proper.hrl").
--include_lib("couch/include/couch_eunit.hrl").
-
--define(PROP_PREFIX, "prop_").
-
--define(CAPACITY, 3).
-
--define(RANDOM_CHANNEL, lists:flatten(io_lib:format("~p", [erlang:timestamp()]))).
-
-setup() ->
- Ctx = test_util:start_couch(),
- Ctx.
-
-teardown(Ctx) ->
- test_util:stop_couch(Ctx).
-
-smoosh_priority_queue_test_() ->
- {
- "smoosh priority queue test",
- {
- setup,
- fun setup/0,
- fun teardown/1,
- [
- fun prop_inverse_test_/0,
- fun no_halt_on_corrupted_file_test/0,
- fun no_halt_on_missing_file_test/0
- ]
- }
- }.
-
-%% ==========
-%% Tests
-%% ----------
-
-%% define all tests to be able to run them individually
-prop_inverse_test_() ->
- ?_test(begin
- test_property(prop_inverse)
- end).
-
-no_halt_on_corrupted_file_test() ->
- ?_test(begin
- Name = ?RANDOM_CHANNEL,
- Q = smoosh_priority_queue:new(Name),
- FilePath = smoosh_priority_queue:file_name(Q),
- ok = file:write_file(FilePath, <<"garbage">>),
- ?assertEqual(Q, smoosh_priority_queue:recover(Q)),
- ok
- end).
-
-no_halt_on_missing_file_test() ->
- ?_test(begin
- Name = ?RANDOM_CHANNEL,
- Q = smoosh_priority_queue:new(Name),
- FilePath = smoosh_priority_queue:file_name(Q),
- ok = file:delete(FilePath),
- ?assertEqual(Q, smoosh_priority_queue:recover(Q)),
- ok
- end).
-
-%% ==========
-%% Properties
-%% ----------
-
-prop_inverse() ->
- ?FORALL(
- Q,
- queue(),
- begin
- List = smoosh_priority_queue:to_list(Q),
- equal(Q, smoosh_priority_queue:from_list(List, Q))
- end
- ).
-
-%% ==========
-%% Generators
-%% ----------
-
-key() ->
- proper_types:oneof([proper_types:binary(), {proper_types:binary(), proper_types:binary()}]).
-value() ->
- proper_types:oneof([proper_types:binary(), {proper_types:binary(), proper_types:binary()}]).
-priority() -> integer().
-item() -> {key(), value(), priority()}.
-
-items_list() ->
- ?LET(L, list(item()), L).
-
-simple_queue() ->
- ?LET(
- L,
- items_list(),
- from_list(L)
- ).
-
-with_deleted() ->
- ?LET(
- Q,
- ?LET(
- {{K0, V0, P0}, Q0},
- {item(), simple_queue()},
- smoosh_priority_queue:in(K0, V0, P0, ?CAPACITY, Q0)
- ),
- frequency([
- {1, Q},
- {2, element(3, smoosh_priority_queue:out(Q))}
- ])
- ).
-
-queue() ->
- with_deleted().
-
-%% ==========================
-%% Proper related boilerplate
-%% --------------------------
-
-test_property(Property) when is_atom(Property) ->
- test_property({atom_to_list(Property), Property});
-test_property({Id, Property}) ->
- Name = string:sub_string(Id, length(?PROP_PREFIX) + 1),
- Opts = [long_result, {numtests, 1000}, {to_file, user}],
- {Name, {timeout, 60, fun() -> test_it(Property, Opts) end}}.
-
-test_it(Property, Opts) ->
- case proper:quickcheck(?MODULE:Property(), Opts) of
- true ->
- true;
- Else ->
- erlang:error(
- {propertyFailed, [
- {module, ?MODULE},
- {property, Property},
- {result, Else}
- ]}
- )
- end.
-
-%% ================
-%% Helper functions
-%% ----------------
-
-new() ->
- Q = smoosh_priority_queue:new("foo"),
- smoosh_priority_queue:recover(Q).
-
-from_list(List) ->
- lists:foldl(
- fun({Key, Value, Priority}, Queue) ->
- smoosh_priority_queue:in(Key, Value, Priority, ?CAPACITY, Queue)
- end,
- new(),
- List
- ).
-
-equal(Q1, Q2) ->
- out_all(Q1) =:= out_all(Q2).
-
-out_all(Q) ->
- out_all(Q, []).
-out_all(Q0, Acc) ->
- case smoosh_priority_queue:out(Q0) of
- {K, V, Q1} -> out_all(Q1, [{K, V} | Acc]);
- false -> lists:reverse(Acc)
- end.
[couchdb] 01/03: Add a new smoosh persistence module
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch optimize-smoosh
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit b19794c5a8898e35cacaa2b3199f2b3234ec2486
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Thu Nov 3 23:01:59 2022 -0400
Add a new smoosh persistence module
* persist/3 : persists the queue, active and starting jobs
* unpersist/1 : unpersist the queue
* check_setup/0 : quick configuration smoke test
The design is a bit simplified from the previous implementation:
* There is a single waiting queue persisted as simple map. Both starting and
active jobs are added to the same map structure. To maintain their position
in the queue use `infinity` as their priority value. This is an established
Erlang pattern and they should sort higher than both float and numeric
values.
* Detailed, unique gb_tree key parts are not persisted. Only the priority
values and the queue objects. This should make structure smaller and faster
a bit.
* Remove versioning. Smoosh queue data is not critical and if need be, we can
just use a new suffix.
* Anticipating setup errors with the persistence state directory, add a
smoke test in check_setup/0 to check writes and read to/from the state
directory and log warning on failure. The idea is to do call it once
during startup in the smoosh application to inform users their setup is
problematic. After that, treat smoosh persistence opportunistically and if
state directory is not setup, opt to keep the rest of CouchDB node up.
There should be 100% test coverage.
---
src/smoosh/src/smoosh_persist.erl | 293 ++++++++++++++++++++++++++++++++++++++
1 file changed, 293 insertions(+)
diff --git a/src/smoosh/src/smoosh_persist.erl b/src/smoosh/src/smoosh_persist.erl
new file mode 100644
index 000000000..3b5418b04
--- /dev/null
+++ b/src/smoosh/src/smoosh_persist.erl
@@ -0,0 +1,293 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(smoosh_persist).
+
+-export([
+ unpersist/1,
+ persist/3,
+ check_setup/0
+]).
+
+-include_lib("kernel/include/file.hrl").
+
+-define(SUFFIX, ".smooshq").
+
+% Public API
+
+unpersist(Name) ->
+ Enabled = config:get_boolean("smoosh", "persist", false),
+ Capacity = smoosh_utils:capacity(Name),
+ unpersist(Enabled, Name, Capacity).
+
+persist(Waiting, #{} = Active, #{} = Starting) ->
+ Enabled = config:get_boolean("smoosh", "persist", false),
+ persist(Enabled, Waiting, Active, Starting).
+
+% Validate peristence setup for read/write/delete access. Emit warnings if
+% there are any failures. Call this function once during startup. During
+% runtime errors are ignored. Smoosh persistence is opportunistic, so if we
+% cannot read or write we just move on.
+%
+check_setup() ->
+ Enabled = config:get_boolean("smoosh", "persist", false),
+ try
+ check_setup(Enabled)
+ catch
+ throw:{fail, Msg, Error} ->
+ LogMsg = "~s : " ++ Msg ++ " failed in directory ~p : ~p",
+ Args = [?MODULE, state_dir(), Error],
+ couch_log:warning(LogMsg, Args),
+ {error, {Msg, Error}}
+ end.
+
+% Private functions
+
+unpersist(false, Name, _Capacity) ->
+ smoosh_priority_queue:new(Name);
+unpersist(true, Name, Capacity) ->
+ Path = file_path(Name),
+ case read(Path) of
+ {ok, Map} -> smoosh_priority_queue:from_map(Name, Capacity, Map);
+ {error, _} -> smoosh_priority_queue:new(Name)
+ end.
+
+persist(false, _Waiting, _Active, _Starting) ->
+ ok;
+persist(true, Waiting, Active, Starting) ->
+ Name = smoosh_priority_queue:name(Waiting),
+ WMap = smoosh_priority_queue:to_map(Waiting),
+ % Starting and active jobs are at priority level `infinity` as they are
+ % already running. We want them to be the first one to continue after
+ % restart. We're relying on infinity sorting higher than float and integer
+ % numeric values here.
+ AMap = maps:map(fun(_, _) -> infinity end, Active),
+ SMap = maps:from_list([{K, infinity} || K <- maps:values(Starting)]),
+ Path = file_path(Name),
+ write(maps:merge(WMap, maps:merge(AMap, SMap)), Path).
+
+check_setup(false) ->
+ disabled;
+check_setup(true) ->
+ StateDir = state_dir(),
+ Path = filename:join(StateDir, "smooshq.test"),
+ Data = #{<<"test">> => 1},
+ case file:read_file_info(StateDir) of
+ {ok, #file_info{access = A}} when A == read; A == read_write ->
+ ok;
+ {ok, #file_info{access = Invalid}} ->
+ throw({fail, "read access", Invalid});
+ {error, Error1} ->
+ throw({fail, "read", Error1})
+ end,
+ case write(Data, Path) of
+ ok -> ok;
+ {error, Error2} -> throw({fail, "write", Error2})
+ end,
+ couch_util:file_delete(Path).
+
+write(#{} = QData, Path) when is_list(Path), map_size(QData) == 0 ->
+ % Save a few bytes by deleting the persisted queue data if
+ % there are no waiting/starting or active jobs
+ couch_util:file_delete(Path);
+write(#{} = QData, Path) when is_list(Path) ->
+ Bin = term_to_binary(QData, [compressed, {minor_version, 2}]),
+ TmpPath = tmp_path(Path),
+ case file:write_file(TmpPath, Bin, [raw]) of
+ ok -> file:rename(TmpPath, Path);
+ {error, _} = Error -> Error
+ end.
+
+read(Path) ->
+ case file:read_file(Path) of
+ {ok, Bin} ->
+ try binary_to_term(Bin, [safe]) of
+ #{} = QData -> {ok, QData};
+ _ -> {error, term_not_a_map}
+ catch
+ _:_ ->
+ {error, invalid_term}
+ end;
+ {error, _} = Error ->
+ Error
+ end.
+
+tmp_path(Path) ->
+ Time = abs(erlang:system_time()),
+ Path ++ "." ++ integer_to_list(Time) ++ ".tmp".
+
+file_path(Name) ->
+ StateDir = filename:absname(state_dir()),
+ filename:join(StateDir, Name ++ ?SUFFIX).
+
+state_dir() ->
+ Dir = config:get("smoosh", "state_dir", "."),
+ filename:absname(Dir).
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+persist_unpersist_test_() ->
+ {
+ foreach,
+ fun() ->
+ meck:expect(config, get, fun(_, _, Default) -> Default end)
+ end,
+ fun(_) ->
+ meck:unload()
+ end,
+ [
+ ?TDEF_FE(t_write_read_delete),
+ ?TDEF_FE(t_fail_write_read_delete),
+ ?TDEF_FE(t_corrupted_read),
+ ?TDEF_FE(t_check_setup),
+ ?TDEF_FE(t_persist_unpersist_disabled),
+ ?TDEF_FE(t_persist_unpersist_enabled),
+ ?TDEF_FE(t_persist_unpersist_errors)
+ ]
+ }.
+
+t_write_read_delete(_) ->
+ Path = file_path("foochan"),
+ Data = #{<<"a">> => 1},
+
+ ?assertEqual(ok, write(Data, Path)),
+ ?assertMatch({ok, _}, file:read_file_info(Path)),
+
+ ReadRes = read(Path),
+ ?assertMatch({ok, #{}}, ReadRes),
+ {ok, ReadData} = ReadRes,
+ ?assertEqual(Data, ReadData),
+
+ ?assertEqual(ok, write(#{}, Path)),
+ ?assertEqual({error, enoent}, file:read_file_info(Path)).
+
+t_fail_write_read_delete(_) ->
+ meck:expect(config, get, fun("smoosh", "state_dir", _) -> "./x" end),
+ Path = file_path("foochan"),
+ ?assertEqual({error, enoent}, write(#{<<"a">> => 1}, Path)),
+ ?assertEqual({error, enoent}, read(Path)),
+ ?assertEqual({error, enoent}, write(#{}, Path)).
+
+t_corrupted_read(_) ->
+ Path = file_path("foochan"),
+ ?assertEqual(ok, write(#{<<"a">> => 1}, Path)),
+
+ ok = file:write_file(Path, term_to_binary(foo), [raw]),
+ ?assertEqual({error, term_not_a_map}, read(Path)),
+
+ ok = file:write_file(Path, <<"42">>, [raw]),
+ ?assertEqual({error, invalid_term}, read(Path)),
+
+ ?assertEqual(ok, write(#{}, Path)),
+ ?assertEqual({error, enoent}, file:read_file_info(Path)).
+
+t_check_setup(_) ->
+ ?assertEqual(disabled, check_setup()),
+
+ meck:expect(config, get_boolean, fun("smoosh", "persist", _) -> true end),
+ ?assertEqual(ok, check_setup()),
+
+ TDir = ?tempfile(),
+ meck:expect(config, get, fun("smoosh", "state_dir", _) -> TDir end),
+ ?assertEqual({error, {"read", enoent}}, check_setup()),
+
+ Dir = state_dir(),
+ ok = file:make_dir(Dir),
+ % Can't write, only read
+ ok = file:change_mode(Dir, 8#500),
+ ?assertEqual({error, {"write", eacces}}, check_setup()),
+ % Can't read, only write
+ ok = file:change_mode(Dir, 8#300),
+ ?assertEqual({error, {"read access", write}}, check_setup()),
+ ok = file:del_dir_r(Dir).
+
+t_persist_unpersist_disabled(_) ->
+ Name = "chan1",
+ Q = smoosh_priority_queue:new(Name),
+ [K1, K2, K3] = [<<"x">>, {<<"x">>, <<"_design/y">>}, {index_cleanup, <<"z">>}],
+ Active = #{K1 => self()},
+ Starting = #{make_ref() => K2},
+ Q1 = smoosh_priority_queue:in(K3, 1.0, 9999, Q),
+
+ ?assertEqual(ok, persist(Q1, Active, Starting)),
+
+ Q2 = unpersist(Name),
+ ?assertEqual(Name, smoosh_priority_queue:name(Q2)),
+ ?assertEqual([{size, 0}], smoosh_priority_queue:info(Q2)).
+
+t_persist_unpersist_enabled(_) ->
+ Name = "chan2",
+ Q = smoosh_priority_queue:new(Name),
+ Keys = [K1, K2, K3] = [<<"x">>, {<<"x">>, <<"y">>}, {index_cleanup, <<"z">>}],
+ Active = #{K1 => self()},
+ Starting = #{make_ref() => K2},
+ Q1 = smoosh_priority_queue:in(K3, 1.0, 9999, Q),
+
+ meck:expect(config, get_boolean, fun("smoosh", "persist", _) -> true end),
+ ?assertEqual(ok, persist(Q1, Active, Starting)),
+
+ Q2 = unpersist(Name),
+ ?assertEqual(Name, smoosh_priority_queue:name(Q2)),
+ Info2 = smoosh_priority_queue:info(Q2),
+ ?assertEqual([{size, 3}, {min, 1.0}, {max, infinity}], Info2),
+ ?assertEqual(Keys, drain_q(Q2)),
+
+ % Try to persisted the already unpersisted queue
+ ?assertEqual(ok, persist(Q2, #{}, #{})),
+ Q3 = unpersist(Name),
+ ?assertEqual(Name, smoosh_priority_queue:name(Q3)),
+ Info3 = smoosh_priority_queue:info(Q2),
+ ?assertEqual([{size, 3}, {min, 1.0}, {max, infinity}], Info3),
+ ?assertEqual(Keys, drain_q(Q3)).
+
+t_persist_unpersist_errors(_) ->
+ Name = "chan3",
+ Q = smoosh_priority_queue:new(Name),
+ Q1 = smoosh_priority_queue:in(<<"x">>, 1.0, 9999, Q),
+
+ meck:expect(config, get_boolean, fun("smoosh", "persist", _) -> true end),
+
+ TDir = ?tempfile(),
+ meck:expect(config, get, fun
+ ("smoosh", "state_dir", _) -> TDir;
+ (_, _, Default) -> Default
+ end),
+
+ Q2 = unpersist(Name),
+ ?assertEqual(Name, smoosh_priority_queue:name(Q2)),
+ ?assertEqual([{size, 0}], smoosh_priority_queue:info(Q2)),
+
+ Dir = state_dir(),
+ ok = file:make_dir(Dir),
+
+ % Can't write, only read
+ ok = file:change_mode(Dir, 8#500),
+ ?assertEqual({error, eacces}, persist(Q1, #{}, #{})),
+
+ Q3 = unpersist(Name),
+ ?assertEqual(Name, smoosh_priority_queue:name(Q3)),
+ ?assertEqual([{size, 0}], smoosh_priority_queue:info(Q3)),
+
+ ok = file:del_dir_r(Dir).
+
+drain_q(Q) ->
+ lists:reverse(drain_q(Q, [])).
+
+drain_q(Q, Acc) ->
+ case smoosh_priority_queue:out(Q) of
+ false -> Acc;
+ {Key, Q1} -> drain_q(Q1, [Key | Acc])
+ end.
+
+-endif.