diff --git a/aws-s3-lwt/io.ml b/aws-s3-lwt/io.ml index c032373..dd9e1e2 100644 --- a/aws-s3-lwt/io.ml +++ b/aws-s3-lwt/io.ml @@ -68,7 +68,7 @@ module Pipe = struct failwith (__LOC__ ^ ": Closed") let flush writer = - match Queue.length writer.queue = 0 && writer.closed with + match writer.closed with | true -> Lwt.return () | false -> let waiter, wakeup = Lwt.wait () in @@ -76,13 +76,28 @@ module Pipe = struct if Queue.length writer.queue = 1 then Lwt_condition.signal writer.cond (); waiter + (* Only a reader reaching a marker resolves the flush that queued it, so a + pipe nobody reads again has to release them itself: a producer bounding + itself on flush waits forever otherwise. *) + let wake_flushers pipe = + let data = Queue.create () in + Queue.iter + (function + | Flush wakeup -> Lwt.wakeup_later wakeup () + | Data _ as elem -> Queue.add elem data) + pipe.queue; + Queue.clear pipe.queue; + Queue.transfer data pipe.queue + let close (writer : 'a writer) = writer.closed <- true; + wake_flushers writer; Lwt_condition.broadcast writer.cond (); on_close writer let close_reader (reader : 'a reader) = reader.closed <- true; + wake_flushers reader; Lwt_condition.broadcast reader.cond (); on_close reader diff --git a/aws-s3.opam b/aws-s3.opam index 4c651fb..03307f2 100644 --- a/aws-s3.opam +++ b/aws-s3.opam @@ -15,6 +15,7 @@ depends: [ "ocaml" {>= "4.08.0"} "dune" {>= "2.0.0"} "inifiles" + "bigstringaf" {>= "0.5.0"} "digestif" {>= "0.7"} "ptime" "uri" diff --git a/aws-s3/authorization.ml b/aws-s3/authorization.ml index 8f4143f..e0a64bd 100644 --- a/aws-s3/authorization.ml +++ b/aws-s3/authorization.ml @@ -12,6 +12,9 @@ let log fmt = match debug with let hash_sha256 s = Digestif.SHA256.digest_string s +let hash_sha256_bigstring s = + Digestif.SHA256.digest_bigstring s + let hmac_sha256 ~key v = Digestif.SHA256.hmac_string ~key v diff --git a/aws-s3/authorization.mli b/aws-s3/authorization.mli index b767ef7..a3fd8e8 100644 --- a/aws-s3/authorization.mli +++ b/aws-s3/authorization.mli @@ -1,5 +1,6 @@ (**/**) val hash_sha256 : string -> Digestif.SHA256.t +val hash_sha256_bigstring : Bigstringaf.t -> Digestif.SHA256.t val hmac_sha256 : key:string -> string -> Digestif.SHA256.t val to_hex : Digestif.SHA256.t -> string diff --git a/aws-s3/aws.ml b/aws-s3/aws.ml index 0e77db2..461045d 100644 --- a/aws-s3/aws.ml +++ b/aws-s3/aws.ml @@ -96,6 +96,37 @@ module Make(Io : Types.Io) = struct in Pipe.create_reader ~f:(transfer initial_signature Digestif.SHA256.empty 0 [] None) + (* Sized to what the transport allocates per read anyway, so a bigstring body + reaches the socket without an object-sized string on the heap. *) + let body_slice_size = 65536 + + (* One slice ahead of the consumer, as {!chunk_writer} does: a pipe write does + not block, so without the flush the whole body would queue up as the + strings this exists to avoid. *) + let bigstring_reader body = + let rec send writer offset = + match Bigstringaf.length body - offset with + | _ when Pipe.is_closed writer -> return () + | 0 -> return () + | remain -> + let flushed = Pipe.flush writer in + let len = min body_slice_size remain in + Pipe.write writer (Bigstringaf.substring body ~off:offset ~len) >>= fun () -> + flushed >>= fun () -> + send writer (offset + len) + in + Pipe.create_reader ~f:(fun writer -> send writer 0) + + let reader_of_body ~chunked = function + | Body.String body -> + let reader, writer = Pipe.create () in + Pipe.write writer body >>= fun () -> + Pipe.close writer; + return (Some reader) + | Body.Bigstring body -> return (Some (bigstring_reader body)) + | Body.Empty -> return None + | Body.Chunked { pipe; chunk_size; _ } -> chunked ~pipe ~chunk_size + let make_request ~(endpoint: Region.endpoint) ?connect_timeout_ms ?(expect=false) ~sink ?(body=Body.Empty) ?(credentials:Credentials.t option) ~headers ~meth ~path ~query () = let (date, time) = Unix.gettimeofday () |> Time.iso8601_of_time in @@ -103,6 +134,7 @@ module Make(Io : Types.Io) = struct let content_length = match meth, body with | (`PUT | `POST), Body.String body -> Some (String.length body |> string_of_int) + | (`PUT | `POST), Body.Bigstring body -> Some (Bigstringaf.length body |> string_of_int) | (`PUT | `POST), Body.Chunked { length; chunk_size; _ } -> Some (get_chunked_length ~chunk_size length |> string_of_int ) | (`PUT | `POST), Body.Empty -> Some "0" @@ -111,6 +143,7 @@ module Make(Io : Types.Io) = struct let payload_sha = match body with | Body.Empty -> empty_sha | Body.String body -> Authorization.hash_sha256 body |> Authorization.to_hex + | Body.Bigstring body -> Authorization.hash_sha256_bigstring body |> Authorization.to_hex | Body.Chunked _ -> "STREAMING-AWS4-HMAC-SHA256-PAYLOAD" in let token = match credentials with @@ -153,32 +186,16 @@ module Make(Io : Types.Io) = struct ~headers ~query:query ~scope ~signing_key ~payload_sha in let auth = (Authorization.make_auth_header ~credentials ~scope ~signed_headers ~signature) in - let body = match body with - | Body.String body -> - let reader, writer = Pipe.create () in - Pipe.write writer body >>= fun () -> - Pipe.close writer; - return (Some reader) - | Body.Empty -> return None - | Body.Chunked { pipe; chunk_size; _ } -> - let pipe = + let body = + reader_of_body body ~chunked:(fun ~pipe ~chunk_size -> (* Get errors if the chunk_writer fails *) - chunk_writer ~signing_key ~scope - ~initial_signature:signature ~date ~time ~chunk_size pipe - in - return (Some pipe) + return (Some (chunk_writer ~signing_key ~scope + ~initial_signature:signature ~date ~time ~chunk_size pipe))) in Some auth, body | None -> - let body = match body with - | Body.String body -> - let reader, writer = Pipe.create () in - Pipe.write writer body >>= fun () -> - Pipe.close writer; - return (Some reader) - | Body.Empty -> return None - | Body.Chunked { pipe; _} -> - return (Some pipe) + let body = + reader_of_body body ~chunked:(fun ~pipe ~chunk_size:_ -> return (Some pipe)) in None, body in diff --git a/aws-s3/body.ml b/aws-s3/body.ml index b5978db..7569816 100644 --- a/aws-s3/body.ml +++ b/aws-s3/body.ml @@ -1,11 +1,67 @@ open StdLabels +(* Bytes gathered off the OCaml heap. It doubles because a sink is never told + the content length it is about to receive. *) +module Accumulator = struct + type t = { mutable buffer: Bigstringaf.t; mutable filled: int } + + let initial_size = 65536 + + let create () = { buffer = Bigstringaf.empty; filled = 0 } + + let reserve t needed = + match Bigstringaf.length t.buffer >= needed with + | true -> () + | false -> + let rec double size = match size >= needed with + | true -> size + | false -> double (size * 2) + in + let size = max (Bigstringaf.length t.buffer) initial_size in + let grown = Bigstringaf.create (double size) in + Bigstringaf.blit t.buffer ~src_off:0 grown ~dst_off:0 ~len:t.filled; + t.buffer <- grown + + let add_string t data = + let len = String.length data in + reserve t (t.filled + len); + Bigstringaf.blit_from_string data ~src_off:0 t.buffer ~dst_off:t.filled ~len; + t.filled <- t.filled + len + + (* A [sub] would share the slack doubling left behind, and the caller holds + the result for as long as it holds the body. *) + let contents t = + match t.filled = Bigstringaf.length t.buffer with + | true -> t.buffer + | false -> Bigstringaf.copy t.buffer ~off:0 ~len:t.filled +end + +let%test "accumulator answers exactly what was added, in order" = + let acc = Accumulator.create () in + let expected = Buffer.create 0 in + (* Pieces of varying length carrying their own index, so a lost, doubled or + misplaced one shows up rather than being covered by its neighbour. *) + for i = 1 to 5000 do + let piece = Printf.sprintf "%d:%s|" i (String.make (i mod 97) 'x') in + Buffer.add_string expected piece; + Accumulator.add_string acc piece + done; + let result = Accumulator.contents acc in + let expected = Buffer.contents expected in + String.length expected > Accumulator.initial_size + && Bigstringaf.length result = String.length expected + && Bigstringaf.to_string result = expected + +let%test "an accumulator nothing was added to is empty" = + Bigstringaf.length (Accumulator.contents (Accumulator.create ())) = 0 + module Make(Io : Types.Io) = struct open Io open Deferred type t = | String of string + | Bigstring of Bigstringaf.t | Empty | Chunked of { pipe: string Pipe.reader; length: int; chunk_size: int } @@ -27,6 +83,19 @@ module Make(Io : Types.Io) = struct in loop [] + (* Each fragment is dropped as soon as it is gathered, where {!to_string} + holds every one of them until the end. *) + let to_bigstring body = + let acc = Accumulator.create () in + let rec loop () = + Pipe.read body >>= function + | Some data -> + Accumulator.add_string acc data; + loop () + | None -> return (Accumulator.contents acc) + in + loop () + let read_string ?start ~length reader = let rec loop acc data remain = match data, remain with diff --git a/aws-s3/body.mli b/aws-s3/body.mli index ee3a585..e7424e5 100644 --- a/aws-s3/body.mli +++ b/aws-s3/body.mli @@ -2,6 +2,7 @@ module Make(Io : Types.Io) : sig open Io type t = | String of string + | Bigstring of Bigstringaf.t | Empty | Chunked of { pipe : string Pipe.reader; length : int; chunk_size : int; } (**/**) @@ -9,6 +10,11 @@ module Make(Io : Types.Io) : sig string Pipe.reader -> string Deferred.t + (** {!to_string} without putting the body on the OCaml heap. *) + val to_bigstring : + string Pipe.reader -> + Bigstringaf.t Deferred.t + val read_string : ?start:string -> length:int -> diff --git a/aws-s3/dune b/aws-s3/dune index 35a61ce..eaa4638 100644 --- a/aws-s3/dune +++ b/aws-s3/dune @@ -2,7 +2,7 @@ (name aws_s3) (public_name aws-s3) (synopsis "Amazon S3 access library") - (libraries ptime inifiles digestif.c + (libraries ptime inifiles bigstringaf digestif.c base64 uri yojson ppx_protocol_conv_json ppx_protocol_conv_xmlm str) diff --git a/aws-s3/http.ml b/aws-s3/http.ml index dac2724..ffae05b 100644 --- a/aws-s3/http.ml +++ b/aws-s3/http.ml @@ -164,5 +164,7 @@ module Make(Io : Types.Io) = struct Pipe.close writer; Pipe.close_reader reader; Pipe.close sink; + (* A body the request never got to send still has a producer behind it. *) + (match body with Some body -> Pipe.close_reader body | None -> ()); return result end diff --git a/aws-s3/s3.ml b/aws-s3/s3.ml index bf0ff0b..62b195b 100644 --- a/aws-s3/s3.ml +++ b/aws-s3/s3.ml @@ -85,10 +85,21 @@ module Protocol(P: sig type 'a result end) = struct last_modified: time [@key "LastModified"]; key: string [@key "Key"]; etag: etag [@key "ETag"]; + response_headers: (string * string) list option; [@default None] meta_headers: (string * string) list option; [@default None] (** Add expiration date option *) } [@@deriving of_protocol ~driver:(module Protocol_conv_xmlm.Xmlm)] + (* The server picks the case it sends a header name in. *) + let find_header ~name content = + let name = String.lowercase_ascii name in + match content.response_headers with + | None -> None + | Some headers -> + match List.find_opt ~f:(fun (key, _) -> String.lowercase_ascii key = name) headers with + | Some (_, value) -> Some value + | None -> None + module Ls = struct type result = { @@ -227,6 +238,10 @@ module Make(Io : Types.Io) = struct let reader, writer = Pipe.create () in Body.to_string reader, writer + let bigstring_sink () = + let reader, writer = Pipe.create () in + Body.to_bigstring reader, writer + include Protocol(struct type nonrec 'a result = ('a, error) result Deferred.t end) type range = { first: int option; last: int option } @@ -346,12 +361,22 @@ module Make(Io : Types.Io) = struct let body = Body.String data in put_common ?credentials ?connect_timeout_ms ?confirm_requester_pays ?content_type ?content_encoding ?acl ?cache_control ?expect ?meta_headers ~endpoint ~bucket ~key ~body () + let put_bigstring ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint ?content_type ?content_encoding ?acl ?cache_control ?expect ?meta_headers ~bucket ~key ~data () = + let body = Body.Bigstring data in + put_common ?credentials ?connect_timeout_ms ?confirm_requester_pays ?content_type ?content_encoding ?acl ?cache_control ?expect ?meta_headers ~endpoint ~bucket ~key ~body () + let get ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint ?range ~bucket ~key () = let body, data = string_sink () in Stream.get ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint ?range ~bucket ~key ~data () >>=? fun () -> body >>= fun body -> Deferred.return (Ok body) + let get_bigstring ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint ?range ~bucket ~key () = + let body, data = bigstring_sink () in + Stream.get ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint ?range ~bucket ~key ~data () >>=? fun () -> + body >>= fun body -> + Deferred.return (Ok body) + let delete ?credentials ?connect_timeout_ms ?(confirm_requester_pays=false) ~endpoint ~bucket ~key () = let path = sprintf "/%s/%s" bucket key in let sink = Body.null () in @@ -371,6 +396,7 @@ module Make(Io : Types.Io) = struct Aws.make_request ?credentials ?connect_timeout_ms ~endpoint ~headers ~meth:`HEAD ~path ~query:[] ~sink () in do_command ~endpoint cmd >>=? fun headers -> + let response_headers = Some (Headers.bindings headers) in let result = let (>>=) a f = match a with | Some x -> f x @@ -388,7 +414,8 @@ module Make(Io : Types.Io) = struct storage_class_of_xmlm_exn (make_xmlm_node "p" [] [`Data s]) ) in - Some { storage_class; size; last_modified; key; etag = unquote etag; meta_headers = Some meta_headers} + Some { storage_class; size; last_modified; key; etag = unquote etag; + response_headers; meta_headers = Some meta_headers } in match result with | Some r -> Deferred.return (Ok r) @@ -500,7 +527,7 @@ module Make(Io : Types.Io) = struct [part_number] specifies the part numer. Parts will be assembled in order, but does not have to be consecutive *) - let upload_part ?credentials ?connect_timeout_ms ?(confirm_requester_pays=false) ~endpoint t ~part_number ?expect ~data () = + let upload_part_common ?credentials ?connect_timeout_ms ?(confirm_requester_pays=false) ~endpoint t ~part_number ?expect ~body () = let path = sprintf "/%s/%s" t.bucket t.key in let query = [ "partNumber", string_of_int part_number; @@ -510,7 +537,7 @@ module Make(Io : Types.Io) = struct let headers = maybe_add_request_payer confirm_requester_pays [] in let cmd () = Aws.make_request ?expect ?credentials ?connect_timeout_ms ~endpoint ~headers ~meth:`PUT ~path - ~body:(Body.String data) ~query ~sink () + ~body ~query ~sink () in do_command ~endpoint cmd >>=? fun headers -> let etag = @@ -521,6 +548,12 @@ module Make(Io : Types.Io) = struct t.parts <- { etag; part_number } :: t.parts; Deferred.return (Ok etag) + let upload_part ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint t ~part_number ?expect ~data () = + upload_part_common ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint t ~part_number ?expect ~body:(Body.String data) () + + let upload_part_bigstring ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint t ~part_number ?expect ~data () = + upload_part_common ?credentials ?connect_timeout_ms ?confirm_requester_pays ~endpoint t ~part_number ?expect ~body:(Body.Bigstring data) () + (** Specify a part to be a file on s3. [range] can be used to only include a part of the s3 file *) @@ -693,6 +726,42 @@ let%test "parse Error_response.t" = let error = Protocol.Error_response.of_xmlm_exn xml in "PermanentRedirect" = error.Protocol.Error_response.code +let%test "response_headers is None in XML-parsed content" = + let module Protocol = Protocol(struct type 'a result = 'a end) in + let data = {| + + s3_osd + + 1 + 1000 + false + + STANDARD + test + 2018-02-27T13:39:35.000Z + "7538d2bd85ea5dfb689ed65a0f60a7aa" + 20 + + + |} + in + let xml = xmlm_of_string data in + let result = Protocol.Ls.result_of_xmlm_exn xml in + let c = List.hd result.Protocol.Ls.contents in + c.Protocol.response_headers = None + +let%test "find_header ignores the case the server chose" = + let module Protocol = Protocol(struct type 'a result = 'a end) in + let content = + { Protocol.storage_class = Protocol.Standard; size = 0; last_modified = 0.; + key = "test"; etag = "e"; + response_headers = Some ["Content-Type", "text/plain"]; + meta_headers = None } + in + Protocol.find_header ~name:"content-type" content = Some "text/plain" + && Protocol.find_header ~name:"content-encoding" content = None + && Protocol.find_header ~name:"content-type" { content with Protocol.response_headers = None } = None + let%test "parse Delete_multi.result" = let module Protocol = Protocol(struct type 'a result = 'a end) in let data = diff --git a/aws-s3/s3.mli b/aws-s3/s3.mli index 051736a..cb7bf6f 100644 --- a/aws-s3/s3.mli +++ b/aws-s3/s3.mli @@ -33,10 +33,17 @@ module Make(Io : Types.Io) : sig last_modified : float; (** Seconds since epoch *) key : string; etag : etag; (** Etag as a string. this us usually the MD5, unless the object was constructed by multi-upload *) + response_headers : (string * string) list option; + (** All headers of the response. If None, the information was not + retrieved: [ls] parses an xml listing, which carries no per-object + headers. *) meta_headers: (string * string) list option; (** Meta headers. If None, the information was not retrieved. *) - } + (** Look up a response header, ignoring the case the server chose for its + name. *) + val find_header : name:string -> content -> string option + type nonrec 'a result = ('a, error) result Deferred.t (** The type of S3 requests. [credentials] refers to AWS @@ -94,6 +101,22 @@ module Make(Io : Types.Io) : sig key:string -> data:string -> unit -> etag result) command + (** {!put} with a body that is never held on the OCaml heap. + + [data] is read for the duration of the request rather than copied up + front, so its bytes must stay valid until the result is determined. + *) + val put_bigstring : + (?content_type:string -> + ?content_encoding:string -> + ?acl:string -> + ?cache_control:string -> + ?expect:bool -> + ?meta_headers:(string * string) list -> + bucket:string -> + key:string -> + data:Bigstringaf.t -> unit -> etag result) command + (** Download [key] from s3 in [bucket] If [range] is specified, only a part of the file is retrieved: - If [first] is None, then start from the beginning of the object. @@ -102,6 +125,10 @@ module Make(Io : Types.Io) : sig val get : (?range:range -> bucket:string -> key:string -> unit -> string result) command + (** {!get} answering off the OCaml heap. *) + val get_bigstring : + (?range:range -> bucket:string -> key:string -> unit -> Bigstringaf.t result) command + (** Call head on the object to retrieve info on a single object *) val head : (bucket:string -> key:string -> unit -> content result) command @@ -218,6 +245,16 @@ module Make(Io : Types.Io) : sig unit -> etag result) command + (** {!upload_part} with a body that is never held on the OCaml heap, under + the same ownership rule as {!Aws_s3.S3.Make.put_bigstring}. *) + val upload_part_bigstring : + (t -> + part_number:int -> + ?expect:bool -> + data:Bigstringaf.t -> + unit -> + etag result) command + (** Specify a part as a copy of an existing object in S3. *) val copy_part : (t -> part_number:int -> ?range:int * int -> bucket:string -> key:string -> unit -> unit result) command diff --git a/cli/aws.ml b/cli/aws.ml index 9cd67ff..9b0ccb9 100644 --- a/cli/aws.ml +++ b/cli/aws.ml @@ -57,11 +57,50 @@ module Make(Io : Aws_s3.Types.Io) = struct Io.Deferred.async (Io.Pipe.closed reader >>= fun () -> close_in ic; return ()); reader + (* Block at a time through one reusable buffer: a [Bytes] the size of the file + would put the whole body on the heap, which is what the bigstring is here + to avoid. *) + let read_file_bigstring ~pos ~len file = + let block = 65536 in + let ic = open_in_bin file in + seek_in ic pos; + let data = Bigstringaf.create len in + let buffer = Bytes.create (min block len) in + let rec read offset = + match len - offset with + | 0 -> () + | remain -> + let n = min block remain in + really_input ic buffer 0 n; + Bigstringaf.blit_from_bytes buffer ~src_off:0 data ~dst_off:offset ~len:n; + read (offset + n) + in + read 0; + close_in ic; + data + let save_file file contents = let oc = open_out file in output_string oc contents; close_out oc + let save_file_bigstring file contents = + let block = 65536 in + let len = Bigstringaf.length contents in + let oc = open_out_bin file in + let buffer = Bytes.create (min block len) in + let rec write offset = + match len - offset with + | 0 -> () + | remain -> + let n = min block remain in + Bigstringaf.blit_to_bytes contents ~src_off:offset buffer ~dst_off:0 ~len:n; + output oc buffer 0 n; + write (offset + n) + in + write 0; + close_out oc + type objekt = { bucket: string; key: string } let objekt_of_uri u = match String.split_on_char ~sep:'/' u with @@ -115,11 +154,18 @@ module Make(Io : Aws_s3.Types.Io) = struct (S3.retry ~endpoint ~retries ~f:(f ~size)) () :: upload_parts t endpoint ~retries ~expect ~credentials ~offset:(offset + size) ~total ~part_number:(part_number + 1) ?chunk_size src - let cp profile endpoint ~retries ~expect ~confirm_requester_pays ?(use_multi=false) ?first ?last ?chunk_size src dst = + let cp profile endpoint ~retries ~expect ~confirm_requester_pays ?(use_multi=false) ?(use_bigstring=false) ?first ?last ?chunk_size src dst = let range = { S3.first; last } in Credentials.Helper.get_credentials ?profile () >>= fun credentials -> let credentials = ok_exn credentials in match determine_paths src dst with + | S3toLocal (src, dst) when use_bigstring && chunk_size = None -> + let f ~endpoint () = + S3.get_bigstring ~endpoint ~credentials ~range ~confirm_requester_pays ~bucket:src.bucket ~key:src.key () + in + S3.retry ~endpoint ~retries ~f () >>=? fun data -> + save_file_bigstring dst data; + Deferred.return (Ok ()) | S3toLocal (src, dst) -> let f ~endpoint () = match chunk_size with | None -> @@ -157,12 +203,16 @@ module Make(Io : Aws_s3.Types.Io) = struct | None -> file_length src - pos | Some l -> l - pos in - let f = match chunk_size with - | None -> + let f = match chunk_size, use_bigstring with + | None, true -> + let data = read_file_bigstring ~pos ~len src in + fun ~endpoint () -> S3.put_bigstring ~endpoint ~expect ~credentials ~bucket:dst.bucket ~key:dst.key + ~data () + | None, false -> let data = read_file ~pos ~len src in fun ~endpoint () -> S3.put ~endpoint ~expect ~credentials ~bucket:dst.bucket ~key:dst.key ~data () - | Some chunk_size -> + | Some chunk_size, _ -> let reader = file_reader ~pos ~len src in fun ~endpoint () -> S3.Stream.put ~endpoint ~expect ~credentials ~bucket:dst.bucket ~key:dst.key ~data:reader ~chunk_size ~length:len () @@ -247,8 +297,8 @@ module Make(Io : Aws_s3.Types.Io) = struct let endpoint = Aws_s3.Region.endpoint ~inet ~scheme region in begin match cmd with - | Cli.Cp { src; dest; first; last; multi; chunk_size } -> - cp profile endpoint ~retries ~confirm_requester_pays ~expect ~use_multi:multi ?first ?last ?chunk_size src dest + | Cli.Cp { src; dest; first; last; multi; chunk_size; bigstring } -> + cp profile endpoint ~retries ~confirm_requester_pays ~expect ~use_multi:multi ~use_bigstring:bigstring ?first ?last ?chunk_size src dest | Rm { bucket; paths } -> rm profile endpoint ~retries ~confirm_requester_pays bucket paths | Ls { ratelimit; bucket; prefix; start_after; max_keys } -> diff --git a/cli/cli.ml b/cli/cli.ml index d5199e9..53107ac 100644 --- a/cli/cli.ml +++ b/cli/cli.ml @@ -5,7 +5,7 @@ type actions = | Ls of { bucket: string; prefix: string option; start_after: string option; ratelimit: int option; max_keys: int option} | Head of { path: string; } | Rm of { bucket: string; paths : string list } - | Cp of { src: string; dest: string; first: int option; last: int option; multi: bool; chunk_size: int option} + | Cp of { src: string; dest: string; first: int option; last: int option; multi: bool; chunk_size: int option; bigstring: bool } type options = { profile: string option; minio: string option; https: bool; retries: int; ipv6: bool; expect: bool; confirm_requester_pays : bool } @@ -69,8 +69,8 @@ let parse exec = in let cp = - let make opts first last multi chunk_size src dest = - opts, Cp { src; dest; first; last; multi; chunk_size } + let make opts first last multi chunk_size bigstring src dest = + opts, Cp { src; dest; first; last; multi; chunk_size; bigstring } in let first = let doc = "first byte of the source object to copy. If omitted means from the start." in @@ -89,9 +89,13 @@ let parse exec = let doc = "Use streaming get / put the given chunk_size" in Arg.(value & opt (some int) None & info ["chunk-size"; "c"] ~docv:"CHUNK SIZE" ~doc) in + let bigstring = + let doc = "Carry the body off the OCaml heap. Ignored when streaming." in + Arg.(value & flag & info ["bigstring"; "B"] ~docv:"BIGSTRING" ~doc) + in Cmd.v Cmd.(info "cp" ~doc:"Copy files to and from S3") - Term.(const make $ common_opts $ first $ last $ multi $ chunk_size $ path 0 "SRC" $ path 1 "DEST") + Term.(const make $ common_opts $ first $ last $ multi $ chunk_size $ bigstring $ path 0 "SRC" $ path 1 "DEST") in let rm = let objects = diff --git a/cli/dune b/cli/dune index a5d9820..979e128 100644 --- a/cli/dune +++ b/cli/dune @@ -1,5 +1,5 @@ (library (name aws_cli) (modules "aws" "cli") - (libraries aws-s3 cmdliner) + (libraries aws-s3 bigstringaf cmdliner) ) diff --git a/integration.sh b/integration.sh index f440f88..f2c7a3d 100755 --- a/integration.sh +++ b/integration.sh @@ -108,6 +108,30 @@ function test_complete () { test "download" ${BIN} cp ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" ${TEMP} test "data" diff -u $LARGE_FILE ${TEMP} + # Whole-body transfers carrying the payload off the OCaml heap. The large + # file is what matters here: it spans many slices in each direction, where + # the small one fits in a single read. + test "upload bigstring" ${BIN} cp -B ${OPTIONS} $LARGE_FILE "s3://${BUCKET}/${PREFIX}test" + test "download bigstring" ${BIN} cp -B ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" ${TEMP} + test "data" diff -u $LARGE_FILE ${TEMP} + + test "upload bigstring expect" ${BIN} cp -B -e ${OPTIONS} $LARGE_FILE "s3://${BUCKET}/${PREFIX}test" + test "download bigstring" ${BIN} cp -B ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" ${TEMP} + test "data" diff -u $LARGE_FILE ${TEMP} + + # Each spelling against the other's, so a body that is merely + # self-consistent does not pass. + test "upload bigstring" ${BIN} cp -B ${OPTIONS} $LARGE_FILE "s3://${BUCKET}/${PREFIX}test" + test "download" ${BIN} cp ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" ${TEMP} + test "data" diff -u $LARGE_FILE ${TEMP} + + test "upload" ${BIN} cp ${OPTIONS} $LARGE_FILE "s3://${BUCKET}/${PREFIX}test" + test "download bigstring" ${BIN} cp -B ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" ${TEMP} + test "data" diff -u $LARGE_FILE ${TEMP} + + test "partial download bigstring" ${BIN} cp -B ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" --first=$FIRST_PART --last=$LAST_PART ${TEMP} + test "partial data" diff -u ${PART} ${TEMP} + test "partial download" ${BIN} cp ${OPTIONS} "s3://${BUCKET}/${PREFIX}test" --first=$FIRST_PART --last=$LAST_PART ${TEMP} test "partial data" diff -u ${PART} ${TEMP} @@ -122,7 +146,12 @@ function test_complete () { } for TYPE in ${TYPES:-lwt}; do - opam exec -- dune build aws-s3-${TYPE}/bin/aws_cli_${TYPE}.exe + # Without this the suite runs against whatever was built last, and a build + # that never compiled the change still reports every test green. + if ! opam exec -- dune build aws-s3-${TYPE}/bin/aws_cli_${TYPE}.exe; then + echo "Failed to build aws-s3-${TYPE}, refusing to test a stale binary" + exit 1 + fi BIN=_build/default/aws-s3-${TYPE}/bin/aws_cli_${TYPE}.exe if [ -z "${MINIO}" ]; then