Kyorai
diff --git a/‎ebin/riak_kv.app
+6-2 b/‎ebin/riak_kv.app
+6-2
diff --git a/‎include/riak_kv_map_phase.hrl
+4 b/‎include/riak_kv_map_phase.hrl
+4
diff --git a/‎include/riak_kv_vnode.hrl
+6 b/‎include/riak_kv_vnode.hrl
+6
diff --git a/‎rebar.config
+2-2 b/‎rebar.config
+2-2
diff --git a/‎src/riak_client.erl
+75-13 b/‎src/riak_client.erl
+75-13
diff --git a/‎src/riak_kv_js_manager.erl
+78-52 b/‎src/riak_kv_js_manager.erl
+78-52
@@ -30,10 +30,15 @@
              riak_kv_keys_fsm,
              riak_kv_legacy_vnode,
              riak_kv_lru,
-             riak_kv_map_executor,
+             riak_kv_map_master,
+             riak_kv_mapper,
+             riak_kv_mapper_sup,
              riak_kv_map_localphase,
              riak_kv_map_phase,
+             riak_kv_mapred_cache,
+             riak_kv_mapred_filters,
              riak_kv_mapred_json,
+             riak_kv_mapred_planner,
              riak_kv_mapred_query,
              riak_kv_mapred_term,
              riak_kv_mapreduce,
@@ -65,7 +70,6 @@
                   riak_core,
                   luke,
                   erlang_js,
-                  bitcask,
                   mochiweb,
                   webmachine,
                   os_mon
 
@@ -0,0 +1,4 @@
+-record(riak_kv_map_input, {bkey,
+                            bprops,
+                            kd,
+                            preflist}).
@@ -11,6 +11,11 @@
           bkey :: {binary(), binary()},
           req_id :: non_neg_integer()}).
 
+-record(riak_kv_mget_req_v1, {
+          bkeys :: list({binary(), binary()}),
+          req_id :: non_neg_integer(),
+          from :: term()}).
+
 -record(riak_kv_listkeys_req_v1, {
           bucket :: binary(),
           req_id :: non_neg_integer()}).
