Method Init

phxqueue/store/syncctrl.cpp:82–114

}

comm::RetCode SyncCtrl::Init() {
    auto opt(impl_->store->GetStoreOption());

    impl_->buf_size = opt->nsub * opt->nqueue * sizeof (SyncCtrlItem_t);

    auto sync_path(opt->data_dir_path + "/sync");
    int fd = open(sync_path.c_str(), O_RDWR | O_CREAT, S_IRUSR | S_IWUSR);
    if (fd < 0) {
        QLErr("open err %s path %s", strerror(errno), sync_path.c_str());
        return comm::RetCode::RET_ERR_SYS;
    }

    if (ftruncate(fd, (off_t)impl_->buf_size) < 0) {
        QLErr("ftruncate err %s", strerror(errno));
        close(fd);
        return comm::RetCode::RET_ERR_SYS;
    }

    void *pa{mmap(nullptr, impl_->buf_size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0)};

    close(fd);

    if ((char *)pa == MAP_FAILED) {
        QLErr("mmap err %s", strerror(errno));
        return comm::RetCode::RET_ERR_SYS;
    }

    impl_->buf = (SyncCtrlItem_t*)pa;

    impl_->locks.reset(new mutex[opt->nsub * opt->nqueue]);

    return comm::RetCode::RET_OK;;
}

static inline void GetIdx(const int nsub, const int sub_id, const int queue_id, size_t &idx) {
    idx = queue_id * nsub + sub_id - 1;