svn commit: r987834 - in /couchdb/branches/new_replicator/src/couchdb: couch_rep_writer.erl couch_replicate.erl couch_view_updater.erl couch_work_queue.erl

[fd...@apache.org]

Author: fdmanana
Date: Sat Aug 21 21:50:06 2010
New Revision: 987834

URL: http://svn.apache.org/viewvc?rev=987834&view=rev
Log:
Merged revision 987824 from trunk:

Small refactoring of the work queue module to accomodate for incoming options (multiple workers)

Modified:
    couchdb/branches/new_replicator/src/couchdb/couch_rep_writer.erl
    couchdb/branches/new_replicator/src/couchdb/couch_replicate.erl
    couchdb/branches/new_replicator/src/couchdb/couch_view_updater.erl
    couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl

Modified: couchdb/branches/new_replicator/src/couchdb/couch_rep_writer.erl
URL: http://svn.apache.org/viewvc/couchdb/branches/new_replicator/src/couchdb/couch_rep_writer.erl?rev=987834&r1=987833&r2=987834&view=diff
==============================================================================
--- couchdb/branches/new_replicator/src/couchdb/couch_rep_writer.erl (original)
+++ couchdb/branches/new_replicator/src/couchdb/couch_rep_writer.erl Sat Aug 21 21:50:06 2010
@@ -97,7 +97,8 @@ write_multi_part_doc(#http_db{headers=He
     {_ContentType, Len} = couch_doc:len_doc_to_multi_part_stream(
         Boundary, JsonBytes, Atts, true
     ),
-    {ok, DataQueue} = couch_work_queue:new(1024*1024, 1000),
+    {ok, DataQueue} = couch_work_queue:new(
+        [{max_size, 1024 * 1024}, {max_items, 1000}]),
     _StreamerPid = spawn_link(
         fun() ->
             couch_doc:doc_to_multi_part_stream(

Modified: couchdb/branches/new_replicator/src/couchdb/couch_replicate.erl
URL: http://svn.apache.org/viewvc/couchdb/branches/new_replicator/src/couchdb/couch_replicate.erl?rev=987834&r1=987833&r2=987834&view=diff
==============================================================================
--- couchdb/branches/new_replicator/src/couchdb/couch_replicate.erl (original)
+++ couchdb/branches/new_replicator/src/couchdb/couch_replicate.erl Sat Aug 21 21:50:06 2010
@@ -176,11 +176,13 @@ do_init([RepId, Src, Tgt, Options, UserC
         start_seq = StartSeq
     } = State = init_state(RepId, Src, Tgt, Options, UserCtx),
 
-    {ok, MissingRevsQueue} = couch_work_queue:new(100000, 500),
+    {ok, MissingRevsQueue} = couch_work_queue:new(
+        [{max_size, 100000}, {max_items, 500}]),
 
     case couch_util:get_value(doc_ids, Options) of
     undefined ->
-        {ok, ChangesQueue} = couch_work_queue:new(100000, 500),
+        {ok, ChangesQueue} = couch_work_queue:new(
+            [{max_size, 100000}, {max_items, 500}]),
 
         % This starts the _changes reader process. It adds the changes from
         % the source db to the ChangesQueue.

Modified: couchdb/branches/new_replicator/src/couchdb/couch_view_updater.erl
URL: http://svn.apache.org/viewvc/couchdb/branches/new_replicator/src/couchdb/couch_view_updater.erl?rev=987834&r1=987833&r2=987834&view=diff
==============================================================================
--- couchdb/branches/new_replicator/src/couchdb/couch_view_updater.erl (original)
+++ couchdb/branches/new_replicator/src/couchdb/couch_view_updater.erl Sat Aug 21 21:50:06 2010
@@ -38,8 +38,10 @@ update(Owner, Group) ->
         couch_task_status:update(<<"Resetting view index due to lost purge entries.">>),
         exit(reset)
     end,
-    {ok, MapQueue} = couch_work_queue:new(100000, 500),
-    {ok, WriteQueue} = couch_work_queue:new(100000, 500),
+    {ok, MapQueue} = couch_work_queue:new(
+        [{max_size, 100000}, {max_items, 500}]),
+    {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),

Modified: couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl
URL: http://svn.apache.org/viewvc/couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl?rev=987834&r1=987833&r2=987834&view=diff
==============================================================================
--- couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl (original)
+++ couchdb/branches/new_replicator/src/couchdb/couch_work_queue.erl Sat Aug 21 21:50:06 2010
@@ -13,7 +13,7 @@
 -module(couch_work_queue).
 -behaviour(gen_server).
 
--export([new/2,queue/2,dequeue/1,dequeue/2,close/1]).
+-export([new/1,queue/2,dequeue/1,dequeue/2,close/1]).
 -export([init/1, terminate/2, handle_call/3, handle_cast/2, code_change/3, handle_info/2]).
 
 -record(q, {
@@ -27,8 +27,8 @@
     close_on_dequeue=false
 }).
 
-new(MaxSize, MaxItems) ->
-    gen_server:start_link(couch_work_queue, {MaxSize, MaxItems}, []).
+new(Options) ->
+    gen_server:start_link(couch_work_queue, Options, []).
 
 queue(Wq, Item) ->
     gen_server:call(Wq, {queue, Item}, infinity).
@@ -46,8 +46,12 @@ close(Wq) ->
     gen_server:cast(Wq, close).
     
 
-init({MaxSize,MaxItems}) ->
-    {ok, #q{max_size=MaxSize, max_items=MaxItems}}.
+init(Options) ->
+    Q = #q{
+        max_size = couch_util:get_value(max_size, Options),
+        max_items = couch_util:get_value(max_items, Options)
+    },
+    {ok, Q}.
 
 terminate(_Reason, #q{work_waiter=nil}) ->
     ok;