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-25 16:36:50 +09:00
commit 0b2b79cd0b
9 changed files with 673 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

224
src/machi.proto
View file

@ -31,7 +31,7 @@ option java_outer_classname = "MachiPB";
//////////////////////////////////////////
//
// enums
// Enums
//
//////////////////////////////////////////
@ -41,6 +41,8 @@ enum Mpb_GeneralStatusCode {
WEDGED = 2;
BAD_CHECKSUM = 3;
PARTITION = 4;
NOT_WRITTEN = 5;
WRITTEN = 6;
BAD_JOSS = 255; // Only for testing by the Taipan
}
@ -52,9 +54,14 @@ enum Mpb_CSumType {
CSUM_TAG_SERVER_REGEN_SHA = 3;
}
enum Mpb_ProjType {
PRIVATE = 20;
PUBLIC = 21;
}
//////////////////////////////////////////
//
// basic data types
// Basic data types
//
//////////////////////////////////////////
@ -72,8 +79,8 @@ message Mpb_ChunkCSum {
}
// epoch_id() type
message Mpb_EpochId {
required uint32 epoch_num = 1;
message Mpb_EpochID {
required uint32 epoch_number = 1;
required bytes epoch_csum = 2;
}
@ -83,9 +90,18 @@ message Mpb_FileInfo {
required string file_name = 2;
}
// #p_srvr{} type
message Mpb_P_Srvr {
required string name = 1;
required string proto_mod = 2;
required string address = 3;
required string port = 4;
required bytes opaque_props = 5;
}
//////////////////////////////////////////
//
// requests & responses
// Requests & responses
//
//////////////////////////////////////////
@ -97,7 +113,21 @@ message Mpb_ErrorResp {
optional bytes extra = 3;
}
// ping() request & response
//////////////////////////////////////////
//
// High level API requests:
//
// echo() : Mpb_EchoReq and Mpb_EchoResp
// auth() : Mpb_AuthReq and Mpb_AuthResp
// append_chunk() : Mpb_AppendChunkReq and Mpb_AppendChunkResp
// write_chunk() : Mpb_WriteChunkReq and Mpb_WriteChunkResp
// read_chunk() : Mpb_ReadChunkReq and Mpb_ReadChunkResp
// checksum_list() : Mpb_ChecksumListReq and Mpb_ChecksumListResp
// list_files() : Mpb_ListFilesReq and Mpb_ListFilesResp
//
//////////////////////////////////////////
// High level API: echo() request & response
message Mpb_EchoReq {
optional string message = 1;
@ -107,7 +137,7 @@ message Mpb_EchoResp {
optional string message = 1;
}
// Authentication request & response
// High level API: auth() request & response (not yet implemented)
message Mpb_AuthReq {
required bytes user = 1;
@ -119,7 +149,7 @@ message Mpb_AuthResp {
// TODO: not implemented yet
}
// append_chunk() request & response
// High level API: append_chunk() request & response
message Mpb_AppendChunkReq {
optional bytes placement_key = 1;
@ -135,7 +165,7 @@ message Mpb_AppendChunkResp {
optional Mpb_ChunkPos chunk_pos = 2;
}
// write_chunk() request & response
// High level API: write_chunk() request & response
message Mpb_WriteChunkReq {
required string file = 1;
@ -148,7 +178,7 @@ message Mpb_WriteChunkResp {
required Mpb_GeneralStatusCode status = 1;
}
// read_chunk() request & response
// High level API: read_chunk() request & response
message Mpb_ReadChunkReq {
required string file = 1;
@ -167,7 +197,7 @@ message Mpb_ReadChunkResp {
optional Mpb_ChunkCSum csum = 3;
}
// checksum_list() request & response
// High level API: checksum_list() request & response
message Mpb_ChecksumListReq {
required string file = 1;
@ -178,7 +208,7 @@ message Mpb_ChecksumListResp {
optional bytes chunk = 2;
}
// list_files() request & response
// High level API: list_files() request & response
message Mpb_ListFilesReq {
}
@ -191,7 +221,7 @@ message Mpb_ListFilesResp {
//////////////////////////////////////////
//
// request & response wrapper
// High level API request & response wrapper
//
//////////////////////////////////////////
@ -225,7 +255,7 @@ message Mpb_Response {
// Generic error response, typically used when something quite
// bad/unexpected happened within the server.
// Clients should always check this response and, if defined,
// ignroe any request-specific response at codes 10+.
// ignore any request-specific response at codes 10+.
optional Mpb_ErrorResp generic = 2;
// Specific responses.
@ -237,3 +267,169 @@ message Mpb_Response {
optional Mpb_ChecksumListResp checksum_list = 15;
optional Mpb_ListFilesResp list_files = 16;
}
//////////////////////////////////////////
//
// Low level API data types
//
//////////////////////////////////////////
enum Mpb_Mode {
AP_MODE = 30;
CP_MODE = 31;
}
message Mpb_Now {
required uint64 sec = 1;
required uint32 usec = 2;
}
message Mpb_MembersDictEntry {
required string key = 1;
required Mpb_P_Srvr val = 2;
}
message Mpb_ProjectionV1 {
required uint32 epoch_number = 1;
required bytes epoch_csum = 2;
required string author_server = 3;
repeated string all_members = 4;
required Mpb_Now creation_time = 5;
required Mpb_Mode mode = 6;
repeated string upi = 7;
repeated string repairing = 8;
repeated string down = 9;
optional bytes opaque_flap = 10;
optional bytes opaque_inner = 11;
required bytes opaque_dbg = 12;
required bytes opaque_dbg2 = 13;
repeated Mpb_MembersDictEntry members_dict = 14;
}
//////////////////////////////////////////
//
// Low level API requests:
//
// echo() : Mpb_EchoReq and Mpb_EchoResp (reused from high level API)
// auth() : Mpb_AuthReq and Mpb_AuthResp (reused from high level API)
// get_latest_epochid() : Mpb_GetLatestEpochIDReq and Mpb_GetLatestEpochIDResp
//
//////////////////////////////////////////
// Low level API: get_latest_epochid() request & response
message Mpb_LL_GetLatestEpochIDReq {
required Mpb_ProjType type = 1;
}
message Mpb_LL_GetLatestEpochIDResp {
required Mpb_GeneralStatusCode status = 1;
optional Mpb_EpochID epoch_id = 2;
}
// Low level API: read_latest_projection() request & response
message Mpb_LL_ReadLatestProjectionReq {
required Mpb_ProjType type = 1;
}
message Mpb_LL_ReadLatestProjectionResp {
required Mpb_GeneralStatusCode status = 1;
optional Mpb_ProjectionV1 proj = 2;
}
// Low level API: read_projection() request & response
message Mpb_LL_ReadProjectionReq {
required Mpb_ProjType type = 1;
required uint32 epoch_number = 2;
}
message Mpb_LL_ReadProjectionResp {
required Mpb_GeneralStatusCode status = 1;
optional Mpb_ProjectionV1 proj = 2;
}
// Low level API: write_projection() request & response
message Mpb_LL_WriteProjectionReq {
required Mpb_ProjType type = 1;
required Mpb_ProjectionV1 proj = 2;
}
message Mpb_LL_WriteProjectionResp {
required Mpb_GeneralStatusCode status = 1;
}
// Low level API: get_all_projections() request & response
message Mpb_LL_GetAllProjectionsReq {
required Mpb_ProjType type = 1;
}
message Mpb_LL_GetAllProjectionsResp {
required Mpb_GeneralStatusCode status = 1;
repeated Mpb_ProjectionV1 projs = 2;
}
// Low level API: list_all_projections() request & response
message Mpb_LL_ListAllProjectionsReq {
required Mpb_ProjType type = 1;
}
message Mpb_LL_ListAllProjectionsResp {
required Mpb_GeneralStatusCode status = 1;
repeated uint32 epochs = 2;
}
//////////////////////////////////////////
//
// Low API request & response wrapper
//
//////////////////////////////////////////
message Mpb_LL_Request {
// TODO: If we wish to support pipelined requests sometime in the
// future, this is the placeholder to do it.
required bytes req_id = 1;
// The client should only define one request message. If the client
// includes multiple requests here, the server may pick/choose an
// arbitrary one.
// NOTE: The erlang protobuffs compiler doesn't support 'oneof'.
// But 'oneof' appears to be a very tiny memory optimization
// that not all languages might care about? (Erlang doesn't)
optional Mpb_EchoReq echo = 10; // Re-use from high level API
optional Mpb_AuthReq auth = 11; // Re-use from high level API
optional Mpb_LL_GetLatestEpochIDReq proj_gl = 12;
optional Mpb_LL_ReadLatestProjectionReq proj_rl = 13;
optional Mpb_LL_ReadProjectionReq proj_rp = 14;
optional Mpb_LL_WriteProjectionReq proj_wp = 15;
optional Mpb_LL_GetAllProjectionsReq proj_ga = 16;
optional Mpb_LL_ListAllProjectionsReq proj_la = 17;
}
message Mpb_LL_Response {
// TODO: If we wish to support pipelined requests sometime in the
// future, this is the placeholder to do it.
required bytes req_id = 1;
// The server will define only one of the optional responses below.
// Generic error response, typically used when something quite
// bad/unexpected happened within the server.
// Clients should always check this response and, if defined,
// ignore any request-specific response at codes 10+.
optional Mpb_ErrorResp generic = 2;
// Specific responses.
optional Mpb_EchoResp echo = 10; // Re-use from high level API
optional Mpb_AuthResp auth = 11; // Re-use from high level API
optional Mpb_LL_GetLatestEpochIDResp proj_gl = 12;
optional Mpb_LL_ReadLatestProjectionResp proj_rl = 13;
optional Mpb_LL_ReadProjectionResp proj_rp = 14;
optional Mpb_LL_WriteProjectionResp proj_wp = 15;
optional Mpb_LL_GetAllProjectionsResp proj_ga = 16;
optional Mpb_LL_ListAllProjectionsResp proj_la = 17;
}

2
src/machi_dt.erl
View file

@ -34,7 +34,7 @@
-type epoch_num() :: -1 | non_neg_integer().
-type epoch_id() :: {epoch_num(), epoch_csum()}.
-type file_info() :: {file_size(), file_name_s()}.
-type file_name() :: float().%%%binary() | list().
-type file_name() :: binary() | list().
-type file_name_s() :: binary(). % server reply
-type file_offset() :: non_neg_integer().
-type file_size() :: non_neg_integer().

13
src/machi_flu1.erl
View file

@ -864,11 +864,14 @@ do_projection_command(Sock, LenHex, S) ->
ok = inet:setopts(Sock, [{packet, raw}]),
{ok, ProjCmdBin} = gen_tcp:recv(Sock, Len),
ok = inet:setopts(Sock, [{packet, line}]),
ProjCmd = binary_to_term(ProjCmdBin),
ProjCmdM = machi_pb:decode_mpb_ll_request(ProjCmdBin),
{ID, ProjCmd} = machi_pb_wrap:unmake_projection_req(ProjCmdM),
ProjOp = element(1, ProjCmd),
put(hack, ProjCmd),
Res = handle_projection_command(ProjCmd, S),
ResBin = term_to_binary(Res),
ResLenHex = machi_util:int_to_hexbin(byte_size(ResBin), 32),
ResM = machi_pb_wrap:make_projection_resp(ID, ProjOp, Res),
ResBin = machi_pb:encode_mpb_ll_response(ResM),
ResLenHex = machi_util:int_to_hexbin(iolist_size(ResBin), 32),
ok = gen_tcp:send(Sock, [<<"OK ">>, ResLenHex, <<"\n">>, ResBin])
catch
What:Why ->
@ -896,9 +899,7 @@ handle_projection_command({get_all_projections, ProjType},
machi_projection_store:get_all_projections(ProjStore, ProjType);
handle_projection_command({list_all_projections, ProjType},
#state{proj_store=ProjStore}) ->
machi_projection_store:list_all_projections(ProjStore, ProjType);
handle_projection_command(Else, _S) ->
{error, unknown_cmd, Else}.
machi_projection_store:list_all_projections(ProjStore, ProjType).
make_listener_regname(BaseName) ->
list_to_atom(atom_to_list(BaseName) ++ "_listener").

43
src/machi_flu1_client.erl
View file

@ -802,38 +802,44 @@ trunc_hack2(Sock, EpochID, File) ->
end.
get_latest_epochid2(Sock, ProjType) ->
ProjCmd = {get_latest_epochid, ProjType},
do_projection_common(Sock, ProjCmd).
Req = machi_pb_wrap:make_projection_req(
<<42>>, {get_latest_epochid, ProjType}),
do_projection_common(Sock, Req).
read_latest_projection2(Sock, ProjType) ->
ProjCmd = {read_latest_projection, ProjType},
do_projection_common(Sock, ProjCmd).
Req = machi_pb_wrap:make_projection_req(
<<42>>, {read_latest_projection, ProjType}),
do_projection_common(Sock, Req).
read_projection2(Sock, ProjType, Epoch) ->
ProjCmd = {read_projection, ProjType, Epoch},
do_projection_common(Sock, ProjCmd).
Req = machi_pb_wrap:make_projection_req(
<<42>>, {read_projection, ProjType, Epoch}),
do_projection_common(Sock, Req).
write_projection2(Sock, ProjType, Proj) ->
ProjCmd = {write_projection, ProjType, Proj},
do_projection_common(Sock, ProjCmd).
Req = machi_pb_wrap:make_projection_req(
<<42>>, {write_projection, ProjType, Proj}),
do_projection_common(Sock, Req).
get_all_projections2(Sock, ProjType) ->
ProjCmd = {get_all_projections, ProjType},
do_projection_common(Sock, ProjCmd).
Req = machi_pb_wrap:make_projection_req(
<<42>>, {get_all_projections, ProjType}),
do_projection_common(Sock, Req).
list_all_projections2(Sock, ProjType) ->
ProjCmd = {list_all_projections, ProjType},
do_projection_common(Sock, ProjCmd).
Req = machi_pb_wrap:make_projection_req(
<<42>>, {list_all_projections, ProjType}),
do_projection_common(Sock, Req).
do_projection_common(Sock, ProjCmd) ->
do_projection_common(Sock, Req) ->
erase(bad_sock),
try
ProjCmdBin = term_to_binary(ProjCmd),
Len = iolist_size(ProjCmdBin),
ReqBin = list_to_binary(machi_pb:encode_mpb_ll_request(Req)),
Len = iolist_size(ReqBin),
true = (Len =< ?MAX_CHUNK_SIZE),
LenHex = machi_util:int_to_hexbin(Len, 32),
Cmd = [<<"PROJ ">>, LenHex, <<"\n">>],
ok = w_send(Sock, [Cmd, ProjCmdBin]),
ok = w_send(Sock, [Cmd, ReqBin]),
ok = w_setopts(Sock, [{packet, line}]),
case w_recv(Sock, 0) of
{ok, Line} ->
@ -841,9 +847,10 @@ do_projection_common(Sock, ProjCmd) ->
<<"OK ", ResLenHex:8/binary, "\n">> ->
ResLen = machi_util:hexstr_to_int(ResLenHex),
ok = w_setopts(Sock, [{packet, raw}]),
{ok, ResBin} = w_recv(Sock, ResLen),
{ok, RespBin} = w_recv(Sock, ResLen),
ok = w_setopts(Sock, [{packet, line}]),
binary_to_term(ResBin);
Resp = machi_pb:decode_mpb_ll_response(RespBin),
machi_pb_wrap:unmake_projection_resp(Resp);
Else ->
{error, Else}
end

7
src/machi_pb_high_client.erl
View file

@ -44,6 +44,7 @@
checksum_list/2, checksum_list/3,
list_files/1, list_files/2
]).
-export([convert_general_status_code/1]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
@ -354,6 +355,8 @@ convert_append_chunk_resp(#mpb_appendchunkresp{status='OK', chunk_pos=CP}) ->
convert_append_chunk_resp(#mpb_appendchunkresp{status=Status}) ->
convert_general_status_code(Status).
convert_general_status_code('OK') ->
ok;
convert_general_status_code('BAD_ARG') ->
{error, bad_arg};
convert_general_status_code('WEDGED') ->
@ -362,6 +365,10 @@ convert_general_status_code('BAD_CHECKSUM') ->
{error, bad_checksum};
convert_general_status_code('PARTITION') ->
{error, partition};
convert_general_status_code('NOT_WRITTEN') ->
{error, not_written};
convert_general_status_code('WRITTEN') ->
{error, written};
convert_general_status_code('BAD_JOSS') ->
throw({error, bad_joss_taipan_fixme}).

367
src/machi_pb_wrap.erl Normal file
View file

@ -0,0 +1,367 @@
%% -------------------------------------------------------------------
%%
%% Machi: a small village of replicated files
%%
%% Copyright (c) 2014-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 Wrappers for Protocol Buffers encoding, including hacks to fix
%% impedance mismatches between Erlang terms and PB encodings.
%%
%% TODO: Any use of enc_sexp() and dec_sexp() should be eliminated,
%% except for the possibility of items where we are 100% sure
%% that a non-Erlang software component can get away with always
%% treating that item as an opaque thing.
-module(machi_pb_wrap).
-include("machi_pb.hrl").
-include("machi_projection.hrl").
-export([enc_p_srvr/1, dec_p_srvr/1,
enc_projection_v1/1, dec_projection_v1/1,
make_projection_req/2, unmake_projection_req/1,
make_projection_resp/3, unmake_projection_resp/1]).
-ifdef(TEST).
-compile(export_all).
-endif. % TEST
enc_p_srvr(P) ->
machi_pb:encode_mpb_p_srvr(conv_from_p_srvr(P)).
dec_p_srvr(Bin) ->
conv_to_p_srvr(machi_pb:decode_mpb_p_srvr(Bin)).
conv_from_p_srvr(#p_srvr{name=Name,
proto_mod=ProtoMod,
address=Address,
port=Port,
props=Props}) ->
#mpb_p_srvr{name=to_list(Name),
proto_mod=to_list(ProtoMod),
address=to_list(Address),
port=to_list(Port),
opaque_props=enc_sexp(Props)}.
conv_to_p_srvr(#mpb_p_srvr{name=Name,
proto_mod=ProtoMod,
address=Address,
port=Port,
opaque_props=Props}) ->
#p_srvr{name=to_atom(Name),
proto_mod=to_atom(ProtoMod),
address=to_list(Address),
port=to_integer(Port),
props=dec_sexp(Props)}.
enc_projection_v1(P) ->
%% Awww, flatten it here
list_to_binary(
machi_pb:encode_mpb_projectionv1(conv_from_projection_v1(P))).
dec_projection_v1(Bin) ->
conv_to_projection_v1(machi_pb:decode_mpb_projectionv1(Bin)).
conv_from_projection_v1(#projection_v1{epoch_number=Epoch,
epoch_csum=CSum,
author_server=Author,
all_members=AllMembers,
creation_time=CTime,
mode=Mode,
upi=UPI,
repairing=Repairing,
down=Down,
flap=Flap,
inner=Inner,
dbg=Dbg,
dbg2=Dbg2,
members_dict=MembersDict}) ->
#mpb_projectionv1{epoch_number=Epoch,
epoch_csum=CSum,
author_server=to_list(Author),
all_members=[to_list(X) || X <- AllMembers],
creation_time=conv_from_now(CTime),
mode=conv_from_mode(Mode),
upi=[to_list(X) || X <- UPI],
repairing=[to_list(X) || X <- Repairing],
down=[to_list(X) || X <- Down],
opaque_flap=enc_optional_sexp(Flap),
opaque_inner=enc_optional_sexp(Inner),
opaque_dbg=enc_sexp(Dbg),
opaque_dbg2=enc_sexp(Dbg2),
members_dict=conv_from_members_dict(MembersDict)}.
conv_to_projection_v1(#mpb_projectionv1{epoch_number=Epoch,
epoch_csum=CSum,
author_server=Author,
all_members=AllMembers,
creation_time=CTime,
mode=Mode,
upi=UPI,
repairing=Repairing,
down=Down,
opaque_flap=Flap,
opaque_inner=Inner,
opaque_dbg=Dbg,
opaque_dbg2=Dbg2,
members_dict=MembersDict}) ->
#projection_v1{epoch_number=Epoch,
epoch_csum=CSum,
author_server=to_atom(Author),
all_members=[to_atom(X) || X <- AllMembers],
creation_time=conv_to_now(CTime),
mode=conv_to_mode(Mode),
upi=[to_atom(X) || X <- UPI],
repairing=[to_atom(X) || X <- Repairing],
down=[to_atom(X) || X <- Down],
flap=dec_optional_sexp(Flap),
inner=dec_optional_sexp(Inner),
dbg=dec_sexp(Dbg),
dbg2=dec_sexp(Dbg2),
members_dict=conv_to_members_dict(MembersDict)}.
make_projection_req(ID, {get_latest_epochid, ProjType}) ->
#mpb_ll_request{req_id=ID,
proj_gl=#mpb_ll_getlatestepochidreq{type=conv_from_type(ProjType)}};
make_projection_req(ID, {read_latest_projection, ProjType}) ->
#mpb_ll_request{req_id=ID,
proj_rl=#mpb_ll_readlatestprojectionreq{type=conv_from_type(ProjType)}};
make_projection_req(ID, {read_projection, ProjType, Epoch}) ->
#mpb_ll_request{req_id=ID,
proj_rp=#mpb_ll_readprojectionreq{type=conv_from_type(ProjType),
epoch_number=Epoch}};
make_projection_req(ID, {write_projection, ProjType, Proj}) ->
ProjM = conv_from_projection_v1(Proj),
#mpb_ll_request{req_id=ID,
proj_wp=#mpb_ll_writeprojectionreq{type=conv_from_type(ProjType),
proj=ProjM}};
make_projection_req(ID, {get_all_projections, ProjType}) ->
#mpb_ll_request{req_id=ID,
proj_ga=#mpb_ll_getallprojectionsreq{type=conv_from_type(ProjType)}};
make_projection_req(ID, {list_all_projections, ProjType}) ->
#mpb_ll_request{req_id=ID,
proj_la=#mpb_ll_listallprojectionsreq{type=conv_from_type(ProjType)}}.
unmake_projection_req(
#mpb_ll_request{req_id=ID,
proj_gl=#mpb_ll_getlatestepochidreq{type=ProjType}}) ->
{ID, {get_latest_epochid, conv_to_type(ProjType)}};
unmake_projection_req(
#mpb_ll_request{req_id=ID,
proj_rl=#mpb_ll_readlatestprojectionreq{type=ProjType}}) ->
{ID, {read_latest_projection, conv_to_type(ProjType)}};
unmake_projection_req(
#mpb_ll_request{req_id=ID,
proj_rp=#mpb_ll_readprojectionreq{type=ProjType,
epoch_number=Epoch}}) ->
{ID, {read_projection, conv_to_type(ProjType), Epoch}};
unmake_projection_req(
#mpb_ll_request{req_id=ID,
proj_wp=#mpb_ll_writeprojectionreq{type=ProjType,
proj=ProjM}}) ->
Proj = conv_to_projection_v1(ProjM),
{ID, {write_projection, conv_to_type(ProjType), Proj}};
unmake_projection_req(
#mpb_ll_request{req_id=ID,
proj_ga=#mpb_ll_getallprojectionsreq{type=ProjType}}) ->
{ID, {get_all_projections, conv_to_type(ProjType)}};
unmake_projection_req(
#mpb_ll_request{req_id=ID,
proj_la=#mpb_ll_listallprojectionsreq{type=ProjType}}) ->
{ID, {list_all_projections, conv_to_type(ProjType)}}.
make_projection_resp(ID, get_latest_epochid, {ok, {Epoch, CSum}}) ->
EID = #mpb_epochid{epoch_number=Epoch, epoch_csum=CSum},
#mpb_ll_response{req_id=ID,
proj_gl=#mpb_ll_getlatestepochidresp{
status='OK', epoch_id=EID}};
make_projection_resp(ID, get_latest_epochid, Status) ->
#mpb_ll_response{req_id=ID,
proj_gl=#mpb_ll_getlatestepochidresp{
status=conv_from_status(Status)}};
make_projection_resp(ID, read_latest_projection, {ok, Proj}) ->
ProjM = conv_from_projection_v1(Proj),
#mpb_ll_response{req_id=ID,
proj_rl=#mpb_ll_readlatestprojectionresp{
status='OK', proj=ProjM}};
make_projection_resp(ID, read_latest_projection, Status) ->
#mpb_ll_response{req_id=ID,
proj_rl=#mpb_ll_readlatestprojectionresp{
status=conv_from_status(Status)}};
make_projection_resp(ID, read_projection, {ok, Proj}) ->
ProjM = conv_from_projection_v1(Proj),
#mpb_ll_response{req_id=ID,
proj_rp=#mpb_ll_readprojectionresp{
status='OK', proj=ProjM}};
make_projection_resp(ID, read_projection, Status) ->
#mpb_ll_response{req_id=ID,
proj_rp=#mpb_ll_readprojectionresp{
status=conv_from_status(Status)}};
make_projection_resp(ID, write_projection, Status) ->
#mpb_ll_response{req_id=ID,
proj_wp=#mpb_ll_writeprojectionresp{
status=conv_from_status(Status)}};
make_projection_resp(ID, get_all_projections, {ok, Projs}) ->
ProjsM = [conv_from_projection_v1(Proj) || Proj <- Projs],
#mpb_ll_response{req_id=ID,
proj_ga=#mpb_ll_getallprojectionsresp{
status='OK', projs=ProjsM}};
make_projection_resp(ID, get_all_projections, Status) ->
#mpb_ll_response{req_id=ID,
proj_ga=#mpb_ll_getallprojectionsresp{
status=conv_from_status(Status)}};
make_projection_resp(ID, list_all_projections, {ok, Epochs}) ->
#mpb_ll_response{req_id=ID,
proj_la=#mpb_ll_listallprojectionsresp{
status='OK', epochs=Epochs}};
make_projection_resp(ID, list_all_projections, Status) ->
#mpb_ll_response{req_id=ID,
proj_la=#mpb_ll_listallprojectionsresp{
status=conv_from_status(Status)}}.
unmake_projection_resp(#mpb_ll_response{proj_gl=#mpb_ll_getlatestepochidresp{
status=Status, epoch_id=EID}}) ->
case Status of
'OK' ->
#mpb_epochid{epoch_number=Epoch, epoch_csum=CSum} = EID,
{ok, {Epoch, CSum}};
_ ->
machi_pb_high_client:convert_general_status_code(Status)
end;
unmake_projection_resp(#mpb_ll_response{proj_rl=#mpb_ll_readlatestprojectionresp{
status=Status, proj=P}}) ->
case Status of
'OK' ->
{ok, conv_to_projection_v1(P)};
_ ->
machi_pb_high_client:convert_general_status_code(Status)
end;
unmake_projection_resp(#mpb_ll_response{proj_rp=#mpb_ll_readprojectionresp{
status=Status, proj=P}}) ->
case Status of
'OK' ->
{ok, conv_to_projection_v1(P)};
_ ->
machi_pb_high_client:convert_general_status_code(Status)
end;
unmake_projection_resp(#mpb_ll_response{proj_wp=#mpb_ll_writeprojectionresp{
status=Status}}) ->
machi_pb_high_client:convert_general_status_code(Status);
unmake_projection_resp(#mpb_ll_response{proj_ga=#mpb_ll_getallprojectionsresp{
status=Status, projs=ProjsM}}) ->
case Status of
'OK' ->
{ok, [conv_to_projection_v1(ProjM) || ProjM <- ProjsM]};
_ ->
machi_pb_high_client:convert_general_status_code(Status)
end;
unmake_projection_resp(#mpb_ll_response{proj_la=#mpb_ll_listallprojectionsresp{
status=Status, epochs=Epochs}}) ->
case Status of
'OK' ->
{ok, Epochs};
_ ->
machi_pb_high_client:convert_general_status_code(Status)
end.
%%%%%%%%%%%%%%%%%%%
enc_sexp(T) ->
term_to_binary(T).
dec_sexp(Bin) when is_binary(Bin) ->
binary_to_term(Bin).
enc_optional_sexp(undefined) ->
undefined;
enc_optional_sexp(T) ->
enc_sexp(T).
dec_optional_sexp(undefined) ->
undefined;
dec_optional_sexp(T) ->
dec_sexp(T).
conv_from_members_dict(D) ->
%% Use list_to_binary() here to "flatten" the serialized #p_srvr{}
[#mpb_membersdictentry{key=to_list(K), val=conv_from_p_srvr(V)} ||
{K, V} <- orddict:to_list(D)].
conv_to_members_dict(List) ->
orddict:from_list([{to_atom(K), conv_to_p_srvr(V)} ||
#mpb_membersdictentry{key=K, val=V} <- List]).
to_list(X) when is_atom(X) ->
atom_to_list(X);
to_list(X) when is_binary(X) ->
binary_to_list(X);
to_list(X) when is_integer(X) ->
integer_to_list(X);
to_list(X) when is_list(X) ->
X.
to_atom(X) when is_list(X) ->
list_to_atom(X);
to_atom(X) when is_binary(X) ->
erlang:binary_to_atom(X, latin1);
to_atom(X) when is_atom(X) ->
X.
to_integer(X) when is_list(X) ->
list_to_integer(X);
to_integer(X) when is_binary(X) ->
list_to_binary(binary_to_list(X));
to_integer(X) when is_integer(X) ->
X.
conv_from_now({A,B,C}) ->
#mpb_now{sec=(1000000 * A) + B,
usec=C}.
conv_to_now(#mpb_now{sec=Sec, usec=USec}) ->
{Sec div 1000000, Sec rem 1000000, USec}.
conv_from_mode(ap_mode) -> 'AP_MODE';
conv_from_mode(cp_mode) -> 'CP_MODE'.
conv_to_mode('AP_MODE') -> ap_mode;
conv_to_mode('CP_MODE') -> cp_mode.
conv_from_type(private) -> 'PRIVATE';
conv_from_type(public) -> 'PUBLIC'.
conv_to_type('PRIVATE') -> private;
conv_to_type('PUBLIC') -> public.
conv_from_status(ok) ->
'OK';
conv_from_status({error, bad_arg}) ->
'BAD_ARG';
conv_from_status({error, wedged}) ->
'WEDGED';
conv_from_status({error, bad_checksum}) ->
'BAD_CHECKSUM';
conv_from_status({error, partition}) ->
'PARTITION';
conv_from_status({error, not_written}) ->
'NOT_WRITTEN';
conv_from_status({error, written}) ->
'WRITTEN';
conv_from_status(_OOPS) ->
io:format(user, "HEY, ~s:~w got ~w\n", [?MODULE, ?LINE, _OOPS]),
'BAD_JOSS'.

39
test/machi_flu1_test.erl
View file

@ -173,7 +173,7 @@ flu_projection_smoke_test() ->
{ok, []} = ?FLU_C:list_all_projections(Host, TcpPort, T),
{ok, []} = ?FLU_C:get_all_projections(Host, TcpPort, T),
P_a = #p_srvr{name=a},
P_a = #p_srvr{name=a, address="localhost", port=4321},
P1 = machi_projection:new(1, a, [P_a], [], [a], [], []),
ok = ?FLU_C:write_projection(Host, TcpPort, T, P1),
{error, written} = ?FLU_C:write_projection(Host, TcpPort, T, P1),
@ -210,5 +210,42 @@ bad_checksum_test() ->
ok = ?FLU:stop(FLU1)
end.
%% The purpose of timing_pb_encoding_test_ and timing_bif_encoding_test_ is
%% to show the relative speed of the PB encoding of something like a
%% projection store command is about 35x slower than simply using the Erlang
%% BIFs term_to_binary() and binary_to_term(). We try to do enough work, at
%% least a couple of seconds, so that any dynamic CPU voltage adjustment
%% might kick into highest speed, in theory.
timing_pb_encoding_test_() ->
{timeout, 60, fun() -> timing_pb_encoding_test2() end}.
timing_pb_encoding_test2() ->
P_a = #p_srvr{name=a, address="localhost", port=4321},
P1 = machi_projection:new(1, a, [P_a], [], [a], [], []),
DoIt1 = fun() ->
Req = machi_pb_wrap:make_projection_req(
<<1,2,3,4>>, {write_projection, public, P1}),
Bin = list_to_binary(machi_pb:encode_mpb_ll_request(Req)),
ZZ = machi_pb:decode_mpb_ll_request(Bin),
_ = machi_pb_wrap:unmake_projection_req(ZZ)
end,
XX = lists:seq(1,70*1000),
erlang:garbage_collect(),
RUN1 = timer:tc(fun() -> begin [_ = DoIt1() || _ <- XX], ok end end),
erlang:garbage_collect(),
DoIt2 = fun() ->
Req = term_to_binary({
<<1,2,3,4>>, {write_projection, public, P1}}),
_ = binary_to_term(Req)
end,
erlang:garbage_collect(),
RUN2 = timer:tc(fun() -> begin [_ = DoIt2() || _ <- XX], ok end end),
erlang:garbage_collect(),
Factor = (element(1, RUN1) / element(1, RUN2)),
io:format(user, " speed factor=~.2f ", [Factor]),
ok.
-endif. % !PULSE
-endif. % TEST

2
test/machi_pb_high_client_test.erl
View file

@ -91,7 +91,7 @@ smoke_test2() ->
end
after
exit(SupPid, normal),
[os:cmd("rm -rf " ++ P#p_srvr.props) || {_,P} <- Ps],
[os:cmd("rm -rf " ++ P#p_srvr.props) || P <- Ps],
machi_util:wait_for_death(SupPid, 100),
ok
end.

17
test/machi_pb_test.erl
View file

@ -22,6 +22,7 @@
-module(machi_pb_test).
-include("machi_pb.hrl").
-include("machi_projection.hrl").
-compile(export_all).
@ -62,6 +63,22 @@ smoke_responses_test() ->
ok.
smoke_p_srvr_test() ->
P1 = #p_srvr{name=a, address="localhost", port=5555,
props=[{dir,"./data.a"}]},
P1 = machi_pb_wrap:dec_p_srvr(
list_to_binary(machi_pb_wrap:enc_p_srvr(P1))),
ok.
smoke_projection_v1_test() ->
P1 = #p_srvr{name=a, address="localhost", port=5555,
props=[{dir,"./data.a"}]},
D = orddict:from_list([ {P1#p_srvr.name, P1} ]),
Proj1 = machi_projection:new(a, D, [a], [], [], [{property, 42}]),
Proj1 = machi_pb_wrap:dec_projection_v1(
machi_pb_wrap:enc_projection_v1(Proj1)),
ok.
encdec_request(M) ->
machi_pb:decode_mpb_request(
list_to_binary(machi_pb:encode_mpb_request(M))).