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 2020/03/16 18:19:01 UTC
[couchdb] 01/01: Handle transaction cancelled errors in
list_dbs_info/3
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch handle-list-dbs-future-tx-aborts
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 1f0f8a0b80f5c874e2769f9963d94e4ee545b67f
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Mon Mar 16 13:50:34 2020 -0400
Handle transaction cancelled errors in list_dbs_info/3
`list_dbs_info/3` maintains a queue of up to 100 futures which are used to
concurrently fetch data. Previously, if the transaction was reset, and the
accumulator inside the fold may have had futures from a previous transaction,
which have gotten their results yet, they threw a transaction_canceled (1025)
error.
To fix this, if we're in a read-only transaction, we return the tx object in
the opaque db info record. Then, if `erlfdb:wait/1` throws a transaction
canceled error, we re-fetch the future from the now restarted transaction.
Potentially, the transaction may also time-out while the futures queues is
drained after the main range fold has finished already. Handle that case by
reseting the transaction and then re-fetching the futures. To avoid an infinite
loop we allow up to 2 retries only.
This approach is not the most optimal but simpler as it hides the complexity
inside the fabric2_fdb module where we already handle these conditions. It
means that every 5 or so seconds we might have to refetch less than 100 extra
futures from the queue (as some or all may have gotten their results back
already).
---
src/fabric/include/fabric2.hrl | 1 +
src/fabric/src/fabric2_fdb.erl | 96 ++++++++++++++++++++++---------
src/fabric/test/fabric2_db_crud_tests.erl | 84 ++++++++++++++++++++++++++-
3 files changed, 153 insertions(+), 28 deletions(-)
diff --git a/src/fabric/include/fabric2.hrl b/src/fabric/include/fabric2.hrl
index f526d7b..a4f68bd 100644
--- a/src/fabric/include/fabric2.hrl
+++ b/src/fabric/include/fabric2.hrl
@@ -68,6 +68,7 @@
-define(TRANSACTION_TOO_OLD, 1007).
-define(FUTURE_VERSION, 1009).
-define(COMMIT_UNKNOWN_RESULT, 1021).
+-define(TRANSACTION_CANCELLED, 1025).
-define(BINARY_CHUNK_SIZE, 100000).
diff --git a/src/fabric/src/fabric2_fdb.erl b/src/fabric/src/fabric2_fdb.erl
index 8bc8792..403b5bb 100644
--- a/src/fabric/src/fabric2_fdb.erl
+++ b/src/fabric/src/fabric2_fdb.erl
@@ -89,6 +89,14 @@
user_acc
}).
+-record(info_future, {
+ tx,
+ db_prefix,
+ changes_future,
+ meta_future,
+ retries = 0
+}).
+
transactional(Fun) ->
do_transaction(Fun, undefined).
@@ -386,36 +394,37 @@ get_info_future(Tx, DbPrefix) ->
StatsPrefix = erlfdb_tuple:pack({?DB_STATS}, DbPrefix),
MetaFuture = erlfdb:get_range_startswith(Tx, StatsPrefix),
- {DbPrefix, ChangesFuture, MetaFuture}.
+ % Save the tx object only if it's read-only as we might retry to get the
+ % future again after the tx was reset
+ SaveTx = case erlfdb:get_writes_allowed(Tx) of
+ true -> undefined;
+ false -> Tx
+ end,
+ #info_future{
+ tx = SaveTx,
+ db_prefix = DbPrefix,
+ changes_future = ChangesFuture,
+ meta_future = MetaFuture
+ }.
-get_info_wait({DbPrefix, ChangesFuture, MetaFuture}) ->
- RawSeq = case erlfdb:wait(ChangesFuture) of
- [] ->
- vs_to_seq(fabric2_util:seq_zero_vs());
- [{SeqKey, _}] ->
- {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(SeqKey, DbPrefix),
- vs_to_seq(SeqVS)
- end,
- CProp = {update_seq, RawSeq},
- MProps = lists:foldl(fun({K, V}, Acc) ->
- case erlfdb_tuple:unpack(K, DbPrefix) of
- {?DB_STATS, <<"doc_count">>} ->
- [{doc_count, ?bin2uint(V)} | Acc];
- {?DB_STATS, <<"doc_del_count">>} ->
- [{doc_del_count, ?bin2uint(V)} | Acc];
- {?DB_STATS, <<"sizes">>, Name} ->
- Val = ?bin2uint(V),
- {_, {Sizes}} = lists:keyfind(sizes, 1, Acc),
- NewSizes = lists:keystore(Name, 1, Sizes, {Name, Val}),
- lists:keystore(sizes, 1, Acc, {sizes, {NewSizes}});
- {?DB_STATS, _} ->
- Acc
- end
- end, [{sizes, {[]}}], erlfdb:wait(MetaFuture)),
+get_info_wait(#info_future{tx = Tx, retries = Retries} = Future)
+ when Tx =:= undefined orelse Retries >= 2 ->
+ get_info_wait_int(Future);
- [CProp | MProps].
+get_info_wait(#info_future{tx = Tx, retries = Retries} = Future) ->
+ try
+ get_info_wait_int(Future)
+ catch
+ error:{erlfdb_error, ?TRANSACTION_CANCELLED} ->
+ Future1 = get_info_future(Tx, Future#info_future.db_prefix),
+ get_info_wait(Future1#info_future{retries = Retries + 1});
+ error:{erlfdb_error, ?TRANSACTION_TOO_OLD} ->
+ ok = erlfdb:reset(Tx),
+ Future1 = get_info_future(Tx, Future#info_future.db_prefix),
+ get_info_wait(Future1#info_future{retries = Retries + 1})
+ end.
load_config(#{} = Db) ->
@@ -1759,6 +1768,41 @@ with_span(Operation, ExtraTags, Fun) ->
end.
+get_info_wait_int(#info_future{} = InfoFuture) ->
+ #info_future{
+ db_prefix = DbPrefix,
+ changes_future = ChangesFuture,
+ meta_future = MetaFuture
+ } = InfoFuture,
+
+ RawSeq = case erlfdb:wait(ChangesFuture) of
+ [] ->
+ vs_to_seq(fabric2_util:seq_zero_vs());
+ [{SeqKey, _}] ->
+ {?DB_CHANGES, SeqVS} = erlfdb_tuple:unpack(SeqKey, DbPrefix),
+ vs_to_seq(SeqVS)
+ end,
+ CProp = {update_seq, RawSeq},
+
+ MProps = lists:foldl(fun({K, V}, Acc) ->
+ case erlfdb_tuple:unpack(K, DbPrefix) of
+ {?DB_STATS, <<"doc_count">>} ->
+ [{doc_count, ?bin2uint(V)} | Acc];
+ {?DB_STATS, <<"doc_del_count">>} ->
+ [{doc_del_count, ?bin2uint(V)} | Acc];
+ {?DB_STATS, <<"sizes">>, Name} ->
+ Val = ?bin2uint(V),
+ {_, {Sizes}} = lists:keyfind(sizes, 1, Acc),
+ NewSizes = lists:keystore(Name, 1, Sizes, {Name, Val}),
+ lists:keystore(sizes, 1, Acc, {sizes, {NewSizes}});
+ {?DB_STATS, _} ->
+ Acc
+ end
+ end, [{sizes, {[]}}], erlfdb:wait(MetaFuture)),
+
+ [CProp | MProps].
+
+
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
diff --git a/src/fabric/test/fabric2_db_crud_tests.erl b/src/fabric/test/fabric2_db_crud_tests.erl
index c0a65eb..3b53dbd 100644
--- a/src/fabric/test/fabric2_db_crud_tests.erl
+++ b/src/fabric/test/fabric2_db_crud_tests.erl
@@ -18,6 +18,9 @@
-include("fabric2_test.hrl").
+-define(PDICT_RAISE_IN_ERLFDB_WAIT, '$pdict_raise_in_erlfdb_wait').
+
+
crud_test_() ->
{
"Test database CRUD operations",
@@ -39,7 +42,9 @@ crud_test_() ->
?TDEF_FE(list_dbs_info),
?TDEF_FE(list_dbs_info_partial),
?TDEF_FE(list_dbs_tx_too_old),
- ?TDEF_FE(list_dbs_info_tx_too_old)
+ ?TDEF_FE(list_dbs_info_tx_too_old),
+ ?TDEF_FE(get_info_wait_retry_on_tx_too_old),
+ ?TDEF_FE(get_info_wait_retry_on_tx_abort)
]
}
}
@@ -62,7 +67,9 @@ setup() ->
cleanup(_) ->
- fabric2_test_util:tx_too_old_reset_errors().
+ fabric2_test_util:tx_too_old_reset_errors(),
+ reset_fail_erfdb_wait(),
+ meck:reset([erlfdb]).
create_db(_) ->
@@ -256,6 +263,79 @@ list_dbs_info_tx_too_old(_) ->
end, DbNames).
+get_info_wait_retry_on_tx_too_old(_) ->
+ DbName = ?tempdb(),
+ ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+
+ {ok, Db} = fabric2_db:open(DbName, []),
+
+ fabric2_fdb:transactional(Db, fun(TxDb) ->
+ #{
+ tx := Tx,
+ db_prefix := DbPrefix
+ } = TxDb,
+
+ % Simulate being in a list_infos callback
+ ok = erlfdb:set_option(Tx, disallow_writes),
+
+ InfoF = fabric2_fdb:get_info_future(Tx, DbPrefix),
+ {info_future, _, _, ChangesF, _, _} = InfoF,
+
+ raise_in_erlfdb_wait(ChangesF, {erlfdb_error, 1007}, 3),
+ ?assertError({erlfdb_error, 1007}, fabric2_fdb:get_info_wait(InfoF)),
+
+ raise_in_erlfdb_wait(ChangesF, {erlfdb_error, 1007}, 2),
+ ?assertMatch([{_, _} | _], fabric2_fdb:get_info_wait(InfoF))
+ end).
+
+
+get_info_wait_retry_on_tx_abort(_)->
+ DbName = ?tempdb(),
+ ?assertMatch({ok, _}, fabric2_db:create(DbName, [])),
+
+ {ok, Db} = fabric2_db:open(DbName, []),
+
+ fabric2_fdb:transactional(Db, fun(TxDb) ->
+ #{
+ tx := Tx,
+ db_prefix := DbPrefix
+ } = TxDb,
+
+ % Simulate being in a list_infos callback
+ ok = erlfdb:set_option(Tx, disallow_writes),
+
+ InfoF = fabric2_fdb:get_info_future(Tx, DbPrefix),
+ {info_future, _, _, ChangesF, _, _} = InfoF,
+
+ raise_in_erlfdb_wait(ChangesF, {erlfdb_error, 1025}, 3),
+ ?assertError({erlfdb_error, 1025}, fabric2_fdb:get_info_wait(InfoF)),
+
+ raise_in_erlfdb_wait(ChangesF, {erlfdb_error, 1025}, 2),
+ ?assertMatch([{_, _} | _], fabric2_fdb:get_info_wait(InfoF))
+ end).
+
+
+reset_fail_erfdb_wait() ->
+ erase(?PDICT_RAISE_IN_ERLFDB_WAIT),
+ meck:expect(erlfdb, wait, fun(F) -> meck:passthrough([F]) end).
+
+
+raise_in_erlfdb_wait(Future, Error, Count) ->
+ put(?PDICT_RAISE_IN_ERLFDB_WAIT, Count),
+ meck:expect(erlfdb, wait, fun
+ (F) when F =:= Future ->
+ case get(?PDICT_RAISE_IN_ERLFDB_WAIT) of
+ N when is_integer(N), N > 0 ->
+ put(?PDICT_RAISE_IN_ERLFDB_WAIT, N - 1),
+ error(Error);
+ _ ->
+ meck:passthrough([F])
+ end;
+ (F) ->
+ meck:passthrough([F])
+ end).
+
+
is_db_info_member(_, []) ->
false;