CouchProjects/CouchDb/couch_fabric.erl

   1 %% CouchDb
   2 %% Copyright (C) 2006  Damien Katz
   3 %%
   4 %% This program is free software; you can redistribute it and/or
   5 %% modify it under the terms of the GNU General Public License
   6 %% as published by the Free Software Foundation; either version 2
   7 %% of the License, or (at your option) any later version.
   8 %%
   9 %% This program is distributed in the hope that it will be useful,
  10 %% but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 %% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 %% GNU General Public License for more details.
  13 %%
  14 %% You should have received a copy of the GNU General Public License
  15 %% along with this program; if not, write to the Free Software
  16 %% Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  17
  18 -module(couch_fabric).
  19 -behaviour(gen_server).
  20
  21 -include("couch_db.hrl").
  22
  23 -export([start_link/1, test/0]).
  24
  25 -export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3,stop/0]).
  26 -export([check/2, check_all/1, table_compute/2, table_columns/1]).
  27
  28 -define(ERR_HANDLE, {Port, {exit_status, Status}} -> {stop, {unknown_error, Status}, {unknown_error, Status}, Port}).
  29
  30 start_link(FabricExec) ->
  31         gen_server:start_link({local, couch_fabric}, couch_fabric, FabricExec, []).
  32
  33 stop() ->
  34         exit(whereis(couch_fabric), close).
  35
  36 check(Formula, Type) ->
  37         {ok, [Result]} = check_all([{Formula,Type}]),
  38         Result.
  39
  40 check_all(Formulas) ->
  41         gen_server:call(couch_fabric, {check, Formulas}).
  42
  43 timeout() ->
  44         % in Msecs. maybe need to make this a configurable value
  45         30000.
  46
  47 table_compute(Docs, Formulas) ->
  48         gen_server:call(couch_fabric, {table_compute, Docs, Formulas}, timeout()).
  49
  50 table_columns(Formulas) ->
  51         gen_server:call(couch_fabric, {table_columns, Formulas}).
  52
  53 init(FabricExec) ->
  54         Cmd =
  55         case erlang:system_info(os_type) of
  56         {unix, _} ->
  57             lists:flatten(io_lib:format("~s launched_from_erlang", [FabricExec]));
  58         {win32, _} ->
  59             lists:flatten(io_lib:format("cmd /c\"~s\" launched_from_erlang", [FabricExec]))
  60     end,
  61         Port = open_port({spawn, Cmd}, [{packet, 4}, exit_status, hide]),
  62         {ok, Port}.
  63
  64 terminate(_Reason, _Server) ->
  65         ok.
  66
  67
  68 handle_call({check, Formulas}, _From, Port) ->
  69         % send the formulas
  70         lists:foreach(fun({Formula, Type}) ->
  71                         send_dword(Port, 1), % indicate compilation
  72                         case Type of
  73                         simple ->
  74                                 send_dword(Port, 0);
  75                         table ->
  76                                 send_dword(Port, 1)
  77                         end,
  78                         send_string(Port, Formula)
  79                 end,
  80                 Formulas),
  81
  82         Results = lists:map(fun(_Formula) ->
  83                         case get_dword(Port) of
  84                         0 ->
  85                                 ok;
  86                         1 ->
  87                                 get_error(Port)
  88                         end
  89                 end,
  90                 Formulas),
  91         {reply, {ok, Results}, Port};
  92 handle_call({table_columns, Formulas}, _From, Port) ->
  93         % indicate table columns
  94         send_dword(Port, 3),
  95         % send the number of formulas
  96         send_dword(Port, length(Formulas)),
  97         % send the formulas
  98         Results = lists:map(fun(Formula) ->
  99                         send_string(Port, Formula),
 100                         case get_dword(Port) of
 101                         0 -> % compile error
 102                                 get_error(Port);
 103                         1 ->
 104                                 Names = get_list(Port),
 105                                 {ok, [list_to_atom(Name) || Name <- Names]}
 106                         end
 107                 end,
 108                 Formulas),
 109         {reply, {ok, Results}, Port};
 110 handle_call({table_compute, Docs, Formulas}, _From, Port) ->
 111         % indicate table computation
 112         send_dword(Port, 2),
 113         % send the number of formulas
 114         send_dword(Port, length(Formulas)),
 115         % send the formulas
 116         lists:foreach(fun(Formula) ->
 117                         send_string(Port, Formula)
 118                 end,
 119                 Formulas),
 120
 121         % send the number of documents
 122         send_dword(Port, length(Docs)),
 123
 124         % send the documents
 125         lists:foreach(fun(Doc) ->
 126                         send_doc(Port, Doc)
 127                 end,
 128                 Docs),
 129
 130         % get the results
 131         DocResults =
 132         lists:map(fun(_Doc) ->
 133                         % loop for each document and receive the corresponding response
 134                         lists:map(fun(_Formula) ->
 135                                 % loop for each formula
 136                                 % did the formula match?
 137                                 case get_dword(Port) of
 138                                 0 -> % no match
 139                                         no_match;
 140                                 1 -> % did match
 141                                         {ok, get_values(Port)}
 142                                 end
 143                         end,
 144                         Formulas)
 145                 end,
 146                 Docs),
 147         {reply, {ok, DocResults}, Port}.
 148
 149 handle_cast(_Whatever, Port) ->
 150         {noreply, Port}.
 151
 152 handle_info({Port, {exit_status, Status}}, Port) ->
 153         {stop, {fabric_server_exited, Status}, Port};
 154 handle_info(_Whatever, Port) ->
 155         {noreply, Port}.
 156
 157 code_change(_OldVsn, State, _Extra) ->
 158         {ok, State}.
 159
 160 send_dword(Port, Num) when Num =< 16#FFFFFFFF, Num >= 0 ->
 161         port_command(Port, <<Num:32/native >>).
 162
 163 get_dword(Port) ->
 164         receive
 165         {Port, {data, Bytes}} ->
 166                 <<Result:32/native>> = list_to_binary(Bytes),
 167                 Result;
 168         ?ERR_HANDLE
 169         end.
 170
 171 get_float(Port) ->
 172         receive
 173         {Port, {data, Bytes}} ->
 174                 <<Result:64/float>> = list_to_binary(Bytes),
 175                 Result;
 176         ?ERR_HANDLE
 177         end.
 178
 179 get_error(Port) ->
 180         ErrorAtom = get_atom(Port),
 181         ErrorString = get_string(Port),
 182         {ErrorAtom, ErrorString}.
 183
 184
 185 send_float(Port, Num) ->
 186         port_command(Port, <<Num:64/float>>).
 187
 188 send_string(Port, []) ->
 189         % can't send null, send a single zeroed byte
 190         port_command(Port, [0]);
 191 send_string(Port, String) ->
 192         port_command(Port, String).
 193
 194 get_string(Port) ->
 195         receive
 196         {Port, {data, [0]}} ->
 197                 % we can't send or receive null strings, so to represent a
 198                 % null string we send a zeroed 1 byte string and then
 199                 % convert to a null string
 200                 [];
 201         {Port, {data, String}} ->
 202                 String;
 203         ?ERR_HANDLE
 204         end.
 205
 206 get_atom(Port) ->
 207         list_to_atom(get_string(Port)).
 208
 209 send_doc(Port, Doc) ->
 210         Fields = [
 211                 {"$docid", [Doc#doc.uuid]},
 212                 {"$revisions", Doc#doc.revisions}
 213                 | dict:to_list(Doc#doc.summary_fields)],
 214         send_fields(Port, Fields).
 215
 216
 217
 218 send_fields(Port, Fields) ->
 219         % send the number of fields
 220         send_dword(Port, length(Fields)),
 221         % send the fields
 222         lists:foreach(fun({FieldName, FieldValue}) ->
 223                         send_string(Port, FieldName),
 224                         send_list(Port, FieldValue)
 225                 end,
 226                 Fields).
 227
 228 get_values(Port) ->
 229         get_values2(Port, get_dword(Port), []).
 230
 231 get_values2(_Port, 0, Acc) ->
 232         lists:reverse(Acc);
 233 get_values2(Port, Count, Acc) ->
 234         Value = get_list(Port),
 235         get_values2(Port, Count-1, [Value | Acc]).
 236
 237
 238 send_list(Port, List) ->
 239         send_dword(Port, length(List)),
 240         send_list2(Port, List).
 241
 242 send_list2(_Port, []) ->
 243         ok;
 244 send_list2(Port, [Element | Rest]) when is_list(Element) ->
 245         send_dword(Port, 0), % indicate string
 246         send_string(Port, Element),
 247         send_list2(Port, Rest);
 248 send_list2(Port, [Element | Rest]) when is_number(Element) ->
 249         send_dword(Port, 1), % indicate float
 250         send_float(Port, Element),
 251         send_list2(Port, Rest).
 252
 253 get_list(Port) ->
 254         get_list2(Port, get_dword(Port), []).
 255
 256 get_list2(_Port, 0, Acc) ->
 257         lists:reverse(Acc);
 258 get_list2(Port, Count, Acc) ->
 259         Element =
 260         case get_dword(Port) of
 261         0 ->
 262                 get_string(Port);
 263         1 ->
 264                 get_float(Port)
 265         end,
 266         get_list2(Port, Count - 1, [Element | Acc]).
 267
 268 test() ->
 269         ok = check("damien", simple),
 270         io:format("Formula compiled~n"),
 271         {ok, Columns} = table_columns([" select Foo == \"bar\"; COLUMN Food := Foo + Bar2", " select Foo != \"bar\";"]),
 272         io:format("Table Columns: ~p~n", [Columns]),
 273         Doc = couch_doc:new([{"Foo", ["bar"]}, {"Bar2", ["bar2"]}, {"Num", [1.0]}, {"gak", ["glopf"]}]),
 274         Doc2 = couch_doc:new([{"Foo", ["bar"]}, {"Bar2", ["bbbbbbbabs"]}, {"Num", [2.0]}, {"gak", ["glopf"]}]),
 275         {ok, [Result, ResultB]} = table_compute([Doc, Doc2], [" select Foo == \"bar\"; COLUMN Food := Num + 2", " select Foo != \"bar\";"]),
 276          io:format("~p~n", [Result]),
 277           io:format("ResultB: ~p~n", [ResultB]),
 278         {ok, [Result2]} = table_compute([Doc], [" select Foo != \"bar\";"]),
 279         io:format("~p~n", [Result2]),
 280         {ok, [Result3]} = table_compute([Doc], [" select Num == 1"]),
 281         io:format("~p~n", [Result3]),
 282         ok.