From 3d05f543df2a7543829ac0d02c8ce34127abd05d Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Mon, 22 Jun 2015 17:49:07 +0900 Subject: [PATCH 01/13] WIP: new test case is failing, quick fix soon --- src/machi_flu1.erl | 3 +- src/machi_pb_high_client.erl | 146 ++++++++++++++++++ ...test.erl => machi_pb_high_client_test.erl} | 31 ++-- 3 files changed, 158 insertions(+), 22 deletions(-) create mode 100644 src/machi_pb_high_client.erl rename test/{machi_pb_cr_client_test.erl => machi_pb_high_client_test.erl} (60%) diff --git a/src/machi_flu1.erl b/src/machi_flu1.erl index af507c2..2955341 100644 --- a/src/machi_flu1.erl +++ b/src/machi_flu1.erl @@ -346,7 +346,8 @@ net_server_loop(Sock, #state{flu_name=FluName, data_dir=DataDir}=S) -> http_server_hack(FluName, PutLine, Sock, S); <<"PROTOCOL-BUFFERS\n">> -> ok = gen_tcp:send(Sock, <<"OK\n">>), - ok = inet:setopts(Sock, [{packet, 4}]), + ok = inet:setopts(Sock, [{packet, 4}, + {packet_size, 33*1024*1024}]), protocol_buffers_loop(Sock, S); _ -> machi_util:verb("Else Got: ~p\n", [Line]), diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl new file mode 100644 index 0000000..def6843 --- /dev/null +++ b/src/machi_pb_high_client.erl @@ -0,0 +1,146 @@ +%% ------------------------------------------------------------------- +%% +%% Copyright (c) 2007-2015 Basho Technologies, Inc. All Rights Reserved. +%% +%% This file is provided to you 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. +%% +%% ------------------------------------------------------------------- + +%% @doc Machi PB (Protocol Buffers) high-level client (prototype, API TBD) + +-module(machi_pb_high_client). + +-include("machi.hrl"). +-include("machi_pb.hrl"). +-include("machi_projection.hrl"). + +-define(DEFAULT_TIMEOUT, 10*1000). + +-export([start_link/1, quit/1, + connected_p/1, + echo/2, echo/3]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-record(state, { + server_list :: p_srvr_dict(), + sock :: 'undefined' | port() + }). + +start_link(P_srvr_list) -> + gen_server:start_link(?MODULE, [P_srvr_list], []). + +quit(PidSpec) -> + gen_server:call(PidSpec, quit, infinity). + +connected_p(PidSpec) -> + gen_server:call(PidSpec, connected_p, infinity). + +echo(PidSpec, String) -> + echo(PidSpec, String, ?DEFAULT_TIMEOUT). + +echo(PidSpec, String, Timeout) -> + send_sync(PidSpec, {echo, String}, Timeout). + +send_sync(PidSpec, Cmd, Timeout) -> + gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). + +%%%%%%%%%%%%%%%%%%%%%%%%%%% + +init([P_srvr_list]) -> + {ok, #state{server_list=P_srvr_list}}. + +handle_call(quit, _From, S) -> + {stop, normal, ok, S}; +handle_call(connected_p, _From, #state{sock=Port}=S) + when is_port(Port) -> + {reply, true, S}; +handle_call(connected_p, _From, #state{sock=undefined}=S) -> + S2 = try_connect(S), + {reply, is_port(S2#state.sock), S2}; +handle_call({send_sync, _Cmd}, _From, #state{sock=undefined}=S) -> + {reply, not_connected, S}; +handle_call({send_sync, Cmd}, _From, S) -> + {Reply, S2} = do_send_sync(Cmd, S), + {reply, Reply, S2}; +handle_call(_Request, _From, S) -> + Reply = whaaaaaaaaaaaaaaaaaaaa, + {reply, Reply, S}. + +handle_cast(_Msg, S) -> + {noreply, S}. + +handle_info(_Info, S) -> + io:format(user, "~s:handle_info: ~p\n", [?MODULE, _Info]), + {noreply, S}. + +terminate(_Reason, _S) -> + ok. + +code_change(_OldVsn, S, _Extra) -> + {ok, S}. + +%%%%%%%%%%%%%%%%%%%%%%%%%%% + +try_connect(#state{sock=Sock}=S) when is_port(Sock) -> + S; +try_connect(#state{server_list=Ps}=S) -> + case lists:foldl(fun(_, Sock) when is_port(Sock) -> + Sock; + (P, _) -> + do_connect_to_pb_listener(P) + end, unused, Ps) of + Sock when is_port(Sock) -> + S#state{sock=Sock}; + _Else -> + S + end. + +do_connect_to_pb_listener(P) -> + try + {ok, Sock} = gen_tcp:connect(P#p_srvr.address, P#p_srvr.port, + [{packet, line}, binary, {active, false}]), + ok = gen_tcp:send(Sock, <<"PROTOCOL-BUFFERS\n">>), + {ok, <<"OK\n">>} = gen_tcp:recv(Sock, 0), + ok = inet:setopts(Sock, [{packet,4}]), + Sock + catch _X:_Y -> + io:format(user, "\n~p ~p @ ~p\n", [_X, _Y, erlang:get_stacktrace()]), + bummer + end. + +%% {Reply, S2} = do_send_sync(Cmd, S), + +do_send_sync({echo, String}, #state{sock=Sock}=S) -> + try + ReqID = <<0>>, + R1a = #mpb_request{req_id=ReqID, + echo=#mpb_echoreq{message=String}}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, echo=Echo} = _R1b -> + io:format(user, "do_send_sync ~p\n", [_R1b]), + {Echo#mpb_echoresp.message, S} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S} + end. + + diff --git a/test/machi_pb_cr_client_test.erl b/test/machi_pb_high_client_test.erl similarity index 60% rename from test/machi_pb_cr_client_test.erl rename to test/machi_pb_high_client_test.erl index 483ab9d..6aa4e6c 100644 --- a/test/machi_pb_cr_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -18,7 +18,7 @@ %% %% ------------------------------------------------------------------- --module(machi_pb_cr_client_test). +-module(machi_pb_high_client_test). -compile(export_all). -ifdef(TEST). @@ -28,45 +28,34 @@ -include("machi_projection.hrl"). -include_lib("eunit/include/eunit.hrl"). --define(C, machi_pb_cr_client). +-define(C, machi_pb_high_client). smoke_test_() -> {timeout, 5*60, fun() -> smoke_test2() end}. smoke_test2() -> Port = 5720, - Ps = [{a,#p_srvr{name=a, address="localhost", port=Port, props="./data.a"}} + Ps = [#p_srvr{name=a, address="localhost", port=Port, props="./data.a"} ], - [os:cmd("rm -rf " ++ P#p_srvr.props) || {_,P} <- Ps], + [os:cmd("rm -rf " ++ P#p_srvr.props) || P <- Ps], {ok, SupPid} = machi_flu_sup:start_link(), try [begin #p_srvr{name=Name, port=Port, props=Dir} = P, {ok, _} = machi_flu_psup:start_flu_package(Name, Port, Dir, []) - end || {_,P} <- Ps], + end || P <- Ps], [machi_chain_manager1:test_react_to_env(a_chmgr) || _ <-lists:seq(1,5)], - {_, P_a} = hd(Ps), - {ok, Sock} = gen_tcp:connect(P_a#p_srvr.address, P_a#p_srvr.port, - [{packet, line}, binary, {active, false}]), + {ok, Clnt} = ?C:start_link(Ps), try - Prefix = <<"pre">>, - Chunk1 = <<"yochunk">>, - ok = gen_tcp:send(Sock, <<"PROTOCOL-BUFFERS\n">>), - {ok, <<"OK\n">>} = gen_tcp:recv(Sock, 0), - ok = inet:setopts(Sock, [{packet,4}]), - R1a = #mpb_request{req_id= <<0>>, - echo=#mpb_echoreq{message = <<"Hello, world!">>}}, - Bin1a = machi_pb:encode_mpb_request(R1a), - ok = gen_tcp:send(Sock, Bin1a), - {ok, Bin1B} = gen_tcp:recv(Sock, 0), - R1b = machi_pb:decode_mpb_response(Bin1B), - true = is_record(R1b, mpb_response), + true = ?C:connected_p(Clnt), + String = <<"yo, dawg">>, + String = ?C:echo(Clnt, String), ok after - (catch gen_tcp:close(Sock)) + (catch ?C:quit(Clnt)) end after exit(SupPid, normal), From db7f1476b9b0fb80d678ee0aeeca6f03ddacaa9d Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Mon, 22 Jun 2015 18:04:17 +0900 Subject: [PATCH 02/13] WIP: 'echo' request works end-to-end, yay! --- src/machi_flu1.erl | 27 +++++++++++++++++++++++---- src/machi_pb_high_client.erl | 3 --- test/machi_pb_high_client_test.erl | 2 +- 3 files changed, 24 insertions(+), 8 deletions(-) diff --git a/src/machi_flu1.erl b/src/machi_flu1.erl index 2955341..63abb8e 100644 --- a/src/machi_flu1.erl +++ b/src/machi_flu1.erl @@ -1017,10 +1017,11 @@ http_harvest_headers({ok, Hdr}, Sock, Acc) -> protocol_buffers_loop(Sock, S) -> case gen_tcp:recv(Sock, 0) of - {ok, _Bin} -> - R = #mpb_response{req_id= <<"not paying any attention">>, - generic=#mpb_errorresp{code=-6, - msg="not implemented"}}, + {ok, Bin} -> + R = do_pb_request(catch machi_pb:decode_mpb_request(Bin)), + %% R = #mpb_response{req_id= <<"not paying any attention">>, + %% generic=#mpb_errorresp{code=-6, + %% msg="not implemented"}}, Resp = machi_pb:encode_mpb_response(R), ok = gen_tcp:send(Sock, Resp), protocol_buffers_loop(Sock, S); @@ -1048,3 +1049,21 @@ split_uri_options(OpsBin) -> [<<"size">>, Bin] -> {size, binary_to_integer(Bin)} end || X <- L]. + +do_pb_request(#mpb_request{req_id=ReqID, + echo=#mpb_echoreq{message=Msg}}) -> + #mpb_response{req_id=ReqID, + echo=#mpb_echoresp{message=Msg}}; +do_pb_request(#mpb_request{req_id=ReqID, + auth=#mpb_authreq{}}) -> + #mpb_response{req_id=ReqID, + generic=#mpb_errorresp{code=1, + msg="AUTH not implemented"}}; +do_pb_request(#mpb_request{req_id=ReqID}) -> + #mpb_response{req_id=ReqID, + generic=#mpb_errorresp{code=66, + msg="Unknown request"}}; +do_pb_request(_Else) -> + #mpb_response{req_id= <<>>, + generic=#mpb_errorresp{code=67, + msg="Unknown PB request"}}. diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index def6843..9e413a4 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -123,8 +123,6 @@ do_connect_to_pb_listener(P) -> bummer end. -%% {Reply, S2} = do_send_sync(Cmd, S), - do_send_sync({echo, String}, #state{sock=Sock}=S) -> try ReqID = <<0>>, @@ -135,7 +133,6 @@ do_send_sync({echo, String}, #state{sock=Sock}=S) -> {ok, Bin1B} = gen_tcp:recv(Sock, 0), case (catch machi_pb:decode_mpb_response(Bin1B)) of #mpb_response{req_id=ReqID, echo=Echo} = _R1b -> - io:format(user, "do_send_sync ~p\n", [_R1b]), {Echo#mpb_echoresp.message, S} end catch X:Y -> diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 6aa4e6c..7b552b9 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -50,7 +50,7 @@ smoke_test2() -> {ok, Clnt} = ?C:start_link(Ps), try true = ?C:connected_p(Clnt), - String = <<"yo, dawg">>, + String = "yo, dawg", String = ?C:echo(Clnt, String), ok From bb8e725c2618ee5d02704519c067d262cfe79193 Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Mon, 22 Jun 2015 18:16:15 +0900 Subject: [PATCH 03/13] WIP: 'auth' request placeholders --- src/machi_pb_high_client.erl | 38 +++++++++++++++++++++++++++--- test/machi_pb_high_client_test.erl | 4 ++++ 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index 9e413a4..a9db840 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -30,7 +30,8 @@ -export([start_link/1, quit/1, connected_p/1, - echo/2, echo/3]). + echo/2, echo/3, + auth/3, auth/4]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, @@ -56,6 +57,15 @@ echo(PidSpec, String) -> echo(PidSpec, String, Timeout) -> send_sync(PidSpec, {echo, String}, Timeout). +%% TODO: auth() is not implemented. Auth requires SSL, and this client +%% doesn't support SSL yet. This is just a placeholder & reminder. + +auth(PidSpec, User, Pass) -> + auth(PidSpec, User, Pass, ?DEFAULT_TIMEOUT). + +auth(PidSpec, User, Pass, Timeout) -> + send_sync(PidSpec, {auth, User, Pass}, Timeout). + send_sync(PidSpec, Cmd, Timeout) -> gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). @@ -132,8 +142,30 @@ do_send_sync({echo, String}, #state{sock=Sock}=S) -> ok = gen_tcp:send(Sock, Bin1a), {ok, Bin1B} = gen_tcp:recv(Sock, 0), case (catch machi_pb:decode_mpb_response(Bin1B)) of - #mpb_response{req_id=ReqID, echo=Echo} = _R1b -> - {Echo#mpb_echoresp.message, S} + #mpb_response{req_id=ReqID, echo=Echo} when Echo /= undefined -> + {Echo#mpb_echoresp.message, S}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S} + end; +do_send_sync({auth, User, Pass}, #state{sock=Sock}=S) -> + try + ReqID = <<0>>, + R1a = #mpb_request{req_id=ReqID, + auth=#mpb_authreq{user=User, password=Pass}}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, auth=Auth} when Auth /= undefined -> + {Auth#mpb_authresp.code, S}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S} end catch X:Y -> Res = {bummer, {X, Y, erlang:get_stacktrace()}}, diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 7b552b9..aa1393e 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -52,6 +52,10 @@ smoke_test2() -> true = ?C:connected_p(Clnt), String = "yo, dawg", String = ?C:echo(Clnt, String), + %% TODO: auth() is not implemented. Auth requires SSL. + %% Probably ought to put client stuff that relies on SSL into + %% a separate test module? Or separate test func? + {error, _} = ?C:auth(Clnt, "foo", "bar"), ok after From 5ef499ec737c3f4b33c8a8f9e17209786524c9bd Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 14:08:10 +0900 Subject: [PATCH 04/13] WIP: append_chunk #1 --- include/machi.hrl | 6 +-- src/machi.proto | 16 ++++---- src/machi_flu1.erl | 57 +++++++++++++++++++++++--- src/machi_flu1_client.erl | 4 +- src/machi_pb_high_client.erl | 59 +++++++++++++++++++++++++-- src/machi_util.erl | 23 +++++++---- test/machi_cr_client_test.erl | 2 +- test/machi_flu1_test.erl | 2 +- test/machi_pb_high_client_test.erl | 6 +++ test/machi_proxy_flu1_client_test.erl | 2 +- 10 files changed, 145 insertions(+), 32 deletions(-) diff --git a/include/machi.hrl b/include/machi.hrl index 64b0fc5..daf13c4 100644 --- a/include/machi.hrl +++ b/include/machi.hrl @@ -26,7 +26,7 @@ %% 0th draft of checksum typing with 1st byte. -define(CSUM_TAG_NONE, 0). % No csum provided by client --define(CSUM_TAG_CLIENT_GEN, 1). % Client-generated csum --define(CSUM_TAG_SERVER_GEN, 2). % Server-genereated csum --define(CSUM_TAG_SERVER_REGEN, 3). % Server-regenerated csum +-define(CSUM_TAG_CLIENT_SHA, 1). % Client-generated SHA1 +-define(CSUM_TAG_SERVER_SHA, 2). % Server-genereated SHA1 +-define(CSUM_TAG_SERVER_REGEN_SHA, 3). % Server-regenerated SHA1 diff --git a/src/machi.proto b/src/machi.proto index 993ff42..d4c3b82 100644 --- a/src/machi.proto +++ b/src/machi.proto @@ -41,14 +41,15 @@ enum Mpb_GeneralStatusCode { WEDGED = 2; BAD_CHECKSUM = 3; PARTITION = 4; + BAD_JOSS = 255; // Only for testing by the Taipan } // Must match with machi.hrl's values! enum Mpb_CSumType { CSUM_TAG_NONE = 0; - CSUM_TAG_CLIENT_GEN = 1; - CSUM_TAG_SERVER_GEN = 2; - CSUM_TAG_SERVER_REGEN = 3; + CSUM_TAG_CLIENT_SHA = 1; + CSUM_TAG_SERVER_SHA = 2; + CSUM_TAG_SERVER_REGEN_SHA = 3; } ////////////////////////////////////////// @@ -67,7 +68,7 @@ message Mpb_ChunkPos { // chunk_csum() type message Mpb_ChunkCSum { required Mpb_CSumType type = 1; - required bytes csum = 2; + optional bytes csum = 2; } // epoch_id() type @@ -115,10 +116,11 @@ message Mpb_AuthResp { // append_chunk() request & response message Mpb_AppendChunkReq { - required string prefix = 1; - optional bytes placement_key = 2; + optional bytes placement_key = 1; + required string prefix = 2; required bytes chunk = 3; - optional uint32 chunk_extra = 4; + required Mpb_ChunkCSum csum = 4; + optional uint32 chunk_extra = 5; } message Mpb_AppendChunkResp { diff --git a/src/machi_flu1.erl b/src/machi_flu1.erl index 63abb8e..3927a01 100644 --- a/src/machi_flu1.erl +++ b/src/machi_flu1.erl @@ -404,8 +404,8 @@ do_net_server_append2(FluName, Sock, CSumHex, LenHex, ExtraHex, Prefix) -> %% should be calculated by the head and passed down %% the chain together with the value. CS = machi_util:checksum_chunk(Chunk), - machi_util:make_tagged_csum(server_gen, CS); - <> -> + machi_util:make_tagged_csum(server_sha, CS); + <> -> CS = machi_util:checksum_chunk(Chunk), if CS == ClientCS -> ClientCSum; @@ -569,8 +569,8 @@ do_net_server_write2(Sock, CSumHex, OffsetHex, LenHex, FileBin, DataDir, FHc, %% should be calculated by the head and passed down %% the chain together with the value. CS = machi_util:checksum_chunk(Chunk), - machi_util:make_tagged_csum(server_gen,CS); - <> -> + machi_util:make_tagged_csum(server_sha,CS); + <> -> CS = machi_util:checksum_chunk(Chunk), if CS == ClientCS -> ClientCSum; @@ -935,7 +935,7 @@ http_server_hack_put(Sock, G, FluName, MyURI) -> try CSum = case G#http_goop.x_csum of undefined -> - machi_util:make_tagged_csum(server_gen, CSum0); + machi_util:make_tagged_csum(server_sha, CSum0); XX when is_binary(XX) -> if XX == CSum0 -> machi_util:make_tagged_csum(client_gen, CSum0); @@ -1059,6 +1059,39 @@ do_pb_request(#mpb_request{req_id=ReqID, #mpb_response{req_id=ReqID, generic=#mpb_errorresp{code=1, msg="AUTH not implemented"}}; +do_pb_request(#mpb_request{req_id=ReqID, + append_chunk=AC=#mpb_appendchunkreq{}}) -> + #mpb_appendchunkreq{placement_key=____PK, + prefix=Prefix, + chunk=ChunkBin, + csum=#mpb_chunkcsum{type=CSumType, csum=CSum}, + chunk_extra=ChunkExtra} = AC, + TaggedCSum = case CSumType of + 'CSUM_TAG_NONE' -> + C = machi_util:checksum_chunk(ChunkBin), + machi_util:make_tagged_csum(server_sha, C); + 'CSUM_TAG_CLIENT_SHA' -> + machi_util:make_tagged_csum(client_sha, CSum) + end, + Chunk = {TaggedCSum, ChunkBin}, + case (catch machi_cr_client:append_chunk(todo_fixme, + Prefix, Chunk)) of + {ok, {Offset, Size, File}} -> + make_append_resp(ReqID, 'OK', + #mpb_chunkpos{offset=Offset, + chunk_size=Size, + file_name=File}); + {error, bad_arg} -> + make_append_resp(ReqID, 'BAD_ARG'); + {error, wedged} -> + make_append_resp(ReqID, 'WEDGED'); + {error, bad_checksum} -> + make_append_resp(ReqID, 'BAD_CHECKSUM'); + {error, partition} -> + make_append_resp(ReqID, 'PARTITION'); + _Else -> + make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) + end; do_pb_request(#mpb_request{req_id=ReqID}) -> #mpb_response{req_id=ReqID, generic=#mpb_errorresp{code=66, @@ -1067,3 +1100,17 @@ do_pb_request(_Else) -> #mpb_response{req_id= <<>>, generic=#mpb_errorresp{code=67, msg="Unknown PB request"}}. + +make_append_resp(ReqID, Status) -> + make_append_resp(ReqID, Status, undefined). + +make_append_resp(ReqID, Status, Where) -> + #mpb_response{req_id=ReqID, + append_chunk=#mpb_appendchunkresp{status=Status, + chunk_pos=Where}}. + +make_error_resp(ReqID, Code, Msg) -> + #mpb_response{req_id=ReqID, + generic=#mpb_errorresp{code=Code, + msg=Msg}}. + diff --git a/src/machi_flu1_client.erl b/src/machi_flu1_client.erl index c19a36a..4a7996b 100644 --- a/src/machi_flu1_client.erl +++ b/src/machi_flu1_client.erl @@ -468,7 +468,7 @@ append_chunk2(Sock, EpochID, Prefix0, Chunk0, ChunkExtra) -> Chunk0; XX when is_binary(XX) -> SHA = machi_util:checksum_chunk(Chunk0), - {<>, Chunk0} + {<>, Chunk0} end, Len = iolist_size(Chunk), true = (Len =< ?MAX_CHUNK_SIZE), @@ -703,7 +703,7 @@ write_chunk2(Sock, EpochID, File0, Offset, Chunk0) -> Chunk0; XX when is_binary(XX) -> SHA = machi_util:checksum_chunk(Chunk0), - {<>, Chunk0} + {<>, Chunk0} end, CSumHex = machi_util:bin_to_hexstr(CSum), Len = iolist_size(Chunk), diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index a9db840..7c62e44 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -19,6 +19,12 @@ %% ------------------------------------------------------------------- %% @doc Machi PB (Protocol Buffers) high-level client (prototype, API TBD) +%% +%% At the moment, this is prototype-quality code: the API is not yet +%% fully specified, there is very little error handling with respect +%% to a single socket connection, and there is no code to deal with +%% multiple connections/load balancing/error handling to several/all +%% Machi cluster servers. -module(machi_pb_high_client). @@ -31,7 +37,9 @@ -export([start_link/1, quit/1, connected_p/1, echo/2, echo/3, - auth/3, auth/4]). + auth/3, auth/4, + append_chunk/6, append_chunk/7 + ]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, @@ -39,7 +47,9 @@ -record(state, { server_list :: p_srvr_dict(), - sock :: 'undefined' | port() + sock :: 'undefined' | port(), + sock_id :: integer(), + count=0 :: non_neg_integer() }). start_link(P_srvr_list) -> @@ -66,6 +76,12 @@ auth(PidSpec, User, Pass) -> auth(PidSpec, User, Pass, Timeout) -> send_sync(PidSpec, {auth, User, Pass}, Timeout). +append_chunk(PidSpec, PlacementKey, Prefix, Chunk, CSum, ChunkExtra) -> + append_chunk(PidSpec, PlacementKey, Prefix, Chunk, CSum, ChunkExtra, ?DEFAULT_TIMEOUT). + +append_chunk(PidSpec, PlacementKey, Prefix, Chunk, CSum, ChunkExtra, Timeout) -> + send_sync(PidSpec, {append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, Timeout). + send_sync(PidSpec, Cmd, Timeout) -> gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). @@ -115,7 +131,8 @@ try_connect(#state{server_list=Ps}=S) -> do_connect_to_pb_listener(P) end, unused, Ps) of Sock when is_port(Sock) -> - S#state{sock=Sock}; + {id, Index} = erlang:port_info(Sock, id), + S#state{sock=Sock, sock_id=Index, count=0}; _Else -> S end. @@ -170,6 +187,42 @@ do_send_sync({auth, User, Pass}, #state{sock=Sock}=S) -> catch X:Y -> Res = {bummer, {X, Y, erlang:get_stacktrace()}}, {Res, S} + end; +do_send_sync({append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, + #state{sock=Sock, sock_id=Index, count=Count}=S) -> + try + ReqID = <>, + PK = if PlacementKey == <<>> -> undefined; + true -> PlacementKey + end, + CSumT = case CSum of + none -> + #mpb_chunkcsum{type='CSUM_TAG_NONE', + csum=undefined}; + {client_sha, CSumBin} -> + #mpb_chunkcsum{type='CSUM_TAG_CLIENT_SHA', + csum=CSumBin} + end, + Req = #mpb_appendchunkreq{placement_key=PK, + prefix=Prefix, + chunk=Chunk, + csum=CSumT, + chunk_extra=ChunkExtra}, + R1a = #mpb_request{req_id=ReqID, + append_chunk=Req}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, append_chunk=R} when R /= undefined -> + {R, S#state{count=Count+1}}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S#state{count=Count+1}} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S#state{count=Count+1}} end. diff --git a/src/machi_util.erl b/src/machi_util.erl index 2b8fb97..81325d0 100644 --- a/src/machi_util.erl +++ b/src/machi_util.erl @@ -24,7 +24,7 @@ -export([ checksum_chunk/1, - make_tagged_csum/2, + make_tagged_csum/1, make_tagged_csum/2, hexstr_to_bin/1, bin_to_hexstr/1, hexstr_to_int/1, int_to_hexstr/2, int_to_hexbin/2, make_binary/1, make_string/1, @@ -220,14 +220,19 @@ checksum_chunk(Chunk) when is_binary(Chunk); is_list(Chunk) -> %% @doc Create a tagged checksum -make_tagged_csum(none, SHA) -> - <>; -make_tagged_csum(client_gen, SHA) -> - <>; -make_tagged_csum(server_gen, SHA) -> - <>; -make_tagged_csum(server_regen, SHA) -> - <>. +make_tagged_csum(none) -> + <>; +make_tagged_csum({Tag, CSum}) -> + make_tagged_csum(Tag, CSum). + +make_tagged_csum(none, _SHA) -> + <>; +make_tagged_csum(client_sha, SHA) -> + <>; +make_tagged_csum(server_sha, SHA) -> + <>; +make_tagged_csum(server_regen_sha, SHA) -> + <>. %% @doc Log a verbose message. diff --git a/test/machi_cr_client_test.erl b/test/machi_cr_client_test.erl index e6db150..23011a2 100644 --- a/test/machi_cr_client_test.erl +++ b/test/machi_cr_client_test.erl @@ -89,7 +89,7 @@ smoke_test2() -> machi_cr_client:append_chunk(C1, Prefix, Chunk1), {ok, {Off1,Size1,File1}} = machi_cr_client:append_chunk(C1, Prefix, Chunk1), - Chunk1_badcs = {<>, Chunk1}, + Chunk1_badcs = {<>, Chunk1}, {error, bad_checksum} = machi_cr_client:append_chunk(C1, Prefix, Chunk1_badcs), {ok, Chunk1} = machi_cr_client:read_chunk(C1, File1, Off1, Size1), diff --git a/test/machi_flu1_test.erl b/test/machi_flu1_test.erl index 153f65a..7a0045d 100644 --- a/test/machi_flu1_test.erl +++ b/test/machi_flu1_test.erl @@ -197,7 +197,7 @@ bad_checksum_test() -> try Prefix = <<"some prefix">>, Chunk1 = <<"yo yo yo">>, - Chunk1_badcs = {<>, Chunk1}, + Chunk1_badcs = {<>, Chunk1}, {error, bad_checksum} = ?FLU_C:append_chunk(Host, TcpPort, ?DUMMY_PV1_EPOCH, Prefix, Chunk1_badcs), diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index aa1393e..dce1395 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -52,11 +52,17 @@ smoke_test2() -> true = ?C:connected_p(Clnt), String = "yo, dawg", String = ?C:echo(Clnt, String), + %% TODO: auth() is not implemented. Auth requires SSL. %% Probably ought to put client stuff that relies on SSL into %% a separate test module? Or separate test func? {error, _} = ?C:auth(Clnt, "foo", "bar"), + PK = <<>>, + Prefix = <<"prefix">>, + Chunk1 = <<"Hello, chunk!">>, + yo = ?C:append_chunk(Clnt, PK, Prefix, Chunk1, none, 0), + ok after (catch ?C:quit(Clnt)) diff --git a/test/machi_proxy_flu1_client_test.erl b/test/machi_proxy_flu1_client_test.erl index e98b129..c0441d3 100644 --- a/test/machi_proxy_flu1_client_test.erl +++ b/test/machi_proxy_flu1_client_test.erl @@ -71,7 +71,7 @@ api_smoke_test() -> ?MUT:append_chunk_extra(Prox1, FakeEpoch, Prefix, MyChunk2, 4242, infinity), {ok, MyChunk2} = ?MUT:read_chunk(Prox1, FakeEpoch, MyFile2, MyOff2, MySize2), - MyChunk_badcs = {<>, MyChunk}, + MyChunk_badcs = {<>, MyChunk}, {error, bad_checksum} = ?MUT:append_chunk(Prox1, FakeEpoch, Prefix, MyChunk_badcs), {error, bad_checksum} = ?MUT:write_chunk(Prox1, FakeEpoch, From cb06c53dc011aa939c13618d15af1437eaf70f8b Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 14:45:24 +0900 Subject: [PATCH 05/13] WIP: PB append_chunk end-to-end works! --- src/machi_cr_client.erl | 7 ++- src/machi_flu1.erl | 88 ++---------------------------- src/machi_pb_high_client.erl | 17 +++++- test/machi_pb_high_client_test.erl | 12 +++- 4 files changed, 39 insertions(+), 85 deletions(-) diff --git a/src/machi_cr_client.erl b/src/machi_cr_client.erl index 09ef18d..ce30722 100644 --- a/src/machi_cr_client.erl +++ b/src/machi_cr_client.erl @@ -372,7 +372,7 @@ do_append_midtail(_RestFLUs, Prefix, File, Offset, Chunk, ChunkExtra, do_append_midtail2([], _Prefix, File, Offset, Chunk, _ChunkExtra, _Ws, _Depth, _STime, S) -> %% io:format(user, "ok!\n", []), - {reply, {ok, {Offset, iolist_size(Chunk), File}}, S}; + {reply, {ok, {Offset, chunk_wrapper_size(Chunk), File}}, S}; do_append_midtail2([FLU|RestFLUs]=FLUs, Prefix, File, Offset, Chunk, ChunkExtra, Ws, Depth, STime, #state{epoch_id=EpochID, proxies_dict=PD}=S) -> @@ -828,3 +828,8 @@ sleep_a_while(1) -> ok; sleep_a_while(Depth) -> timer:sleep(30 + trunc(math:pow(1.9, Depth))). + +chunk_wrapper_size({_TaggedCSum, Chunk}) -> + iolist_size(Chunk); +chunk_wrapper_size(Chunk) -> + iolist_size(Chunk). diff --git a/src/machi_flu1.erl b/src/machi_flu1.erl index 3927a01..cbd4242 100644 --- a/src/machi_flu1.erl +++ b/src/machi_flu1.erl @@ -75,7 +75,6 @@ -include_lib("kernel/include/file.hrl"). -include("machi.hrl"). --include("machi_pb.hrl"). -include("machi_projection.hrl"). -define(SERVER_CMD_READ_TIMEOUT, 600*1000). @@ -348,7 +347,12 @@ net_server_loop(Sock, #state{flu_name=FluName, data_dir=DataDir}=S) -> ok = gen_tcp:send(Sock, <<"OK\n">>), ok = inet:setopts(Sock, [{packet, 4}, {packet_size, 33*1024*1024}]), - protocol_buffers_loop(Sock, S); + {ok, Proj} = machi_projection_store:read_latest_projection( + S#state.proj_store, private), + Ps = [P_srvr || + {_, P_srvr} <- orddict:to_list( + Proj#projection_v1.members_dict)], + machi_pb_server:run_loop(Sock, Ps); _ -> machi_util:verb("Else Got: ~p\n", [Line]), io:format(user, "TODO: Else Got: ~p\n", [Line]), @@ -1015,21 +1019,6 @@ http_harvest_headers({ok, Hdr}, Sock, Acc) -> http_harvest_headers(gen_tcp:recv(Sock, 0, ?SERVER_CMD_READ_TIMEOUT), Sock, [Hdr|Acc]). -protocol_buffers_loop(Sock, S) -> - case gen_tcp:recv(Sock, 0) of - {ok, Bin} -> - R = do_pb_request(catch machi_pb:decode_mpb_request(Bin)), - %% R = #mpb_response{req_id= <<"not paying any attention">>, - %% generic=#mpb_errorresp{code=-6, - %% msg="not implemented"}}, - Resp = machi_pb:encode_mpb_response(R), - ok = gen_tcp:send(Sock, Resp), - protocol_buffers_loop(Sock, S); - {error, _} -> - (catch gen_tcp:close(Sock)), - exit(normal) - end. - digest_header_goop([], G) -> G; digest_header_goop([{http_header, _, 'Content-Length', _, Str}|T], G) -> @@ -1049,68 +1038,3 @@ split_uri_options(OpsBin) -> [<<"size">>, Bin] -> {size, binary_to_integer(Bin)} end || X <- L]. - -do_pb_request(#mpb_request{req_id=ReqID, - echo=#mpb_echoreq{message=Msg}}) -> - #mpb_response{req_id=ReqID, - echo=#mpb_echoresp{message=Msg}}; -do_pb_request(#mpb_request{req_id=ReqID, - auth=#mpb_authreq{}}) -> - #mpb_response{req_id=ReqID, - generic=#mpb_errorresp{code=1, - msg="AUTH not implemented"}}; -do_pb_request(#mpb_request{req_id=ReqID, - append_chunk=AC=#mpb_appendchunkreq{}}) -> - #mpb_appendchunkreq{placement_key=____PK, - prefix=Prefix, - chunk=ChunkBin, - csum=#mpb_chunkcsum{type=CSumType, csum=CSum}, - chunk_extra=ChunkExtra} = AC, - TaggedCSum = case CSumType of - 'CSUM_TAG_NONE' -> - C = machi_util:checksum_chunk(ChunkBin), - machi_util:make_tagged_csum(server_sha, C); - 'CSUM_TAG_CLIENT_SHA' -> - machi_util:make_tagged_csum(client_sha, CSum) - end, - Chunk = {TaggedCSum, ChunkBin}, - case (catch machi_cr_client:append_chunk(todo_fixme, - Prefix, Chunk)) of - {ok, {Offset, Size, File}} -> - make_append_resp(ReqID, 'OK', - #mpb_chunkpos{offset=Offset, - chunk_size=Size, - file_name=File}); - {error, bad_arg} -> - make_append_resp(ReqID, 'BAD_ARG'); - {error, wedged} -> - make_append_resp(ReqID, 'WEDGED'); - {error, bad_checksum} -> - make_append_resp(ReqID, 'BAD_CHECKSUM'); - {error, partition} -> - make_append_resp(ReqID, 'PARTITION'); - _Else -> - make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) - end; -do_pb_request(#mpb_request{req_id=ReqID}) -> - #mpb_response{req_id=ReqID, - generic=#mpb_errorresp{code=66, - msg="Unknown request"}}; -do_pb_request(_Else) -> - #mpb_response{req_id= <<>>, - generic=#mpb_errorresp{code=67, - msg="Unknown PB request"}}. - -make_append_resp(ReqID, Status) -> - make_append_resp(ReqID, Status, undefined). - -make_append_resp(ReqID, Status, Where) -> - #mpb_response{req_id=ReqID, - append_chunk=#mpb_appendchunkresp{status=Status, - chunk_pos=Where}}. - -make_error_resp(ReqID, Code, Msg) -> - #mpb_response{req_id=ReqID, - generic=#mpb_errorresp{code=Code, - msg=Msg}}. - diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index 7c62e44..6529bca 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -215,7 +215,8 @@ do_send_sync({append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, {ok, Bin1B} = gen_tcp:recv(Sock, 0), case (catch machi_pb:decode_mpb_response(Bin1B)) of #mpb_response{req_id=ReqID, append_chunk=R} when R /= undefined -> - {R, S#state{count=Count+1}}; + Result = convert_append_chunk_resp(R), + {Result, S#state{count=Count+1}}; #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, {{error, {Code, Msg, Extra}}, S#state{count=Count+1}} @@ -225,4 +226,18 @@ do_send_sync({append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, {Res, S#state{count=Count+1}} end. +convert_append_chunk_resp(#mpb_appendchunkresp{status='OK', chunk_pos=CP}) -> + #mpb_chunkpos{offset=Offset, chunk_size=Size, file_name=File} = CP, + {ok, {Offset, Size, File}}; +convert_append_chunk_resp(#mpb_appendchunkresp{status='BAD_ARG'}) -> + {error, bad_arg}; +convert_append_chunk_resp(#mpb_appendchunkresp{status='WEDGED'}) -> + {error, wedged}; +convert_append_chunk_resp(#mpb_appendchunkresp{status='BAD_CHECKSUM'}) -> + {error, bad_checksum}; +convert_append_chunk_resp(#mpb_appendchunkresp{status='PARTITION'}) -> + {error, partition}; +convert_append_chunk_resp(#mpb_appendchunkresp{status='BAD_JOSS'}) -> + throw({error, bad_joss_taipan_fixme}). + diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index dce1395..73b35a5 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -37,6 +37,7 @@ smoke_test2() -> Port = 5720, Ps = [#p_srvr{name=a, address="localhost", port=Port, props="./data.a"} ], + D = orddict:from_list([{P#p_srvr.name, P} || P <- Ps]), [os:cmd("rm -rf " ++ P#p_srvr.props) || P <- Ps], {ok, SupPid} = machi_flu_sup:start_link(), @@ -45,6 +46,7 @@ smoke_test2() -> #p_srvr{name=Name, port=Port, props=Dir} = P, {ok, _} = machi_flu_psup:start_flu_package(Name, Port, Dir, []) end || P <- Ps], + ok = machi_chain_manager1:set_chain_members(a_chmgr, D), [machi_chain_manager1:test_react_to_env(a_chmgr) || _ <-lists:seq(1,5)], {ok, Clnt} = ?C:start_link(Ps), @@ -61,7 +63,15 @@ smoke_test2() -> PK = <<>>, Prefix = <<"prefix">>, Chunk1 = <<"Hello, chunk!">>, - yo = ?C:append_chunk(Clnt, PK, Prefix, Chunk1, none, 0), + {ok, {Off1, Size1, File1}} = + ?C:append_chunk(Clnt, PK, Prefix, Chunk1, none, 0), + Chunk2 = "It's another chunk", + CSum2 = {client_sha, machi_util:checksum_chunk(Chunk2)}, + {ok, {Off2, Size2, File2}} = + ?C:append_chunk(Clnt, PK, Prefix, Chunk2, CSum2, 1024), + %% Chunk3 = ["This is a ", <<"test,">>, 32, [["Hello, world!"]]], + %% {ok, {Off3, Size3, File3}} = + %% ?C:write_chunk(Clnt, File2, Off2+iolist_size(Chunk2), Chunk3), ok after From a8782eed5a2fb80e1d304cd3f8b4b8ad79f06f66 Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 15:13:13 +0900 Subject: [PATCH 06/13] WIP: write_chunk #1 --- src/machi.proto | 5 +-- src/machi_pb_high_client.erl | 69 +++++++++++++++++++++++------- test/machi_pb_high_client_test.erl | 6 +-- 3 files changed, 59 insertions(+), 21 deletions(-) diff --git a/src/machi.proto b/src/machi.proto index d4c3b82..59c9242 100644 --- a/src/machi.proto +++ b/src/machi.proto @@ -74,7 +74,7 @@ message Mpb_ChunkCSum { // epoch_id() type message Mpb_EpochId { required uint32 epoch_num = 1; - required Mpb_ChunkCSum epoch_csum = 2; + required bytes epoch_csum = 2; } ////////////////////////////////////////// @@ -135,12 +135,11 @@ message Mpb_WriteChunkReq { required string file = 1; required uint64 offset = 2; required bytes chunk = 3; + required Mpb_ChunkCSum csum = 4; } message Mpb_WriteChunkResp { required Mpb_GeneralStatusCode status = 1; - // If OK, then chunk_pos is defined. - optional Mpb_ChunkPos chunk_pos = 2; } ////////////////////////////////////////// diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index 6529bca..b076959 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -38,7 +38,8 @@ connected_p/1, echo/2, echo/3, auth/3, auth/4, - append_chunk/6, append_chunk/7 + append_chunk/6, append_chunk/7, + write_chunk/5, write_chunk/6 ]). %% gen_server callbacks @@ -82,6 +83,12 @@ append_chunk(PidSpec, PlacementKey, Prefix, Chunk, CSum, ChunkExtra) -> append_chunk(PidSpec, PlacementKey, Prefix, Chunk, CSum, ChunkExtra, Timeout) -> send_sync(PidSpec, {append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, Timeout). +write_chunk(PidSpec, File, Offset, Chunk, CSum) -> + write_chunk(PidSpec, File, Offset, Chunk, CSum, ?DEFAULT_TIMEOUT). + +write_chunk(PidSpec, File, Offset, Chunk, CSum, Timeout) -> + send_sync(PidSpec, {write_chunk, File, Offset, Chunk, CSum}, Timeout). + send_sync(PidSpec, Cmd, Timeout) -> gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). @@ -195,14 +202,7 @@ do_send_sync({append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, PK = if PlacementKey == <<>> -> undefined; true -> PlacementKey end, - CSumT = case CSum of - none -> - #mpb_chunkcsum{type='CSUM_TAG_NONE', - csum=undefined}; - {client_sha, CSumBin} -> - #mpb_chunkcsum{type='CSUM_TAG_CLIENT_SHA', - csum=CSumBin} - end, + CSumT = convert_csum_req(CSum), Req = #mpb_appendchunkreq{placement_key=PK, prefix=Prefix, chunk=Chunk, @@ -224,20 +224,59 @@ do_send_sync({append_chunk, PlacementKey, Prefix, Chunk, CSum, ChunkExtra}, catch X:Y -> Res = {bummer, {X, Y, erlang:get_stacktrace()}}, {Res, S#state{count=Count+1}} + end; +do_send_sync({write_chunk, File, Offset, Chunk, CSum}, + #state{sock=Sock, sock_id=Index, count=Count}=S) -> + try + ReqID = <>, + CSumT = convert_csum_req(CSum), + Req = #mpb_writechunkreq{file=File, + offset=Offset, + chunk=Chunk, + csum=CSumT}, + R1a = #mpb_request{req_id=ReqID, + write_chunk=Req}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, write_chunk=R} when R /= undefined -> + Result = convert_write_chunk_resp(R), + {Result, S#state{count=Count+1}}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S#state{count=Count+1}} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S#state{count=Count+1}} end. +convert_csum_req(none) -> + #mpb_chunkcsum{type='CSUM_TAG_NONE', + csum=undefined}; +convert_csum_req({client_sha, CSumBin}) -> + #mpb_chunkcsum{type='CSUM_TAG_CLIENT_SHA', + csum=CSumBin}. + convert_append_chunk_resp(#mpb_appendchunkresp{status='OK', chunk_pos=CP}) -> #mpb_chunkpos{offset=Offset, chunk_size=Size, file_name=File} = CP, {ok, {Offset, Size, File}}; -convert_append_chunk_resp(#mpb_appendchunkresp{status='BAD_ARG'}) -> +convert_append_chunk_resp(#mpb_appendchunkresp{status=Status}) -> + convert_general_status_code(Status). + +convert_general_status_code('BAD_ARG') -> {error, bad_arg}; -convert_append_chunk_resp(#mpb_appendchunkresp{status='WEDGED'}) -> +convert_general_status_code('WEDGED') -> {error, wedged}; -convert_append_chunk_resp(#mpb_appendchunkresp{status='BAD_CHECKSUM'}) -> +convert_general_status_code('BAD_CHECKSUM') -> {error, bad_checksum}; -convert_append_chunk_resp(#mpb_appendchunkresp{status='PARTITION'}) -> +convert_general_status_code('PARTITION') -> {error, partition}; -convert_append_chunk_resp(#mpb_appendchunkresp{status='BAD_JOSS'}) -> +convert_general_status_code('BAD_JOSS') -> throw({error, bad_joss_taipan_fixme}). - +convert_write_chunk_resp(#mpb_writechunkresp{status='OK'}) -> + ok; +convert_write_chunk_resp(#mpb_writechunkresp{status=Status}) -> + convert_general_status_code(Status). diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 73b35a5..1a3c0ed 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -69,9 +69,9 @@ smoke_test2() -> CSum2 = {client_sha, machi_util:checksum_chunk(Chunk2)}, {ok, {Off2, Size2, File2}} = ?C:append_chunk(Clnt, PK, Prefix, Chunk2, CSum2, 1024), - %% Chunk3 = ["This is a ", <<"test,">>, 32, [["Hello, world!"]]], - %% {ok, {Off3, Size3, File3}} = - %% ?C:write_chunk(Clnt, File2, Off2+iolist_size(Chunk2), Chunk3), + Chunk3 = ["This is a ", <<"test,">>, 32, [["Hello, world!"]]], + ok = ?C:write_chunk(Clnt, File2, Off2+iolist_size(Chunk2), + Chunk3, none), ok after From 44c22bf752a891edf6478c927bd94106c6cbe5b9 Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 15:34:48 +0900 Subject: [PATCH 07/13] WIP: read_chunk #1 --- src/machi.proto | 21 +++++++++++++++++ src/machi_pb_high_client.erl | 38 +++++++++++++++++++++++++++++- test/machi_pb_high_client_test.erl | 11 +++++++-- 3 files changed, 67 insertions(+), 3 deletions(-) diff --git a/src/machi.proto b/src/machi.proto index 59c9242..6593ed1 100644 --- a/src/machi.proto +++ b/src/machi.proto @@ -142,6 +142,25 @@ message Mpb_WriteChunkResp { required Mpb_GeneralStatusCode status = 1; } +// read_chunk() request & response + +message Mpb_ReadChunkReq { + required string file = 1; + required uint64 offset = 2; + required uint32 size = 3; + // Use flag_checksum=non-zero to request the chunk's checksum also + optional uint32 flag_checksum = 4 [default=0]; + // Use flag_no_chunk=non-zero to skip returning the chunk (which + // only makes sense if flag_checksum is set). + optional uint32 flag_no_chunk = 5 [default=0]; +} + +message Mpb_ReadChunkResp { + required Mpb_GeneralStatusCode status = 1; + optional bytes chunk = 2; + optional Mpb_ChunkCSum csum = 3; +} + ////////////////////////////////////////// // // request & response wrapper @@ -163,6 +182,7 @@ message Mpb_Request { optional Mpb_AuthReq auth = 11; optional Mpb_AppendChunkReq append_chunk = 12; optional Mpb_WriteChunkReq write_chunk = 13; + optional Mpb_ReadChunkReq read_chunk = 14; } message Mpb_Response { @@ -183,4 +203,5 @@ message Mpb_Response { optional Mpb_AuthResp auth = 11; optional Mpb_AppendChunkResp append_chunk = 12; optional Mpb_WriteChunkResp write_chunk = 13; + optional Mpb_ReadChunkResp read_chunk = 14; } diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index b076959..a1c2b4b 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -39,7 +39,8 @@ echo/2, echo/3, auth/3, auth/4, append_chunk/6, append_chunk/7, - write_chunk/5, write_chunk/6 + write_chunk/5, write_chunk/6, + read_chunk/4, read_chunk/5 ]). %% gen_server callbacks @@ -89,6 +90,12 @@ write_chunk(PidSpec, File, Offset, Chunk, CSum) -> write_chunk(PidSpec, File, Offset, Chunk, CSum, Timeout) -> send_sync(PidSpec, {write_chunk, File, Offset, Chunk, CSum}, Timeout). +read_chunk(PidSpec, File, Offset, Size) -> + read_chunk(PidSpec, File, Offset, Size, ?DEFAULT_TIMEOUT). + +read_chunk(PidSpec, File, Offset, Size, Timeout) -> + send_sync(PidSpec, {read_chunk, File, Offset, Size}, Timeout). + send_sync(PidSpec, Cmd, Timeout) -> gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). @@ -250,6 +257,30 @@ do_send_sync({write_chunk, File, Offset, Chunk, CSum}, catch X:Y -> Res = {bummer, {X, Y, erlang:get_stacktrace()}}, {Res, S#state{count=Count+1}} + end; +do_send_sync({read_chunk, File, Offset, Size}, + #state{sock=Sock, sock_id=Index, count=Count}=S) -> + try + ReqID = <>, + Req = #mpb_readchunkreq{file=File, + offset=Offset, + size=Size}, + R1a = #mpb_request{req_id=ReqID, + read_chunk=Req}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, read_chunk=R} when R /= undefined -> + Result = convert_read_chunk_resp(R), + {Result, S#state{count=Count+1}}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S#state{count=Count+1}} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S#state{count=Count+1}} end. convert_csum_req(none) -> @@ -280,3 +311,8 @@ convert_write_chunk_resp(#mpb_writechunkresp{status='OK'}) -> ok; convert_write_chunk_resp(#mpb_writechunkresp{status=Status}) -> convert_general_status_code(Status). + +convert_read_chunk_resp(#mpb_readchunkresp{status='OK', chunk=Chunk}) -> + {ok, Chunk}; +convert_read_chunk_resp(#mpb_readchunkresp{status=Status}) -> + convert_general_status_code(Status). diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 1a3c0ed..9d3875d 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -70,8 +70,15 @@ smoke_test2() -> {ok, {Off2, Size2, File2}} = ?C:append_chunk(Clnt, PK, Prefix, Chunk2, CSum2, 1024), Chunk3 = ["This is a ", <<"test,">>, 32, [["Hello, world!"]]], - ok = ?C:write_chunk(Clnt, File2, Off2+iolist_size(Chunk2), - Chunk3, none), + Off3 = Off2 + iolist_size(Chunk2), + Size3 = iolist_size(Chunk3), + ok = ?C:write_chunk(Clnt, File2, Off3, Chunk3, none), + + Reads = [{Chunk1, File1, Off1, Size1}, + {Chunk2, File2, Off2, Size2}, + {Chunk3, File2, Off3, Size3}], + [{ok, Ch} = ?C:read_chunk(Clnt, Fl, Off, Sz) || + {Ch, Fl, Off, Sz} <- Reads], ok after From 6e77a4ea7477170a634adfae0aae8126422e5c9b Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 16:24:08 +0900 Subject: [PATCH 08/13] WIP: read_chunk end-to-end! --- test/machi_pb_high_client_test.erl | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 9d3875d..0af1dbc 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -74,9 +74,9 @@ smoke_test2() -> Size3 = iolist_size(Chunk3), ok = ?C:write_chunk(Clnt, File2, Off3, Chunk3, none), - Reads = [{Chunk1, File1, Off1, Size1}, - {Chunk2, File2, Off2, Size2}, - {Chunk3, File2, Off3, Size3}], + Reads = [{iolist_to_binary(Chunk1), File1, Off1, Size1}, + {iolist_to_binary(Chunk2), File2, Off2, Size2}, + {iolist_to_binary(Chunk3), File2, Off3, Size3}], [{ok, Ch} = ?C:read_chunk(Clnt, Fl, Off, Sz) || {Ch, Fl, Off, Sz} <- Reads], From 6722b3c0f1293b65480e989f2873ff62f2c47b3b Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 16:53:06 +0900 Subject: [PATCH 09/13] WIP: checksum_list incomplete implementation.... --- src/machi.proto | 13 +++++++++++ src/machi_pb_high_client.erl | 36 +++++++++++++++++++++++++++++- test/machi_pb_high_client_test.erl | 5 ++++- 3 files changed, 52 insertions(+), 2 deletions(-) diff --git a/src/machi.proto b/src/machi.proto index 6593ed1..ee43c26 100644 --- a/src/machi.proto +++ b/src/machi.proto @@ -161,6 +161,17 @@ message Mpb_ReadChunkResp { optional Mpb_ChunkCSum csum = 3; } +// read_chunk() request & response + +message Mpb_ChecksumListReq { + required string file = 1; +} + +message Mpb_ChecksumListResp { + required Mpb_GeneralStatusCode status = 1; + optional bytes chunk = 2; +} + ////////////////////////////////////////// // // request & response wrapper @@ -183,6 +194,7 @@ message Mpb_Request { optional Mpb_AppendChunkReq append_chunk = 12; optional Mpb_WriteChunkReq write_chunk = 13; optional Mpb_ReadChunkReq read_chunk = 14; + optional Mpb_ChecksumListReq checksum_list = 15; } message Mpb_Response { @@ -204,4 +216,5 @@ message Mpb_Response { optional Mpb_AppendChunkResp append_chunk = 12; optional Mpb_WriteChunkResp write_chunk = 13; optional Mpb_ReadChunkResp read_chunk = 14; + optional Mpb_ChecksumListResp checksum_list = 15; } diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index a1c2b4b..0363980 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -40,7 +40,8 @@ auth/3, auth/4, append_chunk/6, append_chunk/7, write_chunk/5, write_chunk/6, - read_chunk/4, read_chunk/5 + read_chunk/4, read_chunk/5, + checksum_list/2, checksum_list/3 ]). %% gen_server callbacks @@ -96,6 +97,12 @@ read_chunk(PidSpec, File, Offset, Size) -> read_chunk(PidSpec, File, Offset, Size, Timeout) -> send_sync(PidSpec, {read_chunk, File, Offset, Size}, Timeout). +checksum_list(PidSpec, File) -> + checksum_list(PidSpec, File, ?DEFAULT_TIMEOUT). + +checksum_list(PidSpec, File, Timeout) -> + send_sync(PidSpec, {checksum_list, File}, Timeout). + send_sync(PidSpec, Cmd, Timeout) -> gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). @@ -281,6 +288,28 @@ do_send_sync({read_chunk, File, Offset, Size}, catch X:Y -> Res = {bummer, {X, Y, erlang:get_stacktrace()}}, {Res, S#state{count=Count+1}} + end; +do_send_sync({checksum_list, File}, + #state{sock=Sock, sock_id=Index, count=Count}=S) -> + try + ReqID = <>, + Req = #mpb_checksumlistreq{file=File}, + R1a = #mpb_request{req_id=ReqID, + checksum_list=Req}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, checksum_list=R} when R /= undefined -> + Result = convert_checksum_list_resp(R), + {Result, S#state{count=Count+1}}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S#state{count=Count+1}} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S#state{count=Count+1}} end. convert_csum_req(none) -> @@ -316,3 +345,8 @@ convert_read_chunk_resp(#mpb_readchunkresp{status='OK', chunk=Chunk}) -> {ok, Chunk}; convert_read_chunk_resp(#mpb_readchunkresp{status=Status}) -> convert_general_status_code(Status). + +convert_checksum_list_resp(#mpb_checksumlistresp{status='OK', chunk=Chunk}) -> + {ok, Chunk}; +convert_checksum_list_resp(#mpb_checksumlistresp{status=Status}) -> + convert_general_status_code(Status). diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 0af1dbc..62c0b8e 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -70,16 +70,19 @@ smoke_test2() -> {ok, {Off2, Size2, File2}} = ?C:append_chunk(Clnt, PK, Prefix, Chunk2, CSum2, 1024), Chunk3 = ["This is a ", <<"test,">>, 32, [["Hello, world!"]]], + File3 = File2, Off3 = Off2 + iolist_size(Chunk2), Size3 = iolist_size(Chunk3), ok = ?C:write_chunk(Clnt, File2, Off3, Chunk3, none), Reads = [{iolist_to_binary(Chunk1), File1, Off1, Size1}, {iolist_to_binary(Chunk2), File2, Off2, Size2}, - {iolist_to_binary(Chunk3), File2, Off3, Size3}], + {iolist_to_binary(Chunk3), File3, Off3, Size3}], [{ok, Ch} = ?C:read_chunk(Clnt, Fl, Off, Sz) || {Ch, Fl, Off, Sz} <- Reads], + {ok, _} = ?C:checksum_list(Clnt, File1), + ok after (catch ?C:quit(Clnt)) From 73f71c406e5b5577399ec80a87f66397c67c434c Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 17:08:15 +0900 Subject: [PATCH 10/13] WIP: list_files end-to-end! --- src/machi.proto | 21 ++++++++++++++++- src/machi_pb_high_client.erl | 38 +++++++++++++++++++++++++++++- test/machi_pb_high_client_test.erl | 2 ++ 3 files changed, 59 insertions(+), 2 deletions(-) diff --git a/src/machi.proto b/src/machi.proto index ee43c26..9dfa246 100644 --- a/src/machi.proto +++ b/src/machi.proto @@ -77,6 +77,12 @@ message Mpb_EpochId { required bytes epoch_csum = 2; } +// file_info() type +message Mpb_FileInfo { + required uint64 file_size = 1; + required string file_name = 2; +} + ////////////////////////////////////////// // // requests & responses @@ -161,7 +167,7 @@ message Mpb_ReadChunkResp { optional Mpb_ChunkCSum csum = 3; } -// read_chunk() request & response +// checksum_list() request & response message Mpb_ChecksumListReq { required string file = 1; @@ -172,6 +178,17 @@ message Mpb_ChecksumListResp { optional bytes chunk = 2; } +// list_files() request & response + +message Mpb_ListFilesReq { +} + +message Mpb_ListFilesResp { + required Mpb_GeneralStatusCode status = 1; + // Wow, I cannot use optional & repeated together? + repeated Mpb_FileInfo files = 2; +} + ////////////////////////////////////////// // // request & response wrapper @@ -195,6 +212,7 @@ message Mpb_Request { optional Mpb_WriteChunkReq write_chunk = 13; optional Mpb_ReadChunkReq read_chunk = 14; optional Mpb_ChecksumListReq checksum_list = 15; + optional Mpb_ListFilesReq list_files = 16; } message Mpb_Response { @@ -217,4 +235,5 @@ message Mpb_Response { optional Mpb_WriteChunkResp write_chunk = 13; optional Mpb_ReadChunkResp read_chunk = 14; optional Mpb_ChecksumListResp checksum_list = 15; + optional Mpb_ListFilesResp list_files = 16; } diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index 0363980..bf0e104 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -41,7 +41,8 @@ append_chunk/6, append_chunk/7, write_chunk/5, write_chunk/6, read_chunk/4, read_chunk/5, - checksum_list/2, checksum_list/3 + checksum_list/2, checksum_list/3, + list_files/1, list_files/2 ]). %% gen_server callbacks @@ -103,6 +104,12 @@ checksum_list(PidSpec, File) -> checksum_list(PidSpec, File, Timeout) -> send_sync(PidSpec, {checksum_list, File}, Timeout). +list_files(PidSpec) -> + list_files(PidSpec, ?DEFAULT_TIMEOUT). + +list_files(PidSpec, Timeout) -> + send_sync(PidSpec, {list_files}, Timeout). + send_sync(PidSpec, Cmd, Timeout) -> gen_server:call(PidSpec, {send_sync, Cmd}, Timeout). @@ -310,6 +317,28 @@ do_send_sync({checksum_list, File}, catch X:Y -> Res = {bummer, {X, Y, erlang:get_stacktrace()}}, {Res, S#state{count=Count+1}} + end; +do_send_sync({list_files}, + #state{sock=Sock, sock_id=Index, count=Count}=S) -> + try + ReqID = <>, + Req = #mpb_listfilesreq{}, + R1a = #mpb_request{req_id=ReqID, + list_files=Req}, + Bin1a = machi_pb:encode_mpb_request(R1a), + ok = gen_tcp:send(Sock, Bin1a), + {ok, Bin1B} = gen_tcp:recv(Sock, 0), + case (catch machi_pb:decode_mpb_response(Bin1B)) of + #mpb_response{req_id=ReqID, list_files=R} when R /= undefined -> + Result = convert_list_files_resp(R), + {Result, S#state{count=Count+1}}; + #mpb_response{req_id=ReqID, generic=G} when G /= undefined -> + #mpb_errorresp{code=Code, msg=Msg, extra=Extra} = G, + {{error, {Code, Msg, Extra}}, S#state{count=Count+1}} + end + catch X:Y -> + Res = {bummer, {X, Y, erlang:get_stacktrace()}}, + {Res, S#state{count=Count+1}} end. convert_csum_req(none) -> @@ -350,3 +379,10 @@ convert_checksum_list_resp(#mpb_checksumlistresp{status='OK', chunk=Chunk}) -> {ok, Chunk}; convert_checksum_list_resp(#mpb_checksumlistresp{status=Status}) -> convert_general_status_code(Status). + +convert_list_files_resp(#mpb_listfilesresp{status='OK', files=Files}) -> + FileInfo = [{Size, File} || #mpb_fileinfo{file_size=Size, + file_name=File} <- Files], + {ok, FileInfo}; +convert_list_files_resp(#mpb_listfilesresp{status=Status}) -> + convert_general_status_code(Status). diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl index 62c0b8e..04ff08e 100644 --- a/test/machi_pb_high_client_test.erl +++ b/test/machi_pb_high_client_test.erl @@ -82,6 +82,8 @@ smoke_test2() -> {Ch, Fl, Off, Sz} <- Reads], {ok, _} = ?C:checksum_list(Clnt, File1), + {ok, [{File1Size,File1}]} = ?C:list_files(Clnt), + true = is_integer(File1Size), ok after From ceebe3d49145aaa99ff6943bc758471731190a7e Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 17:17:14 +0900 Subject: [PATCH 11/13] WIP: list_files #2 --- src/machi_pb_high_client.erl | 2 +- test/machi_pb_test.erl | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index bf0e104..375c8b8 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -20,7 +20,7 @@ %% @doc Machi PB (Protocol Buffers) high-level client (prototype, API TBD) %% -%% At the moment, this is prototype-quality code: the API is not yet +%% At the moment, this is brittle-prototype-quality code: the API is not yet %% fully specified, there is very little error handling with respect %% to a single socket connection, and there is no code to deal with %% multiple connections/load balancing/error handling to several/all diff --git a/test/machi_pb_test.erl b/test/machi_pb_test.erl index 41a224f..1044bf6 100644 --- a/test/machi_pb_test.erl +++ b/test/machi_pb_test.erl @@ -53,6 +53,13 @@ smoke_responses_test() -> extra= <<"bar">>}}, R1 = encdec_response(R1), + Files2 = [#mpb_fileinfo{file_size=X, file_name="foo."++integer_to_list(X)} + || X <- lists:seq(1, 5)], + R2 = #mpb_response{req_id= <<"x">>, + list_files=#mpb_listfilesresp{status='OK', + files=Files2}}, + R2 = encdec_response(R2), + ok. encdec_request(M) -> From 727b2a987d24bcdff98b3e850ea760f8409bca34 Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 17:22:45 +0900 Subject: [PATCH 12/13] ROTFL forgot to add src/machi_pb_server.erl --- src/machi_pb_server.erl | 215 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 215 insertions(+) create mode 100644 src/machi_pb_server.erl diff --git a/src/machi_pb_server.erl b/src/machi_pb_server.erl new file mode 100644 index 0000000..7fa9ecb --- /dev/null +++ b/src/machi_pb_server.erl @@ -0,0 +1,215 @@ +%% ------------------------------------------------------------------- +%% +%% Copyright (c) 2007-2015 Basho Technologies, Inc. All Rights Reserved. +%% +%% This file is provided to you 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(machi_pb_server). + +-include("machi.hrl"). +-include("machi_pb.hrl"). +-include("machi_projection.hrl"). + +-define(SERVER_CMD_READ_TIMEOUT, 600*1000). + +-export([run_loop/2]). + +run_loop(Sock, P_srvr_list) -> + ok = inet:setopts(Sock, [{packet, 4}, + {packet_size, 33*1024*1024}]), + {ok, Clnt} = machi_cr_client:start_link(P_srvr_list), + protocol_buffers_loop(Sock, Clnt). + +protocol_buffers_loop(Sock, Clnt) -> + case gen_tcp:recv(Sock, 0, ?SERVER_CMD_READ_TIMEOUT) of + {ok, Bin} -> + R = do_pb_request(catch machi_pb:decode_mpb_request(Bin), Clnt), + %% R = #mpb_response{req_id= <<"not paying any attention">>, + %% generic=#mpb_errorresp{code=-6, + %% msg="not implemented"}}, + Resp = machi_pb:encode_mpb_response(R), + ok = gen_tcp:send(Sock, Resp), + protocol_buffers_loop(Sock, Clnt); + {error, _} -> + (catch gen_tcp:close(Sock)), + exit(normal) + end. + +do_pb_request(#mpb_request{req_id=ReqID, + echo=#mpb_echoreq{message=Msg}}, _Clnt) -> + #mpb_response{req_id=ReqID, + echo=#mpb_echoresp{message=Msg}}; +do_pb_request(#mpb_request{req_id=ReqID, + auth=#mpb_authreq{}}, _Clnt) -> + #mpb_response{req_id=ReqID, + generic=#mpb_errorresp{code=1, + msg="AUTH not implemented"}}; +do_pb_request(#mpb_request{req_id=ReqID, + append_chunk=IR=#mpb_appendchunkreq{}}, Clnt) -> + #mpb_appendchunkreq{placement_key=__todo__PK, + prefix=Prefix, + chunk=ChunkBin, + csum=CSum, + chunk_extra=ChunkExtra} = IR, + TaggedCSum = make_tagged_csum(CSum, ChunkBin), + Chunk = {TaggedCSum, ChunkBin}, + case (catch machi_cr_client:append_chunk_extra(Clnt, Prefix, Chunk, + ChunkExtra)) of + {ok, {Offset, Size, File}} -> + make_append_resp(ReqID, 'OK', + #mpb_chunkpos{offset=Offset, + chunk_size=Size, + file_name=File}); + {error, bad_arg} -> + make_append_resp(ReqID, 'BAD_ARG'); + {error, wedged} -> + make_append_resp(ReqID, 'WEDGED'); + {error, bad_checksum} -> + make_append_resp(ReqID, 'BAD_CHECKSUM'); + {error, partition} -> + make_append_resp(ReqID, 'PARTITION'); + _Else -> + make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) + end; +do_pb_request(#mpb_request{req_id=ReqID, + write_chunk=IR=#mpb_writechunkreq{}}, Clnt) -> + #mpb_writechunkreq{file=File, + offset=Offset, + chunk=ChunkBin, + csum=CSum} = IR, + TaggedCSum = make_tagged_csum(CSum, ChunkBin), + Chunk = {TaggedCSum, ChunkBin}, + case (catch machi_cr_client:write_chunk(Clnt, File, Offset, Chunk)) of + {ok, {_,_,_}} -> + %% machi_cr_client returns ok 2-tuple, convert to simple ok. + make_write_resp(ReqID, 'OK'); + {error, bad_arg} -> + make_write_resp(ReqID, 'BAD_ARG'); + {error, wedged} -> + make_write_resp(ReqID, 'WEDGED'); + {error, bad_checksum} -> + make_write_resp(ReqID, 'BAD_CHECKSUM'); + {error, partition} -> + make_write_resp(ReqID, 'PARTITION'); + _Else -> + make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) + end; +do_pb_request(#mpb_request{req_id=ReqID, + read_chunk=IR=#mpb_readchunkreq{}}, Clnt) -> + #mpb_readchunkreq{file=File, + offset=Offset, + size=Size} = IR, + %% TODO: implement the optional flags in Mpb_ReadChunkReq + case (catch machi_cr_client:read_chunk(Clnt, File, Offset, Size)) of + {ok, Chunk} -> + make_read_resp(ReqID, 'OK', Chunk); + {error, bad_arg} -> + make_read_resp(ReqID, 'BAD_ARG', undefined); + {error, wedged} -> + make_read_resp(ReqID, 'WEDGED', undefined); + {error, bad_checksum} -> + make_read_resp(ReqID, 'BAD_CHECKSUM', undefined); + {error, partition} -> + make_read_resp(ReqID, 'PARTITION', undefined); + _Else -> + make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) + end; +do_pb_request(#mpb_request{req_id=ReqID, + checksum_list=IR=#mpb_checksumlistreq{}}, Clnt) -> + #mpb_checksumlistreq{file=File} = IR, + %% TODO: implement the optional flags in Mpb_ReadChunkReq + case (catch machi_cr_client:checksum_list(Clnt, File)) of + {ok, Chunk} -> + make_checksum_list_resp(ReqID, 'OK', Chunk); + {error, bad_arg} -> + make_checksum_list_resp(ReqID, 'BAD_ARG', undefined); + {error, wedged} -> + make_checksum_list_resp(ReqID, 'WEDGED', undefined); + {error, bad_checksum} -> + make_checksum_list_resp(ReqID, 'BAD_CHECKSUM', undefined); + {error, partition} -> + make_checksum_list_resp(ReqID, 'PARTITION', undefined); + _Else -> + make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) + end; +do_pb_request(#mpb_request{req_id=ReqID, + list_files=IR=#mpb_listfilesreq{}}, Clnt) -> + #mpb_listfilesreq{} = IR, + %% TODO: implement the optional flags in Mpb_ReadChunkReq + case (catch machi_cr_client:list_files(Clnt)) of + {ok, FileInfo} -> + make_list_files_resp(ReqID, 'OK', FileInfo); + {error, bad_arg} -> + make_list_files_resp(ReqID, 'BAD_ARG', []); + {error, wedged} -> + make_list_files_resp(ReqID, 'WEDGED', []); + {error, bad_checksum} -> + make_list_files_resp(ReqID, 'BAD_CHECKSUM', []); + {error, partition} -> + make_list_files_resp(ReqID, 'PARTITION', []); + _Else -> + make_error_resp(ReqID, 66, io_lib:format("err ~p", [_Else])) + end; +do_pb_request(#mpb_request{req_id=ReqID}, _Clnt) -> + #mpb_response{req_id=ReqID, + generic=#mpb_errorresp{code=66, + msg="Unknown request"}}; +do_pb_request(_Else, _Clnt) -> + #mpb_response{req_id= <<>>, + generic=#mpb_errorresp{code=67, + msg="Unknown PB request"}}. + +make_tagged_csum(#mpb_chunkcsum{type='CSUM_TAG_NONE'}, ChunkBin) -> + C = machi_util:checksum_chunk(ChunkBin), + machi_util:make_tagged_csum(server_sha, C); +make_tagged_csum(#mpb_chunkcsum{type='CSUM_TAG_CLIENT_SHA', csum=CSum}, _CB) -> + machi_util:make_tagged_csum(client_sha, CSum). + +make_append_resp(ReqID, Status) -> + make_append_resp(ReqID, Status, undefined). + +make_append_resp(ReqID, Status, Where) -> + #mpb_response{req_id=ReqID, + append_chunk=#mpb_appendchunkresp{status=Status, + chunk_pos=Where}}. + +make_write_resp(ReqID, Status) -> + #mpb_response{req_id=ReqID, + write_chunk=#mpb_writechunkresp{status=Status}}. + +make_read_resp(ReqID, Status, Chunk) -> + #mpb_response{req_id=ReqID, + read_chunk=#mpb_readchunkresp{status=Status, + chunk=Chunk}}. + +make_checksum_list_resp(ReqID, Status, __todo__Chunk) -> + Chunk = <<"TODO item: refactor the checksum_list op to return simply the text file representation of the checksums?">>, + #mpb_response{req_id=ReqID, + checksum_list=#mpb_checksumlistresp{status=Status, + chunk=Chunk}}. + +make_list_files_resp(ReqID, Status, FileInfo) -> + Files = [#mpb_fileinfo{file_size=Size, file_name=Name} || + {Size, Name} <- FileInfo], + #mpb_response{req_id=ReqID, + list_files=#mpb_listfilesresp{status=Status, + files=Files}}. + +make_error_resp(ReqID, Code, Msg) -> + #mpb_response{req_id=ReqID, + generic=#mpb_errorresp{code=Code, + msg=Msg}}. From d3b0b7fdc5e8317793832b051d76b24130ed8a4d Mon Sep 17 00:00:00 2001 From: Scott Lystig Fritchie Date: Tue, 23 Jun 2015 17:26:15 +0900 Subject: [PATCH 13/13] Clean up some dialyzer complaints --- src/machi_flu1.erl | 4 ++-- src/machi_pb_high_client.erl | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/machi_flu1.erl b/src/machi_flu1.erl index cbd4242..9dfe6a6 100644 --- a/src/machi_flu1.erl +++ b/src/machi_flu1.erl @@ -942,7 +942,7 @@ http_server_hack_put(Sock, G, FluName, MyURI) -> machi_util:make_tagged_csum(server_sha, CSum0); XX when is_binary(XX) -> if XX == CSum0 -> - machi_util:make_tagged_csum(client_gen, CSum0); + machi_util:make_tagged_csum(client_sha, CSum0); true -> throw({bad_csum, XX}) end @@ -1025,7 +1025,7 @@ digest_header_goop([{http_header, _, 'Content-Length', _, Str}|T], G) -> digest_header_goop(T, G#http_goop{len=list_to_integer(Str)}); digest_header_goop([{http_header, _, "X-Checksum", _, Str}|T], G) -> SHA = machi_util:hexstr_to_bin(Str), - CSum = machi_util:make_tagged_csum(client_gen, SHA), + CSum = machi_util:make_tagged_csum(client_sha, SHA), digest_header_goop(T, G#http_goop{x_csum=CSum}); digest_header_goop([_H|T], G) -> digest_header_goop(T, G). diff --git a/src/machi_pb_high_client.erl b/src/machi_pb_high_client.erl index 375c8b8..adf380d 100644 --- a/src/machi_pb_high_client.erl +++ b/src/machi_pb_high_client.erl @@ -129,7 +129,7 @@ handle_call(connected_p, _From, #state{sock=undefined}=S) -> handle_call({send_sync, _Cmd}, _From, #state{sock=undefined}=S) -> {reply, not_connected, S}; handle_call({send_sync, Cmd}, _From, S) -> - {Reply, S2} = do_send_sync(Cmd, S), + {Reply, S2} = do_send_sync(Cmd, try_connect(S)), {reply, Reply, S2}; handle_call(_Request, _From, S) -> Reply = whaaaaaaaaaaaaaaaaaaaa,