@@ -36,6 +41,7 @@
 
 -define(KV_PUT_REQ, #riak_kv_put_req_v1).
 -define(KV_GET_REQ, #riak_kv_get_req_v1).
+-define(KV_MGET_REQ, #riak_kv_mget_req_v1).
 -define(KV_LISTKEYS_REQ, #riak_kv_listkeys_req_v2).
 -define(KV_DELETE_REQ, #riak_kv_delete_req_v1).
 -define(KV_MAP_REQ, #riak_kv_map_req_v1).
 
@@ -9,12 +9,12 @@
                                "HEAD"}},
         {luke, "0.2.2", {git, "git://github.com/basho/luke",
                               "HEAD"}},
-        {erlang_js, "0.4.1", {git, "git://github.com/basho/erlang_js",
+        {erlang_js, "0.5", {git, "git://github.com/kevsmith/erlang_js-jaegermonkey",
                              "HEAD"}},
         {bitcask, "1.1.4", {git, "git://github.com/basho/bitcask",
                                  "HEAD"}},
         {ebloom, "1.0.2", {git, "git://github.com/basho/ebloom",
-                                "HEAD"}},                       
+                                "HEAD"}},
         {eper, "0.60", {git, "git://github.com/dizzyd/eper.git",
                         "eper-0.60"}}
        ]}.
@@ -84,18 +84,37 @@ mapred(Inputs,Query,Timeout) ->
 %%       {error, Err :: term()}
 %% @doc Perform a map/reduce job across the cluster.
 %%      See the map/reduce documentation for explanation of behavior.
-mapred(Inputs,Query,ResultTransformer,Timeout)
-  when is_binary(Inputs) ->
-    mapred_bucket(Inputs, Query, ResultTransformer, Timeout);
+mapred(Inputs,Query,ResultTransformer,Timeout) when is_binary(Inputs) orelse
+                                                    is_tuple(Inputs) ->
+    case is_binary(Inputs) orelse is_key_filter(Inputs) of
+        true ->
+            mapred_bucket(Inputs, Query, ResultTransformer, Timeout);
+        false ->
+            Me = self(),
+            case mapred_stream(Query,Me,ResultTransformer,Timeout) of
+                {ok, {ReqId, FlowPid}} ->
+                    case is_list(Inputs) of
+                        true ->
+                            add_inputs(FlowPid, Inputs);
+                        false ->
+                            mapred_dynamic_inputs_stream(FlowPid, Inputs, Timeout)
+                    end,
+                    luke_flow:finish_inputs(FlowPid),
+                    luke_flow:collect_output(ReqId, Timeout);
+                Error ->
+                    Error
+            end
+    end;
 mapred(Inputs,Query,ResultTransformer,Timeout)
   when is_list(Query),
        (is_integer(Timeout) orelse Timeout =:= infinity) ->
     Me = self(),
     case mapred_stream(Query,Me,ResultTransformer,Timeout) of
         {ok, {ReqId, FlowPid}} ->
-            if is_list(Inputs) ->
-                    luke_flow:add_inputs(FlowPid, Inputs);
-               is_tuple(Inputs) ->
+            case is_list(Inputs) of
+                true ->
+                    add_inputs(FlowPid, Inputs);
+                false ->
                     mapred_dynamic_inputs_stream(FlowPid, Inputs, Timeout)
             end,
             luke_flow:finish_inputs(FlowPid),
@@ -186,7 +205,7 @@ mapred_bucket(Bucket, Query, ResultTransformer, Timeout, ErrorTolerance) ->
 %% review of Map/Reduce.
 mapred_dynamic_inputs_stream(FSMPid, InputDef, Timeout) ->
     case InputDef of
-        {modfun, Mod, Fun, Options} -> 
+        {modfun, Mod, Fun, Options} ->
             Mod:Fun(FSMPid, Options, Timeout);
         _ ->
             throw({invalid_inputdef, InputDef})
@@ -224,7 +243,7 @@ get(Bucket, Key, R) -> get(Bucket, Key, R, ?DEFAULT_TIMEOUT).
 %% @doc Fetch the object at Bucket/Key.  Return a value as soon as R
 %%      nodes have responded with a value or error, or TimeoutMillisecs passes.
 get(Bucket, Key, R, Timeout) when is_binary(Bucket), is_binary(Key),
-                                  (is_atom(R) or is_integer(R)), 
+                                  (is_atom(R) or is_integer(R)),
                                   is_integer(Timeout) ->
     Me = self(),
     ReqId = mk_reqid(),
@@ -237,7 +256,7 @@ get(Bucket, Key, R, Timeout) when is_binary(Bucket), is_binary(Key),
 %%       {error, timeout} |
 %%       {error, {n_val_violation, N::integer()}}
 %% @doc Store RObj in the cluster.
-%%      Return as soon as the default W value number of nodes for this bucket 
+%%      Return as soon as the default W value number of nodes for this bucket
 %%      nodes have received the request.
 %% @equiv put(RObj, W, W, default_timeout())
 put(RObj) -> put(RObj, default, default, ?DEFAULT_TIMEOUT).
@@ -387,11 +406,16 @@ stream_list_keys(Bucket, Timeout, ErrorTolerance, Client) ->
 %%      keys in Bucket on any single vnode.
 %%      If ClientType is set to 'mapred' instead of 'plain', then the
 %%      messages will be sent in the form of a MR input stream.
-stream_list_keys(Bucket, Timeout, ErrorTolerance, Client, ClientType) ->
+stream_list_keys(Bucket0, Timeout, ErrorTolerance, Client, ClientType) ->
     ReqId = mk_reqid(),
-    spawn(Node, riak_kv_keys_fsm, start,
-          [ReqId,Bucket,Timeout,ClientType,ErrorTolerance,Client]),
-    {ok, ReqId}.
+    case build_filter(Bucket0) of
+        {ok, Filter} ->
+            spawn(Node, riak_kv_keys_fsm, start,
+                  [ReqId,Filter,Timeout,ClientType,ErrorTolerance,Client]),
+            {ok, ReqId};
+        Error ->
+            Error
+    end.
 
 %% @spec filter_keys(riak_object:bucket(), Fun :: function()) ->
 %%       {ok, [Key :: riak_object:key()]} |
@@ -508,3 +532,41 @@ wait_for_listkeys(ReqId,Timeout,Acc) ->
     after Timeout ->
             {error, timeout, Acc}
     end.
+
+add_inputs(_FlowPid, []) ->
+    ok;
+add_inputs(FlowPid, Inputs) when length(Inputs) < 100 ->
+    luke_flow:add_inputs(FlowPid, Inputs);
+ add_inputs(FlowPid, Inputs) ->
+    {Current, Next} = lists:split(100, Inputs),
+    luke_flow:add_inputs(FlowPid, Current),
+    add_inputs(FlowPid, Next).
+
+is_key_filter({Bucket, Filters}) when is_binary(Bucket),
+                                      is_list(Filters) ->
+    true;
+is_key_filter(_) ->
+    false.
+
+build_filter({Bucket, Exprs}) ->
+    case build_exprs(Exprs) of
+        {ok, Filters} ->
+            {ok, {Bucket, Filters}};
+        Error ->
+            Error
+    end;
+build_filter(Bucket) when is_binary(Bucket) ->
+    {ok, {Bucket, []}}.
+
+build_exprs(Exprs) ->
+    build_exprs(Exprs, []).
+
+build_exprs([], Accum) ->
+    {ok, lists:reverse(Accum)};
+build_exprs([[FunName|Args]|T], Accum) ->
+    case riak_kv_mapred_filters:resolve_name(FunName) of
+        error ->
+            {error, {bad_filter, FunName}};
+        Fun ->
+            build_exprs(T, [{riak_kv_mapred_filters, Fun, Args}|Accum])
+    end.
@@ -22,19 +22,22 @@
 
 %% @doc dispatch work to JavaScript VMs
 -module(riak_kv_js_manager).
+-author('Kevin Smith <[email protected]>').
+-author('John Muellerleile <[email protected]>').
 
 -behaviour(gen_server).
 
 %% API
--export([start_link/1,
-         add_vm/0,
-         reload/0,
+-export([start_link/2,
+         add_vm/1,
          reload/1,
-         mark_idle/0,
-         reserve_vm/0,
-         dispatch/2,
-         blocking_dispatch/2,
-         pool_size/0]).
+         reload/2,
+         mark_idle/1,
+         reserve_vm/1,
+         reserve_batch_vm/2,
+         dispatch/3,
+         blocking_dispatch/3,
+         pool_size/1]).
 
 %% gen_server callbacks
 -export([init/1,
@@ -44,43 +47,44 @@
          terminate/2,
          code_change/3]).
 
--define(SERVER, ?MODULE).
-
 -record('DOWN', {ref, type, pid, info}).
 -record(vm_state, {pid, needs_reload=false}).
--record(state, {master, idle, reserve}).
+-record(state, {name, master, idle, reserve}).
+
+start_link(Name, ChildCount) ->
+    gen_server:start_link({local, Name}, ?MODULE, [Name, ChildCount], []).
 
-start_link(ChildCount) ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [ChildCount], []).
+reload(Name, []) ->
+    reload(Name).
+reload(Name) ->
+    gen_server:call(Name, reload_vms).
 
-reload([]) ->
-    reload().
-reload() ->
-    gen_server:call(?SERVER, reload_vms).
+add_vm(Name) ->
+    gen_server:cast(Name, {add_vm, self()}).
 
-add_vm() ->
-    gen_server:cast(?SERVER, {add_vm, self()}).
+mark_idle(Name) ->
+    gen_server:call(Name, {mark_idle, self()}).
 
-mark_idle() ->
-    gen_server:call(?SERVER, {mark_idle, self()}).
+dispatch(Name, JSCall, Tries) ->
+    dispatch(Name, JSCall, Tries, Tries).
 
-dispatch(JSCall, Tries) ->
-    dispatch(JSCall, Tries, Tries).
+blocking_dispatch(Name, JSCall, Tries) ->
+    blocking_dispatch(Name, JSCall, Tries, Tries).
 
-blocking_dispatch(JSCall, Tries) ->
-    blocking_dispatch(JSCall, Tries, Tries).
+reserve_vm(Name) ->
+    gen_server:call(Name, reserve_vm).
 
-reserve_vm() ->
-    gen_server:call(?SERVER, reserve_vm).
+reserve_batch_vm(Name, Tries) ->
+    reserve_batch_vm(Name, Tries, Tries).
 
-pool_size() ->
-    gen_server:call(?SERVER, pool_size).
+pool_size(Name) ->
+    gen_server:call(Name, pool_size).
 
-init([ChildCount]) ->
-    Master = ets:new(jsvm_master, [private, {keypos, 2}]),
-    Idle = ets:new(jsvm_idle, [private]),
-    start_vms(ChildCount),
-    {ok, #state{master=Master, idle=Idle}}.
+init([Name, ChildCount]) ->
+    Master = ets:new(Name, [private, {keypos, 2}]),
+    Idle = ets:new(Name, [private]),
+    start_vms(Name, ChildCount),
+    {ok, #state{name=Name, master=Master, idle=Idle}}.
 
 handle_call({mark_idle, VM}, _From, #state{master=Master,
                                            idle=Idle}=State) ->
@@ -100,6 +104,16 @@ handle_call(reload_vms, _From, #state{master=Master, idle=Idle}=State) ->
     riak_kv_vnode:purge_mapcaches(),
     {reply, ok, State};
 
+handle_call(reserve_batch_vm, _From, State) ->
+    {Reply, State1} = case handle_call(reserve_vm, _From, State) of
+                          {reply, {ok, VM}, NewState} ->
+                              riak_kv_js_vm:start_batch(VM),
+                              {{ok, VM}, NewState};
+                          {reply, Error, NewState} ->
+                              {Error, NewState}
+                      end,
+    {reply, Reply, State1};
+
 handle_call(reserve_vm, _From, #state{idle=Idle}=State) ->
     Reply = case ets:first(Idle) of
                 '$end_of_table' ->
@@ -159,11 +173,11 @@ is_vm_idle(Idle, VMPid) ->
             true
     end.
 
-start_vms(0) ->
+start_vms(_Pool, 0) ->
     ok;
-start_vms(Count) ->
-    riak_kv_js_sup:start_js(self()),
-    start_vms(Count - 1).
+start_vms(Pool, Count) ->
+    riak_kv_js_sup:start_js(self(), Pool),
+    start_vms(Pool, Count - 1).
 
 reload_idle_vms(Tid) ->
     reload_idle_vms(ets:first(Tid), Tid).
@@ -190,33 +204,45 @@ mark_pending_reloads(VMPid, Master, Idle) ->
     end,
     mark_pending_reloads(ets:next(Master, VMPid), Master, Idle).
 
-dispatch(_JSCall, _MaxCount, 0) ->
+dispatch(_Name, _JSCall, _MaxCount, 0) ->
     error_logger:info_msg("JS call failed: All VMs are busy.~n"),
     {error, no_vms};
-dispatch(JSCall, MaxCount, Count) ->
-    case reserve_vm() of
+dispatch(Name, JSCall, MaxCount, Count) ->
+    case reserve_vm(Name) of
         {ok, VM} ->
             JobId = {VM, make_ref()},
             riak_kv_js_vm:dispatch(VM, self(), JobId, JSCall),
             {ok, JobId};
         {error, no_vms} ->
-             ScalingFactor = (1 + (MaxCount - Count)) *
-                (0.1 + random:uniform(100) * 0.001),
-            timer:sleep(erlang:round(500 * ScalingFactor)),
-            dispatch(JSCall, MaxCount, Count - 1)
+            back_off(MaxCount, Count),
+            dispatch(Name, JSCall, MaxCount, Count - 1)
     end.
 
-blocking_dispatch(_JSCall, _MaxCount, 0) ->
+blocking_dispatch(_Name, _JSCall, _MaxCount, 0) ->
     error_logger:info_msg("JS call failed: All VMs are busy.~n"),
     {error, no_vms};
-blocking_dispatch(JSCall, MaxCount, Count) ->
-    case reserve_vm() of
+blocking_dispatch(Name, JSCall, MaxCount, Count) ->
+    case reserve_vm(Name) of
         {ok, VM} ->
             JobId = {VM, make_ref()},
             riak_kv_js_vm:blocking_dispatch(VM, JobId, JSCall);
         {error, no_vms} ->
-            ScalingFactor = (1 + (MaxCount - Count)) *
-                (0.1 + random:uniform(100) * 0.001),
-            timer:sleep(erlang:round(500 * ScalingFactor)),
-            blocking_dispatch(JSCall, MaxCount, Count - 1)
+            back_off(MaxCount, Count),
+            blocking_dispatch(Name, JSCall, MaxCount, Count - 1)
+    end.
+
+reserve_batch_vm(_Name, _MaxCount, 0) ->
+    {error, no_vms};
+reserve_batch_vm(Name, MaxCount, Count) ->
+    case gen_server:call(Name, reserve_batch_vm) of
+        {error, no_vms} ->
+            back_off(MaxCount, Count),
+            reserve_batch_vm(Name, MaxCount, Count - 1);
+        {ok, VM} ->
+            {ok, VM}
     end.
+
+back_off(MaxCount, Count) ->
+    ScalingFactor = (1 + (MaxCount - Count)) *
+        (0.1 + random:uniform(100) * 0.001),
+    timer:sleep(erlang:round(500 * ScalingFactor)).