crash test and other refactoring (#13704)

Summary:
**Summary:**
AddressSanitizer failed complaining stack-use-after-return while executing AutoSkipCompressorWrapper::CompressBlock. This was caused because the AutoSkipCompressorWrapper was storing const reference pointer to Compression Options. It seems like the life time of the Compression Options can be shorter than the AutoSkipCompressorWrapper thus we need to copy the Compression Options and store it in AutoSkipCompressorWrapper.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13704

Test Plan:
Run the crashtest again to verify that we don’t encounter the issue again.
```bash
make clean
COMPILE_WITH_ASAN=1 make -j80 dbg
mkdir -p /dev/shm/rocksdb_test/rocksdb_crashtest_blackbox
mkdir -p dev/shm/rocksdb_test/rocksdb_crashtest_expected
./db_stress --WAL_size_limit_MB=0 --WAL_ttl_seconds=0 --acquire_snapshot_one_in=100 --adaptive_readahead=0 --adm_policy=2 --advise_random_on_open=1 --allow_data_in_errors=True --allow_fallocate=1 --allow_setting_blob_options_dynamically=1 --allow_unprepared_value=1 --async_io=0 --auto_readahead_size=1 --auto_refresh_iterator_with_snapshot=0 --avoid_flush_during_recovery=0 --avoid_flush_during_shutdown=1 --avoid_unnecessary_blocking_io=1 --backup_max_size=104857600 --backup_one_in=100000 --batch_protection_bytes_per_key=0 --bgerror_resume_retry_interval=1000000 --blob_cache_size=2097152 --blob_compaction_readahead_size=0 --blob_compression_type=snappy --blob_file_size=16777216 --blob_file_starting_level=2 --blob_garbage_collection_age_cutoff=0.5 --blob_garbage_collection_force_threshold=0.5 --block_align=0 --block_protection_bytes_per_key=2 --block_size=16384 --bloom_before_level=2147483646 --bloom_bits=9.703060295811829 --bottommost_compression_type=disable --bottommost_file_compaction_delay=0 --bytes_per_sync=262144 --cache_index_and_filter_blocks=1 --cache_index_and_filter_blocks_with_high_priority=0 --cache_size=33554432 --cache_type=fixed_hyper_clock_cache --charge_compression_dictionary_building_buffer=0 --charge_file_metadata=1 --charge_filter_construction=1 --charge_table_reader=0 --check_multiget_consistency=0 --check_multiget_entity_consistency=0 --checkpoint_one_in=1000000 --checksum_type=kXXH3 --clear_column_family_one_in=0 --compact_files_one_in=1000 --compact_range_one_in=1000000 --compaction_pri=1 --compaction_readahead_size=1048576 --compaction_style=0 --compaction_ttl=0 --compress_format_version=1 --compressed_secondary_cache_size=16777216 --compression_checksum=0 --compression_manager=autoskip --compression_max_dict_buffer_bytes=0 --compression_max_dict_bytes=0 --compression_parallel_threads=1 --compression_type=zlib --compression_use_zstd_dict_trainer=1 --compression_zstd_max_train_bytes=0 --continuous_verification_interval=0 --daily_offpeak_time_utc= --data_block_index_type=1 --db=/dev/shm/rocksdb_test/rocksdb_crashtest_blackbox --db_write_buffer_size=1048576 --decouple_partitioned_filters=1 --default_temperature=kUnknown --default_write_temperature=kWarm --delete_obsolete_files_period_micros=30000000 --delpercent=4 --delrangepercent=1 --destroy_db_initially=0 --detect_filter_construct_corruption=0 --disable_file_deletions_one_in=10000 --disable_manual_compaction_one_in=1000000 --disable_wal=0 --dump_malloc_stats=0 --enable_blob_files=1 --enable_blob_garbage_collection=1 --enable_checksum_handoff=1 --enable_compaction_filter=0 --enable_custom_split_merge=0 --enable_do_not_compress_roles=0 --enable_index_compression=1 --enable_memtable_insert_with_hint_prefix_extractor=0 --enable_pipelined_write=0 --enable_remote_compaction=0 --enable_sst_partitioner_factory=0 --enable_thread_tracking=1 --enable_write_thread_adaptive_yield=0 --error_recovery_with_no_fault_injection=0 --exclude_wal_from_write_fault_injection=0 --expected_values_dir=/dev/shm/rocksdb_test/rocksdb_crashtest_expected --fifo_allow_compaction=1 --file_checksum_impl=xxh64 --file_temperature_age_thresholds= --fill_cache=1 --flush_one_in=1000 --format_version=2 --get_all_column_family_metadata_one_in=10000 --get_current_wal_file_one_in=0 --get_live_files_apis_one_in=1000000 --get_properties_of_all_tables_one_in=100000 --get_property_one_in=100000 --get_sorted_wal_files_one_in=0 --hard_pending_compaction_bytes_limit=274877906944 --high_pri_pool_ratio=0 --index_block_restart_interval=15 --index_shortening=2 --index_type=2 --ingest_external_file_one_in=1000 --ingest_wbwi_one_in=0 --initial_auto_readahead_size=0 --inplace_update_support=0 --iterpercent=10 --key_len_percent_dist=1,30,69 --key_may_exist_one_in=100000 --last_level_temperature=kHot --level_compaction_dynamic_level_bytes=0 --lock_wal_one_in=0 --log_file_time_to_roll=0 --log_readahead_size=16777216 --long_running_snapshots=1 --low_pri_pool_ratio=0 --lowest_used_cache_tier=2 --manifest_preallocation_size=0 --manual_wal_flush_one_in=0 --mark_for_compaction_one_file_in=10 --max_auto_readahead_size=524288 --max_background_compactions=20 --max_bytes_for_level_base=10485760 --max_key=100000 --max_key_len=3 --max_log_file_size=1048576 --max_manifest_file_size=1073741824 --max_sequential_skip_in_iterations=1 --max_total_wal_size=0 --max_write_batch_group_size_bytes=16 --max_write_buffer_number=10 --max_write_buffer_size_to_maintain=1048576 --memtable_avg_op_scan_flush_trigger=0 --memtable_insert_hint_per_batch=1 --memtable_max_range_deletions=0 --memtable_op_scan_flush_trigger=1000 --memtable_prefix_bloom_size_ratio=0.01 --memtable_protection_bytes_per_key=4 --memtable_whole_key_filtering=1 --memtablerep=skip_list --metadata_charge_policy=0 --metadata_read_fault_one_in=0 --metadata_write_fault_one_in=1000 --min_blob_size=16 --min_write_buffer_number_to_merge=2 --mmap_read=1 --mock_direct_io=False --nooverwritepercent=1 --num_bottom_pri_threads=1 --num_file_reads_for_auto_readahead=1 --open_files=100 --open_metadata_read_fault_one_in=0 --open_metadata_write_fault_one_in=0 --open_read_fault_one_in=0 --open_write_fault_one_in=16 --ops_per_thread=100000000 --optimize_filters_for_hits=0 --optimize_filters_for_memory=0 --optimize_multiget_for_io=1 --paranoid_file_checks=0 --paranoid_memory_checks=0 --partition_filters=1 --partition_pinning=3 --pause_background_one_in=1000000 --periodic_compaction_seconds=0 --prefix_size=7 --prefixpercent=5 --prepopulate_blob_cache=0 --prepopulate_block_cache=1 --preserve_internal_time_seconds=0 --progress_reports=0 --promote_l0_one_in=0 --read_amp_bytes_per_bit=0 --read_fault_one_in=0 --readahead_size=16384 --readpercent=45 --recycle_log_file_num=0 --reopen=0 --report_bg_io_stats=0 --reset_stats_one_in=10000 --sample_for_compression=0 --secondary_cache_fault_one_in=32 --secondary_cache_uri= --set_options_one_in=1000 --skip_stats_update_on_db_open=0 --snapshot_hold_ops=100000 --soft_pending_compaction_bytes_limit=1048576 --sqfc_name=bar --sqfc_version=2 --sst_file_manager_bytes_per_sec=0 --sst_file_manager_bytes_per_truncate=0 --stats_dump_period_sec=0 --stats_history_buffer_size=1048576 --strict_bytes_per_sync=0 --subcompactions=3 --sync=0 --sync_fault_injection=0 --table_cache_numshardbits=-1 --target_file_size_base=524288 --target_file_size_multiplier=2 --test_batches_snapshots=0 --test_ingest_standalone_range_deletion_one_in=10 --top_level_index_pinning=2 --uncache_aggressiveness=5100 --universal_max_read_amp=0 --universal_reduce_file_locking=1 --unpartitioned_pinning=2 --use_adaptive_mutex=1 --use_adaptive_mutex_lru=0 --use_attribute_group=1 --use_blob_cache=1 --use_delta_encoding=0 --use_direct_io_for_flush_and_compaction=0 --use_direct_reads=0 --use_full_merge_v1=0 --use_get_entity=0 --use_merge=0 --use_multi_cf_iterator=1 --use_multi_get_entity=0 --use_multiget=1 --use_put_entity_one_in=10 --use_shared_block_and_blob_cache=0 --use_sqfc_for_range_queries=1 --use_timed_put_one_in=0 --use_write_buffer_manager=1 --user_timestamp_size=0 --value_size_mult=32 --verification_only=0 --verify_checksum=1 --verify_checksum_one_in=1000 --verify_compression=0 --verify_db_one_in=100000 --verify_file_checksums_one_in=1000000 --verify_iterator_with_expected_state_one_in=5 --verify_sst_unique_id_in_manifest=1 --wal_bytes_per_sync=0 --wal_compression=zstd --write_buffer_size=1048576 --write_dbid_to_manifest=1 --write_fault_one_in=128 --write_identity_file=0 --writepercent=35
```

Reviewed By: hx235

Differential Revision: D76826904

Pulled By: shubhajeet

fbshipit-source-id: c4a1522d3fed37bdd3e711f4c99c16d7bd1d794f

CMakeLists.txt

@@ -874,13 +874,13 @@ set(SOURCES
         trace_replay/trace_record.cc
         trace_replay/trace_replay.cc
         util/async_file_reader.cc
+        util/auto_skip_compressor.cc
         util/cleanable.cc
         util/coding.cc
         util/compaction_job_stats_impl.cc
         util/comparator.cc
         util/compression.cc
         util/simple_mixed_compressor.cc
-        util/auto_skip_compressor.cc
         util/compression_context_cache.cc
         util/concurrent_task_limiter_impl.cc
         util/crc32c.cc
@@ -1447,7 +1447,6 @@ if(WITH_TESTS)
         table/table_test.cc
         table/block_fetcher_test.cc
         test_util/testutil_test.cc
-        util/compression_test.cc
         trace_replay/block_cache_tracer_test.cc
         trace_replay/io_tracer_test.cc
         tools/block_cache_analyzer/block_cache_trace_analyzer_test.cc
@@ -1459,6 +1458,7 @@ if(WITH_TESTS)
         util/autovector_test.cc
         util/bloom_test.cc
         util/coding_test.cc
+        util/compression_test.cc
         util/crc32c_test.cc
         util/defer_test.cc
         util/dynamic_bloom_test.cc

tools/db_crashtest.py

@@ -1011,7 +1011,7 @@ def finalize_and_sanitize(src_params):
         dest_params["compression_type"] = "zstd"
         dest_params["bottommost_compression_type"] = "zstd"
     elif dest_params.get("compression_manager") == "autoskip":
-        # disabling compression parallel threads if mixed manager is being used as the predictor is not thread safe
+        # disabling compression parallel threads if auto skip manager is being used as the predictor is not thread safe
         dest_params["compression_parallel_threads"] = 1
         # esuring the compression is being used
         if dest_params.get("compression_type") == "none":

util/auto_skip_compressor.cc

@@ -32,9 +32,10 @@ bool CompressionRejectionProbabilityPredictor::Record(
   } else {
     compressed_count_++;
   }
-  if (attempted_compression_count() >= window_size_) {
-    pred_rejection_prob_percentage_ = static_cast<int>(
-        rejected_count_ * 100 / (compressed_count_ + rejected_count_));
+  auto attempted = attempted_compression_count();
+  if (attempted >= window_size_) {
+    pred_rejection_prob_percentage_ =
+        static_cast<int>(rejected_count_ * 100 / attempted);
     compressed_count_ = 0;
     rejected_count_ = 0;
     assert(attempted_compression_count() == 0);
@@ -42,16 +43,11 @@ bool CompressionRejectionProbabilityPredictor::Record(
   return true;
 }
 AutoSkipCompressorWrapper::AutoSkipCompressorWrapper(
-    std::unique_ptr<Compressor> compressor, const CompressionOptions& opts,
-    const CompressionType type)
+    std::unique_ptr<Compressor> compressor, const CompressionOptions& opts)
     : CompressorWrapper::CompressorWrapper(std::move(compressor)),
-      opts_(opts),
-      type_(type),
+      kOpts(opts),
       predictor_(
-          std::make_shared<CompressionRejectionProbabilityPredictor>(10)) {
-  (void)type_;
-  (void)opts_;
-}
+          std::make_shared<CompressionRejectionProbabilityPredictor>(10)) {}
 
 const char* AutoSkipCompressorWrapper::Name() const {
   return "AutoSkipCompressorWrapper";
@@ -89,7 +85,7 @@ Status AutoSkipCompressorWrapper::CompressBlockAndRecord(
   Status status = wrapped_->CompressBlock(uncompressed_data, compressed_output,
                                           out_compression_type, wa);
   // determine if it was rejected or compressed
-  predictor_->Record(uncompressed_data, compressed_output, opts_);
+  predictor_->Record(uncompressed_data, compressed_output, kOpts);
   return status;
 }
 
@@ -105,7 +101,7 @@ std::unique_ptr<Compressor> AutoSkipCompressorManager::GetCompressorForSST(
   assert(GetSupportedCompressions().size() > 1);
   assert(preferred != kNoCompression);
   return std::make_unique<AutoSkipCompressorWrapper>(
-      wrapped_->GetCompressorForSST(context, opts, preferred), opts, preferred);
+      wrapped_->GetCompressorForSST(context, opts, preferred), opts);
 }
 
 std::shared_ptr<CompressionManagerWrapper> CreateAutoSkipCompressionManager(

util/auto_skip_compressor.h

@@ -37,8 +37,7 @@ class AutoSkipCompressorWrapper : public CompressorWrapper {
  public:
   const char* Name() const override;
   explicit AutoSkipCompressorWrapper(std::unique_ptr<Compressor> compressor,
-                                     const CompressionOptions& opts,
-                                     const CompressionType type);
+                                     const CompressionOptions& opts);
 
   Status CompressBlock(Slice uncompressed_data, std::string* compressed_output,
                        CompressionType* out_compression_type,
@@ -51,8 +50,7 @@ class AutoSkipCompressorWrapper : public CompressorWrapper {
                                 ManagedWorkingArea* wa);
   static constexpr int kExplorationPercentage = 10;
   static constexpr int kProbabilityCutOff = 50;
-  const CompressionOptions& opts_;
-  const CompressionType type_;
+  const CompressionOptions kOpts;
   std::shared_ptr<CompressionRejectionProbabilityPredictor> predictor_;
 };