couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fdman...@apache.org
Subject svn commit: r1144844 - in /couchdb/trunk/src/couchdb: couch_os_process.erl couch_query_servers.erl couch_view_group.erl couch_view_updater.erl
Date Sun, 10 Jul 2011 12:11:15 GMT
Author: fdmanana
Date: Sun Jul 10 12:11:15 2011
New Revision: 1144844

URL: http://svn.apache.org/viewvc?rev=1144844&view=rev
Log:
Some view indexer optimizations

Changes:

- Decoding of the JSON received from the external OS process
  (couchjs) is now done in the indexer's process which does the
  writes to disk. This avoids several phases of EJSON term
  copying: couch_os_process -> do_maps process -> work queue ->
  do_writes process;

- The indexer's do_maps process no longer needs to find out
  which seq number is the highest from the input list via
  a list comprehension + lists:max/1 call. The highest is
  always the last one on the queue (this is enforced when
  folding the database's byseq btree);

- Don't update the task status everytime we read a single doc
  from the database. Instead update it only after writing
  a view KV batch to disk;

- Avoid several linear passages over the same lists multiple
  times.

This is part of COUCHDB-1186.


Modified:
    couchdb/trunk/src/couchdb/couch_os_process.erl
    couchdb/trunk/src/couchdb/couch_query_servers.erl
    couchdb/trunk/src/couchdb/couch_view_group.erl
    couchdb/trunk/src/couchdb/couch_view_updater.erl

Modified: couchdb/trunk/src/couchdb/couch_os_process.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_os_process.erl?rev=1144844&r1=1144843&r2=1144844&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_os_process.erl (original)
+++ couchdb/trunk/src/couchdb/couch_os_process.erl Sun Jul 10 12:11:15 2011
@@ -85,22 +85,51 @@ writejson(OsProc, Data) when is_record(O
     true = writeline(OsProc, JsonData).
 
 readjson(OsProc) when is_record(OsProc, os_proc) ->
-    Line = readline(OsProc),
+    Line = iolist_to_binary(readline(OsProc)),
     ?LOG_DEBUG("OS Process ~p Output :: ~s", [OsProc#os_proc.port, Line]),
-    case ?JSON_DECODE(Line) of
-    [<<"log">>, Msg] when is_binary(Msg) ->
-        % we got a message to log. Log it and continue
-        ?LOG_INFO("OS Process ~p Log :: ~s", [OsProc#os_proc.port, Msg]),
-        readjson(OsProc);
-    [<<"error">>, Id, Reason] ->
-        throw({error, {couch_util:to_existing_atom(Id),Reason}});
-    [<<"fatal">>, Id, Reason] ->
-        ?LOG_INFO("OS Process ~p Fatal Error :: ~s ~p",[OsProc#os_proc.port, Id, Reason]),
-        throw({couch_util:to_existing_atom(Id),Reason});
-    Result ->
-        Result
+    try
+        % Don't actually parse the whole JSON. Just try to see if it's
+        % a command or a doc map/reduce/filter/show/list/update output.
+        % If it's a command then parse the whole JSON and execute the
+        % command, otherwise return the raw JSON line to the caller.
+        pick_command(Line)
+    catch
+    throw:abort ->
+        {json, Line};
+    throw:{cmd, _Cmd} ->
+        case ?JSON_DECODE(Line) of
+        [<<"log">>, Msg] when is_binary(Msg) ->
+            % we got a message to log. Log it and continue
+            ?LOG_INFO("OS Process ~p Log :: ~s", [OsProc#os_proc.port, Msg]),
+            readjson(OsProc);
+        [<<"error">>, Id, Reason] ->
+            throw({error, {couch_util:to_existing_atom(Id),Reason}});
+        [<<"fatal">>, Id, Reason] ->
+            ?LOG_INFO("OS Process ~p Fatal Error :: ~s ~p",
+                [OsProc#os_proc.port, Id, Reason]),
+            throw({couch_util:to_existing_atom(Id),Reason});
+        _Result ->
+            {json, Line}
+        end
     end.
 
+pick_command(Line) ->
+    json_stream_parse:events(Line, fun pick_command0/1).
+
+pick_command0(array_start) ->
+    fun pick_command1/1;
+pick_command0(_) ->
+    throw(abort).
+
+pick_command1(<<"log">> = Cmd) ->
+    throw({cmd, Cmd});
+pick_command1(<<"error">> = Cmd) ->
+    throw({cmd, Cmd});
+pick_command1(<<"fatal">> = Cmd) ->
+    throw({cmd, Cmd});
+pick_command1(_) ->
+    throw(abort).
+
 
 % gen_server API
 init([Command, Options, PortOptions]) ->

Modified: couchdb/trunk/src/couchdb/couch_query_servers.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_query_servers.erl?rev=1144844&r1=1144843&r2=1144844&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_query_servers.erl (original)
+++ couchdb/trunk/src/couchdb/couch_query_servers.erl Sun Jul 10 12:11:15 2011
@@ -16,7 +16,7 @@
 -export([start_link/0, config_change/1]).
 
 -export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3]).
--export([start_doc_map/3, map_docs/2, stop_doc_map/1]).
+-export([start_doc_map/3, map_docs/2, map_doc_raw/2, stop_doc_map/1, raw_to_ejson/1]).
 -export([reduce/3, rereduce/3,validate_doc_update/5]).
 -export([filter_docs/5]).
 -export([filter_view/3]).
@@ -82,6 +82,10 @@ map_docs(Proc, Docs) ->
         Docs),
     {ok, Results}.
 
+map_doc_raw(Proc, Doc) ->
+    Json = couch_doc:to_json_obj(Doc, []),
+    {ok, proc_prompt_raw(Proc, [<<"map_doc">>, Json])}.
+
 
 stop_doc_map(nil) ->
     ok;
@@ -494,9 +498,21 @@ proc_with_ddoc(DDoc, DDocKey, LangProcs)
     end.
 
 proc_prompt(Proc, Args) ->
-    {Mod, Func} = Proc#proc.prompt_fun,
+     case proc_prompt_raw(Proc, Args) of
+     {json, Json} ->
+         ?JSON_DECODE(Json);
+     EJson ->
+         EJson
+     end.
+
+proc_prompt_raw(#proc{prompt_fun = {Mod, Func}} = Proc, Args) ->
     apply(Mod, Func, [Proc#proc.pid, Args]).
 
+raw_to_ejson({json, Json}) ->
+    ?JSON_DECODE(Json);
+raw_to_ejson(EJson) ->
+    EJson.
+
 proc_stop(Proc) ->
     {Mod, Func} = Proc#proc.stop_fun,
     apply(Mod, Func, [Proc#proc.pid]).

Modified: couchdb/trunk/src/couchdb/couch_view_group.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_view_group.erl?rev=1144844&r1=1144843&r2=1144844&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_view_group.erl (original)
+++ couchdb/trunk/src/couchdb/couch_view_group.erl Sun Jul 10 12:11:15 2011
@@ -236,16 +236,22 @@ handle_cast({partial_update, Pid, NewGro
         = State) ->
     #group_state{
         db_name = DbName,
-        waiting_commit = WaitingCommit
+        waiting_commit = WaitingCommit,
+        group = Group
     } = State,
     NewSeq = NewGroup#group.current_seq,
-    ?LOG_INFO("checkpointing view update at seq ~p for ~s ~s", [NewSeq,
-        DbName, NewGroup#group.name]),
-    if not WaitingCommit ->
-        erlang:send_after(1000, self(), delayed_commit);
-    true -> ok
-    end,
-    {noreply, State#group_state{group=NewGroup, waiting_commit=true}};
+    case NewSeq > Group#group.current_seq of
+    true ->
+        ?LOG_INFO("checkpointing view update at seq ~p for ~s ~s", [NewSeq,
+            DbName, NewGroup#group.name]),
+        if not WaitingCommit ->
+            erlang:send_after(1000, self(), delayed_commit);
+        true -> ok
+        end,
+        {noreply, State#group_state{group=NewGroup, waiting_commit=true}};
+    false ->
+        {noreply, State}
+    end;
 handle_cast({partial_update, _, _}, State) ->
     %% message from an old (probably pre-compaction) updater; ignore
     {noreply, State}.

Modified: couchdb/trunk/src/couchdb/couch_view_updater.erl
URL: http://svn.apache.org/viewvc/couchdb/trunk/src/couchdb/couch_view_updater.erl?rev=1144844&r1=1144843&r2=1144844&view=diff
==============================================================================
--- couchdb/trunk/src/couchdb/couch_view_updater.erl (original)
+++ couchdb/trunk/src/couchdb/couch_view_updater.erl Sun Jul 10 12:11:15 2011
@@ -24,19 +24,14 @@ update(Owner, Group, DbName) ->
         current_seq = Seq,
         purge_seq = PurgeSeq
     } = Group,
-    couch_task_status:add_task(<<"View Group Indexer">>, <<DbName/binary,"
",GroupName/binary>>, <<"Starting index update">>),
 
     {ok, Db} = couch_db:open_int(DbName, []),
     DbPurgeSeq = couch_db:get_purge_seq(Db),
-    Group2 =
     if DbPurgeSeq == PurgeSeq ->
-        Group;
+        ok;
     DbPurgeSeq == PurgeSeq + 1 ->
-        couch_task_status:update(<<"Removing purged entries from view index.">>),
-        purge_index(Group, Db);
+        ok;
     true ->
-        couch_task_status:update(<<"Resetting view index due to lost purge entries.">>),
-        couch_db:close(Db),
         exit(reset)
     end,
     {ok, MapQueue} = couch_work_queue:new(
@@ -44,13 +39,30 @@ update(Owner, Group, DbName) ->
     {ok, WriteQueue} = couch_work_queue:new(
         [{max_size, 100000}, {max_items, 500}]),
     Self = self(),
-    ViewEmptyKVs = [{View, []} || View <- Group2#group.views],
-    spawn_link(fun() -> do_maps(Group, MapQueue, WriteQueue, ViewEmptyKVs) end),
-    spawn_link(fun() -> do_writes(Self, Owner, Group2, WriteQueue, Seq == 0) end),
-    % compute on all docs modified since we last computed.
+    spawn_link(fun() ->
+        do_maps(add_query_server(Group), MapQueue, WriteQueue)
+    end),
     TotalChanges = couch_db:count_changes_since(Db, Seq),
-    % update status every half second
-    couch_task_status:set_update_frequency(500),
+    spawn_link(fun() ->
+        couch_task_status:add_task(
+            <<"View Group Indexer">>,
+            <<DbName/binary, " ", GroupName/binary>>,
+            <<"Starting index update">>),
+        couch_task_status:set_update_frequency(500),
+        Group2 =
+        if DbPurgeSeq == PurgeSeq + 1 ->
+            couch_task_status:update(<<"Removing purged entries from view index.">>),
+            purge_index(Group, Db);
+        true ->
+            Group
+        end,
+        ViewEmptyKVs = [{View, []} || View <- Group2#group.views],
+        do_writes(Self, Owner, Group2, WriteQueue,
+            Seq == 0, ViewEmptyKVs, 0, TotalChanges),
+        couch_task_status:set_update_frequency(0),
+        couch_task_status:update("Finishing.")
+    end),
+    % compute on all docs modified since we last computed.
     #group{ design_options = DesignOptions } = Group,
     IncludeDesign = couch_util:get_value(<<"include_design">>,
         DesignOptions, false),
@@ -64,15 +76,11 @@ update(Owner, Group, DbName) ->
         = couch_db:enum_docs_since(
             Db,
             Seq,
-            fun(DocInfo, _, ChangesProcessed) ->
-                couch_task_status:update("Processed ~p of ~p changes (~p%)",
-                        [ChangesProcessed, TotalChanges, (ChangesProcessed*100) div TotalChanges]),
+            fun(DocInfo, _, Acc) ->
                 load_doc(Db, DocInfo, MapQueue, DocOpts, IncludeDesign),
-                {ok, ChangesProcessed+1}
+                {ok, Acc}
             end,
-            0, []),
-    couch_task_status:set_update_frequency(0),
-    couch_task_status:update("Finishing."),
+            ok, []),
     couch_work_queue:close(MapQueue),
     couch_db:close(Db),
     receive {new_group, NewGroup} ->
@@ -81,6 +89,16 @@ update(Owner, Group, DbName) ->
     end.
 
 
+add_query_server(#group{query_server = nil} = Group) ->
+    {ok, Qs} = couch_query_servers:start_doc_map(
+        Group#group.def_lang,
+        [View#view.def || View <- Group#group.views],
+        Group#group.lib),
+    Group#group{query_server = Qs};
+add_query_server(Group) ->
+    Group.
+
+
 purge_index(#group{views=Views, id_btree=IdBtree}=Group, Db) ->
     {ok, PurgedIdsRevs} = couch_db:get_last_purged(Db),
     Ids = [Id || {Id, _Revs} <- PurgedIdsRevs],
@@ -133,96 +151,83 @@ load_doc(Db, DocInfo, MapQueue, DocOpts,
         end
     end.
     
-do_maps(Group, MapQueue, WriteQueue, ViewEmptyKVs) ->
+do_maps(#group{query_server = Qs} = Group, MapQueue, WriteQueue) ->
     case couch_work_queue:dequeue(MapQueue) of
     closed ->
         couch_work_queue:close(WriteQueue),
         couch_query_servers:stop_doc_map(Group#group.query_server);
     {ok, Queue} ->
-        Docs = [Doc || {_,#doc{deleted=false}=Doc} <- Queue],
-        DelKVs = [{Id, []} || {_, #doc{deleted=true,id=Id}} <- Queue],
-        LastSeq = lists:max([Seq || {Seq, _Doc} <- Queue]),
-        {Group1, Results} = view_compute(Group, Docs),
-        {ViewKVs, DocIdViewIdKeys} = view_insert_query_results(Docs,
-                    Results, ViewEmptyKVs, DelKVs),
-        couch_work_queue:queue(WriteQueue, {LastSeq, ViewKVs, DocIdViewIdKeys}),
-        do_maps(Group1, MapQueue, WriteQueue, ViewEmptyKVs)
+        lists:foreach(
+            fun({Seq, #doc{id = Id, deleted = true}}) ->
+                Item = {Seq, Id, []},
+                ok = couch_work_queue:queue(WriteQueue, Item);
+            ({Seq, #doc{id = Id, deleted = false} = Doc}) ->
+                {ok, Result} = couch_query_servers:map_doc_raw(Qs, Doc),
+                Item = {Seq, Id, Result},
+                ok = couch_work_queue:queue(WriteQueue, Item)
+            end,
+            Queue),
+        do_maps(Group, MapQueue, WriteQueue)
     end.
 
-do_writes(Parent, Owner, Group, WriteQueue, InitialBuild) ->
+do_writes(Parent, Owner, Group, WriteQueue, InitialBuild, ViewEmptyKVs,
+        ChangesDone, TotalChanges) ->
     case couch_work_queue:dequeue(WriteQueue) of
     closed ->
         Parent ! {new_group, Group};
     {ok, Queue} ->
-        {NewSeq, ViewKeyValues, DocIdViewIdKeys} = lists:foldl(
-            fun({Seq, ViewKVs, DocIdViewIdKeys}, nil) ->
-                {Seq, ViewKVs, DocIdViewIdKeys};
-            ({Seq, ViewKVs, DocIdViewIdKeys}, Acc) ->
-                {Seq2, AccViewKVs, AccDocIdViewIdKeys} = Acc,
-                AccViewKVs2 = lists:zipwith(
-                    fun({View, KVsIn}, {_View, KVsAcc}) ->
-                        {View, KVsIn ++ KVsAcc}
-                    end, ViewKVs, AccViewKVs),
-                {lists:max([Seq, Seq2]),
-                        AccViewKVs2, DocIdViewIdKeys ++ AccDocIdViewIdKeys}
-            end, nil, Queue),
-        Group2 = write_changes(Group, ViewKeyValues, DocIdViewIdKeys, NewSeq,
-                InitialBuild),
+        {ViewKVs, DocIdViewIdKeys} = lists:foldr(
+            fun({_Seq, Id, []}, {ViewKVsAcc, DocIdViewIdKeysAcc}) ->
+                {ViewKVsAcc, [{Id, []} | DocIdViewIdKeysAcc]};
+            ({_Seq, Id, RawQueryResults}, {ViewKVsAcc, DocIdViewIdKeysAcc}) ->
+                QueryResults = [
+                    [list_to_tuple(FunResult) || FunResult <- FunRs] || FunRs <-
+                        couch_query_servers:raw_to_ejson(RawQueryResults)
+                ],
+                {NewViewKVs, NewViewIdKeys} = view_insert_doc_query_results(
+                        Id, QueryResults, ViewKVsAcc, [], []),
+                {NewViewKVs, [{Id, NewViewIdKeys} | DocIdViewIdKeysAcc]}
+            end,
+            {ViewEmptyKVs, []}, Queue),
+        {NewSeq, _, _} = lists:last(Queue),
+        Group2 = write_changes(
+            Group, ViewKVs, DocIdViewIdKeys, NewSeq, InitialBuild),
         case Owner of
-        nil -> ok;
-        _ -> ok = gen_server:cast(Owner, {partial_update, Parent, Group2})
+        nil ->
+            ok;
+        _ ->
+            ok = gen_server:cast(Owner, {partial_update, Parent, Group2})
         end,
-        do_writes(Parent, Owner, Group2, WriteQueue, InitialBuild)
+        ChangesDone2 = ChangesDone + length(Queue),
+        couch_task_status:update("Processed ~p of ~p changes (~p%)",
+              [ChangesDone2, TotalChanges, (ChangesDone2 * 100) div TotalChanges]),
+        do_writes(Parent, Owner, Group2, WriteQueue, InitialBuild, ViewEmptyKVs,
+            ChangesDone2, TotalChanges)
     end.
 
-view_insert_query_results([], [], ViewKVs, DocIdViewIdKeysAcc) ->
-    {ViewKVs, DocIdViewIdKeysAcc};
-view_insert_query_results([Doc|RestDocs], [QueryResults | RestResults], ViewKVs, DocIdViewIdKeysAcc)
->
-    {NewViewKVs, NewViewIdKeys} = view_insert_doc_query_results(Doc, QueryResults, ViewKVs,
[], []),
-    NewDocIdViewIdKeys = [{Doc#doc.id, NewViewIdKeys} | DocIdViewIdKeysAcc],
-    view_insert_query_results(RestDocs, RestResults, NewViewKVs, NewDocIdViewIdKeys).
 
-
-view_insert_doc_query_results(_Doc, [], [], ViewKVsAcc, ViewIdKeysAcc) ->
+view_insert_doc_query_results(_DocId, [], [], ViewKVsAcc, ViewIdKeysAcc) ->
     {lists:reverse(ViewKVsAcc), lists:reverse(ViewIdKeysAcc)};
-view_insert_doc_query_results(#doc{id=DocId}=Doc, [ResultKVs|RestResults], [{View, KVs}|RestViewKVs],
ViewKVsAcc, ViewIdKeysAcc) ->
+view_insert_doc_query_results(DocId, [ResultKVs | RestResults],
+        [{View, KVs} | RestViewKVs], ViewKVsAcc, ViewIdKeysAcc) ->
     % Take any identical keys and combine the values
-    ResultKVs2 = lists:foldl(
-        fun({Key,Value}, [{PrevKey,PrevVal}|AccRest]) ->
-            case Key == PrevKey of
-            true ->
-                case PrevVal of
-                {dups, Dups} ->
-                    [{PrevKey, {dups, [Value|Dups]}} | AccRest];
-                _ ->
-                    [{PrevKey, {dups, [Value,PrevVal]}} | AccRest]
-                end;
-            false ->
-                [{Key,Value},{PrevKey,PrevVal}|AccRest]
-            end;
-        (KV, []) ->
-           [KV]
-        end, [], lists:sort(ResultKVs)),
-    NewKVs = [{{Key, DocId}, Value} || {Key, Value} <- ResultKVs2],
+    {NewKVs, NewViewIdKeys} = lists:foldl(
+        fun({Key, Val}, {[{{Key, _DocId} = Kd, PrevVal} | AccRest], AccVid}) ->
+            AccKv2 = case PrevVal of
+            {dups, Dups} ->
+                [{Kd, {dups, [Val | Dups]}} | AccRest];
+            _ ->
+                [{Kd, {dups, [Val, PrevVal]}} | AccRest]
+            end,
+            {AccKv2, [{View#view.id_num, Key} | AccVid]};
+        ({Key, Val}, {AccKv, AccVid}) ->
+            {[{{Key, DocId}, Val} | AccKv], [{View#view.id_num, Key} | AccVid]}
+        end,
+        {[], []}, lists:sort(ResultKVs)),
     NewViewKVsAcc = [{View, NewKVs ++ KVs} | ViewKVsAcc],
-    NewViewIdKeys = [{View#view.id_num, Key} || {Key, _Value} <- ResultKVs2],
     NewViewIdKeysAcc = NewViewIdKeys ++ ViewIdKeysAcc,
-    view_insert_doc_query_results(Doc, RestResults, RestViewKVs, NewViewKVsAcc, NewViewIdKeysAcc).
-
-view_compute(Group, []) ->
-    {Group, []};
-view_compute(#group{def_lang=DefLang, lib=Lib, query_server=QueryServerIn}=Group, Docs) ->
-    {ok, QueryServer} =
-    case QueryServerIn of
-    nil -> % doc map not started
-        Definitions = [View#view.def || View <- Group#group.views],
-        couch_query_servers:start_doc_map(DefLang, Definitions, Lib);
-    _ ->
-        {ok, QueryServerIn}
-    end,
-    {ok, Results} = couch_query_servers:map_docs(QueryServer, Docs),
-    {Group#group{query_server=QueryServer}, Results}.
-
+    view_insert_doc_query_results(
+        DocId, RestResults, RestViewKVs, NewViewKVsAcc, NewViewIdKeysAcc).
 
 
 write_changes(Group, ViewKeyValuesToAdd, DocIdViewIdKeys, NewSeq, InitialBuild) ->



Mime
View raw message