Merge pull request #24 from lpgauth/perf/decode

Optimize decode

include/marina.hrl

@@ -3,6 +3,7 @@
 -define(CLIENT, marina_client).
 -define(GET_ENV(Key, Default), application:get_env(?APP, Key, Default)).
 -define(ETS_TABLE_CACHE, marina_cache).
+-define(LOOKUP(Key, List), shackle_utils:lookup(Key, List, undefined)).
 
 %% defaults
 -define(DEFAULT_BACKLOG_SIZE, 1024).

src/marina_body.erl

@@ -11,27 +11,22 @@
 %% public
 -spec decode(frame()) -> {ok, term()} | {error, atom()}.
 
-decode(#frame {opcode = ?OP_READY}) ->
-    {ok, undefined};
-decode(#frame {
-        opcode = ?OP_ERROR,
-        body = Body
-    }) ->
+decode(#frame {flags = 0, body = Body, opcode = Opcode}) ->
+    decode(Opcode, Body);
+decode(#frame {flags = 1, body = Body, opcode = Opcode}) ->
+    {ok, Body2} = marina_utils:unpack(Body),
+    decode(Opcode, Body2).
 
+%% private
+decode(?OP_READY, _) ->
+    {ok, undefined};
+decode(?OP_ERROR, Body) ->
     {Code, Rest} = marina_types:decode_int(Body),
     {Msg, _Rest2} = marina_types:decode_string(Rest),
     {error, {Code, Msg}};
-decode(#frame {
-        opcode = ?OP_RESULT,
-        body = <<1:32/integer>>
-    }) ->
-
+decode(?OP_RESULT, <<1:32/integer>>) ->
     {ok, undefined};
-decode(#frame {
-        opcode = ?OP_RESULT,
-        body = <<2:32/integer, Rest/binary>>
-    }) ->
-
+decode(?OP_RESULT, <<2:32/integer, Rest/binary>>) ->
     {Metadata, Rest2} = decode_result_metadata(Rest),
     {RowsCount, Rest3} = marina_types:decode_int(Rest2),
     ColumnsCount = Metadata#result_metadata.columns_count,
@@ -42,28 +37,15 @@ decode(#frame {
         rows_count = RowsCount,
         rows = Rows
     }};
-decode(#frame {
-        opcode = ?OP_RESULT,
-        body = <<3:32/integer, Rest/binary>>
-    }) ->
-
+decode(?OP_RESULT, <<3:32/integer, Rest/binary>>) ->
     {Keyspace, <<>>} = marina_types:decode_string(Rest),
     {ok, Keyspace};
-decode(#frame {
-        opcode = ?OP_RESULT,
-        body = <<4:32/integer, Rest/binary>>
-    }) ->
-
+decode(?OP_RESULT, <<4:32/integer, Rest/binary>>) ->
     {Id, Rest2} = marina_types:decode_short_bytes(Rest),
     {_Metadata, Rest3} = decode_result_metadata(Rest2),
     {_ResultMetadata, <<>>} = decode_result_metadata(Rest3),
-
     {ok, Id};
-decode(#frame {
-        opcode = ?OP_RESULT,
-        body = <<5:32/integer, Rest/binary>>
-    }) ->
-
+decode(?OP_RESULT, <<5:32/integer, Rest/binary>>) ->
     {ChangeType, Rest2} = marina_types:decode_string(Rest),
     {Target, Rest3} = marina_types:decode_string(Rest2),
 
@@ -83,7 +65,6 @@ decode(#frame {
 
     {ok, {ChangeType, Target, Options}}.
 
-%% private
 decode_columns(Bin, Count) ->
     decode_columns(Bin, Count, []).
 

src/marina_frame.erl

@@ -18,27 +18,16 @@ decode(Bin) ->
 
 -spec encode(frame()) -> iolist().
 
-encode(#frame {
-        stream = Stream,
-        opcode = ?OP_STARTUP,
-        body = Body
-    }) ->
-
-    [<<0:1, ?PROTO_VERSION:7/unsigned-integer,
-        ?DEFAULT_FLAGS:8/unsigned-integer, Stream:16/signed-integer,
-        ?OP_STARTUP:8/unsigned-integer,
-        (iolist_size(Body)):32/unsigned-integer>>, Body];
 encode(#frame {
         flags = Flags,
         stream = Stream,
         opcode = Opcode,
         body = Body
     }) ->
 
-    Body2 = encode_body(Flags, Body),
     [<<0:1, ?PROTO_VERSION:7/unsigned-integer, Flags:8/unsigned-integer,
         Stream:16/signed-integer, Opcode:8/unsigned-integer,
-        (iolist_size(Body2)):32/unsigned-integer>>, Body2].
+        (iolist_size(Body)):32/unsigned-integer>>, Body].
 
 -spec pending_size(binary()) -> pos_integer() | undefined.
 
