realm/realm-core | Build 0 | src/realm/db.cpp | Coveralls

149,818,695✔

            return version != 0;

149,818,695✔

99

149,818,695✔

101

10,106,505✔

102

            version = 0;

10,106,505✔

103

            count_live = count_frozen = count_full = 0;

10,106,505✔

104

10,106,505✔

106

1,498,878✔

107

            version = v;

1,498,878✔

108

1,498,878✔

112

276,879✔

113

        for (auto i = entries; i < size; ++i)

9,136,764✔

114

            data()[i].deactivate();

8,859,885✔

115

        if (size > entries) {

276,879✔

116

            // Fence preventing downward motion of above writes

136,608✔

117

            std::atomic_signal_fence(std::memory_order_release);

276,879✔

118

            entries = size;

276,879✔

119

276,879✔

120

276,879✔

123

137,526✔

124

        newest = nil; // empty

137,526✔

125

        entries = 0;

137,526✔

126

        reserve(init_readers_size);

137,526✔

127

137,526✔

130

138,168✔

131

        // get space required for given number of entries beyond the initial count.

67,920✔

132

        // NB: this not the size of the VersionList, it is the size minus whatever was

67,920✔

133

        // the initial size.

67,920✔

134

        return sizeof(ReadCount) * (num_entries - init_readers_size);

138,168✔

135

138,168✔

138

2,719,182✔

139

        return entries;

2,719,182✔

140

2,719,182✔

143

6,220,500✔

144

        return data()[idx];

6,220,500✔

145

6,220,500✔

153

1,498,848✔

154

        auto i = allocating.load();

1,498,848✔

155

        if (i == newest.load()) {

1,498,848✔

156

            // if newest != allocating we are recovering from a crash and MUST complete the earlier allocation

685,890✔

157

            // but if not, find lowest free entry by linear search.

685,890✔

158

            uint32_t k = 0;

1,359,528✔

159

            while (k < entries && data()[k].is_active()) {

2,490,201✔

160

                ++k;

1,130,673✔

161

1,130,673✔

162

            if (k == entries)

1,359,528✔

163

                return nullptr;     // no free entries

18✔

164

            allocating.exchange(k); // barrier: prevent upward movement of instructions below

1,359,510✔

165

            i = k;

1,359,510✔

166

1,359,510✔

167

        auto& rc = data()[i];

1,498,839✔

168

        REALM_ASSERT(rc.count_frozen == 0);

1,498,830✔

169

        REALM_ASSERT(rc.count_live == 0);

1,498,830✔

170

        REALM_ASSERT(rc.count_full == 0);

1,498,830✔

171

        rc.current_top = top;

1,498,830✔

172

        rc.filesize = size;

1,498,830✔

173

        rc.activate(version);

1,498,830✔

174

        newest.store(i); // barrier: prevent downward movement of instructions above

1,498,830✔

175

        return &rc;

1,498,830✔

176

1,498,848✔

179

1,246,599✔

180

        return (uint32_t)(&rc - data());

1,246,599✔

181

1,246,599✔

184

1,246,602✔

185

        rc->current_top = rc->filesize = -1ULL; // easy to recognize in debugger

1,246,602✔

186

        rc->deactivate();

1,246,602✔

187

1,246,602✔

197

139,335✔

198

        newest = nil;

139,335✔

199

        allocating = 0;

139,335✔

200

        auto t_free = entries;

139,335✔

201

        entries = 0;

139,335✔

202

        reserve(t_free);

139,335✔

203

        return *try_allocate_entry(top, filesize, version);

139,335✔

204

139,335✔

207

1,359,510✔

208

        oldest_live_v = std::numeric_limits<uint64_t>::max();

1,359,510✔

209

        auto oldest_full_v = std::numeric_limits<uint64_t>::max();

1,359,510✔

210

        any_new_unreachables = false;

1,359,510✔

211

        // correct case where an earlier crash may have left the entry at 'allocating' partially initialized:

685,848✔

212

        const auto index_of_newest = newest.load();

1,359,510✔

213

        if (auto a = allocating.load(); a != index_of_newest) {

1,359,510✔

216

        // determine fully locked versions - after one of those all versions are considered live.

685,848✔

217

        for (auto* rc = data(); rc < data() + entries; ++rc) {

45,347,142✔

218

            if (!rc->is_active())

43,987,632✔

219

                continue;

40,626,162✔

220

            if (rc->count_full) {

3,361,470✔

224

3,361,470✔

225

        // collect reachable versions and determine oldest live reachable version

685,848✔

226

        // (oldest reachable version is the first entry in the top_refs map, so no need to find it explicitly)

685,848✔

227

        for (auto* rc = data(); rc < data() + entries; ++rc) {

45,348,246✔

228

            if (!rc->is_active())

43,988,736✔

229

                continue;

40,627,374✔

230

            if (rc->count_frozen || rc->count_live || rc->version >= oldest_full_v) {

3,361,362✔

231

                // entry is still reachable

1,081,692✔

232

                top_refs.emplace(rc->version, VersionInfo{to_ref(rc->current_top), to_ref(rc->filesize)});

2,115,309✔

233

2,115,309✔

234

            if (rc->count_live || rc->version >= oldest_full_v) {

3,361,362✔

235

                if (rc->version < oldest_live_v)

1,559,301✔

236

                    oldest_live_v = rc->version;

1,423,311✔

237

1,559,301✔

238

3,361,362✔

239

        // we must have found at least one reachable version

685,848✔

240

        REALM_ASSERT(top_refs.size());

1,359,510✔

241

        // free unreachable entries and determine if we want to trigger backdating

685,848✔

242

        uint64_t oldest_v = top_refs.begin()->first;

1,359,510✔

243

        for (auto* rc = data(); rc < data() + entries; ++rc) {

45,347,544✔

244

            if (!rc->is_active())

43,988,034✔

245

                continue;

40,626,813✔

246

            if (rc->count_frozen == 0 && rc->count_live == 0 && rc->version < oldest_full_v) {

3,361,221✔

247

                // entry is becoming unreachable.

630,270✔

248

                // if it is also younger than a reachable version, then set 'any_new_unreachables' to trigger

630,270✔

249

                // backdating

630,270✔

250

                if (rc->version > oldest_v) {

1,246,602✔

251

                    any_new_unreachables = true;

63,681✔

252

63,681✔

253

                REALM_ASSERT(index_of(*rc) != index_of_newest);

1,246,602✔

254

                free_entry(rc);

1,246,602✔

255

1,246,602✔

256

3,361,221✔

257

        REALM_ASSERT(oldest_v != std::numeric_limits<uint64_t>::max());

1,359,510✔

258

        REALM_ASSERT(oldest_live_v != std::numeric_limits<uint64_t>::max());

1,359,510✔

259

1,359,510✔

288

160,401,585✔

289

        return m_data;

160,401,585✔

290

160,401,585✔

298

constexpr auto TransactionDeleter = [](Transaction* t) {

2,777,520✔

299

    t->close();

2,777,520✔

300

    delete t;

2,777,520✔

301

};

2,777,520✔

305

2,779,458✔

306

    return TransactionRef(new Transaction(std::forward<Args>(args)...), TransactionDeleter);

2,779,458✔

307

2,779,458✔

446

    ~SharedInfo() noexcept {}

25,302✔

449

139,335✔

450

        // Create our first versioning entry:

68,793✔

451

        readers.init_versioning(top_ref, file_size, initial_version);

139,335✔

452

139,335✔

461

137,526✔

462

    durability = static_cast<uint16_t>(dura); // durability level is fixed from creation

137,526✔

463

    REALM_ASSERT(!util::int_cast_has_overflow<decltype(history_type)>(ht + 0));

137,526✔

464

    REALM_ASSERT(!util::int_cast_has_overflow<decltype(history_schema_version)>(hsv));

137,526✔

465

    history_type = ht;

137,526✔

466

    history_schema_version = static_cast<uint16_t>(hsv);

137,526✔

467

    InterprocessCondVar::init_shared_part(new_commit_available); // Throws

137,526✔

468

    InterprocessCondVar::init_shared_part(pick_next_writer);     // Throws

137,526✔

469

    next_ticket = 0;

137,526✔

470

67,806✔

471

// IMPORTANT: The offsets, types (, and meanings) of these members must

67,806✔

472

// never change, not even when the SharedInfo layout version is bumped. The

67,806✔

473

// eternal constancy of this part of the layout is what ensures that a

67,806✔

474

// joining session participant can reliably verify that the actual format is

67,806✔

475

// as expected.

67,806✔

476

#ifndef _WIN32

137,526✔

477

#pragma GCC diagnostic push

137,526✔

478

#pragma GCC diagnostic ignored "-Winvalid-offsetof"

137,526✔

479

#endif

137,526✔

480

    static_assert(offsetof(SharedInfo, init_complete) == 0 && ATOMIC_BOOL_LOCK_FREE == 2 &&

137,526✔

481

                      std::is_same<decltype(init_complete), std::atomic<uint8_t>>::value &&

137,526✔

482

                      offsetof(SharedInfo, shared_info_version) == 6 &&

137,526✔

483

                      std::is_same<decltype(shared_info_version), uint16_t>::value,

137,526✔

484

                  "Forbidden change in SharedInfo layout");

137,526✔

485

67,806✔

486

    // Try to catch some of the memory layout changes that requires bumping of

67,806✔

487

    // the SharedInfo file format version (shared_info_version).

67,806✔

488

    static_assert(

137,526✔

489

        offsetof(SharedInfo, size_of_mutex) == 1 && std::is_same<decltype(size_of_mutex), uint8_t>::value &&

137,526✔

490

            offsetof(SharedInfo, size_of_condvar) == 2 && std::is_same<decltype(size_of_condvar), uint8_t>::value &&

137,526✔

491

            offsetof(SharedInfo, commit_in_critical_phase) == 3 &&

137,526✔

492

            std::is_same<decltype(commit_in_critical_phase), std::atomic<uint8_t>>::value &&

137,526✔

493

            offsetof(SharedInfo, file_format_version) == 4 &&

137,526✔

494

            std::is_same<decltype(file_format_version), uint8_t>::value && offsetof(SharedInfo, history_type) == 5 &&

137,526✔

495

            std::is_same<decltype(history_type), int8_t>::value && offsetof(SharedInfo, durability) == 8 &&

137,526✔

496

            std::is_same<decltype(durability), uint16_t>::value && offsetof(SharedInfo, free_write_slots) == 10 &&

137,526✔

497

            std::is_same<decltype(free_write_slots), uint16_t>::value &&

137,526✔

498

            offsetof(SharedInfo, num_participants) == 12 &&

137,526✔

499

            std::is_same<decltype(num_participants), uint32_t>::value &&

137,526✔

500

            offsetof(SharedInfo, latest_version_number) == 16 &&

137,526✔

501

            std::is_same<decltype(latest_version_number), uint64_t>::value &&

137,526✔

502

            offsetof(SharedInfo, session_initiator_pid) == 24 &&

137,526✔

503

            std::is_same<decltype(session_initiator_pid), uint64_t>::value &&

137,526✔

504

            offsetof(SharedInfo, number_of_versions) == 32 &&

137,526✔

505

            std::is_same<decltype(number_of_versions), std::atomic<uint64_t>>::value &&

137,526✔

506

            offsetof(SharedInfo, sync_agent_present) == 40 &&

137,526✔

507

            std::is_same<decltype(sync_agent_present), uint8_t>::value &&

137,526✔

508

            offsetof(SharedInfo, daemon_started) == 41 && std::is_same<decltype(daemon_started), uint8_t>::value &&

137,526✔

509

            offsetof(SharedInfo, daemon_ready) == 42 && std::is_same<decltype(daemon_ready), uint8_t>::value &&

137,526✔

510

            offsetof(SharedInfo, filler_1) == 43 && std::is_same<decltype(filler_1), uint8_t>::value &&

137,526✔

511

            offsetof(SharedInfo, history_schema_version) == 44 &&

137,526✔

512

            std::is_same<decltype(history_schema_version), uint16_t>::value && offsetof(SharedInfo, filler_2) == 46 &&

137,526✔

513

            std::is_same<decltype(filler_2), uint16_t>::value && offsetof(SharedInfo, shared_writemutex) == 48 &&

137,526✔

514

            std::is_same<decltype(shared_writemutex), InterprocessMutex::SharedPart>::value,

137,526✔

515

        "Caught layout change requiring SharedInfo file format bumping");

137,526✔

516

    static_assert(std::atomic<uint64_t>::is_always_lock_free);

137,526✔

517

#ifndef _WIN32

137,526✔

518

#pragma GCC diagnostic pop

137,526✔

519

#endif

137,526✔

520

137,526✔

526

163,452✔

527

163,452✔

528

    virtual ~VersionManager() {}

163,449✔

532

1,359,528✔

533

        std::lock_guard lock(m_mutex);

1,359,528✔

534

        util::CheckedLockGuard info_lock(m_info_mutex);

1,359,528✔

535

        ensure_reader_mapping();

1,359,528✔

536

        m_info->readers.purge_versions(oldest_live_version, top_refs, any_new_unreachables);

1,359,528✔

537

1,359,528✔

540

1,359,519✔

541

        return get_version_id_of_latest_snapshot().version;

1,359,519✔

542

1,359,519✔

545

2,545,737✔

546

2,545,737✔

547

            // First check the local cache. This is an unlocked read, so it may

1,740,813✔

548

            // race with adding a new version. If this happens we'll either see

1,740,813✔

549

            // a stale value (acceptable for a racing write on one thread and

1,740,813✔

550

            // a read on another), or a new value which is guaranteed to not

1,740,813✔

551

            // be an active index in the local cache.

1,740,813✔

552

            util::CheckedLockGuard lock(m_local_readers_mutex);

2,545,737✔

553

            util::CheckedLockGuard info_lock(m_info_mutex);

2,545,737✔

554

            auto index = m_info->readers.newest.load();

2,545,737✔

555

            if (index < m_local_readers.size()) {

2,545,746✔

556

                auto& r = m_local_readers[index];

2,545,710✔

557

                if (r.is_active()) {

2,545,710✔

558

                    return {r.version, index};

2,542,005✔

559

2,542,005✔

560

3,732✔

561

3,732✔

562

1,899✔

563

        std::lock_guard lock(m_mutex);

3,732✔

564

        util::CheckedLockGuard info_lock(m_info_mutex);

3,732✔

565

        auto index = m_info->readers.newest.load();

3,732✔

566

        ensure_reader_mapping(index);

3,732✔

567

        return {m_info->readers.get(index).version, index};

3,732✔

568

3,732✔

571

10,024,539✔

572

10,024,539✔

573

            util::CheckedLockGuard lock(m_local_readers_mutex);

10,024,539✔

574

            REALM_ASSERT(read_lock.m_reader_idx < m_local_readers.size());

10,024,539✔

575

            auto& r = m_local_readers[read_lock.m_reader_idx];

10,024,539✔

576

            auto& f = field_for_type(r, read_lock.m_type);

10,024,539✔

577

            REALM_ASSERT(f > 0);

10,024,539✔

578

            if (--f > 0)

10,024,539✔

579

                return;

6,916,056✔

580

            if (r.count_live == 0 && r.count_full == 0 && r.count_frozen == 0)

3,108,483✔

581

                r.version = 0;

3,083,262✔

582

3,108,483✔

583

1,576,914✔

584

        std::lock_guard lock(m_mutex);

3,108,483✔

585

        util::CheckedLockGuard info_lock(m_info_mutex);

3,108,483✔

586

        // we should not need to call ensure_full_reader_mapping,

1,576,914✔

587

        // since releasing a read lock means it has been grabbed

1,576,914✔

588

        // earlier - and hence must reside in mapped memory:

1,576,914✔

589

        REALM_ASSERT(read_lock.m_reader_idx < m_local_max_entry);

3,108,483✔

590

        auto& r = m_info->readers.get(read_lock.m_reader_idx);

3,108,483✔

591

        REALM_ASSERT(read_lock.m_version == r.version);

3,108,483✔

592

        --field_for_type(r, read_lock.m_type);

3,108,483✔

593

3,108,483✔

597

10,024,524✔

598

        ReadLockInfo read_lock;

10,024,524✔

599

        if (try_grab_local_read_lock(read_lock, type, version_id))

10,024,524✔

600

            return read_lock;

6,916,440✔

601

1,576,569✔

602

3,108,084✔

603

            const bool pick_specific = version_id.version != VersionID().version;

3,108,084✔

604

            std::lock_guard lock(m_mutex);

3,108,084✔

605

            util::CheckedLockGuard info_lock(m_info_mutex);

3,108,084✔

606

            auto newest = m_info->readers.newest.load();

3,108,084✔

607

            REALM_ASSERT(newest != VersionList::nil);

3,108,084✔

608

            read_lock.m_reader_idx = pick_specific ? version_id.index : newest;

3,101,310✔

609

            ensure_reader_mapping((unsigned int)read_lock.m_reader_idx);

3,108,084✔

610

            bool picked_newest = read_lock.m_reader_idx == (unsigned)newest;

3,108,084✔

611

            auto& r = m_info->readers.get(read_lock.m_reader_idx);

3,108,084✔

612

            if (pick_specific && version_id.version != r.version)

3,108,084✔

613

                throw BadVersion(version_id.version);

72✔

614

            if (!picked_newest) {

3,108,012✔

615

                if (type == ReadLockInfo::Frozen && r.count_frozen == 0 && r.count_live == 0)

702✔

617

                if (type != ReadLockInfo::Frozen && r.count_live == 0)

702✔

618

                    throw BadVersion(version_id.version);

60✔

619

3,107,952✔

620

            populate_read_lock(read_lock, r, type);

3,107,952✔

621

3,107,952✔

622

1,576,503✔

623

3,107,952✔

624

            util::CheckedLockGuard local_lock(m_local_readers_mutex);

3,107,952✔

625

            grow_local_cache(read_lock.m_reader_idx + 1);

3,107,952✔

626

            auto& r2 = m_local_readers[read_lock.m_reader_idx];

3,107,952✔

627

            if (!r2.is_active()) {

3,107,952✔

628

                r2.version = read_lock.m_version;

3,083,211✔

629

                r2.filesize = read_lock.m_file_size;

3,083,211✔

630

                r2.current_top = read_lock.m_top_ref;

3,083,211✔

631

                r2.count_full = r2.count_live = r2.count_frozen = 0;

3,083,211✔

632

3,083,211✔

633

            REALM_ASSERT_EX(field_for_type(r2, type) == 0, type, r2.count_full, r2.count_live, r2.count_frozen);

3,107,952✔

634

            field_for_type(r2, type) = 1;

3,107,952✔

635

3,107,952✔

636

1,576,503✔

637

        return read_lock;

3,107,952✔

638

3,107,952✔

641

114,033✔

642

        std::lock_guard lock(m_mutex);

114,033✔

643

        util::CheckedLockGuard info_lock(m_info_mutex);

114,033✔

644

        m_info->init_versioning(top_ref, file_size, initial_version);

114,033✔

645

114,033✔

648

1,359,510✔

649

        std::lock_guard lock(m_mutex);

1,359,510✔

650

        util::CheckedLockGuard info_lock(m_info_mutex);

1,359,510✔

651

        ensure_reader_mapping();

1,359,510✔

652

        if (m_info->readers.try_allocate_entry(new_top_ref, new_file_size, new_version)) {

1,359,513✔

653

            return;

1,359,504✔

654

1,359,504✔

655

        // allocation failed, expand VersionList (and lockfile) and retry

9✔

656

        auto entries = m_info->readers.capacity();

2,147,483,656✔

657

        auto new_entries = entries + 32;

2,147,483,656✔

658

        expand_version_list(new_entries);

2,147,483,656✔

659

        m_local_max_entry = new_entries;

2,147,483,656✔

660

        m_info->readers.reserve(new_entries);

2,147,483,656✔

661

        auto success = m_info->readers.try_allocate_entry(new_top_ref, new_file_size, new_version);

2,147,483,656✔

662

        REALM_ASSERT_EX(success, new_entries, new_version);

2,147,483,656✔

663

2,147,483,656✔

668

3,108,558✔

669

        if (new_size > m_local_readers.size())

3,108,558✔

670

            m_local_readers.resize(new_size, VersionList::ReadCount{});

279,531✔

671

3,108,558✔

674

10,024,842✔

675

        ++field_for_type(r, type);

10,024,842✔

676

        read_lock.m_type = type;

10,024,842✔

677

        read_lock.m_version = r.version;

10,024,842✔

678

        read_lock.m_top_ref = static_cast<ref_type>(r.current_top);

10,024,842✔

679

        read_lock.m_file_size = static_cast<size_t>(r.filesize);

10,024,842✔

680

10,024,842✔

684

10,024,302✔

685

        const bool pick_specific = version_id.version != VersionID().version;

10,024,302✔

686

        auto index = version_id.index;

10,024,302✔

687

        if (!pick_specific) {

10,024,302✔

688

            util::CheckedLockGuard lock(m_info_mutex);

9,732,096✔

689

            index = m_info->readers.newest.load();

9,732,096✔

690

9,732,096✔

691

        util::CheckedLockGuard local_lock(m_local_readers_mutex);

10,024,302✔

692

        if (index >= m_local_readers.size())

10,024,302✔

693

            return false;

279,537✔

694

7,765,842✔

695

        auto& r = m_local_readers[index];

9,744,765✔

696

        if (!r.is_active())

9,744,765✔

697

            return false;

2,803,821✔

698

        if (pick_specific && r.version != version_id.version)

6,940,944✔

700

        if (field_for_type(r, type) == 0)

6,940,944✔

701

            return false;

25,296✔

702

6,326,388✔

703

        read_lock.m_reader_idx = index;

6,915,648✔

704

        populate_read_lock(read_lock, r, type);

6,915,648✔

705

        return true;

6,915,648✔

706

6,915,648✔

709

36,311,127✔

710

        switch (type) {

36,311,127✔

711

            case ReadLockInfo::Frozen:

172,965✔

712

                return r.count_frozen;

172,965✔

713

            case ReadLockInfo::Live:

36,138,015✔

714

                return r.count_live;

36,138,015✔

719

36,311,127✔

720

36,311,127✔

723

2,310✔

724

        util::CheckedLockGuard info_lock(m_info_mutex);

2,310✔

725

        m_info->writing_page_offset = page_offset + 1;

2,310✔

726

        m_info->write_counter++;

2,310✔

727

2,310✔

729

2,310✔

730

        util::CheckedLockGuard info_lock(m_info_mutex);

2,310✔

731

        m_info->write_counter++;

2,310✔

732

        m_info->writing_page_offset = 0;

2,310✔

733

2,310✔

738

90✔

739

        util::CheckedLockGuard info_lock(m_info_mutex);

90✔

740

        if (write_counter) {

90✔

741

            *write_counter = m_info->write_counter;

90✔

742

90✔

743

        uint64_t marked = m_info->writing_page_offset;

90✔

747

        return marked != 0;

90✔

748

90✔

769

138,150✔

770

        size_t size = 0, required_size = sizeof(SharedInfo);

138,150✔

771

        while (size < required_size) {

276,300✔

772

            // Map the file without the lock held. This could result in the

67,911✔

773

            // mapping being too small and having to remap if the file is grown

67,911✔

774

            // concurrently, but if this is the case we should always see a bigger

67,911✔

775

            // size the next time.

67,911✔

776

            auto new_size = static_cast<size_t>(m_file.get_size());

138,150✔

777

            REALM_ASSERT(new_size > size);

138,150✔

778

            size = new_size;

138,150✔

779

            m_reader_map.remap(m_file, File::access_ReadWrite, size, File::map_NoSync);

138,150✔

780

            m_info = m_reader_map.get_addr();

138,150✔

781

67,911✔

782

            std::lock_guard lock(m_mutex);

138,150✔

783

            m_local_max_entry = m_info->readers.capacity();

138,150✔

784

            required_size = sizeof(SharedInfo) + m_info->readers.compute_required_space(m_local_max_entry);

138,150✔

785

            REALM_ASSERT(required_size >= size);

138,150✔

786

138,150✔

787

138,150✔

790

18✔

791

        size_t new_info_size = sizeof(SharedInfo) + m_info->readers.compute_required_space(new_entries);

18✔

792

        m_file.prealloc(new_info_size);                                          // Throws

18✔

793

        m_reader_map.remap(m_file, util::File::access_ReadWrite, new_info_size); // Throws

18✔

794

        m_info = m_reader_map.get_addr();

18✔

795

18✔

799

5,497,653✔

800

        using _impl::SimulatedFailure;

5,497,653✔

801

        SimulatedFailure::trigger(SimulatedFailure::shared_group__grow_reader_mapping); // Throws

5,497,653✔

802

2,784,147✔

803

        if (required < m_local_max_entry)

5,497,653✔

804

            return;

2,942,091✔

805

1,290,300✔

806

        auto new_max_entry = m_info->readers.capacity();

2,555,562✔

807

        if (new_max_entry > m_local_max_entry) {

2,555,562✔

814

2,555,562✔

827

138,150✔

828

138,150✔

830

90✔

831

        uint64_t tmp_write_count;

90✔

832

        auto page_may_have_been_written = vm.observe_writer(nullptr, &tmp_write_count);

90✔

833

        if (tmp_write_count != last_seen_count) {

90✔

837

        if (page_may_have_been_written) {

90✔

841

        ++calls_since_last_writer_observed;

90✔

842

        constexpr size_t max_calls = 5; // an arbitrary handful, > 1

90✔

843

        return (calls_since_last_writer_observed >= max_calls);

90✔

844

90✔

846

2,310✔

847

        vm.mark_page_for_writing(pos);

2,310✔

848

2,310✔

850

2,310✔

851

        vm.clear_writing_marker();

2,310✔

852

2,310✔

864

25,302✔

865

        m_info = info;

25,302✔

866

        m_local_max_entry = m_info->readers.capacity();

25,302✔

867

25,302✔

874

    void ensure_reader_mapping(unsigned int) override {}

333,462✔

902

138,069✔

903

    // Exception safety: Since do_open() is called from constructors, if it

67,881✔

904

    // throws, it must leave the file closed.

67,881✔

905

    using util::format;

138,069✔

906

67,881✔

907

    REALM_ASSERT(!is_attached());

138,069✔

908

    REALM_ASSERT(path.size());

138,069✔

909

67,881✔

910

    m_db_path = path;

138,069✔

911

67,881✔

912

    set_logger(options.logger);

138,069✔

913

    if (m_replication) {

138,069✔

914

        m_replication->set_logger(m_logger.get());

113,238✔

915

113,238✔

916

    if (m_logger)

138,069✔

917

        m_logger->log(util::Logger::Level::detail, "Open file: %1", path);

123,744✔

918

    SlabAlloc& alloc = m_alloc;

138,069✔

919

    if (options.is_immutable) {

138,069✔

920

        SlabAlloc::Config cfg;

180✔

921

        cfg.read_only = true;

180✔

922

        cfg.no_create = true;

180✔

923

        cfg.encryption_key = options.encryption_key;

180✔

924

        auto top_ref = alloc.attach_file(path, cfg);

180✔

925

        SlabAlloc::DetachGuard dg(alloc);

180✔

926

        Group::read_only_version_check(alloc, top_ref, path);

180✔

927

        m_fake_read_lock_if_immutable = ReadLockInfo::make_fake(top_ref, m_alloc.get_baseline());

180✔

928

        dg.release();

180✔

929

        return;

180✔

930

180✔

931

    std::string lockfile_path = get_core_file(path, CoreFileType::Lock);

137,889✔

932

    std::string coordination_dir = get_core_file(path, CoreFileType::Management);

137,889✔

933

    std::string lockfile_prefix = coordination_dir + "/access_control";

137,889✔

934

    m_alloc.set_read_only(false);

137,889✔

935

67,791✔

936

    Replication::HistoryType openers_hist_type = Replication::hist_None;

137,889✔

937

    int openers_hist_schema_version = 0;

137,889✔

938

    if (Replication* repl = get_replication()) {

137,889✔

939

        openers_hist_type = repl->get_history_type();

113,238✔

940

        openers_hist_schema_version = repl->get_history_schema_version();

113,238✔

941

113,238✔

942

67,791✔

943

    int current_file_format_version;

137,889✔

944

    int target_file_format_version;

137,889✔

945

    int stored_hist_schema_version = -1; // Signals undetermined

137,889✔

946

67,791✔

947

    int retries_left = 10; // number of times to retry before throwing exceptions

137,889✔

948

    // in case there is something wrong with the .lock file... the retries allows

67,791✔

949

    // us to pick a new lockfile initializer in case the first one crashes without

67,791✔

950

    // completing the initialization

67,791✔

951

    std::default_random_engine random_gen;

137,889✔

952

    for (;;) {

221,619✔

953

105,948✔

954

        // if we're retrying, we first wait a random time

105,948✔

955

        if (retries_left < 10) {

221,619✔

956

            if (retries_left == 9) { // we seed it from a true random source if possible

240✔

957

                std::random_device r;

24✔

958

                random_gen.seed(r());

24✔

959

24✔

960

            int max_delay = (10 - retries_left) * 10;

240✔

961

            int msecs = random_gen() % max_delay;

240✔

962

            millisleep(msecs);

240✔

963

240✔

964

105,948✔

965

        m_file.open(lockfile_path, File::access_ReadWrite, File::create_Auto, 0); // Throws

221,619✔

966

        File::CloseGuard fcg(m_file);

221,619✔

967

        m_file.set_fifo_path(coordination_dir, "lock.fifo");

221,619✔

968

105,948✔

969

        if (m_file.try_rw_lock_exclusive()) { // Throws

221,619✔

970

            File::UnlockGuard ulg(m_file);

112,224✔

971

55,176✔

972

            // We're alone in the world, and it is Ok to initialize the

55,176✔

973

            // file. Start by truncating the file to zero to ensure that

55,176✔

974

            // the following resize will generate a file filled with zeroes.

55,176✔

975

//

55,176✔

976

            // This will in particular set m_init_complete to 0.

55,176✔

977

            m_file.resize(0);

112,224✔

978

            m_file.prealloc(sizeof(SharedInfo));

112,224✔

979

55,176✔

980

            // We can crash anytime during this process. A crash prior to

55,176✔

981

            // the first resize could allow another thread which could not

55,176✔

982

            // get the exclusive lock because we hold it, and hence were

55,176✔

983

            // waiting for the shared lock instead, to observe and use an

55,176✔

984

            // old lock file.

55,176✔

985

            m_file_map.map(m_file, File::access_ReadWrite, sizeof(SharedInfo), File::map_NoSync); // Throws

112,224✔

986

            File::UnmapGuard fug(m_file_map);

112,224✔

987

            SharedInfo* info = m_file_map.get_addr();

112,224✔

988

55,176✔

989

            new (info) SharedInfo{options.durability, openers_hist_type, openers_hist_schema_version}; // Throws

112,224✔

990

55,176✔

991

            // Because init_complete is an std::atomic, it's guaranteed not to be observable by others

55,176✔

992

            // as being 1 before the entire SharedInfo header has been written.

55,176✔

993

            info->init_complete = 1;

112,224✔

994

112,224✔

995

105,948✔

996

// We hold the shared lock from here until we close the file!

105,948✔

997

#if REALM_PLATFORM_APPLE

115,671✔

1001

        while (!m_file.try_rw_lock_shared()) {

117,123✔

1002

            sched_yield();

1,452✔

1003

1,452✔

1005

        m_file.rw_lock_shared(); // Throws

105,948✔

1006

#endif

105,948✔

1007

        File::UnlockGuard ulg(m_file);

221,619✔

1008

105,948✔

1009

        // The coordination/management dir is created as a side effect of the lock

105,948✔

1010

        // operation above if needed for lock emulation. But it may also be needed

105,948✔

1011

        // for other purposes, so make sure it exists.

105,948✔

1012

        // in worst case there'll be a race on creating this directory.

105,948✔

1013

        // This should be safe but a waste of resources.

105,948✔

1014

        // Unfortunately it cannot be created at an earlier point, because

105,948✔

1015

        // it may then be deleted during the above lock_shared() operation.

105,948✔

1016

        try_make_dir(coordination_dir);

221,619✔

1017

105,948✔

1018

        // If the file is not completely initialized at this point in time, the

105,948✔

1019

        // preceeding initialization attempt must have failed. We know that an

105,948✔

1020

        // initialization process was in progress, because this thread (or

105,948✔

1021

        // process) failed to get an exclusive lock on the file. Because this

105,948✔

1022

        // thread (or process) currently has a shared lock on the file, we also

105,948✔

1023

        // know that the initialization process can no longer be in progress, so

105,948✔

1024

        // the initialization must either have completed or failed at this time.

105,948✔

1025

105,948✔

1026

        // The file is taken to be completely initialized if it is large enough

105,948✔

1027

        // to contain the `init_complete` field, and `init_complete` is true. If

105,948✔

1028

        // the file was not completely initialized, this thread must give up its

105,948✔

1029

        // shared lock, and retry to become the initializer. Eventually, one of

105,948✔

1030

        // two things must happen; either this thread, or another thread

105,948✔

1031

        // succeeds in completing the initialization, or this thread becomes the

105,948✔

1032

        // initializer, and fails the initialization. In either case, the retry

105,948✔

1033

        // loop will eventually terminate.

105,948✔

1034

105,948✔

1035

        // An empty file is (and was) never a successfully initialized file.

105,948✔

1036

        size_t info_size = sizeof(SharedInfo);

221,619✔

1037

221,619✔

1038

            auto file_size = m_file.get_size();

221,619✔

1039

            if (util::int_less_than(file_size, info_size)) {

221,619✔

1040

                if (file_size == 0)

83,265✔

1041

                    continue; // Retry

59,025✔

1042

                info_size = size_t(file_size);

24,240✔

1043

24,240✔

1044

221,619✔

1045

105,948✔

1046

        // Map the initial section of the SharedInfo file that corresponds to

105,948✔

1047

        // the SharedInfo struct, or less if the file is smaller. We know that

105,948✔

1048

        // we have at least one byte, and that is enough to read the

105,948✔

1049

        // `init_complete` flag.

105,948✔

1050

        m_file_map.map(m_file, File::access_ReadWrite, info_size, File::map_NoSync);

193,074✔

1051

        File::UnmapGuard fug_1(m_file_map);

162,594✔

1052

        SharedInfo* info = m_file_map.get_addr();

162,594✔

1053

75,468✔

1054

#ifndef _WIN32

162,594✔

1055

#pragma GCC diagnostic push

162,594✔

1056

#pragma GCC diagnostic ignored "-Winvalid-offsetof"

162,594✔

1057

#endif

162,594✔

1058

        static_assert(offsetof(SharedInfo, init_complete) + sizeof SharedInfo::init_complete <= 1,

162,594✔

1059

                      "Unexpected position or size of SharedInfo::init_complete");

162,594✔

1060

#ifndef _WIN32

162,594✔

1061

#pragma GCC diagnostic pop

162,594✔

1062

#endif

162,594✔

1063

        if (info->init_complete == 0)

162,594✔

1064

            continue;

24,174✔

1065

        REALM_ASSERT(info->init_complete == 1);

138,420✔

1066

68,049✔

1067

        // At this time, we know that the file was completely initialized, but

68,049✔

1068

        // we still need to verify that is was initialized with the memory

68,049✔

1069

        // layout expected by this session participant. We could find that it is

68,049✔

1070

        // initializaed with a different memory layout if other concurrent

68,049✔

1071

        // session participants use different versions of the core library.

68,049✔

1072

        if (info_size < sizeof(SharedInfo)) {

138,420✔

1073

            if (retries_left) {

66✔

1074

                --retries_left;

60✔

1075

                continue;

60✔

1076

60✔

1077

            throw IncompatibleLockFile(path, format("Architecture mismatch: SharedInfo size is %1 but should be %2.",

6✔

1078

                                                    info_size, sizeof(SharedInfo)));

6✔

1079

6✔

1080

        if (info->shared_info_version != g_shared_info_version) {

138,354✔

1081

            if (retries_left) {

66✔

1082

                --retries_left;

60✔

1083

                continue;

60✔

1084

60✔

1085

            throw IncompatibleLockFile(path, format("Version mismatch: SharedInfo version is %1 but should be %2.",

6✔

1086

                                                    info->shared_info_version, g_shared_info_version));

6✔

1087

6✔

1088

        // Validate compatible sizes of mutex and condvar types. Sizes of all

67,983✔

1089

        // other fields are architecture independent, so if condvar and mutex

67,983✔

1090

        // sizes match, the entire struct matches. The offsets of

67,983✔

1091

        // `size_of_mutex` and `size_of_condvar` are known to be as expected due

67,983✔

1092

        // to the preceeding check in `shared_info_version`.

67,983✔

1093

        if (info->size_of_mutex != sizeof info->shared_controlmutex) {

138,288✔

1094

            if (retries_left) {

66✔

1095

                --retries_left;

60✔

1096

                continue;

60✔

1097

60✔

1098

            throw IncompatibleLockFile(path, format("Architecture mismatch: Mutex size is %1 but should be %2.",

6✔

1099

                                                    info->size_of_mutex, sizeof(info->shared_controlmutex)));

6✔

1100

6✔

1101

67,950✔

1102

        if (info->size_of_condvar != sizeof info->room_to_write) {

138,222✔

1103

            if (retries_left) {

66✔

1104

                --retries_left;

60✔

1105

                continue;

60✔

1106

60✔

1107

            throw IncompatibleLockFile(

6✔

1108

                path, format("Architecture mismatch: Condition variable size is %1 but should be %2.",

6✔

1109

                             info->size_of_condvar, sizeof(info->room_to_write)));

6✔

1110

6✔

1111

        m_writemutex.set_shared_part(info->shared_writemutex, lockfile_prefix, "write");

138,156✔

1112

        m_controlmutex.set_shared_part(info->shared_controlmutex, lockfile_prefix, "control");

138,156✔

1113

        m_versionlist_mutex.set_shared_part(info->shared_versionlist_mutex, lockfile_prefix, "versions");

138,156✔

1114

67,917✔

1115

        // even though fields match wrt alignment and size, there may still be incompatibilities

67,917✔

1116

        // between implementations, so lets ask one of the mutexes if it thinks it'll work.

67,917✔

1117

        if (!m_controlmutex.is_valid()) {

138,156✔

1121

67,917✔

1122

        // OK! lock file appears valid. We can now continue operations under the protection

67,917✔

1123

        // of the controlmutex. The controlmutex protects the following activities:

67,917✔

1124

        // - attachment of the database file

67,917✔

1125

        // - start of the async daemon

67,917✔

1126

        // - stop of the async daemon

67,917✔

1127

        // - restore of a backup, if desired

67,917✔

1128

        // - backup of the realm file in preparation of file format upgrade

67,917✔

1129

        // - DB beginning/ending a session

67,917✔

1130

        // - Waiting for and signalling database changes

67,917✔

1131

138,156✔

1132

            std::lock_guard<InterprocessMutex> lock(m_controlmutex); // Throws

138,156✔

1133

            auto version_manager = std::make_unique<FileVersionManager>(m_file, m_versionlist_mutex);

138,156✔

1134

67,917✔

1135

            // proceed to initialize versioning and other metadata information related to

67,917✔

1136

            // the database. Also create the database if we're beginning a new session

67,917✔

1137

            bool begin_new_session = (info->num_participants == 0);

138,156✔

1138

            SlabAlloc::Config cfg;

138,156✔

1139

            cfg.session_initiator = begin_new_session;

138,156✔

1140

            cfg.is_shared = true;

138,156✔

1141

            cfg.read_only = false;

138,156✔

1142

            cfg.skip_validate = !begin_new_session;

138,156✔

1143

            cfg.disable_sync = options.durability == Durability::MemOnly || options.durability == Durability::Unsafe;

138,156✔

1144

67,917✔

1145

            // only the session initiator is allowed to create the database, all other

67,917✔

1146

            // must assume that it already exists.

67,917✔

1147

            cfg.no_create = (begin_new_session ? no_create_file : true);

125,916✔

1148

67,917✔

1149

            // if we're opening a MemOnly file that isn't already opened by

67,917✔

1150

            // someone else then it's a file which should have been deleted on

67,917✔

1151

            // close previously, but wasn't (perhaps due to the process crashing)

67,917✔

1152

            cfg.clear_file = (options.durability == Durability::MemOnly && begin_new_session);

138,156✔

1153

67,917✔

1154

            cfg.encryption_key = options.encryption_key;

138,156✔

1155

            ref_type top_ref;

138,156✔

1156

            m_marker_observer = std::make_unique<EncryptionMarkerObserver>(*version_manager);

138,156✔

1157

            try {

138,156✔

1158

                top_ref = alloc.attach_file(path, cfg, m_marker_observer.get()); // Throws

138,156✔

1159

138,156✔

1160

            catch (const SlabAlloc::Retry&) {

67,917✔

1165

67,872✔

1166

            // Determine target file format version for session (upgrade

67,872✔

1167

            // required if greater than file format version of attached file).

67,872✔

1168

            current_file_format_version = alloc.get_committed_file_format_version();

138,072✔

1169

            target_file_format_version =

138,072✔

1170

                Group::get_target_file_format_version_for_session(current_file_format_version, openers_hist_type);

138,072✔

1171

            BackupHandler backup(path, options.accepted_versions, options.to_be_deleted);

138,072✔

1172

            if (backup.must_restore_from_backup(current_file_format_version)) {

138,072✔

1173

                // we need to unmap before any file ops that'll change the realm

6✔

1174

                // file:

6✔

1175

                // (only strictly needed for Windows)

6✔

1176

                alloc.detach();

12✔

1177

                backup.restore_from_backup();

12✔

1178

                // finally, retry with the restored file instead of the original

6✔

1179

                // one:

6✔

1180

                continue;

12✔

1181

12✔

1182

            backup.cleanup_backups();

138,060✔

1183

67,866✔

1184

            // From here on, if we fail in any way, we must detach the

67,866✔

1185

            // allocator.

67,866✔

1186

            SlabAlloc::DetachGuard alloc_detach_guard(alloc);

138,060✔

1187

            alloc.note_reader_start(this);

138,060✔

1188

            // must come after the alloc detach guard

67,866✔

1189

            auto handler = [this, &alloc]() noexcept {

138,060✔

1190

                alloc.note_reader_end(this);

138,060✔

1191

};

138,060✔

1192

            auto reader_end_guard = make_scope_exit(handler);

138,060✔

1193

67,866✔

1194

            // Check validity of top array (to give more meaningful errors

67,866✔

1195

            // early)

67,866✔

1196

            if (top_ref) {

138,060✔

1197

                try {

87,654✔

1198

                    alloc.note_reader_start(this);

87,654✔

1199

                    auto reader_end_guard = make_scope_exit([&]() noexcept {

87,657✔

1200

                        alloc.note_reader_end(this);

87,657✔

1201

});

87,657✔

1202

                    Array top{alloc};

87,654✔

1203

                    top.init_from_ref(top_ref);

87,654✔

1204

                    Group::validate_top_array(top, alloc);

87,654✔

1205

87,654✔

1206

                catch (const InvalidDatabase& e) {

43,332✔

1212

87,654✔

1213

            if (options.backup_at_file_format_change) {

138,063✔

1214

                backup.backup_realm_if_needed(current_file_format_version, target_file_format_version);

138,051✔

1215

138,051✔

1216

            using gf = _impl::GroupFriend;

138,063✔

1217

            bool file_format_ok;

138,063✔

1218

            // In shared mode (Realm file opened via a DB instance) this

67,866✔

1219

            // version of the core library is able to open Realms using file format

67,866✔

1220

            // versions listed below. Please see Group::get_file_format_version() for

67,866✔

1221

            // information about the individual file format versions.

67,866✔

1222

            if (current_file_format_version == 0) {

138,063✔

1223

                file_format_ok = (top_ref == 0);

50,403✔

1224

50,403✔

1225

            else {

87,660✔

1226

                file_format_ok = backup.is_accepted_file_format(current_file_format_version);

87,660✔

1227

87,660✔

1228

67,866✔

1229

            if (REALM_UNLIKELY(!file_format_ok)) {

138,063✔

1230

                throw UnsupportedFileFormatVersion(current_file_format_version);

12✔

1231

12✔

1232

67,860✔

1233

            if (begin_new_session) {

138,051✔

1234

                // Determine version (snapshot number) and check history

56,235✔

1235

                // compatibility

56,235✔

1236

                version_type version = 0;

114,192✔

1237

                int stored_hist_type = 0;

114,192✔

1238

                gf::get_version_and_history_info(alloc, top_ref, version, stored_hist_type,

114,192✔

1239

                                                 stored_hist_schema_version);

114,192✔

1240

                bool good_history_type = false;

114,192✔

1241

                switch (openers_hist_type) {

114,192✔

1242

                    case Replication::hist_None:

7,332✔

1243

                        good_history_type = (stored_hist_type == Replication::hist_None);

7,332✔

1244

                        if (!good_history_type)

7,332✔

1245

                            throw IncompatibleHistories(

6✔

1246

                                util::format("Realm file at path '%1' has history type '%2', but is being opened "

6✔

1247

                                             "with replication disabled.",

6✔

1248

                                             path, Replication::history_type_name(stored_hist_type)),

6✔

1249

                                path);

6✔

1250

                        break;

7,326✔

1251

                    case Replication::hist_OutOfRealm:

3,813✔

1254

                    case Replication::hist_InRealm:

78,945✔

1255

                        good_history_type = (stored_hist_type == Replication::hist_InRealm ||

78,945✔

1256

                                             stored_hist_type == Replication::hist_None);

48,153✔

1257

                        if (!good_history_type)

78,945✔

1258

                            throw IncompatibleHistories(

6✔

1259

                                util::format("Realm file at path '%1' has history type '%2', but is being opened in "

6✔

1260

                                             "local history mode.",

6✔

1261

                                             path, Replication::history_type_name(stored_hist_type)),

6✔

1262

                                path);

6✔

1263

                        break;

78,939✔

1264

                    case Replication::hist_SyncClient:

52,260✔

1265

                        good_history_type = ((stored_hist_type == Replication::hist_SyncClient) || (top_ref == 0));

26,229✔

1266

                        if (!good_history_type)

26,229✔

1267

                            throw IncompatibleHistories(

6✔

1268

                                util::format("Realm file at path '%1' has history type '%2', but is being opened in "

6✔

1269

                                             "synchronized history mode.",

6✔

1270

                                             path, Replication::history_type_name(stored_hist_type)),

6✔

1271

                                path);

6✔

1272

                        break;

26,223✔

1273

                    case Replication::hist_SyncServer:

13,845✔

1274

                        good_history_type = ((stored_hist_type == Replication::hist_SyncServer) || (top_ref == 0));

1,686✔

1275

                        if (!good_history_type)

1,686✔

1281

                        break;

1,686✔

1282

114,174✔

1283

56,226✔

1284

                REALM_ASSERT(stored_hist_schema_version >= 0);

114,174✔

1285

                if (stored_hist_schema_version > openers_hist_schema_version)

114,174✔

1290

                bool need_hist_schema_upgrade =

114,174✔

1291

                    (stored_hist_schema_version < openers_hist_schema_version && top_ref != 0);

114,174✔

1292

                if (need_hist_schema_upgrade) {

114,174✔

1293

                    Replication* repl = get_replication();

186✔

1294

                    if (!repl->is_upgradable_history_schema(stored_hist_schema_version))

186✔

1299

114,174✔

1300

56,226✔

1301

                bool need_file_format_upgrade =

114,174✔

1302

                    current_file_format_version < target_file_format_version && top_ref != 0;

114,174✔

1303

                if (!options.allow_file_format_upgrade && (need_hist_schema_upgrade || need_file_format_upgrade)) {

114,174✔

1304

                    throw FileFormatUpgradeRequired(m_db_path);

6✔

1305

6✔

1306

56,223✔

1307

                alloc.convert_from_streaming_form(top_ref);

114,168✔

1308

                try {

114,168✔

1309

                    bool file_changed_size = alloc.align_filesize_for_mmap(top_ref, cfg);

114,168✔

1310

                    if (file_changed_size) {

114,168✔

1311

                        // we need to re-establish proper mappings after file size change.

132✔

1312

                        // we do this simply by aborting and starting all over:

132✔

1313

                        continue;

279✔

1314

279✔

1320

                if (options.encryption_key) {

113,889✔

1324

                    static_assert(sizeof(pid_t) <= sizeof(uint64_t), "process identifiers too large");

180✔

1325

                    uint64_t pid = getpid();

180✔

1326

#endif

180✔

1327

                    info->session_initiator_pid = pid;

180✔

1328

180✔

1329

56,091✔

1330

                info->file_format_version = uint_fast8_t(target_file_format_version);

113,889✔

1331

56,091✔

1332

                // Initially there is a single version in the file

56,091✔

1333

                info->number_of_versions = 1;

113,889✔

1334

56,091✔

1335

                info->latest_version_number = version;

113,889✔

1336

                alloc.init_mapping_management(version);

113,889✔

1337

56,091✔

1338

                size_t file_size = 24;

113,889✔

1339

                if (top_ref) {

113,889✔

1340

                    Array top(alloc);

64,968✔

1341

                    top.init_from_ref(top_ref);

64,968✔

1342

                    file_size = Group::get_logical_file_size(top);

64,968✔

1343

64,968✔

1344

                version_manager->init_versioning(top_ref, file_size, version);

113,889✔

1345

113,889✔

1346

            else { // Not the session initiator

23,859✔

1347

                // Durability setting must be consistent across a session. An

11,625✔

1348

                // inconsistency is a logic error, as the user is required to

11,625✔

1349

                // make sure that all possible concurrent session participants

11,625✔

1350

                // use the same durability setting for the same Realm file.

11,625✔

1351

                if (Durability(info->durability) != options.durability)

23,859✔

1352

                    throw RuntimeError(ErrorCodes::IncompatibleSession, "Durability not consistent");

6✔

1353

11,622✔

1354

                // History type must be consistent across a session. An

11,622✔

1355

                // inconsistency is a logic error, as the user is required to

11,622✔

1356

                // make sure that all possible concurrent session participants

11,622✔

1357

                // use the same history type for the same Realm file.

11,622✔

1358

                if (info->history_type != openers_hist_type)

23,853✔

1359

                    throw RuntimeError(ErrorCodes::IncompatibleSession, "History type not consistent");

6✔

1360

11,619✔

1361

                // History schema version must be consistent across a

11,619✔

1362

                // session. An inconsistency is a logic error, as the user is

11,619✔

1363

                // required to make sure that all possible concurrent session

11,619✔

1364

                // participants use the same history schema version for the same

11,619✔

1365

                // Realm file.

11,619✔

1366

                if (info->history_schema_version != openers_hist_schema_version)

23,847✔

1368

11,619✔

1369

                // We need per session agreement among all participants on the

11,619✔

1370

                // target Realm file format. From a technical perspective, the

11,619✔

1371

                // best way to ensure that, would be to require a bumping of the

11,619✔

1372

                // SharedInfo file format version on any change that could lead

11,619✔

1373

                // to a different result from

11,619✔

1374

                // get_target_file_format_for_session() given the same current

11,619✔

1375

                // Realm file format version and the same history type, as that

11,619✔

1376

                // would prevent the outcome of the Realm opening process from

11,619✔

1377

                // depending on race conditions. However, for practical reasons,

11,619✔

1378

                // we shall instead simply check that there is agreement, and

11,619✔

1379

                // throw the same kind of exception, as would have been thrown

11,619✔

1380

                // with a bumped SharedInfo file format version, if there isn't.

11,619✔

1381

                if (info->file_format_version != target_file_format_version) {

23,847✔

1386

11,619✔

1387

                // Even though this session participant is not the session initiator,

11,619✔

1388

                // it may be the one that has to perform the history schema upgrade.

11,619✔

1389

                // See upgrade_file_format(). However we cannot get the actual value

11,619✔

1390

                // at this point as the allocator is not synchronized with the file.

11,619✔

1391

                // The value will be read in a ReadTransaction later.

11,619✔

1392

11,619✔

1393

                // We need to setup the allocators version information, as it is needed

11,619✔

1394

                // to correctly age and later reclaim memory mappings.

11,619✔

1395

                version_type version = info->latest_version_number;

23,847✔

1396

                alloc.init_mapping_management(version);

23,847✔

1397

23,847✔

1398

67,860✔

1399

            m_new_commit_available.set_shared_part(info->new_commit_available, lockfile_prefix, "new_commit",

137,886✔

1400

                                                   options.temp_dir);

137,736✔

1401

            m_pick_next_writer.set_shared_part(info->pick_next_writer, lockfile_prefix, "pick_writer",

137,736✔

1402

                                               options.temp_dir);

137,736✔

1403

67,710✔

1404

            // make our presence noted:

67,710✔

1405

            ++info->num_participants;

137,736✔

1406

            m_info = info;

137,736✔

1407

67,710✔

1408

            // Keep the mappings and file open:

67,710✔

1409

            m_version_manager = std::move(version_manager);

137,736✔

1410

            alloc_detach_guard.release();

137,736✔

1411

            fug_1.release(); // Do not unmap

137,736✔

1412

            fcg.release();   // Do not close

137,736✔

1413

137,736✔

1414

        ulg.release(); // Do not release shared lock

137,736✔

1415

        break;

137,736✔

1416

138,051✔

1417

67,791✔

1418

    // Upgrade file format and/or history schema

67,791✔

1419

    try {

137,817✔

1420

        if (stored_hist_schema_version == -1) {

137,736✔

1421

            // current_hist_schema_version has not been read. Read it now

11,619✔

1422

            stored_hist_schema_version = start_read()->get_history_schema_version();

23,844✔

1423

23,844✔

1424

        if (current_file_format_version == 0) {

137,736✔

1425

            // If the current file format is still undecided, no upgrade is

24,525✔

1426

            // necessary, but we still need to make the chosen file format

24,525✔

1427

            // visible to the rest of the core library by updating the value

24,525✔

1428

            // that will be subsequently returned by

24,525✔

1429

            // Group::get_file_format_version(). For this to work, all session

24,525✔

1430

            // participants must adopt the chosen target Realm file format when

24,525✔

1431

            // the stored file format version is zero regardless of the version

24,525✔

1432

            // of the core library used.

24,525✔

1433

            m_file_format_version = target_file_format_version;

50,388✔

1434

50,388✔

1435

        else {

87,348✔

1436

            m_file_format_version = current_file_format_version;

87,348✔

1437

            upgrade_file_format(options.allow_file_format_upgrade, target_file_format_version,

87,348✔

1438

                                stored_hist_schema_version, openers_hist_schema_version); // Throws

87,348✔

1439

87,348✔

1440

137,736✔

1441

    catch (...) {

67,713✔

1442

        close();

6✔

1443

        throw;

6✔

1444

6✔

1445

    m_alloc.set_read_only(true);

137,727✔

1446

137,727✔

1449

6✔

1450

    auto top_ref = m_alloc.attach_buffer(buffer.data(), buffer.size());

6✔

1451

    m_fake_read_lock_if_immutable = ReadLockInfo::make_fake(top_ref, buffer.size());

6✔

1452

    if (take_ownership)

6✔

1454

6✔

1457

113,232✔

1458

    // Exception safety: Since open() is called from constructors, if it throws,

55,461✔

1459

    // it must leave the file closed.

55,461✔

1460

55,461✔

1461

    REALM_ASSERT(!is_attached());

113,232✔

1462

55,461✔

1463

    repl.initialize(*this); // Throws

113,232✔

1464

55,461✔

1465

    set_replication(&repl);

113,232✔

1466

55,461✔

1467

    bool no_create = false;

113,232✔

1468

    open(file, no_create, options); // Throws

113,232✔

1469

113,232✔

1475

73,656✔

1476

149,034✔

1477

75,378✔

1480

679,998✔

1481

        std::ostringstream ostr;

5,993,133✔

1482

        auto id = std::this_thread::get_id();

5,993,133✔

1483

        ostr << "DB: " << m_hash << " Thread " << id << ": ";

5,993,133✔

1484

        Logger::do_log(*m_base_logger_ptr, level, ostr.str() + message);

5,993,133✔

1485

5,993,133✔

1486

5,313,135✔

1492

80,511✔

1493

    if (logger)

80,511✔

1494

        m_logger = std::make_shared<DBLogger>(logger, m_log_id);

156,516✔

1495

163,371✔

1496

75,375✔

1497

void DB::open(Replication& repl, const DBOptions options)

82,860✔

1498

12,630✔

1499

    REALM_ASSERT(!is_attached());

12,630✔

1500

    repl.initialize(*this); // Throws

25,302✔

1501

    set_replication(&repl);

25,302✔

1502

25,302✔

1503

    m_alloc.init_in_memory_buffer();

25,302✔

1504

12,630✔

1505

    set_logger(options.logger);

25,302✔

1506

    m_replication->set_logger(m_logger.get());

12,630✔

1507

    if (m_logger)

25,302✔

1508

        m_logger->log(util::Logger::Level::detail, "Open memory-only realm");

25,296✔

1509

25,302✔

1510

    auto hist_type = repl.get_history_type();

25,296✔

1511

    m_in_memory_info =

12,630✔

1512

        std::make_unique<SharedInfo>(DBOptions::Durability::MemOnly, hist_type, repl.get_history_schema_version());

25,302✔

1513

    SharedInfo* info = m_in_memory_info.get();

25,302✔

1514

    m_writemutex.set_shared_part(info->shared_writemutex, "", "write");

25,302✔

1515

    m_controlmutex.set_shared_part(info->shared_controlmutex, "", "control");

25,302✔

1516

    m_new_commit_available.set_shared_part(info->new_commit_available, "", "new_commit", options.temp_dir);

25,302✔

1517

    m_pick_next_writer.set_shared_part(info->pick_next_writer, "", "pick_writer", options.temp_dir);

25,302✔

1518

    m_versionlist_mutex.set_shared_part(info->shared_versionlist_mutex, "", "versions");

25,302✔

1519

25,302✔

1520

    auto target_file_format_version = uint_fast8_t(Group::get_target_file_format_version_for_session(0, hist_type));

25,302✔

1521

    info->file_format_version = target_file_format_version;

12,630✔

1522

    info->number_of_versions = 1;

25,302✔

1523

    info->latest_version_number = 1;

25,302✔

1524

    info->init_versioning(0, m_alloc.get_baseline(), 1);

25,302✔

1525

    ++info->num_participants;

25,302✔

1526

25,302✔

1527

    m_version_manager = std::make_unique<InMemoryVersionManager>(info, m_versionlist_mutex);

25,302✔

1528

12,630✔

1529

    m_file_format_version = target_file_format_version;

25,302✔

1530

12,630✔

1531

    m_info = info;

25,302✔

1532

    m_alloc.set_read_only(true);

12,630✔

1533

25,302✔

1534

12,672✔

1535

void DB::create_new_history(Replication& repl)

12,672✔

1536

18✔

1537

    Replication* old_repl = get_replication();

18✔

1538

    try {

36✔

1539

        repl.initialize(*this);

36✔

1540

        set_replication(&repl);

36✔

1541

36✔

1542

        auto tr = start_write();

36✔

1543

        tr->clear_history();

18✔

1544

        tr->replicate(tr.get(), repl);

36✔

1545

        tr->commit();

36✔

1546

36✔

1547

    catch (...) {

36✔

1548

        set_replication(old_repl);

18✔

1551

18✔

1553

void DB::create_new_history(std::unique_ptr<Replication> repl)

18✔

1554

18✔

1555

    create_new_history(*repl);

18✔

1556

    m_history = std::move(repl);

36✔

1557

36✔

1558

18✔

1559

// WARNING / FIXME: compact() should NOT be exposed publicly on Windows because it's not crash safe! It may

18✔

1587

75✔

1588

    REALM_ASSERT(!m_fake_read_lock_if_immutable);

75✔

1589

    std::string tmp_path = m_db_path + ".tmp_compaction_space";

150✔

1590

150✔

1591

    // To enter compact, the DB object must already have been attached to a file,

150✔

1592

    // since this happens in DB::create().

75✔

1593

75✔

1594

    // Verify that the lock file is still attached. There is no attempt to guard against

75✔

1595

    // a race between close() and compact().

75✔

1596

    if (is_attached() == false) {

75✔

1598

75✔

1599

    auto info = m_info;

75✔

1600

    Durability dura = Durability(info->durability);

75✔

1601

    const char* write_key = bool(output_encryption_key) ? *output_encryption_key : get_encryption_key();

150✔

1602

150✔

1603

        std::unique_lock<InterprocessMutex> lock(m_controlmutex); // Throws

144✔

1604

        auto t1 = std::chrono::steady_clock::now();

150✔

1605

150✔

1606

        // We must be the ONLY DB object attached if we're to do compaction

150✔

1607

        if (info->num_participants > 1)

75✔

1609

150✔

1610

        // Holding the controlmutex prevents any other DB from attaching to the file.

75✔

1611

75✔

1612

        // Using start_read here ensures that we have access to the latest entry

75✔

1613

        // in the VersionList. We need to have access to that later to update top_ref and file_size.

75✔

1614

        // This is also needed to attach the group (get the proper top pointer, etc)

75✔

1615

        TransactionRef tr = start_read();

75✔

1616

        auto file_size_before = tr->get_logical_file_size();

75✔

1617

150✔

1618

        // local lock blocking any transaction from starting (and stopping)

150✔

1619

        CheckedLockGuard local_lock(m_mutex);

75✔

1620

75✔

1621

        // We should be the only transaction active - otherwise back out

150✔

1622

        if (m_transaction_count != 1)

75✔

1623

            return false;

3✔

1624

147✔

1625

        // group::write() will throw if the file already exists.

75✔

1626

        // To prevent this, we have to remove the file (should it exist)

72✔

1627

        // before calling group::write().

72✔

1628

        File::try_remove(tmp_path);

72✔

1629

72✔

1630

        // Compact by writing a new file holding only live data, then renaming the new file

144✔

1631

        // so it becomes the database file, replacing the old one in the process.

72✔

1632

        try {

72✔

1633

            File file;

72✔

1634

            file.open(tmp_path, File::access_ReadWrite, File::create_Must, 0);

144✔

1635

            int incr = bump_version_number ? 1 : 0;

144✔

1636

            Group::DefaultTableWriter writer;

144✔

1637

            tr->write(file, write_key, info->latest_version_number + incr, writer); // Throws

138✔

1638

            // Data needs to be flushed to the disk before renaming.

144✔

1639

            bool disable_sync = get_disable_sync_to_disk();

144✔

1640

            if (!disable_sync && dura != Durability::Unsafe)

72✔

1641

                file.sync(); // Throws

72✔

1643

        catch (...) {

72✔

1644

            // If writing the compact version failed in any way, delete the partially written file to clean up disk

72✔

1651

        // if we've written a file with a bumped version number, we need to update the lock file to match.

72✔

1652

        if (bump_version_number) {

72✔

1653

            ++info->latest_version_number;

6✔

1654

78✔

1655

        // We need to release any shared mapping *before* releasing the control mutex.

78✔

1656

        // When someone attaches to the new database file, they *must* *not* see and

78✔

1657

        // reuse any existing memory mapping of the stale file.

72✔

1658

        tr->close_read_with_lock();

72✔

1659

        m_alloc.detach();

72✔

1660

144✔

1661

        util::File::move(tmp_path, m_db_path);

144✔

1662

72✔

1663

        SlabAlloc::Config cfg;

144✔

1664

        cfg.session_initiator = true;

72✔

1665

        cfg.is_shared = true;

144✔

1666

        cfg.read_only = false;

144✔

1667

        cfg.skip_validate = false;

144✔

1668

        cfg.no_create = true;

144✔

1669

        cfg.clear_file = false;

144✔

1670

        cfg.encryption_key = write_key;

144✔

1671

        ref_type top_ref;

144✔

1672

        top_ref = m_alloc.attach_file(m_db_path, cfg, m_marker_observer.get());

144✔

1673

        m_alloc.convert_from_streaming_form(top_ref);

144✔

1674

        m_alloc.init_mapping_management(info->latest_version_number);

144✔

1675

        info->number_of_versions = 1;

144✔

1676

        size_t logical_file_size = sizeof(SlabAlloc::Header);

144✔

1677

        if (top_ref) {

144✔

1678

            Array top(m_alloc);

141✔

1679

            top.init_from_ref(top_ref);

141✔

1680

            logical_file_size = Group::get_logical_file_size(top);

138✔

1681

138✔

1682

        m_version_manager->init_versioning(top_ref, logical_file_size, info->latest_version_number);

141✔

1683

        if (m_logger) {

141✔

1684

            auto t2 = std::chrono::steady_clock::now();

102✔

1685

            m_logger->log(util::Logger::Level::info, "DB compacted from: %1 to %2 in %3 us", file_size_before,

102✔

1686

                          logical_file_size, std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count());

60✔

1687

60✔

1688

102✔

1689

    return true;

102✔

1690

144✔

1691

72✔

1692

void DB::write_copy(StringData path, const char* output_encryption_key)

72✔

1693

45✔

1694

    auto tr = start_read();

45✔

1695

    if (auto hist = tr->get_history()) {

90✔

1696

        if (!hist->no_pending_local_changes(tr->get_version())) {

90✔

1697

            throw Exception(ErrorCodes::IllegalOperation,

48✔

1698

                            "All client changes must be integrated in server before writing copy");

48✔

1699

6✔

1700

45✔

1701

45✔

1702

    class NoClientFileIdWriter : public Group::DefaultTableWriter {

84✔

1703

    public:

42✔

1704

        NoClientFileIdWriter()

84✔

1705

            : Group::DefaultTableWriter(true)

84✔

1706

84✔

1707

84✔

1708

        HistoryInfo write_history(_impl::OutputStream& out) override

84✔

1709

84✔

1710

            auto hist = Group::DefaultTableWriter::write_history(out);

81✔

1711

            hist.sync_file_id = 0;

78✔

1712

            return hist;

78✔

1713

78✔

1714

    } writer;

81✔

1715

81✔

1716

    File file;

84✔

1717

    file.open(path, File::access_ReadWrite, File::create_Must, 0);

42✔

1718

    file.resize(0);

84✔

1719

84✔

1720

    auto t1 = std::chrono::steady_clock::now();

84✔

1721

    tr->write(file, output_encryption_key, m_info->latest_version_number, writer);

42✔

1722

    if (m_logger) {

84✔

1723

        auto t2 = std::chrono::steady_clock::now();

72✔

1724

        m_logger->log(util::Logger::Level::info, "DB written to '%1' in %2 us", path,

72✔

1725

                      std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count());

60✔

1726

60✔

1727

72✔

1728

30✔

1729

uint_fast64_t DB::get_number_of_versions()

42✔

1730

184,089✔

1731

    if (m_fake_read_lock_if_immutable)

184,089✔

1732

        return 1;

192,636✔

1733

    return m_info->number_of_versions;

376,719✔

1734

184,089✔

1735

192,630✔

1736

size_t DB::get_allocated_size() const

192,630✔

1737

3✔

1738

    return m_alloc.get_allocated_size();

3✔

1739

6✔

1740

3✔

1741

DB::~DB() noexcept

3✔

1742

80,514✔

1743

    close();

80,514✔

1744

163,383✔

1745

82,869✔

1746

void DB::release_all_read_locks() noexcept

82,869✔

1747

80,340✔

1748

    REALM_ASSERT(!m_fake_read_lock_if_immutable);

80,340✔

1749

    CheckedLockGuard local_lock(m_mutex); // mx on m_local_locks_held

163,029✔

1750

    for (auto& read_lock : m_local_locks_held) {

163,029✔

1751

        --m_transaction_count;

82,692✔

1752

        m_version_manager->release_read_lock(read_lock);

6✔

1753

6✔

1754

    m_local_locks_held.clear();

80,343✔

1755

    REALM_ASSERT(m_transaction_count == 0);

80,343✔

1756

163,029✔

1757

82,689✔

1758

// Note: close() and close_internal() may be called from the DB::~DB().

82,689✔

1762

80,643✔

1763

    // make helper thread(s) terminate

80,643✔

1764

    m_commit_helper.reset();

163,656✔

1765

80,643✔

1766

    if (m_fake_read_lock_if_immutable) {

163,656✔

1767

        if (!is_attached())

93✔

1768

            return;

83,013✔

1769

186✔

1770

            CheckedLockGuard local_lock(m_mutex);

93✔

1771

            if (!allow_open_read_transactions && m_transaction_count)

186✔

1772

                throw WrongTransactionState("Closing with open read transactions");

93✔

1773

186✔

1774

        if (m_alloc.is_attached())

93✔

1775

            m_alloc.detach();

186✔

1776

        m_fake_read_lock_if_immutable.reset();

186✔

1777

186✔

1778

    else {

80,643✔

1779

        close_internal(std::unique_lock<InterprocessMutex>(m_controlmutex, std::defer_lock),

80,643✔

1780

                       allow_open_read_transactions);

163,470✔

1781

163,470✔

1782

163,563✔

1783

82,920✔

1784

void DB::close_internal(std::unique_lock<InterprocessMutex> lock, bool allow_open_read_transactions)

83,013✔

1785

80,550✔

1786

    if (!is_attached())

80,550✔

1787

        return;

83,124✔

1788

163,266✔

1789

80,565✔

1790

        CheckedLockGuard local_lock(m_mutex);

80,346✔

1791

        if (m_write_transaction_open)

163,047✔

1792

            throw WrongTransactionState("Closing with open write transactions");

82,704✔

1793

        if (!allow_open_read_transactions && m_transaction_count)

163,044✔

1794

            throw WrongTransactionState("Closing with open read transactions");

6✔

1795

163,038✔

1796

    SharedInfo* info = m_info;

80,343✔

1797

163,035✔

1798

        if (!lock.owns_lock())

163,035✔

1799

            lock.lock();

163,035✔

1800

163,035✔

1801

        if (m_alloc.is_attached())

163,035✔

1802

            m_alloc.detach();

80,340✔

1803

163,035✔

1804

        if (m_is_sync_agent) {

163,035✔

1805

            REALM_ASSERT(info->sync_agent_present);

609✔

1806

            info->sync_agent_present = 0; // Set to false

83,304✔

1807

1,542✔

1808

        release_all_read_locks();

81,273✔

1809

        --info->num_participants;

81,273✔

1810

        bool end_of_session = info->num_participants == 0;

163,035✔

1811

        // std::cerr << "closing" << std::endl;

163,035✔

1812

        if (end_of_session) {

163,035✔

1813

68,712✔

1814

            // If the db file is just backing for a transient data structure,

151,407✔

1815

            // we can delete it when done.

68,712✔

1816

            if (Durability(info->durability) == Durability::MemOnly && !m_in_memory_info) {

68,712✔

1817

                try {

10,047✔

1818

                    util::File::remove(m_db_path.c_str());

80,505✔

1819

20,094✔

1820

                catch (...) {

20,094✔

1821

                } // ignored on purpose.

10,059✔

1822

10,059✔

1823

68,724✔

1824

        lock.unlock();

90,387✔

1825

150,798✔

1826

163,035✔

1827

        CheckedLockGuard local_lock(m_mutex);

163,035✔

1828

163,035✔

1829

        m_new_commit_available.close();

163,035✔

1830

        m_pick_next_writer.close();

80,340✔

1831

163,035✔

1832

        if (m_in_memory_info) {

163,035✔

1833

            m_in_memory_info.reset();

12,630✔

1834

95,325✔

1835

        else {

80,382✔

1836

            // On Windows it is important that we unmap before unlocking, else a SetEndOfFile() call from another

80,382✔

1837

            // thread may interleave which is not permitted on Windows. It is permitted on *nix.

137,733✔

1838

            m_file_map.unmap();

67,710✔

1839

            m_version_manager.reset();

67,710✔

1840

            m_file.rw_unlock();

137,733✔

1841

            // info->~SharedInfo(); // DO NOT Call destructor

137,733✔

1842

            m_file.close();

137,733✔

1843

67,710✔

1844

        m_info = nullptr;

150,363✔

1845

        if (m_logger)

150,363✔

1846

            m_logger->log(util::Logger::Level::detail, "DB closed");

156,246✔

1847

163,035✔

1848

155,595✔

1849

82,695✔

1850

bool DB::other_writers_waiting_for_lock() const

82,695✔

1851

33,174✔

1852

    SharedInfo* info = m_info;

33,174✔

1853

64,725✔

1854

    uint32_t next_ticket = info->next_ticket.load(std::memory_order_relaxed);

64,725✔

1855

    uint32_t next_served = info->next_served.load(std::memory_order_relaxed);

33,174✔

1856

    // When holding the write lock, next_ticket = next_served + 1, hence, if the diference between 'next_ticket' and

64,725✔

1857

    // 'next_served' is greater than 1, there is at least one thread waiting to acquire the write lock.

64,725✔

1858

    return next_ticket > next_served + 1;

33,174✔

1859

33,174✔

1860

31,551✔

1861

class DB::AsyncCommitHelper {

31,551✔

1865

61,599✔

1866

61,599✔

1867

    ~AsyncCommitHelper()

63,309✔

1868

124,908✔

1869

61,599✔

1870

            std::unique_lock lg(m_mutex);

124,908✔

1871

            if (!m_running) {

124,908✔

1872

                return;

77,148✔

1873

77,148✔

1874

            m_running = false;

110,895✔

1875

            m_cv_worker.notify_one();

110,895✔

1876

47,934✔

1877

        m_thread.join();

47,934✔

1878

47,934✔

1879

174✔

1880

    void begin_write(util::UniqueFunction<void()> fn)

174✔

1881

393✔

1882

        std::unique_lock lg(m_mutex);

393✔

1883

        start_thread();

1,614✔

1884

        m_pending_writes.emplace_back(std::move(fn));

1,614✔

1885

        m_cv_worker.notify_one();

1,614✔

1886

1,614✔

1887

1,221✔

1888

    void blocking_begin_write()

1,221✔

1889

126,033✔

1890

        std::unique_lock lg(m_mutex);

126,033✔

1891

256,356✔

1892

        // If we support unlocking InterprocessMutex from a different thread

256,356✔

1893

        // than it was locked on, we can sometimes just begin the write on

126,033✔

1894

        // the current thread. This requires that no one is currently waiting

126,033✔

1895

        // for the worker thread to acquire the write lock, as we'll deadlock

126,033✔

1896

        // if we try to async commit while the worker is waiting for the lock.

126,033✔

1897

        bool can_lock_on_caller =

126,033✔

1898

            !InterprocessMutex::is_thread_confined && (!m_owns_write_mutex && m_pending_writes.empty() &&

126,033✔

1899

                                                       m_write_lock_claim_ticket == m_write_lock_claim_fulfilled);

130,323✔

1900

256,356✔

1901

        // If we support cross-thread unlocking and m_running is false,

256,296✔

1902

        // can_lock_on_caller should always be true or we forgot to launch the thread

126,033✔

1903

        REALM_ASSERT(can_lock_on_caller || m_running || InterprocessMutex::is_thread_confined);

126,033✔

1904

126,033✔

1905

        // If possible, just begin the write on the current thread

256,356✔

1906

        if (can_lock_on_caller) {

126,033✔

1908

            lg.unlock();

130,323✔

1909

            m_db->do_begin_write();

130,263✔

1910

            lg.lock();

130,263✔

1911

            m_waiting_for_write_mutex = false;

130,263✔

1912

            m_has_write_mutex = true;

130,263✔

1913

            m_owns_write_mutex = false;

130,263✔

1914

            return;

130,263✔

1915

130,263✔

1916

256,296✔

1917

        // Otherwise we have to ask the worker thread to acquire it and wait

256,296✔

1918

        // for that

126,033✔

1919

        start_thread();

126,033✔

1920

        size_t ticket = ++m_write_lock_claim_ticket;

126,033✔

1921

        m_cv_worker.notify_one();

126,093✔

1922

        m_cv_callers.wait(lg, [this, ticket] {

253,668✔

1923

            return ticket == m_write_lock_claim_fulfilled;

253,668✔

1924

});

253,896✔

1925

126,321✔

1926

288✔

1927

    void end_write()

60✔

1928

27✔

1929

        std::unique_lock lg(m_mutex);

27✔

1930

        REALM_ASSERT(m_has_write_mutex);

54✔

1931

        REALM_ASSERT(m_owns_write_mutex || !InterprocessMutex::is_thread_confined);

54✔

1932

54✔

1933

        // If we acquired the write lock on the worker thread, also release it

54✔

1934

        // there even if our mutex supports unlocking cross-thread as it simplifies things.

27✔

1935

        if (m_owns_write_mutex) {

27✔

1936

            m_pending_mx_release = true;

27✔

1937

            m_cv_worker.notify_one();

54✔

1938

51✔

1939

        else {

24✔

1940

            m_db->do_end_write();

24✔

1941

            m_has_write_mutex = false;

3✔

1942

3✔

1943

30✔

1944

3✔

1945

    bool blocking_end_write()

27✔

1946

148,347✔

1947

        std::unique_lock lg(m_mutex);

148,347✔

1948

        if (!m_has_write_mutex) {

302,388✔

1949

            return false;

176,295✔

1950

176,295✔

1951

        REALM_ASSERT(m_owns_write_mutex || !InterprocessMutex::is_thread_confined);

149,667✔

1952

149,667✔

1953

        // If we acquired the write lock on the worker thread, also release it

256,560✔

1954

        // there even if our mutex supports unlocking cross-thread as it simplifies things.

126,093✔

1955

        if (m_owns_write_mutex) {

126,093✔

1956

            m_pending_mx_release = true;

126,093✔

1957

            m_cv_worker.notify_one();

256,560✔

1958

            m_cv_callers.wait(lg, [this] {

252,711✔

1959

                return !m_pending_mx_release;

252,711✔

1960

});

253,236✔

1961

127,143✔

1962

        else {

1,050✔

1963

            m_db->do_end_write();

525✔

1964

            m_has_write_mutex = false;

129,942✔

1965

129,942✔

1966

            // The worker thread may have ignored a request for the write mutex

129,942✔

1970

                m_cv_worker.notify_one();

129,942✔

1973

        return true;

126,093✔

1974

256,035✔

1975

130,467✔

1976

130,467✔

1978

306✔

1979

        REALM_ASSERT(fn);

306✔

1980

        std::unique_lock lg(m_mutex);

1,356✔

1981

        REALM_ASSERT(!m_pending_sync);

1,356✔

1982

        start_thread();

1,356✔

1983

        m_pending_sync = std::move(fn);

1,356✔

1984

        m_cv_worker.notify_one();

1,356✔

1985

1,356✔

1986

1,050✔

1987

private:

1,050✔

2006

126,732✔

2007

        if (m_running) {

126,732✔

2008

            return;

81,303✔

2009

81,303✔

2010

        m_running = true;

49,920✔

2011

        m_thread = std::thread([this]() {

49,920✔

2012

            main();

47,931✔

2013

});

47,931✔

2014

47,931✔

2015

171✔

2016

    bool has_pending_write_requests()

171✔

2017

246,408✔

2018

        return m_write_lock_claim_fulfilled < m_write_lock_claim_ticket || !m_pending_writes.empty();

246,408✔

2019

378,189✔

2020

};

131,781✔

2021

131,781✔

2023

47,760✔

2024

    std::unique_lock lg(m_mutex);

47,760✔

2025

    while (m_running) {

553,605✔

2026

#if 0 // Enable for testing purposes

171✔

2027

        std::this_thread::sleep_for(std::chrono::milliseconds(10));

5,592✔

2029

        if (m_has_write_mutex) {

505,674✔

2030

            if (auto cb = std::move(m_pending_sync)) {

259,266✔

2031

                // Only one of sync_to_disk(), end_write(), or blocking_end_write()

5,727✔

2032

                // should be called, so we should never have both a pending sync

3,750✔

2033

                // and pending release.

306✔

2034

                REALM_ASSERT(!m_pending_mx_release);

306✔

2035

                lg.unlock();

306✔

2036

                cb();

1,356✔

2037

                cb = nullptr; // Release things captured by the callback before reacquiring the lock

1,356✔

2038

                lg.lock();

1,356✔

2039

                m_pending_mx_release = true;

1,356✔

2040

1,356✔

2041

            if (m_pending_mx_release) {

260,316✔

2042

                REALM_ASSERT(!InterprocessMutex::is_thread_confined || m_owns_write_mutex);

127,476✔

2043

                m_db->do_end_write();

129,870✔

2045

                m_has_write_mutex = false;

128,025✔

2046

                m_owns_write_mutex = false;

128,025✔

2047

128,025✔

2048

                lg.unlock();

128,025✔

2049

                m_cv_callers.notify_all();

126,426✔

2050

                lg.lock();

128,025✔

2051

                continue;

128,025✔

2052

128,025✔

2053

248,007✔

2054

        else {

248,007✔

2055

            REALM_ASSERT(!m_pending_sync && !m_pending_mx_release);

248,385✔

2056

248,385✔

2057

            // Acquire the write lock if anyone has requested it, but only if

248,385✔

2058

            // another thread is not already waiting for it. If there's another

246,408✔

2059

            // thread requesting and they get it while we're waiting, we'll

246,408✔

2060

            // deadlock if they ask us to perform the sync.

246,408✔

2061

            if (!m_waiting_for_write_mutex && has_pending_write_requests()) {

246,408✔

2062

                lg.unlock();

126,426✔

2063

                m_db->do_begin_write();

128,403✔

2064

                lg.lock();

127,707✔

2065

127,707✔

2066

                REALM_ASSERT(!m_has_write_mutex);

127,707✔

2067

                m_has_write_mutex = true;

126,426✔

2068

                m_owns_write_mutex = true;

127,707✔

2069

127,707✔

2070

                // Synchronous transaction requests get priority over async

127,707✔

2071

                if (m_write_lock_claim_fulfilled < m_write_lock_claim_ticket) {

126,426✔

2072

                    ++m_write_lock_claim_fulfilled;

126,033✔

2073

                    m_cv_callers.notify_all();

127,314✔

2074

                    continue;

126,093✔

2075

126,093✔

2076

453✔

2077

                REALM_ASSERT(!m_pending_writes.empty());

453✔

2078

                auto callback = std::move(m_pending_writes.front());

393✔

2079

                m_pending_writes.pop_front();

1,614✔

2080

                lg.unlock();

1,614✔

2081

                callback();

1,614✔

2082

                // Release things captured by the callback before reacquiring the lock

1,614✔

2083

                callback = nullptr;

1,614✔

2084

                lg.lock();

393✔

2085

                continue;

1,614✔

2086

1,614✔

2087

247,629✔

2088

        m_cv_worker.wait(lg);

254,043✔

2089

254,799✔

2090

    if (m_has_write_mutex && m_owns_write_mutex) {

50,301✔

2091

        m_db->do_end_write();

2,541✔

2093

47,760✔

2095

171✔

2097

393✔

2098

    REALM_ASSERT(m_commit_helper);

393✔

2099

    m_commit_helper->begin_write(std::move(fn));

1,614✔

2100

1,614✔

2101

1,221✔

2102

void DB::async_end_write()

1,221✔

2103

27✔

2104

    REALM_ASSERT(m_commit_helper);

27✔

2105

    m_commit_helper->end_write();

54✔

2106

54✔

2107

27✔

2108

void DB::async_sync_to_disk(util::UniqueFunction<void()> fn)

27✔

2109

306✔

2110

    REALM_ASSERT(m_commit_helper);

306✔

2111

    m_commit_helper->sync_to_disk(std::move(fn));

1,356✔

2112

1,356✔

2113

1,050✔

2114

bool DB::has_changed(TransactionRef& tr)

1,050✔

2115

953,220✔

2116

    if (m_fake_read_lock_if_immutable)

953,220✔

2117

        return false; // immutable doesn't change

18,687✔

2118

    bool changed = tr->m_read_lock.m_version != get_version_of_latest_snapshot();

971,907✔

2119

    return changed;

953,220✔

2120

971,907✔

2121

18,687✔

2122

bool DB::wait_for_change(TransactionRef& tr)

18,687✔

2152

43,182✔

2153

    // In a multithreaded scenario multiple threads may initially see a need to

43,182✔

2154

    // upgrade (maybe_upgrade == true) even though one onw thread is supposed to

87,342✔

2155

    // perform the upgrade, but that is ok, because the condition is rechecked

43,182✔

2156

    // in a fully reliable way inside a transaction.

43,182✔

2157

43,182✔

2158

    // First a non-threadsafe but fast check

43,182✔

2159

    int current_file_format_version = m_file_format_version;

43,182✔

2160

    REALM_ASSERT(current_file_format_version <= target_file_format_version);

43,182✔

2161

    REALM_ASSERT(current_hist_schema_version <= target_hist_schema_version);

87,342✔

2162

    bool maybe_upgrade_file_format = (current_file_format_version < target_file_format_version);

87,342✔

2163

    bool maybe_upgrade_hist_schema = (current_hist_schema_version < target_hist_schema_version);

87,342✔

2164

    bool maybe_upgrade = maybe_upgrade_file_format || maybe_upgrade_hist_schema;

87,342✔

2165

    if (maybe_upgrade) {

87,342✔

2166

44,280✔

2167

#ifdef REALM_DEBUG

44,280✔

2168

// This sleep() only exists in order to increase the quality of the

120✔

2169

// TEST(Upgrade_Database_2_3_Writes_New_File_Format_new) unit test.

258✔

2170

// The unit test creates multiple threads that all call

120✔

2171

// upgrade_file_format() simultaneously. This sleep() then acts like

120✔

2172

// a simple thread barrier that makes sure the threads meet here, to

120✔

2173

// increase the likelyhood of detecting any potential race problems.

120✔

2174

// See the unit test for details.

120✔

2175

//

120✔

2176

// NOTE: This sleep has been disabled because no problems have been found with

120✔

2177

// this code in a long while, and it was dramatically slowing down a unit test

120✔

2178

// in realm-sync.

120✔

2179

120✔

2180

// millisleep(200);

120✔

2181

#endif

120✔

2182

120✔

2183

        // WriteTransaction wt(*this);

258✔

2184

        auto wt = start_write();

120✔

2185

        bool dirty = false;

120✔

2186

258✔

2187

        // We need to upgrade history first. We may need to access it during migration

258✔

2188

        // when processing the !OID columns

120✔

2189

        int current_hist_schema_version_2 = wt->get_history_schema_version();

120✔

2190

        // The history must either still be using its initial schema or have

120✔

2191

        // been upgraded already to the chosen target schema version via a

258✔

2192

        // concurrent DB object.

120✔

2193

        REALM_ASSERT(current_hist_schema_version_2 == current_hist_schema_version ||

120✔

2194

                     current_hist_schema_version_2 == target_hist_schema_version);

120✔

2196

        if (need_hist_schema_upgrade) {

258✔

2197

            if (!allow_file_format_upgrade)

144✔

2198

                throw FileFormatUpgradeRequired(this->m_db_path);

138✔

2199

102✔

2200

            Replication* repl = get_replication();

6✔

2201

            repl->upgrade_history_schema(current_hist_schema_version_2); // Throws

6✔

2202

            wt->set_history_schema_version(target_hist_schema_version);  // Throws

102✔

2203

            dirty = true;

102✔

2204

102✔

2205

216✔

2206

        // File format upgrade

216✔

2207

        int current_file_format_version_2 = m_alloc.get_committed_file_format_version();

120✔

2208

        // The file must either still be using its initial file_format or have

120✔

2209

        // been upgraded already to the chosen target file format via a

258✔

2210

        // concurrent DB object.

120✔

2211

        REALM_ASSERT(current_file_format_version_2 == current_file_format_version ||

120✔

2212

                     current_file_format_version_2 == target_file_format_version);

120✔

2214

        if (need_file_format_upgrade) {

258✔

2215

            if (!allow_file_format_upgrade)

255✔

2216

                throw FileFormatUpgradeRequired(this->m_db_path);

138✔

2217

            wt->upgrade_file_format(target_file_format_version); // Throws

195✔

2218

            // Note: The file format version stored in the Realm file will be

117✔

2219

            // updated to the new file format version as part of the following

195✔

2220

            // commit operation. This happens in GroupWriter::commit().

117✔

2221

            if (m_upgrade_callback)

117✔

2222

                m_upgrade_callback(current_file_format_version_2, target_file_format_version); // Throws

9✔

2223

            dirty = true;

195✔

2224

126✔

2225

        wt->set_file_format_version(target_file_format_version);

198✔

2226

        m_file_format_version = target_file_format_version;

198✔

2227

258✔

2228

        if (dirty)

258✔

2229

            wt->commit(); // Throws

117✔

2230

258✔

2231

43,317✔

2232

138✔

2233

void DB::release_read_lock(ReadLockInfo& read_lock) noexcept

44,160✔

2234

7,898,931✔

2235

    // ignore if opened with immutable file (then we have no lockfile)

7,898,931✔

2236

    if (m_fake_read_lock_if_immutable)

10,019,751✔

2237

        return;

183✔

2238

    CheckedLockGuard lock(m_mutex); // mx on m_local_locks_held

10,019,568✔

2239

    do_release_read_lock(read_lock);

7,898,931✔

2240

10,019,385✔

2241

2,120,637✔

2242

// this is called with m_mutex locked

2,120,637✔

2244

7,903,776✔

2245

    REALM_ASSERT(!m_fake_read_lock_if_immutable);

7,903,776✔

2246

    bool found_match = false;

10,024,530✔

2247

    // simple linear search and move-last-over if a match is found.

10,024,530✔

2248

    // common case should have only a modest number of transactions in play..

10,024,530✔

2249

    for (size_t j = 0; j < m_local_locks_held.size(); ++j) {

13,080,684✔

2250

        if (m_local_locks_held[j].m_version == read_lock.m_version) {

13,080,339✔

2251

            m_local_locks_held[j] = m_local_locks_held.back();

10,877,937✔

2252

            m_local_locks_held.pop_back();

10,877,937✔

2253

            found_match = true;

10,024,221✔

2254

            break;

10,024,221✔

2255

10,024,221✔

2256

15,201,129✔

2257

    if (!found_match) {

10,024,566✔

2258

        REALM_ASSERT(!is_attached());

2,974,509✔

2259

        // it's OK, someone called close() and all locks where released

2,120,757✔

2260

        return;

6✔

2261

3✔

2262

    --m_transaction_count;

7,903,776✔

2263

    m_version_manager->release_read_lock(read_lock);

7,903,776✔

2264

10,024,524✔

2265

2,120,751✔

2266

2,120,751✔

2268

7,884,435✔

2269

    CheckedLockGuard lock(m_mutex); // mx on m_local_locks_held

7,884,435✔

2270

    REALM_ASSERT_RELEASE(is_attached());

10,004,520✔

2271

    auto read_lock = m_version_manager->grab_read_lock(type, version_id);

10,004,520✔

2272

10,004,520✔

2273

    m_local_locks_held.emplace_back(read_lock);

10,004,520✔

2274

    ++m_transaction_count;

7,884,435✔

2275

    REALM_ASSERT(read_lock.m_file_size > read_lock.m_top_ref);

10,004,520✔

2276

    return read_lock;

10,004,520✔

2277

10,004,520✔

2278

2,120,085✔

2279

void DB::leak_read_lock(ReadLockInfo& read_lock) noexcept

2,120,085✔

2280

3✔

2281

    CheckedLockGuard lock(m_mutex); // mx on m_local_locks_held

3✔

2282

    // simple linear search and move-last-over if a match is found.

6✔

2283

    // common case should have only a modest number of transactions in play..

6✔

2284

    for (size_t j = 0; j < m_local_locks_held.size(); ++j) {

3✔

2285

        if (m_local_locks_held[j].m_version == read_lock.m_version) {

3✔

2286

            m_local_locks_held[j] = m_local_locks_held.back();

6✔

2287

            m_local_locks_held.pop_back();

6✔

2288

            --m_transaction_count;

6✔

2289

            return;

6✔

2290

6✔

2291

6✔

2292

6✔

2293

3✔

2294

bool DB::do_try_begin_write()

3✔

2295

42✔

2296

    // In the non-blocking case, we will only succeed if there is no contention for

42✔

2297

    // the write mutex. For this case we are trivially fair and can ignore the

84✔

2298

    // fairness machinery.

42✔

2299

    bool got_the_lock = m_writemutex.try_lock();

42✔

2300

    if (got_the_lock) {

42✔

2301

        finish_begin_write();

78✔

2302

78✔

2303

    return got_the_lock;

78✔

2304

78✔

2305

42✔

2306

void DB::do_begin_write()

42✔

2307

697,113✔

2308

    if (m_logger) {

697,113✔

2309

        m_logger->log(util::Logger::Level::trace, "acquire writemutex");

834,051✔

2310

834,051✔

2311

852,390✔

2312

    SharedInfo* info = m_info;

852,390✔

2313

697,113✔

2314

    // Get write lock - the write lock is held until do_end_write().

1,382,412✔

2315

//

697,113✔

2316

    // We use a ticketing scheme to ensure fairness wrt performing write transactions.

697,113✔

2317

    // (But cannot do that on Windows until we have interprocess condition variables there)

697,113✔

2318

    uint32_t my_ticket = info->next_ticket.fetch_add(1, std::memory_order_relaxed);

697,113✔

2319

    m_writemutex.lock(); // Throws

697,113✔

2320

1,382,412✔

2321

    // allow for comparison even after wrap around of ticket numbering:

1,382,412✔

2322

    int32_t diff = int32_t(my_ticket - info->next_served.load(std::memory_order_relaxed));

697,113✔

2323

    bool should_yield = diff > 0; // ticket is in the future

697,113✔

2324

    // a) the above comparison is only guaranteed to be correct, if the distance

1,382,412✔

2325

    //    between my_ticket and info->next_served is less than 2^30. This will

1,382,412✔

2326

    //    be the case since the distance will be bounded by the number of threads

697,113✔

2327

    //    and each thread cannot ever hold more than one ticket.

697,113✔

2328

    // b) we could use 64 bit counters instead, but it is unclear if all platforms

697,113✔

2329

    //    have support for interprocess atomics for 64 bit values.

697,113✔

2330

697,113✔

2331

    timespec time_limit; // only compute the time limit if we're going to use it:

697,113✔

2332

    if (should_yield) {

697,113✔

2333

        // This clock is not monotonic, so time can move backwards. This can lead

703,719✔

2334

        // to a wrong time limit, but the only effect of a wrong time limit is that

703,719✔

2335

        // we momentarily lose fairness, so we accept it.

18,420✔

2336

        timeval tv;

18,420✔

2337

        gettimeofday(&tv, nullptr);

18,420✔

2338

        time_limit.tv_sec = tv.tv_sec;

18,654✔

2339

        time_limit.tv_nsec = tv.tv_usec * 1000;

18,654✔

2340

        time_limit.tv_nsec += 500000000;        // 500 msec wait

18,654✔

2341

        if (time_limit.tv_nsec >= 1000000000) { // overflow

18,654✔

2342

            time_limit.tv_nsec -= 1000000000;

9,669✔

2343

            time_limit.tv_sec += 1;

9,669✔

2344

9,438✔

2345

18,423✔

2346

697,116✔

2347

    while (should_yield) {

803,451✔

2348

107,694✔

2349

        m_pick_next_writer.wait(m_writemutex, &time_limit);

793,245✔

2350

        timeval tv;

107,694✔

2351

        gettimeofday(&tv, nullptr);

107,946✔

2352

        if (time_limit.tv_sec < tv.tv_sec ||

107,946✔

2353

            (time_limit.tv_sec == tv.tv_sec && time_limit.tv_nsec < tv.tv_usec * 1000)) {

107,946✔

2354

            // Timeout!

1,842✔

2355

            break;

1,842✔

2356

1,590✔

2357

        diff = int32_t(my_ticket - info->next_served);

106,104✔

2358

        should_yield = diff > 0; // ticket is in the future, so yield to someone else

106,104✔

2359

106,356✔

2360

697,365✔

2361

    // we may get here because a) it's our turn, b) we timed out

697,365✔

2362

    // we don't distinguish, satisfied that event b) should be rare.

697,113✔

2363

    // In case b), we have to *make* it our turn. Failure to do so could leave us

697,113✔

2364

    // with 'next_served' permanently trailing 'next_ticket'.

697,113✔

2365

//

697,113✔

2366

    // In doing so, we may bypass other waiters, hence the condition for yielding

697,113✔

2367

    // should take this situation into account by comparing with '>' instead of '!='

697,113✔

2368

    info->next_served = my_ticket;

697,113✔

2369

    finish_begin_write();

697,113✔

2370

    if (m_logger) {

1,382,412✔

2371

        m_logger->log(util::Logger::Level::trace, "writemutex acquired");

834,051✔

2372

834,051✔

2373

852,390✔

2374

155,277✔

2375

void DB::finish_begin_write()

685,299✔

2376

697,176✔

2377

    if (m_info->commit_in_critical_phase) {

697,176✔

2378

        m_writemutex.unlock();

685,302✔

2379

        throw RuntimeError(ErrorCodes::BrokenInvariant, "Crash of other process detected, session restart required");

685,302✔

2381

697,176✔

2382

697,176✔

2383

697,176✔

2384

        CheckedLockGuard local_lock(m_mutex);

697,176✔

2385

        m_write_transaction_open = true;

1,382,478✔

2386

1,382,478✔

2387

    m_alloc.set_read_only(false);

1,382,478✔

2388

1,382,478✔

2389

685,302✔

2390

void DB::do_end_write() noexcept

685,302✔

2391

697,185✔

2392

    m_info->next_served.fetch_add(1, std::memory_order_relaxed);

697,185✔

2393

1,382,496✔

2394

    CheckedLockGuard local_lock(m_mutex);

1,382,496✔

2395

    REALM_ASSERT(m_write_transaction_open);

697,185✔

2396

    m_alloc.set_read_only(true);

1,382,496✔

2397

    m_write_transaction_open = false;

1,382,496✔

2398

    m_pick_next_writer.notify_all();

1,382,496✔

2399

    m_writemutex.unlock();

1,382,496✔

2400

    if (m_logger) {

1,382,496✔

2401

        m_logger->log(util::Logger::Level::trace, "writemutex released");

834,087✔

2402

834,087✔

2403

852,486✔

2404

155,301✔

2405

685,311✔

2407

685,857✔

2408

    version_type current_version;

685,857✔

2409

1,359,540✔

2410

        current_version = m_version_manager->get_newest_version();

1,359,540✔

2411

1,359,540✔

2412

    version_type new_version = current_version + 1;

1,359,540✔

2413

1,359,540✔

2414

    if (!transaction.m_objects_to_delete.empty()) {

1,359,540✔

2415

        for (auto it : transaction.m_objects_to_delete) {

630✔

2416

            transaction.get_table(it.table_key)->remove_object(it.obj_key);

674,313✔

2417

1,260✔

2418

        transaction.m_objects_to_delete.clear();

957✔

2419

957✔

2420

    if (Replication* repl = get_replication()) {

686,184✔

2421

        // If Replication::prepare_commit() fails, then the entire transaction

677,232✔

2422

        // fails. The application then has the option of terminating the

1,350,588✔

2423

        // transaction with a call to Transaction::Rollback(), which in turn

676,905✔

2424

        // must call Replication::abort_transact().

676,905✔

2425

        new_version = repl->prepare_commit(current_version);        // Throws

676,905✔

2426

        low_level_commit(new_version, transaction, commit_to_disk); // Throws

676,905✔

2427

        repl->finalize_commit();

1,341,861✔

2428

1,341,861✔

2429

    else {

673,908✔

2430

        low_level_commit(new_version, transaction); // Throws

673,908✔

2431

17,679✔

2432

    return new_version;

694,584✔

2433

694,584✔

2434

673,683✔

2435

VersionID DB::get_version_id_of_latest_snapshot()

673,683✔

2436

1,054,983✔

2437

    if (m_fake_read_lock_if_immutable)

1,054,983✔

2438

        return {m_fake_read_lock_if_immutable->m_version, 0};

131,271✔

2439

    return m_version_manager->get_version_id_of_latest_snapshot();

1,186,242✔

2440

1,054,983✔

2441

131,259✔

2442

131,259✔

2444

1,054,707✔

2445

    return get_version_id_of_latest_snapshot().version;

1,054,707✔

2446

1,185,705✔

2447

130,998✔

2448

130,998✔

2450

685,866✔

2451

    SharedInfo* info = m_info;

685,866✔

2452

1,359,543✔

2453

    // Version of oldest snapshot currently (or recently) bound in a transaction

1,359,543✔

2454

    // of the current session.

685,866✔

2455

    uint64_t oldest_version = 0, oldest_live_version = 0;

685,866✔

2456

    TopRefMap top_refs;

685,866✔

2457

    bool any_new_unreachables;

1,359,543✔

2458

1,359,543✔

2459

        CheckedLockGuard lock(m_mutex);

1,359,543✔

2460

        m_version_manager->cleanup_versions(oldest_live_version, top_refs, any_new_unreachables);

1,359,543✔

2461

        oldest_version = top_refs.begin()->first;

1,359,543✔

2462

        // Allow for trimming of the history. Some types of histories do not

1,359,543✔

2463

        // need store changesets prior to the oldest *live* bound snapshot.

1,359,543✔

2464

        if (auto hist = transaction.get_history()) {

685,866✔

2465

            hist->set_oldest_bound_version(oldest_live_version); // Throws

676,872✔

2466

1,350,549✔

2467

        // Cleanup any stale mappings

1,350,801✔

2468

        m_alloc.purge_old_mappings(oldest_version, new_version);

1,350,801✔

2469

685,866✔

2470

    // save number of live versions for later:

1,359,543✔

2471

    // (top_refs is std::moved into GroupWriter so we'll loose it in the call to set_versions below)

1,359,543✔

2472

    auto live_versions = top_refs.size();

685,866✔

2473

    // Do the actual commit

685,866✔

2474

    REALM_ASSERT(oldest_version <= new_version);

1,359,543✔

2475

685,866✔

2476

    GroupWriter out(transaction, Durability(info->durability), m_marker_observer.get()); // Throws

1,359,543✔

2477

    out.set_versions(new_version, top_refs, any_new_unreachables);

685,866✔

2478

    out.prepare_evacuation();

1,359,543✔

2479

    auto t1 = std::chrono::steady_clock::now();

1,359,543✔

2480

    auto commit_size = m_alloc.get_commit_size();

1,359,543✔

2481

    if (m_logger) {

1,359,543✔

2482

        m_logger->log(util::Logger::Level::debug, "Initiate commit version: %1", new_version);

811,986✔

2483

811,986✔

2484

    if (auto limit = out.get_evacuation_limit()) {

830,583✔

2485

        // Get a work limit based on the size of the transaction we're about to commit

147,402✔

2486

        // Add 4k to ensure progress on small commits

676,362✔

2487

        size_t work_limit = commit_size / 2 + out.get_free_list_size() + 0x1000;

2,685✔

2488

        transaction.cow_outliers(out.get_evacuation_progress(), limit, work_limit);

2,685✔

2489

5,343✔

2490

688,524✔

2491

    ref_type new_top_ref;

688,524✔

2492

    // Recursively write all changed arrays to end of file

685,866✔

2493

1,359,543✔

2494

        // protect against race with any other DB trying to attach to the file

685,866✔

2495

        std::lock_guard<InterprocessMutex> lock(m_controlmutex); // Throws

1,359,543✔

2496

        new_top_ref = out.write_group();                         // Throws

685,866✔

2497

1,359,543✔

2498

1,359,543✔

2499

        // protect access to shared variables and m_reader_mapping from here

1,359,543✔

2500

        CheckedLockGuard lock_guard(m_mutex);

1,359,543✔

2501

        m_free_space = out.get_free_space_size();

685,866✔

2502

        m_locked_space = out.get_locked_space_size();

1,359,543✔

2503

        m_used_space = out.get_logical_size() - m_free_space;

1,359,543✔

2504

        m_evac_stage.store(EvacStage(out.get_evacuation_stage()));

1,359,543✔

2505

        out.sync_according_to_durability();

1,359,543✔

2506

        if (Durability(info->durability) == Durability::Full || Durability(info->durability) == Durability::Unsafe) {

1,359,543✔

2507

            if (commit_to_disk) {

1,278,807✔

2508

                GroupCommitter cm(transaction, Durability(info->durability), m_marker_observer.get());

1,275,414✔

2509

                cm.commit(new_top_ref);

1,194,525✔

2510

1,190,388✔

2511

1,193,781✔

2512

        size_t new_file_size = out.get_logical_size();

1,274,517✔

2513

        // We must reset the allocators free space tracking before communicating the new

1,278,654✔

2514

        // version through the ring buffer. If not, a reader may start updating the allocators

1,359,543✔

2515

        // mappings while the allocator is in dirty state.

685,866✔

2516

        reset_free_space_tracking();

685,866✔

2517

        // Add the new version. If this fails in any way, the VersionList may be corrupted.

685,866✔

2518

        // This can lead to readers seing invalid data which is likely to cause them

1,359,543✔

2519

        // to crash. Other writers *must* be prevented from writing any further updates

685,866✔

2520

        // to the database. The flag "commit_in_critical_phase" is used to prevent such updates.

685,866✔

2521

        info->commit_in_critical_phase = 1;

685,866✔

2522

685,866✔

2523

            m_version_manager->add_version(new_top_ref, new_file_size, new_version);

1,359,543✔

2524

1,359,543✔

2525

            // REALM_ASSERT(m_alloc.matches_section_boundary(new_file_size));

1,359,543✔

2526

            REALM_ASSERT(new_top_ref < new_file_size);

685,866✔

2527

685,866✔

2528

        // At this point, the VersionList has been succesfully updated, and the next writer

1,359,543✔

2529

        // can safely proceed once the writemutex has been lifted.

1,359,543✔

2530

        info->commit_in_critical_phase = 0;

685,866✔

2531

685,866✔

2532

1,359,543✔

2533

        // protect against concurrent updates to the .lock file.

1,359,543✔

2534

        // must release m_mutex before this point to obey lock order

1,359,543✔

2535

        std::lock_guard<InterprocessMutex> lock(m_controlmutex);

685,866✔

2536

685,866✔

2537

        info->number_of_versions = live_versions + 1;

1,359,543✔

2538

        info->latest_version_number = new_version;

685,866✔

2539

1,359,543✔

2540

        m_new_commit_available.notify_all();

1,359,543✔

2541

685,866✔

2542

    auto t2 = std::chrono::steady_clock::now();

1,359,543✔

2543

    if (m_logger) {

1,359,543✔

2544

        std::string to_disk_str = commit_to_disk ? util::format(" ref %1", new_top_ref) : " (no commit to disk)";

811,986✔

2545

        m_logger->log(util::Logger::Level::debug, "Commit of size %1 done in %2 us%3", commit_size,

811,986✔

2546

                      std::chrono::duration_cast<std::chrono::microseconds>(t2 - t1).count(), to_disk_str);

278,889✔

2547

283,026✔

2548

830,583✔

2549

144,717✔

2550

#ifdef REALM_DEBUG

673,677✔

2552

18✔

2553

    REALM_ASSERT(is_attached());

18✔

2554

    m_alloc.reserve_disk_space(size); // Throws

36✔

2555

36✔

2556

#endif

18✔

2557

18✔

2559

306✔

2560

    auto lockfile_path = get_core_file(realm_path, CoreFileType::Lock);

306✔

2561

693✔

2562

    File lockfile;

693✔

2563

    lockfile.open(lockfile_path, File::access_ReadWrite, File::create_Auto, 0); // Throws

306✔

2564

    File::CloseGuard fcg(lockfile);

693✔

2565

    lockfile.set_fifo_path(realm_path + ".management", "lock.fifo");

693✔

2566

    if (lockfile.try_rw_lock_exclusive()) { // Throws

693✔

2567

        callback(realm_path);

681✔

2568

        return true;

681✔

2569

651✔

2570

    return false;

369✔

2571

369✔

2572

30✔

2573

std::string DB::get_core_file(const std::string& base_path, CoreFileType type)

30✔

2574

145,914✔

2575

    switch (type) {

145,914✔

2576

        case CoreFileType::Lock:

219,072✔

2577

            return base_path + ".lock";

219,072✔

2578

        case CoreFileType::Storage:

70,953✔

2579

            return base_path;

70,953✔

2580

        case CoreFileType::Management:

68,529✔

2581

            return base_path + ".management";

68,529✔

2582

        case CoreFileType::Note:

79,302✔

2583

            return base_path + ".note";

79,302✔

2584

        case CoreFileType::Log:

9,450✔

2585

            return base_path + ".log";

9,450✔

2586

369✔

2587

    REALM_UNREACHABLE();

369✔

2591

366✔

2592

    if (File::try_remove(get_core_file(base_path, CoreFileType::Storage)) && did_delete) {

366✔

2593

        *did_delete = true;

453✔

2594

453✔

2595

453✔

2596

    File::try_remove(get_core_file(base_path, CoreFileType::Note));

453✔

2597

    File::try_remove(get_core_file(base_path, CoreFileType::Log));

366✔

2598

    util::try_remove_dir_recursive(get_core_file(base_path, CoreFileType::Management));

732✔

2599

732✔

2600

    if (delete_lockfile) {

732✔

2601

        File::try_remove(get_core_file(base_path, CoreFileType::Lock));

93✔

2602

459✔

2603

459✔

2604

93✔

2605

TransactionRef DB::start_read(VersionID version_id)

366✔

2606

1,194,021✔

2607

    if (!is_attached())

1,194,021✔

2608

        throw StaleAccessor("Stale transaction");

554,085✔

2609

    TransactionRef tr;

1,748,100✔

2610

    if (m_fake_read_lock_if_immutable) {

1,194,021✔

2611

        tr = make_transaction_ref(shared_from_this(), &m_alloc, *m_fake_read_lock_if_immutable, DB::transact_Reading);

554,256✔

2612

554,256✔

2613

    else {

1,194,018✔

2614

        ReadLockInfo read_lock = grab_read_lock(ReadLockInfo::Live, version_id);

1,194,018✔

2615

        ReadLockGuard g(*this, read_lock);

1,747,743✔

2616

        read_lock.check();

1,747,743✔

2617

        tr = make_transaction_ref(shared_from_this(), &m_alloc, read_lock, DB::transact_Reading);

1,747,743✔

2618

        g.release();

1,747,743✔

2619

1,747,743✔

2620

    tr->set_file_format_version(get_file_format_version());

1,747,920✔

2621

    return tr;

1,747,920✔

2622

1,748,097✔

2623

554,079✔

2624

TransactionRef DB::start_frozen(VersionID version_id)

554,079✔

2625

15,000✔

2626

    if (!is_attached())

15,000✔

2627

        throw StaleAccessor("Stale transaction");

15,030✔

2628

    TransactionRef tr;

30,030✔

2629

    if (m_fake_read_lock_if_immutable) {

15,000✔

2630

        tr = make_transaction_ref(shared_from_this(), &m_alloc, *m_fake_read_lock_if_immutable, DB::transact_Frozen);

15,036✔

2631

15,036✔

2632

    else {

15,000✔

2633

        ReadLockInfo read_lock = grab_read_lock(ReadLockInfo::Frozen, version_id);

15,000✔

2634

        ReadLockGuard g(*this, read_lock);

30,018✔

2635

        read_lock.check();

30,018✔

2636

        tr = make_transaction_ref(shared_from_this(), &m_alloc, read_lock, DB::transact_Frozen);

30,018✔

2637

        g.release();

30,018✔

2638

30,018✔

2639

    tr->set_file_format_version(get_file_format_version());

30,024✔

2640

    return tr;

30,024✔

2641

30,030✔

2642

15,030✔

2643

TransactionRef DB::start_write(bool nonblocking)

15,030✔

2644

506,883✔

2645

    if (m_fake_read_lock_if_immutable) {

506,883✔

2646

        REALM_ASSERT(false && "Can't write an immutable DB");

492,975✔

2647

492,975✔

2648

    if (nonblocking) {

506,883✔

2649

        bool success = do_try_begin_write();

42✔

2650

        if (!success) {

493,017✔

2651

            return TransactionRef();

48✔

2652

48✔

2653

506,847✔

2654

    else {

506,847✔

2655

        do_begin_write();

999,774✔

2656

999,774✔

2657

999,816✔

2658

        CheckedUniqueLock local_lock(m_mutex);

999,810✔

2659

        if (!is_attached()) {

999,846✔

2660

            local_lock.unlock();

492,969✔

2661

            end_write_on_correct_thread();

492,969✔

2664

        m_write_transaction_open = true;

506,877✔

2665

506,877✔

2666

    TransactionRef tr;

999,846✔

2667

    try {

999,846✔

2668

        ReadLockInfo read_lock = grab_read_lock(ReadLockInfo::Live, VersionID());

999,846✔

2669

        ReadLockGuard g(*this, read_lock);

999,846✔

2670

        read_lock.check();

999,846✔

2671

999,846✔

2672

        tr = make_transaction_ref(shared_from_this(), &m_alloc, read_lock, DB::transact_Writing);

999,846✔

2673

        tr->set_file_format_version(get_file_format_version());

506,877✔

2674

        version_type current_version = read_lock.m_version;

999,846✔

2675

        m_alloc.init_mapping_management(current_version);

999,846✔

2676

        if (Replication* repl = get_replication()) {

999,846✔

2677

            bool history_updated = false;

991,041✔

2678

            repl->initiate_transact(*tr, current_version, history_updated); // Throws

991,041✔

2679

982,281✔

2680

        g.release();

991,086✔

2681

991,086✔

2682

    catch (...) {

999,846✔

2683

        end_write_on_correct_thread();

492,969✔

2686

506,856✔

2687

    return tr;

506,856✔

2688

506,856✔

2689

492,966✔

2690

void DB::async_request_write_mutex(TransactionRef& tr, util::UniqueFunction<void()>&& when_acquired)

492,966✔

2691

393✔

2692

393✔

2693

        util::CheckedLockGuard lck(tr->m_async_mutex);

1,614✔

2694

        REALM_ASSERT(tr->m_async_stage == Transaction::AsyncState::Idle);

1,614✔

2695

        tr->m_async_stage = Transaction::AsyncState::Requesting;

1,614✔

2696

        tr->m_request_time_point = std::chrono::steady_clock::now();

1,614✔

2697

        if (tr->db->m_logger) {

1,614✔

2698

            tr->db->m_logger->log(util::Logger::Level::trace, "Tr %1: Async request write lock", tr->m_log_id);

1,614✔

2699

1,614✔

2700

1,614✔

2701

    std::weak_ptr<Transaction> weak_tr = tr;

1,614✔

2702

    async_begin_write([weak_tr, cb = std::move(when_acquired)]() {

1,614✔

2703

        if (auto tr = weak_tr.lock()) {

1,614✔

2704

            util::CheckedLockGuard lck(tr->m_async_mutex);

1,614✔

2705

            // If a synchronous transaction happened while we were pending

1,614✔

2706

            // we may be in HasCommits

1,614✔

2707

            if (tr->m_async_stage == Transaction::AsyncState::Requesting) {

393✔

2708

                tr->m_async_stage = Transaction::AsyncState::HasLock;

393✔

2709

1,614✔

2710

            if (tr->db->m_logger) {

1,614✔

2711

                auto t2 = std::chrono::steady_clock::now();

1,614✔

2712

                tr->db->m_logger->log(

1,614✔

2713

                    util::Logger::Level::trace, "Tr %1, Got write lock in %2 us", tr->m_log_id,

1,614✔

2714

                    std::chrono::duration_cast<std::chrono::microseconds>(t2 - tr->m_request_time_point).count());

1,614✔

2715

1,614✔

2716

            if (tr->m_waiting_for_write_lock) {

1,614✔

2717

                tr->m_waiting_for_write_lock = false;

1,245✔

2718

                tr->m_async_cv.notify_one();

1,245✔

2719

132✔

2720

            else if (cb) {

477✔

2721

                cb();

477✔

2722

1,482✔

2723

            tr.reset(); // Release pointer while lock is held

1,506✔

2724

1,506✔

2725

});

1,614✔

2726

1,614✔

2727

1,221✔

2728

inline DB::DB(const DBOptions& options)

1,221✔

2731

80,514✔

2732

    if (options.enable_async_writes) {

80,514✔

2733

        m_commit_helper = std::make_unique<AsyncCommitHelper>(this);

144,465✔

2734

144,465✔

2735

143,823✔

2736

63,309✔

2737

namespace {

82,866✔

2742

80,514✔

2743

80,514✔

2744

};

82,866✔

2745

} // namespace

82,866✔

2748

12,420✔

2749

    DBRef retval = std::make_shared<DBInit>(options);

12,420✔

2750

    retval->open(file, no_create, options);

24,837✔

2751

    return retval;

24,837✔

2752

24,837✔

2753

12,417✔

2754

DBRef DB::create(Replication& repl, const std::string& file, const DBOptions& options) NO_THREAD_SAFETY_ANALYSIS

12,417✔

2755

4,863✔

2756

    DBRef retval = std::make_shared<DBInit>(options);

4,863✔

2757

    retval->open(repl, file, options);

10,323✔

2758

    return retval;

10,323✔

2759

10,323✔

2760

5,460✔

2761

DBRef DB::create(std::unique_ptr<Replication> repl, const std::string& file,

5,460✔

2763

50,598✔

2764

    REALM_ASSERT(repl);

50,598✔

2765

    DBRef retval = std::make_shared<DBInit>(options);

102,915✔

2766

    retval->m_history = std::move(repl);

102,915✔

2767

    retval->open(*retval->m_history, file, options);

102,915✔

2768

    return retval;

102,915✔

2769

102,915✔

2770

52,317✔

2771

DBRef DB::create(std::unique_ptr<Replication> repl, const DBOptions& options) NO_THREAD_SAFETY_ANALYSIS

52,317✔

2772

12,630✔

2773

    REALM_ASSERT(repl);

12,630✔

2774

    DBRef retval = std::make_shared<DBInit>(options);

25,302✔

2775

    retval->m_history = std::move(repl);

25,302✔

2776

    retval->open(*retval->m_history, options);

25,302✔

2777

    return retval;

25,302✔

2778

25,302✔

2779

12,672✔

2780

DBRef DB::create_in_memory(std::unique_ptr<Replication> repl, const std::string& in_memory_path,

12,672✔

2789

3✔

2790

    DBOptions options;

3✔

2791

    options.is_immutable = true;

6✔

2792

    DBRef retval = std::make_shared<DBInit>(options);

6✔

2793

    retval->open(buffer, take_ownership);

6✔

2794

    return retval;

6✔

2795

6✔

2796

3✔

2797

void DB::claim_sync_agent()

3✔

2798

7,548✔

2799

    REALM_ASSERT(is_attached());

7,548✔

2800

    std::unique_lock<InterprocessMutex> lock(m_controlmutex);

15,939✔

2801

    if (m_info->sync_agent_present)

15,939✔

2802

        throw MultipleSyncAgents{};

8,394✔

2803

    m_info->sync_agent_present = 1; // Set to true

15,936✔

2804

    m_is_sync_agent = true;

7,548✔

2805

15,933✔

2806

8,388✔

2807

void DB::release_sync_agent()

8,388✔

2808

7,074✔

2809

    REALM_ASSERT(is_attached());

7,074✔

2810

    std::unique_lock<InterprocessMutex> lock(m_controlmutex);

14,658✔

2811

    if (!m_is_sync_agent)

14,658✔

2812

        return;

7,719✔

2813

    REALM_ASSERT(m_info->sync_agent_present);

14,523✔

2814

    m_info->sync_agent_present = 0;

7,068✔

2815

    m_is_sync_agent = false;

14,394✔

2816

14,394✔

2817

7,455✔

2818

void DB::do_begin_possibly_async_write()

7,455✔

2819

189,897✔

2820

    if (m_commit_helper) {

189,897✔

2821

        m_commit_helper->blocking_begin_write();

317,169✔

2822

317,169✔

2823

    else {

194,187✔

2824

        do_begin_write();

194,187✔

2825

124,677✔

2826

250,710✔

2827

60,813✔

2828

void DB::end_write_on_correct_thread() noexcept

191,136✔

2829

696,852✔

2830

    //    m_local_write_mutex.unlock();

696,852✔

2831

    if (!m_commit_helper || !m_commit_helper->blocking_end_write()) {

1,381,083✔

2832

        do_end_write();

570,759✔

2833

1,254,990✔

2834

1,250,628✔

2835

553,776✔

2836

DisableReplication::DisableReplication(Transaction& t)

684,231✔

2841

51✔

2842

    m_owner->set_replication(nullptr);

51✔

2843

    t.m_history = nullptr;

51✔

2844

51✔

2847

51✔

2848

    m_owner->set_replication(m_repl);

51✔

2849

    if (m_version != m_tr.get_version())

51✔

2850

        m_tr.initialize_replication();

51✔

realm / realm-core / github_pull_request_275914

Source File
Press 'n' to go to next uncovered line, 'b' for previous

realm / realm-core / github_pull_request_275914

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous