node/src/dataqueue/queue.cc at buffer-integer · nxtedition/node · GitHub

History

1149 lines (968 loc) · 39.5 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

653

654

655

656

657

658

659

660

661

662

663

664

665

666

667

668

669

670

671

672

673

674

675

676

677

678

679

680

681

682

683

684

685

686

687

688

689

690

691

692

693

694

695

696

697

698

699

700

701

702

703

704

705

706

707

708

709

710

711

712

713

714

715

716

717

718

719

720

721

722

723

724

725

726

727

728

729

730

731

732

733

734

735

736

737

738

739

740

741

742

743

744

745

746

747

748

749

750

751

752

753

754

755

756

757

758

759

760

761

762

763

764

765

766

767

768

769

770

771

772

773

774

775

776

777

778

779

780

781

782

783

784

785

786

787

788

789

790

791

792

793

794

795

796

797

798

799

800

801

802

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817

818

819

820

821

822

823

824

825

826

827

828

829

830

831

832

833

834

835

836

837

838

839

840

841

842

843

844

845

846

847

848

849

850

851

852

853

854

855

856

857

858

859

860

861

862

863

864

865

866

867

868

869

870

871

872

873

874

875

876

877

878

879

880

881

882

883

884

885

886

887

888

889

890

891

892

893

894

895

896

897

898

899

900

901

902

903

904

905

906

907

908

909

910

911

912

913

914

915

916

917

918

919

920

921

922

923

924

925

926

927

928

929

930

931

932

933

934

935

936

937

938

939

940

941

942

943

944

945

946

947

948

949

950

951

952

953

954

955

956

957

958

959

960

961

962

963

964

965

966

967

968

969

970

971

972

973

974

975

976

977

978

979

980

981

982

983

984

985

986

987

988

989

990

991

992

993

994

995

996

997

998

999

1000

#include "queue.h"

#include <async_wrap-inl.h>

#include <base_object-inl.h>

#include <env-inl.h>

#include <memory_tracker-inl.h>

#include <node.h>

#include <node_bob-inl.h>

#include <node_errors.h>

#include <node_external_reference.h>

#include <node_file-inl.h>

#include <stream_base-inl.h>

#include <util-inl.h>

#include <uv.h>

#include <v8.h>

#include <algorithm>

#include <deque>

#include <initializer_list>

#include <memory>

#include <vector>

