You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2020/06/29 20:25:27 UTC
[couchdb-ioq] branch optimize-ioq-server updated (396c77c ->
a877a72)
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a change to branch optimize-ioq-server
in repository https://gitbox.apache.org/repos/asf/couchdb-ioq.git.
from 396c77c Use ioq_q in ioq_server
new 4039fe6 Implement request deduplication
new c1ba0db Add backpressure to ioq_server
new a877a72 Shed load in ioq_server
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
src/ioq_benchmark.erl | 9 +++-
src/ioq_q.erl | 39 +++++++++++-----
src/ioq_server.erl | 120 +++++++++++++++++++++++++++++++++++++++++++++-----
3 files changed, 145 insertions(+), 23 deletions(-)
[couchdb-ioq] 01/03: Implement request deduplication
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch optimize-ioq-server
in repository https://gitbox.apache.org/repos/asf/couchdb-ioq.git
commit 4039fe6d721ea6b444c9550550ca29f8d4ec72dd
Author: Paul J. Davis <da...@us.ibm.com>
AuthorDate: Mon Jun 29 13:54:36 2020 -0500
Implement request deduplication
This adds a hash to each ioq_q which is then used by ioq_server to
deduplicate read requests to a given channel.
---
src/ioq_q.erl | 39 +++++++++++++++++++++++++++++----------
src/ioq_server.erl | 22 +++++++++++++++++++++-
2 files changed, 50 insertions(+), 11 deletions(-)
diff --git a/src/ioq_q.erl b/src/ioq_q.erl
index 8b592b6..1889d81 100644
--- a/src/ioq_q.erl
+++ b/src/ioq_q.erl
@@ -20,26 +20,31 @@
in/2,
out/1,
+ lookup/2,
+ put/3,
+ del/2,
+
to_list/1
]).
new() ->
{ok, Q} = khash:new(),
+ {ok, H} = khash:new(),
ok = khash:put(Q, next_id, 0),
ok = khash:put(Q, oldest_id, undefined),
- Q.
+ {Q, H}.
-len(Q) ->
+len({Q, _}) ->
khash:size(Q) - 2. % two metadata keys
-is_empty(Q) ->
- len(Q) == 0.
+is_empty({Q, H}) ->
+ len({Q, H}) == 0.
-in(Item, Q) ->
+in(Item, {Q, H}) ->
NextId = khash:get(Q, next_id),
ok = khash:put(Q, NextId, Item),
ok = khash:put(Q, next_id, NextId + 1),
@@ -49,14 +54,14 @@ in(Item, Q) ->
_ ->
ok
end,
- Q.
+ {Q, H}.
-out(Q) ->
+out({Q, H}) ->
NextId = khash:get(Q, next_id),
case khash:get(Q, oldest_id) of
undefined ->
- {empty, Q};
+ {empty, {Q, H}};
OldestId ->
{value, Item} = khash:lookup(Q, OldestId),
khash:del(Q, OldestId),
@@ -67,11 +72,25 @@ out(Q) ->
false ->
khash:put(Q, oldest_id, OldestId + 1)
end,
- {{value, Item}, Q}
+ {{value, Item}, {Q, H}}
end.
-to_list(Q) ->
+lookup({_Q, H}, Key) ->
+ khash:lookup(H, Key).
+
+
+put({Q, H}, Key, Value) ->
+ khash:put(H, Key, Value),
+ {Q, H}.
+
+
+del({Q, H}, Key) ->
+ khash:del(H, Key),
+ {Q, H}.
+
+
+to_list({Q, _H}) ->
Entries = khash:to_list(Q),
NoMeta = [{K, V} || {K, V} <- Entries, is_integer(K)],
Sorted = lists:sort(NoMeta),
diff --git a/src/ioq_server.erl b/src/ioq_server.erl
index 1bce2d3..c66bb30 100644
--- a/src/ioq_server.erl
+++ b/src/ioq_server.erl
@@ -259,7 +259,6 @@ find_channel(Account, #state{} = State) ->
} = State,
case khash:lookup(ChannelsByName, Account) of
{value, Channel} ->
- couch_log:error("XKCD: found channel", []),
Channel;
not_found ->
Channel = #channel{name = Account},
@@ -275,9 +274,26 @@ update_channel(Ch, Req, Dedupe) ->
% everything else is interactive IO class
Ch#channel{qI = update_queue(Req, Ch#channel.qI, Dedupe)}.
+update_queue(#request{from=From, fd=Fd, msg={pread_iolist, Pos}}=R, Q, true) ->
+ Key = {dedupe, Fd, Pos},
+ Req = case ioq_q:lookup(Q, Key) of
+ {value, #request{from = PrevFrom} = PrevReq} ->
+ PrevReq#request{
+ from = append(From, PrevFrom)
+ };
+ not_found ->
+ ioq_q:in(Key, Q),
+ R
+ end,
+ ioq_q:put(Q, Key, Req);
update_queue(Req, Q, _Dedupe) ->
ioq_q:in(Req, Q).
+append(A, B) when is_list(B) ->
+ [A|B];
+append(A, B) ->
+ [A, B].
+
enqueue_channel(#request{channel=Account} = Req, #state{channels=Q} = State) ->
DD = State#state.dedupe,
% ioq_q's are update-in-place
@@ -408,6 +424,10 @@ choose_prioritized_request([Q | Rest], Empties) ->
case ioq_q:out(Q) of
{empty, _} ->
choose_prioritized_request(Rest, [Q | Empties]);
+ {{value, {dedupe, _Fd, _Pos} = Key}, NewQ} ->
+ {value, Req} = ioq_q:lookup(Q, Key),
+ ioq_q:del(Q, Key),
+ {Req, lists:reverse([NewQ | Empties], Rest)};
{{value, Item}, NewQ} ->
{Item, lists:reverse([NewQ | Empties], Rest)}
end.
[couchdb-ioq] 02/03: Add backpressure to ioq_server
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch optimize-ioq-server
in repository https://gitbox.apache.org/repos/asf/couchdb-ioq.git
commit c1ba0db845dd6e6c679558acb2bbe42ce64a26f4
Author: Paul J. Davis <da...@us.ibm.com>
AuthorDate: Mon Jun 29 14:53:38 2020 -0500
Add backpressure to ioq_server
For interactive IOQ requests, the first time a process calls ioq_server
we use a 5s timeout. If we don't have a response within that timeout we
return a signal that IO has been saturated on this node. If the first
request succeeds we remove the timeout so that we don't start failing
IO once a request has started processing.
---
src/ioq_server.erl | 25 +++++++++++++++++++++++++
1 file changed, 25 insertions(+)
diff --git a/src/ioq_server.erl b/src/ioq_server.erl
index c66bb30..1c8a2c6 100644
--- a/src/ioq_server.erl
+++ b/src/ioq_server.erl
@@ -52,6 +52,9 @@
tsub
}).
+-define(INITIAL_TIMEOUT, 5000).
+
+
start_link() ->
Opts = [{spawn_opt, [{fullsweep_after, 10}]}],
gen_server:start_link({local, ?MODULE}, ?MODULE, [], Opts).
@@ -71,6 +74,28 @@ call(Fd, Msg, Priority) ->
catch couch_stats:increment_counter([couchdb, io_queue_bypassed, Class]),
catch couch_stats:increment_counter([couchdb, io_queue_bypassed, RW]),
gen_server:call(Fd, Msg, infinity);
+ _ when Class == interactive ->
+ % For interactive requests we use a timeout on
+ % the first request to determine if ioq_server
+ % is overloaded.
+ Timeout = case get('$ioq_called') of
+ true -> infinity;
+ false -> ?INITIAL_TIMEOUT
+ end,
+ try
+ gen_server:call(?MODULE, Request, Timeout)
+ catch exit:{timeout, {gen_server, call, _Args}} ->
+ case get(rexi_from) of
+ undefined ->
+ exit({io_saturated, node()});
+ _ ->
+ % Done to silence error logging by rexi_server
+ rexi:reply({rexi_EXIT, {io_saturated, node()}}),
+ exit(normal)
+ end;
+ after
+ put('$ioq_called', true)
+ end;
_ ->
gen_server:call(?MODULE, Request, infinity)
end.
[couchdb-ioq] 03/03: Shed load in ioq_server
Posted by da...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch optimize-ioq-server
in repository https://gitbox.apache.org/repos/asf/couchdb-ioq.git
commit a877a724cc47dd269796a30bdf623814126b40af
Author: Paul J. Davis <da...@us.ibm.com>
AuthorDate: Mon Jun 29 15:24:23 2020 -0500
Shed load in ioq_server
This avoids processing any request that we find either in our message
box or in our internal queuing system that has already timedout and had
the client exit.
---
src/ioq_benchmark.erl | 9 +++++--
src/ioq_server.erl | 73 ++++++++++++++++++++++++++++++++++++++++++++-------
2 files changed, 70 insertions(+), 12 deletions(-)
diff --git a/src/ioq_benchmark.erl b/src/ioq_benchmark.erl
index 4e2ea2d..8ef63e9 100644
--- a/src/ioq_benchmark.erl
+++ b/src/ioq_benchmark.erl
@@ -77,7 +77,8 @@ run(Opts) ->
ets:new(?WORKERS, [public, set, named_table]),
create_files(Opts),
%run_static(Opts),
- run_dynamic(Opts).
+ %run_dynamic(Opts).
+ run_static(Opts).
run_static(Opts) ->
@@ -328,7 +329,11 @@ gen_priority_and_message(UserBin, ClassPerc, OpPerc) ->
view ->
case rand:uniform() < 0.6 of
true ->
- {pread_iolist, rand:uniform()};
+ Pos = case rand:uniform() of
+ V when V < 0.05 -> 0.0;
+ V -> V
+ end,
+ {pread_iolist, Pos};
false ->
{append_binary, rand:uniform()}
end
diff --git a/src/ioq_server.erl b/src/ioq_server.erl
index 1c8a2c6..4bc6eae 100644
--- a/src/ioq_server.erl
+++ b/src/ioq_server.erl
@@ -319,15 +319,21 @@ append(A, B) when is_list(B) ->
append(A, B) ->
[A, B].
-enqueue_channel(#request{channel=Account} = Req, #state{channels=Q} = State) ->
+enqueue_channel(#request{name = Account} = Req, #state{channels = Q} = State) ->
DD = State#state.dedupe,
- % ioq_q's are update-in-place
- case find_channel(Account, State) of
- {new, Channel} ->
- update_channel(Channel, Req, DD),
- ioq_q:in(Channel, Q);
- Channel ->
- update_channel(Channel, Req, DD)
+ ShouldProcess = should_process_request(Req),
+ case should_process_request(Req) of
+ true ->
+ % ioq_q's are update-in-place
+ case find_channel(Account, State) of
+ {new, Channel} ->
+ update_channel(Channel, Req, DD),
+ ioq_q:in(Channel, Q);
+ Channel ->
+ update_channel(Channel, Req, DD)
+ end;
+ false ->
+ ok
end,
State.
@@ -391,8 +397,26 @@ make_next_request(State) ->
false ->
NewCh2 = ioq_q:in(Channel#channel{qI=QI2, qU=QU2, qV=QV2}, NewCh)
end,
- submit_request(Item2, State#state{channels=NewCh2, qC=NewQC,
- qR=NewQR, qL=NewQL});
+ NewState = State#state{
+ channels = NewCh2,
+ qC = NewQC,
+ qR = NewQR,
+ qL = NewQL
+ },
+ ShouldProcess = should_process_request(Item2),
+ case ShouldProcess of
+ true ->
+ submit_request(Item2, NewState);
+ false ->
+ % We're recursing here because of the logic
+ % in maybe_submit_request where returning
+ % NewState may actually end up being equal
+ % to the old state due to ioq_q instances
+ % now being references. This would be a bug
+ % that prevented from using all of the
+ % available concurrent request slots.
+ make_next_request(NewState)
+ end;
_ ->
% Item is a background (compaction or internal replication) task
submit_request(Item, State#state{channels=NewCh, qC=NewQC, qR=NewQR,
@@ -423,6 +447,35 @@ submit_request(Request, State) ->
update_counter(Counters, Channel, IOClass, RW),
State#state{reqs = [Request#request{tsub=SubmitTime, ref=Ref} | Reqs]}.
+
+should_process_request(#request{class = interactive} = Request) ->
+ #request{
+ from = Froms,
+ t0 = T0
+ } = Req,
+
+ % Calculate our timeout
+ RawDiff = erlang:monotonic_time() - T0,
+ MilliDiff = erlang:convert_time_unit(Diff, native, millisecond),
+ TimedOut = MilliDiff > ?INITIAL_TIMEOUT,
+
+ if not TimedOut -> true; true ->
+ % Gather the list of pids
+ ClientPids = case Froms of
+ {ClientPid, _Tag} ->
+ [ClientPid];
+ Froms when is_list(Froms) ->
+ [ClientPid || {ClientPid | _From} <- Froms]
+ end,
+
+ AnyAlive = lists:any(fun erlang:is_process_alive/1, ClientPids),
+ if AnyAlive -> true; true ->
+ false
+ end
+ end;
+should_process_request(_Request) ->
+ true.
+
update_counter(Tab, Channel, IOClass, RW) ->
upsert(Tab, {Channel, IOClass, RW}, 1).