@@ -52,28 +41,15 @@ pending_size(_) ->
     undefined.
 
 %% private
-decode_body(0, Body) ->
-    Body;
-decode_body(1, Body) ->
-    {ok, Body2} = marina_utils:unpack(Body),
-    Body2.
-
 decode(<<1:1, ?PROTO_VERSION:7/unsigned-integer, Flags:8/unsigned-integer,
     Stream:16/signed-integer, Opcode:8/unsigned-integer,
     Length:32/unsigned-integer, Body:Length/binary, Rest/binary>>, Acc) ->
 
-    Body2 = decode_body(Flags, Body),
     decode(Rest, [#frame {
         flags = Flags,
         stream = Stream,
         opcode = Opcode,
-        body = Body2
+        body = Body
     } | Acc]);
 decode(Rest, Acc) ->
     {Rest, Acc}.
-
-encode_body(0, Body) ->
-    Body;
-encode_body(1, Body) ->
-    {ok, Body2} = marina_utils:pack(Body),
-    Body2.

src/marina_request.erl

@@ -11,59 +11,68 @@
     startup/1
 ]).
 
--define(LOOKUP(Key, List), shackle_utils:lookup(Key, List, undefined)).
-
 %% public
 -spec execute(stream(), [frame_flag()], statement_id(), [value()],
     consistency(), [flag()]) -> iolist().
 
 execute(Stream, FrameFlags, StatementId, Values, ConsistencyLevel, Flags) ->
+    FrameFlags2 = frame_flags(FrameFlags),
     {Flags2, Values2} = flags_and_values(Flags, Values),
     {PageSize, PagingState} = paging(Flags),
+    Body2 = encode_body(FrameFlags2,
+        [marina_types:encode_short_bytes(StatementId),
+        marina_types:encode_short(ConsistencyLevel),
+        Flags2, Values2, PageSize, PagingState]),
 
     marina_frame:encode(#frame {
         stream = Stream,
         opcode = ?OP_EXECUTE,
-        flags = frame_flags(FrameFlags),
-        body = [marina_types:encode_short_bytes(StatementId),
-            marina_types:encode_short(ConsistencyLevel),
-            Flags2, Values2, PageSize, PagingState]
+        flags = FrameFlags2,
+        body = Body2
     }).
 
 -spec prepare(stream(), [frame_flag()], query()) -> iolist().
 
 prepare(Stream, FrameFlags, Query) ->
+    FrameFlags2 = frame_flags(FrameFlags),
+    Body2 = encode_body(FrameFlags2, [marina_types:encode_long_string(Query)]),
+
     marina_frame:encode(#frame {
         stream = Stream,
         opcode = ?OP_PREPARE,
-        flags = frame_flags(FrameFlags),
-        body = [marina_types:encode_long_string(Query)]
+        flags = FrameFlags2,
+        body = Body2
     }).
 
 -spec query(stream(), [frame_flag()], query(), [value()], consistency(),
     [flag()]) -> iolist().
 
 query(Stream, FrameFlags, Query, Values, ConsistencyLevel, Flags) ->
+    FrameFlags2 = frame_flags(FrameFlags),
     {Flags2, Values2} = flags_and_values(Flags, Values),
     {PageSize, PagingState} = paging(Flags),
+    Body2 = encode_body(FrameFlags2, [marina_types:encode_long_string(Query),
+        marina_types:encode_short(ConsistencyLevel), Flags2, Values2, PageSize,
+        PagingState]),
 
     marina_frame:encode(#frame {
         stream = Stream,
         opcode = ?OP_QUERY,
-        flags = frame_flags(FrameFlags),
-        body = [marina_types:encode_long_string(Query),
-            marina_types:encode_short(ConsistencyLevel), Flags2,
-            Values2, PageSize, PagingState]
+        flags = FrameFlags2,
+        body = Body2
     }).
 
 -spec startup([frame_flag()]) -> iolist().
 
 startup(FrameFlags) ->
     FrameFlags2 = frame_flags(FrameFlags),
     Body = case FrameFlags2 of
-        1 -> [?CQL_VERSION, ?LZ4_COMPRESSION];
-        0 -> [?CQL_VERSION]
+        1 ->
+            [?CQL_VERSION, ?LZ4_COMPRESSION];
+        0 ->
+            [?CQL_VERSION]
     end,
+
     marina_frame:encode(#frame {
         stream = ?DEFAULT_STREAM,
         opcode = ?OP_STARTUP,
@@ -72,6 +81,12 @@ startup(FrameFlags) ->
     }).
 
 %% private
