(* $Id: mr_io.ml 437 2011-10-09 18:12:51Z gerd $ *)
(* Testing Mapred_io.read_file *)
(* We create test files:
F1: empty file
F2: fill the file with 26 lines. Each line has a certain length:
F2.1: bigblock/2 - 2
F2.2: bigblock/2 - 1
F2.3: bigblock/2
F2.4: bigblock - 2
F2.5: bigblock - 1
F2.6: bigblock
(line length includes here the final LF)
F3: fill the file with a first line of specific length, followed by
26 lines with a different length:
F3.1: 1 and bigblock/2 - 3
F3.2: 2 and bigblock/2 - 3
F3.3: 3 and bigblock/2 - 3
F3.4: 11 and bigblock/2 - 3
F3.5: 12 and bigblock/2 - 3
F3.6: 13 and bigblock/2 - 3
Requested region:
R1: request nothing
R2: request complete file
R3: request all whole bigblocks (but not the trailing partial bigblock)
R4: request n-1 whole bigblocks
R5: request bigblock #1 (i.e. the second bigblock)
We always use bigblock=2*block.
Tests are repeated with various buffer sizes (which should not affect
the outcome).
Expected outcome: The test checks that a certain number of lines N are
read, and that these lines contain the right contents.
R1 + all F: N=0
R2 + F1: N=0
R2 + F2: N=26
R2 + F3: N=27
R3 + F1: N=0 (actually nonsense - no whole bigblocks)
R3 + F2.1, F2.2: N=25
R3 + F2.3, F2.4, F2.5, F2.6: N=26
R3 + F3: N=26
R4 + F1: N=0 (actually nonsense - no whole bigblocks)
R4 + F2.1, F2.2: N=23
R4 + F2.3: N=25
R4 + F2.4, F2.5: N=25
R4 + F2.6: N=26
R4 + F3: N=24
R5 + F1: invalid
R5 + F2.1, F2.2: N=2
R5 + F2.3: N=3
R5 + F2.4, 2.5, 2.6: N=1
R5 + F3: N=2
*)
open Printf
open Plasma_rpcapi_aux
open Uq_engines.Operators
let f_cases =
[ "F1";
"F2.1"; "F2.2"; "F2.3"; "F2.4"; "F2.5"; "F2.6";
"F3.1"; "F3.2"; "F3.3"; "F3.4"; "F3.5"; "F3.6";
]
let r_cases =
[ "R1"; "R2"; "R3"; "R4"; "R5" ]
let write_file cluster name s =
let trans = Plasma_client.start cluster in
( try Plasma_client.unlink trans name
with Plasma_client.Plasma_error `enoent -> ()
);
let ii = Plasma_client.regular_ii cluster 0o666 in
let inode = Plasma_client.create_file trans name ii in
let () = Plasma_client.commit trans in
let rec wrloop p =
let l = String.length s - p in
if l > 0 then (
let n =
Plasma_client.write cluster inode (Int64.of_int p) (`String s) p l in
wrloop (p+n)
) in
wrloop 0;
Plasma_client.flush cluster inode 0L 0L
let create_line k l =
(* line number k of the file (k>=0). l is the length including LF *)
assert(l > 0);
String.make (l-1) (Char.chr (64+k)) ^ "\n"
let check_line k line =
let c = Char.chr (64+k) in
for j = 0 to String.length line - 1 do
if line.[j] <> c then (
failwith (sprintf "bad line: expected #%d found #%d"
k
(Char.code line.[j] - 64));
)
done
let name_fcase fcase =
"/mr_io_" ^ fcase
let create_fcase cluster fcase =
let name = name_fcase fcase in
let b = Buffer.create 500 in
let bsize = Plasma_client.blocksize cluster in
let bigsize = 2*bsize in
let f2 n =
for k = 0 to 25 do
let line = create_line k n in
Buffer.add_string b line
done in
let f3 n1 n2 =
let line = create_line 0 n1 in
Buffer.add_string b line;
for k = 1 to 26 do
let line = create_line k n2 in
Buffer.add_string b line
done in
( match fcase with
| "F1" -> ()
| "F2.1" -> f2 (bigsize/2-2)
| "F2.2" -> f2 (bigsize/2-1)
| "F2.3" -> f2 (bigsize/2-0)
| "F2.4" -> f2 (bigsize-2)
| "F2.5" -> f2 (bigsize-1)
| "F2.6" -> f2 (bigsize-0)
| "F3.1" -> f3 1 (bigsize/2-3)
| "F3.2" -> f3 2 (bigsize/2-3)
| "F3.3" -> f3 3 (bigsize/2-3)
| "F3.4" -> f3 11 (bigsize/2-3)
| "F3.5" -> f3 12 (bigsize/2-3)
| "F3.6" -> f3 13 (bigsize/2-3)
| _ -> failwith "unknown fcase"
);
let s = Buffer.contents b in
write_file cluster name s
let request_length cluster name rcase =
let trans = Plasma_client.start cluster in
let ii = Plasma_client.get_inodeinfo trans
(Plasma_client.lookup trans name false) in
let () = Plasma_client.abort trans in
let bsize = Plasma_client.blocksize cluster in
let bigsize = 2*bsize in
match rcase with
| "R1" -> 0L
| "R2" -> ii.blocklimit
| "R3" -> Int64.mul 2L (Int64.div ii.eof (Int64.of_int bigsize))
| "R4" -> max (Int64.mul 2L (Int64.pred (Int64.div ii.eof (Int64.of_int bigsize)))) 0L
| "R5" -> 2L
| _ -> failwith "unknown rcase"
let request_offset cluster name rcase =
(* at which block we begin to read *)
match rcase with
| "R1" | "R2" | "R3" | "R4" -> 0L
| "R5" -> 2L
| _ -> failwith "unknown rcase"
let line_offset fcase rcase =
(* the line number of the first read line *)
match rcase with
| "R1" | "R2" | "R3" | "R4" -> 0
| "R5" ->
( match fcase with
| "F1" -> assert false
| "F2.1" | "F2.2" | "F2.3" -> 3
| "F2.4" | "F2.5" | "F2.6" -> 2
| "F3.1" | "F3.2" | "F3.3" -> 4
| "F3.4" | "F3.5" | "F3.6" -> 3
| _ -> failwith "unknown fcase"
)
| _ -> failwith "unknown rcase"
let expected_num_lines fcase rcase =
match rcase with
| "R1" ->
( match fcase with
| "F1" -> 0
| _ -> 1
)
| "R2" ->
( match fcase with
| "F1" -> 0
| "F2.1" | "F2.2" | "F2.3" | "F2.4" | "F2.5" | "F2.6" -> 26
| "F3.1" | "F3.2" | "F3.3" | "F3.4" | "F3.5" | "F3.6" -> 27
| _ -> failwith "unknown fcase"
)
| "R3" ->
( match fcase with
| "F1" -> 0
| "F2.1" | "F2.2" -> 25
| "F2.3" | "F2.4" | "F2.5" | "F2.6" -> 26
| "F3.1" | "F3.2" | "F3.3" | "F3.4" | "F3.5" | "F3.6" -> 26
| _ -> failwith "unknown fcase"
)
| "R4" ->
( match fcase with
| "F1" -> 0
| "F2.1" | "F2.2" -> 23
| "F2.3" -> 25
| "F2.4" | "F2.5" -> 25
| "F2.6" -> 26
| "F3.1" | "F3.2" | "F3.3" | "F3.4" | "F3.5" | "F3.6" -> 24
| _ -> failwith "unknown fcase"
)
| "R5" ->
( match fcase with
| "F1" -> assert false
| "F2.1" | "F2.2" | "F2.3" -> 2
| "F2.4" | "F2.5" | "F2.6" -> 1
| "F3.4" | "F3.5" -> 3
| "F3.1" | "F3.2" | "F3.3" | "F3.6" -> 2
| _ -> failwith "unknown fcase"
)
| _ ->
failwith "unknown rcase"
let run_test cluster mr_buf_size fcase rcase =
printf "Test: %s + %s for buf_size=%d\n%!" rcase fcase mr_buf_size;
try
let name = name_fcase fcase in
let bsize = Plasma_client.blocksize cluster in
let rc =
( object
method bigblock_size = 2*bsize
method mr_buffer_size = mr_buf_size
method mr_buffer_size_tight = mr_buf_size
end
) in
let reqlen = request_length cluster name rcase in
let reqoff = request_offset cluster name rcase in
let lineoff = line_offset fcase rcase in
printf " reqoff=%Ld reqlen=%Ld lineoff=%d\n%!" reqoff reqlen lineoff;
let rr =
Mapred_io.read_file cluster rc name reqoff reqlen in
( try
while true do
let k = rr#pos_in in
let line = rr#input_record () in
check_line (lineoff+k) line
done
with
| End_of_file -> ()
);
let n = expected_num_lines fcase rcase in
if n <> rr#pos_in then
failwith (sprintf "bad number of lines - expected: %d - actual: %d"
n rr#pos_in);
printf " test passed\n%!"
with
| Failure msg ->
printf " TEST FAILED: %s\n%!" msg
let run_test_fd cluster mr_buf_size fcase rcase =
printf "FD Test: %s + %s for buf_size=%d\n%!" rcase fcase mr_buf_size;
let esys = Plasma_client.event_system cluster in
try
let name = name_fcase fcase in
let bsize = Plasma_client.blocksize cluster in
let rc =
( object
method bigblock_size = 2*bsize
method mr_buffer_size = mr_buf_size
method mr_buffer_size_tight = mr_buf_size
end
) in
let reqlen = request_length cluster name rcase in
let reqoff = request_offset cluster name rcase in
let lineoff = line_offset fcase rcase in
printf " reqoff=%Ld reqlen=%Ld lineoff=%d\n%!" reqoff reqlen lineoff;
let rr =
Mapred_io.read_file cluster rc name reqoff reqlen in
let (fd_rd, fd_wr) = Unix.pipe() in
Unix.set_nonblock fd_rd;
Unix.set_nonblock fd_wr;
let fd_wr_closed = ref false in
let d1 = `Polldescr(`Read_write, fd_rd, esys) in
let d2 = `Buffer_in(Uq_io.create_in_buffer d1) in
let pump_e =
rr # to_fd_e fd_wr
++ (fun () ->
Unix.close fd_wr; fd_wr_closed := true; eps_e (`Done ()) esys) in
let rec read_e k =
(Uq_io.input_line_e d2 >> Uq_io.eof_as_none)
++ (fun line_opt ->
match line_opt with
| None -> (* EOF *)
eps_e (`Done k) esys
| Some line ->
check_line (lineoff + k) line;
read_e (k + 1)
)
in
let e =
Uq_engines.sync_engine
pump_e
(read_e 0)
++ (fun ((), k_end) ->
let n = expected_num_lines fcase rcase in
if n <> k_end then
failwith (sprintf
"bad number of lines - expected: %d - actual: %d"
n k_end);
eps_e (`Done ()) esys
) in
Unixqueue.run esys;
( match e#state with
| `Working _ -> assert false
| `Aborted -> assert false
| `Error err -> raise err
| `Done () -> ()
);
Unix.close fd_rd;
if not !fd_wr_closed then Unix.close fd_wr;
printf " test passed\n%!"
with
| Failure msg ->
printf " TEST FAILED: %s\n%!" msg
let run ?clustername ?nn_nodes() =
let cconf = Plasma_client_config.get_config ?clustername ?nn_nodes() in
let esys = Unixqueue.create_unix_event_system() in
let cluster = Plasma_client.open_cluster_cc cconf esys in
Plasma_client.configure_buffer cluster 100;
Plasma_client.configure_auth_daemon cluster;
printf "Creating files...\n%!";
(* Create files *)
List.iter
(fun fcase -> create_fcase cluster fcase)
f_cases;
let bsize = Plasma_client.blocksize cluster in
List.iter
(fun bufsize ->
List.iter
(fun rcase ->
List.iter
(fun fcase ->
if rcase<>"R5" || fcase <> "F1" then (
run_test cluster bufsize fcase rcase;
run_test_fd cluster bufsize fcase rcase;
)
)
f_cases
)
r_cases
)
[ bsize; 2*bsize; 3*bsize; 8*bsize ];
printf "End of test sequence.\n%!"
let main () =
let clustername = ref None in
let nn_nodes = ref [] in
let debug = ref false in
Arg.parse
[ "-namenode", Arg.String (fun s -> nn_nodes := s :: !nn_nodes),
"<host>:<port> Access this namenode";
"-cluster", Arg.String (fun s -> clustername := Some s),
"<name> name of the cluster";
"-debug", Arg.Set debug,
" run with debugging enabled"
]
(fun s -> raise(Arg.Bad("Unexpected arg: " ^ s)))
"usage: mr_io <options>";
Netlog.current_logger :=
Netlog.channel_logger stderr (if !debug then `Debug else `Info);
Plasma_util.debug := !debug;
try
run
?clustername:!clustername
?nn_nodes:(if !nn_nodes = [] then None else Some !nn_nodes)
()
with
| error ->
eprintf "Exception %s\n%!" (Netexn.to_string error)
let () = main()
