couchdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From beno...@apache.org
Subject [49/57] [abbrv] [partial] inital move to rebar compilation
Date Tue, 07 Jan 2014 00:37:09 GMT
http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch.erl b/apps/couch/src/couch.erl
new file mode 100644
index 0000000..80e3261
--- /dev/null
+++ b/apps/couch/src/couch.erl
@@ -0,0 +1,58 @@
+% 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).
+
+-export([get_app_env/2,
+         version/0,
+         start/0,
+         stop/0,
+         restart/0,
+         reload/0]).
+
+get_app_env(Env, Default) ->
+    case application:get_env(couch, Env) of
+        {ok, Val} -> Val;
+        undefined -> Default
+    end.
+
+version() ->
+    case application:get_key(couch, vsn) of
+        {ok, FullVersion} ->
+            hd(string:tokens(FullVersion, "-"));
+        _ ->
+            "0.0.0"
+    end.
+
+start() ->
+    application:start(couch).
+
+stop() ->
+    application:stop(couch).
+
+restart() ->
+    case stop() of
+    ok ->
+        start();
+    {error, {not_started,couch}} ->
+        start();
+    {error, Reason} ->
+        {error, Reason}
+    end.
+
+reload() ->
+    case supervisor:terminate_child(couch_server_sup, couch_config) of
+    ok ->
+        supervisor:restart_child(couch_server_sup, couch_config);
+    {error, Reason} ->
+        {error, Reason}
+    end.

http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch_app.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_app.erl b/apps/couch/src/couch_app.erl
new file mode 100644
index 0000000..a8d215e
--- /dev/null
+++ b/apps/couch/src/couch_app.erl
@@ -0,0 +1,36 @@
+% 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_app).
+
+-behaviour(application).
+
+-include("couch_db.hrl").
+
+-export([start/2, stop/1]).
+
+-define(CONF_FILES, ["couch.ini", "couch_httpd.ini", "local.ini"]).
+
+start(_Type, _Args) ->
+    couch_util:start_app_deps(couch),
+    IniFiles = get_ini_files(),
+    couch_server_sup:start_link(IniFiles).
+
+stop(_) ->
+    ok.
+
+get_ini_files() ->
+    DefaultConfDir =  filename:join([code:root_dir(), "./etc"]),
+    Defaults = lists:map(fun(FName) ->
+                    filename:join(DefaultConfDir, FName)
+            end, ?CONF_FILES),
+    couch:get_app_env(config_files, Defaults).

