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 2017/04/04 13:41:25 UTC
[couchdb] branch 63012-scheduler updated (4fc28f5 -> 53ce0eb)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch 63012-scheduler
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
from 4fc28f5 Cluster ownership module implementation
new 7eeebca Implement multi-db shard change monitoring
new 8fb04ec Share connections between replications
new 53ce0eb AIMD based rate limiter implementation
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "adds" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/couch_replicator/src/couch_multidb_changes.erl | 761 +++++++++++++++++++++
.../src/couch_replicator_connection.erl | 211 ++++++
.../src/couch_replicator_httpc_pool.erl | 78 +--
.../src/couch_replicator_rate_limiter.erl | 274 ++++++++
.../test/couch_replicator_connection_tests.erl | 128 ++++
.../test/couch_replicator_rate_limiter_tests.erl | 89 +++
6 files changed, 1494 insertions(+), 47 deletions(-)
create mode 100644 src/couch_replicator/src/couch_multidb_changes.erl
create mode 100644 src/couch_replicator/src/couch_replicator_connection.erl
create mode 100644 src/couch_replicator/src/couch_replicator_rate_limiter.erl
create mode 100644 src/couch_replicator/test/couch_replicator_connection_tests.erl
create mode 100644 src/couch_replicator/test/couch_replicator_rate_limiter_tests.erl
--
To stop receiving notification emails like this one, please contact
['"commits@couchdb.apache.org" <co...@couchdb.apache.org>'].
[couchdb] 03/03: AIMD based rate limiter implementation
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch 63012-scheduler
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 53ce0ebc8af63dff1e01022810b6bdc6ecf25e5f
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Sun Sep 25 03:49:13 2016 -0400
AIMD based rate limiter implementation
AIMD: additive increase / multiplicative decrease feedback control algorithm.
https://en.wikipedia.org/wiki/Additive_increase/multiplicative_decrease
This is an algorithm which converges on the available channel capacity.
Each participant doesn't a priori know the capacity and participants don't
communicate or know about each other (so they don't coordinate to divide
the capacity among themselves).
A variation of this is used in TCP congestion control algorithm. This is proven
to converge, while for example, additive increase / additive decrease or
multiplicative increase / multiplicative decrease won't.
A few tweaks were applied to the base control logic:
* Estimated value is an interval (period) instead of a rate. This is for
convenience, as users will probably want to know how much to sleep. But,
rate is just 1000 / interval, so it is easy to transform.
* There is a hard max limit for estimated period. Mainly as a practical concern
as connections sleeping too long will timeout and / or jobs will waste time
sleeping and consume scheduler slots, while others could be running.
* There is a time decay component used to handle large pauses between updates.
In case of large update interval, assume (optimistically) some successful
requests have been made. Intuitively, the more time passes, the less accurate
the estimated period probably is.
* The rate of updates applied to the algorithm is limited. This effectively
acts as a low pass filter and make the algorithm handle better spikes and
short bursts of failures. This is not a novel idea, some alternative TCP
control algorithms like Westwood+ do something similar.
* There is a large downward pressure applied to the increasing interval as it
approaches the max limit. This is done by tweaking the additive factor via
a step function. In practice this has effect of trying to make it a bit
harder for jobs to cross the maximum backoff threshold, as they would be
killed and potentially lose intermediate work.
Main API functions are:
success(Key) -> IntervalInMilliseconds
failure(Key) -> IntervalInMilliseconds
interval(Key) -> IntervalInMilliseconds
Key is any (hashable by phash2) term. Typically would be something like
{Method, Url}. The result from the function is the current period value. Caller
would then presumably choose to sleep for that amount of time before or after
making requests. The current interval can be read with interval(Key) function.
Implementation is sharded ETS tables based on the key and there is a periodic
timer which cleans unused items.
Jira: COUCHDB-3324
---
.../src/couch_replicator_rate_limiter.erl | 274 +++++++++++++++++++++
.../test/couch_replicator_rate_limiter_tests.erl | 89 +++++++
2 files changed, 363 insertions(+)
diff --git a/src/couch_replicator/src/couch_replicator_rate_limiter.erl b/src/couch_replicator/src/couch_replicator_rate_limiter.erl
new file mode 100644
index 0000000..f4ae8d2
--- /dev/null
+++ b/src/couch_replicator/src/couch_replicator_rate_limiter.erl
@@ -0,0 +1,274 @@
+% 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.
+
+
+% This module implements rate limiting based on a variation the additive
+% increase / multiplicative decrease feedback control algorithm.
+%
+% https://en.wikipedia.org/wiki/Additive_increase/multiplicative_decrease
+%
+% This is an adaptive algorithm which converges on available channel
+% capacity where each participant (client) doesn't a priori know the
+% capacity, and participants don't communicate or know about each other (so they
+% don't coordinate to divide the capacity among themselves).
+%
+% The algorithm referenced above estimates a rate, whereas the implemented
+% algorithm uses an interval (in milliseconds). It preserves the original
+% semantics, that is the failure part is multplicative and the success part is
+% additive. The relationship between rate and interval is: rate = 1000 / interval.
+%
+% There are two main API functions:
+%
+% success(Key) -> IntervalInMilliseconds
+% failure(Key) -> IntervalInMilliseconds
+%
+% Key is any term, typically something like {Method, Url}. The result from the
+% function is the current period value. Caller then might decide to sleep for
+% that amount of time before or after each request.
+
+
+-module(couch_replicator_rate_limiter).
+-behaviour(gen_server).
+
+
+% public API
+
+-export([start_link/0]).
+-export([interval/1, max_interval/0, failure/1, success/1]).
+
+
+% gen_server callbacks
+
+-export([init/1, handle_call/3, handle_info/2, handle_cast/2,
+ code_change/3, terminate/2]).
+
+
+% Types
+
+-type key() :: any().
+-type interval() :: non_neg_integer().
+-type msec() :: non_neg_integer().
+
+
+% Definitions
+
+-define(SHARDS_N, 16).
+
+% Main parameters of the algorithm. The factor is the multiplicative part and
+% base interval is the additive.
+-define(BASE_INTERVAL, 20).
+-define(BACKOFF_FACTOR, 1.2).
+
+% If estimated period exceeds a limit, it is clipped to this value. This
+% defines a practical limit of this algorithm. This is driven by real world
+% concerns such as having a connection which sleeps for too long and ends up
+% with socket timeout errors, or replication jobs which occupy a scheduler
+% slot without making any progress.
+-define(MAX_INTERVAL, 25000).
+
+% Specify when (threshold) and how much (factor) to decay the estimated period.
+% If there is a long pause between consecutive updates, the estimated period
+% would become less accurate as more time passes. In such case choose to
+% optimistically decay the estimated value. That is assume there a certain
+% rate of successful requests happened. (For reference, TCP congestion algorithm
+% also handles a variation of this in RFC 5681 under "Restarting Idle Connections"
+% section).
+-define(TIME_DECAY_FACTOR, 2).
+-define(TIME_DECAY_THRESHOLD, 1000).
+
+% Limit the rate of updates applied. This controls the rate of change of the
+% estimated value. In colloquial terms it defines how "twitchy" the algorithm
+% is. Or, another way to look at it, this is as a poor version of a low pass
+% filter. (Some alternative TCP congestion control algorithms, like Westwood+
+% use something similar to solve the ACK compression problem).
+-define(SENSITIVITY_TIME_WINDOW, 80).
+
+
+-record(state, {timer}).
+-record(rec, {id, backoff, ts}).
+
+
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+-spec interval(key()) -> interval().
+interval(Key) ->
+ {Interval, _Timestamp} = interval_and_timestamp(Key),
+ Interval.
+
+
+-spec max_interval() -> interval().
+max_interval() ->
+ ?MAX_INTERVAL.
+
+
+-spec failure(key()) -> interval().
+failure(Key) ->
+ {Interval, Timestamp} = interval_and_timestamp(Key),
+ update_failure(Key, Interval, Timestamp, now_msec()).
+
+
+-spec success(key()) -> interval().
+success(Key) ->
+ {Interval, Timestamp} = interval_and_timestamp(Key),
+ update_success(Key, Interval, Timestamp, now_msec()).
+
+
+% gen_server callbacks
+
+init([]) ->
+ Opts = [named_table, public, {keypos, #rec.id}, {read_concurrency, true}],
+ [ets:new(list_to_atom(TableName), Opts) || TableName <- table_names()],
+ {ok, #state{timer = new_timer()}}.
+
+
+terminate(_Reason, _State) ->
+ ok.
+
+
+handle_call(_Msg, _From, State) ->
+ {reply, invalid, State}.
+
+
+handle_cast(_, State) ->
+ {noreply, State}.
+
+
+handle_info(cleanup, #state{timer = Timer}) ->
+ timer:cancel(Timer),
+ TIds = [list_to_existing_atom(TableName) || TableName <- table_names()],
+ [cleanup_table(TId, now_msec() - ?MAX_INTERVAL) || TId <- TIds],
+ {noreply, #state{timer = new_timer()}}.
+
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+
+% Private functions
+
+-spec update_success(any(), interval(), msec(), msec()) -> interval().
+update_success(_Key, _Interval, _Timestamp = 0, _Now) ->
+ 0; % No ets entry. Keep it that way and don't insert a new one.
+
+update_success(_Key, Interval, Timestamp, Now)
+ when Now - Timestamp =< ?SENSITIVITY_TIME_WINDOW ->
+ Interval; % Ignore too frequent updates.
+
+update_success(Key, Interval, Timestamp, Now) ->
+ DecayedInterval = time_decay(Now - Timestamp, Interval),
+ AdditiveFactor = additive_factor(DecayedInterval),
+ NewInterval = DecayedInterval - AdditiveFactor,
+ if
+ NewInterval =< 0 ->
+ ets:delete(term_to_table(Key), Key),
+ 0;
+ NewInterval =< ?BASE_INTERVAL ->
+ insert(Key, ?BASE_INTERVAL, Now);
+ NewInterval > ?BASE_INTERVAL ->
+ insert(Key, NewInterval, Now)
+ end.
+
+
+-spec update_failure(any(), interval(), msec(), msec()) -> interval().
+update_failure(_Key, Interval, Timestamp, Now)
+ when Now - Timestamp =< ?SENSITIVITY_TIME_WINDOW ->
+ Interval; % Ignore too frequent updates.
+
+update_failure(Key, Interval, _Timestamp, Now) ->
+ Interval1 = erlang:max(Interval, ?BASE_INTERVAL),
+ Interval2 = round(Interval1 * ?BACKOFF_FACTOR),
+ Interval3 = erlang:min(Interval2, ?MAX_INTERVAL),
+ insert(Key, Interval3, Now).
+
+
+-spec insert(any(), interval(), msec()) -> interval().
+insert(Key, Interval, Timestamp) ->
+ Entry = #rec{id = Key, backoff = Interval, ts = Timestamp},
+ ets:insert(term_to_table(Key), Entry),
+ Interval.
+
+
+-spec interval_and_timestamp(key()) -> {interval(), msec()}.
+interval_and_timestamp(Key) ->
+ case ets:lookup(term_to_table(Key), Key) of
+ [] ->
+ {0, 0};
+ [#rec{backoff = Interval, ts = Timestamp}] ->
+ {Interval, Timestamp}
+ end.
+
+
+-spec time_decay(msec(), interval()) -> interval().
+time_decay(Dt, Interval) when Dt > ?TIME_DECAY_THRESHOLD ->
+ DecayedInterval = Interval - ?TIME_DECAY_FACTOR * Dt,
+ erlang:max(round(DecayedInterval), 0);
+
+time_decay(_Dt, Interval) ->
+ Interval.
+
+
+% Calculate additive factor. Ideally it would be a constant but in this case
+% it is a step function to help handle larger values as they are approaching
+% the backoff limit. Large success values closer to the limit add some
+% pressure against the limit, which is useful, as at the backoff limit the
+% whole replication job is killed which can be costly in time and temporary work
+% lost by those jobs.
+-spec additive_factor(interval()) -> interval().
+additive_factor(Interval) when Interval > 10000 ->
+ ?BASE_INTERVAL * 50;
+
+ additive_factor(Interval) when Interval > 1000 ->
+ ?BASE_INTERVAL * 5;
+
+additive_factor(Interval) when Interval > 100 ->
+ ?BASE_INTERVAL * 2;
+
+additive_factor(_Interval) ->
+ ?BASE_INTERVAL.
+
+
+-spec table_name(non_neg_integer()) -> string().
+table_name(Id) when is_integer(Id), Id >= 0 andalso Id < ?SHARDS_N ->
+ atom_to_list(?MODULE) ++ "_" ++ integer_to_list(Id).
+
+
+-spec table_names() -> [string()].
+table_names() ->
+ [table_name(N) || N <- lists:seq(0, ?SHARDS_N - 1)].
+
+
+-spec term_to_table(any()) -> atom().
+term_to_table(Term) ->
+ PHash = erlang:phash2(Term),
+ list_to_existing_atom(table_name(PHash rem ?SHARDS_N)).
+
+
+-spec new_timer() -> timer:tref().
+new_timer() ->
+ {ok, Timer} = timer:send_after(?MAX_INTERVAL * 2, cleanup),
+ Timer.
+
+
+-spec now_msec() -> msec().
+now_msec() ->
+ {Mega, Sec, Micro} = os:timestamp(),
+ ((Mega * 1000000) + Sec) * 1000 + round(Micro / 1000).
+
+
+-spec cleanup_table(atom(), msec()) -> non_neg_integer().
+cleanup_table(Tid, LimitMSec) ->
+ Head = #rec{ts = '$1', _ = '_'},
+ Guard = {'<', '$1', LimitMSec},
+ ets:select_delete(Tid, [{Head, [Guard], [true]}]).
diff --git a/src/couch_replicator/test/couch_replicator_rate_limiter_tests.erl b/src/couch_replicator/test/couch_replicator_rate_limiter_tests.erl
new file mode 100644
index 0000000..034550a
--- /dev/null
+++ b/src/couch_replicator/test/couch_replicator_rate_limiter_tests.erl
@@ -0,0 +1,89 @@
+-module(couch_replicator_rate_limiter_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+
+rate_limiter_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ t_new_key(),
+ t_1_failure(),
+ t_2_failures_back_to_back(),
+ t_2_failures(),
+ t_success_threshold(),
+ t_1_failure_2_successes()
+ ]
+ }.
+
+
+t_new_key() ->
+ ?_test(begin
+ ?assertEqual(0, couch_replicator_rate_limiter:interval({"foo", get}))
+ end).
+
+
+t_1_failure() ->
+ ?_test(begin
+ ?assertEqual(24, couch_replicator_rate_limiter:failure({"foo", get}))
+ end).
+
+
+t_2_failures() ->
+ ?_test(begin
+ couch_replicator_rate_limiter:failure({"foo", get}),
+ low_pass_filter_delay(),
+ Interval = couch_replicator_rate_limiter:failure({"foo", get}),
+ ?assertEqual(29, Interval)
+ end).
+
+
+t_2_failures_back_to_back() ->
+ ?_test(begin
+ couch_replicator_rate_limiter:failure({"foo", get}),
+ Interval = couch_replicator_rate_limiter:failure({"foo", get}),
+ ?assertEqual(24, Interval)
+ end).
+
+
+t_success_threshold() ->
+ ?_test(begin
+ Interval = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(0, Interval),
+ Interval = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(0, Interval)
+ end).
+
+
+t_1_failure_2_successes() ->
+ ?_test(begin
+ couch_replicator_rate_limiter:failure({"foo", get}),
+ low_pass_filter_delay(),
+ Succ1 = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(20, Succ1),
+ low_pass_filter_delay(),
+ Succ2 = couch_replicator_rate_limiter:success({"foo", get}),
+ ?assertEqual(0, Succ2)
+ end).
+
+
+low_pass_filter_delay() ->
+ timer:sleep(100).
+
+
+setup() ->
+ {ok, Pid} = couch_replicator_rate_limiter:start_link(),
+ Pid.
+
+
+teardown(Pid) ->
+ Ref = erlang:monitor(process, Pid),
+ unlink(Pid),
+ exit(Pid, kill),
+ receive
+ {'DOWN', Ref, process, Pid, _} ->
+ ok
+ end,
+ ok.
--
To stop receiving notification emails like this one, please contact
"commits@couchdb.apache.org" <co...@couchdb.apache.org>.
[couchdb] 02/03: Share connections between replications
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch 63012-scheduler
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 8fb04ecfb1e9089658860a64da0af97d31db824b
Author: Benjamin Bastian <be...@gmail.com>
AuthorDate: Tue Aug 2 15:01:23 2016 -0700
Share connections between replications
This commit adds functionality to share connections between
replications. This is to solve two problems:
- Prior to this commit, each replication would create a pool of
connections and hold onto those connections as long as the replication
existed. This was wasteful and cause CouchDB to use many unnecessary
connections.
- When the pool was being terminated, the pool would block while the
socket was closed. This would cause the entire replication scheduler
to block. By reusing connections, connections are never closed by
clients. They are only ever relinquished. This operation is always
fast.
This commit adds an intermediary process which tracks which connection
processes are being used by which client. It monitors clients and
connections. If a client or connection crashes, the paired
client/connection will be terminated.
A client can gracefully relinquish ownership of a connection. If that
happens, the connection will be shared with another client. If the
connection remains idle for too long, it will be closed.
Jira: COUCHDB-3324
---
.../src/couch_replicator_connection.erl | 211 +++++++++++++++++++++
.../src/couch_replicator_httpc_pool.erl | 78 +++-----
.../test/couch_replicator_connection_tests.erl | 128 +++++++++++++
3 files changed, 370 insertions(+), 47 deletions(-)
diff --git a/src/couch_replicator/src/couch_replicator_connection.erl b/src/couch_replicator/src/couch_replicator_connection.erl
new file mode 100644
index 0000000..e580663
--- /dev/null
+++ b/src/couch_replicator/src/couch_replicator_connection.erl
@@ -0,0 +1,211 @@
+% 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(couch_replicator_connection).
+
+-behavior(gen_server).
+-behavior(config_listener).
+
+-export([start_link/0]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2]).
+-export([code_change/3, terminate/2]).
+
+-export([acquire/1, relinquish/1]).
+
+-export([handle_config_change/5, handle_config_terminate/3]).
+
+-define(DEFAULT_CLOSE_INTERVAL, 90000).
+-define(RELISTEN_DELAY, 5000).
+
+-record(state, {
+ close_interval,
+ timer
+}).
+
+-record(connection, {
+ worker,
+ host,
+ port,
+ mref
+}).
+
+-include_lib("ibrowse/include/ibrowse.hrl").
+
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+init([]) ->
+ process_flag(trap_exit, true),
+ ?MODULE = ets:new(?MODULE, [named_table, public, {keypos, #connection.worker}]),
+ ok = config:listen_for_changes(?MODULE, nil),
+ Interval = config:get_integer("replicator", "connection_close_interval", ?DEFAULT_CLOSE_INTERVAL),
+ {ok, Timer} = timer:send_after(Interval, close_idle_connections),
+ ibrowse:add_config([{inactivity_timeout, Interval}]),
+ {ok, #state{close_interval=Interval, timer=Timer}}.
+
+
+acquire(URL) when is_binary(URL) ->
+ acquire(binary_to_list(URL));
+
+acquire(URL) ->
+ case gen_server:call(?MODULE, {acquire, URL}) of
+ {ok, Worker} ->
+ link(Worker),
+ {ok, Worker};
+ {error, all_allocated} ->
+ {ok, Pid} = ibrowse:spawn_link_worker_process(URL),
+ ok = gen_server:call(?MODULE, {create, URL, Pid}),
+ {ok, Pid};
+ {error, Reason} ->
+ {error, Reason}
+ end.
+
+
+relinquish(Worker) ->
+ unlink(Worker),
+ gen_server:cast(?MODULE, {relinquish, Worker}).
+
+
+handle_call({acquire, URL}, From, State) ->
+ {Pid, _Ref} = From,
+ case ibrowse_lib:parse_url(URL) of
+ #url{host=Host, port=Port} ->
+ case ets:match_object(?MODULE, #connection{host=Host, port=Port, mref=undefined, _='_'}, 1) of
+ '$end_of_table' ->
+ {reply, {error, all_allocated}, State};
+ {[Worker], _Cont} ->
+ couch_stats:increment_counter([couch_replicator, connection, acquires]),
+ ets:insert(?MODULE, Worker#connection{mref=monitor(process, Pid)}),
+ {reply, {ok, Worker#connection.worker}, State}
+ end;
+ {error, invalid_uri} ->
+ {reply, {error, invalid_uri}, State}
+ end;
+
+handle_call({create, URL, Worker}, From, State) ->
+ {Pid, _Ref} = From,
+ case ibrowse_lib:parse_url(URL) of
+ #url{host=Host, port=Port} ->
+ link(Worker),
+ couch_stats:increment_counter([couch_replicator, connection, creates]),
+ true = ets:insert_new(
+ ?MODULE,
+ #connection{host=Host, port=Port, worker=Worker, mref=monitor(process, Pid)}
+ ),
+ {reply, ok, State}
+ end.
+
+
+handle_cast({relinquish, WorkerPid}, State) ->
+ couch_stats:increment_counter([couch_replicator, connection, relinquishes]),
+ case ets:lookup(?MODULE, WorkerPid) of
+ [Worker] ->
+ case Worker#connection.mref of
+ MRef when is_reference(MRef) -> demonitor(MRef, [flush]);
+ undefined -> ok
+ end,
+ ets:insert(?MODULE, Worker#connection{mref=undefined});
+ [] ->
+ ok
+ end,
+ {noreply, State};
+
+handle_cast({connection_close_interval, V}, State) ->
+ {ok, cancel} = timer:cancel(State#state.timer),
+ {ok, NewTimer} = timer:send_after(V, close_idle_connections),
+ ibrowse:add_config([{inactivity_timeout, V}]),
+ {noreply, State#state{close_interval=V, timer=NewTimer}}.
+
+
+% owner crashed
+handle_info({'DOWN', Ref, process, _Pid, _Reason}, State) ->
+ couch_stats:increment_counter([couch_replicator, connection, owner_crashes]),
+ ets:match_delete(?MODULE, #connection{mref=Ref, _='_'}),
+ {noreply, State};
+
+% worker crashed
+handle_info({'EXIT', Pid, Reason}, State) ->
+ couch_stats:increment_counter([couch_replicator, connection, worker_crashes]),
+ case ets:lookup(?MODULE, Pid) of
+ [] ->
+ ok;
+ [Worker] ->
+ #connection{host=Host, port=Port} = Worker,
+ maybe_log_worker_death(Host, Port, Reason),
+ case Worker#connection.mref of
+ MRef when is_reference(MRef) -> demonitor(MRef, [flush]);
+ undefined -> ok
+ end,
+ ets:delete(?MODULE, Pid)
+ end,
+ {noreply, State};
+
+handle_info(close_idle_connections, State) ->
+ #state{
+ close_interval=Interval,
+ timer=Timer
+ } = State,
+ Conns = ets:match_object(?MODULE, #connection{mref=undefined, _='_'}),
+ lists:foreach(fun(Conn) ->
+ couch_stats:increment_counter([couch_replicator, connection, closes]),
+ delete_worker(Conn)
+ end, Conns),
+ {ok, cancel} = timer:cancel(Timer),
+ {ok, NewTimer} = timer:send_after(Interval, close_idle_connections),
+ {noreply, State#state{timer=NewTimer}};
+
+handle_info(restart_config_listener, State) ->
+ ok = config:listen_for_changes(?MODULE, nil),
+ {noreply, State}.
+
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+
+terminate(_Reason, _State) ->
+ ok.
+
+
+maybe_log_worker_death(_Host, _Port, normal) ->
+ ok;
+
+maybe_log_worker_death(Host, Port, Reason) ->
+ ErrMsg = "Replication connection to: ~p:~p died with reason ~p",
+ couch_log:info(ErrMsg, [Host, Port, Reason]).
+
+
+-spec delete_worker(#connection{}) -> ok.
+delete_worker(Worker) ->
+ ets:delete(?MODULE, Worker#connection.worker),
+ unlink(Worker#connection.worker),
+ spawn(fun() -> ibrowse_http_client:stop(Worker#connection.worker) end),
+ ok.
+
+
+handle_config_change("replicator", "connection_close_interval", V, _, S) ->
+ ok = gen_server:cast(?MODULE, {connection_close_interval,
+ list_to_integer(V)}),
+ {ok, S};
+
+handle_config_change(_, _, _, _, S) ->
+ {ok, S}.
+
+
+handle_config_terminate(_, stop, _) ->
+ ok;
+
+handle_config_terminate(_, _, _) ->
+ Pid = whereis(?MODULE),
+ erlang:send_after(?RELISTEN_DELAY, Pid, restart_config_listener).
diff --git a/src/couch_replicator/src/couch_replicator_httpc_pool.erl b/src/couch_replicator/src/couch_replicator_httpc_pool.erl
index 09e3b23..9f1f1a8 100644
--- a/src/couch_replicator/src/couch_replicator_httpc_pool.erl
+++ b/src/couch_replicator/src/couch_replicator_httpc_pool.erl
@@ -31,8 +31,7 @@
-record(state, {
url,
limit, % max # of workers allowed
- free = [], % free workers (connections)
- busy = [], % busy workers (connections)
+ conns = [],
waiting = queue:new(), % blocked clients waiting for a worker
callers = [] % clients who've been given a worker
}).
@@ -70,23 +69,17 @@ handle_call(get_worker, From, State) ->
callers = Callers,
url = Url,
limit = Limit,
- busy = Busy,
- free = Free
+ conns = Conns
} = State,
- case length(Busy) >= Limit of
+ case length(Conns) >= Limit of
true ->
{noreply, State#state{waiting = queue:in(From, Waiting)}};
false ->
- case Free of
- [] ->
- {ok, Worker} = ibrowse:spawn_link_worker_process(Url),
- Free2 = Free;
- [Worker | Free2] ->
- ok
- end,
+ % If the call to acquire fails, the worker pool will crash with a
+ % badmatch.
+ {ok, Worker} = couch_replicator_connection:acquire(Url),
NewState = State#state{
- free = Free2,
- busy = [Worker | Busy],
+ conns = [Worker | Conns],
callers = monitor_client(Callers, Worker, From)
},
{reply, {ok, Worker}, NewState}
@@ -104,35 +97,29 @@ handle_cast({release_worker, Worker}, State) ->
handle_info({'EXIT', Pid, _Reason}, State) ->
#state{
url = Url,
- busy = Busy,
- free = Free,
+ conns = Conns,
waiting = Waiting,
callers = Callers
} = State,
NewCallers0 = demonitor_client(Callers, Pid),
- case Free -- [Pid] of
- Free ->
- case Busy -- [Pid] of
- Busy ->
+ case Conns -- [Pid] of
+ Conns ->
{noreply, State#state{callers = NewCallers0}};
- Busy2 ->
+ Conns2 ->
case queue:out(Waiting) of
- {empty, _} ->
- {noreply, State#state{busy = Busy2, callers = NewCallers0}};
- {{value, From}, Waiting2} ->
- {ok, Worker} = ibrowse:spawn_link_worker_process(Url),
- NewCallers1 = monitor_client(NewCallers0, Worker, From),
- gen_server:reply(From, {ok, Worker}),
- NewState = State#state{
- busy = [Worker | Busy2],
- waiting = Waiting2,
- callers = NewCallers1
- },
- {noreply, NewState}
+ {empty, _} ->
+ {noreply, State#state{conns = Conns2, callers = NewCallers0}};
+ {{value, From}, Waiting2} ->
+ {ok, Worker} = couch_replicator_connection:acquire(Url),
+ NewCallers1 = monitor_client(NewCallers0, Worker, From),
+ gen_server:reply(From, {ok, Worker}),
+ NewState = State#state{
+ conns = [Worker | Conns2],
+ waiting = Waiting2,
+ callers = NewCallers1
+ },
+ {noreply, NewState}
end
- end;
- Free2 ->
- {noreply, State#state{free = Free2, callers = NewCallers0}}
end;
handle_info({'DOWN', Ref, process, _, _}, #state{callers = Callers} = State) ->
@@ -147,9 +134,8 @@ code_change(_OldVsn, #state{}=State, _Extra) ->
{ok, State}.
-terminate(_Reason, State) ->
- lists:foreach(fun ibrowse_http_client:stop/1, State#state.free),
- lists:foreach(fun ibrowse_http_client:stop/1, State#state.busy).
+terminate(_Reason, _State) ->
+ ok.
monitor_client(Callers, Worker, {ClientPid, _}) ->
[{Worker, erlang:monitor(process, ClientPid)} | Callers].
@@ -167,22 +153,20 @@ release_worker_internal(Worker, State) ->
#state{waiting = Waiting, callers = Callers} = State,
NewCallers0 = demonitor_client(Callers, Worker),
case is_process_alive(Worker) andalso
- lists:member(Worker, State#state.busy) of
+ lists:member(Worker, State#state.conns) of
true ->
- case queue:out(Waiting) of
+ Conns = case queue:out(Waiting) of
{empty, Waiting2} ->
NewCallers1 = NewCallers0,
- Busy2 = State#state.busy -- [Worker],
- Free2 = [Worker | State#state.free];
+ couch_replicator_connection:relinquish(Worker),
+ State#state.conns -- [Worker];
{{value, From}, Waiting2} ->
NewCallers1 = monitor_client(NewCallers0, Worker, From),
gen_server:reply(From, {ok, Worker}),
- Busy2 = State#state.busy,
- Free2 = State#state.free
+ State#state.conns
end,
NewState = State#state{
- busy = Busy2,
- free = Free2,
+ conns = Conns,
waiting = Waiting2,
callers = NewCallers1
},
diff --git a/src/couch_replicator/test/couch_replicator_connection_tests.erl b/src/couch_replicator/test/couch_replicator_connection_tests.erl
new file mode 100644
index 0000000..323a7de
--- /dev/null
+++ b/src/couch_replicator/test/couch_replicator_connection_tests.erl
@@ -0,0 +1,128 @@
+% 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(couch_replicator_connection_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+-define(TIMEOUT, 1000).
+
+
+setup() ->
+ Host = config:get("httpd", "bind_address", "127.0.0.1"),
+ Port = config:get("httpd", "port", "5984"),
+ {Host, Port}.
+
+teardown(_) ->
+ ok.
+
+
+httpc_pool_test_() ->
+ {
+ "replicator connection sharing tests",
+ {
+ setup,
+ fun() -> test_util:start_couch([couch_replicator]) end, fun test_util:stop_couch/1,
+ {
+ foreach,
+ fun setup/0, fun teardown/1,
+ [
+ fun connections_shared_after_relinquish/1,
+ fun connections_not_shared_after_owner_death/1,
+ fun idle_connections_closed/1,
+ fun test_owner_monitors/1
+ ]
+ }
+ }
+ }.
+
+
+connections_shared_after_relinquish({Host, Port}) ->
+ ?_test(begin
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ Self = self(),
+ {ok, Pid} = couch_replicator_connection:acquire(URL),
+ couch_replicator_connection:relinquish(Pid),
+ spawn(fun() ->
+ Self ! couch_replicator_connection:acquire(URL)
+ end),
+ receive
+ {ok, Pid2} ->
+ ?assertEqual(Pid, Pid2)
+ end
+ end).
+
+
+connections_not_shared_after_owner_death({Host, Port}) ->
+ ?_test(begin
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ Self = self(),
+ spawn(fun() ->
+ Self ! couch_replicator_connection:acquire(URL),
+ 1/0
+ end),
+ receive
+ {ok, Pid} ->
+ {ok, Pid2} = couch_replicator_connection:acquire(URL),
+ ?assertNotEqual(Pid, Pid2),
+ MRef = monitor(process, Pid),
+ receive {'DOWN', MRef, process, Pid, _Reason} ->
+ ?assert(not is_process_alive(Pid));
+ Other -> throw(Other)
+ end
+ end
+ end).
+
+
+idle_connections_closed({Host, Port}) ->
+ ?_test(begin
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ {ok, Pid} = couch_replicator_connection:acquire(URL),
+ couch_replicator_connection ! close_idle_connections,
+ ?assert(ets:member(couch_replicator_connection, Pid)),
+ % block until idle connections have closed
+ sys:get_status(couch_replicator_connection),
+ couch_replicator_connection:relinquish(Pid),
+ couch_replicator_connection ! close_idle_connections,
+ % block until idle connections have closed
+ sys:get_status(couch_replicator_connection),
+ ?assert(not ets:member(couch_replicator_connection, Pid))
+ end).
+
+
+test_owner_monitors({Host, Port}) ->
+ ?_test(begin
+ URL = "http://" ++ Host ++ ":" ++ Port,
+ {ok, Worker0} = couch_replicator_connection:acquire(URL),
+ assert_monitors_equal([{process, self()}]),
+ couch_replicator_connection:relinquish(Worker0),
+ assert_monitors_equal([]),
+ {Workers, Monitors} = lists:foldl(fun(_, {WAcc, MAcc}) ->
+ {ok, Worker1} = couch_replicator_connection:acquire(URL),
+ MAcc1 = [{process, self()} | MAcc],
+ assert_monitors_equal(MAcc1),
+ {[Worker1 | WAcc], MAcc1}
+ end, {[], []}, lists:seq(1,5)),
+ lists:foldl(fun(Worker2, Acc) ->
+ [_ | NewAcc] = Acc,
+ couch_replicator_connection:relinquish(Worker2),
+ assert_monitors_equal(NewAcc),
+ NewAcc
+ end, Monitors, Workers)
+ end).
+
+
+assert_monitors_equal(ShouldBe) ->
+ sys:get_status(couch_replicator_connection),
+ {monitors, Monitors} = process_info(whereis(couch_replicator_connection), monitors),
+ ?assertEqual(Monitors, ShouldBe).
--
To stop receiving notification emails like this one, please contact
"commits@couchdb.apache.org" <co...@couchdb.apache.org>.
[couchdb] 01/03: Implement multi-db shard change monitoring
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch 63012-scheduler
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 7eeebca138b6b5c20279baa208e984047e406a2f
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Thu Apr 21 02:57:43 2016 -0400
Implement multi-db shard change monitoring
Monitor shards which match a suffix for creation, deletion, and doc updates.
To use implement `couch_multidb_changes` behavior. Call `start_link` with
DbSuffix, with an option to skip design docs (`skip_ddocs`). Behavior
callback functions will be called when shards are created, deleted, found and
updated.
Jira: COUCHDB-3324
---
src/couch_replicator/src/couch_multidb_changes.erl | 761 +++++++++++++++++++++
1 file changed, 761 insertions(+)
diff --git a/src/couch_replicator/src/couch_multidb_changes.erl b/src/couch_replicator/src/couch_multidb_changes.erl
new file mode 100644
index 0000000..b9c9ad5
--- /dev/null
+++ b/src/couch_replicator/src/couch_multidb_changes.erl
@@ -0,0 +1,761 @@
+% 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(couch_multidb_changes).
+-behaviour(gen_server).
+
+-export([start_link/4]).
+
+-export([init/1, handle_call/3, handle_info/2, handle_cast/2]).
+-export([code_change/3, terminate/2]).
+
+-export([changes_reader/3, changes_reader_cb/3]).
+
+-include_lib("couch/include/couch_db.hrl").
+
+-define(CTX, {user_ctx, #user_ctx{roles=[<<"_admin">>, <<"_replicator">>]}}).
+
+-define(AVG_DELAY_MSEC, 100).
+-define(MAX_DELAY_MSEC, 60000).
+
+-record(state, {
+ tid :: ets:tid(),
+ mod :: atom(),
+ ctx :: term(),
+ suffix :: binary(),
+ event_server :: reference(),
+ scanner :: nil | pid(),
+ pids :: [{binary(),pid()}],
+ skip_ddocs :: boolean()
+}).
+
+
+
+% Behavior API
+
+% For each db shard with a matching suffix, report created,
+% deleted, found (discovered) and change events.
+
+-callback db_created(DbName :: binary(), Context :: term()) ->
+ Context :: term().
+
+-callback db_deleted(DbName :: binary(), Context :: term()) ->
+ Context :: term().
+
+-callback db_found(DbName :: binary(), Context :: term()) ->
+ Context :: term().
+
+-callback db_change(DbName :: binary(), Change :: term(), Context :: term()) ->
+ Context :: term().
+
+
+
+% External API
+
+
+% Opts list can contain:
+% - `skip_ddocs` : Skip design docs
+
+-spec start_link(binary(), module(), term(), list()) ->
+ {ok, pid()} | ignore | {error, term()}.
+start_link(DbSuffix, Module, Context, Opts) when
+ is_binary(DbSuffix), is_atom(Module), is_list(Opts) ->
+ gen_server:start_link(?MODULE, [DbSuffix, Module, Context, Opts], []).
+
+
+% gen_server callbacks
+
+init([DbSuffix, Module, Context, Opts]) ->
+ process_flag(trap_exit, true),
+ Server = self(),
+ {ok, #state{
+ tid = ets:new(?MODULE, [set, protected]),
+ mod = Module,
+ ctx = Context,
+ suffix = DbSuffix,
+ event_server = register_with_event_server(Server),
+ scanner = spawn_link(fun() -> scan_all_dbs(Server, DbSuffix) end),
+ pids = [],
+ skip_ddocs = proplists:is_defined(skip_ddocs, Opts)
+ }}.
+
+
+terminate(_Reason, _State) ->
+ ok.
+
+
+handle_call({change, DbName, Change}, _From,
+ #state{skip_ddocs=SkipDDocs, mod=Mod, ctx=Ctx} = State) ->
+ case {SkipDDocs, is_design_doc(Change)} of
+ {true, true} ->
+ {reply, ok, State};
+ {_, _} ->
+ {reply, ok, State#state{ctx=Mod:db_change(DbName, Change, Ctx)}}
+ end;
+
+handle_call({checkpoint, DbName, EndSeq}, _From, #state{tid=Ets} = State) ->
+ case ets:lookup(Ets, DbName) of
+ [] ->
+ true = ets:insert(Ets, {DbName, EndSeq, false});
+ [{DbName, _OldSeq, Rescan}] ->
+ true = ets:insert(Ets, {DbName, EndSeq, Rescan})
+ end,
+ {reply, ok, State}.
+
+
+handle_cast({resume_scan, DbName}, State) ->
+ {noreply, resume_scan(DbName, State)}.
+
+
+handle_info({'$couch_event', DbName, Event}, #state{suffix = Suf} = State) ->
+ case Suf =:= couch_db:dbname_suffix(DbName) of
+ true ->
+ {noreply, db_callback(Event, DbName, State)};
+ _ ->
+ {noreply, State}
+ end;
+
+handle_info({'DOWN', Ref, _, _, Info}, #state{event_server = Ref} = State) ->
+ {stop, {couch_event_server_died, Info}, State};
+
+handle_info({'EXIT', From, normal}, #state{scanner = From} = State) ->
+ {noreply, State#state{scanner=nil}};
+
+handle_info({'EXIT', From, Reason}, #state{scanner = From} = State) ->
+ {stop, {scanner_died, Reason}, State};
+
+handle_info({'EXIT', From, Reason}, #state{pids = Pids} = State) ->
+ couch_log:info("~p change feed exited ~p",[State#state.suffix, From]),
+ case lists:keytake(From, 2, Pids) of
+ {value, {DbName, From}, NewPids} ->
+ if Reason == normal -> ok; true ->
+ Fmt = "~s : Known change feed ~w died :: ~w",
+ couch_log:error(Fmt, [?MODULE, From, Reason])
+ end,
+ NewState = State#state{pids = NewPids},
+ case ets:lookup(State#state.tid, DbName) of
+ [{DbName, _EndSeq, true}] ->
+ {noreply, resume_scan(DbName, NewState)};
+ _ ->
+ {noreply, NewState}
+ end;
+ false when Reason == normal ->
+ {noreply, State};
+ false ->
+ Fmt = "~s(~p) : Unknown pid ~w died :: ~w",
+ couch_log:error(Fmt, [?MODULE, State#state.suffix, From, Reason]),
+ {stop, {unexpected_exit, From, Reason}, State}
+ end;
+
+handle_info(_Msg, State) ->
+ {noreply, State}.
+
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+
+% Private functions
+
+-spec register_with_event_server(pid()) -> reference().
+register_with_event_server(Server) ->
+ Ref = erlang:monitor(process, couch_event_server),
+ couch_event:register_all(Server),
+ Ref.
+
+
+-spec db_callback(created | deleted | updated, binary(), #state{}) -> #state{}.
+db_callback(created, DbName, #state{mod = Mod, ctx = Ctx} = State) ->
+ State#state{ctx = Mod:db_created(DbName, Ctx)};
+
+db_callback(deleted, DbName, #state{mod = Mod, ctx = Ctx} = State) ->
+ State#state{ctx = Mod:db_deleted(DbName, Ctx)};
+
+db_callback(updated, DbName, State) ->
+ resume_scan(DbName, State);
+
+db_callback(_Other, _DbName, State) ->
+ State.
+
+
+-spec resume_scan(binary(), #state{}) -> #state{}.
+resume_scan(DbName, #state{pids=Pids, tid=Ets} = State) ->
+ case {lists:keyfind(DbName, 1, Pids), ets:lookup(Ets, DbName)} of
+ {{DbName, _}, []} ->
+ % Found existing change feed, but not entry in ETS
+ % Flag a need to rescan from begining
+ true = ets:insert(Ets, {DbName, 0, true}),
+ State;
+ {{DbName, _}, [{DbName, EndSeq, _}]} ->
+ % Found existing change feed and entry in ETS
+ % Flag a need to rescan from last ETS checkpoint
+ true = ets:insert(Ets, {DbName, EndSeq, true}),
+ State;
+ {false, []} ->
+ % No existing change feed running. No entry in ETS.
+ % This is first time seeing this db shard.
+ % Notify user with a found callback. Insert checkpoint
+ % entry in ETS to start from 0. And start a change feed.
+ true = ets:insert(Ets, {DbName, 0, false}),
+ Mod = State#state.mod,
+ Ctx = Mod:db_found(DbName, State#state.ctx),
+ Pid = start_changes_reader(DbName, 0),
+ State#state{ctx=Ctx, pids=[{DbName, Pid} | Pids]};
+ {false, [{DbName, EndSeq, _}]} ->
+ % No existing change feed running. Found existing checkpoint.
+ % Start a new change reader from last checkpoint.
+ true = ets:insert(Ets, {DbName, EndSeq, false}),
+ Pid = start_changes_reader(DbName, EndSeq),
+ State#state{pids=[{DbName, Pid} | Pids]}
+ end.
+
+
+
+start_changes_reader(DbName, Since) ->
+ spawn_link(?MODULE, changes_reader, [self(), DbName, Since]).
+
+
+changes_reader(Server, DbName, Since) ->
+ {ok, Db} = couch_db:open_int(DbName, [?CTX, sys_db]),
+ ChFun = couch_changes:handle_db_changes(
+ #changes_args{
+ include_docs = true,
+ since = Since,
+ feed = "normal",
+ timeout = infinity
+ }, {json_req, null}, Db),
+ ChFun({fun ?MODULE:changes_reader_cb/3, {Server, DbName}}).
+
+
+changes_reader_cb({change, Change, _}, _, {Server, DbName}) ->
+ ok = gen_server:call(Server, {change, DbName, Change}, infinity),
+ {Server, DbName};
+
+changes_reader_cb({stop, EndSeq}, _, {Server, DbName}) ->
+ ok = gen_server:call(Server, {checkpoint, DbName, EndSeq}, infinity),
+ {Server, DbName};
+
+changes_reader_cb(_, _, Acc) ->
+ Acc.
+
+
+scan_all_dbs(Server, DbSuffix) when is_pid(Server) ->
+ Root = config:get("couchdb", "database_dir", "."),
+ NormRoot = couch_util:normpath(Root),
+ Pat = io_lib:format("~s(\\.[0-9]{10,})?.couch$", [DbSuffix]),
+ filelib:fold_files(Root, lists:flatten(Pat), true,
+ fun(Filename, Acc) ->
+ % shamelessly stolen from couch_server.erl
+ NormFilename = couch_util:normpath(Filename),
+ case NormFilename -- NormRoot of
+ [$/ | RelativeFilename] -> ok;
+ RelativeFilename -> ok
+ end,
+ DbName = ?l2b(filename:rootname(RelativeFilename, ".couch")),
+ Jitter = jitter(Acc),
+ spawn_link(fun() ->
+ timer:sleep(Jitter),
+ gen_server:cast(Server, {resume_scan, DbName})
+ end),
+ Acc + 1
+ end, 1).
+
+
+% calculate random delay proportional to the number of replications
+% on current node, in order to prevent a stampede:
+% - when a source with multiple replication targets fails
+% - when we restart couch_replication_manager
+jitter(N) ->
+ Range = min(2 * N * ?AVG_DELAY_MSEC, ?MAX_DELAY_MSEC),
+ random:uniform(Range).
+
+
+is_design_doc({Change}) ->
+ case lists:keyfind(<<"id">>, 1, Change) of
+ false ->
+ false;
+ {_, Id} ->
+ is_design_doc_id(Id)
+ end.
+
+
+is_design_doc_id(<<?DESIGN_DOC_PREFIX, _/binary>>) ->
+ true;
+
+is_design_doc_id(_) ->
+ false.
+
+
+
+-ifdef(TEST).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+
+-define(MOD, multidb_test_module).
+-define(SUFFIX, <<"suff">>).
+-define(DBNAME, <<"shards/40000000-5fffffff/acct/suff.0123456789">>).
+
+couch_multidb_changes_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ t_handle_call_change(),
+ t_handle_call_change_filter_design_docs(),
+ t_handle_call_checkpoint_new(),
+ t_handle_call_checkpoint_existing(),
+ t_handle_info_created(),
+ t_handle_info_deleted(),
+ t_handle_info_updated(),
+ t_handle_info_other_event(),
+ t_handle_info_created_other_db(),
+ t_handle_info_scanner_exit_normal(),
+ t_handle_info_scanner_crashed(),
+ t_handle_info_event_server_exited(),
+ t_handle_info_unknown_pid_exited(),
+ t_handle_info_change_feed_exited(),
+ t_handle_info_change_feed_exited_and_need_rescan(),
+ t_spawn_changes_reader(),
+ t_changes_reader_cb_change(),
+ t_changes_reader_cb_stop(),
+ t_changes_reader_cb_other(),
+ t_handle_call_resume_scan_no_chfeed_no_ets_entry(),
+ t_handle_call_resume_scan_chfeed_no_ets_entry(),
+ t_handle_call_resume_scan_chfeed_ets_entry(),
+ t_handle_call_resume_scan_no_chfeed_ets_entry(),
+ t_start_link(),
+ t_start_link_no_ddocs(),
+ t_scanner_finds_shard(),
+ t_misc_gen_server_callbacks()
+ ]
+ }.
+
+setup() ->
+ mock_logs(),
+ mock_callback_mod(),
+ meck:expect(couch_event, register_all, 1, ok),
+ meck:expect(config, get, fun("couchdb", "database_dir", _) -> ?TEMPDIR end),
+ mock_changes_reader(),
+ % create process to stand in for couch_ever_server
+ % mocking erlang:monitor doesn't, so give it real process to monitor
+ EvtPid = spawn_link(fun() -> receive looper -> ok end end),
+ true = register(couch_event_server, EvtPid),
+ EvtPid.
+
+
+teardown(EvtPid) ->
+ unlink(EvtPid),
+ exit(EvtPid, kill),
+ meck:unload(),
+ delete_shard_file(?DBNAME).
+
+
+t_handle_call_change() ->
+ ?_test(begin
+ State = mock_state(),
+ Change = change_row(<<"blah">>),
+ handle_call_ok({change, ?DBNAME, Change}, State),
+ ?assert(meck:validate(?MOD)),
+ ?assert(meck:called(?MOD, db_change, [?DBNAME, Change, zig]))
+ end).
+
+t_handle_call_change_filter_design_docs() ->
+ ?_test(begin
+ State0 = mock_state(),
+ State = State0#state{skip_ddocs = true},
+ Change = change_row(<<"_design/blah">>),
+ handle_call_ok({change, ?DBNAME, Change}, State),
+ ?assert(meck:validate(?MOD)),
+ ?assertNot(meck:called(?MOD, db_change, [?DBNAME, Change, zig]))
+ end).
+
+t_handle_call_checkpoint_new() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ State = mock_state(Tid),
+ handle_call_ok({checkpoint, ?DBNAME, 1}, State),
+ ?assertEqual([{?DBNAME, 1, false}], ets:tab2list(Tid)),
+ ets:delete(Tid)
+ end).
+
+t_handle_call_checkpoint_existing() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ State = mock_state(Tid),
+ true = ets:insert(Tid, {?DBNAME, 1, true}),
+ handle_call_ok({checkpoint, ?DBNAME, 2}, State),
+ ?assertEqual([{?DBNAME, 2, true}], ets:tab2list(Tid)),
+ ets:delete(Tid)
+ end).
+
+t_handle_info_created() ->
+ ?_test(begin
+ State = mock_state(),
+ handle_info_check({'$couch_event', ?DBNAME, created}, State),
+ ?assert(meck:validate(?MOD)),
+ ?assert(meck:called(?MOD, db_created, [?DBNAME, zig]))
+ end).
+
+t_handle_info_deleted() ->
+ ?_test(begin
+ State = mock_state(),
+ handle_info_check({'$couch_event', ?DBNAME, deleted}, State),
+ ?assert(meck:validate(?MOD)),
+ ?assert(meck:called(?MOD, db_deleted, [?DBNAME, zig]))
+ end).
+
+t_handle_info_updated() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ State = mock_state(Tid),
+ handle_info_check({'$couch_event', ?DBNAME, updated}, State),
+ ?assert(meck:validate(?MOD)),
+ ?assert(meck:called(?MOD, db_found, [?DBNAME, zig]))
+ end).
+
+t_handle_info_other_event() ->
+ ?_test(begin
+ State = mock_state(),
+ handle_info_check({'$couch_event', ?DBNAME, somethingelse}, State)
+ end).
+
+t_handle_info_created_other_db() ->
+ ?_test(begin
+ State = mock_state(),
+ handle_info_check({'$couch_event', <<"otherdb">>, created}, State),
+ ?assertNot(meck:called(?MOD, db_created, [?DBNAME, zig]))
+ end).
+
+t_handle_info_scanner_exit_normal() ->
+ ?_test(begin
+ Res = handle_info({'EXIT', spid, normal}, mock_state()),
+ ?assertMatch({noreply, _}, Res),
+ {noreply, RState} = Res,
+ ?assertEqual(nil, RState#state.scanner)
+ end).
+
+t_handle_info_scanner_crashed() ->
+ ?_test(begin
+ Res = handle_info({'EXIT', spid, oops}, mock_state()),
+ ?assertMatch({stop, {scanner_died, oops}, _State}, Res)
+ end).
+
+t_handle_info_event_server_exited() ->
+ ?_test(begin
+ Res = handle_info({'DOWN', esref, type, espid, reason}, mock_state()),
+ ?assertMatch({stop, {couch_event_server_died, reason},_}, Res)
+ end).
+
+t_handle_info_unknown_pid_exited() ->
+ ?_test(begin
+ State0 = mock_state(),
+ Res0 = handle_info({'EXIT', somepid, normal}, State0),
+ ?assertMatch({noreply, State0}, Res0),
+ State1 = mock_state(),
+ Res1 = handle_info({'EXIT', somepid, oops}, State1),
+ ?assertMatch({stop, {unexpected_exit, somepid, oops}, State1}, Res1)
+ end).
+
+t_handle_info_change_feed_exited() ->
+ ?_test(begin
+ Tid0 = mock_ets(),
+ State0 = mock_state(Tid0, cpid),
+ Res0 = handle_info({'EXIT', cpid, normal}, State0),
+ ?assertMatch({noreply, _}, Res0),
+ {noreply, RState0} = Res0,
+ ?assertEqual([], RState0#state.pids),
+ ets:delete(Tid0),
+ Tid1 = mock_ets(),
+ State1 = mock_state(Tid1, cpid),
+ Res1 = handle_info({'EXIT', cpid, oops}, State1),
+ ?assertMatch({noreply, _}, Res1),
+ {noreply, RState1} = Res1,
+ ?assertEqual([], RState1#state.pids),
+ ets:delete(Tid1)
+ end).
+
+t_handle_info_change_feed_exited_and_need_rescan() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ true = ets:insert(Tid, {?DBNAME, 1, true}),
+ State = mock_state(Tid, cpid),
+ Res = handle_info({'EXIT', cpid, normal}, State),
+ ?assertMatch({noreply, _}, Res),
+ {noreply, RState} = Res,
+ % rescan flag should have been reset to false
+ ?assertEqual([{?DBNAME, 1, false}], ets:tab2list(Tid)),
+ % a mock change feed process should be running
+ [{?DBNAME, Pid}] = RState#state.pids,
+ ?assert(is_pid(Pid)),
+ ChArgs = kill_mock_change_reader_and_get_its_args(Pid),
+ ?assertEqual({self(), ?DBNAME}, ChArgs),
+ ets:delete(Tid)
+ end).
+
+t_spawn_changes_reader() ->
+ ?_test(begin
+ Pid = start_changes_reader(?DBNAME, 3),
+ ?assert(erlang:is_process_alive(Pid)),
+ ChArgs = kill_mock_change_reader_and_get_its_args(Pid),
+ ?assertEqual({self(), ?DBNAME}, ChArgs),
+ ?assert(meck:validate(couch_db)),
+ ?assert(meck:validate(couch_changes)),
+ ?assert(meck:called(couch_db, open_int, [?DBNAME, [?CTX, sys_db]])),
+ ?assert(meck:called(couch_changes, handle_db_changes, [
+ #changes_args{
+ include_docs = true,
+ since = 3,
+ feed = "normal",
+ timeout = infinity
+ }, {json_req, null}, db]))
+ end).
+
+t_changes_reader_cb_change() ->
+ ?_test(begin
+ {ok, Pid} = start_link(?SUFFIX, ?MOD, zig, []),
+ Change = change_row(<<"blah">>),
+ ChArg = {change, Change, ignore},
+ {Pid, ?DBNAME} = changes_reader_cb(ChArg, chtype, {Pid, ?DBNAME}),
+ ?assert(meck:called(?MOD, db_change, [?DBNAME, Change, zig])),
+ unlink(Pid),
+ exit(Pid, kill)
+ end).
+
+t_changes_reader_cb_stop() ->
+ ?_test(begin
+ {ok, Pid} = start_link(?SUFFIX, ?MOD, zig, []),
+ ChArg = {stop, 11},
+ {Pid, ?DBNAME} = changes_reader_cb(ChArg, chtype, {Pid, ?DBNAME}),
+ % We checkpoint on stop, check if checkpointed at correct sequence
+ #state{tid = Tid} = sys:get_state(Pid),
+ ?assertEqual([{?DBNAME, 11, false}], ets:tab2list(Tid)),
+ unlink(Pid),
+ exit(Pid, kill)
+ end).
+
+t_changes_reader_cb_other() ->
+ ?_assertEqual(acc, changes_reader_cb(other, chtype, acc)).
+
+t_handle_call_resume_scan_no_chfeed_no_ets_entry() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ State = mock_state(Tid),
+ RState = resume_scan(?DBNAME, State),
+ % Check if inserted checkpoint entry in ets starting at 0
+ ?assertEqual([{?DBNAME, 0, false}], ets:tab2list(Tid)),
+ % Check if called db_found callback
+ ?assert(meck:called(?MOD, db_found, [?DBNAME, zig])),
+ % Check if started a change reader
+ [{?DBNAME, Pid}] = RState#state.pids,
+ ChArgs = kill_mock_change_reader_and_get_its_args(Pid),
+ ?assertEqual({self(), ?DBNAME}, ChArgs),
+ ?assert(meck:called(couch_changes, handle_db_changes, [
+ #changes_args{
+ include_docs = true,
+ since = 0,
+ feed = "normal",
+ timeout = infinity
+ }, {json_req, null}, db])),
+ ets:delete(Tid)
+ end).
+
+t_handle_call_resume_scan_chfeed_no_ets_entry() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ Pid = start_changes_reader(?DBNAME, 0),
+ State = mock_state(Tid, Pid),
+ resume_scan(?DBNAME, State),
+ % Check ets checkpoint is set to 0 and rescan = true
+ ?assertEqual([{?DBNAME, 0, true}], ets:tab2list(Tid)),
+ ets:delete(Tid),
+ kill_mock_change_reader_and_get_its_args(Pid)
+ end).
+
+t_handle_call_resume_scan_chfeed_ets_entry() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ true = ets:insert(Tid, [{?DBNAME, 2, false}]),
+ Pid = start_changes_reader(?DBNAME, 1),
+ State = mock_state(Tid, Pid),
+ resume_scan(?DBNAME, State),
+ % Check ets checkpoint is set to same endseq but rescan = true
+ ?assertEqual([{?DBNAME, 2, true}], ets:tab2list(Tid)),
+ ets:delete(Tid),
+ kill_mock_change_reader_and_get_its_args(Pid)
+ end).
+
+t_handle_call_resume_scan_no_chfeed_ets_entry() ->
+ ?_test(begin
+ Tid = mock_ets(),
+ true = ets:insert(Tid, [{?DBNAME, 1, true}]),
+ State = mock_state(Tid),
+ RState = resume_scan(?DBNAME, State),
+ % Check if reset rescan to false but kept same endseq
+ ?assertEqual([{?DBNAME, 1, false}], ets:tab2list(Tid)),
+ % Check if started a change reader
+ [{?DBNAME, Pid}] = RState#state.pids,
+ ChArgs = kill_mock_change_reader_and_get_its_args(Pid),
+ ?assertEqual({self(), ?DBNAME}, ChArgs),
+ ?assert(meck:called(couch_changes, handle_db_changes, [
+ #changes_args{
+ include_docs = true,
+ since = 1,
+ feed = "normal",
+ timeout = infinity
+ }, {json_req, null}, db])),
+ ets:delete(Tid)
+ end).
+
+t_start_link() ->
+ ?_test(begin
+ {ok, Pid} = start_link(?SUFFIX, ?MOD, nil, []),
+ ?assert(is_pid(Pid)),
+ ?assertMatch(#state{
+ mod = ?MOD,
+ suffix = ?SUFFIX,
+ ctx = nil,
+ pids = [],
+ skip_ddocs = false
+ }, sys:get_state(Pid)),
+ unlink(Pid),
+ exit(Pid, kill),
+ ?assert(meck:called(couch_event, register_all, [Pid]))
+ end).
+
+t_start_link_no_ddocs() ->
+ ?_test(begin
+ {ok, Pid} = start_link(?SUFFIX, ?MOD, nil, [skip_ddocs]),
+ ?assert(is_pid(Pid)),
+ ?assertMatch(#state{
+ mod = ?MOD,
+ suffix = ?SUFFIX,
+ ctx = nil,
+ pids = [],
+ skip_ddocs = true
+ }, sys:get_state(Pid)),
+ unlink(Pid),
+ exit(Pid, kill)
+ end).
+
+t_scanner_finds_shard() ->
+ ?_test(begin
+ ok = create_shard_file(?DBNAME),
+ {ok, Pid} = start_link(?SUFFIX, ?MOD, zig, []),
+ ok = meck:wait(?MOD, db_found, [?DBNAME, zig], 2000),
+ unlink(Pid),
+ exit(Pid, kill)
+ end).
+
+t_misc_gen_server_callbacks() ->
+ ?_test(begin
+ ?assertEqual(ok, terminate(reason, state)),
+ ?assertEqual({ok, state}, code_change(old, state, extra))
+ end).
+
+
+% Test helper functions
+
+mock_logs() ->
+ meck:expect(couch_log, error, 2, ok),
+ meck:expect(couch_log, notice, 2, ok),
+ meck:expect(couch_log, info, 2, ok),
+ meck:expect(couch_log, debug, 2, ok).
+
+
+mock_callback_mod() ->
+ meck:new(?MOD, [non_strict]),
+ meck:expect(?MOD, db_created, fun(_DbName, Ctx) -> Ctx end),
+ meck:expect(?MOD, db_deleted, fun(_DbName, Ctx) -> Ctx end),
+ meck:expect(?MOD, db_found, fun(_DbName, Ctx) -> Ctx end),
+ meck:expect(?MOD, db_change, fun(_DbName, _Change, Ctx) -> Ctx end).
+
+
+mock_changes_reader_loop({_CbFun, {Server, DbName}}) ->
+ receive
+ die ->
+ exit({Server, DbName})
+ end.
+
+kill_mock_change_reader_and_get_its_args(Pid) ->
+ Ref = monitor(process, Pid),
+ unlink(Pid),
+ Pid ! die,
+ receive
+ {'DOWN', Ref, _, Pid, {Server, DbName}} ->
+ {Server, DbName}
+ after 1000 ->
+ erlang:error(spawn_change_reader_timeout)
+ end.
+
+mock_changes_reader() ->
+ meck:expect(couch_db, open_int, 2, {ok, db}),
+ meck:expect(couch_changes, handle_db_changes,
+ fun(_ChArgs, _Req, db) ->
+ fun mock_changes_reader_loop/1
+ end).
+
+mock_ets() ->
+ ets:new(multidb_test_ets, [set, public]).
+
+mock_state() ->
+ #state{
+ mod = ?MOD,
+ ctx = zig,
+ suffix = ?SUFFIX,
+ event_server = esref,
+ scanner = spid,
+ pids = []}.
+
+mock_state(Ets) ->
+ State = mock_state(),
+ State#state{tid = Ets}.
+
+mock_state(Ets, Pid) ->
+ State = mock_state(Ets),
+ State#state{pids = [{?DBNAME, Pid}]}.
+
+
+change_row(Id) when is_binary(Id) ->
+ {[
+ {<<"seq">>, 1},
+ {<<"id">>, Id},
+ {<<"changes">>, [{[{<<"rev">>, <<"1-f00">>}]}]},
+ {doc, {[{<<"_id">>, Id}, {<<"_rev">>, <<"1-f00">>}]}}
+ ]}.
+
+shard_fname(DbName) ->
+ filename:join([?TEMPDIR, binary_to_list(DbName) ++ ".couch"]).
+
+delete_shard_file(DbName) ->
+ Path = shard_fname(DbName),
+ case filelib:is_file(Path) of
+ true ->
+ ok = file:delete(Path);
+ false ->
+ ok
+ end.
+
+create_shard_file(DbName) ->
+ Path = shard_fname(DbName),
+ ok = filelib:ensure_dir(Path),
+ ok = file:write_file(Path, <<>>).
+
+
+handle_call_ok(Msg, State) ->
+ ?assertMatch({reply, ok, _}, handle_call(Msg, from, State)).
+
+handle_info_check(Msg, State) ->
+ ?assertMatch({noreply, _}, handle_info(Msg, State)).
+
+-endif.
--
To stop receiving notification emails like this one, please contact
"commits@couchdb.apache.org" <co...@couchdb.apache.org>.