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..9dfa246 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,13 +68,19 @@ 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 message Mpb_EpochId { required uint32 epoch_num = 1; - required Mpb_ChunkCSum epoch_csum = 2; + required bytes epoch_csum = 2; +} + +// file_info() type +message Mpb_FileInfo { + required uint64 file_size = 1; + required string file_name = 2; } ////////////////////////////////////////// @@ -115,10 +122,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 { @@ -133,12 +141,52 @@ 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; +} + +// 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; +} + +// checksum_list() request & response + +message Mpb_ChecksumListReq { + required string file = 1; +} + +message Mpb_ChecksumListResp { + required Mpb_GeneralStatusCode status = 1; + 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; } ////////////////////////////////////////// @@ -162,6 +210,9 @@ 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; + optional Mpb_ChecksumListReq checksum_list = 15; + optional Mpb_ListFilesReq list_files = 16; } message Mpb_Response { @@ -182,4 +233,7 @@ 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; + optional Mpb_ChecksumListResp checksum_list = 15; + optional Mpb_ListFilesResp list_files = 16; } 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 af507c2..9dfe6a6 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). @@ -346,8 +345,14 @@ 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}]), - protocol_buffers_loop(Sock, S); + ok = inet:setopts(Sock, [{packet, 4}, + {packet_size, 33*1024*1024}]), + {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]), @@ -403,8 +408,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; @@ -568,8 +573,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; @@ -934,10 +939,10 @@ 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); + machi_util:make_tagged_csum(client_sha, CSum0); true -> throw({bad_csum, XX}) end @@ -1014,27 +1019,13 @@ 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 = #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) -> 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_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 new file mode 100644 index 0000000..adf380d --- /dev/null +++ b/src/machi_pb_high_client.erl @@ -0,0 +1,388 @@ +%% ------------------------------------------------------------------- +%% +%% 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) +%% +%% 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 +%% Machi cluster servers. + +-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, + auth/3, auth/4, + append_chunk/6, append_chunk/7, + write_chunk/5, write_chunk/6, + read_chunk/4, read_chunk/5, + checksum_list/2, checksum_list/3, + list_files/1, list_files/2 + ]). + +%% 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(), + sock_id :: integer(), + count=0 :: non_neg_integer() + }). + +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). + +%% 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). + +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). + +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). + +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). + +checksum_list(PidSpec, File) -> + checksum_list(PidSpec, File, ?DEFAULT_TIMEOUT). + +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). + +%%%%%%%%%%%%%%%%%%%%%%%%%%% + +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, try_connect(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) -> + {id, Index} = erlang:port_info(Sock, id), + S#state{sock=Sock, sock_id=Index, count=0}; + _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. + +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} 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()}}, + {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 = convert_csum_req(CSum), + 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 -> + 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}} + end + 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; +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; +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; +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) -> + #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=Status}) -> + convert_general_status_code(Status). + +convert_general_status_code('BAD_ARG') -> + {error, bad_arg}; +convert_general_status_code('WEDGED') -> + {error, wedged}; +convert_general_status_code('BAD_CHECKSUM') -> + {error, bad_checksum}; +convert_general_status_code('PARTITION') -> + {error, partition}; +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). + +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). + +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/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}}. 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_cr_client_test.erl b/test/machi_pb_cr_client_test.erl deleted file mode 100644 index 483ab9d..0000000 --- a/test/machi_pb_cr_client_test.erl +++ /dev/null @@ -1,79 +0,0 @@ -%% ------------------------------------------------------------------- -%% -%% 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_cr_client_test). --compile(export_all). - --ifdef(TEST). --ifndef(PULSE). - --include("machi_pb.hrl"). --include("machi_projection.hrl"). --include_lib("eunit/include/eunit.hrl"). - --define(C, machi_pb_cr_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"}} - ], - - [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], - [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}]), - 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), - - ok - after - (catch gen_tcp:close(Sock)) - end - after - exit(SupPid, normal), - [os:cmd("rm -rf " ++ P#p_srvr.props) || {_,P} <- Ps], - machi_util:wait_for_death(SupPid, 100), - ok - end. - --endif. % !PULSE --endif. % TEST diff --git a/test/machi_pb_high_client_test.erl b/test/machi_pb_high_client_test.erl new file mode 100644 index 0000000..04ff08e --- /dev/null +++ b/test/machi_pb_high_client_test.erl @@ -0,0 +1,100 @@ +%% ------------------------------------------------------------------- +%% +%% 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_high_client_test). +-compile(export_all). + +-ifdef(TEST). +-ifndef(PULSE). + +-include("machi_pb.hrl"). +-include("machi_projection.hrl"). +-include_lib("eunit/include/eunit.hrl"). + +-define(C, machi_pb_high_client). + +smoke_test_() -> + {timeout, 5*60, fun() -> smoke_test2() end}. + +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(), + try + [begin + #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), + try + 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!">>, + {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!"]]], + 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), File3, Off3, Size3}], + [{ok, Ch} = ?C:read_chunk(Clnt, Fl, Off, Sz) || + {Ch, Fl, Off, Sz} <- Reads], + + {ok, _} = ?C:checksum_list(Clnt, File1), + {ok, [{File1Size,File1}]} = ?C:list_files(Clnt), + true = is_integer(File1Size), + + ok + after + (catch ?C:quit(Clnt)) + end + after + exit(SupPid, normal), + [os:cmd("rm -rf " ++ P#p_srvr.props) || {_,P} <- Ps], + machi_util:wait_for_death(SupPid, 100), + ok + end. + +-endif. % !PULSE +-endif. % TEST 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) -> 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,