http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch_auth_cache.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_auth_cache.erl b/apps/couch/src/couch_auth_cache.erl
new file mode 100644
index 0000000..42ccd44
--- /dev/null
+++ b/apps/couch/src/couch_auth_cache.erl
@@ -0,0 +1,425 @@
+% 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_auth_cache).
+-behaviour(gen_server).
+
+% public API
+-export([get_user_creds/1]).
+
+% gen_server API
+-export([start_link/0, init/1, handle_call/3, handle_info/2, handle_cast/2]).
+-export([code_change/3, terminate/2]).
+
+-include("couch_db.hrl").
+-include("couch_js_functions.hrl").
+
+-define(STATE, auth_state_ets).
+-define(BY_USER, auth_by_user_ets).
+-define(BY_ATIME, auth_by_atime_ets).
+
+-record(state, {
+    max_cache_size = 0,
+    cache_size = 0,
+    db_notifier = nil,
+    db_mon_ref = nil
+}).
+
+
+-spec get_user_creds(UserName::string() | binary()) ->
+    Credentials::list() | nil.
+
+get_user_creds(UserName) when is_list(UserName) ->
+    get_user_creds(?l2b(UserName));
+
+get_user_creds(UserName) ->
+    UserCreds = case couch_config:get("admins", ?b2l(UserName)) of
+    "-hashed-" ++ HashedPwdAndSalt ->
+        % the name is an admin, now check to see if there is a user doc
+        % which has a matching name, salt, and password_sha
+        [HashedPwd, Salt] = string:tokens(HashedPwdAndSalt, ","),
+        case get_from_cache(UserName) of
+        nil ->
+            make_admin_doc(HashedPwd, Salt, []);
+        UserProps when is_list(UserProps) ->
+            make_admin_doc(HashedPwd, Salt, couch_util:get_value(<<"roles">>, UserProps))
+        end;
+    "-pbkdf2-" ++ HashedPwdSaltAndIterations ->
+        [HashedPwd, Salt, Iterations] = string:tokens(HashedPwdSaltAndIterations, ","),
+        case get_from_cache(UserName) of
+        nil ->
+            make_admin_doc(HashedPwd, Salt, Iterations, []);
+        UserProps when is_list(UserProps) ->
+            make_admin_doc(HashedPwd, Salt, Iterations, couch_util:get_value(<<"roles">>, UserProps))
+    end;
+    _Else ->
+        get_from_cache(UserName)
+    end,
+    validate_user_creds(UserCreds).
+
+make_admin_doc(HashedPwd, Salt, ExtraRoles) ->
+    [{<<"roles">>, [<<"_admin">>|ExtraRoles]},
+     {<<"salt">>, ?l2b(Salt)},
+     {<<"password_scheme">>, <<"simple">>},
+     {<<"password_sha">>, ?l2b(HashedPwd)}].
+
+make_admin_doc(DerivedKey, Salt, Iterations, ExtraRoles) ->
+    [{<<"roles">>, [<<"_admin">>|ExtraRoles]},
+     {<<"salt">>, ?l2b(Salt)},
+     {<<"iterations">>, list_to_integer(Iterations)},
+     {<<"password_scheme">>, <<"pbkdf2">>},
+     {<<"derived_key">>, ?l2b(DerivedKey)}].
+
+get_from_cache(UserName) ->
+    exec_if_auth_db(
+        fun(_AuthDb) ->
+            maybe_refresh_cache(),
+            case ets:lookup(?BY_USER, UserName) of
+            [] ->
+                gen_server:call(?MODULE, {fetch, UserName}, infinity);
+            [{UserName, {Credentials, _ATime}}] ->
+                couch_stats_collector:increment({couchdb, auth_cache_hits}),
+                gen_server:cast(?MODULE, {cache_hit, UserName}),
+                Credentials
+            end
+        end,
+        nil
+    ).
+
+
+validate_user_creds(nil) ->
+    nil;
+validate_user_creds(UserCreds) ->
+    case couch_util:get_value(<<"_conflicts">>, UserCreds) of
+    undefined ->
+        ok;
+    _ConflictList ->
+        throw({unauthorized,
+            <<"User document conflicts must be resolved before the document",
+              " is used for authentication purposes.">>
+        })
+    end,
+    UserCreds.
+
+
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+init(_) ->
+    ?STATE = ets:new(?STATE, [set, protected, named_table]),
+    ?BY_USER = ets:new(?BY_USER, [set, protected, named_table]),
+    ?BY_ATIME = ets:new(?BY_ATIME, [ordered_set, private, named_table]),
+    process_flag(trap_exit, true),
+    ok = couch_config:register(
+        fun("couch_httpd_auth", "auth_cache_size", SizeList) ->
+            Size = list_to_integer(SizeList),
+            ok = gen_server:call(?MODULE, {new_max_cache_size, Size}, infinity);
+        ("couch_httpd_auth", "authentication_db", _DbName) ->
+            ok = gen_server:call(?MODULE, reinit_cache, infinity)
+        end
+    ),
+    {ok, Notifier} = couch_db_update_notifier:start_link(fun handle_db_event/1),
+    State = #state{
+        db_notifier = Notifier,
+        max_cache_size = list_to_integer(
+            couch_config:get("couch_httpd_auth", "auth_cache_size", "50")
+        )
+    },
+    {ok, reinit_cache(State)}.
+
+
+handle_db_event({Event, DbName}) ->
+    [{auth_db_name, AuthDbName}] = ets:lookup(?STATE, auth_db_name),
+    case DbName =:= AuthDbName of
+    true ->
+        case Event of
+        created -> gen_server:call(?MODULE, reinit_cache, infinity);
+        compacted -> gen_server:call(?MODULE, auth_db_compacted, infinity);
+        _Else   -> ok
+        end;
+    false ->
+        ok
+    end.
+
+
+handle_call(reinit_cache, _From, State) ->
+    catch erlang:demonitor(State#state.db_mon_ref, [flush]),
+    exec_if_auth_db(fun(AuthDb) -> catch couch_db:close(AuthDb) end),
+    {reply, ok, reinit_cache(State)};
+
+handle_call(auth_db_compacted, _From, State) ->
+    exec_if_auth_db(
+        fun(AuthDb) ->
+            true = ets:insert(?STATE, {auth_db, reopen_auth_db(AuthDb)})
+        end
+    ),
+    {reply, ok, State};
+
+handle_call({new_max_cache_size, NewSize},
+        _From, #state{cache_size = Size} = State) when NewSize >= Size ->
+    {reply, ok, State#state{max_cache_size = NewSize}};
+
+handle_call({new_max_cache_size, NewSize}, _From, State) ->
+    free_mru_cache_entries(State#state.cache_size - NewSize),
+    {reply, ok, State#state{max_cache_size = NewSize, cache_size = NewSize}};
+
+handle_call({fetch, UserName}, _From, State) ->
+    {Credentials, NewState} = case ets:lookup(?BY_USER, UserName) of
+    [{UserName, {Creds, ATime}}] ->
+        couch_stats_collector:increment({couchdb, auth_cache_hits}),
+        cache_hit(UserName, Creds, ATime),
+        {Creds, State};
+    [] ->
+        couch_stats_collector:increment({couchdb, auth_cache_misses}),
+        Creds = get_user_props_from_db(UserName),
+        State1 = add_cache_entry(UserName, Creds, erlang:now(), State),
+        {Creds, State1}
+    end,
+    {reply, Credentials, NewState};
+
+handle_call(refresh, _From, State) ->
+    exec_if_auth_db(fun refresh_entries/1),
+    {reply, ok, State}.
+
+
+handle_cast({cache_hit, UserName}, State) ->
+    case ets:lookup(?BY_USER, UserName) of
+    [{UserName, {Credentials, ATime}}] ->
+        cache_hit(UserName, Credentials, ATime);
+    _ ->
+        ok
+    end,
+    {noreply, State}.
+
+
+handle_info({'DOWN', Ref, _, _, _Reason}, #state{db_mon_ref = Ref} = State) ->
+    {noreply, reinit_cache(State)}.
+
+
+terminate(_Reason, #state{db_notifier = Notifier}) ->
+    couch_db_update_notifier:stop(Notifier),
+    exec_if_auth_db(fun(AuthDb) -> catch couch_db:close(AuthDb) end),
+    true = ets:delete(?BY_USER),
+    true = ets:delete(?BY_ATIME),
+    true = ets:delete(?STATE).
+
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+
+clear_cache(State) ->
+    exec_if_auth_db(fun(AuthDb) -> catch couch_db:close(AuthDb) end),
+    true = ets:delete_all_objects(?BY_USER),
+    true = ets:delete_all_objects(?BY_ATIME),
+    State#state{cache_size = 0}.
+
+
+reinit_cache(State) ->
+    NewState = clear_cache(State),
+    AuthDbName = ?l2b(couch_config:get("couch_httpd_auth", "authentication_db")),
+    true = ets:insert(?STATE, {auth_db_name, AuthDbName}),
+    AuthDb = open_auth_db(),
+    true = ets:insert(?STATE, {auth_db, AuthDb}),
+    NewState#state{db_mon_ref = couch_db:monitor(AuthDb)}.
+
+
+add_cache_entry(_, _, _, #state{max_cache_size = 0} = State) ->
+    State;
+add_cache_entry(UserName, Credentials, ATime, State) ->
+    case State#state.cache_size >= State#state.max_cache_size of
+    true ->
+        free_mru_cache_entry();
+    false ->
+        ok
+    end,
+    true = ets:insert(?BY_ATIME, {ATime, UserName}),
+    true = ets:insert(?BY_USER, {UserName, {Credentials, ATime}}),
+    State#state{cache_size = couch_util:get_value(size, ets:info(?BY_USER))}.
+
+free_mru_cache_entries(0) ->
+    ok;
+free_mru_cache_entries(N) when N > 0 ->
+    free_mru_cache_entry(),
+    free_mru_cache_entries(N - 1).
+
+free_mru_cache_entry() ->
+    MruTime = ets:last(?BY_ATIME),
+    [{MruTime, UserName}] = ets:lookup(?BY_ATIME, MruTime),
+    true = ets:delete(?BY_ATIME, MruTime),
+    true = ets:delete(?BY_USER, UserName).
+
+
+cache_hit(UserName, Credentials, ATime) ->
+    NewATime = erlang:now(),
+    true = ets:delete(?BY_ATIME, ATime),
+    true = ets:insert(?BY_ATIME, {NewATime, UserName}),
+    true = ets:insert(?BY_USER, {UserName, {Credentials, NewATime}}).
+
+
+refresh_entries(AuthDb) ->
+    case reopen_auth_db(AuthDb) of
+    nil ->
+        ok;
+    AuthDb2 ->
+        case AuthDb2#db.update_seq > AuthDb#db.update_seq of
+        true ->
+            {ok, _, _} = couch_db:enum_docs_since(
+                AuthDb2,
+                AuthDb#db.update_seq,
+                fun(DocInfo, _, _) -> refresh_entry(AuthDb2, DocInfo) end,
+                AuthDb#db.update_seq,
+                []
+            ),
+            true = ets:insert(?STATE, {auth_db, AuthDb2});
+        false ->
+            ok
+        end
+    end.
+
+
+refresh_entry(Db, #doc_info{high_seq = DocSeq} = DocInfo) ->
+    case is_user_doc(DocInfo) of
+    {true, UserName} ->
+        case ets:lookup(?BY_USER, UserName) of
+        [] ->
+            ok;
+        [{UserName, {_OldCreds, ATime}}] ->
+            {ok, Doc} = couch_db:open_doc(Db, DocInfo, [conflicts, deleted]),
+            NewCreds = user_creds(Doc),
+            true = ets:insert(?BY_USER, {UserName, {NewCreds, ATime}})
+        end;
+    false ->
+        ok
+    end,
+    {ok, DocSeq}.
+
+
+user_creds(#doc{deleted = true}) ->
+    nil;
+user_creds(#doc{} = Doc) ->
+    {Creds} = couch_doc:to_json_obj(Doc, []),
+    Creds.
+
+
+is_user_doc(#doc_info{id = <<"org.couchdb.user:", UserName/binary>>}) ->
+    {true, UserName};
+is_user_doc(_) ->
+    false.
+
+
+maybe_refresh_cache() ->
+    case cache_needs_refresh() of
+    true ->
+        ok = gen_server:call(?MODULE, refresh, infinity);
+    false ->
+        ok
+    end.
+
+
+cache_needs_refresh() ->
+    exec_if_auth_db(
+        fun(AuthDb) ->
+            case reopen_auth_db(AuthDb) of
+            nil ->
+                false;
+            AuthDb2 ->
+                AuthDb2#db.update_seq > AuthDb#db.update_seq
+            end
+        end,
+        false
+    ).
+
+
+reopen_auth_db(AuthDb) ->
+    case (catch couch_db:reopen(AuthDb)) of
+    {ok, AuthDb2} ->
+        AuthDb2;
+    _ ->
+        nil
+    end.
+
+
+exec_if_auth_db(Fun) ->
+    exec_if_auth_db(Fun, ok).
+
+exec_if_auth_db(Fun, DefRes) ->
+    case ets:lookup(?STATE, auth_db) of
+    [{auth_db, #db{} = AuthDb}] ->
+        Fun(AuthDb);
+    _ ->
+        DefRes
+    end.
+
+
+open_auth_db() ->
+    [{auth_db_name, DbName}] = ets:lookup(?STATE, auth_db_name),
+    {ok, AuthDb} = ensure_users_db_exists(DbName, [sys_db]),
+    AuthDb.
+
+
+get_user_props_from_db(UserName) ->
+    exec_if_auth_db(
+        fun(AuthDb) ->
+            Db = reopen_auth_db(AuthDb),
+            DocId = <<"org.couchdb.user:", UserName/binary>>,
+            try
+                {ok, Doc} = couch_db:open_doc(Db, DocId, [conflicts]),
+                {DocProps} = couch_doc:to_json_obj(Doc, []),
+                DocProps
+            catch
+            _:_Error ->
+                nil
+            end
+        end,
+        nil
+    ).
+
+ensure_users_db_exists(DbName, Options) ->
+    Options1 = [{user_ctx, #user_ctx{roles=[<<"_admin">>]}}, nologifmissing | Options],
+    case couch_db:open(DbName, Options1) of
+    {ok, Db} ->
+        ensure_auth_ddoc_exists(Db, <<"_design/_auth">>),
+        {ok, Db};
+    _Error ->
+        {ok, Db} = couch_db:create(DbName, Options1),
+        ok = ensure_auth_ddoc_exists(Db, <<"_design/_auth">>),
+        {ok, Db}
+    end.
+
+ensure_auth_ddoc_exists(Db, DDocId) ->
+    case couch_db:open_doc(Db, DDocId) of
+    {not_found, _Reason} ->
+        {ok, AuthDesign} = auth_design_doc(DDocId),
+        {ok, _Rev} = couch_db:update_doc(Db, AuthDesign, []);
+    {ok, Doc} ->
+        {Props} = couch_doc:to_json_obj(Doc, []),
+        case couch_util:get_value(<<"validate_doc_update">>, Props, []) of
+            ?AUTH_DB_DOC_VALIDATE_FUNCTION ->
+                ok;
+            _ ->
+                Props1 = lists:keyreplace(<<"validate_doc_update">>, 1, Props,
+                    {<<"validate_doc_update">>,
+                    ?AUTH_DB_DOC_VALIDATE_FUNCTION}),
+                couch_db:update_doc(Db, couch_doc:from_json_obj({Props1}), [])
+        end
+    end,
+    ok.
+
+auth_design_doc(DocId) ->
+    DocProps = [
+        {<<"_id">>, DocId},
+        {<<"language">>,<<"javascript">>},
+        {<<"validate_doc_update">>, ?AUTH_DB_DOC_VALIDATE_FUNCTION}
+    ],
+    {ok, couch_doc:from_json_obj({DocProps})}.

http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch_btree.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_btree.erl b/apps/couch/src/couch_btree.erl
new file mode 100644
index 0000000..789819e
--- /dev/null
+++ b/apps/couch/src/couch_btree.erl
@@ -0,0 +1,714 @@
+% 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_btree).
+
+-export([open/2, open/3, query_modify/4, add/2, add_remove/3]).
+-export([fold/4, full_reduce/1, final_reduce/2, size/1, foldl/3, foldl/4]).
+-export([fold_reduce/4, lookup/2, get_state/1, set_options/2]).
+-export([less/3]).
+
+-include("couch_db.hrl").
+-define(CHUNK_THRESHOLD, 16#4ff).
+
+extract(#btree{extract_kv=Extract}, Value) ->
+    Extract(Value).
+
+assemble(#btree{assemble_kv=Assemble}, Key, Value) ->
+    Assemble(Key, Value).
+
+less(#btree{less=Less}, A, B) ->
+    Less(A, B).
+
+% pass in 'nil' for State if a new Btree.
+open(State, Fd) ->
+    {ok, #btree{root=State, fd=Fd}}.
+
+set_options(Bt, []) ->
+    Bt;
+set_options(Bt, [{split, Extract}|Rest]) ->
+    set_options(Bt#btree{extract_kv=Extract}, Rest);
+set_options(Bt, [{join, Assemble}|Rest]) ->
+    set_options(Bt#btree{assemble_kv=Assemble}, Rest);
+set_options(Bt, [{less, Less}|Rest]) ->
+    set_options(Bt#btree{less=Less}, Rest);
+set_options(Bt, [{reduce, Reduce}|Rest]) ->
+    set_options(Bt#btree{reduce=Reduce}, Rest);
+set_options(Bt, [{compression, Comp}|Rest]) ->
+    set_options(Bt#btree{compression=Comp}, Rest).
+
+open(State, Fd, Options) ->
+    {ok, set_options(#btree{root=State, fd=Fd}, Options)}.
+
+get_state(#btree{root=Root}) ->
+    Root.
+
+final_reduce(#btree{reduce=Reduce}, Val) ->
+    final_reduce(Reduce, Val);
+final_reduce(Reduce, {[], []}) ->
+    Reduce(reduce, []);
+final_reduce(_Bt, {[], [Red]}) ->
+    Red;
+final_reduce(Reduce, {[], Reductions}) ->
+    Reduce(rereduce, Reductions);
+final_reduce(Reduce, {KVs, Reductions}) ->
+    Red = Reduce(reduce, KVs),
+    final_reduce(Reduce, {[], [Red | Reductions]}).
+
+fold_reduce(#btree{root=Root}=Bt, Fun, Acc, Options) ->
+    Dir = couch_util:get_value(dir, Options, fwd),
+    StartKey = couch_util:get_value(start_key, Options),
+    InEndRangeFun = make_key_in_end_range_function(Bt, Dir, Options),
+    KeyGroupFun = couch_util:get_value(key_group_fun, Options, fun(_,_) -> true end),
+    try
+        {ok, Acc2, GroupedRedsAcc2, GroupedKVsAcc2, GroupedKey2} =
+            reduce_stream_node(Bt, Dir, Root, StartKey, InEndRangeFun, undefined, [], [],
+            KeyGroupFun, Fun, Acc),
+        if GroupedKey2 == undefined ->
+            {ok, Acc2};
+        true ->
+            case Fun(GroupedKey2, {GroupedKVsAcc2, GroupedRedsAcc2}, Acc2) of
+            {ok, Acc3} -> {ok, Acc3};
+            {stop, Acc3} -> {ok, Acc3}
+            end
+        end
+    catch
+        throw:{stop, AccDone} -> {ok, AccDone}
+    end.
+
+full_reduce(#btree{root=nil,reduce=Reduce}) ->
+    {ok, Reduce(reduce, [])};
+full_reduce(#btree{root=Root}) ->
+    {ok, element(2, Root)}.
+
+size(#btree{root = nil}) ->
+    0;
+size(#btree{root = {_P, _Red}}) ->
+    % pre 1.2 format
+    nil;
+size(#btree{root = {_P, _Red, Size}}) ->
+    Size.
+
+% wraps a 2 arity function with the proper 3 arity function
+convert_fun_arity(Fun) when is_function(Fun, 2) ->
+    fun
+        (visit, KV, _Reds, AccIn) -> Fun(KV, AccIn);
+        (traverse, _K, _Red, AccIn) -> {ok, AccIn}
+    end;
+convert_fun_arity(Fun) when is_function(Fun, 3) ->
+    fun
+        (visit, KV, Reds, AccIn) -> Fun(KV, Reds, AccIn);
+        (traverse, _K, _Red, AccIn) -> {ok, AccIn}
+    end;
+convert_fun_arity(Fun) when is_function(Fun, 4) ->
+    Fun.    % Already arity 4
+
+make_key_in_end_range_function(#btree{less=Less}, fwd, Options) ->
+    case couch_util:get_value(end_key_gt, Options) of
+    undefined ->
+        case couch_util:get_value(end_key, Options) of
+        undefined ->
+            fun(_Key) -> true end;
+        LastKey ->
+            fun(Key) -> not Less(LastKey, Key) end
+        end;
+    EndKey ->
+        fun(Key) -> Less(Key, EndKey) end
+    end;
+make_key_in_end_range_function(#btree{less=Less}, rev, Options) ->
+    case couch_util:get_value(end_key_gt, Options) of
+    undefined ->
+        case couch_util:get_value(end_key, Options) of
+        undefined ->
+            fun(_Key) -> true end;
+        LastKey ->
+            fun(Key) -> not Less(Key, LastKey) end
+        end;
+    EndKey ->
+        fun(Key) -> Less(EndKey, Key) end
+    end.
+
+
+foldl(Bt, Fun, Acc) ->
+    fold(Bt, Fun, Acc, []).
+
+foldl(Bt, Fun, Acc, Options) ->
+    fold(Bt, Fun, Acc, Options).
+
+
+fold(#btree{root=nil}, _Fun, Acc, _Options) ->
+    {ok, {[], []}, Acc};
+fold(#btree{root=Root}=Bt, Fun, Acc, Options) ->
+    Dir = couch_util:get_value(dir, Options, fwd),
+    InRange = make_key_in_end_range_function(Bt, Dir, Options),
+    Result =
+    case couch_util:get_value(start_key, Options) of
+    undefined ->
+        stream_node(Bt, [], Bt#btree.root, InRange, Dir,
+                convert_fun_arity(Fun), Acc);
+    StartKey ->
+        stream_node(Bt, [], Bt#btree.root, StartKey, InRange, Dir,
+                convert_fun_arity(Fun), Acc)
+    end,
+    case Result of
+    {ok, Acc2}->
+        FullReduction = element(2, Root),
+        {ok, {[], [FullReduction]}, Acc2};
+    {stop, LastReduction, Acc2} ->
+        {ok, LastReduction, Acc2}
+    end.
+
+add(Bt, InsertKeyValues) ->
+    add_remove(Bt, InsertKeyValues, []).
+
+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}}.
+
+% for ordering different operations with the same key.
+% fetch < remove < insert
+op_order(fetch) -> 1;
+op_order(remove) -> 2;
+op_order(insert) -> 3.
+
+lookup(#btree{root=Root, less=Less}=Bt, Keys) ->
+    SortedKeys = lists:sort(Less, Keys),
+    {ok, SortedResults} = lookup(Bt, Root, SortedKeys),
+    % We want to return the results in the same order as the keys were input
+    % but we may have changed the order when we sorted. So we need to put the
+    % order back into the results.
+    couch_util:reorder_results(Keys, SortedResults).
+
+lookup(_Bt, nil, Keys) ->
+    {ok, [{Key, not_found} || Key <- Keys]};
+lookup(Bt, Node, Keys) ->
+    Pointer = element(1, Node),
+    {NodeType, NodeList} = get_node(Bt, Pointer),
+    case NodeType of
+    kp_node ->
+        lookup_kpnode(Bt, list_to_tuple(NodeList), 1, Keys, []);
+    kv_node ->
+        lookup_kvnode(Bt, list_to_tuple(NodeList), 1, Keys, [])
+    end.
+
+lookup_kpnode(_Bt, _NodeTuple, _LowerBound, [], Output) ->
+    {ok, lists:reverse(Output)};
+lookup_kpnode(_Bt, NodeTuple, LowerBound, Keys, Output) when tuple_size(NodeTuple) < LowerBound ->
+    {ok, lists:reverse(Output, [{Key, not_found} || Key <- Keys])};
+lookup_kpnode(Bt, NodeTuple, LowerBound, [FirstLookupKey | _] = LookupKeys, Output) ->
+    N = find_first_gteq(Bt, NodeTuple, LowerBound, tuple_size(NodeTuple), FirstLookupKey),
+    {Key, PointerInfo} = element(N, NodeTuple),
+    SplitFun = fun(LookupKey) -> not less(Bt, Key, LookupKey) end,
+    case lists:splitwith(SplitFun, LookupKeys) of
+    {[], GreaterQueries} ->
+        lookup_kpnode(Bt, NodeTuple, N + 1, GreaterQueries, Output);
+    {LessEqQueries, GreaterQueries} ->
+        {ok, Results} = lookup(Bt, PointerInfo, LessEqQueries),
+        lookup_kpnode(Bt, NodeTuple, N + 1, GreaterQueries, lists:reverse(Results, Output))
+    end.
+
+
+lookup_kvnode(_Bt, _NodeTuple, _LowerBound, [], Output) ->
+    {ok, lists:reverse(Output)};
+lookup_kvnode(_Bt, NodeTuple, LowerBound, Keys, Output) when tuple_size(NodeTuple) < LowerBound ->
+    % keys not found
+    {ok, lists:reverse(Output, [{Key, not_found} || Key <- Keys])};
+lookup_kvnode(Bt, NodeTuple, LowerBound, [LookupKey | RestLookupKeys], Output) ->
+    N = find_first_gteq(Bt, NodeTuple, LowerBound, tuple_size(NodeTuple), LookupKey),
+    {Key, Value} = element(N, NodeTuple),
+    case less(Bt, LookupKey, Key) of
+    true ->
+        % LookupKey is less than Key
+        lookup_kvnode(Bt, NodeTuple, N, RestLookupKeys, [{LookupKey, not_found} | Output]);
+    false ->
+        case less(Bt, Key, LookupKey) of
+        true ->
+            % LookupKey is greater than Key
+            lookup_kvnode(Bt, NodeTuple, N+1, RestLookupKeys, [{LookupKey, not_found} | Output]);
+        false ->
+            % LookupKey is equal to Key
+            lookup_kvnode(Bt, NodeTuple, N, RestLookupKeys, [{LookupKey, {ok, assemble(Bt, LookupKey, Value)}} | Output])
+        end
+    end.
+
+
+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).
+
+%%%%%%%%%%%%% The chunkify function sucks! %%%%%%%%%%%%%
+% It is inaccurate as it does not account for compression when blocks are
+% written. Plus with the "case byte_size(term_to_binary(InList)) of" code
+% it's probably really inefficient.
+
+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.
+
+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.
+
+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]).
+
+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).
+
+get_node(#btree{fd = Fd}, NodePos) ->
+    {ok, {NodeType, NodeList}} = couch_file:pread_term(Fd, NodePos),
+    {NodeType, NodeList}.
+
+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}.
+
+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.
+
+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]).
+
+bounded_tuple_to_list(Tuple, Start, End, Tail) ->
+    bounded_tuple_to_list2(Tuple, Start, End, [], Tail).
+
+bounded_tuple_to_list2(_Tuple, Start, End, Acc, Tail) when Start > End ->
+    lists:reverse(Acc, Tail);
+bounded_tuple_to_list2(Tuple, Start, End, Acc, Tail) ->
+    bounded_tuple_to_list2(Tuple, Start + 1, End, [element(Start, Tuple) | Acc], Tail).
+
+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.
+
+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.
+
+
+reduce_stream_node(_Bt, _Dir, nil, _KeyStart, _InEndRangeFun, GroupedKey, GroupedKVsAcc,
+        GroupedRedsAcc, _KeyGroupFun, _Fun, Acc) ->
+    {ok, Acc, GroupedRedsAcc, GroupedKVsAcc, GroupedKey};
+reduce_stream_node(Bt, Dir, Node, KeyStart, InEndRangeFun, GroupedKey, GroupedKVsAcc,
+        GroupedRedsAcc, KeyGroupFun, Fun, Acc) ->
+    P = element(1, Node),
+    case get_node(Bt, P) of
+    {kp_node, NodeList} ->
+        NodeList2 = adjust_dir(Dir, NodeList),
+        reduce_stream_kp_node(Bt, Dir, NodeList2, KeyStart, InEndRangeFun, GroupedKey,
+                GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc);
+    {kv_node, KVs} ->
+        KVs2 = adjust_dir(Dir, KVs),
+        reduce_stream_kv_node(Bt, Dir, KVs2, KeyStart, InEndRangeFun, GroupedKey,
+                GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc)
+    end.
+
+reduce_stream_kv_node(Bt, Dir, KVs, KeyStart, InEndRangeFun,
+                        GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
+                        KeyGroupFun, Fun, Acc) ->
+
+    GTEKeyStartKVs =
+    case KeyStart of
+    undefined ->
+        KVs;
+    _ ->
+        DropFun = case Dir of
+        fwd ->
+            fun({Key, _}) -> less(Bt, Key, KeyStart) end;
+        rev ->
+            fun({Key, _}) -> less(Bt, KeyStart, Key) end
+        end,
+        lists:dropwhile(DropFun, KVs)
+    end,
+    KVs2 = lists:takewhile(
+        fun({Key, _}) -> InEndRangeFun(Key) end, GTEKeyStartKVs),
+    reduce_stream_kv_node2(Bt, KVs2, GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
+                        KeyGroupFun, Fun, Acc).
+
+
+reduce_stream_kv_node2(_Bt, [], GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
+        _KeyGroupFun, _Fun, Acc) ->
+    {ok, Acc, GroupedRedsAcc, GroupedKVsAcc, GroupedKey};
+reduce_stream_kv_node2(Bt, [{Key, Value}| RestKVs], GroupedKey, GroupedKVsAcc,
+        GroupedRedsAcc, KeyGroupFun, Fun, Acc) ->
+    case GroupedKey of
+    undefined ->
+        reduce_stream_kv_node2(Bt, RestKVs, Key,
+                [assemble(Bt,Key,Value)], [], KeyGroupFun, Fun, Acc);
+    _ ->
+
+        case KeyGroupFun(GroupedKey, Key) of
+        true ->
+            reduce_stream_kv_node2(Bt, RestKVs, GroupedKey,
+                [assemble(Bt,Key,Value)|GroupedKVsAcc], GroupedRedsAcc, KeyGroupFun,
+                Fun, Acc);
+        false ->
+            case Fun(GroupedKey, {GroupedKVsAcc, GroupedRedsAcc}, Acc) of
+            {ok, Acc2} ->
+                reduce_stream_kv_node2(Bt, RestKVs, Key, [assemble(Bt,Key,Value)],
+                    [], KeyGroupFun, Fun, Acc2);
+            {stop, Acc2} ->
+                throw({stop, Acc2})
+            end
+        end
+    end.
+
+reduce_stream_kp_node(Bt, Dir, NodeList, KeyStart, InEndRangeFun,
+                        GroupedKey, GroupedKVsAcc, GroupedRedsAcc,
+                        KeyGroupFun, Fun, Acc) ->
+    Nodes =
+    case KeyStart of
+    undefined ->
+        NodeList;
+    _ ->
+        case Dir of
+        fwd ->
+            lists:dropwhile(fun({Key, _}) -> less(Bt, Key, KeyStart) end, NodeList);
+        rev ->
+            RevKPs = lists:reverse(NodeList),
+            case lists:splitwith(fun({Key, _}) -> less(Bt, Key, KeyStart) end, RevKPs) of
+            {_Before, []} ->
+                NodeList;
+            {Before, [FirstAfter | _]} ->
+                [FirstAfter | lists:reverse(Before)]
+            end
+        end
+    end,
+    {InRange, MaybeInRange} = lists:splitwith(
+        fun({Key, _}) -> InEndRangeFun(Key) end, Nodes),
+    NodesInRange = case MaybeInRange of
+    [FirstMaybeInRange | _] when Dir =:= fwd ->
+        InRange ++ [FirstMaybeInRange];
+    _ ->
+        InRange
+    end,
+    reduce_stream_kp_node2(Bt, Dir, NodesInRange, KeyStart, InEndRangeFun,
+        GroupedKey, GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc).
+
+
+reduce_stream_kp_node2(Bt, Dir, [{_Key, NodeInfo} | RestNodeList], KeyStart, InEndRangeFun,
+                        undefined, [], [], KeyGroupFun, Fun, Acc) ->
+    {ok, Acc2, GroupedRedsAcc2, GroupedKVsAcc2, GroupedKey2} =
+            reduce_stream_node(Bt, Dir, NodeInfo, KeyStart, InEndRangeFun, undefined,
+                [], [], KeyGroupFun, Fun, Acc),
+    reduce_stream_kp_node2(Bt, Dir, RestNodeList, KeyStart, InEndRangeFun, GroupedKey2,
+            GroupedKVsAcc2, GroupedRedsAcc2, KeyGroupFun, Fun, Acc2);
+reduce_stream_kp_node2(Bt, Dir, NodeList, KeyStart, InEndRangeFun,
+        GroupedKey, GroupedKVsAcc, GroupedRedsAcc, KeyGroupFun, Fun, Acc) ->
+    {Grouped0, Ungrouped0} = lists:splitwith(fun({Key,_}) ->
+        KeyGroupFun(GroupedKey, Key) end, NodeList),
+    {GroupedNodes, UngroupedNodes} =
+    case Grouped0 of
+    [] ->
+        {Grouped0, Ungrouped0};
+    _ ->
+        [FirstGrouped | RestGrouped] = lists:reverse(Grouped0),
+        {RestGrouped, [FirstGrouped | Ungrouped0]}
+    end,
+    GroupedReds = [element(2, Node) || {_, Node} <- GroupedNodes],
+    case UngroupedNodes of
+    [{_Key, NodeInfo}|RestNodes] ->
+        {ok, Acc2, GroupedRedsAcc2, GroupedKVsAcc2, GroupedKey2} =
+            reduce_stream_node(Bt, Dir, NodeInfo, KeyStart, InEndRangeFun, GroupedKey,
+                GroupedKVsAcc, GroupedReds ++ GroupedRedsAcc, KeyGroupFun, Fun, Acc),
+        reduce_stream_kp_node2(Bt, Dir, RestNodes, KeyStart, InEndRangeFun, GroupedKey2,
+                GroupedKVsAcc2, GroupedRedsAcc2, KeyGroupFun, Fun, Acc2);
+    [] ->
+        {ok, Acc, GroupedReds ++ GroupedRedsAcc, GroupedKVsAcc, GroupedKey}
+    end.
+
+adjust_dir(fwd, List) ->
+    List;
+adjust_dir(rev, List) ->
+    lists:reverse(List).
+
+stream_node(Bt, Reds, Node, StartKey, InRange, Dir, Fun, Acc) ->
+    Pointer = element(1, Node),
+    {NodeType, NodeList} = get_node(Bt, Pointer),
+    case NodeType of
+    kp_node ->
+        stream_kp_node(Bt, Reds, adjust_dir(Dir, NodeList), StartKey, InRange, Dir, Fun, Acc);
+    kv_node ->
+        stream_kv_node(Bt, Reds, adjust_dir(Dir, NodeList), StartKey, InRange, Dir, Fun, Acc)
+    end.
+
+stream_node(Bt, Reds, Node, InRange, Dir, Fun, Acc) ->
+    Pointer = element(1, Node),
+    {NodeType, NodeList} = get_node(Bt, Pointer),
+    case NodeType of
+    kp_node ->
+        stream_kp_node(Bt, Reds, adjust_dir(Dir, NodeList), InRange, Dir, Fun, Acc);
+    kv_node ->
+        stream_kv_node2(Bt, Reds, [], adjust_dir(Dir, NodeList), InRange, Dir, Fun, Acc)
+    end.
+
+stream_kp_node(_Bt, _Reds, [], _InRange, _Dir, _Fun, Acc) ->
+    {ok, Acc};
+stream_kp_node(Bt, Reds, [{Key, Node} | Rest], InRange, Dir, Fun, Acc) ->
+    Red = element(2, Node),
+    case Fun(traverse, Key, Red, Acc) of
+    {ok, Acc2} ->
+        case stream_node(Bt, Reds, Node, InRange, Dir, Fun, Acc2) of
+        {ok, Acc3} ->
+            stream_kp_node(Bt, [Red | Reds], Rest, InRange, Dir, Fun, Acc3);
+        {stop, LastReds, Acc3} ->
+            {stop, LastReds, Acc3}
+        end;
+    {skip, Acc2} ->
+        stream_kp_node(Bt, [Red | Reds], Rest, InRange, Dir, Fun, Acc2)
+    end.
+
+drop_nodes(_Bt, Reds, _StartKey, []) ->
+    {Reds, []};
+drop_nodes(Bt, Reds, StartKey, [{NodeKey, Node} | RestKPs]) ->
+    case less(Bt, NodeKey, StartKey) of
+    true ->
+        drop_nodes(Bt, [element(2, Node) | Reds], StartKey, RestKPs);
+    false ->
+        {Reds, [{NodeKey, Node} | RestKPs]}
+    end.
+
+stream_kp_node(Bt, Reds, KPs, StartKey, InRange, Dir, Fun, Acc) ->
+    {NewReds, NodesToStream} =
+    case Dir of
+    fwd ->
+        % drop all nodes sorting before the key
+        drop_nodes(Bt, Reds, StartKey, KPs);
+    rev ->
+        % keep all nodes sorting before the key, AND the first node to sort after
+        RevKPs = lists:reverse(KPs),
+         case lists:splitwith(fun({Key, _Pointer}) -> less(Bt, Key, StartKey) end, RevKPs) of
+        {_RevsBefore, []} ->
+            % everything sorts before it
+            {Reds, KPs};
+        {RevBefore, [FirstAfter | Drop]} ->
+            {[element(2, Node) || {_K, Node} <- Drop] ++ Reds,
+                 [FirstAfter | lists:reverse(RevBefore)]}
+        end
+    end,
+    case NodesToStream of
+    [] ->
+        {ok, Acc};
+    [{_Key, Node} | Rest] ->
+        case stream_node(Bt, NewReds, Node, StartKey, InRange, Dir, Fun, Acc) of
+        {ok, Acc2} ->
+            Red = element(2, Node),
+            stream_kp_node(Bt, [Red | NewReds], Rest, InRange, Dir, Fun, Acc2);
+        {stop, LastReds, Acc2} ->
+            {stop, LastReds, Acc2}
+        end
+    end.
+
+stream_kv_node(Bt, Reds, KVs, StartKey, InRange, Dir, Fun, Acc) ->
+    DropFun =
+    case Dir of
+    fwd ->
+        fun({Key, _}) -> less(Bt, Key, StartKey) end;
+    rev ->
+        fun({Key, _}) -> less(Bt, StartKey, Key) end
+    end,
+    {LTKVs, GTEKVs} = lists:splitwith(DropFun, KVs),
+    AssembleLTKVs = [assemble(Bt,K,V) || {K,V} <- LTKVs],
+    stream_kv_node2(Bt, Reds, AssembleLTKVs, GTEKVs, InRange, Dir, Fun, Acc).
+
+stream_kv_node2(_Bt, _Reds, _PrevKVs, [], _InRange, _Dir, _Fun, Acc) ->
+    {ok, Acc};
+stream_kv_node2(Bt, Reds, PrevKVs, [{K,V} | RestKVs], InRange, Dir, Fun, Acc) ->
+    case InRange(K) of
+    false ->
+        {stop, {PrevKVs, Reds}, Acc};
+    true ->
+        AssembledKV = assemble(Bt, K, V),
+        case Fun(visit, AssembledKV, {PrevKVs, Reds}, Acc) of
+        {ok, Acc2} ->
+            stream_kv_node2(Bt, Reds, [AssembledKV | PrevKVs], RestKVs, InRange, Dir, Fun, Acc2);
+        {stop, Acc2} ->
+            {stop, {PrevKVs, Reds}, Acc2}
+        end
+    end.

http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch_changes.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_changes.erl b/apps/couch/src/couch_changes.erl
new file mode 100644
index 0000000..6edde32
--- /dev/null
+++ b/apps/couch/src/couch_changes.erl
@@ -0,0 +1,577 @@
+% 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_changes).
+-include("couch_db.hrl").
+
+-export([handle_changes/3]).
+
+% For the builtin filter _docs_ids, this is the maximum number
+% of documents for which we trigger the optimized code path.
+-define(MAX_DOC_IDS, 100).
+
+-record(changes_acc, {
+    db,
+    seq,
+    prepend,
+    filter,
+    callback,
+    user_acc,
+    resp_type,
+    limit,
+    include_docs,
+    doc_options,
+    conflicts,
+    timeout,
+    timeout_fun
+}).
+
+%% @type Req -> #httpd{} | {json_req, JsonObj()}
+handle_changes(Args1, Req, Db0) ->
+    #changes_args{
+        style = Style,
+        filter = FilterName,
+        feed = Feed,
+        dir = Dir,
+        since = Since
+    } = Args1,
+    {FilterFun, FilterArgs} = make_filter_fun(FilterName, Style, Req, Db0),
+    Args = Args1#changes_args{filter_fun = FilterFun, filter_args = FilterArgs},
+    Start = fun() ->
+        {ok, Db} = couch_db:reopen(Db0),
+        StartSeq = case Dir of
+        rev ->
+            couch_db:get_update_seq(Db);
+        fwd ->
+            Since
+        end,
+        {Db, StartSeq}
+    end,
+    % begin timer to deal with heartbeat when filter function fails
+    case Args#changes_args.heartbeat of
+    undefined ->
+        erlang:erase(last_changes_heartbeat);
+    Val when is_integer(Val); Val =:= true ->
+        put(last_changes_heartbeat, now())
+    end,
+
+    case lists:member(Feed, ["continuous", "longpoll", "eventsource"]) of
+    true ->
+        fun(CallbackAcc) ->
+            {Callback, UserAcc} = get_callback_acc(CallbackAcc),
+            Self = self(),
+            {ok, Notify} = couch_db_update_notifier:start_link(
+                fun({_, DbName}) when  Db0#db.name == DbName ->
+                    Self ! db_updated;
+                (_) ->
+                    ok
+                end
+            ),
+            {Db, StartSeq} = Start(),
+            UserAcc2 = start_sending_changes(Callback, UserAcc, Feed),
+            {Timeout, TimeoutFun} = get_changes_timeout(Args, Callback),
+            Acc0 = build_acc(Args, Callback, UserAcc2, Db, StartSeq,
+                             <<"">>, Timeout, TimeoutFun),
+            try
+                keep_sending_changes(
+                    Args#changes_args{dir=fwd},
+                    Acc0,
+                    true)
+            after
+                couch_db_update_notifier:stop(Notify),
+                get_rest_db_updated(ok) % clean out any remaining update messages
+            end
+        end;
+    false ->
+        fun(CallbackAcc) ->
+            {Callback, UserAcc} = get_callback_acc(CallbackAcc),
+            UserAcc2 = start_sending_changes(Callback, UserAcc, Feed),
+            {Timeout, TimeoutFun} = get_changes_timeout(Args, Callback),
+            {Db, StartSeq} = Start(),
+            Acc0 = build_acc(Args#changes_args{feed="normal"}, Callback,
+                             UserAcc2, Db, StartSeq, <<>>, Timeout, TimeoutFun),
+            {ok, #changes_acc{seq = LastSeq, user_acc = UserAcc3}} =
+                send_changes(
+                    Args#changes_args{feed="normal"},
+                    Acc0,
+                    true),
+            end_sending_changes(Callback, UserAcc3, LastSeq, Feed)
+        end
+    end.
+
+get_callback_acc({Callback, _UserAcc} = Pair) when is_function(Callback, 3) ->
+    Pair;
+get_callback_acc(Callback) when is_function(Callback, 2) ->
+    {fun(Ev, Data, _) -> Callback(Ev, Data) end, ok}.
+
+%% @type Req -> #httpd{} | {json_req, JsonObj()}
+make_filter_fun([$_ | _] = FilterName, Style, Req, Db) ->
+    builtin_filter_fun(FilterName, Style, Req, Db);
+make_filter_fun(FilterName, Style, Req, Db) ->
+    {os_filter_fun(FilterName, Style, Req, Db), []}.
+
+os_filter_fun(FilterName, Style, Req, Db) ->
+    case [list_to_binary(couch_httpd:unquote(Part))
+            || Part <- string:tokens(FilterName, "/")] of
+    [] ->
+        fun(_Db2, #doc_info{revs=Revs}) ->
+                builtin_results(Style, Revs)
+        end;
+    [DName, FName] ->
+        DesignId = <<"_design/", DName/binary>>,
+        DDoc = couch_httpd_db:couch_doc_open(Db, DesignId, nil, [ejson_body]),
+        % validate that the ddoc has the filter fun
+        #doc{body={Props}} = DDoc,
+        couch_util:get_nested_json_value({Props}, [<<"filters">>, FName]),
+        fun(Db2, DocInfo) ->
+            DocInfos =
+            case Style of
+            main_only ->
+                [DocInfo];
+            all_docs ->
+                [DocInfo#doc_info{revs=[Rev]}|| Rev <- DocInfo#doc_info.revs]
+            end,
+            Docs = [Doc || {ok, Doc} <- [
+                    couch_db:open_doc(Db2, DocInfo2, [deleted, conflicts])
+                        || DocInfo2 <- DocInfos]],
+            {ok, Passes} = couch_query_servers:filter_docs(
+                Req, Db2, DDoc, FName, Docs
+            ),
+            [{[{<<"rev">>, couch_doc:rev_to_str({RevPos,RevId})}]}
+                || {Pass, #doc{revs={RevPos,[RevId|_]}}}
+                <- lists:zip(Passes, Docs), Pass == true]
+        end;
+    _Else ->
+        throw({bad_request,
+            "filter parameter must be of the form `designname/filtername`"})
+    end.
+
+builtin_filter_fun("_doc_ids", Style, {json_req, {Props}}, _Db) ->
+    DocIds = couch_util:get_value(<<"doc_ids">>, Props),
+    {filter_docids(DocIds, Style), DocIds};
+builtin_filter_fun("_doc_ids", Style, #httpd{method='POST'}=Req, _Db) ->
+    {Props} = couch_httpd:json_body_obj(Req),
+    DocIds =  couch_util:get_value(<<"doc_ids">>, Props, nil),
+    {filter_docids(DocIds, Style), DocIds};
+builtin_filter_fun("_doc_ids", Style, #httpd{method='GET'}=Req, _Db) ->
+    DocIds = ?JSON_DECODE(couch_httpd:qs_value(Req, "doc_ids", "null")),
+    {filter_docids(DocIds, Style), DocIds};
+builtin_filter_fun("_design", Style, _Req, _Db) ->
+    {filter_designdoc(Style), []};
+builtin_filter_fun("_view", Style, Req, Db) ->
+    ViewName = couch_httpd:qs_value(Req, "view", ""),
+    {filter_view(ViewName, Style, Db), []};
+builtin_filter_fun(_FilterName, _Style, _Req, _Db) ->
+    throw({bad_request, "unknown builtin filter name"}).
+
+filter_docids(DocIds, Style) when is_list(DocIds)->
+    fun(_Db, #doc_info{id=DocId, revs=Revs}) ->
+            case lists:member(DocId, DocIds) of
+                true ->
+                    builtin_results(Style, Revs);
+                _ -> []
+            end
+    end;
+filter_docids(_, _) ->
+    throw({bad_request, "`doc_ids` filter parameter is not a list."}).
+
+filter_designdoc(Style) ->
+    fun(_Db, #doc_info{id=DocId, revs=Revs}) ->
+            case DocId of
+            <<"_design", _/binary>> ->
+                    builtin_results(Style, Revs);
+                _ -> []
+            end
+    end.
+
+filter_view("", _Style, _Db) ->
+    throw({bad_request, "`view` filter parameter is not provided."});
+filter_view(ViewName, Style, Db) ->
+    case [list_to_binary(couch_httpd:unquote(Part))
+            || Part <- string:tokens(ViewName, "/")] of
+        [] ->
+            throw({bad_request, "Invalid `view` parameter."});
+        [DName, VName] ->
+            DesignId = <<"_design/", DName/binary>>,
+            DDoc = couch_httpd_db:couch_doc_open(Db, DesignId, nil, [ejson_body]),
+            % validate that the ddoc has the filter fun
+            #doc{body={Props}} = DDoc,
+            couch_util:get_nested_json_value({Props}, [<<"views">>, VName]),
+            fun(Db2, DocInfo) ->
+                DocInfos =
+                case Style of
+                main_only ->
+                    [DocInfo];
+                all_docs ->
+                    [DocInfo#doc_info{revs=[Rev]}|| Rev <- DocInfo#doc_info.revs]
+                end,
+                Docs = [Doc || {ok, Doc} <- [
+                        couch_db:open_doc(Db2, DocInfo2, [deleted, conflicts])
+                            || DocInfo2 <- DocInfos]],
+                {ok, Passes} = couch_query_servers:filter_view(
+                    DDoc, VName, Docs
+                ),
+                [{[{<<"rev">>, couch_doc:rev_to_str({RevPos,RevId})}]}
+                    || {Pass, #doc{revs={RevPos,[RevId|_]}}}
+                    <- lists:zip(Passes, Docs), Pass == true]
+            end
+        end.
+
+builtin_results(Style, [#rev_info{rev=Rev}|_]=Revs) ->
+    case Style of
+        main_only ->
+            [{[{<<"rev">>, couch_doc:rev_to_str(Rev)}]}];
+        all_docs ->
+            [{[{<<"rev">>, couch_doc:rev_to_str(R)}]}
+                || #rev_info{rev=R} <- Revs]
+    end.
+
+get_changes_timeout(Args, Callback) ->
+    #changes_args{
+        heartbeat = Heartbeat,
+        timeout = Timeout,
+        feed = ResponseType
+    } = Args,
+    DefaultTimeout = list_to_integer(
+        couch_config:get("httpd", "changes_timeout", "60000")
+    ),
+    case Heartbeat of
+    undefined ->
+        case Timeout of
+        undefined ->
+            {DefaultTimeout, fun(UserAcc) -> {stop, UserAcc} end};
+        infinity ->
+            {infinity, fun(UserAcc) -> {stop, UserAcc} end};
+        _ ->
+            {lists:min([DefaultTimeout, Timeout]),
+                fun(UserAcc) -> {stop, UserAcc} end}
+        end;
+    true ->
+        {DefaultTimeout,
+            fun(UserAcc) -> {ok, Callback(timeout, ResponseType, UserAcc)} end};
+    _ ->
+        {lists:min([DefaultTimeout, Heartbeat]),
+            fun(UserAcc) -> {ok, Callback(timeout, ResponseType, UserAcc)} end}
+    end.
+
+start_sending_changes(_Callback, UserAcc, ResponseType)
+        when ResponseType =:= "continuous"
+        orelse ResponseType =:= "eventsource" ->
+    UserAcc;
+start_sending_changes(Callback, UserAcc, ResponseType) ->
+    Callback(start, ResponseType, UserAcc).
+
+build_acc(Args, Callback, UserAcc, Db, StartSeq, Prepend, Timeout, TimeoutFun) ->
+    #changes_args{
+        include_docs = IncludeDocs,
+        doc_options = DocOpts,
+        conflicts = Conflicts,
+        limit = Limit,
+        feed = ResponseType,
+        filter_fun = FilterFun
+    } = Args,
+    #changes_acc{
+        db = Db,
+        seq = StartSeq,
+        prepend = Prepend,
+        filter = FilterFun,
+        callback = Callback,
+        user_acc = UserAcc,
+        resp_type = ResponseType,
+        limit = Limit,
+        include_docs = IncludeDocs,
+        doc_options = DocOpts,
+        conflicts = Conflicts,
+        timeout = Timeout,
+        timeout_fun = TimeoutFun
+    }.
+
+send_changes(Args, Acc0, FirstRound) ->
+    #changes_args{
+        dir = Dir,
+        filter = FilterName,
+        filter_args = FilterArgs
+    } = Args,
+    #changes_acc{
+        db = Db,
+        seq = StartSeq
+    } = Acc0,
+    case FirstRound of
+    true ->
+        case FilterName of
+        "_doc_ids" when length(FilterArgs) =< ?MAX_DOC_IDS ->
+            send_changes_doc_ids(
+                FilterArgs, Db, StartSeq, Dir, fun changes_enumerator/2, Acc0);
+        "_design" ->
+            send_changes_design_docs(
+                Db, StartSeq, Dir, fun changes_enumerator/2, Acc0);
+        _ ->
+            couch_db:changes_since(
+                Db, StartSeq, fun changes_enumerator/2, [{dir, Dir}], Acc0)
+        end;
+    false ->
+        couch_db:changes_since(
+            Db, StartSeq, fun changes_enumerator/2, [{dir, Dir}], Acc0)
+    end.
+
+
+send_changes_doc_ids(DocIds, Db, StartSeq, Dir, Fun, Acc0) ->
+    Lookups = couch_btree:lookup(Db#db.fulldocinfo_by_id_btree, DocIds),
+    FullDocInfos = lists:foldl(
+        fun({ok, FDI}, Acc) ->
+            [FDI | Acc];
+        (not_found, Acc) ->
+            Acc
+        end,
+        [], Lookups),
+    send_lookup_changes(FullDocInfos, StartSeq, Dir, Db, Fun, Acc0).
+
+
+send_changes_design_docs(Db, StartSeq, Dir, Fun, Acc0) ->
+    FoldFun = fun(FullDocInfo, _, Acc) ->
+        {ok, [FullDocInfo | Acc]}
+    end,
+    KeyOpts = [{start_key, <<"_design/">>}, {end_key_gt, <<"_design0">>}],
+    {ok, _, FullDocInfos} = couch_btree:fold(
+        Db#db.fulldocinfo_by_id_btree, FoldFun, [], KeyOpts),
+    send_lookup_changes(FullDocInfos, StartSeq, Dir, Db, Fun, Acc0).
+
+
+send_lookup_changes(FullDocInfos, StartSeq, Dir, Db, Fun, Acc0) ->
+    FoldFun = case Dir of
+    fwd ->
+        fun lists:foldl/3;
+    rev ->
+        fun lists:foldr/3
+    end,
+    GreaterFun = case Dir of
+    fwd ->
+        fun(A, B) -> A > B end;
+    rev ->
+        fun(A, B) -> A =< B end
+    end,
+    DocInfos = lists:foldl(
+        fun(FDI, Acc) ->
+            DI = couch_doc:to_doc_info(FDI),
+            case GreaterFun(DI#doc_info.high_seq, StartSeq) of
+            true ->
+                [DI | Acc];
+            false ->
+                Acc
+            end
+        end,
+        [], FullDocInfos),
+    SortedDocInfos = lists:keysort(#doc_info.high_seq, DocInfos),
+    FinalAcc = try
+        FoldFun(
+            fun(DocInfo, Acc) ->
+                case Fun(DocInfo, Acc) of
+                {ok, NewAcc} ->
+                    NewAcc;
+                {stop, NewAcc} ->
+                    throw({stop, NewAcc})
+                end
+            end,
+            Acc0, SortedDocInfos)
+    catch
+    throw:{stop, Acc} ->
+        Acc
+    end,
+    case Dir of
+    fwd ->
+        {ok, FinalAcc#changes_acc{seq = couch_db:get_update_seq(Db)}};
+    rev ->
+        {ok, FinalAcc}
+    end.
+
+
+keep_sending_changes(Args, Acc0, FirstRound) ->
+    #changes_args{
+        feed = ResponseType,
+        limit = Limit,
+        db_open_options = DbOptions
+    } = Args,
+
+    {ok, ChangesAcc} = send_changes(
+        Args#changes_args{dir=fwd},
+        Acc0,
+        FirstRound),
+    #changes_acc{
+        db = Db, callback = Callback, timeout = Timeout, timeout_fun = TimeoutFun,
+        seq = EndSeq, prepend = Prepend2, user_acc = UserAcc2, limit = NewLimit
+    } = ChangesAcc,
+
+    couch_db:close(Db),
+    if Limit > NewLimit, ResponseType == "longpoll" ->
+        end_sending_changes(Callback, UserAcc2, EndSeq, ResponseType);
+    true ->
+        case wait_db_updated(Timeout, TimeoutFun, UserAcc2) of
+        {updated, UserAcc4} ->
+            DbOptions1 = [{user_ctx, Db#db.user_ctx} | DbOptions],
+            case couch_db:open(Db#db.name, DbOptions1) of
+            {ok, Db2} ->
+                keep_sending_changes(
+                  Args#changes_args{limit=NewLimit},
+                  ChangesAcc#changes_acc{
+                    db = Db2,
+                    user_acc = UserAcc4,
+                    seq = EndSeq,
+                    prepend = Prepend2,
+                    timeout = Timeout,
+                    timeout_fun = TimeoutFun},
+                  false);
+            _Else ->
+                end_sending_changes(Callback, UserAcc2, EndSeq, ResponseType)
+            end;
+        {stop, UserAcc4} ->
+            end_sending_changes(Callback, UserAcc4, EndSeq, ResponseType)
+        end
+    end.
+
+end_sending_changes(Callback, UserAcc, EndSeq, ResponseType) ->
+    Callback({stop, EndSeq}, ResponseType, UserAcc).
+
+changes_enumerator(DocInfo, #changes_acc{resp_type = ResponseType} = Acc)
+        when ResponseType =:= "continuous"
+        orelse ResponseType =:= "eventsource" ->
+    #changes_acc{
+        filter = FilterFun, callback = Callback,
+        user_acc = UserAcc, limit = Limit, db = Db,
+        timeout = Timeout, timeout_fun = TimeoutFun
+    } = Acc,
+    #doc_info{high_seq = Seq} = DocInfo,
+    Results0 = FilterFun(Db, DocInfo),
+    Results = [Result || Result <- Results0, Result /= null],
+    %% TODO: I'm thinking this should be < 1 and not =< 1
+    Go = if Limit =< 1 -> stop; true -> ok end,
+    case Results of
+    [] ->
+        {Done, UserAcc2} = maybe_heartbeat(Timeout, TimeoutFun, UserAcc),
+        case Done of
+        stop ->
+            {stop, Acc#changes_acc{seq = Seq, user_acc = UserAcc2}};
+        ok ->
+            {Go, Acc#changes_acc{seq = Seq, user_acc = UserAcc2}}
+        end;
+    _ ->
+        ChangesRow = changes_row(Results, DocInfo, Acc),
+        UserAcc2 = Callback({change, ChangesRow, <<>>}, ResponseType, UserAcc),
+        reset_heartbeat(),
+        {Go, Acc#changes_acc{seq = Seq, user_acc = UserAcc2, limit = Limit - 1}}
+    end;
+changes_enumerator(DocInfo, Acc) ->
+    #changes_acc{
+        filter = FilterFun, callback = Callback, prepend = Prepend,
+        user_acc = UserAcc, limit = Limit, resp_type = ResponseType, db = Db,
+        timeout = Timeout, timeout_fun = TimeoutFun
+    } = Acc,
+    #doc_info{high_seq = Seq} = DocInfo,
+    Results0 = FilterFun(Db, DocInfo),
+    Results = [Result || Result <- Results0, Result /= null],
+    Go = if (Limit =< 1) andalso Results =/= [] -> stop; true -> ok end,
+    case Results of
+    [] ->
+        {Done, UserAcc2} = maybe_heartbeat(Timeout, TimeoutFun, UserAcc),
+        case Done of
+        stop ->
+            {stop, Acc#changes_acc{seq = Seq, user_acc = UserAcc2}};
+        ok ->
+            {Go, Acc#changes_acc{seq = Seq, user_acc = UserAcc2}}
+        end;
+    _ ->
+        ChangesRow = changes_row(Results, DocInfo, Acc),
+        UserAcc2 = Callback({change, ChangesRow, Prepend}, ResponseType, UserAcc),
+        reset_heartbeat(),
+        {Go, Acc#changes_acc{
+            seq = Seq, prepend = <<",\n">>,
+            user_acc = UserAcc2, limit = Limit - 1}}
+    end.
+
+
+changes_row(Results, DocInfo, Acc) ->
+    #doc_info{
+        id = Id, high_seq = Seq, revs = [#rev_info{deleted = Del} | _]
+    } = DocInfo,
+    #changes_acc{
+        db = Db,
+        include_docs = IncDoc,
+        doc_options = DocOpts,
+        conflicts = Conflicts
+    } = Acc,
+    {[{<<"seq">>, Seq}, {<<"id">>, Id}, {<<"changes">>, Results}] ++
+        deleted_item(Del) ++ case IncDoc of
+            true ->
+                Opts = case Conflicts of
+                    true -> [deleted, conflicts];
+                    false -> [deleted]
+                end,
+                Doc = couch_index_util:load_doc(Db, DocInfo, Opts),
+                case Doc of
+                    null ->
+                        [{doc, null}];
+                    _ ->
+                        [{doc, couch_doc:to_json_obj(Doc, DocOpts)}]
+                end;
+            false ->
+                []
+        end}.
+
+deleted_item(true) -> [{<<"deleted">>, true}];
+deleted_item(_) -> [].
+
+% waits for a db_updated msg, if there are multiple msgs, collects them.
+wait_db_updated(Timeout, TimeoutFun, UserAcc) ->
+    receive
+    db_updated ->
+        get_rest_db_updated(UserAcc)
+    after Timeout ->
+        {Go, UserAcc2} = TimeoutFun(UserAcc),
+        case Go of
+        ok ->
+            wait_db_updated(Timeout, TimeoutFun, UserAcc2);
+        stop ->
+            {stop, UserAcc2}
+        end
+    end.
+
+get_rest_db_updated(UserAcc) ->
+    receive
+    db_updated ->
+        get_rest_db_updated(UserAcc)
+    after 0 ->
+        {updated, UserAcc}
+    end.
+
+reset_heartbeat() ->
+    case get(last_changes_heartbeat) of
+    undefined ->
+        ok;
+    _ ->
+        put(last_changes_heartbeat, now())
+    end.
+
+maybe_heartbeat(Timeout, TimeoutFun, Acc) ->
+    Before = get(last_changes_heartbeat),
+    case Before of
+    undefined ->
+        {ok, Acc};
+    _ ->
+        Now = now(),
+        case timer:now_diff(Now, Before) div 1000 >= Timeout of
+        true ->
+            Acc2 = TimeoutFun(Acc),
+            put(last_changes_heartbeat, Now),
+            Acc2;
+        false ->
+            {ok, Acc}
+        end
+    end.

http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch_compaction_daemon.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_compaction_daemon.erl b/apps/couch/src/couch_compaction_daemon.erl
new file mode 100644
index 0000000..18a51a4
--- /dev/null
+++ b/apps/couch/src/couch_compaction_daemon.erl
@@ -0,0 +1,504 @@
+% 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_compaction_daemon).
+-behaviour(gen_server).
+
+% public API
+-export([start_link/0, config_change/3]).
+
+% gen_server callbacks
+-export([init/1, handle_call/3, handle_info/2, handle_cast/2]).
+-export([code_change/3, terminate/2]).
+
+-include("couch_db.hrl").
+
+-define(CONFIG_ETS, couch_compaction_daemon_config).
+
+-record(state, {
+    loop_pid
+}).
+
+-record(config, {
+    db_frag = nil,
+    view_frag = nil,
+    period = nil,
+    cancel = false,
+    parallel_view_compact = false
+}).
+
+-record(period, {
+    from = nil,
+    to = nil
+}).
+
+
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+init(_) ->
+    process_flag(trap_exit, true),
+    ?CONFIG_ETS = ets:new(?CONFIG_ETS, [named_table, set, protected]),
+    ok = couch_config:register(fun ?MODULE:config_change/3),
+    load_config(),
+    Server = self(),
+    Loop = spawn_link(fun() -> compact_loop(Server) end),
+    {ok, #state{loop_pid = Loop}}.
+
+
+config_change("compactions", DbName, NewValue) ->
+    ok = gen_server:cast(?MODULE, {config_update, DbName, NewValue}).
+
+
+handle_cast({config_update, DbName, deleted}, State) ->
+    true = ets:delete(?CONFIG_ETS, ?l2b(DbName)),
+    {noreply, State};
+
+handle_cast({config_update, DbName, Config}, #state{loop_pid = Loop} = State) ->
+    case parse_config(DbName, Config) of
+    {ok, NewConfig} ->
+        WasEmpty = (ets:info(?CONFIG_ETS, size) =:= 0),
+        true = ets:insert(?CONFIG_ETS, {?l2b(DbName), NewConfig}),
+        case WasEmpty of
+        true ->
+            Loop ! {self(), have_config};
+        false ->
+            ok
+        end;
+    error ->
+        ok
+    end,
+    {noreply, State}.
+
+
+handle_call(Msg, _From, State) ->
+    {stop, {unexpected_call, Msg}, State}.
+
+
+handle_info({'EXIT', Pid, Reason}, #state{loop_pid = Pid} = State) ->
+    {stop, {compaction_loop_died, Reason}, State}.
+
+
+terminate(_Reason, _State) ->
+    true = ets:delete(?CONFIG_ETS).
+
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+
+compact_loop(Parent) ->
+    {ok, _} = couch_server:all_databases(
+        fun(DbName, Acc) ->
+            case ets:info(?CONFIG_ETS, size) =:= 0 of
+            true ->
+                {stop, Acc};
+            false ->
+                case get_db_config(DbName) of
+                nil ->
+                    ok;
+                {ok, Config} ->
+                    case check_period(Config) of
+                    true ->
+                        maybe_compact_db(DbName, Config);
+                    false ->
+                        ok
+                    end
+                end,
+                {ok, Acc}
+            end
+        end, ok),
+    case ets:info(?CONFIG_ETS, size) =:= 0 of
+    true ->
+        receive {Parent, have_config} -> ok end;
+    false ->
+        PausePeriod = list_to_integer(
+            couch_config:get("compaction_daemon", "check_interval", "300")),
+        ok = timer:sleep(PausePeriod * 1000)
+    end,
+    compact_loop(Parent).
+
+
+maybe_compact_db(DbName, Config) ->
+    case (catch couch_db:open_int(DbName, [{user_ctx, #user_ctx{roles=[<<"_admin">>]}}])) of
+    {ok, Db} ->
+        DDocNames = db_ddoc_names(Db),
+        case can_db_compact(Config, Db) of
+        true ->
+            {ok, DbCompactPid} = couch_db:start_compact(Db),
+            TimeLeft = compact_time_left(Config),
+            case Config#config.parallel_view_compact of
+            true ->
+                ViewsCompactPid = spawn_link(fun() ->
+                    maybe_compact_views(DbName, DDocNames, Config)
+                end),
+                ViewsMonRef = erlang:monitor(process, ViewsCompactPid);
+            false ->
+                ViewsCompactPid = nil,
+                ViewsMonRef = nil
+            end,
+            DbMonRef = erlang:monitor(process, DbCompactPid),
+            receive
+            {'DOWN', DbMonRef, process, _, normal} ->
+                couch_db:close(Db),
+                case Config#config.parallel_view_compact of
+                true ->
+                    ok;
+                false ->
+                    maybe_compact_views(DbName, DDocNames, Config)
+                end;
+            {'DOWN', DbMonRef, process, _, Reason} ->
+                couch_db:close(Db),
+                ?LOG_ERROR("Compaction daemon - an error ocurred while"
+                    " compacting the database `~s`: ~p", [DbName, Reason])
+            after TimeLeft ->
+                ?LOG_INFO("Compaction daemon - canceling compaction for database"
+                    " `~s` because it's exceeding the allowed period.",
+                    [DbName]),
+                erlang:demonitor(DbMonRef, [flush]),
+                ok = couch_db:cancel_compact(Db),
+                couch_db:close(Db)
+            end,
+            case ViewsMonRef of
+            nil ->
+                ok;
+            _ ->
+                receive
+                {'DOWN', ViewsMonRef, process, _, _Reason} ->
+                    ok
+                after TimeLeft + 1000 ->
+                    % Under normal circunstances, the view compaction process
+                    % should have finished already.
+                    erlang:demonitor(ViewsMonRef, [flush]),
+                    unlink(ViewsCompactPid),
+                    exit(ViewsCompactPid, kill)
+                end
+            end;
+        false ->
+            couch_db:close(Db),
+            maybe_compact_views(DbName, DDocNames, Config)
+        end;
+    _ ->
+        ok
+    end.
+
+
+maybe_compact_views(_DbName, [], _Config) ->
+    ok;
+maybe_compact_views(DbName, [DDocName | Rest], Config) ->
+    case check_period(Config) of
+    true ->
+        case maybe_compact_view(DbName, DDocName, Config) of
+        ok ->
+            maybe_compact_views(DbName, Rest, Config);
+        timeout ->
+            ok
+        end;
+    false ->
+        ok
+    end.
+
+
+db_ddoc_names(Db) ->
+    {ok, _, DDocNames} = couch_db:enum_docs(
+        Db,
+        fun(#full_doc_info{id = <<"_design/", _/binary>>, deleted = true}, _, Acc) ->
+            {ok, Acc};
+        (#full_doc_info{id = <<"_design/", Id/binary>>}, _, Acc) ->
+            {ok, [Id | Acc]};
+        (_, _, Acc) ->
+            {stop, Acc}
+        end, [], [{start_key, <<"_design/">>}, {end_key_gt, <<"_design0">>}]),
+    DDocNames.
+
+
+maybe_compact_view(DbName, GroupId, Config) ->
+    DDocId = <<"_design/", GroupId/binary>>,
+    case (catch couch_mrview:get_info(DbName, DDocId)) of
+    {ok, GroupInfo} ->
+        case can_view_compact(Config, DbName, GroupId, GroupInfo) of
+        true ->
+            {ok, MonRef} = couch_mrview:compact(DbName, DDocId, [monitor]),
+            TimeLeft = compact_time_left(Config),
+            receive
+            {'DOWN', MonRef, process, _, normal} ->
+                ok;
+            {'DOWN', MonRef, process, _, Reason} ->
+                ?LOG_ERROR("Compaction daemon - an error ocurred while compacting"
+                    " the view group `~s` from database `~s`: ~p",
+                    [GroupId, DbName, Reason]),
+                ok
+            after TimeLeft ->
+                ?LOG_INFO("Compaction daemon - canceling the compaction for the "
+                    "view group `~s` of the database `~s` because it's exceeding"
+                    " the allowed period.", [GroupId, DbName]),
+                erlang:demonitor(MonRef, [flush]),
+                ok = couch_mrview:cancel_compaction(DbName, DDocId),
+                timeout
+            end;
+        false ->
+            ok
+        end;
+    Error ->
+        ?LOG_ERROR("Error opening view group `~s` from database `~s`: ~p",
+            [GroupId, DbName, Error]),
+        ok
+    end.
+
+
+compact_time_left(#config{cancel = false}) ->
+    infinity;
+compact_time_left(#config{period = nil}) ->
+    infinity;
+compact_time_left(#config{period = #period{to = {ToH, ToM} = To}}) ->
+    {H, M, _} = time(),
+    case To > {H, M} of
+    true ->
+        ((ToH - H) * 60 * 60 * 1000) + (abs(ToM - M) * 60 * 1000);
+    false ->
+        ((24 - H + ToH) * 60 * 60 * 1000) + (abs(ToM - M) * 60 * 1000)
+    end.
+
+
+get_db_config(DbName) ->
+    case ets:lookup(?CONFIG_ETS, DbName) of
+    [] ->
+        case ets:lookup(?CONFIG_ETS, <<"_default">>) of
+        [] ->
+            nil;
+        [{<<"_default">>, Config}] ->
+            {ok, Config}
+        end;
+    [{DbName, Config}] ->
+        {ok, Config}
+    end.
+
+
+can_db_compact(#config{db_frag = Threshold} = Config, Db) ->
+    case check_period(Config) of
+    false ->
+        false;
+    true ->
+        {ok, DbInfo} = couch_db:get_db_info(Db),
+        {Frag, SpaceRequired} = frag(DbInfo),
+        ?LOG_DEBUG("Fragmentation for database `~s` is ~p%, estimated space for"
+           " compaction is ~p bytes.", [Db#db.name, Frag, SpaceRequired]),
+        case check_frag(Threshold, Frag) of
+        false ->
+            false;
+        true ->
+            Free = free_space(couch_config:get("couchdb", "database_dir")),
+            case Free >= SpaceRequired of
+            true ->
+                true;
+            false ->
+                ?LOG_WARN("Compaction daemon - skipping database `~s` "
+                    "compaction: the estimated necessary disk space is about ~p"
+                    " bytes but the currently available disk space is ~p bytes.",
+                   [Db#db.name, SpaceRequired, Free]),
+                false
+            end
+        end
+    end.
+
+can_view_compact(Config, DbName, GroupId, GroupInfo) ->
+    case check_period(Config) of
+    false ->
+        false;
+    true ->
+        case couch_util:get_value(updater_running, GroupInfo) of
+        true ->
+            false;
+        false ->
+            {Frag, SpaceRequired} = frag(GroupInfo),
+            ?LOG_DEBUG("Fragmentation for view group `~s` (database `~s`) is "
+                "~p%, estimated space for compaction is ~p bytes.",
+                [GroupId, DbName, Frag, SpaceRequired]),
+            case check_frag(Config#config.view_frag, Frag) of
+            false ->
+                false;
+            true ->
+                Free = free_space(couch_index_util:root_dir()),
+                case Free >= SpaceRequired of
+                true ->
+                    true;
+                false ->
+                    ?LOG_WARN("Compaction daemon - skipping view group `~s` "
+                        "compaction (database `~s`): the estimated necessary "
+                        "disk space is about ~p bytes but the currently available"
+                        " disk space is ~p bytes.",
+                        [GroupId, DbName, SpaceRequired, Free]),
+                    false
+                end
+            end
+        end
+    end.
+
+
+check_period(#config{period = nil}) ->
+    true;
+check_period(#config{period = #period{from = From, to = To}}) ->
+    {HH, MM, _} = erlang:time(),
+    case From < To of
+    true ->
+        ({HH, MM} >= From) andalso ({HH, MM} < To);
+    false ->
+        ({HH, MM} >= From) orelse ({HH, MM} < To)
+    end.
+
+
+check_frag(nil, _) ->
+    true;
+check_frag(Threshold, Frag) ->
+    Frag >= Threshold.
+
+
+frag(Props) ->
+    FileSize = couch_util:get_value(disk_size, Props),
+    MinFileSize = list_to_integer(
+        couch_config:get("compaction_daemon", "min_file_size", "131072")),
+    case FileSize < MinFileSize of
+    true ->
+        {0, FileSize};
+    false ->
+        case couch_util:get_value(data_size, Props) of
+        null ->
+            {100, FileSize};
+        0 ->
+            {0, FileSize};
+        DataSize ->
+            Frag = round(((FileSize - DataSize) / FileSize * 100)),
+            {Frag, space_required(DataSize)}
+        end
+    end.
+
+% Rough, and pessimistic, estimation of necessary disk space to compact a
+% database or view index.
+space_required(DataSize) ->
+    round(DataSize * 2.0).
+
+
+load_config() ->
+    lists:foreach(
+        fun({DbName, ConfigString}) ->
+            case parse_config(DbName, ConfigString) of
+            {ok, Config} ->
+                true = ets:insert(?CONFIG_ETS, {?l2b(DbName), Config});
+            error ->
+                ok
+            end
+        end,
+        couch_config:get("compactions")).
+
+parse_config(DbName, ConfigString) ->
+    case (catch do_parse_config(ConfigString)) of
+    {ok, Conf} ->
+        {ok, Conf};
+    incomplete_period ->
+        ?LOG_ERROR("Incomplete period ('to' or 'from' missing) in the compaction"
+            " configuration for database `~s`", [DbName]),
+        error;
+    _ ->
+        ?LOG_ERROR("Invalid compaction configuration for database "
+            "`~s`: `~s`", [DbName, ConfigString]),
+        error
+    end.
+
+do_parse_config(ConfigString) ->
+    {ok, ConfProps} = couch_util:parse_term(ConfigString),
+    {ok, #config{period = Period} = Conf} = config_record(ConfProps, #config{}),
+    case Period of
+    nil ->
+        {ok, Conf};
+    #period{from = From, to = To} when From =/= nil, To =/= nil ->
+        {ok, Conf};
+    #period{} ->
+        incomplete_period
+    end.
+
+config_record([], Config) ->
+    {ok, Config};
+
+config_record([{db_fragmentation, V} | Rest], Config) ->
+    [Frag] = string:tokens(V, "%"),
+    config_record(Rest, Config#config{db_frag = list_to_integer(Frag)});
+
+config_record([{view_fragmentation, V} | Rest], Config) ->
+    [Frag] = string:tokens(V, "%"),
+    config_record(Rest, Config#config{view_frag = list_to_integer(Frag)});
+
+config_record([{from, V} | Rest], #config{period = Period0} = Config) ->
+    Time = parse_time(V),
+    Period = case Period0 of
+    nil ->
+        #period{from = Time};
+    #period{} ->
+        Period0#period{from = Time}
+    end,
+    config_record(Rest, Config#config{period = Period});
+
+config_record([{to, V} | Rest], #config{period = Period0} = Config) ->
+    Time = parse_time(V),
+    Period = case Period0 of
+    nil ->
+        #period{to = Time};
+    #period{} ->
+        Period0#period{to = Time}
+    end,
+    config_record(Rest, Config#config{period = Period});
+
+config_record([{strict_window, true} | Rest], Config) ->
+    config_record(Rest, Config#config{cancel = true});
+
+config_record([{strict_window, false} | Rest], Config) ->
+    config_record(Rest, Config#config{cancel = false});
+
+config_record([{parallel_view_compaction, true} | Rest], Config) ->
+    config_record(Rest, Config#config{parallel_view_compact = true});
+
+config_record([{parallel_view_compaction, false} | Rest], Config) ->
+    config_record(Rest, Config#config{parallel_view_compact = false}).
+
+
+parse_time(String) ->
+    [HH, MM] = string:tokens(String, ":"),
+    {list_to_integer(HH), list_to_integer(MM)}.
+
+
+free_space(Path) ->
+    DiskData = lists:sort(
+        fun({PathA, _, _}, {PathB, _, _}) ->
+            length(filename:split(PathA)) > length(filename:split(PathB))
+        end,
+        disksup:get_disk_data()),
+    free_space_rec(abs_path(Path), DiskData).
+
+free_space_rec(_Path, []) ->
+    undefined;
+free_space_rec(Path, [{MountPoint0, Total, Usage} | Rest]) ->
+    MountPoint = abs_path(MountPoint0),
+    case MountPoint =:= string:substr(Path, 1, length(MountPoint)) of
+    false ->
+        free_space_rec(Path, Rest);
+    true ->
+        trunc(Total - (Total * (Usage / 100))) * 1024
+    end.
+
+abs_path(Path0) ->
+    Path = filename:absname(Path0),
+    case lists:last(Path) of
+    $/ ->
+        Path;
+    _ ->
+        Path ++ "/"
+    end.

http://git-wip-us.apache.org/repos/asf/couchdb/blob/add91738/apps/couch/src/couch_compress.erl
----------------------------------------------------------------------
diff --git a/apps/couch/src/couch_compress.erl b/apps/couch/src/couch_compress.erl
new file mode 100644
index 0000000..ac386fd
--- /dev/null
+++ b/apps/couch/src/couch_compress.erl
@@ -0,0 +1,84 @@
+% 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_compress).
+
+-export([compress/2, decompress/1, is_compressed/2]).
+-export([get_compression_method/0]).
+
+-include("couch_db.hrl").
+
+% binaries compressed with snappy have their first byte set to this value
+-define(SNAPPY_PREFIX, 1).
+% Term prefixes documented at:
+%      http://www.erlang.org/doc/apps/erts/erl_ext_dist.html
+-define(TERM_PREFIX, 131).
+-define(COMPRESSED_TERM_PREFIX, 131, 80).
+
+
+get_compression_method() ->
+    case couch_config:get("couchdb", "file_compression") of
+    undefined ->
+        ?DEFAULT_COMPRESSION;
+    Method1 ->
+        case string:tokens(Method1, "_") of
+        [Method] ->
+            list_to_existing_atom(Method);
+        [Method, Level] ->
+            {list_to_existing_atom(Method), list_to_integer(Level)}
+        end
+    end.
+
+
+compress(<<?SNAPPY_PREFIX, _/binary>> = Bin, snappy) ->
+    Bin;
+compress(<<?SNAPPY_PREFIX, _/binary>> = Bin, Method) ->
+    compress(decompress(Bin), Method);
+compress(<<?TERM_PREFIX, _/binary>> = Bin, Method) ->
+    compress(decompress(Bin), Method);
+compress(Term, none) ->
+    ?term_to_bin(Term);
+compress(Term, {deflate, Level}) ->
+    term_to_binary(Term, [{minor_version, 1}, {compressed, Level}]);
+compress(Term, snappy) ->
+    Bin = ?term_to_bin(Term),
+    try
+        {ok, CompressedBin} = snappy:compress(Bin),
+        case byte_size(CompressedBin) < byte_size(Bin) of
+        true ->
+            <<?SNAPPY_PREFIX, CompressedBin/binary>>;
+        false ->
+            Bin
+        end
+    catch exit:snappy_nif_not_loaded ->
+        Bin
+    end.
+
+
+decompress(<<?SNAPPY_PREFIX, Rest/binary>>) ->
+    {ok, TermBin} = snappy:decompress(Rest),
+    binary_to_term(TermBin);
+decompress(<<?TERM_PREFIX, _/binary>> = Bin) ->
+    binary_to_term(Bin).
+
+
+is_compressed(<<?SNAPPY_PREFIX, _/binary>>, Method) ->
+    Method =:= snappy;
+is_compressed(<<?COMPRESSED_TERM_PREFIX, _/binary>>, {deflate, _Level}) ->
+    true;
+is_compressed(<<?COMPRESSED_TERM_PREFIX, _/binary>>, _Method) ->
+    false;
+is_compressed(<<?TERM_PREFIX, _/binary>>, Method) ->
+    Method =:= none;
+is_compressed(Term, _Method) when not is_binary(Term) ->
+    false.
+


Mime
View raw message