namespace node {

using v8::ArrayBufferView;

using v8::BackingStore;

using v8::Local;

using v8::Object;

using v8::Value;

namespace {

// ============================================================================

class IdempotentDataQueueReader;

class NonIdempotentDataQueueReader;

class EntryImpl : public DataQueue::Entry {

public:

virtual std::shared_ptr<DataQueue::Reader> get_reader() = 0;

};

class DataQueueImpl final : public DataQueue,

public std::enable_shared_from_this<DataQueueImpl> {

public:

// Constructor for an idempotent, fixed sized DataQueue.

DataQueueImpl(std::vector<std::unique_ptr<Entry>>&& list, uint64_t size)

: entries_(std::move(list)),

idempotent_(true),

size_(size),

capped_size_(0) {}

// Constructor for a non-idempotent DataQueue. This kind of queue can have

// entries added to it over time. The size is set to 0 initially. The queue

// can be capped immediately on creation. Depending on the entries that are

// added, the size can be cleared if any of the entries are not capable of

// providing a size.

DataQueueImpl(std::optional<uint64_t> cap = std::nullopt)

: idempotent_(false), size_(0), capped_size_(cap) {}

// Disallow moving and copying.

DataQueueImpl(const DataQueueImpl&) = delete;

DataQueueImpl(DataQueueImpl&&) = delete;

DataQueueImpl& operator=(const DataQueueImpl&) = delete;

DataQueueImpl& operator=(DataQueueImpl&&) = delete;

std::shared_ptr<DataQueue> slice(

uint64_t start,

std::optional<uint64_t> maybeEnd = std::nullopt) override {

// If the data queue is not idempotent, or the size cannot be determined,

// we cannot reasonably create a slice. Therefore, return nothing.

if (!idempotent_ || !size_.has_value()) return nullptr;

uint64_t size = size_.value();

// start cannot be greater than the size.

start = std::min(start, size);

uint64_t end = std::max(start, std::min(maybeEnd.value_or(size), size));

DCHECK_LE(start, end);

uint64_t len = end - start;

uint64_t remaining = end - start;

std::vector<std::unique_ptr<Entry>> slices;

if (remaining > 0) {

for (const auto& entry : entries_) {

// The size of every entry should be known since this is an

// idempotent queue.

uint64_t entrySize = entry->size().value();

if (start > entrySize) {

start -= entrySize;

continue;

}

uint64_t chunkStart = start;

uint64_t len = std::min(remaining, entrySize - chunkStart);

slices.emplace_back(entry->slice(chunkStart, chunkStart + len));

remaining -= len;

start = 0;

if (remaining == 0) break;

}

return std::make_shared<DataQueueImpl>(std::move(slices), len);

}

std::optional<uint64_t> size() const override { return size_; }

bool is_idempotent() const override { return idempotent_; }

bool is_capped() const override { return capped_size_.has_value(); }

std::optional<bool> append(std::unique_ptr<Entry> entry) override {

if (idempotent_) return std::nullopt;

if (!entry) return false;

// If this entry successfully provides a size, we can add it to our size_

// if that has a value, otherwise, we keep uint64_t empty.

if (entry->size().has_value() && size_.has_value()) {

uint64_t entrySize = entry->size().value();

uint64_t size = size_.value();

// If capped_size_ is set, size + entrySize cannot exceed capped_size_

// or the entry cannot be added.

if (capped_size_.has_value() &&

(capped_size_.value() < entrySize + size)) {

return false;

}

size_ = size + entrySize;

} else {

// This entry cannot provide a size. We can still add it but we have to

// clear the known size.

size_ = std::nullopt;

}

entries_.push_back(std::move(entry));

return true;

}

void cap(uint64_t limit = 0) override {

if (is_idempotent()) return;

// If the data queue is already capped, it is possible to call

// cap again with a smaller size.

if (capped_size_.has_value()) {

capped_size_ = std::min(limit, capped_size_.value());

return;

}

// Otherwise just set the limit.

capped_size_ = limit;

}

std::optional<uint64_t> maybeCapRemaining() const override {

if (capped_size_.has_value() && size_.has_value()) {

uint64_t capped_size = capped_size_.value();

uint64_t size = size_.value();

return capped_size > size ? capped_size - size : 0UL;

}

return std::nullopt;

}

void MemoryInfo(node::MemoryTracker* tracker) const override {

tracker->TrackField(

"entries", entries_, "std::vector<std::unique_ptr<Entry>>");

}

void addBackpressureListener(BackpressureListener* listener) override {

if (idempotent_) return;

DCHECK_NOT_NULL(listener);

backpressure_listeners_.insert(listener);

}

void removeBackpressureListener(BackpressureListener* listener) override {

if (idempotent_) return;

DCHECK_NOT_NULL(listener);

backpressure_listeners_.erase(listener);

}

void NotifyBackpressure(size_t amount) {

if (idempotent_) return;

for (auto& listener : backpressure_listeners_) listener->EntryRead(amount);

}

bool HasBackpressureListeners() const noexcept {

return !backpressure_listeners_.empty();

}

std::shared_ptr<Reader> get_reader() override;

SET_MEMORY_INFO_NAME(DataQueue)

SET_SELF_SIZE(DataQueueImpl)

private:

std::vector<std::unique_ptr<Entry>> entries_;

bool idempotent_;

std::optional<uint64_t> size_ = std::nullopt;

std::optional<uint64_t> capped_size_ = std::nullopt;

bool locked_to_reader_ = false;

std::unordered_set<BackpressureListener*> backpressure_listeners_;

friend class DataQueue;

friend class IdempotentDataQueueReader;

friend class NonIdempotentDataQueueReader;

};

// An IdempotentDataQueueReader always reads the entire content of the

// DataQueue with which it is associated, and always from the beginning.

// Reads are non-destructive, meaning that the state of the DataQueue

// will not and cannot be changed.

class IdempotentDataQueueReader final

: public DataQueue::Reader,

public std::enable_shared_from_this<IdempotentDataQueueReader> {

public:

IdempotentDataQueueReader(std::shared_ptr<DataQueueImpl> data_queue)

: data_queue_(std::move(data_queue)) {

CHECK(data_queue_->is_idempotent());

}

// Disallow moving and copying.

IdempotentDataQueueReader(const IdempotentDataQueueReader&) = delete;

IdempotentDataQueueReader(IdempotentDataQueueReader&&) = delete;

IdempotentDataQueueReader& operator=(const IdempotentDataQueueReader&) =

delete;

IdempotentDataQueueReader& operator=(IdempotentDataQueueReader&&) = delete;

int Pull(Next next,

int options,

DataQueue::Vec* data,

size_t count,

size_t max_count_hint = bob::kMaxCountHint) override {

std::shared_ptr<DataQueue::Reader> self = shared_from_this();

// If ended is true, this reader has already reached the end and cannot

// provide any more data.

if (ended_) {

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::Status::STATUS_EOS;

}

// If this is the first pull from this reader, we are first going to

// check to see if there is anything at all to actually do.

if (!current_index_.has_value()) {

// First, let's check the number of entries. If there are no entries,

// we've reached the end and have nothing to do.

// Because this is an idempotent dataqueue, we should always know the

// size...

if (data_queue_->entries_.empty()) {

ended_ = true;

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::Status::STATUS_EOS;

}

current_index_ = 0;

}

// We have current_index_, awesome, we are going to keep reading from

// it until we receive and end.

auto current_reader = getCurrentReader();

if (current_reader == nullptr) {

// Getting the current reader for an entry could fail for several

// reasons. For an FdEntry, for instance, getting the reader may

// fail if the file has been modified since the FdEntry was created.

// We handle the case simply by erroring.

std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});

return UV_EINVAL;

}

CHECK(!pull_pending_);

pull_pending_ = true;

int status = current_reader->Pull(

[this, next = std::move(next)](

int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {

pull_pending_ = false;

// In each of these cases, we do not expect that the source will

// actually have provided any actual data.

CHECK_IMPLIES(status == bob::Status::STATUS_BLOCK ||

status == bob::Status::STATUS_WAIT ||

status == bob::Status::STATUS_EOS,

vecs == nullptr && count == 0);

if (status == bob::Status::STATUS_EOS) {

uint32_t current = current_index_.value() + 1;

current_reader_ = nullptr;

// We have reached the end of this entry. If this is the last entry,

// then we are done. Otherwise, we advance the current_index_, clear

// the current_reader_ and wait for the next read.

if (current == data_queue_->entries_.size()) {

// Yes, this was the final entry. We're all done.

ended_ = true;

} else {

// This was not the final entry, so we update the index and

// continue on by performing another read.

current_index_ = current;

status = bob::STATUS_CONTINUE;

}

std::move(next)(status, nullptr, 0, [](uint64_t) {});

return;

}

std::move(next)(status, vecs, count, std::move(done));

options,

data,

count,

max_count_hint);

// The pull was handled synchronously. If we're not ended, we want to

// make sure status returned is CONTINUE.

if (!pull_pending_) {

if (!ended_) return bob::Status::STATUS_CONTINUE;

// For all other status, we just fall through and return it straightaway.

}

// The other statuses that can be returned by the pull are:

// bob::Status::STATUS_CONTINUE - means that the entry has more data

// to pull.

// bob::Status::STATUS_BLOCK - means that the entry has more data to

// pull but it is not available yet. The

// caller should not keep calling pull for

// now but may check again later.

// bob::Status::STATUS_WAIT - means that the entry has more data to

// pull but it won't be provided

// synchronously, instead the next() callback

// will be called when the data is available.

// For any of these statuses, we want to keep the current index and

// current_reader_ set for the next pull.

return status;

}

DataQueue::Reader* getCurrentReader() {

CHECK(!ended_);

CHECK(current_index_.has_value());

if (current_reader_ == nullptr) {

auto& entry = data_queue_->entries_[current_index_.value()];

// Because this is an idempotent reader, let's just be sure to

// doublecheck that the entry itself is actually idempotent

DCHECK(entry->is_idempotent());

current_reader_ = static_cast<EntryImpl&>(*entry).get_reader();

}

return current_reader_.get();

}

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(IdempotentDataQueueReader)

SET_SELF_SIZE(IdempotentDataQueueReader)

private:

std::shared_ptr<DataQueueImpl> data_queue_;

std::optional<uint32_t> current_index_ = std::nullopt;

std::shared_ptr<DataQueue::Reader> current_reader_ = nullptr;

bool ended_ = false;

bool pull_pending_ = false;

};

// A NonIdempotentDataQueueReader reads entries from the DataEnqueue

// and removes those entries from the queue as they are fully consumed.

// This means that reads are destructive and the state of the DataQueue

// is mutated as the read proceeds.

class NonIdempotentDataQueueReader final

: public DataQueue::Reader,

public std::enable_shared_from_this<NonIdempotentDataQueueReader> {

public:

NonIdempotentDataQueueReader(std::shared_ptr<DataQueueImpl> data_queue)

: data_queue_(std::move(data_queue)) {

CHECK(!data_queue_->is_idempotent());

}

// Disallow moving and copying.

NonIdempotentDataQueueReader(const NonIdempotentDataQueueReader&) = delete;

NonIdempotentDataQueueReader(NonIdempotentDataQueueReader&&) = delete;

NonIdempotentDataQueueReader& operator=(const NonIdempotentDataQueueReader&) =

delete;

NonIdempotentDataQueueReader& operator=(NonIdempotentDataQueueReader&&) =

delete;

int Pull(Next next,

int options,

DataQueue::Vec* data,

size_t count,

size_t max_count_hint = bob::kMaxCountHint) override {

std::shared_ptr<DataQueue::Reader> self = shared_from_this();

// If ended is true, this reader has already reached the end and cannot

// provide any more data.

if (ended_) {

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::Status::STATUS_EOS;

}

// If the collection of entries is empty, there's nothing currently left to

// read. How we respond depends on whether the data queue has been capped

// or not.

if (data_queue_->entries_.empty()) {

// If the data_queue_ is empty, and not capped, then we can reasonably

// expect more data to be provided later, but we don't know exactly when

// that'll happe, so the proper response here is to return a blocked

// status.

if (!data_queue_->is_capped()) {

std::move(next)(bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});

return bob::STATUS_BLOCK;

}

// However, if we are capped, the status will depend on whether the size

// of the data_queue_ is known or not.

if (data_queue_->size().has_value()) {

// If the size is known, and it is still less than the cap, then we

// still might get more data. We just don't know exactly when that'll

// come, so let's return a blocked status.

if (data_queue_->size().value() < data_queue_->capped_size_.value()) {

std::move(next)(

bob::Status::STATUS_BLOCK, nullptr, 0, [](uint64_t) {});

return bob::STATUS_BLOCK;

}

// Otherwise, if size is equal to or greater than capped, we are done.

// Fall through to allow the end handling to run.

}

// If the size is not known, and the data queue is capped, no additional

// entries are going to be added to the queue. Since we are all out of

// entries, we're done. There's nothing left to read.

current_reader_ = nullptr;

ended_ = true;

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::STATUS_EOS;

}

auto current_reader = getCurrentReader();

if (current_reader == nullptr) {

std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});

return UV_EINVAL;

}

// If we got here, we have an entry to read from.

CHECK(!pull_pending_);

pull_pending_ = true;

int status = current_reader->Pull(

[this, next = std::move(next)](

int status, const DataQueue::Vec* vecs, uint64_t count, Done done) {

pull_pending_ = false;

// In each of these cases, we do not expect that the source will

// actually have provided any actual data.

CHECK_IMPLIES(status == bob::Status::STATUS_BLOCK ||

status == bob::Status::STATUS_WAIT ||

status == bob::Status::STATUS_EOS,

vecs == nullptr && count == 0);

if (status == bob::Status::STATUS_EOS) {

data_queue_->entries_.erase(data_queue_->entries_.begin());

ended_ = data_queue_->entries_.empty();

current_reader_ = nullptr;

if (!ended_) status = bob::Status::STATUS_CONTINUE;

std::move(next)(status, nullptr, 0, [](uint64_t) {});

return;

}

// If there is a backpressure listener, lets report on how much data

// was actually read.

if (data_queue_->HasBackpressureListeners()) {

// How much did we actually read?

size_t read = 0;

for (uint64_t n = 0; n < count; n++) {

read += vecs[n].len;

}

data_queue_->NotifyBackpressure(read);

}

// Now that we have updated this readers state, we can forward

// everything on to the outer next.

std::move(next)(status, vecs, count, std::move(done));

options,

data,

count,

max_count_hint);

if (!pull_pending_) {

// The callback was resolved synchronously. Let's check our status.

if (!ended_) return bob::Status::STATUS_CONTINUE;

// For all other status, we just fall through and return it straightaway.

}

// The other statuses that can be returned by the pull are:

// bob::Status::STATUS_CONTINUE - means that the entry has more data

// to pull.

// bob::Status::STATUS_BLOCK - means that the entry has more data to

// pull but it is not available yet. The

// caller should not keep calling pull for

// now but may check again later.

// bob::Status::STATUS_WAIT - means that the entry has more data to

// pull but it won't be provided

// synchronously, instead the next() callback

// will be called when the data is available.

// For any of these statuses, we want to keep the current index and

// current_reader_ set for the next pull.

return status;

}

DataQueue::Reader* getCurrentReader() {

CHECK(!ended_);

CHECK(!data_queue_->entries_.empty());

if (current_reader_ == nullptr) {

auto& entry = data_queue_->entries_.front();

current_reader_ = static_cast<EntryImpl&>(*entry).get_reader();

}

return current_reader_.get();

}

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(NonIdempotentDataQueueReader)

SET_SELF_SIZE(NonIdempotentDataQueueReader)

private:

std::shared_ptr<DataQueueImpl> data_queue_;

std::shared_ptr<DataQueue::Reader> current_reader_ = nullptr;

bool ended_ = false;

bool pull_pending_ = false;

};

std::shared_ptr<DataQueue::Reader> DataQueueImpl::get_reader() {

if (is_idempotent()) {

return std::make_shared<IdempotentDataQueueReader>(shared_from_this());

}

if (locked_to_reader_) return nullptr;

locked_to_reader_ = true;

return std::make_shared<NonIdempotentDataQueueReader>(shared_from_this());

}

// ============================================================================

// An empty, always idempotent entry.

class EmptyEntry final : public EntryImpl {

public:

class EmptyReader final : public DataQueue::Reader,

public std::enable_shared_from_this<EmptyReader> {

public:

int Pull(Next next,

int options,

DataQueue::Vec* data,

size_t count,

size_t max_count_hint = bob::kMaxCountHint) override {

auto self = shared_from_this();

if (ended_) {

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::Status::STATUS_EOS;

}

ended_ = true;

std::move(next)(

bob::Status::STATUS_CONTINUE, nullptr, 0, [](uint64_t) {});

return bob::Status::STATUS_CONTINUE;

}

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(EmptyReader)

SET_SELF_SIZE(EmptyReader)

private:

bool ended_ = false;

};

EmptyEntry() = default;

// Disallow moving and copying.

EmptyEntry(const EmptyEntry&) = delete;

EmptyEntry(EmptyEntry&&) = delete;

EmptyEntry& operator=(const EmptyEntry&) = delete;

EmptyEntry& operator=(EmptyEntry&&) = delete;

std::shared_ptr<DataQueue::Reader> get_reader() override {

return std::make_shared<EmptyReader>();

}

std::unique_ptr<Entry> slice(

uint64_t start,

std::optional<uint64_t> maybeEnd = std::nullopt) override {

if (start != 0) return nullptr;

return std::make_unique<EmptyEntry>();

}

std::optional<uint64_t> size() const override { return 0; }

bool is_idempotent() const override { return true; }

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(EmptyEntry)

SET_SELF_SIZE(EmptyEntry)

};

// ============================================================================

// An entry that consists of a single memory resident v8::BackingStore.

// These are always idempotent and always a fixed, known size.

class InMemoryEntry final : public EntryImpl {

public:

struct InMemoryFunctor final {

std::shared_ptr<BackingStore> backing_store;

void operator()(uint64_t) { backing_store = nullptr; }

};

class InMemoryReader final

: public DataQueue::Reader,

public std::enable_shared_from_this<InMemoryReader> {

public:

InMemoryReader(InMemoryEntry& entry) : entry_(entry) {}

int Pull(Next next,

int options,

DataQueue::Vec* data,

size_t count,

size_t max_count_hint = bob::kMaxCountHint) override {

auto self = shared_from_this();

if (ended_) {

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::Status::STATUS_EOS;

}

ended_ = true;

DataQueue::Vec vec{

reinterpret_cast<uint8_t*>(entry_.backing_store_->Data()) +

entry_.offset_,

entry_.byte_length_,

};

std::move(next)(bob::Status::STATUS_CONTINUE,

&vec,

InMemoryFunctor({entry_.backing_store_}));

return bob::Status::STATUS_CONTINUE;

}

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(InMemoryReader)

SET_SELF_SIZE(InMemoryReader)

private:

InMemoryEntry& entry_;

bool ended_ = false;

};

InMemoryEntry(std::shared_ptr<BackingStore> backing_store,

uint64_t offset,

uint64_t byte_length)

: backing_store_(std::move(backing_store)),

offset_(offset),

byte_length_(byte_length) {

// The offset_ + byte_length_ cannot extend beyond the size of the

// backing store, because that would just be silly.

CHECK_LE(offset_ + byte_length_, backing_store_->ByteLength());

}

// Disallow moving and copying.

InMemoryEntry(const InMemoryEntry&) = delete;

InMemoryEntry(InMemoryEntry&&) = delete;

InMemoryEntry& operator=(const InMemoryEntry&) = delete;

InMemoryEntry& operator=(InMemoryEntry&&) = delete;

std::shared_ptr<DataQueue::Reader> get_reader() override {

return std::make_shared<InMemoryReader>(*this);

}

std::unique_ptr<Entry> slice(

uint64_t start,

std::optional<uint64_t> maybeEnd = std::nullopt) override {

const auto makeEntry = [&](uint64_t start,

uint64_t len) -> std::unique_ptr<Entry> {

if (len == 0) {

return std::make_unique<EmptyEntry>();

}

return std::make_unique<InMemoryEntry>(backing_store_, start, len);

};

start += offset_;

// The start cannot extend beyond the maximum end point of this entry.

start = std::min(start, offset_ + byte_length_);

if (maybeEnd.has_value()) {

uint64_t end = maybeEnd.value();

// The end cannot extend beyond the maximum end point of this entry,

// and the end must be equal to or greater than the start.

end = std::max(start, std::min(offset_ + end, offset_ + byte_length_));

return makeEntry(start, end - start);

}

// If no end is given, then the new length is the current length

// minus the adjusted start.

return makeEntry(start, byte_length_ - start);

}

std::optional<uint64_t> size() const override { return byte_length_; }

bool is_idempotent() const override { return true; }

void MemoryInfo(node::MemoryTracker* tracker) const override {

tracker->TrackField(

"store", backing_store_, "std::shared_ptr<v8::BackingStore>");

}

SET_MEMORY_INFO_NAME(InMemoryEntry)

SET_SELF_SIZE(InMemoryEntry)

private:

std::shared_ptr<BackingStore> backing_store_;

uint64_t offset_;

uint64_t byte_length_;

friend class InMemoryReader;

};

// ============================================================================

// An entry that wraps a DataQueue. The entry takes on the characteristics

// of the wrapped dataqueue.

class DataQueueEntry : public EntryImpl {

public:

explicit DataQueueEntry(std::shared_ptr<DataQueue> data_queue)

: data_queue_(std::move(data_queue)) {

CHECK(data_queue_);

}

// Disallow moving and copying.

DataQueueEntry(const DataQueueEntry&) = delete;

DataQueueEntry(DataQueueEntry&&) = delete;

DataQueueEntry& operator=(const DataQueueEntry&) = delete;

DataQueueEntry& operator=(DataQueueEntry&&) = delete;

std::shared_ptr<DataQueue::Reader> get_reader() override {

return std::make_shared<ReaderImpl>(data_queue_->get_reader());

}

std::unique_ptr<Entry> slice(

uint64_t start, std::optional<uint64_t> end = std::nullopt) override {

std::shared_ptr<DataQueue> sliced = data_queue_->slice(start, end);

if (!sliced) return nullptr;

return std::make_unique<DataQueueEntry>(std::move(sliced));

}

// Returns the number of bytes represented by this Entry if it is

// known. Certain types of entries, such as those backed by streams

// might not know the size in advance and therefore cannot provide

// a value. In such cases, size() must return std::nullopt.

// If the entry is idempotent, a size should always be available.

std::optional<uint64_t> size() const override { return data_queue_->size(); }

// When true, multiple reads on the object must produce the exact

// same data or the reads will fail. Some sources of entry data,

// such as streams, may not be capable of preserving idempotency

// and therefore must not claim to be. If an entry claims to be

// idempotent and cannot preserve that quality, subsequent reads

// must fail with an error when a variance is detected.

bool is_idempotent() const override { return data_queue_->is_idempotent(); }

void MemoryInfo(node::MemoryTracker* tracker) const override {

tracker->TrackField(

"data_queue", data_queue_, "std::shared_ptr<DataQueue>");

}

DataQueue& getDataQueue() { return *data_queue_; }

SET_MEMORY_INFO_NAME(DataQueueEntry)

SET_SELF_SIZE(DataQueueEntry)

private:

std::shared_ptr<DataQueue> data_queue_;

class ReaderImpl : public DataQueue::Reader,

public std::enable_shared_from_this<ReaderImpl> {

public:

explicit ReaderImpl(std::shared_ptr<DataQueue::Reader> inner)

: inner_(std::move(inner)) {}

int Pull(DataQueue::Reader::Next next,

int options,

DataQueue::Vec* data,

size_t count,

size_t max_count_hint) override {

auto self = shared_from_this();

return inner_->Pull(

std::move(next), options, data, count, max_count_hint);

}

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(ReaderImpl)

SET_SELF_SIZE(ReaderImpl)

private:

std::shared_ptr<DataQueue::Reader> inner_;

};

// ============================================================================

// An FdEntry reads from a file descriptor. A check is made before each read

// to determine if the fd has changed on disc. This is a best-effort check

// that only looks at file size, creation, and modification times. The stat

// check is also async, so there's a natural race condition there where the

// file could be modified between the stat and actual read calls. That's

// a tolerable risk here. While FdEntry is considered idempotent, this race

// means that it is indeed possible for multiple reads to return different

// results if the file just happens to get modified.

class FdEntry final : public EntryImpl {

// TODO(@jasnell, @flakey5):

// * This should only allow reading from regular files. No directories, no

// pipes, etc.

// * The reader should support accepting the buffer(s) from the pull, if any.

// It should

// only allocate a managed buffer if the pull doesn't provide any.

// * We might want to consider making the stat on each read sync to eliminate

// the race

// condition described in the comment above.

public:

static std::unique_ptr<FdEntry> Create(Environment* env, Local<Value> path) {

// We're only going to create the FdEntry if the file exists.

uv_fs_t req = uv_fs_t();

auto cleanup = OnScopeLeave([&] { uv_fs_req_cleanup(&req); });

auto buf = std::make_shared<BufferValue>(env->isolate(), path);

if (uv_fs_stat(nullptr, &req, buf->out(), nullptr) < 0) return nullptr;

return std::make_unique<FdEntry>(

env, std::move(buf), req.statbuf, 0, req.statbuf.st_size);

}

FdEntry(Environment* env,

std::shared_ptr<BufferValue> path_,

uv_stat_t stat,

uint64_t start,

uint64_t end)

: env_(env),

path_(std::move(path_)),

stat_(stat),

start_(start),

end_(end) {

CHECK_LE(start, end);

}

std::shared_ptr<DataQueue::Reader> get_reader() override {

return ReaderImpl::Create(this);

}

std::unique_ptr<Entry> slice(

uint64_t start, std::optional<uint64_t> end = std::nullopt) override {

uint64_t new_start = start_ + start;

uint64_t new_end = end_;

if (end.has_value()) {

new_end = std::min(end.value() + start_, end_);

}

CHECK(new_start >= start_);

CHECK(new_end <= end_);

return std::make_unique<FdEntry>(env_, path_, stat_, new_start, new_end);

}

std::optional<uint64_t> size() const override { return end_ - start_; }

bool is_idempotent() const override { return true; }

Environment* env() const { return env_; }

SET_NO_MEMORY_INFO()

SET_MEMORY_INFO_NAME(FdEntry)

SET_SELF_SIZE(FdEntry)

private:

Environment* env_;

std::shared_ptr<BufferValue> path_;

uv_stat_t stat_;

uint64_t start_ = 0;

uint64_t end_ = 0;

bool is_modified(const uv_stat_t& other) {

return other.st_size != stat_.st_size ||

other.st_mtim.tv_nsec != stat_.st_mtim.tv_nsec;

}

static bool CheckModified(FdEntry* entry, int fd) {

uv_fs_t req = uv_fs_t();

auto cleanup = OnScopeLeave([&] { uv_fs_req_cleanup(&req); });

// TODO(jasnell): Note the use of a sync fs call here is a bit unfortunate.

// Doing this asynchronously creates a bit of a race condition tho, a file

// could be unmodified when we call the operation but then by the time the

// async callback is triggered to give us that answer the file is modified.

// While such silliness is still possible here, the sync call at least makes

// it less likely to hit the race.

if (uv_fs_fstat(nullptr, &req, fd, nullptr) < 0) return true;

return entry->is_modified(req.statbuf);

}

class ReaderImpl final : public DataQueue::Reader,

public StreamListener,

public std::enable_shared_from_this<ReaderImpl> {

public:

static std::shared_ptr<ReaderImpl> Create(FdEntry* entry) {

uv_fs_t req;

auto cleanup = OnScopeLeave([&] { uv_fs_req_cleanup(&req); });

int file =

uv_fs_open(nullptr, &req, entry->path_->out(), O_RDONLY, 0, nullptr);

if (file < 0 || FdEntry::CheckModified(entry, file)) {

uv_fs_close(nullptr, &req, file, nullptr);

return nullptr;

}

Realm* realm = entry->env()->principal_realm();

return std::make_shared<ReaderImpl>(

BaseObjectPtr<fs::FileHandle>(

fs::FileHandle::New(realm->GetBindingData<fs::BindingData>(),

file,

Local<Object>(),

entry->start_,

entry->end_ - entry->start_)),

entry);

}

explicit ReaderImpl(BaseObjectPtr<fs::FileHandle> handle, FdEntry* entry)

: env_(handle->env()), handle_(std::move(handle)), entry_(entry) {

handle_->PushStreamListener(this);

handle_->env()->AddCleanupHook(cleanup, this);

}

~ReaderImpl() override {

handle_->env()->RemoveCleanupHook(cleanup, this);

DrainAndClose();

handle_->RemoveStreamListener(this);

}

uv_buf_t OnStreamAlloc(size_t suggested_size) override {

return env_->allocate_managed_buffer(suggested_size);

}

void OnStreamRead(ssize_t nread, const uv_buf_t& buf) override {

std::shared_ptr<v8::BackingStore> store =

env_->release_managed_buffer(buf);

if (ended_) {

// If we got here and ended_ is true, it means we ended and drained

// while the read was pending. We're just going to do nothing.

CHECK(pending_pulls_.empty());

return;

}

CHECK(reading_);

auto pending = DequeuePendingPull();

if (CheckModified(entry_, handle_->GetFD())) {

DrainAndClose();

// The file was modified while the read was pending. We need to error.

std::move(pending.next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});

return;

}

if (nread < 0) {

if (nread == UV_EOF) {

std::move(pending.next)(bob::STATUS_EOS, nullptr, 0, [](uint64_t) {});

} else {

std::move(pending.next)(nread, nullptr, 0, [](uint64_t) {});

}

return DrainAndClose();

}

DataQueue::Vec vec;

vec.base = static_cast<uint8_t*>(store->Data());

vec.len = static_cast<uint64_t>(nread);

std::move(pending.next)(

bob::STATUS_CONTINUE, &vec, 1, [store](uint64_t) {});

if (pending_pulls_.empty()) {

reading_ = false;

if (handle_->IsAlive()) handle_->ReadStop();

}

int Pull(Next next,

int options,

DataQueue::Vec* data,

size_t count,

size_t max_count_hint = bob::kMaxCountHint) override {

if (ended_ || !handle_->IsAlive()) {

std::move(next)(bob::STATUS_EOS, nullptr, 0, [](uint64_t) {});

return bob::STATUS_EOS;

}

if (FdEntry::CheckModified(entry_, handle_->GetFD())) {

DrainAndClose();

std::move(next)(UV_EINVAL, nullptr, 0, [](uint64_t) {});

return UV_EINVAL;

}

pending_pulls_.emplace_back(std::move(next), shared_from_this());

if (!reading_) {

View remainder of file in raw view

node/src/dataqueue/queue.cc at buffer-integer · nxtedition/node · GitHub

Uh oh!

Expand file tree

Breadcrumbs

queue.cc

Latest commit

History

Breadcrumbs

queue.cc

File metadata and controls