Repository: milvus-io/milvus
File: internal/core/src/storage/ChunkCache.cpp (C++)

Code before:
    if (!columns_.find(ca, path)) {
        return;
    }
    auto column = ca->second;
    auto ok =
        madvise(reinterpret_cast<void*>(const_cast<char*>(column->Data())),
                column->ByteSize(),
                read_ahead_policy_);
    AssertInfo(ok == 0,
               fmt::format("failed to madvise to the data file {}, err: {}",
                           path.c_str(),
                           strerror(errno)));
}

std::shared_ptr<ColumnBase>
ChunkCache::Mmap(const std::filesystem::path& path,
                 const FieldDataPtr& field_data) {
    std::unique_lock lck(mutex_);
    auto dir = path.parent_path();
    std::filesystem::create_directories(dir);

    auto dim = field_data->get_dim();
    auto data_type = field_data->get_data_type();

    auto file = File::Open(path.string(), O_CREAT | O_TRUNC | O_RDWR);

    // write the field data to disk
    auto data_size = field_data->Size();
    // unused
    std::vector<std::vector<uint64_t>> element_indices{};
    auto written = WriteFieldData(file, data_type, field_data, element_indices);
    AssertInfo(written == data_size,
               fmt::format("failed to write data file {}, written "
                           "{} but total {}, err: {}",
                           path.c_str(),
                           written,
                           data_size,
                           strerror(errno)));

    std::shared_ptr<ColumnBase> column{};

    if (datatype_is_variable(data_type)) {
        AssertInfo(false, "TODO: unimplemented for variable data type");
    } else {
        column = std::make_shared<Column>(file, data_size, dim, data_type);
    }

    // unlink
    auto ok = unlink(path.c_str());
    AssertInfo(ok == 0,
               fmt::format("failed to unlink mmap data file {}, err: {}",

Review comment (suggestion):
the two lines are not atomic, two threads are at line 56 and they can't find the path, then they both insert new mutex, one of the thread's `acc` will point to a dangling object

Code after:
}

void
ChunkCache::Prefetch(const std::string& filepath) {
    auto path = std::filesystem::path(path_prefix_) / filepath;
    ColumnTable::const_accessor ca;
    if (!columns_.find(ca, path)) {
        return;
    }
    auto column = ca->second;
    auto ok =
        madvise(reinterpret_cast<void*>(const_cast<char*>(column->Data())),
                column->ByteSize(),
                read_ahead_policy_);
    AssertInfo(ok == 0,
               fmt::format("failed to madvise to the data file {}, err: {}",
                           path.c_str(),
                           strerror(errno)));
}

std::shared_ptr<ColumnBase>
ChunkCache::Mmap(const std::filesystem::path& path,
                 const FieldDataPtr& field_data) {
    MmapFileLocks::accessor acc;
    if (!mmap_file_locks_.find(acc, path)) {
        mmap_file_locks_.insert(
            acc, std::make_pair(path, std::make_unique<std::mutex>()));
    }
    std::unique_lock lck(*acc->second.get());
    acc.release();

    auto dir = path.parent_path();
    std::filesystem::create_directories(dir);

    auto dim = field_data->get_dim();
    auto data_type = field_data->get_data_type();

    auto file = File::Open(path.string(), O_CREAT | O_TRUNC | O_RDWR);

    // write the field data to disk
    auto data_size = field_data->Size();
    // unused
    std::vector<std::vector<uint64_t>> element_indices{};
    auto written = WriteFieldData(file, data_type, field_data, element_indices);
    AssertInfo(written == data_size,
               fmt::format("failed to write data file {}, written "
                           "{} but total {}, err: {}",
                           path.c_str(),
                           written,
                           data_size,
                           strerror(errno)));