You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2019/05/20 18:00:29 UTC
[couchdb] branch prototype/rfc-couch-jobs updated (f8c2411 ->
7d54227)
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a change to branch prototype/rfc-couch-jobs
in repository https://gitbox.apache.org/repos/asf/couchdb.git.
discard f8c2411 CouchDB background jobs WIP
new 7d54227 CouchDB background jobs WIP2
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (f8c2411)
\
N -- N -- N refs/heads/prototype/rfc-couch-jobs (7d54227)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/couch_jobs/.gitignore | 2 +-
src/couch_jobs/src/couch_jobs.app.src | 4 +-
src/couch_jobs/src/couch_jobs.hrl | 36 ++++
src/couch_jobs/src/couch_jobs_fdb.erl | 268 +++++++++++++++---------------
src/couch_jobs/src/couch_jobs_pending.erl | 150 +++++++++++++++++
5 files changed, 325 insertions(+), 135 deletions(-)
create mode 100644 src/couch_jobs/src/couch_jobs.hrl
create mode 100644 src/couch_jobs/src/couch_jobs_pending.erl
[couchdb] 01/01: CouchDB background jobs WIP2
Posted by va...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch prototype/rfc-couch-jobs
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 7d542276afb85080df2e7312714aec6a458b0352
Author: Nick Vatamaniuc <va...@apache.org>
AuthorDate: Thu May 16 19:11:50 2019 -0400
CouchDB background jobs WIP2
So far got:
- Main API module (couch_jobs.er)
- Supervisor and app structures
- FDB read/write code in couch_jobs_fdb.erl
In couch_jobs_fdb.erl have:
- All jobs creation API: add(), remove(), submit(), get_job()
- Workers API: finish(), resubmit(), update() and accept()
- Directory path caching
- Metadata changes check
Still need:
- Activity monitor implemenation
- An example worker of some sort (Maybe activity monitor can be worker as well...)
---
rebar.config.script | 1 +
rel/reltool.config | 2 +
src/couch_jobs/.gitignore | 4 +
src/couch_jobs/README.md | 5 +
src/couch_jobs/src/couch_jobs.app.src | 28 +++
src/couch_jobs/src/couch_jobs.erl | 96 ++++++++
src/couch_jobs/src/couch_jobs.hrl | 36 +++
src/couch_jobs/src/couch_jobs_app.erl | 26 +++
src/couch_jobs/src/couch_jobs_fdb.erl | 364 ++++++++++++++++++++++++++++++
src/couch_jobs/src/couch_jobs_pending.erl | 150 ++++++++++++
src/couch_jobs/src/couch_jobs_sup.erl | 46 ++++
11 files changed, 758 insertions(+)
diff --git a/rebar.config.script b/rebar.config.script
index 3b58bcb..2def724 100644
--- a/rebar.config.script
+++ b/rebar.config.script
@@ -76,6 +76,7 @@ SubDirs = [
"src/couch_tests",
"src/ddoc_cache",
"src/fabric",
+ "src/couch_jobs",
"src/global_changes",
"src/mango",
"src/rexi",
diff --git a/rel/reltool.config b/rel/reltool.config
index 1051d2e..afebc44 100644
--- a/rel/reltool.config
+++ b/rel/reltool.config
@@ -34,6 +34,7 @@
couch,
couch_epi,
couch_index,
+ couch_jobs,
couch_log,
couch_mrview,
couch_plugins,
@@ -90,6 +91,7 @@
{app, config, [{incl_cond, include}]},
{app, couch, [{incl_cond, include}]},
{app, couch_epi, [{incl_cond, include}]},
+ {app, couch_jobs, [{incl_cond, include}]},
{app, couch_index, [{incl_cond, include}]},
{app, couch_log, [{incl_cond, include}]},
{app, couch_mrview, [{incl_cond, include}]},
diff --git a/src/couch_jobs/.gitignore b/src/couch_jobs/.gitignore
new file mode 100644
index 0000000..6ef4c52
--- /dev/null
+++ b/src/couch_jobs/.gitignore
@@ -0,0 +1,4 @@
+*.beam
+.eunit
+ebin/couch_jobs.app
+.DS_Store
\ No newline at end of file
diff --git a/src/couch_jobs/README.md b/src/couch_jobs/README.md
new file mode 100644
index 0000000..b2910a5
--- /dev/null
+++ b/src/couch_jobs/README.md
@@ -0,0 +1,5 @@
+CouchDB Jobs Application
+=========================
+
+Run background jobs in CouchDB
+
diff --git a/src/couch_jobs/src/couch_jobs.app.src b/src/couch_jobs/src/couch_jobs.app.src
new file mode 100644
index 0000000..c9b8b2c
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs.app.src
@@ -0,0 +1,28 @@
+% 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.
+
+{application, couch_jobs, [
+ {description, "CouchDB Jobs"},
+ {vsn, git},
+ {mod, {couch_jobs_app, []}},
+ {registered, [
+ couch_jobs_sup
+ ]},
+ {applications, [
+ kernel,
+ stdlib,
+ erlfdb,
+ couch_log,
+ config,
+ fabric
+ ]}
+]}.
diff --git a/src/couch_jobs/src/couch_jobs.erl b/src/couch_jobs/src/couch_jobs.erl
new file mode 100644
index 0000000..6166706
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs.erl
@@ -0,0 +1,96 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_jobs).
+
+-export([
+ add/3,
+ remove/2,
+ resubmit/2,
+ get_job/2,
+
+ accept/1,
+ accept/2,
+ finish/5,
+ resubmit/4,
+ update/5
+]).
+
+
+%% Job Creation API
+
+add(Type, JobId, JobOpts) ->
+ try
+ ok = validate_jobopts(JobOpts)
+ catch
+ Tag:Err -> {error, {invalid_job_args, Tag, Err}}
+ end,
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_job_fdb:add(JTx, Type, JobId, JobOpts)
+ end).
+
+
+remove(Type, JobId) ->
+ % Add the bit about cancelling the job if it is running
+ % and waiting for it to stop, then remove it.
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:remove(JTx, Type, JobId)
+ end).
+
+
+resubmit(Type, JobId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:resubmit(JTx, Type, JobId)
+ end).
+
+
+get_job(Type, JobId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:get_job(JTx, Type, JobId)
+ end).
+
+
+%% Worker Implementation API
+
+accept(Type) ->
+ accept(Type, undefined).
+
+
+accept(Type, MaxPriority) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(), fun(JTx) ->
+ couch_jobs_fdb:accept(JTx, Type, MaxPriority)
+ end).
+
+
+finish(Tx, Type, JobId, JobOpts, WorkerLockId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+ couch_jobs_fdb:finish(JTx, Type, JobId, JobOpts, WorkerLockId)
+ end).
+
+
+resubmit(Tx, Type, JobId, WorkerLockId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+ couch_jobs_fdb:resubmit(JTx, Type, JobId, WorkerLockId)
+ end).
+
+
+update(Tx, Type, JobId, JobOpts, WorkerLockId) ->
+ couch_jobs_fdb:tx(couch_jobs_fdb:get_jtx(Tx), fun(JTx) ->
+ couch_jobs_fdb:update(JTx, Type, JobId, JobOpts, WorkerLockId)
+ end).
+
+
+%% Private utils
+
+validate_jobopts(#{} = JobOpts) ->
+ jiffy:encode(JobOpts),
+ ok.
diff --git a/src/couch_jobs/src/couch_jobs.hrl b/src/couch_jobs/src/couch_jobs.hrl
new file mode 100644
index 0000000..43cf98c
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs.hrl
@@ -0,0 +1,36 @@
+% 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.
+
+
+% JobOpts field definitions
+%
+-define(OPT_PRIORITY, <<"priority">>).
+-define(OPT_DATA, <<"data">>).
+-define(OPT_CANCEL, <<"cancel">>).
+-define(OPT_RESUBMIT, <<"resubmit">>).
+
+% These might be in a fabric public .hrl eventually
+%
+-define(uint2bin(I), binary:encode_unsigned(I, little)).
+-define(bin2uint(I), binary:decode_unsigned(I, little)).
+-define(UNSET_VS, {versionstamp, 16#FFFFFFFFFFFFFFFF, 16#FFFF}).
+-define(METADATA_VERSION_KEY, <<"$metadata_version_key$">>).
+
+% Data model definitions
+% Switch these to numbers eventually.
+%
+-define(JOBS, <<"couch_jobs">>).
+-define(DATA, <<"data">>).
+-define(PENDING, <<"pending">>).
+-define(WATCHES, <<"watches">>).
+-define(ACTIVITY_TIMEOUT, <<"activity_timeout">>).
+-define(ACTIVITY, <<"activity">>).
diff --git a/src/couch_jobs/src/couch_jobs_app.erl b/src/couch_jobs/src/couch_jobs_app.erl
new file mode 100644
index 0000000..720b948
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_app.erl
@@ -0,0 +1,26 @@
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_jobs_app).
+
+
+-behaviour(application).
+
+
+-export([
+ start/2,
+ stop/1
+]).
+
+
+start(_Type, []) ->
+ couch_jobs_sup:start_link().
+
+
+stop([]) ->
+ ok.
diff --git a/src/couch_jobs/src/couch_jobs_fdb.erl b/src/couch_jobs/src/couch_jobs_fdb.erl
new file mode 100644
index 0000000..573144d
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_fdb.erl
@@ -0,0 +1,364 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_jobs_fdb).
+
+
+-export([
+ add/4,
+ remove/3,
+ resubmit/3,
+ get_job/3,
+
+ accept/2,
+ accept/3,
+ finish/5,
+ resubmit/4,
+ update/5,
+
+ init_cache/0,
+
+ get_jtx/0,
+ get_jtx/1,
+
+ tx/2
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+% Data model
+%
+% (?JOBS, ?DATA, Type, JobId) = (Sequence, WorkerLockId, Priority, JobOpts)
+% (?JOBS, ?PENDING, Type, Priority, JobId) = ""
+% (?JOBS, ?WATCHES, Type) = Sequence
+% (?JOBS, ?ACTIVITY_TIMEOUT, Type) = ActivityTimeout
+% (?JOBS, ?ACTIVITY, Type, Sequence) = JobId
+
+
+add(#{jtx := true} = JTx0, Type, JobId, JobOpts) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case erlfdb:wait(erlfdb:get(Tx, Key)) of
+ <<_/binary>> ->
+ {error, duplicate_job};
+ not_found ->
+ maybe_submit(JTx, Type, JobId, JobOpts)
+ end.
+
+
+remove(#{jtx := true} = JTx0, Type, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {_, WorkerLockId, _, _} when WorkerLockId =/= null ->
+ {error, job_is_running};
+ {_, _, null, _, _} ->
+ erlfdb:clear(Tx, Key);
+ {_, _, Priority, _, _} ->
+ couch_jobs_pending:remove(JTx, Type, Priority, JobId),
+ erlfdb:clear(Tx, Key);
+ not_found ->
+ not_found
+ end.
+
+
+resubmit(#{jtx := true} = JTx, Type, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {_, _, _, #{?OPT_RESUBMIT := true}} ->
+ ok;
+ {Seq, WorkerLockId, Priority, #{} = JobOpts} ->
+ set_resubmit(Tx, Key, Seq, WorkerLockId, Priority, JobOpts);
+ not_found ->
+ not_found
+ end.
+
+
+get_job(#{jtx := true} = JTx, Type, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {_, WorkerLockId, Priority, JobOpts} ->
+ {ok, JobOpts, job_state(WorkerLockId, Priority)};
+ not_found ->
+ not_found
+ end.
+
+
+% Worker public API
+
+accept(#{jtx := true} = JTx, Type) ->
+ accept(JTx, Type, undefined).
+
+
+accept(#{jtx := true} = JTx0, Type, MaxPriority) ->
+ #{jtx := true} = JTx = get_jtx(JTx0),
+ case couch_jobs_pending:dequeue(JTx, Type, MaxPriority) of
+ not_found ->
+ not_found;
+ <<_/binary>> = JobId ->
+ WorkerLockId = fabric2_util:uuid(),
+ update_lock(JTx, Type, JobId, WorkerLockId),
+ update_activity(JTx, Type, JobId),
+ update_watch(JTx, Type),
+ {ok, JobId, WorkerLockId}
+ end.
+
+
+finish(#{jtx := true} = JTx0, Type, JobId, JobOpts, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx = get_jtx(JTx0),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ try
+ {Seq, _, _, JobOptsCur} = get_job_or_raise_status(Tx, Key, WorkerLockId),
+ clear_activity(JTx, Type, Seq),
+ maybe_submit(JTx, Type, JobId, maps:merge(JobOptsCur, JobOpts))
+ catch
+ throw:{job_status, JobStatus} -> JobStatus
+ end.
+
+
+resubmit(#{jtx := true} = JTx, Type, JobId, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ try
+ case get_job_or_raise_status(Tx, Key, WorkerLockId) of
+ {_, _, _, #{?OPT_RESUBMIT := true}} ->
+ ok;
+ {Seq, WorkerLockId, Priority, #{} = JobOpts} ->
+ set_resubmit(Tx, Key, Seq, WorkerLockId, Priority, JobOpts)
+ end
+ catch
+ throw:{job_status, JobStatus} -> JobStatus
+ end.
+
+
+update(#{jtx := true} = JTx, Type, JobId, JobOpts, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = get_jtx(JTx),
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ try
+ case get_job_or_raise_status(Tx, Key, WorkerLockId) of
+ {null, WorkerLockId, null, JobOptsCur} ->
+ update_activity(JTx, Type, JobId),
+ JobOpts1 = maps:merge(JobOptsCur, JobOpts),
+ JobOptsEnc = jiffy:encode(JobOpts1),
+ Val = erlfdb_tuple:pack({null, WorkerLockId, null, JobOptsEnc}),
+ erlfdb:set(Tx, Key, Val);
+ InvalidState ->
+ error({couch_job_invalid_updata_state, InvalidState})
+ end
+ catch
+ throw:{job_status, JobStatus} -> JobStatus
+ end,
+ ok.
+
+
+% Cache initialization API. Called from the supervisor just to create the ETS
+% table. It returns `ignore` to tell supervisor it won't actually start any
+% process, which is what we want here.
+%
+init_cache() ->
+ ConcurrencyOpts = [{read_concurrency, true}, {write_concurrency, true}],
+ ets:new(?MODULE, [public, named_table] ++ ConcurrencyOpts),
+ ignore.
+
+
+% Cached job transaction object. This object wraps a transaction, caches the
+% directory lookup path, and the metadata version. The function can be used from
+% or outside the transaction. When used from a transaction it will verify if
+% the metadata was changed, and will refresh automatically.
+%
+get_jtx() ->
+ get_jtx(undefined).
+
+
+get_jtx(#{tx := Tx} = _TxDb) ->
+ get_jtx(Tx);
+
+get_jtx(undefined = _Tx) ->
+ case ets:lookup(?MODULE, ?JOBS) of
+ [{_, #{} = JTx}] -> JTx;
+ [] -> update_jtx_cache(init_jtx(undefined))
+ end;
+
+get_jtx({erlfdb_transaction, _} = Tx) ->
+ case ets:lookup(?MODULE, ?JOBS) of
+ [{_, #{} = JTx}] -> ensure_current(JTx#{tx := Tx});
+ [] -> update_jtx_cache(init_jtx(Tx))
+ end.
+
+
+% Transaction processing to be used with couch jobs' specific transaction
+% contexts
+%
+tx(#{jtx := true} = Tx, Fun) when is_function(Fun, 1) ->
+ fabric2_fdb:transactional(Tx, Fun).
+
+
+% Utility fdb functions used by other module in couch_job. Maybe move these to a
+% separate module if the list keep growing
+
+has_versionstamp(?UNSET_VS) ->
+ true;
+
+has_versionstamp(Tuple) when is_tuple(Tuple) ->
+ has_versionstamp(tuple_to_list(Tuple));
+
+has_versionstamp([Elem | Rest]) ->
+ has_versionstamp(Elem) orelse has_versionstamp(Rest);
+
+has_versionstamp(_Other) ->
+ false.
+
+
+
+% Private API helper functions
+
+
+maybe_submit(#{jtx := true} = JTx, Type, JobId, JobOpts) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ Resubmit = maps:get(?OPT_RESUBMIT, JobOpts, false) =/= true,
+ JobOpts1 = maps:without([?OPT_RESUBMIT], JobOpts),
+ Priority = get_jobopts_priority(JobOpts1),
+ JobOptsEnc = jiffy:encode(JobOpts1),
+ case Resubmit of
+ true ->
+ Val = erlfdb_tuple:pack_vs({null, null, Priority, JobOptsEnc}),
+ case couch_jobs_util:has_versionstamp(Priority) of
+ true -> erlfdb:set_versionstamped_value(Tx, Key, Val);
+ false -> erlfdb:set(Tx, Key, Val)
+ end,
+ couch_jobs_pending:enqueue(JTx, Type, Priority, JobId);
+ false ->
+ Val = erlfdb_tuple:pack({null, null, null, JobOptsEnc}),
+ erlfdb:set(Tx, Key, Val)
+ end,
+ ok.
+
+
+get_job(Tx = {erlfdb_transaction, _}, Key) ->
+ case erlfdb:wait(erlfdb:get(Tx, Key)) of
+ <<_/binary>> = Val ->
+ {Seq, WorkerLockId, JobOptsBin} = erlfdb_tuple:unpack(Val),
+ JobOpts = jiffy:decode(JobOptsBin, [return_maps]),
+ {Seq, WorkerLockId, JobOpts};
+ not_found ->
+ not_found
+ end.
+
+
+get_jobopts_priority(JobOpts) ->
+ case maps:get(?OPT_PRIORITY, JobOpts, undefined) of
+ undefined -> ?UNSET_VS;
+ Priority -> Priority
+ end.
+
+
+get_job_or_raise_status(Tx, Key, WorkerLockId) ->
+ case get_job(Tx, Key) of
+ {_, CurWorkerLockId, _, _} when WorkerLockId =/= CurWorkerLockId ->
+ throw({job_status, worker_conflict});
+ {_, _, _, #{?OPT_CANCEL := true}} ->
+ throw({job_status, canceled});
+ {_, _, _, #{}} = Res ->
+ Res
+ end.
+
+
+set_resubmit(Tx, Key, Seq, WorkerLockId, Priority, JobOpts0) ->
+ JobOpts = JobOpts0#{?OPT_RESUBMIT => true},
+ Val = erlfdb_tuple:pack({Seq, WorkerLockId, Priority, jiffy:encode(JobOpts)}),
+ erlfdb:set(Tx, Key, Val).
+
+
+update_activity(#{jtx := true} = JTx, Type, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack_vs({?ACTIVITY, Type, ?UNSET_VS}, Jobs),
+ erlfdb:set_versionstamped_key(Tx, Key, JobId).
+
+
+clear_activity(#{jtx := true} = JTx, Type, Seq) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?ACTIVITY, Type, Seq}, Jobs),
+ erlfdb:clear(Tx, Key).
+
+
+update_watch(#{jtx := true} = JTx, Type) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?WATCHES, Type}, Jobs),
+ erlfdb:set_versionstamped_value(Tx, Key, ?UNSET_VS).
+
+
+update_lock(#{jtx := true} = JTx, Type, JobId, WorkerLockId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?DATA, Type, JobId}, Jobs),
+ case get_job(Tx, Key) of
+ {null, null, _Priority, JobOpts} ->
+ ValTup = {?UNSET_VS, WorkerLockId, null, jiffy:encode(JobOpts)},
+ Val = erlfdb_tuple:pack_vs(ValTup),
+ erlfdb:set_versionstamped_value(Tx, Key, Val);
+ InvalidState ->
+ error({couch_job_invalid_accept_state, InvalidState})
+ end.
+
+
+job_state(WorkerLockId, Priority) ->
+ case {WorkerLockId, Priority} of
+ {null, null} ->
+ finished;
+ {WorkerLockId, _} when WorkerLockId =/= null ->
+ running;
+ {_, Priority} when Priority =/= null ->
+ pending;
+ ErrorState ->
+ error({invalid_job_state, ErrorState})
+ end.
+
+
+% This a transaction context object similar to the Db = #{} one from fabric2_fdb.
+% It's is used to cache the jobs path directory (to avoid extra lookups on every
+% operation) and to check for metadata changes (in case directory changes).
+%
+init_jtx(undefined) ->
+ fabric2_fdb:transactional(fun(Tx) -> init_jtx(Tx) end);
+
+init_jtx({erlfdb_transaction, _} = Tx) ->
+ Root = erlfdb_directory:root(),
+ CouchDB = erlfdb_directory:create_or_open(Tx, Root, [<<"couchdb">>]),
+ LayerPrefix = erlfdb_directory:get_name(CouchDB),
+ JobsPrefix = erlfdb_tuple:pack({?JOBS}, LayerPrefix),
+ Version = erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)),
+ % layer_prefix, md_version and tx here match db map fields in fabric2_fdb
+ % but we also assert that this is a job transaction using the jtx => true field
+ #{
+ jtx => true,
+ tx => Tx,
+ layer_prefix => LayerPrefix,
+ jobs_prefix => JobsPrefix,
+ md_version => Version
+ }.
+
+
+ensure_current(#{jtx := true, tx := Tx, md_version := Version} = JTx) ->
+ case erlfdb:wait(erlfdb:get(Tx, ?METADATA_VERSION_KEY)) of
+ Version -> JTx;
+ _NewVersion -> update_jtx_cache(init_jtx(Tx))
+ end.
+
+
+update_jtx_cache(#{jtx := true} = JTx) ->
+ CachedJTx = JTx#{tx := undefined},
+ ets:insert(?MODULE, {?JOBS, CachedJTx}),
+ JTx.
diff --git a/src/couch_jobs/src/couch_jobs_pending.erl b/src/couch_jobs/src/couch_jobs_pending.erl
new file mode 100644
index 0000000..d015c04
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_pending.erl
@@ -0,0 +1,150 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_jobs_pending).
+
+
+-export([
+ enqueue/4,
+ dequeue/3,
+ remove/4
+]).
+
+
+-include("couch_jobs.hrl").
+
+
+% Make this configurable or auto-adjustable based on retries
+%
+-define(RANGE_LIMIT, 256).
+
+
+% Data model
+%
+% (?JOBS, ?PENDING, Type, Priority, JobId) = ""
+
+
+% Public API
+
+% Enqueue a job into the pending queue. Priority determines the place in the
+% queue.
+%
+enqueue(#{jtx := true} = JTx, Type, Priority, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ case couch_jobs_fdb:has_versionstamp(Priority) of
+ true ->
+ Key = erlfdb_tuple:pack_vs({?PENDING, Type, Priority, JobId}, Jobs),
+ erlfdb:set_versionstamped_key(Tx, Key, null);
+ false ->
+ Key = erlfdb_tuple:pack({?PENDING, Type, Priority, JobId}, Jobs),
+ erlfdb:set(Tx, Key, null)
+ end.
+
+
+% Dequeue a job from the front of the queue.
+%
+% If MaxPriority is specified, any job between the front of the queue and
+% up until MaxPriority is considered. Workers may randomly pick jobs in that
+% range. That can be used to avoid contention at the expense of strict dequeue
+% ordering. For instance if priorities are 0-high, 1-normal and 2-low, and we wish
+% to process normal and urgent jobs, then MaxPriority=1-normal would
+% accomplish that.
+%
+dequeue(#{jtx := true} = JTx, Type, undefined) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Prefix = erlfdb_tuple:pack({?PENDING, Type}, Jobs),
+ case get_front_priority(Tx, Prefix) of
+ not_found ->
+ not_found;
+ Priority ->
+ {Start, End} = erlang_tuple:range(Priority, Prefix),
+ case clear_random_key_from_range(Tx, Start, End) of
+ not_found ->
+ not_found;
+ <<_/binary>> = PendingKey ->
+ {_, JobId} = erlfdb_tuple:unpack(PendingKey, Prefix),
+ JobId
+ end
+ end;
+
+% If MaxPriority is not specified, only jobs with the same priority as the item
+% at the front of the queue are considered. Two extremes are useful to
+% consider:
+%
+% * Priority is just one static value (say null, or "normal"). In that case,
+% the queue is effectively a bag of tasks that can be grabbed in any order,
+% which should minimize contention.
+%
+% * Each job has a unique priority value, for example a versionstamp. In that
+% case, the queue has strict FIFO behavior, but there will be more contention
+% at the front of the queue.
+%
+dequeue(#{jtx := true} = JTx, Type, MaxPriority) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Prefix = erlfdb_tuple:pack({?PENDING, Type}, Jobs),
+ StartKeySel = erlfdb_key:first_greater_than(Prefix),
+ End = erlfdb_tuple:pack({?PENDING, Type, MaxPriority, <<16#FF>>}, Prefix),
+ EndKeySel = erlfdb_key:last_less_than(End),
+ case clear_random_key_from_range(Tx, StartKeySel, EndKeySel) of
+ not_found ->
+ not_found;
+ <<_/binary>> = PendingKey ->
+ {_, JobId} = erlfdb_tuple:unpack(PendingKey, Prefix),
+ JobId
+ end.
+
+
+% Remove a job from the pending queue. This is used, for example, when a job is
+% canceled while it was waiting in the pending queue.
+%
+remove(#{jtx := true} = JTx, Type, Priority, JobId) ->
+ #{tx := Tx, jobs_path := Jobs} = JTx,
+ Key = erlfdb_tuple:pack({?PENDING, Type, Priority, JobId}, Jobs),
+ erlfdb:clear(Tx, Key).
+
+
+% Private functions
+
+% The priority of the item at the front. If there are multiple
+% items with the same priority, workers can randomly pick between them to
+% avoid contention.
+%
+get_front_priority(Tx, Prefix) ->
+ Opts = [{limit, 1}, {snapshot, true}],
+ case erlfdb:wait(erfldb:get_range_startswith(Tx, Prefix, Opts)) of
+ [] ->
+ not_found;
+ [{FrontKey, _}] ->
+ {Priority, _} = erlfdb_tuple:unpack(FrontKey, Prefix),
+ Priority
+ end.
+
+
+% Pick a random key from the range snapshot. Then radomly pick a key to
+% clear. Before clearing, ensure there is a read conflict on the key in
+% in case other workers have picked the same key.
+%
+clear_random_key_from_range(Tx, Start, End) ->
+ Opts = [
+ {limit, ?RANGE_LIMIT},
+ {snapshot, true}
+ ],
+ case erlfdb:wait(erlfdb:get_range(Tx, Start, End, Opts)) of
+ [] ->
+ not_found;
+ [{_, _} | _] = KVs ->
+ Index = rand:uniform(length(KVs)),
+ {Key, _} = lists:nth(Index),
+ erlfdb:add_read_conflict_key(Tx, Key),
+ erlfdb:clear(Tx, Key),
+ Key
+ end.
diff --git a/src/couch_jobs/src/couch_jobs_sup.erl b/src/couch_jobs/src/couch_jobs_sup.erl
new file mode 100644
index 0000000..8a7832a
--- /dev/null
+++ b/src/couch_jobs/src/couch_jobs_sup.erl
@@ -0,0 +1,46 @@
+%
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_jobs_sup).
+
+
+-behaviour(supervisor).
+
+
+-export([
+ start_link/0
+]).
+
+-export([
+ init/1
+]).
+
+
+start_link() ->
+ supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+init([]) ->
+ Flags = #{
+ strategy => one_for_one,
+ intensity => 1,
+ period => 5
+ },
+ Children = [
+ #{
+ id => couch_jobs_fdb,
+ restart => transient,
+ start => {couch_jobs_fdb, init_cache, []}
+ }
+ ],
+ {ok, {Flags, Children}}.