You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@couchdb.apache.org by GitBox <gi...@apache.org> on 2021/02/08 20:43:32 UTC
[GitHub] [couchdb] nickva opened a new pull request #3364: Fair share scheduler 3.x
nickva opened a new pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364
This is just a wip/draft PR to help see the diff between 3.x and the branch
RFC PR is https://github.com/apache/couchdb-documentation/pull/617
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585076188
##########
File path: src/couch_replicator/src/couch_replicator_scheduler.erl
##########
@@ -1621,13 +1662,32 @@ mock_state(MaxJobs, MaxChurn) ->
}.
+rep() ->
+ #rep{options = [], user_ctx = #user_ctx{}}.
+
+
+rep(Src, Tgt) ->
+ Rep = rep(),
+ Rep#rep{source = Src, target = Tgt}.
+
+
+continuous_rep() ->
+ #rep{options = [{continuous, true}], user_ctx = #user_ctx{}}.
+
+
+continuous_rep(Src, Tgt) ->
+ Rep = continuous_rep(),
+ Rep#rep{source = Src, target = Tgt}.
+
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585079388
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,805 @@
+% 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 the "Fair Share" algorithm by Judy Kay and Piers
+% Lauder [1] and applies it to the scheduling of replication jobs.
+%
+% The main idea is _replicator dbs can have a configurable number of "shares"
+% assigned to them. Shares is an abstract quantity from 1 to 1000. The default
+% is 100. Jobs from _replicator databases with more shares get proportionally a
+% higher chance to run than those from databases with a lower number of shares.
+%
+% Every scheduler cycle running jobs are "charged" based on how much time they
+% spent running during that cycle. At the end of the cycle the accumulated
+% charges for each job, the number of shares configured, and the total number
+% of jobs in the pending queue from the same _replicator db, are used to
+% calculate new priority values for all the jobs. To match the algorithm from
+% the paper, jobs with lower priority values are the ones at the front of the
+% run queue and have a higher chance of running.
+%
+% Here is how charges, shares, and number of sibling jobs affect the
+% priority value:
+%
+% 1) Jobs from dbs with higher configured shares get assigned lower
+% priority values and so stay closer to the front of the queue.
+%
+% 2) Jobs from dbs with many other jobs (many siblings) get assigned a
+% higher priority value, so they get pushed further down the queue
+% and have a lower chance of running.
+%
+% 3) Jobs which run longer accumulate more charges and get assigned a
+% higher priority value and get to wait longer to run.
+%
+% In order to prevent job starvation, all job priorities are periodicaly
+% decayed (decreased). This effectively moves all the jobs towards the front of
+% the run queue. So, in effect, there are two competing processes: one
+% uniformly moves all jobs to the front, and the other throws them back in
+% proportion to those factors mentioned above. The speed of this uniform
+% priority decay is controlled by the priority_coeff parameter.
+%
+% In order to prevent jobs from low shares dbs from "cheating" by getting
+% deleted and immediately re-added, charges are accumulated using a
+% historically decayed usage value. The speed of the usage decay is controlled
+% by the `usage_coeff = 0.5` parameter.
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+ update_shares/2,
+ reset_shares/1,
+ job_added/1,
+ job_removed/1,
+ update/3,
+ priority/1,
+ charge/3
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute), then if the job stops running it will take about 26 cycles (minutes)
+% for it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. This coefficient defines a maximum
+% time window over which this algorithm would operate. For example, if this
+% value is too small (0.1), after a few cycles quite a few jobs would end up at
+% priority 0, and would render this algorithm useless. The default value of
+% 0.98 is picked such that if a job ran for one scheduler cycle, then didn't
+% get to run for 7 hours, it would still have priority > 0. 7 hours was picked
+% as it was close enought to 8 hours which is the default maximum error backoff
+% interval.
+%
+% Example calculation:
+% shares = 100
+% usage after 1 minute cycle run = 60000000
+% initial priority = 60000000 / (100 * 100) = 6000
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 431 / 60 ~= 7 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) -> update_shares(K, V) end, get_config_shares()).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, bounded(Shares, ?MIN_SHARES, ?MAX_SHARES)}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ % If the entry is not present {Key, 0} is used as the default
Review comment:
I thought that initially too but he ets table docs say: https://erlang.org/doc/man/ets.html#update_counter-4
> If a default object Default is specified, it is used as the object to be updated if the key is missing from the table.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581433522
##########
File path: src/couch_replicator/src/couch_replicator_scheduler.erl
##########
@@ -734,6 +775,24 @@ reset_job_process(#job{} = Job) ->
-spec reschedule(#state{}) -> ok.
reschedule(State) ->
+ % Charge all running jobs for the current interval
+ RunningJobs = running_jobs(),
+ Now = os:timestamp(),
+ lists:foreach(fun(Job) ->
+ couch_replicator_share:charge(Job, State#state.interval, Now)
+ end, RunningJobs),
+
+ % Update usage table
+ couch_replicator_share:update_usage(),
+
+ % Decay all the process priorities
+ couch_replicator_share:decay_priorities(),
+
+ % Adjust running process priorities
+ lists:foreach(fun(Job) ->
+ couch_replicator_share:update_priority(Job)
+ end, RunningJobs),
Review comment:
Good idea. I was slightly worried about passing in the whole list of running jobs into the call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581438436
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
Review comment:
Oh good point. That is a subtle behavior. I'll add a comment there
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] davisp commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
davisp commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585073555
##########
File path: src/couch_replicator/src/couch_replicator_scheduler.erl
##########
@@ -1621,13 +1662,32 @@ mock_state(MaxJobs, MaxChurn) ->
}.
+rep() ->
+ #rep{options = [], user_ctx = #user_ctx{}}.
+
+
+rep(Src, Tgt) ->
+ Rep = rep(),
+ Rep#rep{source = Src, target = Tgt}.
+
+
+continuous_rep() ->
+ #rep{options = [{continuous, true}], user_ctx = #user_ctx{}}.
+
+
+continuous_rep(Src, Tgt) ->
+ Rep = continuous_rep(),
+ Rep#rep{source = Src, target = Tgt}.
+
Review comment:
Whoa whoa whoa. Extra blank line!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r582159719
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
Review comment:
Yap. I added a `new get_config_shares/0` function which validates the input.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] davisp commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
davisp commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585033588
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
Review comment:
Excellent update.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva merged pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva merged pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#issuecomment-796938965
Will merge the PR and noting that I still owe a documentation update for it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] davisp commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
davisp commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581400024
##########
File path: rel/overlay/etc/default.ini
##########
@@ -482,6 +482,33 @@ ssl_certificate_max_depth = 3
; or 403 response this setting is not needed.
;session_refresh_interval_sec = 550
+; Usage coefficient decays historic fair share usage every scheduling
+; cycle. The value must be between 0.0 and 1.0. Lower values will
+; ensure historic usage decays quicker and higher values means it will
+; be remembered longer.
+;usage_coeff = 0.5
+
+; Priority coefficient decays all the fair share job priorities such
+; that they uniformly drift towards the front of the run queue. At the
+; default value of 0.98 it will take about 430 scheduler cycle (7
+; hours) for a single job which ran for 1 minute to drift towards the
+; front of the queue (get assigned priority 0). 7 hours was picked as
+; it is close the maximum error backoff interval of about 8 hours. The
+; value must be between 0.0 and 1.0. A too low of a value, coupled
+; with a lower max jobs or churn parameter could end up making the
+; majority of job priority 0 too quickly and canceling the effect of
+; the fair share algorithm.
+;priority_coeff = 0.98
+
+
+[replicator.shares]
+; Fair share configuration section. More shares result in a higher
+; chance that jobs from that db get to run. The default value is 100,
+; minimum is 1 and maximum is 1000. The configuration may be set even
+; if the database wasn't created yet.
Review comment:
Minor language nit, I'd simplify to "if the database does not exist."
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
Review comment:
This 7 hour comment is slightly worrisome. My guess is that you're talking about in some extreme case where a cluster has significantly more replication jobs than bandwidth for processing them, however, folks reading that their replications make pause for 7 hours at a time is likely not good. I'd wager we just need to reword in what situations this 7 hour thing comes into effect.
##########
File path: src/couch_replicator/src/couch_replicator_scheduler.erl
##########
@@ -734,6 +775,24 @@ reset_job_process(#job{} = Job) ->
-spec reschedule(#state{}) -> ok.
reschedule(State) ->
+ % Charge all running jobs for the current interval
+ RunningJobs = running_jobs(),
+ Now = os:timestamp(),
+ lists:foreach(fun(Job) ->
+ couch_replicator_share:charge(Job, State#state.interval, Now)
+ end, RunningJobs),
+
+ % Update usage table
+ couch_replicator_share:update_usage(),
+
+ % Decay all the process priorities
+ couch_replicator_share:decay_priorities(),
+
+ % Adjust running process priorities
+ lists:foreach(fun(Job) ->
+ couch_replicator_share:update_priority(Job)
+ end, RunningJobs),
Review comment:
This whole hunk feels as though it should be a single call to `couch_replicator_share` and that the multiple calls are leaking implementation internals outside of that module.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
Review comment:
How does this behave if I delete the shares key when jobs exist? Granted I may be about to read what happens.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
Review comment:
Probably want to guard on user input here.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
Review comment:
Reckon it makes sense to do a `=<` comparison instead?
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
Review comment:
Sneaky sneaky sneaky.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
+
+
+key(#job{} = Job) ->
+ Rep = Job#job.rep,
+ case is_binary(Rep#rep.db_name) of
+ true -> mem3:dbname(Rep#rep.db_name);
+ false -> (Rep#rep.user_ctx)#user_ctx.name
+ end.
+
+
+% Jobs are charged based on the amount of time the job was running during the
+% last scheduling interval. The time units used are microseconds in order to
+% have a large enough usage values so that when priority is calculated the
+% rounded value won't be rounded off to 0 easily. The formula for the priority
+% calculation is:
+%
+% Priority = (Usage * NumJobs) / Shares^2
+%
+% Then in the worst case of a single job in the db, running only for one
+% second,for one job, with 1000 (max) shares, the priority would be:
+%
+% 1000000 * 1 / (1000^2) = 1
+%
+job_charges(#job{} = Job, IntervalMSec, {_, _, _} = Now) ->
+ TimeRunning = timer:now_diff(Now, last_started(Job)),
+ IntervalUSec = IntervalMSec * 1000,
Review comment:
The unit conversion here strikes me as not awesome since the conversion logic is implicitly shared between modules rather than being an internal detail of this module.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
Review comment:
You might leave a note that update_counter applies the increment even if the default `{Key, 0}` is needed. I had to double check the docs to make sure my memory was correct.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
+
+
+key(#job{} = Job) ->
+ Rep = Job#job.rep,
+ case is_binary(Rep#rep.db_name) of
+ true -> mem3:dbname(Rep#rep.db_name);
+ false -> (Rep#rep.user_ctx)#user_ctx.name
+ end.
+
+
+% Jobs are charged based on the amount of time the job was running during the
+% last scheduling interval. The time units used are microseconds in order to
+% have a large enough usage values so that when priority is calculated the
+% rounded value won't be rounded off to 0 easily. The formula for the priority
+% calculation is:
+%
+% Priority = (Usage * NumJobs) / Shares^2
+%
+% Then in the worst case of a single job in the db, running only for one
+% second,for one job, with 1000 (max) shares, the priority would be:
+%
+% 1000000 * 1 / (1000^2) = 1
+%
+job_charges(#job{} = Job, IntervalMSec, {_, _, _} = Now) ->
+ TimeRunning = timer:now_diff(Now, last_started(Job)),
+ IntervalUSec = IntervalMSec * 1000,
+ min(IntervalUSec, max(0, TimeRunning)).
+
+
+last_started(#job{} = Job) ->
+ case lists:keyfind(started, 1, Job#job.history) of
+ false -> {0, 0, 0}; % In case user set too low of a max history
+ {started, When} -> When
+ end.
+
+
+% Config helper functions
+
+priority_coeff() ->
+ % This is the K2 coefficient from [1]
+ Default = ?DEFAULT_PRIORITY_COEFF,
+ Val = float_val(config:get("replicator", "priority_coeff"), Default),
+ max(0.0, min(1.0, Val)).
Review comment:
Minor nit, but a helper of `bounded(0.0, 1.0, Val)` or something of that nature might be slightly easier to read since this min/max snippet is repeated a number of times.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r582158885
##########
File path: rel/overlay/etc/default.ini
##########
@@ -482,6 +482,33 @@ ssl_certificate_max_depth = 3
; or 403 response this setting is not needed.
;session_refresh_interval_sec = 550
+; Usage coefficient decays historic fair share usage every scheduling
+; cycle. The value must be between 0.0 and 1.0. Lower values will
+; ensure historic usage decays quicker and higher values means it will
+; be remembered longer.
+;usage_coeff = 0.5
+
+; Priority coefficient decays all the fair share job priorities such
+; that they uniformly drift towards the front of the run queue. At the
+; default value of 0.98 it will take about 430 scheduler cycle (7
+; hours) for a single job which ran for 1 minute to drift towards the
+; front of the queue (get assigned priority 0). 7 hours was picked as
+; it is close the maximum error backoff interval of about 8 hours. The
+; value must be between 0.0 and 1.0. A too low of a value, coupled
+; with a lower max jobs or churn parameter could end up making the
+; majority of job priority 0 too quickly and canceling the effect of
+; the fair share algorithm.
+;priority_coeff = 0.98
+
+
+[replicator.shares]
+; Fair share configuration section. More shares result in a higher
+; chance that jobs from that db get to run. The default value is 100,
+; minimum is 1 and maximum is 1000. The configuration may be set even
+; if the database wasn't created yet.
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581441924
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
Review comment:
Since we are guarding the coefficient values to be between `0.0` and `1.0` (and never negative) and the tables we are decaying are never decremented with `-1`, thought `0` might be more clear `=<` would imply they could be negative. If `=<` is easier to read and understand, I can change it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581437094
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
Review comment:
Good point, yeah that does sound confusing. I'll try to reword it.
The idea with 7 hours is that, if the jobs doesn't get to run again it will still have a usable priority value so it can be ordered against other jobs, so this algorithm will be "usable" in that time period. Bu if, say, this coefficient was 0.1, jobs might decay their priorities to too quickly and the sharing algorithm won't be any use as it will simply order jobs by their last_started timestamp like it did before.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581437094
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
Review comment:
Good point, yeah that does sound confusing. I'll try to reword it.
The idea with 7 hours is that, if the jobs doesn't get to run again it will still have a usable priority value so it can be ordered against other jobs, so this algorithm will be "usable" in that time period. Bu if, say, this coefficient was 0.1, jobs might decay their priorities to too quickly and the sharing algorithm won't be any use as it will simply order jobs by their last_started timestamp like it did before.
Updated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] davisp commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
davisp commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585075990
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,805 @@
+% 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 the "Fair Share" algorithm by Judy Kay and Piers
+% Lauder [1] and applies it to the scheduling of replication jobs.
+%
+% The main idea is _replicator dbs can have a configurable number of "shares"
+% assigned to them. Shares is an abstract quantity from 1 to 1000. The default
+% is 100. Jobs from _replicator databases with more shares get proportionally a
+% higher chance to run than those from databases with a lower number of shares.
+%
+% Every scheduler cycle running jobs are "charged" based on how much time they
+% spent running during that cycle. At the end of the cycle the accumulated
+% charges for each job, the number of shares configured, and the total number
+% of jobs in the pending queue from the same _replicator db, are used to
+% calculate new priority values for all the jobs. To match the algorithm from
+% the paper, jobs with lower priority values are the ones at the front of the
+% run queue and have a higher chance of running.
+%
+% Here is how charges, shares, and number of sibling jobs affect the
+% priority value:
+%
+% 1) Jobs from dbs with higher configured shares get assigned lower
+% priority values and so stay closer to the front of the queue.
+%
+% 2) Jobs from dbs with many other jobs (many siblings) get assigned a
+% higher priority value, so they get pushed further down the queue
+% and have a lower chance of running.
+%
+% 3) Jobs which run longer accumulate more charges and get assigned a
+% higher priority value and get to wait longer to run.
+%
+% In order to prevent job starvation, all job priorities are periodicaly
+% decayed (decreased). This effectively moves all the jobs towards the front of
+% the run queue. So, in effect, there are two competing processes: one
+% uniformly moves all jobs to the front, and the other throws them back in
+% proportion to those factors mentioned above. The speed of this uniform
+% priority decay is controlled by the priority_coeff parameter.
+%
+% In order to prevent jobs from low shares dbs from "cheating" by getting
+% deleted and immediately re-added, charges are accumulated using a
+% historically decayed usage value. The speed of the usage decay is controlled
+% by the `usage_coeff = 0.5` parameter.
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+ update_shares/2,
+ reset_shares/1,
+ job_added/1,
+ job_removed/1,
+ update/3,
+ priority/1,
+ charge/3
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute), then if the job stops running it will take about 26 cycles (minutes)
+% for it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. This coefficient defines a maximum
+% time window over which this algorithm would operate. For example, if this
+% value is too small (0.1), after a few cycles quite a few jobs would end up at
+% priority 0, and would render this algorithm useless. The default value of
+% 0.98 is picked such that if a job ran for one scheduler cycle, then didn't
+% get to run for 7 hours, it would still have priority > 0. 7 hours was picked
+% as it was close enought to 8 hours which is the default maximum error backoff
+% interval.
+%
+% Example calculation:
+% shares = 100
+% usage after 1 minute cycle run = 60000000
+% initial priority = 60000000 / (100 * 100) = 6000
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 431 / 60 ~= 7 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) -> update_shares(K, V) end, get_config_shares()).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, bounded(Shares, ?MIN_SHARES, ?MAX_SHARES)}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ % If the entry is not present {Key, 0} is used as the default
Review comment:
Technically, I think the 0 is ignored and the default value is the increment. I dunno if that's important enough of a distinction or not.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581438436
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
Review comment:
Oh good point. That is a subtle behavior. I added comments to all the update_counter/4 calls
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] davisp commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
davisp commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585036181
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
Review comment:
My personal preference is to always use the range condition rather than a point condition when it makes sense like here. Even though we expect it to never go negative because math, its possible the math ends up changing and this assumption doesn't hold.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585037305
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
Review comment:
Excellent point about it changing in the future. Agree. Will update it
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] davisp commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
davisp commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585081401
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,805 @@
+% 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 the "Fair Share" algorithm by Judy Kay and Piers
+% Lauder [1] and applies it to the scheduling of replication jobs.
+%
+% The main idea is _replicator dbs can have a configurable number of "shares"
+% assigned to them. Shares is an abstract quantity from 1 to 1000. The default
+% is 100. Jobs from _replicator databases with more shares get proportionally a
+% higher chance to run than those from databases with a lower number of shares.
+%
+% Every scheduler cycle running jobs are "charged" based on how much time they
+% spent running during that cycle. At the end of the cycle the accumulated
+% charges for each job, the number of shares configured, and the total number
+% of jobs in the pending queue from the same _replicator db, are used to
+% calculate new priority values for all the jobs. To match the algorithm from
+% the paper, jobs with lower priority values are the ones at the front of the
+% run queue and have a higher chance of running.
+%
+% Here is how charges, shares, and number of sibling jobs affect the
+% priority value:
+%
+% 1) Jobs from dbs with higher configured shares get assigned lower
+% priority values and so stay closer to the front of the queue.
+%
+% 2) Jobs from dbs with many other jobs (many siblings) get assigned a
+% higher priority value, so they get pushed further down the queue
+% and have a lower chance of running.
+%
+% 3) Jobs which run longer accumulate more charges and get assigned a
+% higher priority value and get to wait longer to run.
+%
+% In order to prevent job starvation, all job priorities are periodicaly
+% decayed (decreased). This effectively moves all the jobs towards the front of
+% the run queue. So, in effect, there are two competing processes: one
+% uniformly moves all jobs to the front, and the other throws them back in
+% proportion to those factors mentioned above. The speed of this uniform
+% priority decay is controlled by the priority_coeff parameter.
+%
+% In order to prevent jobs from low shares dbs from "cheating" by getting
+% deleted and immediately re-added, charges are accumulated using a
+% historically decayed usage value. The speed of the usage decay is controlled
+% by the `usage_coeff = 0.5` parameter.
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+ update_shares/2,
+ reset_shares/1,
+ job_added/1,
+ job_removed/1,
+ update/3,
+ priority/1,
+ charge/3
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute), then if the job stops running it will take about 26 cycles (minutes)
+% for it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. This coefficient defines a maximum
+% time window over which this algorithm would operate. For example, if this
+% value is too small (0.1), after a few cycles quite a few jobs would end up at
+% priority 0, and would render this algorithm useless. The default value of
+% 0.98 is picked such that if a job ran for one scheduler cycle, then didn't
+% get to run for 7 hours, it would still have priority > 0. 7 hours was picked
+% as it was close enought to 8 hours which is the default maximum error backoff
+% interval.
+%
+% Example calculation:
+% shares = 100
+% usage after 1 minute cycle run = 60000000
+% initial priority = 60000000 / (100 * 100) = 6000
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 431 / 60 ~= 7 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) -> update_shares(K, V) end, get_config_shares()).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, bounded(Shares, ?MIN_SHARES, ?MAX_SHARES)}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ % If the entry is not present {Key, 0} is used as the default
Review comment:
Aha, I misremembered that. Its the key that's replaced:
> The value in place of the key is ignored and replaced by the proper key value.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581438060
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
Review comment:
It will just use the default value. In other words the ?SHARES table should only hold configured (non-default) values. The value may exist before the db is even created.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#issuecomment-782437514
To help test the PR I created a silly script which runs with a local `./dev/run -n1` cluster. It configures replication jobs and shares for three `_replicator` dbs and then continuously samples the proportion of running jobs in each db:
https://gist.github.com/nickva/7e86b3df19537a60372217e0f68b693a
So for the default parameters it might output something like:
```
** run stats num dbs: 3, total jobs: 100
rdb1/_replicator 66 66%
rdb2/_replicator 23 23%
rdb3/_replicator 11 11%
** run stats num dbs: 3, total jobs: 100
rdb1/_replicator 61 61%
rdb2/_replicator 31 31%
rdb3/_replicator 8 8%
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r585037305
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
Review comment:
Excellent point about it changing in the future. Agree. Updated in a fixup
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581438900
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
Review comment:
I'll add a comment on why we keep them as integers (to make update_counter work)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva edited a comment on pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva edited a comment on pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#issuecomment-782437514
To help test the PR I created a silly script which runs with a local `./dev/run -n1` cluster. It configures replication jobs and shares for three `_replicator` dbs and then continuously samples the proportion of running jobs in each db:
https://gist.github.com/nickva/7e86b3df19537a60372217e0f68b693a
So for the default parameters it might output something like:
```
** run stats num dbs: 3, total jobs: 100
rdb1/_replicator 66 66%
rdb2/_replicator 23 23%
rdb3/_replicator 11 11%
** run stats num dbs: 3, total jobs: 100
rdb1/_replicator 61 61%
rdb2/_replicator 31 31%
rdb3/_replicator 8 8%
```
With a configuration where the shares are all even (100) but the dbs have an uneven number of jobs, after 5-10 minutes there should be a roughly even share of running jobs from each db, even though one db has 500 jobs and others have less.
```python
INTERVAL = 20000
CONFIG = {
"replicator": {
"max_jobs": "100",
"max_churn": "20",
"interval" : str(INTERVAL),
},
"replicator.shares": {
RDB1 : "100",
RDB2 : "100",
RDB3 : "100",
}
}
REPS = {
RDB1 : 200,
RDB2 : 300,
RDB3 : 500,
}
```
```
** run stats num dbs: 3, total jobs: 100
rdb1/_replicator 30 30%
rdb2/_replicator 34 34%
rdb3/_replicator 36 36%
** run stats num dbs: 3, total jobs: 100
rdb1/_replicator 29 29%
rdb2/_replicator 33 33%
rdb3/_replicator 38 38%
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581444649
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
+
+
+key(#job{} = Job) ->
+ Rep = Job#job.rep,
+ case is_binary(Rep#rep.db_name) of
+ true -> mem3:dbname(Rep#rep.db_name);
+ false -> (Rep#rep.user_ctx)#user_ctx.name
+ end.
+
+
+% Jobs are charged based on the amount of time the job was running during the
+% last scheduling interval. The time units used are microseconds in order to
+% have a large enough usage values so that when priority is calculated the
+% rounded value won't be rounded off to 0 easily. The formula for the priority
+% calculation is:
+%
+% Priority = (Usage * NumJobs) / Shares^2
+%
+% Then in the worst case of a single job in the db, running only for one
+% second,for one job, with 1000 (max) shares, the priority would be:
+%
+% 1000000 * 1 / (1000^2) = 1
+%
+job_charges(#job{} = Job, IntervalMSec, {_, _, _} = Now) ->
+ TimeRunning = timer:now_diff(Now, last_started(Job)),
+ IntervalUSec = IntervalMSec * 1000,
+ min(IntervalUSec, max(0, TimeRunning)).
+
+
+last_started(#job{} = Job) ->
+ case lists:keyfind(started, 1, Job#job.history) of
+ false -> {0, 0, 0}; % In case user set too low of a max history
+ {started, When} -> When
+ end.
+
+
+% Config helper functions
+
+priority_coeff() ->
+ % This is the K2 coefficient from [1]
+ Default = ?DEFAULT_PRIORITY_COEFF,
+ Val = float_val(config:get("replicator", "priority_coeff"), Default),
+ max(0.0, min(1.0, Val)).
Review comment:
Good call. Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581444594
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
+
+
+key(#job{} = Job) ->
+ Rep = Job#job.rep,
+ case is_binary(Rep#rep.db_name) of
+ true -> mem3:dbname(Rep#rep.db_name);
+ false -> (Rep#rep.user_ctx)#user_ctx.name
+ end.
+
+
+% Jobs are charged based on the amount of time the job was running during the
+% last scheduling interval. The time units used are microseconds in order to
+% have a large enough usage values so that when priority is calculated the
+% rounded value won't be rounded off to 0 easily. The formula for the priority
+% calculation is:
+%
+% Priority = (Usage * NumJobs) / Shares^2
+%
+% Then in the worst case of a single job in the db, running only for one
+% second,for one job, with 1000 (max) shares, the priority would be:
+%
+% 1000000 * 1 / (1000^2) = 1
+%
+job_charges(#job{} = Job, IntervalMSec, {_, _, _} = Now) ->
+ TimeRunning = timer:now_diff(Now, last_started(Job)),
+ IntervalUSec = IntervalMSec * 1000,
Review comment:
The "charges" here is an abstract quantity as defined internally to this module. I added a [note](4f831d7decc#diff-db809547f022d6df8a8a55d281ffd03db38734923b242369aac80f9a48c9814eR221) why we are charging by the microsecond in order to have a large enough range and keep priorities integers and not floats. But in principle this could be any other measure, like say minutes or number of requests made (or time spent without making requests if we wanted to punish "lazy" jobs). Since we are mostly plugging this into the existing replicator, I thought of passing in the `Interval` as is without converting it to microseconds or seconds outside this module.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581437094
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
Review comment:
Good point, yeah that does sound confusing. I'll try to reword it.
The idea with 7 hours is that, if the jobs doesn't get to run again it will still have a usable priority value so it can be ordered against other jobs, so this algorithm will be "usable" in that time period. But say, this coefficient was 0.1, jobs might decay their priorities too quickly and the algorithm won't be any use as it will simply order jobs by their last_started timestamp like it did before.
Updated.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
Review comment:
Yap. I added a `get_config_shares/0` function which validates the input.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [couchdb] nickva commented on a change in pull request #3364: Fair Share Replication Scheduler Implementation (3.x)
Posted by GitBox <gi...@apache.org>.
nickva commented on a change in pull request #3364:
URL: https://github.com/apache/couchdb/pull/3364#discussion_r581444594
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
+
+
+key(#job{} = Job) ->
+ Rep = Job#job.rep,
+ case is_binary(Rep#rep.db_name) of
+ true -> mem3:dbname(Rep#rep.db_name);
+ false -> (Rep#rep.user_ctx)#user_ctx.name
+ end.
+
+
+% Jobs are charged based on the amount of time the job was running during the
+% last scheduling interval. The time units used are microseconds in order to
+% have a large enough usage values so that when priority is calculated the
+% rounded value won't be rounded off to 0 easily. The formula for the priority
+% calculation is:
+%
+% Priority = (Usage * NumJobs) / Shares^2
+%
+% Then in the worst case of a single job in the db, running only for one
+% second,for one job, with 1000 (max) shares, the priority would be:
+%
+% 1000000 * 1 / (1000^2) = 1
+%
+job_charges(#job{} = Job, IntervalMSec, {_, _, _} = Now) ->
+ TimeRunning = timer:now_diff(Now, last_started(Job)),
+ IntervalUSec = IntervalMSec * 1000,
Review comment:
The "charges" here is an abstract quantity as defined internally to this module. I added a [note](4f831d7decc#diff-db809547f022d6df8a8a55d281ffd03db38734923b242369aac80f9a48c9814eR221) why we are charging by the microsecond in order to have a large enough range and keep priorities integers and not floats. But in principle this could be any other measure, like say minutes or number of requests made (or time spent without making requests if we wanted to punish "lazy" jobs). Since we are mostly plugging this into the existing replicator, I thought of passing in the `Interval` as is without convert it again to microseconds or seconds.
##########
File path: src/couch_replicator/src/couch_replicator_share.erl
##########
@@ -0,0 +1,271 @@
+% 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.
+
+% Algorithm implemented here is based on the "A Fair Share Scheduler" by Judy
+% Kay and Piers Lauder [1].
+%
+% [1] : https://proteusmaster.urcf.drexel.edu/urcfwiki/images/KayLauderFairShare.pdf
+%
+
+-module(couch_replicator_share).
+
+-export([
+ init/0,
+ clear/0,
+
+ update_shares/2,
+ reset_shares/1,
+
+ job_added/1,
+ job_removed/1,
+
+ priority/1,
+ usage/1,
+ num_jobs/1,
+ shares/1,
+
+ charge/3,
+
+ decay_priorities/0,
+ update_priority/1,
+ update_usage/0
+]).
+
+
+-include_lib("couch/include/couch_db.hrl").
+-include("couch_replicator.hrl").
+
+
+% Usage coefficient decays historic usage every scheduling cycle. For example,
+% the usage value for a job running 1 minute is 60000000 (i.e microseconds /
+% minute). If the job stops running it will take about 26 cycles (minutes) for
+% it to decay to 0 and the system to "forget" about it completely:
+%
+% trunc(60000000 * math:pow(0.5, 26)) = 0
+%
+-define(DEFAULT_USAGE_COEFF, 0.5).
+
+% Priority coefficient decays all the job priorities such that they slowly
+% drift towards the front of the run queue. The priority value for a single job
+% which ran one for 1 minute scheduler cycle and has the default number of 100
+% shares is 60000000 / (100 * 100) = 6000. If coefficient is 0.98 it wil take
+% about 430 cycles i.e. about 7 hours for the job to drift towards the front of
+% the queue:
+%
+% trunc(6000 * math:pow(0.98, 431)) = 0
+% 430 / 60 = 7.2 hrs
+%
+-define(DEFAULT_PRIORITY_COEFF, 0.98).
+
+
+-define(MIN_SHARES, 1).
+-define(MAX_SHARES, 1000).
+-define(DEFAULT_SHARES, 100).
+
+-define(SHARES, couch_replicator_shares).
+-define(PRIORITIES, couch_replicator_priorities).
+-define(USAGE, couch_replicator_usage).
+-define(CHARGES, couch_replicator_stopped_usage).
+-define(NUM_JOBS, couch_replicator_num_jobs).
+
+
+init() ->
+ EtsOpts = [named_table, public],
+ ?SHARES = ets:new(?SHARES, EtsOpts), % {Key, Shares}
+ ?PRIORITIES = ets:new(?PRIORITIES, EtsOpts), % {JobId, Priority}
+ ?USAGE = ets:new(?USAGE, EtsOpts), % {Key, Usage}
+ ?CHARGES = ets:new(?CHARGES, EtsOpts), % {Key, Charges}
+ ?NUM_JOBS = ets:new(?NUM_JOBS, EtsOpts), % {Key, NumJobs}
+ lists:foreach(fun({K, V}) ->
+ update_shares(list_to_binary(K), list_to_integer(V))
+ end, config:get("replicator.shares")).
+
+
+clear() ->
+ Tables = [?SHARES, ?PRIORITIES, ?USAGE, ?CHARGES, ?NUM_JOBS],
+ lists:foreach(fun(T) -> catch ets:delete(T) end, Tables).
+
+
+% This should be called when user updates the replicator.shares config section
+%
+update_shares(Key, Shares) when is_integer(Shares) ->
+ ets:insert(?SHARES, {Key, min(?MAX_SHARES, max(?MIN_SHARES, Shares))}).
+
+
+% Called when the config value is deleted and shares are reset to the default
+% value.
+reset_shares(Key) ->
+ ets:delete(?SHARES, Key).
+
+
+job_added(#job{} = Job) ->
+ Key = key(Job),
+ ets:update_counter(?NUM_JOBS, Key, 1, {Key, 0}),
+ % Update job's priority as if it ran during one scheduler cycle. This is so
+ % new jobs don't get to be at priority 0 (highest).
+ update_priority(Job).
+
+
+job_removed(#job{} = Job) ->
+ Key = key(Job),
+ ets:delete(?PRIORITIES, Job#job.id),
+ case ets:update_counter(?NUM_JOBS, Key, -1, {Key, 0}) of
+ N when is_integer(N), N =< 0 ->
+ ets:delete(?NUM_JOBS, Key);
+ N when is_integer(N), N > 0 ->
+ ok
+ end,
+ ok.
+
+
+priority(JobId) ->
+ % Not found means it was removed because it's value was 0
+ case ets:lookup(?PRIORITIES, JobId) of
+ [{_, Priority}] -> Priority;
+ [] -> 0
+ end.
+
+
+usage(Key) ->
+ case ets:lookup(?USAGE, Key) of
+ [{_, Usage}] -> Usage;
+ [] -> 0
+ end.
+
+
+num_jobs(Key) ->
+ case ets:lookup(?NUM_JOBS, Key) of
+ [{_, NumJobs}] -> NumJobs;
+ [] -> 0
+ end.
+
+
+shares(Key) ->
+ case ets:lookup(?SHARES, Key) of
+ [{_, Shares}] -> Shares;
+ [] -> ?DEFAULT_SHARES
+ end.
+
+
+charge(#job{pid = undefined}, _, _) ->
+ 0;
+
+charge(#job{} = Job, Interval, {_, _, _} = Now) when is_integer(Interval) ->
+ Key = key(Job),
+ Charges = job_charges(Job, Interval, Now),
+ ets:update_counter(?CHARGES, Key, Charges, {Key, 0}).
+
+
+% In [1] this described in the "Decay of Process Priorities" section
+%
+decay_priorities() ->
+ decay(?PRIORITIES, priority_coeff()),
+ % If priority becomes 0, it's removed. When looking it up, if it
+ % is missing we assume it is 0
+ clear_zero(?PRIORITIES).
+
+
+% This is the main part of the alrgorithm. In [1] it is described in the
+% "Priority Adjustment" section.
+%
+update_priority(#job{} = Job) ->
+ Id = Job#job.id,
+ Key = key(Job),
+ Shares = shares(Key),
+ Priority = (usage(Key) * num_jobs(Key)) / (Shares * Shares),
+ ets:update_counter(?PRIORITIES, Id, trunc(Priority), {Id, 0}).
+
+
+% This is the "User-Level Scheduling" part from [1]
+%
+update_usage() ->
+ decay(?USAGE, usage_coeff()),
+ clear_zero(?USAGE),
+ ets:foldl(fun({Key, Charges}, _) ->
+ ets:update_counter(?USAGE, Key, Charges, {Key, 0})
+ end, 0, ?CHARGES),
+ % Start each interval with a fresh charges table
+ ets:delete_all_objects(?CHARGES).
+
+
+% Private helper functions
+
+decay(Ets, Coeff) when is_atom(Ets) ->
+ Head = {'$1', '$2'},
+ Result = {{'$1', {trunc, {'*', '$2', {const, Coeff}}}}},
+ ets:select_replace(Ets, [{Head, [], [Result]}]).
+
+
+clear_zero(Ets) when is_atom(Ets) ->
+ ets:select_delete(Ets, [{{'_', 0}, [], [true]}]).
+
+
+key(#job{} = Job) ->
+ Rep = Job#job.rep,
+ case is_binary(Rep#rep.db_name) of
+ true -> mem3:dbname(Rep#rep.db_name);
+ false -> (Rep#rep.user_ctx)#user_ctx.name
+ end.
+
+
+% Jobs are charged based on the amount of time the job was running during the
+% last scheduling interval. The time units used are microseconds in order to
+% have a large enough usage values so that when priority is calculated the
+% rounded value won't be rounded off to 0 easily. The formula for the priority
+% calculation is:
+%
+% Priority = (Usage * NumJobs) / Shares^2
+%
+% Then in the worst case of a single job in the db, running only for one
+% second,for one job, with 1000 (max) shares, the priority would be:
+%
+% 1000000 * 1 / (1000^2) = 1
+%
+job_charges(#job{} = Job, IntervalMSec, {_, _, _} = Now) ->
+ TimeRunning = timer:now_diff(Now, last_started(Job)),
+ IntervalUSec = IntervalMSec * 1000,
+ min(IntervalUSec, max(0, TimeRunning)).
+
+
+last_started(#job{} = Job) ->
+ case lists:keyfind(started, 1, Job#job.history) of
+ false -> {0, 0, 0}; % In case user set too low of a max history
+ {started, When} -> When
+ end.
+
+
+% Config helper functions
+
+priority_coeff() ->
+ % This is the K2 coefficient from [1]
+ Default = ?DEFAULT_PRIORITY_COEFF,
+ Val = float_val(config:get("replicator", "priority_coeff"), Default),
+ max(0.0, min(1.0, Val)).
Review comment:
Good call
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org