More work

Author: willdealtry

cpp/arcticdb/CMakeLists.txt

@@ -511,7 +511,7 @@ set(arcticdb_srcs
         version/symbol_list.cpp
         version/version_map_batch_methods.cpp
         storage/s3/ec2_utils.cpp
-        util/buffer_holder.cpp storage/async_storage.hpp)
+        util/buffer_holder.cpp storage/async_storage.hpp util/allocation_tracing.hpp util/allocation_tracing.cpp storage/s3/http_client.cpp storage/s3/http_client.hpp storage/s3/http_client_factory.hpp storage/s3/http_client_factory.cpp storage/s3/http_request.hpp storage/s3/http_request.cpp storage/s3/http_response.hpp storage/s3/http_response.cpp)
 
 add_library(arcticdb_core_object OBJECT ${arcticdb_srcs})
 
@@ -662,6 +662,7 @@ set (arcticdb_core_libraries
         ${Zstd_LIBRARY}
         ${ARCTICDB_MONGO_LIBS}
         spdlog::spdlog
+        cpptrace::cpptrace
         Folly::folly # Transitively includes: double-conversion, gflags, glog, libevent, libssl, libcrypto, libiberty, libsodium
         Azure::azure-identity
         Azure::azure-storage-blobs

cpp/arcticdb/async/async_store.hpp

@@ -332,6 +332,7 @@ folly::Future<std::vector<VariantKey>> batch_read_compressed(
 std::vector<folly::Future<pipelines::SegmentAndSlice>> batch_read_uncompressed(
         std::vector<pipelines::RangesAndKey>&& ranges_and_keys,
         std::shared_ptr<std::unordered_set<std::string>> columns_to_decode) override {
+        log::version().debug("Reading {} keys", ranges_and_keys.size());
     return folly::window(
         std::move(ranges_and_keys),
         [this, columns_to_decode](auto&& ranges_and_key) {
@@ -341,7 +342,7 @@ std::vector<folly::Future<pipelines::SegmentAndSlice>> batch_read_uncompressed(
                 library_,
                 storage::ReadKeyOpts{},
                 DecodeSliceTask{std::forward<decltype(ranges_and_key)>(ranges_and_key), columns_to_decode});
-        }, async::TaskScheduler::instance()->io_thread_count() * 2);
+        }, async::TaskScheduler::instance()->io_thread_count() * 4);
 }
 
 std::vector<folly::Future<bool>> batch_key_exists(

cpp/arcticdb/async/tasks.hpp

@@ -429,6 +429,7 @@ struct DecodeSliceTask : BaseTask {
 
     pipelines::SegmentAndSlice operator()(storage::KeySegmentPair&& key_segment_pair) {
         ARCTICDB_SAMPLE(DecodeSliceTask, 0)
+        //log::version().info("Decode into slice {}", key_segment_pair.variant_key());
         return decode_into_slice(std::move(key_segment_pair));
     }
 

cpp/arcticdb/entity/performance_tracing.hpp

@@ -13,7 +13,7 @@
 
 #include <memory>
 
-#define ARCTICDB_LOG_PERFORMANCE
+//#define ARCTICDB_LOG_PERFORMANCE
 
 #define ARCTICDB_RUNTIME_SAMPLE(name, flags) \
 static bool _scoped_timer_active_ = ConfigsMap::instance()->get_int("Logging.timings", 0) == 1 || ConfigsMap::instance()->get_int("Logging.ALL", 0) == 1; \

cpp/arcticdb/log/log.hpp

@@ -11,7 +11,6 @@
 
 #include <spdlog/spdlog.h>
 
-#define DEBUG_BUILD
 #ifdef DEBUG_BUILD
 #define ARCTICDB_DEBUG(logger, ...) logger.debug(__VA_ARGS__)
 #define ARCTICDB_TRACE(logger, ...) logger.trace(__VA_ARGS__)

cpp/arcticdb/processing/clause.cpp

@@ -106,6 +106,7 @@ std::vector<EntityId> FilterClause::process(std::vector<EntityId>&& entity_ids)
     }
     auto proc = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(*component_manager_, std::move(entity_ids));
     proc.set_expression_context(expression_context_);
+    //log::version().info("Doing filter {} for entity ids {}", expression_context_->root_node_name_, entity_ids);
     auto variant_data = proc.get(expression_context_->root_node_name_);
     std::vector<EntityId> output;
     util::variant_match(variant_data,
@@ -468,6 +469,7 @@ void ResampleClause<closed_boundary>::set_processing_config(const ProcessingConf
 template<ResampleBoundary closed_boundary>
 std::vector<std::vector<size_t>> ResampleClause<closed_boundary>::structure_for_processing(
         std::vector<RangesAndKey>& ranges_and_keys) {
+    //log::version().info("ResampleClause: structure for processing 1");
     if (ranges_and_keys.empty()) {
         return {};
     }
@@ -503,6 +505,7 @@ std::vector<std::vector<EntityId>> ResampleClause<closed_boundary>::structure_fo
     if (entity_ids.empty()) {
         return {};
     }
+    //log::version().info("ResampleClause: structure for processing 2");
     auto [segments, row_ranges, col_ranges] = component_manager_->get_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>>(entity_ids, false);
     std::vector<RangesAndEntity> ranges_and_entities;
     ranges_and_entities.reserve(entity_ids.size());
@@ -559,6 +562,7 @@ std::vector<EntityId> ResampleClause<closed_boundary>::process(std::vector<Entit
         return {};
     }
     auto proc = gather_entities<std::shared_ptr<SegmentInMemory>, std::shared_ptr<RowRange>, std::shared_ptr<ColRange>, EntityFetchCount>(*component_manager_, std::move(entity_ids));
+    //log::version().info("ResampleClause: processing entities {}", entity_ids);
     auto row_slices = split_by_row_slice(std::move(proc));
     // If the entity fetch counts for the entities in the first row slice are 2, the first bucket overlapping this row
     // slice is being computed by the call to process dealing with the row slices above these. Otherwise, this call

cpp/arcticdb/processing/component_manager.hpp

@@ -143,4 +143,18 @@ class ComponentManager {
     std::shared_mutex mtx_;
 };
 
-} // namespace arcticdb
+} // namespace arcticdb
+
+namespace fmt {
+template<>
+struct formatter<arcticdb::EntityId> {
+    template<typename ParseContext>
+    constexpr auto parse(ParseContext& ctx) { return ctx.begin(); }
+
+    template<typename FormatContext>
+    auto format(const arcticdb::EntityId& id, FormatContext& ctx) const {
+        return fmt::format_to(ctx.out(), "{}", static_cast<uint64_t>(id));
+    }
+};
+
+} //namespace fmt

cpp/arcticdb/storage/azure/azure_storage.cpp

@@ -199,13 +199,17 @@ KeySegmentPair do_read_impl(
                                 static_cast<int>(e.StatusCode),
                                 e.ReasonPhrase);
         }
-        throw KeyNotFoundException(variant_key,
-                                   fmt::format("Failed to read azure segment with key '{}' {} {}: {}",
-                                               variant_key,
-                                               blob_name,
-                                               static_cast<int>(e.StatusCode),
-                                               e.ReasonPhrase));
+        throw KeyNotFoundException(
+           variant_key,
+           fmt::format("Failed to read azure segment with key '{}' {} {}: {}",
+                       variant_key,
+                       blob_name,
+                       static_cast<int>(e.StatusCode),
+                       e.ReasonPhrase));
+    } catch(const std::exception& ex) {
+        throw KeyNotFoundException(variant_key);
     }
