RecordManager.h

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#pragma once
 
#include <list>
#include <memory>
#include <mutex>
 
#include <vrs/ForwardDefinitions.h>
#include <vrs/Record.h>
#include <vrs/VrsExport.h>
 
namespace vrs {
 
using std::list;
 
class DataLayout;
 
class VRS_API RecordManager {
  friend class Record;
 
 public:
  RecordManager();
  RecordManager(const RecordManager&) = delete;
  RecordManager(RecordManager&&) = delete;
  ~RecordManager();
 
  void operator=(const RecordManager&) = delete;
  void operator=(RecordManager&&) = delete;
 
  Record*
  createRecord(double timestamp, Record::Type type, uint32_t formatVersion, const DataSource& data);
 
  Record* createUncompressedRecord(
      double timestamp,
      Record::Type type,
      uint32_t formatVersion,
      const DataSource& data,
      std::unique_ptr<DirectWriteRecordData>&& directWriteData);
 
  Record* createRecord(
      double timestamp,
      Record::Type type,
      uint32_t formatVersion,
      const DataSource& data,
      std::unique_ptr<BorrowedData>&& borrowedData);
 
  uint32_t purgeOldRecords(double oldestTimestamp, bool recycleBuffers = true);
 
  void purgeCache();
 
  void collectOldRecords(double maxAge, list<Record*>& outCollectedRecords);
 
  CompressionPreset getCompression() const {
    return compression_;
  }
 
  void setCompression(CompressionPreset compression) {
    compression_ = compression;
  }
 
  void setMaxCacheSize(size_t max) {
    maxCacheSize_ = max;
  }
 
  size_t getCurrentCacheSize() const {
    return cache_.size();
  }
 
 private:
  Record* createRecord(
      double timestamp,
      Record::Type type,
      uint32_t formatVersion,
      const DataSource& data,
      std::unique_ptr<DirectWriteRecordData> directWriteData,
      std::unique_ptr<BorrowedData> borrowedData);
  static size_t getAcceptableOverCapacity(size_t capacity);
  void recycle(Record* record);
 
  std::mutex mutex_;
  list<Record*> activeRecords_; // Records ready to be written
  list<Record*> cache_; // Written/recycled records available for reuse
 
  CompressionPreset compression_;
  size_t maxCacheSize_;
  uint64_t creationOrder_;
};
 
} // namespace vrs