implement proper Abort() for composite lambdas

file/luajit_node.cc

@@ -286,7 +286,12 @@ class Node::Lambda final : public nf7::Node::Lambda,
         auto self = d.self.lock();
         if (!self) return luaL_error(L, "context expired");
 
-        d.caller->Handle(name, nf7::luajit::CheckValue(L, 3), self);
+        auto v = nf7::luajit::CheckValue(L, 3);
+
+        auto caller = d.caller;
+        caller->env().ExecSub(self, [self, caller, v = std::move(v)]() mutable {
+          caller->Handle(name, std::move(value), self);
+        });
         return 0;
       });
       lua_setfield(L, -2, "__call");

file/node_network.cc

@@ -421,6 +421,9 @@ class Network::Lambda : public Node::Lambda,
   }
   void Abort() noexcept override {
     abort_ = true;
+    for (auto& p : lamap_) {
+      p.second->Abort();
+    }
   }
   size_t GetMemoryUsage() const noexcept override {
     return 0;
@@ -438,7 +441,7 @@ class Network::Lambda : public Node::Lambda,
   std::unordered_map<ItemId, std::shared_ptr<Node::Lambda>> lamap_;
   std::unordered_map<Node::Lambda*, ItemId> idmap_;
 
-  std::atomic<bool> abort_ = false;
+  bool abort_ = false;
 };
 void Network::DetachLambda() noexcept {
   if (lambda_ && lambda_->isRoot()) {

file/node_ref.cc

@@ -164,6 +164,8 @@ class Ref final : public nf7::File, public nf7::Node {
 class Ref::Lambda final : public Node::Lambda,
     public std::enable_shared_from_this<Ref::Lambda> {
  public:
+  static constexpr size_t kMaxDepth = 1024;
+
   Lambda(Ref& f, const std::shared_ptr<Node::Lambda>& parent) :
       Node::Lambda(f, parent), ref_(&f), log_(f.log_) {
   }
@@ -180,12 +182,22 @@ class Ref::Lambda final : public Node::Lambda,
     }
     if (caller == parent) {
       if (!base_) {
+        if (depth() > kMaxDepth) {
+          log_->Error("stack overflow");
+          return;
+        }
         base_ = ref_->target().CreateLambda(shared_from_this());
       }
       base_->Handle(name, v, shared_from_this());
     }
   }
 
+  void Abort() noexcept override {
+    if (base_) {
+      base_->Abort();
+    }
+  }
+
  private:
   Ref* const ref_;
   std::shared_ptr<nf7::LoggerRef> log_;

file/sequencer_call.cc

@@ -121,7 +121,8 @@ class Call::Lambda final : public nf7::Sequencer::Lambda,
       Sequencer::Lambda(f, ctx), file_(f.life_) {
   }
 
-  void Run(const std::shared_ptr<Sequencer::Session>& ss) noexcept;
+  void Run(const std::shared_ptr<Sequencer::Session>& ss) noexcept override;
+  void Abort() noexcept override;
 
  private:
   nf7::Life<Call>::Ref file_;
@@ -130,17 +131,14 @@ class Call::Lambda final : public nf7::Sequencer::Lambda,
 
   nf7::Node* cached_node_ = nullptr;
   std::shared_ptr<Node::Lambda> la_;
+
+  bool abort_ = false;
 };
 class Call::SessionLambda final : public nf7::Node::Lambda {
  public:
   SessionLambda(Call& f, const std::shared_ptr<Call::Lambda>& parent) noexcept :
       nf7::Node::Lambda(f, parent) {
   }
-  ~SessionLambda() noexcept {
-    if (ss_ && expects_.size() > 0) {
-      ss_->Finish();
-    }
-  }
 
   void Listen(Call& f, const std::shared_ptr<Sequencer::Session>& ss) noexcept {
     assert(!ss_);
@@ -167,6 +165,13 @@ class Call::SessionLambda final : public nf7::Node::Lambda {
     expects_.erase(std::string {name});
     FinishIf();
   }
+  void Abort() noexcept override {
+    if (ss_) {
+      ss_->Finish();
+      ss_ = nullptr;
+      expects_.clear();
+    }
+  }
 
  private:
   std::shared_ptr<Sequencer::Session> ss_;
@@ -186,6 +191,7 @@ std::shared_ptr<Sequencer::Lambda> Call::CreateLambda(
 }
 void Call::Lambda::Run(const std::shared_ptr<Sequencer::Session>& ss) noexcept
 try {
+  if (abort_) return;
   file_.EnforceAlive();
 
   auto& data   = file_->data();
@@ -219,6 +225,16 @@ try {
 } catch (nf7::File::NotImplementedException&) {
   ss->Finish();
 }
+void Call::Lambda::Abort() noexcept {
+  if (ssla_) {
+    ssla_->Abort();
+    ssla_ = nullptr;
+  }
+  if (la_) {
+    la_->Abort();
+    la_ = nullptr;
+  }
+}
 
 
 void Call::UpdateItem(Sequencer::Editor&) noexcept {