graph.h

Go to the documentation of this file.

/*
 * Copyright 2026 The A11 Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef A11_FLOW_GRAPH_H_
#define A11_FLOW_GRAPH_H_
 
#include <cstddef>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
 
#include <absl/base/nullability.h>
#include <absl/container/flat_hash_map.h>
#include <absl/container/flat_hash_set.h>
#include <absl/time/time.h>
 
#include "a11/flow/syntax.h"
#include "a11/flow/vocabulary.h"
 
namespace a11::flow::graph {
 
using RefId = size_t;
using StepId = size_t;
using BodyId = size_t;
using ExprId = size_t;
 
inline constexpr size_t kNone = static_cast<size_t>(-1);
 
enum class RefKind {
  kFlowPort,
  kCallPort,
  kNode,
  kNodeId,
  kStatus,
  kHeader,
  kExpr,
  kDerived,
  kZip,
  kMerge,
  kWinner,
  kBound,
};
 
struct LogTail {
  std::string level;
  std::string format;
  bool has_format = false;
  std::vector<ExprId> arguments;
  int line = 0;
};
 
struct Stage {
  std::string name;
  vocabulary::StageArgument takes = vocabulary::StageArgument::kNone;
  long long count = 0;
  std::string text;
  ExprId expr = kNone;
  RefId stream = kNone;
  LogTail log;
  bool indexed = false;
  long long index = 0;
  bool named_or_indexed = false;
  absl::Duration duration;
  bool descending = false;
  // `fold`: what the first pass carries, the name it is bound to, and the ref
  // that name resolves to inside the fold's expression.
  syntax::Constant start;
  std::string carried;
  RefId carry = kNone;
  bool tolerant = false;
  RefId failures = kNone;
  int parallel = 1;
  bool ordered = true;
};
 
struct Ref {
  RefKind kind = RefKind::kExpr;
  std::string label;
  BodyId owner = kNone;
  bool writable = false;
  bool unary = false;
  long long skip = 0;
 
  std::string name;
  syntax::PortDirection direction = syntax::PortDirection::kInput;
  StepId call = kNone;
  ExprId id_expr = kNone;
  std::string node_map;
  RefId subject = kNone;
  StepId subject_step = kNone;
  std::string header;
  syntax::Constant fallback;
  bool has_fallback = false;
  ExprId expr = kNone;
  RefId source = kNone;
  Stage stage;
  std::string role;
  StepId bound_by = kNone;
  // `kZip`: the streams read in step, in the order they were written, which is
  // the order their values appear in each tuple.
  std::vector<RefId> sources;
};
 
enum class StepKind {
  kCall,
  kPipe,
  kSkip,
  kWait,
  kDrain,
  kCancel,
  kAbort,
  kFail,
  kLog,
  kCapture,
  kForEach,
  kRepeat,
  kIf,
  kBlock,
};
 
std::string_view StepKindName(StepKind kind);
 
inline bool RecordsOutcome(StepKind kind) {
  return kind == StepKind::kBlock || kind == StepKind::kForEach ||
         kind == StepKind::kRepeat || kind == StepKind::kPipe;
}
 
struct Step {
  StepKind kind = StepKind::kPipe;
  std::string label;
  BodyId body = kNone;
  std::vector<StepId> after;
  syntax::Location location;
 
  std::string action;
  std::string mode;
  std::string node_map;
  std::optional<absl::Duration> timeout;
  bool tee = false;
  bool tolerant = false;
  std::vector<std::pair<std::string, ExprId>> headers;
  std::vector<std::string> forward;
  ExprId action_id = kNone;
  absl::flat_hash_map<std::string, RefId> ports;
  RefId status = kNone;
 
  RefId source = kNone;
  RefId destination = kNone;
  bool discard = false;
  std::optional<long long> count;
  std::string slot;
 
  RefId outcome = kNone;
  // `kWait`: the outcomes of a `wait first of` / `wait all of`, in the order
  // they were written.
  std::vector<RefId> subjects;
  bool race = false;
  RefId winner = kNone;
 
  StepId target = kNone;
 
  ExprId code = kNone;
  ExprId message = kNone;
  std::string code_name;
 
  LogTail log;
 
  std::vector<BodyId> bodies;
  RefId item = kNone;
  RefId index = kNone;
  int parallel = 1;
  RefId carry = kNone;
  RefId carry_source = kNone;
  syntax::Constant start;
  std::optional<int> max_iterations;
  ExprId condition = kNone;
  bool stop_when = true;
};
 
struct Body {
  std::string label;
  BodyId parent = kNone;
  StepId owner_step = kNone;
  std::vector<StepId> steps;
};
 
struct Expr {
  const syntax::Node* absl_nullable node = nullptr;
  std::vector<std::pair<const syntax::Node* absl_nonnull, RefId>> bound;
  std::vector<RefId> refs;
};
 
struct FlowGraph {
  std::string name;
  std::vector<Ref> refs;
  std::vector<Step> steps;
  std::vector<Body> bodies;
  std::vector<Expr> exprs;
  BodyId root = kNone;
 
  [[nodiscard]] std::vector<RefId> Upstreams(RefId ref) const;
  [[nodiscard]] std::vector<RefId> ValueRefs(RefId ref) const;
 
  [[nodiscard]] std::vector<RefId> Sources(StepId step) const;
  [[nodiscard]] std::vector<RefId> ValueSources(StepId step) const;
  [[nodiscard]] std::vector<RefId> Destinations(StepId step) const;
  // The refs the *stages* a step reads write to: a `try ...
  [[nodiscard]] std::vector<RefId> StageDestinations(StepId step) const;
  [[nodiscard]] std::vector<RefId> Observed(StepId step) const;
  [[nodiscard]] std::vector<BodyId> NestedBodies(BodyId body) const;
};
 
struct Analysis {
  BodyId body = kNone;
  std::vector<RefId> refs;
  absl::flat_hash_map<RefId, int> readers;
  // The ones read from inside a nested body, which are buffered once and
  // replayed to each reader:
  absl::flat_hash_set<RefId> materialise;
  absl::flat_hash_map<RefId, int> writers;
  std::vector<RefId> destinations;
  absl::flat_hash_map<StepId, std::vector<RefId>> held;
  std::vector<RefId> nodes;
};
 
Analysis Analyse(const FlowGraph& flow, BodyId body);
 
class GraphBuilder {
 public:
  explicit GraphBuilder(FlowGraph& flow) : flow_(&flow) {}
 
  FlowGraph& flow() { return *flow_; }
 
  BodyId AddBody(std::string label, BodyId parent = kNone,
                 StepId owner_step = kNone) {
    Body body;
    body.label = std::move(label);
    body.parent = parent;
    body.owner_step = owner_step;
    flow_->bodies.push_back(std::move(body));
    return flow_->bodies.size() - 1;
  }
 
  [[nodiscard]] bool Carries(RefId ref) const {
    return ref != kNone && ref < flow_->refs.size() && flow_->refs[ref].unary;
  }
 
  static bool StageMakesOne(const Stage& stage) {
    if (vocabulary::ReducingStages().contains(stage.name)) {
      {
        return true;
      }
    }
    return stage.name == "first" && stage.count == 1;
  }
 
  static bool StagePreservesCount(const Stage& stage) {
    static const auto* const kPerValue = new absl::flat_hash_set<std::string>{
        "map", "at", "truncate", "text", "json", "packb", "strformat", "where",
        "mime", "distinct", "first", "last", "drop", "log", "logf", "scan",
        // `sort` reorders and keeps every value; `timeout` and `pace` say
        // *when* a value may pass and change nothing about which do.
        "sort", "timeout", "pace"};
    return kPerValue->contains(stage.name);
  }
 
  RefId AddRef(Ref ref) {
    switch (ref.kind) {
      case RefKind::kHeader:
      case RefKind::kNodeId:
      case RefKind::kStatus:
      case RefKind::kWinner:
      case RefKind::kExpr:
      case RefKind::kBound:
        // One header, one id, one status record, one winner, one evaluated
        // expression, and one value bound per pass of a loop.
        ref.unary = true;
        break;
      case RefKind::kNode:
        // A node is a stream the flow may write from anywhere, including from
        // inside a loop, so nothing about it is provable from the plan.
        ref.unary = false;
        break;
      case RefKind::kDerived:
        ref.unary = StageMakesOne(ref.stage) ||
                    (Carries(ref.source) && StagePreservesCount(ref.stage));
        break;
      case RefKind::kZip:
        // A tuple per round, and the rounds run until every source has ended,
        // so one round is only certain when every source has at most one value.
        ref.unary = !ref.sources.empty();
        for (const RefId source : ref.sources) {
          if (!Carries(source)) {
            {
              ref.unary = false;
            }
          }
        }
        break;
      case RefKind::kMerge:
        // Every value of every source, so one value only when they all have at
        // most one *and* there is one of them: two unary sources interleave
        // into two values.
        ref.unary = ref.sources.size() == 1 && Carries(ref.sources.front());
        break;
      case RefKind::kFlowPort:
      case RefKind::kCallPort:
        // Whatever the declaration said, which only the caller knows.
        break;
    }
    flow_->refs.push_back(std::move(ref));
    return flow_->refs.size() - 1;
  }
 
  StepId AddStep(Step step) {
    const BodyId body = step.body;
    flow_->steps.push_back(std::move(step));
    const StepId id = flow_->steps.size() - 1;
    if (body != kNone) {
      {
        flow_->bodies[body].steps.push_back(id);
      }
    }
    return id;
  }
 
  ExprId AddExpr(Expr expr) {
    flow_->exprs.push_back(std::move(expr));
    return flow_->exprs.size() - 1;
  }
 
  Ref& ref(RefId id) { return flow_->refs[id]; }
 
  Step& step(StepId id) { return flow_->steps[id]; }
 
  Expr& expr(ExprId id) { return flow_->exprs[id]; }
 
 private:
  FlowGraph* absl_nonnull flow_;
};
 
}  // namespace a11::flow::graph
 
#endif  // A11_FLOW_GRAPH_H_