S3Source and Outputer

CMakeLists.txt

@@ -24,11 +24,13 @@ set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED True)
 
 #set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g3")
+add_compile_definitions(TBB_USE_THREADING_TOOLS=1)
 
 find_package(ROOT REQUIRED COMPONENTS Core RIO Tree)
 find_package(TBB REQUIRED)
 find_package(zstd REQUIRED)
 find_package(lz4 REQUIRED)
+find_package(LibLZMA REQUIRED)
 
 set(THREADS_PREFER_PTHREAD_FLAG ON)
 find_package(Threads REQUIRED)
@@ -84,6 +86,7 @@ add_executable(threaded_io_test
   outputerFactoryGenerator.cc
   WaiterFactory.cc
   waiterFactoryGenerator.cc
+  WaitingTaskList.cc
   ScaleWaiter.cc
   EventSleepWaiter.cc
   EventUnevenSleepWaiter.cc
@@ -99,6 +102,7 @@ target_compile_definitions(threaded_io_test PUBLIC TBB_PREVIEW_TASK_GROUP_EXTENS
 
 target_link_libraries(threaded_io_test
                       PRIVATE LZ4::lz4
+                              LibLZMA::LibLZMA
                               ROOT::Core
                               ROOT::RIO
                               ROOT::Tree
@@ -188,3 +192,26 @@ if(ENABLE_HDF5)
   add_test(NAME TestProductsHDFEvent COMMAND bash -c "${CMAKE_CURRENT_BINARY_DIR}/threaded_io_test -s TestProductsSource -t 1 -n 10 -o HDFEventOutputer=test_prod_e.h5")
   #; ${CMAKE_CURRENT_BINARY_DIR}/threaded_io_test -s HDFSource=test_prodi_e.h5 -t 1 -n 10 -o TestProductsOutputer")
 endif()
+
+option(ENABLE_S3 "Build S3 Sources and Outputers" OFF) # default OFF
+if(ENABLE_S3)
+  if(NOT DEFINED LIBS3_DIR)
+    message(FATAL_ERROR "You must provide LIBS3_DIR variable")
+  endif()
+  find_package(CURL REQUIRED)
+  find_package(Protobuf REQUIRED)
+  include_directories(${Protobuf_INCLUDE_DIRS})
+  include_directories(${CMAKE_CURRENT_BINARY_DIR})
+  protobuf_generate_cpp(PROTO_SRCS PROTO_HDRS objectstripe.proto)
+  target_link_libraries(threaded_io_test PRIVATE ${Protobuf_LIBRARIES})
+  target_sources(threaded_io_test PRIVATE
+    S3Outputer.cc
+    S3Source.cc
+    S3Common.cc
+    ${PROTO_SRCS}
+    )
+  target_include_directories(threaded_io_test PRIVATE ${LIBS3_DIR}/include)
+  target_link_directories(threaded_io_test PRIVATE ${LIBS3_DIR}/lib)
+  target_link_libraries(threaded_io_test PRIVATE s3 curl)
+  # add_test(NAME S3OutputerEmptyTest COMMAND threaded_io_test EmptySource 1 1 0 10 S3Outputer)
+endif()

ConfigurationParameters.cc

@@ -21,6 +21,11 @@ namespace cce::tf {
     return std::stof(iValue);
   }
 
+  template<>
+    unsigned long ConfigurationParameters::convert<unsigned long>(std::string const& iValue) {
+    return std::stoul(iValue);
+  }
+
   template<>
   bool ConfigurationParameters::convert<bool>(std::string const& iValue) {
     return (iValue.empty()) or (

README.md

@@ -148,6 +148,17 @@ This is similar to SharedRootEventSource except this time each entry in the `Eve
 > threaded_io_test -s SharedRootBatchEventsSource=test.eroot -t 1 -n 10
 ```
 
+#### S3Source
+Reads individual data product _stripes_--compressed concatenated serialized data products--from an S3 server, along with the appropriate
+event-level metadata to index them. See s3io.md for further details of the data layout.
+
+- verbose (int): increase number to get more detail
+- prefix (string): the object prefix in the S3 bucket
+- conn (string): path to the S3 connection configuration file
+```
+> threaded_io_test -s S3Source=prefix=testproducts:conn=s3conn.ini -t 1 -n 10
+```
+
 ### Outputers
 
 #### DummyOutputer
@@ -311,6 +322,19 @@ or
 > threaded_io_test -s ReplicatedRootSource=test.root -t 1 -n 10 -o RootBatchEventsOutputer=test.root:batchSize=4
 ```
 
+#### S3Outputer
+Outputs individual data product _stripes_--compressed concatenated serialized data products--to an S3 server, along with the appropriate
+event-level metadata to index them. See s3io.md for further details of the data layout.
+
+- verbose (int): increase number to get more detail
+- prefix (string): the object prefix to use when storing data in the S3 bucket
+- productFlush (int): the minimum number of (possibly compressed) bytes to accumulate in the product stripe output buffer before flushing it to S3
+- eventFlush (int): the maximum number of events that can be contained in a single product stripe.
+- conn (string): path to the S3 connection configuration file
+```
+> threaded_io_test -s TestProductsSource -t 1 -n 10 -o S3Outputer=prefix=testproducts:conn=s3conn.ini
+```
+
 ### Waiters
 
 #### ScaleWaiter