This is an automated email from the ASF dual-hosted git repository. nickva pushed a commit to branch also-fix-sws-for-plain-attachment-uploads in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 3bbbea5010ac05a3c996f517cbcc040ed819d125 Author: Nick Vatamaniuc <[email protected]> AuthorDate: Wed Jul 22 16:24:49 2026 -0400 Fix all stream attachments for serialize_worker_startup In #6071 we fixed multipart attachment uploads but we forgot to fix plain attachment uploads with `PUT /db/doc/attachment`. We should apply the same fix for these too, otherwise they'll also buffer whole attachment in memory when sws=true, defeating the whole purpose of incremental attachment streaming. And add a few parsing tests, especially testing how workers wait on others and how the parser should behave if the attachment is too short --- .../test/eunit/couch_httpd_multipart_tests.erl | 148 +++++++++++++++++++++ src/fabric/src/fabric_doc_update.erl | 51 +++++-- 2 files changed, 185 insertions(+), 14 deletions(-) diff --git a/src/couch/test/eunit/couch_httpd_multipart_tests.erl b/src/couch/test/eunit/couch_httpd_multipart_tests.erl new file mode 100644 index 000000000..7411f7611 --- /dev/null +++ b/src/couch/test/eunit/couch_httpd_multipart_tests.erl @@ -0,0 +1,148 @@ +% 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_httpd_multipart_tests). + +-include_lib("couch/include/couch_eunit.hrl"). + +-define(CONTENT_TYPE, ~S'multipart/related;boundary="abc123"'). +-define(DOC_JSON, + ~B'{"_attachments":{"ohai":{"follows":true,"content_type":"text/plain","length":4}}}' +). +-define(ATT_LEN, 4). + +couch_httpd_multipart_test_() -> + { + foreach, + fun setup/0, + fun teardown/1, + [ + ?TDEF_FE(t_single_writer_short_part_fails_fast), + ?TDEF_FE(t_single_writer_empty_part_fails_fast), + ?TDEF_FE(t_single_writer_exact_part_waits_for_more_writers), + ?TDEF_FE(t_all_writers_short_part_fail) + ] + }. + +setup() -> + test_util:start_applications([config]). + +teardown(Apps) -> + test_util:stop_applications(Apps). + +% Doc with on attachment declaring length 4 but only having length 3. The +% writer which detects doesn't have enough bytes exits. +t_single_writer_short_part_fails_fast(_) -> + {Parser, ParserRef, Ref} = parse(3, body(<<"oha">>)), + Writer = spawn_writer(Parser, Ref, ?ATT_LEN), + ?assertEqual({parser_died, normal}, wait_writer(Writer, 3000)), + ?assertEqual(normal, wait_down(ParserRef, 3000)), + stop_writers([Writer]). + +% Empty body part. Writer won't get any chunk before it starves. +t_single_writer_empty_part_fails_fast(_) -> + {Parser, ParserRef, Ref} = parse(3, body(<<>>)), + Writer = spawn_writer(Parser, Ref, ?ATT_LEN), + ?assertEqual({parser_died, normal}, wait_writer(Writer, 3000)), + ?assertEqual(normal, wait_down(ParserRef, 3000)), + stop_writers([Writer]). + +% First writer consuming all bytes should not exit and wait for other to also +% connect and fetch their copies. +t_single_writer_exact_part_waits_for_more_writers(_) -> + {Parser, ParserRef, Ref} = parse(3, body(<<"ohai">>)), + Writer1 = spawn_writer(Parser, Ref, ?ATT_LEN), + ?assertEqual(got_all_bytes, wait_writer(Writer1, 3000)), + % Check it didn't die + ?assertEqual(timeout, wait_down(ParserRef, 300)), + % Late-arriving writers + Writer2 = spawn_writer(Parser, Ref, ?ATT_LEN), + ?assertEqual(got_all_bytes, wait_writer(Writer2, 3000)), + % Still shouldn't die. One more left + ?assertEqual(timeout, wait_down(ParserRef, 300)), + Writer3 = spawn_writer(Parser, Ref, ?ATT_LEN), + ?assertEqual(got_all_bytes, wait_writer(Writer3, 3000)), + % Everyone got their bytes, now it can die + ?assertEqual(normal, wait_down(ParserRef, 3000)), + stop_writers([Writer1, Writer2, Writer3]). + +% Multuple writers get the "too short" condition. That's fine, everyone dies +% without blocking. +t_all_writers_short_part_fail(_) -> + {Parser, ParserRef, Ref} = parse(3, body(<<"oha">>)), + Writers = [spawn_writer(Parser, Ref, ?ATT_LEN) || _ <- lists:seq(1, 3)], + [?assertEqual({parser_died, normal}, wait_writer(W, 3000)) || W <- Writers], + ?assertEqual(normal, wait_down(ParserRef, 3000)), + stop_writers(Writers). + +body(AttBytes) -> + << + "--abc123\r\n" + "Content-Type: application/json\r\n" + "\r\n", + ?DOC_JSON/binary, + "\r\n" + "--abc123\r\n" + "\r\n", + AttBytes/binary, + "\r\n" + "--abc123--" + >>. + +% Spawn the parser like chttpd_db for MP PUT +parse(NumWriters, Body) -> + Ref = make_ref(), + couch_httpd_multipart:num_mp_writers(NumWriters), + DataFun = fun() -> {Body, fun() -> throw(<<"expected more data">>) end} end, + {{doc_bytes, Ref, DocBytes}, Parser, ParserRef} = + couch_httpd_multipart:decode_multipart_stream(?CONTENT_TYPE, DataFun, Ref), + ?assertEqual(?DOC_JSON, iolist_to_binary(DocBytes)), + {Parser, ParserRef, Ref}. + +% This is like the reader loop in fabric_rpc:make_att_reader() +spawn_writer(Parser, Ref, Need) -> + ReportTo = self(), + spawn_link(fun() -> + ParserRef = monitor(process, Parser), + Parser ! {hello_from_writer, Ref, self()}, + ReportTo ! {self(), writer_read(Parser, ParserRef, Ref, Need)}, + receive + stop -> ok + end + end). + +writer_read(_Parser, _ParserRef, _Ref, Need) when Need =< 0 -> + got_all_bytes; +writer_read(Parser, ParserRef, Ref, Need) -> + Parser ! {get_bytes, Ref, self()}, + receive + {bytes, Ref, Bytes} -> + writer_read(Parser, ParserRef, Ref, Need - iolist_size(Bytes)); + {'DOWN', ParserRef, _, _, Reason} -> + {parser_died, Reason} + end. + +wait_writer(Writer, Timeout) -> + receive + {Writer, Result} -> Result + after Timeout -> timeout + end. + +wait_down(MonitorRef, Timeout) -> + receive + {'DOWN', MonitorRef, _, _, Reason} -> Reason + after Timeout -> timeout + end. + +stop_writers(Writers) -> + [Writer ! stop || Writer <- Writers], + ok. diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl index f1199c6a7..a73e09353 100644 --- a/src/fabric/src/fabric_doc_update.erl +++ b/src/fabric/src/fabric_doc_update.erl @@ -434,24 +434,26 @@ validate_atomic_update(_DbName, AllDocs, true) -> ), throw({aborted, PreCommitFailures}). -% Replicated changes and multipart attachment are always in parallel. MP parser -% is designed to distribute attachment chunks to num_mp_writers concurrently, -% so if we serialize them we make the MP parser buffer all the attachment -% chunks (say 1GB of data) before the subsequent workers will be started. +% Replicated changes streamed attachments are always in parallel. Both MP +% parser and fabric_doc_atts are designed to distribute attachment chunks +% concurrently. If we serialize them they would always buffer the whole +% attachment in memory (say 1GB of data) until all workers have consumed the +% byte ranges. serialize_worker_startup(AllDocs, Options) -> Replicated = proplists:get_value(?REPLICATED_CHANGES, Options) =:= true, - case Replicated orelse any_multipart_atts(AllDocs) of + case Replicated orelse any_streamed_atts(AllDocs) of true -> false; false -> config:get_boolean("fabric", "serialize_worker_startup", true) end. -any_multipart_atts(Docs) -> - DocHasMpAtt = fun(#doc{atts = Atts}) -> lists:any(fun is_multipart_att/1, Atts) end, - lists:any(DocHasMpAtt, Docs). +any_streamed_atts(Docs) -> + HasStreamedAtt = fun(#doc{atts = Atts}) -> lists:any(fun is_streamed_att/1, Atts) end, + lists:any(HasStreamedAtt, Docs). -is_multipart_att(Att) -> +is_streamed_att(Att) -> case couch_att:fetch(data, Att) of {follows, Parser, Ref} when is_pid(Parser), is_reference(Ref) -> true; + {fabric_attachment_receiver, Middleman, _} when is_pid(Middleman) -> true; _ -> false end. @@ -561,7 +563,7 @@ doc_update_test_() -> fun filter_conflicts_drops_seen_docs/0, fun parallel_in_flight_after_conflict/0, fun serial_filters_conflicts_at_cast/0, - fun sws_multipart_atts_check/0, + fun sws_streamed_atts_check/0, fun sws_false_mode_conflict_not_final/0, fun sws_false_mode_ok_can_outvote_conflict/0, fun group_docs_content_and_order/0, @@ -1174,8 +1176,10 @@ serial_filters_conflicts_at_cast() -> ?assertEqual([Tagged2], Stored). % Docs with streaming attachment don't serialize workers -sws_multipart_atts_check() -> +sws_streamed_atts_check() -> MpDoc = #doc{id = <<"m">>, atts = [mp_att()]}, + ReceiverDoc = #doc{id = <<"r">>, atts = [receiver_att()]}, + ChunkedReceiverDoc = #doc{id = <<"cr">>, atts = [chunked_receiver_att()]}, InlineAtt = couch_att:new([ {name, <<"b">>}, {type, <<"text/plain">>}, {att_len, 1}, {data, <<"x">>} ]), @@ -1184,12 +1188,15 @@ sws_multipart_atts_check() -> ]), InlineDoc = #doc{id = <<"i">>, atts = [InlineAtt, StubAtt]}, PlainDoc = #doc{id = <<"p">>}, - % Without multipart atts the config default applies + % Without streamed atts the config default applies ?assert(serialize_worker_startup([PlainDoc], [])), ?assert(serialize_worker_startup([InlineDoc], [])), - % Any doc with a mp att forces parallel startup + % Parallel for mp parser and fabric attachment receiver ?assertNot(serialize_worker_startup([MpDoc], [])), - ?assertNot(serialize_worker_startup([PlainDoc, MpDoc], [])). + ?assertNot(serialize_worker_startup([PlainDoc, MpDoc], [])), + ?assertNot(serialize_worker_startup([ReceiverDoc], [])), + ?assertNot(serialize_worker_startup([PlainDoc, ReceiverDoc], [])), + ?assertNot(serialize_worker_startup([ChunkedReceiverDoc], [])). mp_att() -> couch_att:new([ @@ -1199,6 +1206,22 @@ mp_att() -> {data, {follows, self(), make_ref()}} ]). +receiver_att() -> + couch_att:new([ + {name, <<"a">>}, + {type, <<"text/plain">>}, + {att_len, 4}, + {data, {fabric_attqachment_receiver, self(), 4}} + ]). + +chunked_receiver_att() -> + couch_att:new([ + {name, <<"a">>}, + {type, <<"text/plain">>}, + {att_len, undefined}, + {data, {fabric_attachment_receiver, self(), chunked}} + ]). + sws_false_mode_conflict_not_final() -> Docs = [Doc1, Doc2] = tag_docs([
