greg/machi
1
0
Fork 0
Code Issues 11 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Merge branch 'slf/pb-api-experiment1'

Browse source
This commit is contained in:
Scott Lystig Fritchie 2015-06-23 17:26:55 +09:00
commit 22ae33e298
14 changed files with 817 additions and 131 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
include/machi.hrl
View file

@ -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

74
src/machi.proto
View file

@ -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;
}

7
src/machi_cr_client.erl
View file

@ -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).

39
src/machi_flu1.erl
View file

@ -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);
<<?CSUM_TAG_CLIENT_GEN:8, ClientCS/binary>> ->
machi_util:make_tagged_csum(server_sha, CS);
<<?CSUM_TAG_CLIENT_SHA:8, ClientCS/binary>> ->
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);
<<?CSUM_TAG_CLIENT_GEN:8, ClientCS/binary>> ->
machi_util:make_tagged_csum(server_sha,CS);
<<?CSUM_TAG_CLIENT_SHA:8, ClientCS/binary>> ->
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).

4
src/machi_flu1_client.erl
View file

@ -468,7 +468,7 @@ append_chunk2(Sock, EpochID, Prefix0, Chunk0, ChunkExtra) ->
Chunk0;
XX when is_binary(XX) ->
SHA = machi_util:checksum_chunk(Chunk0),
{<<?CSUM_TAG_CLIENT_GEN:8, SHA/binary>>, Chunk0}
{<<?CSUM_TAG_CLIENT_SHA:8, SHA/binary>>, 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),
{<<?CSUM_TAG_CLIENT_GEN:8, SHA/binary>>, Chunk0}
{<<?CSUM_TAG_CLIENT_SHA:8, SHA/binary>>, Chunk0}
end,
CSumHex = machi_util:bin_to_hexstr(CSum),
Len = iolist_size(Chunk),

388
src/machi_pb_high_client.erl Normal file
View file

@ -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 = <<Index:64/big, Count:64/big>>,
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 = <<Index:64/big, Count:64/big>>,
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 = <<Index:64/big, Count:64/big>>,
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 = <<Index:64/big, Count:64/big>>,
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 = <<Index:64/big, Count:64/big>>,
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).

215
src/machi_pb_server.erl Normal file
View file

@ -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}}.

23
src/machi_util.erl
View file

@ -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) ->
<<?CSUM_TAG_NONE:8, SHA/binary>>;
make_tagged_csum(client_gen, SHA) ->
<<?CSUM_TAG_CLIENT_GEN:8, SHA/binary>>;
make_tagged_csum(server_gen, SHA) ->
<<?CSUM_TAG_SERVER_GEN:8, SHA/binary>>;
make_tagged_csum(server_regen, SHA) ->
<<?CSUM_TAG_SERVER_REGEN:8, SHA/binary>>.
make_tagged_csum(none) ->
<<?CSUM_TAG_NONE:8>>;
make_tagged_csum({Tag, CSum}) ->
make_tagged_csum(Tag, CSum).
make_tagged_csum(none, _SHA) ->
<<?CSUM_TAG_NONE:8>>;
make_tagged_csum(client_sha, SHA) ->
<<?CSUM_TAG_CLIENT_SHA:8, SHA/binary>>;
make_tagged_csum(server_sha, SHA) ->
<<?CSUM_TAG_SERVER_SHA:8, SHA/binary>>;
make_tagged_csum(server_regen_sha, SHA) ->
<<?CSUM_TAG_SERVER_REGEN_SHA:8, SHA/binary>>.
%% @doc Log a verbose message.

2
test/machi_cr_client_test.erl
View file

@ -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 = {<<?CSUM_TAG_CLIENT_GEN:8, 0:(8*20)>>, Chunk1},
Chunk1_badcs = {<<?CSUM_TAG_CLIENT_SHA:8, 0:(8*20)>>, Chunk1},
{error, bad_checksum} =
machi_cr_client:append_chunk(C1, Prefix, Chunk1_badcs),
{ok, Chunk1} = machi_cr_client:read_chunk(C1, File1, Off1, Size1),

2
test/machi_flu1_test.erl
View file

@ -197,7 +197,7 @@ bad_checksum_test() ->
try
Prefix = <<"some prefix">>,
Chunk1 = <<"yo yo yo">>,
Chunk1_badcs = {<<?CSUM_TAG_CLIENT_GEN:8, 0:(8*20)>>, Chunk1},
Chunk1_badcs = {<<?CSUM_TAG_CLIENT_SHA:8, 0:(8*20)>>, Chunk1},
{error, bad_checksum} = ?FLU_C:append_chunk(Host, TcpPort,
?DUMMY_PV1_EPOCH,
Prefix, Chunk1_badcs),

79
test/machi_pb_cr_client_test.erl
View file

@ -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

100
test/machi_pb_high_client_test.erl Normal file
View file

@ -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

7
test/machi_pb_test.erl
View file

@ -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) ->

2
test/machi_proxy_flu1_client_test.erl
View file

@ -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 = {<<?CSUM_TAG_CLIENT_GEN:8, 0:(8*20)>>, MyChunk},
MyChunk_badcs = {<<?CSUM_TAG_CLIENT_SHA:8, 0:(8*20)>>, MyChunk},
{error, bad_checksum} = ?MUT:append_chunk(Prox1, FakeEpoch,
Prefix, MyChunk_badcs),
{error, bad_checksum} = ?MUT:write_chunk(Prox1, FakeEpoch,