src/connectivity/overnet/lib/datagram_stream/linearizer.h

// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#pragma once

#include <map>
#include "src/connectivity/overnet/lib/environment/trace.h"
#include "src/connectivity/overnet/lib/vocabulary/callback.h"
#include "src/connectivity/overnet/lib/vocabulary/optional.h"
#include "src/connectivity/overnet/lib/vocabulary/slice.h"
#include "src/connectivity/overnet/lib/vocabulary/status.h"

namespace overnet {

class Linearizer final {
 public:
  explicit Linearizer(uint64_t max_buffer);
  ~Linearizer();

  // Input interface.

  // Add a new slice to the input queue.
  // Returns true if successful, false on failure.
  [[nodiscard]] bool Push(Chunk chunk);

  // Output interface.
  void Pull(StatusOrCallback<Optional<Slice>> ready);
  void PullAll(StatusOrCallback<Optional<std::vector<Slice>>> ready);

  // Returns a finalized status (safe to ignore).
  Status Close(const Status& status, Callback<void> quiesced);
  Status Close(const Status& status);

  bool IsComplete() const { return offset_ == length_; }

 private:
  void IntegratePush(Chunk chunk);
  void AssertValid(const char* marker, const char* pretty_function,
                   const char* file, int line) const;

#ifndef NDEBUG
  class CheckValid {
   public:
    CheckValid(const Linearizer* linearizer, const char* pretty_function,
               const char* file, int line)
        : linearizer_(linearizer),
          pretty_function_(pretty_function),
          file_(file),
          line_(line),
          last_(current_) {
      current_ = this;
      linearizer_->AssertValid("BEGIN", pretty_function_, file_, line_);
    }

    ~CheckValid() {
      if (linearizer_ != nullptr) {
        linearizer_->AssertValid("END", pretty_function_, file_, line_);
      }
      current_ = last_;
    }

    static void CedeChecksTo(Linearizer* linearizer) {
      for (auto* p = current_; p != nullptr; p = p->last_) {
        if (p->linearizer_ == linearizer) {
          p->linearizer_ = nullptr;
        }
      }
    }

   private:
    const Linearizer* linearizer_;
    const char* const pretty_function_;
    const char* const file_;
    int line_;
    // Not threadsafe (but nothing is).
    // It's not clear that we need to support nesting, but it's conceivable
    // through some callback sequence that we do, so current_/last_ keep track
    // of the current nesting structure.
    static inline CheckValid* current_ = nullptr;
    CheckValid* const last_;
  };
#endif

  const uint64_t max_buffer_;
  uint64_t offset_ = 0;
  Optional<uint64_t> length_;
  std::map<uint64_t, Slice> pending_push_;

  enum class ReadMode {
    Closed,
    Idle,
    ReadSlice,
    ReadAll,
  };

  inline friend std::ostream& operator<<(std::ostream& out, ReadMode m) {
    switch (m) {
      case ReadMode::Closed:
        return out << "Closed";
      case ReadMode::Idle:
        return out << "Idle";
      case ReadMode::ReadSlice:
        return out << "ReadSlice";
      case ReadMode::ReadAll:
        return out << "ReadAll";
    }
  }

  struct Closed {
    Status status;
  };
  struct ReadSlice {
    StatusOrCallback<Optional<Slice>> done;
  };
  struct ReadAll {
    std::vector<Slice> building;
    StatusOrCallback<Optional<std::vector<Slice>>> done;
  };
  union ReadData {
    ReadData() {}
    ~ReadData() {}

    Closed closed;
    ReadSlice read_slice;
    ReadAll read_all;
  };

  ReadMode read_mode_ = ReadMode::Idle;
  ReadData read_data_;

  void IdleToClosed(const Status& status);
  void IdleToReadSlice(StatusOrCallback<Optional<Slice>> done);
  void IdleToReadAll(StatusOrCallback<Optional<std::vector<Slice>>> done);
  ReadSlice ReadSliceToIdle();
  ReadAll ReadAllToIdle();
  void ContinueReadAll();
};

}  // namespace overnet