+encode_body(0, Body) ->
+    Body;
+encode_body(1, Body) ->
+    {ok, Body2} = marina_utils:pack(Body),
+    Body2.
+
 flags([]) ->
     0;
 flags([{values, true} | T]) ->

src/marina_types.erl

@@ -34,10 +34,8 @@
 
 decode_bytes(<<255, 255, 255, 255, Rest/binary>>) ->
     {null, Rest};
-decode_bytes(Bin) ->
-    {Pos, Rest} = decode_int(Bin),
-    <<Value:Pos/binary, Rest2/binary>> = Rest,
-    {Value, Rest2}.
+decode_bytes(<<Pos:32, Value:Pos/binary, Rest/binary>>) ->
+    {Value, Rest}.
 
 -spec decode_int(binary()) -> {integer(), binary()}.
 
@@ -63,10 +61,8 @@ decode_short(<<Value:16, Rest/binary>>) ->
 
 decode_short_bytes(<<255, 255, Rest/binary>>) ->
     {null, Rest};
-decode_short_bytes(Bin) ->
-    {Pos, Rest} = decode_short(Bin),
-    <<Value:Pos/binary, Rest2/binary>> = Rest,
-    {Value, Rest2}.
+decode_short_bytes(<<Pos:16, Value:Pos/binary, Rest/binary>>) ->
+    {Value, Rest}.
 
 -spec decode_string(binary()) -> {binary(), binary()}.
 
@@ -75,20 +71,17 @@ decode_string(Bin) ->
 
 -spec decode_string_list(binary()) -> {[binary()], binary()}.
 
-decode_string_list(Bin) ->
-    {Length, Rest} = decode_short(Bin),
+decode_string_list(<<Length:16, Rest/binary>>) ->
     decode_string_list(Rest, Length, []).
 
 -spec decode_string_map(binary()) -> {[{binary(), binary()}], binary()}.
 
-decode_string_map(Bin) ->
-    {Length, Rest} = decode_short(Bin),
+decode_string_map(<<Length:16, Rest/binary>>) ->
     decode_string_map(Rest, Length, []).
 
 -spec decode_string_multimap(binary()) -> {[{binary(), [binary()]}], binary()}.
 
-decode_string_multimap(Bin) ->
-    {Length, Rest} = decode_short(Bin),
+decode_string_multimap(<<Length:16, Rest/binary>>) ->
     decode_string_multimap(Rest, Length, []).
 
 -spec decode_uuid(binary()) -> {binary(), binary()}.
@@ -158,21 +151,19 @@ encode_string_multimap(KeyValues) ->
 %% private
 decode_string_list(Bin, 0, Acc) ->
     {lists:reverse(Acc), Bin};
-decode_string_list(Bin, Length, Acc) ->
-    {String, Rest} = decode_string(Bin),
+decode_string_list(<<Pos:16, String:Pos/binary, Rest/binary>>, Length, Acc) ->
     decode_string_list(Rest, Length - 1, [String | Acc]).
 
 decode_string_map(Bin, 0, Acc) ->
     {lists:reverse(Acc), Bin};
-decode_string_map(Bin, Length, Acc) ->
-    {Key, Rest} = decode_string(Bin),
-    {Value, Rest2} = decode_string(Rest),
-    decode_string_map(Rest2, Length - 1, [{Key, Value} | Acc]).
+decode_string_map(<<Pos:16, Key:Pos/binary, Pos2:16, Value:Pos2/binary,
+    Rest/binary>>, Length, Acc) ->
+
+    decode_string_map(Rest, Length - 1, [{Key, Value} | Acc]).
 
 decode_string_multimap(Bin, 0, Acc) ->
     {lists:reverse(Acc), Bin};
-decode_string_multimap(Bin, Length, Acc) ->
-    {Key, Rest} = decode_string(Bin),
+decode_string_multimap(<<Pos:16, Key:Pos/binary, Rest/binary>>, Length, Acc) ->
     {Values, Rest2} = decode_string_list(Rest),
     decode_string_multimap(Rest2, Length - 1, [{Key, Values} | Acc]).
 

test/marina_tests.erl

@@ -238,6 +238,7 @@ tuples_subtest() ->
 %% utils
 bootstrap() ->
     query(<<"DROP KEYSPACE test;">>),
+    timer:sleep(250),
     {ok, _} = query(<<"CREATE KEYSPACE test WITH REPLICATION =
         {'class':'SimpleStrategy', 'replication_factor':1};">>),
     {ok, _} = query(<<"CREATE TABLE test.users (key uuid, column1 text,