+    return KeySegmentPair{};
 }
 
 namespace fg = folly::gen;

cpp/arcticdb/storage/library_manager.cpp

@@ -250,6 +250,13 @@ std::vector<LibraryPath> LibraryManager::get_library_paths() const {
 }
 
 bool LibraryManager::has_library(const LibraryPath& path) const {
+    {
+        std::lock_guard<std::mutex> lock{open_libraries_mutex_};
+        if (auto cached = open_libraries_.get(path); cached) {
+           return true;
+        }
+    }
+
     return store_->key_exists_sync(RefKey{StreamId(path.to_delim_path()), entity::KeyType::LIBRARY_CONFIG});
 }
 

cpp/arcticdb/storage/library_manager.hpp

@@ -70,7 +70,7 @@ namespace arcticdb::storage {
 
         std::shared_ptr<Store> store_;
         LRUCache<LibraryPath, std::shared_ptr<Library>> open_libraries_;
-        std::mutex open_libraries_mutex_;  // for open_libraries_
+        mutable std::mutex open_libraries_mutex_;  // for open_libraries_
     };
 }
 

cpp/arcticdb/storage/s3/detail-inl.hpp

@@ -145,9 +145,10 @@ KeySegmentPair do_read_impl(
         KeyBucketizer&& bucketizer,
         ReadKeyOpts opts) {
     ARCTICDB_SAMPLE(S3StorageRead, 0)
+    //log::version().info("Pre-bucketize {}", variant_key);
     auto key_type_dir = key_type_folder(root_folder, variant_key_type(variant_key));
     auto s3_object_name = object_path(bucketizer.bucketize(key_type_dir, variant_key), variant_key);
-
+    //log::version().info("Looking for key {}", variant_key);
     auto get_object_result = s3_client.get_object(s3_object_name, bucket_name);
 
     if (get_object_result.is_success()) {
@@ -167,6 +168,7 @@ KeySegmentPair do_read_impl(
 
         throw KeyNotFoundException(variant_key);
     }
+    return KeySegmentPair{};
 }
 
 /*template <typename KeyBucketizer>
@@ -193,7 +195,7 @@ folly::Future<KeySegmentPair> do_async_read_impl(
     ReadKeyOpts) {
     auto key_type_dir = key_type_folder(root_folder, variant_key_type(variant_key));
     auto s3_object_name = object_path(bucketizer.bucketize(key_type_dir, variant_key), variant_key);
-    return s3_client.get_object_async(bucket_name, s3_object_name).thenValue([vk=std::move(variant_key)] (auto&& result) mutable -> KeySegmentPair {
+    return s3_client.get_object_async(s3_object_name, bucket_name).thenValue([vk=std::move(variant_key)] (auto&& result) mutable -> KeySegmentPair {
         if(result.is_success())
             return KeySegmentPair(std::move(vk), std::move(result.get_output()));
         else