act-act about projects rss

CouchDB Source Code Reading part11

remainging code of couch_db:update_docs/4

前回まででcouch_db:doc_flush_atts/2の一連の処理を読み終えました。久々にcouch_db:update_docs/4に戻り、続きを見ていきます。

couch_db.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
update_docs(Db, Docs, Options, interactive_edit) ->
...
        DocBuckets3 = [[
                {doc_flush_atts(set_new_att_revpos(
                        check_dup_atts(Doc)), Db#db.updater_fd), Ref}
                || {Doc, Ref} <- B] || B <- DocBuckets2],
        {DocBuckets4, IdRevs} = new_revs(DocBuckets3, [], []),

        {ok, CommitResults} = write_and_commit(Db, DocBuckets4, NonRepDocs, Options2),

        ResultsDict = dict:from_list(IdRevs ++ CommitResults ++ PreCommitFailures),
        {ok, lists:map(
            fun({#doc{}, Ref}) ->
                {ok, Result} = dict:find(Ref, ResultsDict),
                Result
            end, Docs2)}
    end.

DockBucket3はアタッチメント情報をディスクに書き込み、そのポインタを持ったDocによって構成されています。DocBucket3の構成はcouch_db:goup_alike_docs/1のコメントに記載されています。具体的には以下の矢印右の構造になります。

couch_db.erl

1
% ([DocA, DocA, DocB, DocC], []) -> [[DocA, DocA], [DocB], [DocC]]

これを第一引数に取っているcouch_db:new_revs/3を見ていきます。

couch_db.erl

1
2
3
4
5
6
7
8
9
10
new_revs([], OutBuckets, IdRevsAcc) ->
    {lists:reverse(OutBuckets), IdRevsAcc};
new_revs([Bucket|RestBuckets], OutBuckets, IdRevsAcc) ->
    {NewBucket, IdRevsAcc3} = lists:mapfoldl(
        fun({#doc{revs={Start, RevIds}}=Doc, Ref}, IdRevsAcc2)->
        NewRevId = new_revid(Doc),
        {{Doc#doc{revs={Start+1, [NewRevId | RevIds]}}, Ref},
            [{Ref, {ok, {Start+1, NewRevId}}} | IdRevsAcc2]}
    end, IdRevsAcc, Bucket),
    new_revs(RestBuckets, [NewBucket|OutBuckets], IdRevsAcc3).

lists:mapfoldl/3lists:map/2lists:foldl/3を組み合わせた関数で、1回のリスト走査でmapfoldlを実行します。

1
2
3
Eshell V5.10.3  (abort with ^G)
1> lists:mapfoldl(fun(X, Sum) -> {X*2, X*2+Sum} end, 0, [1,2,3,4,5]).
{[2,4,6,8,10],30}

Bucketリストを指定してlists:mapfoldl/3します。couch_db:new_revid/1で新しいRevIdを生成し、#doc.revsRevsIdに追加しています。Startに1を加えているのが良く分からない…。この値はTreの階層を表すと思っていたのですが、違うのかな。mapの方には新しく生成した#docを、foldlの方には{Ref, {ok, {Start+1, NewRevId}}}を加えます。最終的に、呼び出し元に対して、新しいリビジョンを追加したドキュメント群をDockBuckets4、新しく生成したRevIdの情報をIdRevsとして返します。

couch_db:write_and_commit/4

次にcouch_db:write_and_commit/4を見てみます。

couch_db.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
write_and_commit(#db{update_pid=UpdatePid}=Db, DocBuckets1,
        NonRepDocs, Options0) ->
    DocBuckets = prepare_doc_summaries(Db, DocBuckets1),
    Options = set_commit_option(Options0),
    MergeConflicts = lists:member(merge_conflicts, Options),
    FullCommit = lists:member(full_commit, Options),
    MRef = erlang:monitor(process, UpdatePid),
    try
        UpdatePid ! {update_docs, self(), DocBuckets, NonRepDocs, MergeConflicts, FullCommit},
        case collect_results(UpdatePid, MRef, []) of
        {ok, Results} -> {ok, Results};
        retry ->
            % This can happen if the db file we wrote to was swapped out by
            % compaction. Retry by reopening the db and writing to the current file
            {ok, Db2} = open_ref_counted(Db#db.main_pid, self()),
            DocBuckets2 = [
                [{doc_flush_atts(Doc, Db2#db.updater_fd), Ref} || {Doc, Ref} <- Bucket] ||
                Bucket <- DocBuckets1
            ],
            % We only retry once
            DocBuckets3 = prepare_doc_summaries(Db2, DocBuckets2),
            close(Db2),
            UpdatePid ! {update_docs, self(), DocBuckets3, NonRepDocs, MergeConflicts, FullCommit},
            case collect_results(UpdatePid, MRef, []) of
            {ok, Results} -> {ok, Results};
            retry -> throw({update_error, compaction_retry})
            end
        end
    after
        erlang:demonitor(MRef, [flush])
    end.

そのままcouch_db:prepare_doc_summaries/2を見てみます。

couch_db.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
prepare_doc_summaries(Db, BucketList) ->
    [lists:map(
        fun({#doc{body = Body, atts = Atts} = Doc, Ref}) ->
            DiskAtts = [{N, T, P, AL, DL, R, M, E} ||
                #att{name = N, type = T, data = {_, P}, md5 = M, revpos = R,
                    att_len = AL, disk_len = DL, encoding = E} <- Atts],
            AttsFd = case Atts of
            [#att{data = {Fd, _}} | _] ->
                Fd;
            [] ->
                nil
            end,
            SummaryChunk = couch_db_updater:make_doc_summary(Db, {Body, DiskAtts}),
            {Doc#doc{body = {summary, SummaryChunk, AttsFd}}, Ref}
        end,
        Bucket) || Bucket <- BucketList].

#doc.attsをディスクに書き込む為に変換し、また#doc.attsからFdを取り出しています。ディスクに書き込む為に変換したDiskAtts#doc.bodyを指定し、couch_db_updater:make_doc_summary/2を呼び出しています。この関数を見てみます。

couch_db.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
make_doc_summary(#db{compression = Comp}, {Body0, Atts0}) ->
    Body = case couch_compress:is_compressed(Body0, Comp) of
    true ->
        Body0;
    false ->
        % pre 1.2 database file format
        couch_compress:compress(Body0, Comp)
    end,
    Atts = case couch_compress:is_compressed(Atts0, Comp) of
    true ->
        Atts0;
    false ->
        couch_compress:compress(Atts0, Comp)
    end,
    SummaryBin = ?term_to_bin({Body, Atts}),
    couch_file:assemble_file_chunk(SummaryBin, couch_util:md5(SummaryBin)).

引数で渡された{Body0, Atts0}をそれぞれcompressしてバイナリに変換しているようです。続けてcouch_file:assemble_file_chunk/2を見てみます。

couch_file.erl

1
2
3
4
5
assemble_file_chunk(Bin) ->
    [<<0:1/integer, (iolist_size(Bin)):31/integer>>, Bin].

assemble_file_chunk(Bin, Md5) ->
    [<<1:1/integer, (iolist_size(Bin)):31/integer>>, Md5, Bin].

前回書いたように、データブロックに書き出す為にヘッダを付けてます。

couch_db:write_and_commit/4の冒頭に戻り、couch_db:prepare_doc_summaries/2の呼び出しによりDocBucketに格納された#docbody{summary, SummaryChunk, AttsFd}になっています。前記のデータブロックにかい出す為のヘッダを付けたバイナリはSummaryChunkにバインドされています。

MergeConflictsFullCommitを設定し、下記のcouch_db_updater:handle_info/2が呼び出されます。

couch_db_updater.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
handle_info({update_docs, Client, GroupedDocs, NonRepDocs, MergeConflicts,
        FullCommit}, Db) ->
    GroupedDocs2 = [[{Client, D} || D <- DocGroup] || DocGroup <- GroupedDocs],
    if NonRepDocs == [] ->
        {GroupedDocs3, Clients, FullCommit2} = collect_updates(GroupedDocs2,
                [Client], MergeConflicts, FullCommit);
    true ->
        GroupedDocs3 = GroupedDocs2,
        FullCommit2 = FullCommit,
        Clients = [Client]
    end,
    NonRepDocs2 = [{Client, NRDoc} || NRDoc <- NonRepDocs],
    try update_docs_int(Db, GroupedDocs3, NonRepDocs2, MergeConflicts,
                FullCommit2) of
    {ok, Db2, UpdatedDDocIds} ->
        ok = gen_server:call(Db#db.main_pid, {db_updated, Db2}),
        if Db2#db.update_seq /= Db#db.update_seq ->
            couch_db_update_notifier:notify({updated, Db2#db.name});
        true -> ok
        end,
        [catch(ClientPid ! {done, self()}) || ClientPid <- Clients],
        lists:foreach(fun(DDocId) ->
            couch_db_update_notifier:notify({ddoc_updated, {Db#db.name, DDocId}})
        end, UpdatedDDocIds),
        {noreply, Db2}
    catch
        throw: retry ->
            [catch(ClientPid ! {retry, self()}) || ClientPid <- Clients],
            {noreply, Db}
    end;

GroupDocsGroupDcos2NonRepDocsNonRepDocs2への変換後、couch_db_updater:update_docs_int/5を呼び出していま。

couch_db_updater:update_docs_int/5

couch_db_updater:update_docs_int/5を見てみます。

couch_db_updater.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
update_docs_int(Db, DocsList, NonRepDocs, MergeConflicts, FullCommit) ->
    #db{
        fulldocinfo_by_id_btree = DocInfoByIdBTree,
        docinfo_by_seq_btree = DocInfoBySeqBTree,
        update_seq = LastSeq,
        revs_limit = RevsLimit
        } = Db,
    Ids = [Id || [{_Client, {#doc{id=Id}, _Ref}}|_] <- DocsList],
    % lookup up the old documents, if they exist.
    OldDocLookups = couch_btree:lookup(DocInfoByIdBTree, Ids),
    OldDocInfos = lists:zipwith(
        fun(_Id, {ok, FullDocInfo}) ->
            FullDocInfo;
        (Id, not_found) ->
            #full_doc_info{id=Id}
        end,
        Ids, OldDocLookups),
    % Merge the new docs into the revision trees.
    {ok, NewFullDocInfos, RemoveSeqs, NewSeq} = merge_rev_trees(RevsLimit,
            MergeConflicts, DocsList, OldDocInfos, [], [], LastSeq),

    % All documents are now ready to write.

    {ok, Db2}  = update_local_docs(Db, NonRepDocs),

    % Write out the document summaries (the bodies are stored in the nodes of
    % the trees, the attachments are already written to disk)
    {ok, FlushedFullDocInfos} = flush_trees(Db2, NewFullDocInfos, []),

    {IndexFullDocInfos, IndexDocInfos, UpdatedDDocIds} =
            new_index_entries(FlushedFullDocInfos, [], [], []),

    % and the indexes
    {ok, DocInfoByIdBTree2} = couch_btree:add_remove(DocInfoByIdBTree, IndexFullDocInfos, []),
    {ok, DocInfoBySeqBTree2} = couch_btree:add_remove(DocInfoBySeqBTree, IndexDocInfos, RemoveSeqs),

    Db3 = Db2#db{
        fulldocinfo_by_id_btree = DocInfoByIdBTree2,
        docinfo_by_seq_btree = DocInfoBySeqBTree2,
        update_seq = NewSeq},

    % Check if we just updated any design documents, and update the validation
    % funs if we did.
    Db4 = case UpdatedDDocIds of
    [] ->
        Db3;
    _ ->
        refresh_validate_doc_funs(Db3)
    end,

    {ok, commit_data(Db4, not FullCommit), UpdatedDDocIds}.

大分長いので大まかに見ていきます。更新対象のDocListからIdを取り出し、そのIdを指定して#db.fulldocinfo_by_id_btreeからFullDocInfoを取り出します。該当するFullDocInfoが無かった場合はIdだけ設定した空のFullDocInfoを生成し、OldDocInfosとします。更新対象のDocListOldDocInfosを指定し、couch_db_updater:merge_rev_trees/7を呼び出します。

couch_db_updater:merge_rev_trees/7

couch_db_updater:merge_rev_trees/7を見て行きます。

couch_db_updater.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
merge_rev_trees(_Limit, _Merge, [], [], AccNewInfos, AccRemoveSeqs, AccSeq) ->
    {ok, lists:reverse(AccNewInfos), AccRemoveSeqs, AccSeq};
merge_rev_trees(Limit, MergeConflicts, [NewDocs|RestDocsList],
        [OldDocInfo|RestOldInfo], AccNewInfos, AccRemoveSeqs, AccSeq) ->
    #full_doc_info{id=Id,rev_tree=OldTree,deleted=OldDeleted0,update_seq=OldSeq}
            = OldDocInfo,
    {NewRevTree, _} = lists:foldl(
        fun({Client, {#doc{revs={Pos,[_Rev|PrevRevs]}}=NewDoc, Ref}}, {AccTree, OldDeleted}) ->
            if not MergeConflicts ->
                case couch_key_tree:merge(AccTree, couch_doc:to_path(NewDoc),
                    Limit) of
                {_NewTree, conflicts} when (not OldDeleted) ->
                    send_result(Client, Ref, conflict),
                    {AccTree, OldDeleted};
                {NewTree, conflicts} when PrevRevs /= [] ->
                    % Check to be sure if prev revision was specified, it's
                    % a leaf node in the tree
                    Leafs = couch_key_tree:get_all_leafs(AccTree),
                    IsPrevLeaf = lists:any(fun({_, {LeafPos, [LeafRevId|_]}}) ->
                            {LeafPos, LeafRevId} == {Pos-1, hd(PrevRevs)}
                        end, Leafs),
                    if IsPrevLeaf ->
                        {NewTree, OldDeleted};
                    true ->
                        send_result(Client, Ref, conflict),
                        {AccTree, OldDeleted}
                    end;
                {NewTree, no_conflicts} when  AccTree == NewTree ->
                    % the tree didn't change at all
                    % meaning we are saving a rev that's already
                    % been editted again.
                    if (Pos == 1) and OldDeleted ->
                        % this means we are recreating a brand new document
                        % into a state that already existed before.
                        % put the rev into a subsequent edit of the deletion
                        #doc_info{revs=[#rev_info{rev={OldPos,OldRev}}|_]} =
                                couch_doc:to_doc_info(OldDocInfo),
                        NewRevId = couch_db:new_revid(
                                NewDoc#doc{revs={OldPos, [OldRev]}}),
                        NewDoc2 = NewDoc#doc{revs={OldPos + 1, [NewRevId, OldRev]}},
                        {NewTree2, _} = couch_key_tree:merge(AccTree,
                                couch_doc:to_path(NewDoc2), Limit),
                        % we changed the rev id, this tells the caller we did
                        send_result(Client, Ref, {ok, {OldPos + 1, NewRevId}}),
                        {NewTree2, OldDeleted};
                    true ->
                        send_result(Client, Ref, conflict),
                        {AccTree, OldDeleted}
                    end;
                {NewTree, _} ->
                    {NewTree, NewDoc#doc.deleted}
                end;
            true ->
                {NewTree, _} = couch_key_tree:merge(AccTree,
                            couch_doc:to_path(NewDoc), Limit),
                {NewTree, OldDeleted}
            end
        end,
        {OldTree, OldDeleted0}, NewDocs),
    if NewRevTree == OldTree ->
        % nothing changed
        merge_rev_trees(Limit, MergeConflicts, RestDocsList, RestOldInfo,
            AccNewInfos, AccRemoveSeqs, AccSeq);
    true ->
        % we have updated the document, give it a new seq #
        NewInfo = #full_doc_info{id=Id,update_seq=AccSeq+1,rev_tree=NewRevTree},
        RemoveSeqs = case OldSeq of
            0 -> AccRemoveSeqs;
            _ -> [OldSeq | AccRemoveSeqs]
        end,
        merge_rev_trees(Limit, MergeConflicts, RestDocsList, RestOldInfo,
            [NewInfo|AccNewInfos], RemoveSeqs, AccSeq+1)
    end.

更新対象のドキュメントと、それに対する#full_doc_infoをマージする処理で、実際のマージはcouch_doc:merge/3を呼び出すことによって実現しています。その関数を呼び出す前にcouch_doc:to_path/1を呼び出しているので、先にこの関数を見てみます。

couch_doc.erl

1
2
3
4
5
6
7
8
9
10
-spec to_path(#doc{}) -> path().
to_path(#doc{revs={Start, RevIds}}=Doc) ->
    [Branch] = to_branch(Doc, lists:reverse(RevIds)),
    {Start - length(RevIds) + 1, Branch}.

-spec to_branch(#doc{}, [RevId::binary()]) -> [branch()].
to_branch(Doc, [RevId]) ->
    [{RevId, Doc, []}];
to_branch(Doc, [RevId | Rest]) ->
    [{RevId, ?REV_MISSING, to_branch(Doc, Rest)}].

上記関数を呼び出すと、#docは以下のように変換されます。

1
{3, {RevId3, [], {RevId2, [], {RevId1, Doc, []}}}}}.

また、#full_doc_info.rev_treeは以下のような構造になっています。

1
[{Pos, {Key, Value, SubTree}}, {Pos, {Key, Value, SubTree}}, ...}]

上記より、couch_doc:to_path/1#full_doc_info.rev_treeとマージできるように#doc{Key, Value, SubTree}の形式に変換しているようです。

ここからcouch_doc:merge/3couch_doc:merge_one/4couch_doc:merge_at/3couch_doc:merge_simp2と続き、rev_treeのマージが行われますが、かなり長いので割愛します。更新の衝突がなければ、新しいリビジョンが増えるだけだと思います。

couch_db_updater:flush_trees/3

couch_db_updater:update_docs_int/5に戻ります。次に実行されるのはcouch_db_updater:update_local_docs/2ですが、この関数はrev_treeは関係しないので、後まわしにして先にcouch_db_updater:flush_trees/3を見てみます。

couch_db_updater.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
flush_trees(_Db, [], AccFlushedTrees) ->
    {ok, lists:reverse(AccFlushedTrees)};
flush_trees(#db{updater_fd = Fd} = Db,
        [InfoUnflushed | RestUnflushed], AccFlushed) ->
    #full_doc_info{update_seq=UpdateSeq, rev_tree=Unflushed} = InfoUnflushed,
    {Flushed, LeafsSize} = couch_key_tree:mapfold(
        fun(_Rev, Value, Type, Acc) ->
            case Value of
            #doc{deleted = IsDeleted, body = {summary, Summary, AttsFd}} ->
                % this node value is actually an unwritten document summary,
                % write to disk.
                % make sure the Fd in the written bins is the same Fd we are
                % and convert bins, removing the FD.
                % All bins should have been written to disk already.
                case {AttsFd, Fd} of
                {nil, _} ->
                    ok;
                {SameFd, SameFd} ->
                    ok;
                _ ->
                    % Fd where the attachments were written to is not the same
                    % as our Fd. This can happen when a database is being
                    % switched out during a compaction.
                    ?LOG_DEBUG("File where the attachments are written has"
                            " changed. Possibly retrying.", []),
                    throw(retry)
                end,
                {ok, NewSummaryPointer, SummarySize} =
                    couch_file:append_raw_chunk(Fd, Summary),
                TotalSize = lists:foldl(
                    fun(#att{att_len = L}, A) -> A + L end,
                    SummarySize, Value#doc.atts),
                NewValue = {IsDeleted, NewSummaryPointer, UpdateSeq, TotalSize},
                case Type of
                leaf ->
                    {NewValue, Acc + TotalSize};
                branch ->
                    {NewValue, Acc}
                end;
             {_, _, _, LeafSize} when Type =:= leaf, LeafSize =/= nil ->
                {Value, Acc + LeafSize};
             _ ->
                {Value, Acc}
            end
        end, 0, Unflushed),
    InfoFlushed = InfoUnflushed#full_doc_info{
        rev_tree = Flushed,
        leafs_size = LeafsSize
    },
    flush_trees(Db, RestUnflushed, [InfoFlushed | AccFlushed]).

#full_doc_info.rev_treecouch_key_tree/3でまわしていきます。このループの中で#doc.bodySummaryをファイルに書き出していき、そのファイルポインタを#docの代わりにValueとして設定します。つまりこの関数では#doc.bodyをファイルに書き出していることになります。

couch_db_updater:new_index_entries/4

couch_db_updater:update_docs_int/5に戻ります。次に実行されるのはcouch_db_updater:new_index_entries/4なので、この関数を見てみます。

couch_db_updater.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
new_index_entries([], AccById, AccBySeq, AccDDocIds) ->
    {AccById, AccBySeq, AccDDocIds};
new_index_entries([FullDocInfo|RestInfos], AccById, AccBySeq, AccDDocIds) ->
    #doc_info{revs=[#rev_info{deleted=Deleted}|_], id=Id} = DocInfo =
            couch_doc:to_doc_info(FullDocInfo),
    AccDDocIds2 = case Id of
    <<?DESIGN_DOC_PREFIX, _/binary>> ->
        [Id | AccDDocIds];
    _ ->
        AccDDocIds
    end,
    new_index_entries(RestInfos,
        [FullDocInfo#full_doc_info{deleted=Deleted}|AccById],
        [DocInfo|AccBySeq],
        AccDDocIds2).

これは#full_doc_infoAccByIdに、#doc_infoAccBySeqに、DesignDocのIdをAccDDocIdsに追加している関数のようです。#doc_info.rev_infoの先頭の要素のdeleted#full_doc_info.deletedに設定しているので、削除時はこの契機で#full_doc_info.deletedが設定される、ということになるのかな。

couch_btree:add_remove/3

couch_db_updater:update_docs_int/5に戻ります。couch_db_updater:new_index_entries/4の戻り値{IndexFullDocInfos, IndexDocInfos, UpdatedDDocIds}を使い、DocInfoByIdBTreeDocInfoBySeqBTreeを更新します。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
add_remove(Bt, InsertKeyValues, RemoveKeys) ->
    {ok, [], Bt2} = query_modify(Bt, [], InsertKeyValues, RemoveKeys),
    {ok, Bt2}.

query_modify(Bt, LookupKeys, InsertValues, RemoveKeys) ->
    #btree{root=Root} = Bt,
    InsertActions = lists:map(
        fun(KeyValue) ->
            {Key, Value} = extract(Bt, KeyValue),
            {insert, Key, Value}
        end, InsertValues),
    RemoveActions = [{remove, Key, nil} || Key <- RemoveKeys],
    FetchActions = [{fetch, Key, nil} || Key <- LookupKeys],
    SortFun =
        fun({OpA, A, _}, {OpB, B, _}) ->
            case A == B of
            % A and B are equal, sort by op.
            true -> op_order(OpA) < op_order(OpB);
            false ->
                less(Bt, A, B)
            end
        end,
    Actions = lists:sort(SortFun, lists:append([InsertActions, RemoveActions, FetchActions])),
    {ok, KeyPointers, QueryResults} = modify_node(Bt, Root, Actions, []),
    {ok, NewRoot} = complete_root(Bt, KeyPointers),
    {ok, QueryResults, Bt#btree{root=NewRoot}}.

couch_btree:extract/2を呼び出しKeyValueに分けています。このあたりについては以前見たように、#full_doc_infoをディスクに書き込み可能な形式に変換しています。それぞれのActionを準備した後、couch_btree:modify_node/4を呼び出します。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
modify_node(Bt, RootPointerInfo, Actions, QueryOutput) ->
    case RootPointerInfo of
    nil ->
        NodeType = kv_node,
        NodeList = [];
    _Tuple ->
        Pointer = element(1, RootPointerInfo),
        {NodeType, NodeList} = get_node(Bt, Pointer)
    end,
    NodeTuple = list_to_tuple(NodeList),

    {ok, NewNodeList, QueryOutput2} =
    case NodeType of
    kp_node -> modify_kpnode(Bt, NodeTuple, 1, Actions, [], QueryOutput);
    kv_node -> modify_kvnode(Bt, NodeTuple, 1, Actions, [], QueryOutput)
    end,
    case NewNodeList of
    [] ->  % no nodes remain
        {ok, [], QueryOutput2};
    NodeList ->  % nothing changed
        {LastKey, _LastValue} = element(tuple_size(NodeTuple), NodeTuple),
        {ok, [{LastKey, RootPointerInfo}], QueryOutput2};
    _Else2 ->
        {ok, ResultList} = write_node(Bt, NodeType, NewNodeList),
        {ok, ResultList, QueryOutput2}
    end.

更新対象のTreeからNodeTypeを取り出し、それによってmodify_kpnode/6modify_kvnode/6のどちらかを呼び出します。couch_db_updater:update_docs_int/5からの呼び出しを見ると結局どちらも呼び出されるのですが、modify_kvnode/6の方を見ていきます。

couch_btree:modify_kvnode/6

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
modify_kvnode(_Bt, NodeTuple, LowerBound, [], ResultNode, QueryOutput) ->
    {ok, lists:reverse(ResultNode, bounded_tuple_to_list(NodeTuple, LowerBound, tuple_size(NodeTuple), [])), QueryOutput};
modify_kvnode(Bt, NodeTuple, LowerBound, [{ActionType, ActionKey, ActionValue} | RestActions], ResultNode, QueryOutput) when LowerBound > tuple_size(NodeTuple) ->
    case ActionType of
    insert ->
        modify_kvnode(Bt, NodeTuple, LowerBound, RestActions, [{ActionKey, ActionValue} | ResultNode], QueryOutput);
    remove ->
        % just drop the action
        modify_kvnode(Bt, NodeTuple, LowerBound, RestActions, ResultNode, QueryOutput);
    fetch ->
        % the key/value must not exist in the tree
        modify_kvnode(Bt, NodeTuple, LowerBound, RestActions, ResultNode, [{not_found, {ActionKey, nil}} | QueryOutput])
    end;
modify_kvnode(Bt, NodeTuple, LowerBound, [{ActionType, ActionKey, ActionValue} | RestActions], AccNode, QueryOutput) ->
    N = find_first_gteq(Bt, NodeTuple, LowerBound, tuple_size(NodeTuple), ActionKey),
    {Key, Value} = element(N, NodeTuple),
    ResultNode =  bounded_tuple_to_revlist(NodeTuple, LowerBound, N - 1, AccNode),
    case less(Bt, ActionKey, Key) of
    true ->
        case ActionType of
        insert ->
            % ActionKey is less than the Key, so insert
            modify_kvnode(Bt, NodeTuple, N, RestActions, [{ActionKey, ActionValue} | ResultNode], QueryOutput);
        remove ->
            % ActionKey is less than the Key, just drop the action
            modify_kvnode(Bt, NodeTuple, N, RestActions, ResultNode, QueryOutput);
        fetch ->
            % ActionKey is less than the Key, the key/value must not exist in the tree
            modify_kvnode(Bt, NodeTuple, N, RestActions, ResultNode, [{not_found, {ActionKey, nil}} | QueryOutput])
        end;
    false ->
        % ActionKey and Key are maybe equal.
        case less(Bt, Key, ActionKey) of
        false ->
            case ActionType of
            insert ->
                modify_kvnode(Bt, NodeTuple, N+1, RestActions, [{ActionKey, ActionValue} | ResultNode], QueryOutput);
            remove ->
                modify_kvnode(Bt, NodeTuple, N+1, RestActions, ResultNode, QueryOutput);
            fetch ->
                % ActionKey is equal to the Key, insert into the QueryOuput, but re-process the node
                % since an identical action key can follow it.
                modify_kvnode(Bt, NodeTuple, N, RestActions, ResultNode, [{ok, assemble(Bt, Key, Value)} | QueryOutput])
            end;
        true ->
            modify_kvnode(Bt, NodeTuple, N + 1, [{ActionType, ActionKey, ActionValue} | RestActions], [{Key, Value} | ResultNode], QueryOutput)
        end
    end.

2番目のwhen LowerBound > tuple_size(NodeTuple)というガードが付いている方は、現在のFullDocInfoに当該Keyが含まれていない時に呼び出される関数のようです。メインは3番目の関数。最初に呼び出しているcouch_btree:find_first_gteq/5を見てみます。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
find_first_gteq(_Bt, _Tuple, Start, End, _Key) when Start == End ->
    End;
find_first_gteq(Bt, Tuple, Start, End, Key) ->
    Mid = Start + ((End - Start) div 2),
    {TupleKey, _} = element(Mid, Tuple),
    case less(Bt, TupleKey, Key) of
    true ->
        find_first_gteq(Bt, Tuple, Mid+1, End, Key);
    false ->
        find_first_gteq(Bt, Tuple, Start, Mid, Key)
    end.

Keyでバイナリサーチしています。そのままcouch_btree:bounded_tuple_to_revlist/4を見てみます。

couch_btree.erl

1
2
3
4
bounded_tuple_to_revlist(_Tuple, Start, End, Tail) when Start > End ->
    Tail;
bounded_tuple_to_revlist(Tuple, Start, End, Tail) ->
    bounded_tuple_to_revlist(Tuple, Start+1, End, [element(Start, Tuple)|Tail]).

couch_btree:bounded_tuple_to_revlist/4は、couch_btree:find_first_gteq/5を実行して取得したNに対して、LowerBoundからN-1までのノードをアキュムレータに追加しています。

その後、ActionKeyN番目のKeyと比較します。その結果によって次の操作をNN+1のどちらかから始めるかを決めています。そして、ActionTypeinsertならアキュムレータResultNodeに追加、deleteであれば追加せず、fetchであればQueryOutputの方に追加します。

最終的にcouch_btree:modify_kvnode/6の番目の関数が呼び出されます。ここで、couch_btree:bounded_tuple_to_list/4を呼び出してNodeTupleのうち走査しなかった残りの部分を取得し、ResultNodeをreverseしたリストと連結して返します。つまり、この関数の中でkvnodeのリストに対してActionsに入っている操作を適用し、リストを再構成しています。ActionTypedeleteの時にアキュムレータに追加していないのは、追加しないことによって再構成後のリストにエントリが含まれなくなり、消去されたことになります。また、insertの時のActionKeyNodeTupleKeyと完全に一致する場合、NodeTuple側のエントリはアキュムレータに追加されず、{ActionKey, ActionValue}が追加される為、既存のエントリは再構成後のエントリに含まれず、結果新しい値で上書きしたようになります。

couch_btree:modify_kpnode/6

次にcouch_btree:modify_kpnode/6を見てみます。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
modify_kpnode(Bt, {}, _LowerBound, Actions, [], QueryOutput) ->
    modify_node(Bt, nil, Actions, QueryOutput);
modify_kpnode(_Bt, NodeTuple, LowerBound, [], ResultNode, QueryOutput) ->
    {ok, lists:reverse(ResultNode, bounded_tuple_to_list(NodeTuple, LowerBound,
            tuple_size(NodeTuple), [])), QueryOutput};
modify_kpnode(Bt, NodeTuple, LowerBound,
        [{_, FirstActionKey, _}|_]=Actions, ResultNode, QueryOutput) ->
    Sz = tuple_size(NodeTuple),
    N = find_first_gteq(Bt, NodeTuple, LowerBound, Sz, FirstActionKey),
    case N =:= Sz of
    true  ->
        % perform remaining actions on last node
        {_, PointerInfo} = element(Sz, NodeTuple),
        {ok, ChildKPs, QueryOutput2} =
            modify_node(Bt, PointerInfo, Actions, QueryOutput),
        NodeList = lists:reverse(ResultNode, bounded_tuple_to_list(NodeTuple, LowerBound,
            Sz - 1, ChildKPs)),
        {ok, NodeList, QueryOutput2};
    false ->
        {NodeKey, PointerInfo} = element(N, NodeTuple),
        SplitFun = fun({_ActionType, ActionKey, _ActionValue}) ->
                not less(Bt, NodeKey, ActionKey)
            end,
        {LessEqQueries, GreaterQueries} = lists:splitwith(SplitFun, Actions),
        {ok, ChildKPs, QueryOutput2} =
                modify_node(Bt, PointerInfo, LessEqQueries, QueryOutput),
        ResultNode2 = lists:reverse(ChildKPs, bounded_tuple_to_revlist(NodeTuple,
                LowerBound, N - 1, ResultNode)),
        modify_kpnode(Bt, NodeTuple, N+1, GreaterQueries, ResultNode2, QueryOutput2)
    end.

3番目の関数を見ていくと、まずNodeTupleのサイズとcouch_btree:find_first_gteq/5の結果を比較しています。値が一致する場合は、残りのActionsを指定してcouch_btree:modify_node/4を呼び出し、その結果をアキュムレータに加えて返しています。値が一致しない場合は、couch_btree:find_first_gteq/5の結果で返ってきたインデックスを使ってNodeTupleからタプルを取り出し、そのKeyを境としてActionsを2つに分割します。Key以下のActionKeyを持つActionsを指定してcouch_btree:modify_node/4を呼び出し、その結果をアキュムレータに加えて、Keyより大きいActionKeyを持つActionsを指定してcouch_btree:modify_kpnode/6が呼び出されます。

正直きちんと理解できていないのですが、この関数でcouch_btree:modify_node/4を呼び出すと、恐らくそのその先でNodeTypekv_nodeに変わり、couch_btree:modify_kvnode/4が呼び出され、前記のように更新後のリストが取れるようになるのだと思います。

couch_btree:write_node/3

couch_btree:modify_node/4に戻り、couch_btree:write_node/3を見てみます。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
write_node(#btree{fd = Fd, compression = Comp} = Bt, NodeType, NodeList) ->
    % split up nodes into smaller sizes
    NodeListList = chunkify(NodeList),
    % now write out each chunk and return the KeyPointer pairs for those nodes
    ResultList = [
        begin
            {ok, Pointer, Size} = couch_file:append_term(
                Fd, {NodeType, ANodeList}, [{compression, Comp}]),
            {LastKey, _} = lists:last(ANodeList),
            SubTreeSize = reduce_tree_size(NodeType, Size, ANodeList),
            {LastKey, {Pointer, reduce_node(Bt, NodeType, ANodeList), SubTreeSize}}
        end
    ||
        ANodeList <- NodeListList
    ],
    {ok, ResultList}.

couch_btree:chunkify/1NodeListを分割し、その後に分割した単位でデータベースファイルに追記(couch_file:append_term/3)しています。couch_btree:chunkify/1を見てみます。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
chunkify(InList) ->
    case ?term_size(InList) of
    Size when Size > ?CHUNK_THRESHOLD ->
        NumberOfChunksLikely = ((Size div ?CHUNK_THRESHOLD) + 1),
        ChunkThreshold = Size div NumberOfChunksLikely,
        chunkify(InList, ChunkThreshold, [], 0, []);
    _Else ->
        [InList]
    end.

chunkify([], _ChunkThreshold, [], 0, OutputChunks) ->
    lists:reverse(OutputChunks);
chunkify([], _ChunkThreshold, OutList, _OutListSize, OutputChunks) ->
    lists:reverse([lists:reverse(OutList) | OutputChunks]);
chunkify([InElement | RestInList], ChunkThreshold, OutList, OutListSize, OutputChunks) ->
    case ?term_size(InElement) of
    Size when (Size + OutListSize) > ChunkThreshold andalso OutList /= [] ->
        chunkify(RestInList, ChunkThreshold, [], 0, [lists:reverse([InElement | OutList]) | OutputChunks]);
    Size ->
        chunkify(RestInList, ChunkThreshold, [InElement | OutList], OutListSize + Size, OutputChunks)
    end.

InListのサイズを?CHUNK_THRESHOLD(1279)と比較しています。いくつに分割するべきか(NumberOfChunksLikely)を求め、InListのサイズから分割後の各リストに含まれる要素数を求め、分割していきます。

データベースファイルに追加した後、分割されたリスト内のの最後のKeyを取得します。そしてreduce_tree_size/3を呼び出し、SubTreeSizeを求めます。couch_btree:reduce_tree_size/3を見てみます。

couch_btree.erl

1
2
3
4
5
6
7
8
9
10
11
reduce_tree_size(kv_node, NodeSize, _KvList) ->
    NodeSize;
reduce_tree_size(kp_node, NodeSize, []) ->
    NodeSize;
reduce_tree_size(kp_node, _NodeSize, [{_K, {_P, _Red}} | _]) ->
    % pre 1.2 format
    nil;
reduce_tree_size(kp_node, _NodeSize, [{_K, {_P, _Red, nil}} | _]) ->
    nil;
reduce_tree_size(kp_node, NodeSize, [{_K, {_P, _Red, Sz}} | NodeList]) ->
    reduce_tree_size(kp_node, NodeSize + Sz, NodeList).

NodeTypeによって処理が分かれています。kv_nodeの場合はNodeSizeが返るだけです。kp_nodeの場合は、NodeListの各要素に含まれるTreeSizeNodeSizeに加えています。

続けてcouch_btree:reduce_node/3を見てみます。

couch_btree.erl

1
2
3
4
5
6
reduce_node(#btree{reduce=nil}, _NodeType, _NodeList) ->
    [];
reduce_node(#btree{reduce=R}, kp_node, NodeList) ->
    R(rereduce, [element(2, Node) || {_K, Node} <- NodeList]);
reduce_node(#btree{reduce=R}=Bt, kv_node, NodeList) ->
    R(reduce, [assemble(Bt, K, V) || {K, V} <- NodeList]).

これもNodeTypeによって処理が分かれています。Bt#btree.reduceにバインドされている関数に関しては、kp_nodekv_node共に以前見ていますのでここでは割愛します。

couch_btree:write_node/3を呼び出した結果、以下の値が返ります。

[ok, [{LastKey, {Pointer, reduce_node(Bt, NodeType, ANodeList), SubTreeSize}}, ...]]   

couch_btree:complete_root/1

couch_btree:modify_node/4まで一通り見たので、couch_btree:add_remove/3に戻り、couch_btree:complete_root/2を見てみます。

couch_btree.erl

1
2
3
4
5
6
7
complete_root(_Bt, []) ->
    {ok, nil};
complete_root(_Bt, [{_Key, PointerInfo}])->
    {ok, PointerInfo};
complete_root(Bt, KPs) ->
    {ok, ResultKeyPointers} = write_node(Bt, kp_node, KPs),
    complete_root(Bt, ResultKeyPointers).

第二引数の要素数によって処理が違っています。この第二引数のリストの要素数は、couch_btree:write_node/3中で呼び出したcouch_btree:chunkify/1によって分割された数になります。つまり複数のリストに分割された場合は3番目の関数が、分割されなかった場合は2番目の関数が呼ばれます。

couch_btree:add_remove/3の最後で、この関数の戻り値{ok, PointerInfo}Bt#btree.rootにバインドします。ようやくcouch_db_updater:update_docs_int/5に戻り、更新されたそれぞれの#btreeを新しいBtreeとします。

couch_db_updater:commit_data/3

couch_db_updater:update_docs_int/5で最後に呼び出していいる、couch_db_updater:commit_data/3を見てみます。

couch_db_updater.erl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
commit_data(Db) ->
    commit_data(Db, false).

db_to_header(Db, Header) ->
    Header#db_header{
        update_seq = Db#db.update_seq,
        docinfo_by_seq_btree_state = couch_btree:get_state(Db#db.docinfo_by_seq_btree),
        fulldocinfo_by_id_btree_state = couch_btree:get_state(Db#db.fulldocinfo_by_id_btree),
        local_docs_btree_state = couch_btree:get_state(Db#db.local_docs_btree),
        security_ptr = Db#db.security_ptr,
        revs_limit = Db#db.revs_limit}.

commit_data(#db{waiting_delayed_commit=nil} = Db, true) ->
    Db#db{waiting_delayed_commit=erlang:send_after(1000,self(),delayed_commit)};
commit_data(Db, true) ->
    Db;
commit_data(Db, _) ->
    #db{
        updater_fd = Fd,
        filepath = Filepath,
        header = OldHeader,
        fsync_options = FsyncOptions,
        waiting_delayed_commit = Timer
    } = Db,
    if is_reference(Timer) -> erlang:cancel_timer(Timer); true -> ok end,
    case db_to_header(Db, OldHeader) of
    OldHeader ->
        Db#db{waiting_delayed_commit=nil};
    Header ->
        case lists:member(before_header, FsyncOptions) of
        true -> ok = couch_file:sync(Filepath);
        _    -> ok
        end,

        ok = couch_file:write_header(Fd, Header),

        case lists:member(after_header, FsyncOptions) of
        true -> ok = couch_file:sync(Filepath);
        _    -> ok
        end,

        Db#db{waiting_delayed_commit=nil,
            header=Header,
            committed_update_seq=Db#db.update_seq}
    end.

Db#db.updater_fdのデータファイルに新しいヘッダを書き出し、完了となります。これによって、更新後のデータが見れるようになります。

Conclusion

データ更新処理の残りの部分を見てきました。CouchDBはデータ更新時に、データベースファイルにある既存のデータは書き換えず、追記していくだけになっている、という部分を読むことが、ようやくできました。

本当はこんな風に読んだ際のメモをダラダラと貼り付けることなく、ササっと読んで纏められれば良いのですが、Erlangや関数型言語そのものに慣れていないこともあって、今回はこういう形となりました。

まだ、viewやcompactin、replication等を見ていないので、kp_nodeの意味などはよく理解できていません。このあたりはまた気になった時に見てみたいと思います。