Update to object_store_ffi v0.3 which supports multipart get and put (#23)

Author: jfunstonRAI

Project.toml

@@ -1,6 +1,6 @@
 name = "RustyObjectStore"
 uuid = "1b5eed3d-1f46-4baa-87f3-a4a892b23610"
-version = "0.1.0"
+version = "0.2.0"
 
 [deps]
 DocStringExtensions = "ffbed154-4ef7-542d-bbb7-c09d3a79fcae"
@@ -16,7 +16,7 @@ ReTestItems = "1"
 Sockets = "1"
 Test = "1"
 julia = "1.8"
-object_store_ffi_jll = "0.2"
+object_store_ffi_jll = "0.3"
 
 [extras]
 CloudBase = "85eb1798-d7c4-4918-bb13-c944d38e27ed"

README.md

@@ -41,6 +41,39 @@ nbytes_read = get_object!(buffer, "path/to/example.csv", config)
 @assert String(buffer[1:nbytes_read]) == input
 ```
 
+One-time global configuration can be set using a StaticConfig object passed to init\_object\_store():
+```julia
+test_config = StaticConfig(
+    n_threads=0,
+    cache_capacity=20,
+    cache_ttl_secs=30 * 60,
+    cache_tti_secs=5 * 60,
+    multipart_put_threshold=8 * 1024 * 1024,
+    multipart_get_threshold=8 * 1024 * 1024,
+    multipart_get_part_size=8 * 1024 * 1024
+)
+init_object_store(test_config)
+```
+n\_threads is the number of rust executor threads to use. The default 0 means to use threads equal
+to the number of cores.
+
+cache\_capacity is the size of the LRU cache rust uses to cache connection objects. Here a connection
+means a unique combination of destination URL, credentials, and per-connection configuration such as
+timeouts; it does not mean an HTTP connection.
+
+cache\_ttl\_secs is the time-to-live in seconds for the rust connection cache.
+
+cache\_tti\_secs is the time in seconds that a connection can be idle before it is removed from the
+rust cache.
+
+multipart\_put\_threshold is the size in bytes for which any put request over this size will use a
+multipart upload. The put part size is determined by the rust object\_store implementation, which
+uses 10MB.
+
+multipart\_get\_threshold and multipart\_get\_part\_size configure automatic multipart gets. The part
+size can be greater than the threshold without breaking anything, but it may not make sense to do so.
+The default 8MB for these values was borrowed from CloudStore.jl.
+
 ## Design
 
 #### Packaging

src/RustyObjectStore.jl

@@ -45,21 +45,34 @@ $TYPEDFIELDS
     cache_ttl_secs::Culonglong
     "The time-to-idle in seconds for entries in the client cache"
     cache_tti_secs::Culonglong
+    "Put requests with a size in bytes greater than this will use multipart operations"
+    multipart_put_threshold::Culonglong
+    "Get requests with a size in bytes greater than this will use multipart operations"
+    multipart_get_threshold::Culonglong
+    "The size in bytes for each part of multipart get operations"
+    multipart_get_part_size::Culonglong
+
 end
 
 function Base.show(io::IO, config::StaticConfig)
     print(io, "StaticConfig("),
     print(io, "n_threads=", Int(config.n_threads), ",")
     print(io, "cache_capacity=", Int(config.cache_capacity), ",")
     print(io, "cache_ttl_secs=", Int(config.cache_ttl_secs), ",")
-    print(io, "cache_tti_secs=", Int(config.cache_tti_secs), ")")
+    print(io, "cache_tti_secs=", Int(config.cache_tti_secs), ",")
+    print(io, "multipart_put_threshold=", Int(config.multipart_put_threshold), ",")
+    print(io, "multipart_get_threshold=", Int(config.multipart_get_threshold), ",")
+    print(io, "multipart_get_part_size=", Int(config.multipart_get_part_size), ")")
 end
 
 const DEFAULT_CONFIG = StaticConfig(
     n_threads=0,
     cache_capacity=20,
     cache_ttl_secs=30 * 60,
-    cache_tti_secs=5 * 60
+    cache_tti_secs=5 * 60,
+    multipart_put_threshold=10 * 1024 * 1024,
+    multipart_get_threshold=8 * 1024 * 1024,
+    multipart_get_part_size=8 * 1024 * 1024
 )
 
 const _OBJECT_STORE_STARTED = Ref(false)

test/azure_blobs_exception_tests.jl

@@ -35,6 +35,26 @@
             end
         end
 
+        @testset "Insufficient output buffer size multipart" begin
+            input = "1,2,3,4,5,6,7,8,9,1\n" ^ 1_000_000
+            buffer = Vector{UInt8}(undef, 20_000_000)
+            @assert sizeof(input) == 20_000_000
+            @assert sizeof(buffer) == sizeof(input)
+
+            nbytes_written = put_object(codeunits(input), "test100B.csv", config)
+            @test nbytes_written == 20_000_000
+
+            try
+                # Buffer is over multipart threshold but too small for object
+                buffer = Vector{UInt8}(undef, 10_000_000)
+                nbytes_read = get_object!(buffer, "test100B.csv", config)
+                @test false # Should have thrown an error
+            catch err
+                @test err isa RustyObjectStore.GetException
+                @test err.msg == "failed to process get with error: Supplied buffer was too small"
+            end
+        end
+
         @testset "Malformed credentials" begin
             input = "1,2,3,4,5,6,7,8,9,1\n" ^ 5
             buffer = Vector{UInt8}(undef, 100)

test/basic_unified_tests.jl

@@ -67,7 +67,7 @@ function run_read_write_test_cases(read_config::AbstractConfig, write_config::Ab
         @test String(buffer[1:nbytes_read]) == input
     end
 
-    # Large files should eventually use multipart upload / download requests
+    # Large files should use multipart upload / download requests
     @testset "20MB file, 20MB buffer" begin
         input = "1,2,3,4,5,6,7,8,9,1\n" ^ 1_000_000
         buffer = Vector{UInt8}(undef, 20_000_000)
@@ -93,6 +93,20 @@ function run_read_write_test_cases(read_config::AbstractConfig, write_config::Ab
         @test nbytes_read == 20_000_000
         @test String(buffer[1:nbytes_read]) == input
     end
+
+    @testset "1MB file, 20MB buffer" begin
+        input = "1,2,3,4,5,6,7,8,9,1\n" ^ 50_000
+
+        nbytes_written = put_object(codeunits(input), "test100B.csv", write_config)
+        @test nbytes_written == 1_000_000
+
+        # Edge case for multpart download, file is less than threshold but buffer is greater
+        buffer = Vector{UInt8}(undef, 20_000_000)
+        nbytes_read = get_object!(buffer, "test100B.csv", read_config)
+        @test nbytes_read == 1_000_000
+        @test String(buffer[1:nbytes_read]) == input
+    end
+
 end
 end # @testsetup
 

test/common_testsetup.jl

@@ -1,4 +1,13 @@
 @testsetup module InitializeObjectStore
     using RustyObjectStore
-    init_object_store()
+    test_config = StaticConfig(
+        n_threads=0,
+        cache_capacity=20,
+        cache_ttl_secs=30 * 60,
+        cache_tti_secs=5 * 60,
+        multipart_put_threshold=8 * 1024 * 1024,
+        multipart_get_threshold=8 * 1024 * 1024,
+        multipart_get_part_size=8 * 1024 * 1024
+    )
+    init_object_store(test_config)
 end