text:
#include "tensorflow/core/common_runtime/function.h"

#include <deque>
#include <vector>

#include "tensorflow/core/common_runtime/device.h"
#include "tensorflow/core/common_runtime/executor.h"
#include "tensorflow/core/common_runtime/executor_factory.h"
#include "tensorflow/core/common_runtime/graph_optimizer.h"
#include "tensorflow/core/common_runtime/memory_types.h"
#include "tensorflow/core/common_runtime/rendezvous_mgr.h"
#include "tensorflow/core/framework/collective.h"
#include "tensorflow/core/framework/function.h"
#include "tensorflow/core/framework/node_def.pb.h"
#include "tensorflow/core/framework/node_def_util.h"
#include "tensorflow/core/framework/op.h"
#include "tensorflow/core/framework/op_kernel.h"
#include "tensorflow/core/framework/versions.pb.h"
#include "tensorflow/core/graph/algorithm.h"
#include "tensorflow/core/graph/control_flow.h"
#include "tensorflow/core/graph/gradients.h"
#include "tensorflow/core/graph/graph_constructor.h"
#include "tensorflow/core/graph/optimizer_cse.h"
#include "tensorflow/core/lib/core/threadpool.h"
#include "tensorflow/core/lib/gtl/map_util.h"
#include "tensorflow/core/platform/macros.h"

// See core/kernels/function_ops.cc for related kernels.

namespace tensorflow {

// A few string constant used throughout this module.
static constexpr const char* const kArgOp = FunctionLibraryDefinition::kArgOp;
static constexpr const char* const kRetOp = FunctionLibraryDefinition::kRetOp;
static constexpr const char* const kGradientOp =
    FunctionLibraryDefinition::kGradientOp;
static constexpr const char* const kNodeLabel = "Func";
static constexpr const char* const kFuncAttr =
    FunctionLibraryDefinition::kFuncAttr;

// Represents the index-th output of a node.
struct Endpoint {
  Node* node;
  int index;

  // Returns the string name represents this endpoint.
  string name() const {
    if (index == 0) {
      return node->name();
    } else {
      return strings::StrCat(node->name(), ":", index);
    }
  }

  DataType dtype() const { return node->output_type(index); }
};

struct EndpointHash {
  uint64 operator()(const Endpoint& x) const {
    return Hash64(reinterpret_cast<const char*>(&x.node), sizeof(Node*),
                  x.index);
  }
};

struct EndpointEq {
  bool operator()(const Endpoint& x, const Endpoint& y) const {
    return (x.node == y.node) && (x.index == y.index);
  }
};

// The following Add* routines are used to add a few graph nodes while
// functions are transformed.
static Node* AddNoOp(Graph* g) {
  NodeDef ndef;
  ndef.set_name(g->NewName(kNodeLabel));
  ndef.set_op("NoOp");
  Status s;
  Node* ret = g->AddNode(ndef, &s);
  TF_CHECK_OK(s);
  return ret;
}

static Node* AddIdentity(Graph* g, Endpoint input) {
  DCHECK_LT(0, input.dtype());
  NodeDef ndef;
  ndef.set_name(g->NewName(kNodeLabel));
  ndef.set_op("Identity");
  ndef.add_input(input.name());
  AddNodeAttr("T", BaseType(input.dtype()), &ndef);
  Status s;
  Node* ret = g->AddNode(ndef, &s);
  TF_CHECK_OK(s);
  g->AddEdge(input.node, input.index, ret, 0);
  return ret;
}

static Node* AddArg(Graph* g, DataType dtype, int index) {
  DCHECK_LT(0, dtype);
  DCHECK_LT(dtype, DT_FLOAT_REF);
  NodeDef ndef;
  ndef.set_name(g->NewName(kNodeLabel));
  ndef.set_op(kArgOp);
  AddNodeAttr("T", dtype, &ndef);
  AddNodeAttr("index", index, &ndef);
  Status s;
  Node* ret = g->AddNode(ndef, &s);
  TF_CHECK_OK(s);
  return ret;
}

static Node* AddRet(Graph* g, Endpoint input, int index) {
  DCHECK_LT(0, input.dtype());
  DCHECK_LT(input.dtype(), DT_FLOAT_REF);
  NodeDef ndef;
  ndef.set_name(g->NewName(kNodeLabel));
  ndef.set_op(kRetOp);
  ndef.add_input(input.name());
  AddNodeAttr("T", input.dtype(), &ndef);
  AddNodeAttr("index", index, &ndef);
  Status s;
  Node* ret = g->AddNode(ndef, &s);
  TF_CHECK_OK(s);
  g->AddEdge(input.node, input.index, ret, 0);
  return ret;
}

class FunctionLibraryRuntimeImpl : public FunctionLibraryRuntime {
 public:
  FunctionLibraryRuntimeImpl(const DeviceMgr* dmgr, Env* env, Device* device,
                             int graph_def_version,
                             const FunctionLibraryDefinition* lib_def,
                             thread::ThreadPool* default_thread_pool,
                             const OptimizerOptions& optimizer_options,
                             CustomKernelCreator custom_kernel_creator,
                             ProcessFunctionLibraryRuntime* parent);

  ~FunctionLibraryRuntimeImpl() override;

  Status Instantiate(const string& function_name, AttrSlice attrs,
                     const InstantiateOptions& options,
                     Handle* handle) override;

  Status ReleaseHandle(Handle handle) override;

  const FunctionBody* GetFunctionBody(Handle handle) override;

  Status CreateKernel(const NodeDef& ndef, OpKernel** kernel) override;

  void Run(const Options& opts, Handle handle, gtl::ArraySlice<Tensor> args,
           std::vector<Tensor>* rets, DoneCallback done) override;
  // NOTE(mrry): This overload is currently only implemented for local function
  // execution.
  // TODO(b/70346412): Implement support for remote function execution when
  // passing a call frame.
  void Run(const Options& opts, Handle handle, CallFrameInterface* frame,
           DoneCallback done) override;

  bool IsStateful(const string& function) override;

  const FunctionLibraryDefinition* GetFunctionLibraryDefinition()
      const override {
    return base_lib_def_;
  }

  Device* device() override { return device_; }
  const DeviceMgr* device_mgr() const override { return device_mgr_; }
  Env* env() override { return env_; }
  int graph_def_version() override { return graph_def_version_; }

  string DebugString(Handle h) override;

  Status Clone(std::unique_ptr<FunctionLibraryDefinition>* out_lib_def,
               std::unique_ptr<ProcessFunctionLibraryRuntime>* out_pflr,
               FunctionLibraryRuntime** out_flr) override;

 private:
  typedef FunctionLibraryRuntimeImpl ME;

  const DeviceMgr* const device_mgr_;
  Device* const device_;
  Env* const env_;
  const int graph_def_version_;
  const FunctionLibraryDefinition* const base_lib_def_;
  GraphOptimizer optimizer_;
  const CustomKernelCreator custom_kernel_creator_;
  Executor::Args::Runner default_runner_;
  const string device_name_;

  std::function<Status(const string&, const OpDef**)> get_func_sig_;
  std::function<Status(const NodeDef&, OpKernel**)> create_kernel_;

  mutable mutex mu_;

  int next_handle_ GUARDED_BY(mu_);

  // The instantiated and transformed function is encoded as a Graph
  // object, and an executor is created for the graph.
  struct Item {
    uint64 instantiation_counter = 0;
    const Graph* graph = nullptr;                            // Owned by exec.
    const FunctionLibraryDefinition* overlay_lib = nullptr;  // Not owned.
    FunctionBody* func_graph = nullptr;
    Executor* exec = nullptr;
    string executor_type;

    ~Item() {
      delete this->func_graph;
      delete this->exec;
    }
  };
  std::unordered_map<Handle, std::unique_ptr<Item>> items_ GUARDED_BY(mu_);

  ProcessFunctionLibraryRuntime* parent_ = nullptr;  // not owned.

  Status CreateKernel(const NodeDef& ndef,
                      const FunctionLibraryDefinition* lib_def,
                      OpKernel** kernel);
  Status FunctionDefToBody(const FunctionDef& fdef, AttrSlice attrs,
                           const FunctionLibraryDefinition* lib_def,
                           FunctionBody** fbody);
  Status CreateItem(Handle handle, Item** item);
  Status GetOrCreateItem(Handle handle, Item** item);
  Status InstantiateSymbolicGradient(const NameAttrList& func,
                                     const FunctionLibraryDefinition* lib_def,
                                     FunctionBody** g_body);
  bool IsLocalTarget(const InstantiateOptions& options);
  AttrValueMap FixAttrs(const AttrSlice& attrs);
  void RunRemote(const Options& opts, Handle handle,
                 gtl::ArraySlice<Tensor> args, std::vector<Tensor>* rets,
                 Executor::Args* exec_args, Item* item, DoneCallback done);

  TF_DISALLOW_COPY_AND_ASSIGN(FunctionLibraryRuntimeImpl);
};

FunctionLibraryRuntimeImpl::FunctionLibraryRuntimeImpl(
    const DeviceMgr* dmgr, Env* env, Device* device, int graph_def_version,
    const FunctionLibraryDefinition* lib_def,
    thread::ThreadPool* default_thread_pool,
    const OptimizerOptions& optimizer_options,
    CustomKernelCreator custom_kernel_creator,
    ProcessFunctionLibraryRuntime* parent)
    : device_mgr_(dmgr),
      device_(device),
      env_(env),
      graph_def_version_(graph_def_version),
      base_lib_def_(lib_def),
      optimizer_(optimizer_options),
      custom_kernel_creator_(std::move(custom_kernel_creator)),
      default_runner_(nullptr),
      device_name_(device_ == nullptr
                       ? ProcessFunctionLibraryRuntime::kDefaultFLRDevice
                       : device_->name()),
      next_handle_(0),
      parent_(parent) {
  get_func_sig_ = [this](const string& op, const OpDef** sig) {
    return base_lib_def_->LookUpOpDef(op, sig);
  };
  create_kernel_ = [this](const NodeDef& ndef, OpKernel** kernel) {
    return CreateKernel(ndef, kernel);
  };
  thread::ThreadPool* pool = nullptr;
  if (device_ != nullptr) {
    pool = device_->tensorflow_device_thread_pool();
  }
  if (pool == nullptr) {
    pool = default_thread_pool;
  }
  if (pool != nullptr) {
    default_runner_ = [pool](Executor::Args::Closure c) {
      pool->Schedule(std::move(c));
    };
  }
}

FunctionLibraryRuntimeImpl::~FunctionLibraryRuntimeImpl() {}

// An asynchronous op kernel which executes an instantiated function
// defined in a library.
class CallOp : public AsyncOpKernel {
 public:
  CallOp(FunctionLibraryRuntime::Handle handle, OpKernelConstruction* ctx)
      : AsyncOpKernel(ctx), handle_(handle) {}

  ~CallOp() override {}

  void ComputeAsync(OpKernelContext* ctx, DoneCallback done) override {
    FunctionLibraryRuntime* lib = ctx->function_library();
    OP_REQUIRES_ASYNC(ctx, lib != nullptr,
                      errors::Internal("No function library is provided."),
                      done);
    FunctionLibraryRuntime::Options opts;
    opts.step_id = ctx->step_id();
    opts.rendezvous = ctx->rendezvous();
    opts.cancellation_manager = ctx->cancellation_manager();
    opts.step_container = ctx->step_container();
    opts.stats_collector = ctx->stats_collector();
    opts.runner = ctx->runner();
    std::vector<Tensor> args;
    args.reserve(ctx->num_inputs());
    for (int i = 0; i < ctx->num_inputs(); ++i) {
      args.push_back(ctx->input(i));
    }
    std::vector<Tensor>* rets = new std::vector<Tensor>;
    lib->Run(opts, handle_, args, rets,
             [ctx, done, rets](const Status& status) {
               if (!status.ok()) {
                 ctx->SetStatus(status);
               } else {
                 const int ret_size = static_cast<int>(rets->size());
                 CHECK_EQ(ret_size, ctx->num_outputs());
                 for (int i = 0; i < ret_size; ++i) {
                   ctx->set_output(i, (*rets)[i]);
                 }
               }
               delete rets;
               done();
             });
  }

 private:
  FunctionLibraryRuntime::Handle handle_;

  TF_DISALLOW_COPY_AND_ASSIGN(CallOp);
};

const FunctionBody* FunctionLibraryRuntimeImpl::GetFunctionBody(Handle h) {
  LocalHandle local_handle = parent_->GetHandleOnDevice(device_name_, h);
  if (local_handle == kInvalidLocalHandle) {
    LOG(ERROR) << "Could not find Handle: " << h
               << " on device: " << device_name_;
    return nullptr;
  }

  mutex_lock l(mu_);
  CHECK_EQ(1, items_.count(local_handle));
  return items_[local_handle]->func_graph;
}

Status FunctionLibraryRuntimeImpl::CreateKernel(const NodeDef& ndef,
                                                OpKernel** kernel) {
  return CreateKernel(ndef, base_lib_def_, kernel);
}

Status FunctionLibraryRuntimeImpl::CreateKernel(
    const NodeDef& ndef, const FunctionLibraryDefinition* lib_def,
    OpKernel** kernel) {
  // If a custom kernel creator is given, try that.
  Status s;
  if (custom_kernel_creator_) {
    std::unique_ptr<OpKernel> ret;
    s = custom_kernel_creator_(this, ndef, &ret);
    if (s.ok()) {
      *kernel = ret.release();
      return s;
    } else {
      VLOG(2) << "Custom creator error: " << s;
      // Falls through.
      s = Status::OK();
    }
  }

  if (lib_def->Find(ndef.op()) == nullptr) {
    // A primitive operation. Creates the registered kernel.
    return CreateNonCachedKernel(device_, this, ndef, graph_def_version_,
                                 kernel);
  }

  // Try to instantiate this function for the func/attr. Maybe it's
  // cached already.
  InstantiateOptions options;
  if (lib_def != base_lib_def_) {
    options.overlay_lib = lib_def;
  }
  Handle handle;
  TF_RETURN_IF_ERROR(
      Instantiate(ndef.op(), AttrSlice(&ndef.attr()), options, &handle));

  const FunctionBody* fbody = GetFunctionBody(handle);
  CHECK_NOTNULL(fbody);

  // TODO(zhifengc): For now, we assume int32 and resources are always on host
  // memory and other types are always on device memory. We should do type
  // inference over function body to derive the correct input/output memory
  // types.
  MemoryTypeVector input_memory_types;
  for (const auto& t : fbody->arg_types) {
    input_memory_types.push_back(
        (t == DT_INT32 || t == DT_RESOURCE) ? HOST_MEMORY : DEVICE_MEMORY);
  }
  MemoryTypeVector output_memory_types;
  for (const auto& t : fbody->ret_types) {
    output_memory_types.push_back(t == DT_INT32 ? HOST_MEMORY : DEVICE_MEMORY);
  }

  // Constructs a CallOp kernel for running the instantiated function.
  auto device_type = DeviceType(device_->attributes().device_type());
  OpKernelConstruction construction(
      device_type, device_, device_->GetAllocator(AllocatorAttributes()), &ndef,
      &fbody->fdef.signature(), this, fbody->arg_types, input_memory_types,
      fbody->ret_types, output_memory_types, graph_def_version_, &s);
  *kernel = new CallOp(handle, &construction);
  if (!s.ok()) {
    delete *kernel;
  }
  return s;
}

Status FunctionLibraryRuntimeImpl::FunctionDefToBody(
    const FunctionDef& fdef, AttrSlice attrs,
    const FunctionLibraryDefinition* lib_def, FunctionBody** fbody) {
  if (lib_def == base_lib_def_) {
    return FunctionDefToBodyHelper(fdef, attrs, lib_def, get_func_sig_, fbody);
  } else {
    auto get_func_sig = [lib_def](const string& op, const OpDef** sig) {
      return lib_def->LookUpOpDef(op, sig);
    };
    return FunctionDefToBodyHelper(fdef, attrs, lib_def, get_func_sig, fbody);
  }
}

Status FunctionLibraryRuntimeImpl::InstantiateSymbolicGradient(
    const NameAttrList& func, const FunctionLibraryDefinition* lib_def,
    FunctionBody** g_body) {
  const FunctionDef* fdef = lib_def->Find(func.name());
  if (fdef == nullptr) {
    // f is a primitive op.
    gradient::Creator creator;
    TF_RETURN_IF_ERROR(gradient::GetOpGradientCreator(func.name(), &creator));
    if (creator == nullptr) {
      return errors::InvalidArgument("No gradient is defined for ",
                                     func.name());
    }
    FunctionDef grad_fdef;
    // TODO(josh11b): Should filter out the attrs from func that aren't used
    // by the gradient function.
    TF_RETURN_IF_ERROR(creator(AttrSlice(&func.attr()), &grad_fdef));
    TF_RETURN_IF_ERROR(
        FunctionDefToBody(grad_fdef, AttrSlice(&func.attr()), lib_def, g_body));
  } else {
    // f is a user-defined function.
    InstantiateOptions options;
    if (lib_def != base_lib_def_) {
      options.overlay_lib = lib_def;
    }
    Handle f_handle;
    TF_RETURN_IF_ERROR(
        Instantiate(func.name(), AttrSlice(&func.attr()), options, &f_handle));
    const FunctionBody* f_body = GetFunctionBody(f_handle);
    CHECK_NOTNULL(f_body);
    *g_body = SymbolicGradient(*f_body);
  }
  return Status::OK();
}

bool FunctionLibraryRuntimeImpl::IsLocalTarget(
    const InstantiateOptions& options) {
  if (device_ == nullptr) return true;
  if (options.target.empty()) return true;
  Device* target_device;
  if (!device_mgr_->LookupDevice(options.target, &target_device).ok()) {
    return false;
  }
  return target_device == device_;
}

Status FunctionLibraryRuntimeImpl::Instantiate(
    const string& function_name, AttrSlice attrs,
    const InstantiateOptions& options, Handle* handle) {
  if (!IsLocalTarget(options)) {
    return parent_->Instantiate(function_name, attrs, options, handle);
  }

  // Since this is a local target, ensure that the local `device_name_` appears
  // in the canonical key.
  InstantiateOptions options_copy(options);
  options_copy.target = device_name_;
  const string key = Canonicalize(function_name, attrs, options_copy);

  {
    mutex_lock l(mu_);
    *handle = parent_->GetHandle(key);
    if (*handle != kInvalidHandle) {
      FunctionLibraryRuntime::LocalHandle handle_on_device =
          parent_->GetHandleOnDevice(device_name_, *handle);
      if (handle_on_device == kInvalidLocalHandle) {
        return errors::Internal("LocalHandle not found for handle ", *handle,
                                ".");
      }
      auto item_handle = items_.find(handle_on_device);
      if (item_handle == items_.end()) {
        return errors::Internal("LocalHandle ", handle_on_device,
                                " for handle ", *handle,
                                " not found in items.");
      }
      ++item_handle->second->instantiation_counter;
      return Status::OK();
    }
  }

  Status s;
  const FunctionLibraryDefinition* lib_def =
      options.overlay_lib ? options.overlay_lib : base_lib_def_;
  FunctionBody* fbody = nullptr;
  if (function_name == kGradientOp) {
    const AttrValue* f = attrs.Find(kFuncAttr);
    if (f == nullptr) {
      return errors::InvalidArgument("SymbolicGradient is missing attr: f");
    }
    const auto& func = f->func();
    if (func.name() == kGradientOp) {
      return errors::InvalidArgument("Can't take gradient of SymbolicGradient");
    }
    const string grad = lib_def->FindGradient(func.name());
    if (!grad.empty()) {
      return Instantiate(grad, AttrSlice(&func.attr()), options, handle);
    }
    TF_RETURN_IF_ERROR(InstantiateSymbolicGradient(func, lib_def, &fbody));
  } else {
    const FunctionDef* fdef = lib_def->Find(function_name);
    if (fdef == nullptr) {
      return errors::NotFound("Function ", function_name, " is not defined.");
    }
    TF_RETURN_IF_ERROR(FunctionDefToBody(*fdef, attrs, lib_def, &fbody));
  }

  {
    mutex_lock l(mu_);
    *handle = parent_->GetHandle(key);
    if (*handle != kInvalidHandle) {
      delete fbody;
      ++items_[parent_->GetHandleOnDevice(device_name_, *handle)]
            ->instantiation_counter;
    } else {
      *handle = parent_->AddHandle(key, device_name_, next_handle_);
      Item* item = new Item;
      item->func_graph = fbody;
      item->overlay_lib = options.overlay_lib;
      item->instantiation_counter = 1;
      item->executor_type = options.executor_type;
      items_.emplace(next_handle_, std::unique_ptr<Item>(item));
      next_handle_++;
    }
  }
  return Status::OK();
}

Status FunctionLibraryRuntimeImpl::ReleaseHandle(Handle handle) {
  if (!parent_->IsInstantiatedOnDevice(device_name_, handle)) {
    return parent_->ReleaseHandle(handle);
  }

  LocalHandle h = parent_->GetHandleOnDevice(device_name_, handle);
  CHECK_NE(h, kInvalidLocalHandle);
  mutex_lock l(mu_);
  CHECK_EQ(1, items_.count(h));
  std::unique_ptr<Item>& item = items_[h];
  --item->instantiation_counter;
  if (item->instantiation_counter == 0) {
    items_.erase(h);
    TF_RETURN_IF_ERROR(parent_->RemoveHandle(handle));
  }
  return Status::OK();
}

void DumpGraph(StringPiece label, const Graph* g) {
  // TODO(zhifengc): Change Graph to record #nodes.
  VLOG(1) << "Graph " << label << " #nodes " << g->num_nodes() << " #edges "
          << g->num_edges();
  if (VLOG_IS_ON(2)) {
    for (const auto& line : str_util::Split(DebugString(g), '\n')) {
      VLOG(2) << "|| " << line;
    }
  }
}

void OptimizeGraph(FunctionLibraryRuntime* lib, std::unique_ptr<Graph>* g) {
  OptimizerOptions opts;
  opts.set_do_common_subexpression_elimination(true);
  opts.set_do_function_inlining(true);
  opts.set_do_constant_folding(true);
  GraphOptimizer optimizer(opts);
  optimizer.Optimize(lib, lib->env(), lib->device(), g, /*shape_map=*/nullptr);
}

namespace {
// Removes all stateless nodes that do not contribute to a return
// value from the function body.  Unlike `RemoveDeadNodes()`, which is
// triggered by `OptimizerOptions.do_function_inlining`, this pass
// ignores the SINK node, from which (by definition) all nodes are
// reverse reachable.
void PruneFunctionBody(Graph* g) {
  VLOG(2) << "Pruning function body";
  std::unordered_set<const Node*> nodes;
  for (auto n : g->nodes()) {
    // NOTE(mrry): "_Retval" nodes are stateful, and so will be added
    // to the seed set of `nodes`.
    // TODO(mrry): Investigate whether the `n->IsControlFlow()` test is
    // still needed. It would be preferable to prune entire loops and/or
    // conditionals if they are not used in the graph.
    if (n->IsControlFlow() || n->op_def().is_stateful()) {
      nodes.insert(n);
    }
  }
  bool changed = PruneForReverseReachability(g, std::move(nodes));
  if (changed) {
    FixupSourceAndSinkEdges(g);
  }
}
}  // namespace

Status FunctionLibraryRuntimeImpl::CreateItem(Handle handle, Item** item) {
  const FunctionBody* fbody;
  const FunctionLibraryDefinition* lib_def;
  string executor_type;
  {
    mutex_lock l(mu_);
    fbody = (*item)->func_graph;
    lib_def = (*item)->overlay_lib;
    executor_type = (*item)->executor_type;
  }
  if (!lib_def) {
    lib_def = base_lib_def_;
  }
  std::unique_ptr<Graph> g(new Graph(lib_def));
  CopyGraph(*fbody->graph, g.get());

  PruneFunctionBody(g.get());
  optimizer_.Optimize(this, env(), device(), &g, /*shape_map=*/nullptr);
  TF_RETURN_IF_ERROR(EnsureMemoryTypes(DeviceType(device()->device_type()),
                                       device()->name(), g.get()));

  // Creates an executor based on the g.  This must be done without
  // holding mu_ because create_kernel_ calls back into the library.
  LocalExecutorParams params;
  params.device = device_;
  params.function_library = this;
  if (lib_def == base_lib_def_) {
    params.create_kernel = create_kernel_;
  } else {
    params.create_kernel = [this, lib_def](const NodeDef& ndef,
                                           OpKernel** kernel) {
      return CreateKernel(ndef, lib_def, kernel);
    };
  }
  params.delete_kernel = [](OpKernel* kernel) {
    DeleteNonCachedKernel(kernel);
  };
  Graph* graph = g.get();
  std::unique_ptr<Executor> exec;
  TF_RETURN_IF_ERROR(NewExecutor(executor_type, params, std::move(g), &exec));
  {
    // Guard item since it is already inserted in items_.
    mutex_lock l(mu_);
    if ((*item)->exec == nullptr) {
      (*item)->graph = graph;
      (*item)->exec = exec.release();
    }
  }
  return Status::OK();
}

Status FunctionLibraryRuntimeImpl::GetOrCreateItem(Handle handle, Item** item) {
  LocalHandle local_handle = parent_->GetHandleOnDevice(device_name_, handle);
  {
    mutex_lock l(mu_);
    if (items_.count(local_handle) == 0) {
      return errors::NotFound("Function handle ", handle,
                              " is not valid. Likely an internal error.");
    }
    *item = items_[local_handle].get();
    if ((*item)->exec != nullptr) {
      return Status::OK();
    }
  }
  // NOTE: We need to call CreateItem out of mu_ because creating an
  // executor needs to call CreateKernel.
  return CreateItem(handle, item);
}

void FunctionLibraryRuntimeImpl::RunRemote(const Options& opts, Handle handle,
                                           gtl::ArraySlice<Tensor> args,
                                           std::vector<Tensor>* rets,
                                           Executor::Args* exec_args,
                                           Item* item, DoneCallback done) {
  DCHECK(exec_args->call_frame == nullptr);
  string target_device = parent_->GetDeviceName(handle);
  string source_device = opts.source_device;
  Rendezvous* rendezvous = opts.rendezvous;
  DeviceContext* device_context;
  Status s = parent_->GetDeviceContext(target_device, &device_context);
  if (!s.ok()) {
    delete exec_args;
    done(s);
    return;
  }
  int64 src_incarnation, target_incarnation;
  s = parent_->GetDeviceIncarnation(source_device, &src_incarnation);
  s.Update(parent_->GetDeviceIncarnation(target_device, &target_incarnation));
  if (!s.ok()) {
    delete exec_args;
    done(s);
    return;
  }

  const FunctionBody* fbody = GetFunctionBody(handle);
  FunctionCallFrame* frame =
      new FunctionCallFrame(fbody->arg_types, fbody->ret_types);
  exec_args->call_frame = frame;
  if (!s.ok()) {
    delete frame;
    delete exec_args;
    done(s);
    return;
  }

  std::vector<AllocatorAttributes> args_alloc_attrs, rets_alloc_attrs;
  args_alloc_attrs.reserve(fbody->arg_types.size());
  rets_alloc_attrs.reserve(fbody->ret_types.size());
  for (const auto& arg_type : fbody->arg_types) {
    AllocatorAttributes arg_alloc_attrs;
    if (DataTypeAlwaysOnHost(arg_type)) {
      arg_alloc_attrs.set_on_host(true);
    }
    args_alloc_attrs.push_back(arg_alloc_attrs);
  }
  for (const auto& ret_type : fbody->ret_types) {
    AllocatorAttributes ret_alloc_attrs;
    if (DataTypeAlwaysOnHost(ret_type)) {
      ret_alloc_attrs.set_on_host(true);
    }
    rets_alloc_attrs.push_back(ret_alloc_attrs);
  }

  // The ProcFLR sends the arguments to the function from the source_device to
  // the target_device. So here we receive those arguments. Similarly, when the
  // computation is done and stored in *rets, we send the return values back
  // to the source_device (caller) so that the ProcFLR can receive them later.
  std::vector<Tensor>* remote_args = new std::vector<Tensor>;
  ProcessFunctionLibraryRuntime::ReceiveTensorsAsync(
      source_device, target_device, "arg_", src_incarnation, args.size(),
      device_context, args_alloc_attrs, rendezvous, remote_args,
      [frame, remote_args, item, source_device, target_device,
       target_incarnation, rendezvous, device_context, rets, done, exec_args,
       rets_alloc_attrs](const Status& status) {
        Status s = status;
        if (s.ok()) {
          s = frame->SetArgs(*remote_args);
        }
        if (!s.ok()) {
          delete frame;
          delete remote_args;
          delete exec_args;
          done(s);
          return;
        }
        item->exec->RunAsync(
            *exec_args,
            [frame, rets, done, source_device, target_device,
             target_incarnation, rendezvous, device_context, remote_args,
             exec_args, rets_alloc_attrs](const Status& status) {
              Status s = status;
              if (s.ok()) {
                s = frame->ConsumeRetvals(rets);
              }
              delete frame;
              if (!s.ok()) {
                delete remote_args;
                delete exec_args;
                done(s);
                return;
              }
              s = ProcessFunctionLibraryRuntime::SendTensors(
                  target_device, source_device, "ret_", target_incarnation,
                  *rets, device_context, rets_alloc_attrs, rendezvous);
              delete remote_args;
              delete exec_args;
              done(s);
            });
      });
}

void FunctionLibraryRuntimeImpl::Run(const Options& opts, Handle handle,
                                     gtl::ArraySlice<Tensor> args,
                                     std::vector<Tensor>* rets,
                                     DoneCallback done) {
  if (opts.cancellation_manager && opts.cancellation_manager->IsCancelled()) {
    done(errors::Cancelled(""));
    return;
  }
  Options run_opts = opts;
  if (opts.create_rendezvous) {
    Rendezvous* rendezvous = new IntraProcessRendezvous(device_mgr_);
    run_opts.rendezvous = rendezvous;
    run_opts.create_rendezvous = false;
    done = [done, rendezvous](const Status& status) {
      rendezvous->Unref();
      done(status);
    };
  }

  if (!parent_->IsInstantiatedOnDevice(device_name_, handle)) {
    parent_->Run(run_opts, handle, args, rets, done);
    return;
  }

  if (run_opts.runner == nullptr) {
    run_opts.runner = &default_runner_;
  }
  DCHECK(run_opts.runner != nullptr);

  Executor::Args* exec_args = new Executor::Args;
  // Inherit the step_id from the caller.
  exec_args->step_id = run_opts.step_id;
  exec_args->rendezvous = run_opts.rendezvous;
  exec_args->stats_collector = run_opts.stats_collector;
  exec_args->cancellation_manager = run_opts.cancellation_manager;
  exec_args->step_container = run_opts.step_container;
  exec_args->runner = *run_opts.runner;
  exec_args->collective_executor = run_opts.collective_executor;

  Item* item = nullptr;
  Status s = GetOrCreateItem(handle, &item);
  if (!s.ok()) {
    delete exec_args;
    done(s);
    return;
  }

  if (run_opts.remote_execution) {
    // NOTE(mrry): `RunRemote()` will set `exec_args->call_frame` for us.
    RunRemote(run_opts, handle, args, rets, exec_args, item, done);
    return;
  }

  const FunctionBody* fbody = GetFunctionBody(handle);
  FunctionCallFrame* frame =
      new FunctionCallFrame(fbody->arg_types, fbody->ret_types);
  exec_args->call_frame = frame;
  s = frame->SetArgs(args);
  if (!s.ok()) {
    delete frame;
    delete exec_args;
    done(s);
    return;
  }

  item->exec->RunAsync(
      // Executor args
      *exec_args,
      // Done callback.
      [frame, rets, done, exec_args](const Status& status) {
        Status s = status;
        if (s.ok()) {
          s = frame->ConsumeRetvals(rets);
        }
        delete frame;
        delete exec_args;
        done(s);
      });
}

void FunctionLibraryRuntimeImpl::Run(const Options& opts, Handle handle,
                                     CallFrameInterface* frame,
                                     DoneCallback done) {
  if (opts.cancellation_manager && opts.cancellation_manager->IsCancelled()) {
    done(errors::Cancelled(""));
    return;
  }
  if (!parent_->IsInstantiatedOnDevice(device_name_, handle) ||
      opts.remote_execution) {
    done(errors::Unimplemented("Remote calling with CallFrameInterface"));
    return;
  }

  Options run_opts = opts;
  if (opts.create_rendezvous) {
    Rendezvous* rendezvous = new IntraProcessRendezvous(device_mgr_);
    run_opts.rendezvous = rendezvous;
    run_opts.create_rendezvous = false;
    done = std::bind(
        [rendezvous](DoneCallback done,
                     // Begin unbound arguments.
                     const Status& status) {
          rendezvous->Unref();
          done(status);
        },
        std::move(done), std::placeholders::_1);
  }

  Item* item = nullptr;
  Status s = GetOrCreateItem(handle, &item);
  if (!s.ok()) {
    done(s);
    return;
  }
  if (run_opts.runner == nullptr) {
    run_opts.runner = &default_runner_;
  }
  DCHECK(run_opts.runner != nullptr);

  Executor::Args* exec_args = new Executor::Args;
  // Inherit the step_id from the caller.
  exec_args->step_id = run_opts.step_id;
  exec_args->rendezvous = run_opts.rendezvous;
  exec_args->stats_collector = run_opts.stats_collector;
  exec_args->cancellation_manager = run_opts.cancellation_manager;
  exec_args->collective_executor = run_opts.collective_executor;
  exec_args->step_container = run_opts.step_container;
  exec_args->runner = *run_opts.runner;
  exec_args->call_frame = frame;

  item->exec->RunAsync(
      // Executor args
      *exec_args,
      // Done callback.
      std::bind(
          [item, frame, exec_args](DoneCallback done,
                                   // Start unbound arguments.
                                   const Status& status) {
            delete exec_args;
            done(status);
          },
          std::move(done), std::placeholders::_1));
}

bool FunctionLibraryRuntimeImpl::IsStateful(const string& func) {
  const OpDef* op_def;
  const Status s = base_lib_def_->LookUpOpDef(func, &op_def);
  return s.ok() && op_def->is_stateful();
}

string FunctionLibraryRuntimeImpl::DebugString(Handle handle) {
  Item* item = nullptr;
  Status s = GetOrCreateItem(handle, &item);
  if (s.ok()) {
    return tensorflow::DebugString(item->graph);
  } else {
    return s.ToString();
  }
}

Status FunctionLibraryRuntimeImpl::Clone(
    std::unique_ptr<FunctionLibraryDefinition>* out_lib_def,
    std::unique_ptr<ProcessFunctionLibraryRuntime>* out_pflr,
    FunctionLibraryRuntime** out_flr) {
  TF_RETURN_IF_ERROR(
      parent_->Clone(env_, graph_def_version_, optimizer_.options(),
                     custom_kernel_creator_, out_lib_def, out_pflr));
  *out_flr = (*out_pflr)->GetFLR(device_->name());
  if (out_flr != nullptr) {
    return Status::OK();
  } else {
    return errors::Internal("Cloning FunctionLibraryRuntime failed.");
  }
}

namespace {

struct CustomCreatorSingleton {
  mutex mu;
  CustomKernelCreator custom_creator = nullptr;

  void Set(CustomKernelCreator cb) {
    mutex_lock l(mu);
    custom_creator = std::move(cb);
  }

  CustomKernelCreator Get() {
    mutex_lock l(mu);
    return custom_creator;
  }
};

CustomCreatorSingleton* GetCustomCreatorSingleton() {
  static CustomCreatorSingleton* ccs = new CustomCreatorSingleton;
  return ccs;
}

}  // namespace

void RegisterDefaultCustomKernelCreator(CustomKernelCreator cb) {
  GetCustomCreatorSingleton()->Set(std::move(cb));
}

std::unique_ptr<FunctionLibraryRuntime> NewFunctionLibraryRuntime(
    const DeviceMgr* device_mgr, Env* env, Device* device,
    int graph_def_version, const FunctionLibraryDefinition* lib_def,
    thread::ThreadPool* thread_pool, const OptimizerOptions& optimizer_options,
    CustomKernelCreator custom_kernel_creator,
    ProcessFunctionLibraryRuntime* parent) {
  return std::unique_ptr<FunctionLibraryRuntime>(new FunctionLibraryRuntimeImpl(
      device_mgr, env, device, graph_def_version, lib_def, thread_pool,
      optimizer_options, std::move(custom_kernel_creator), parent));
}

std::unique_ptr<FunctionLibraryRuntime> NewFunctionLibraryRuntime(
    const DeviceMgr* device_mgr, Env* env, Device* device,
    int graph_def_version, const FunctionLibraryDefinition* lib_def,
    thread::ThreadPool* thread_pool, const OptimizerOptions& optimizer_options,
    ProcessFunctionLibraryRuntime* parent) {
  return NewFunctionLibraryRuntime(device_mgr, env, device, graph_def_version,
                                   lib_def, thread_pool, optimizer_options,
                                   GetCustomCreatorSingleton()->Get(), parent);
}

bool RemoveDeadNodes(Graph* g) {
  VLOG(2) << "Removing dead nodes";
  std::unordered_set<const Node*> nodes;
  for (auto n : g->nodes()) {
    if (n->IsSource() || n->IsSink() || n->IsControlFlow() ||
        n->op_def().is_stateful()) {
      nodes.insert(n);
    }
  }
  return PruneForReverseReachability(g, std::move(nodes));
}

namespace {
// If 'edges' contains only 1 non-control edge, returns it. Otherwise,
// returns a nullptr.
const Edge* GetTheOnlyDataEdge(const EdgeSet& edges) {
  const Edge* ret = nullptr;
  for (const Edge* e : edges) {
    if (e->IsControlEdge() || ret) {
      // Don't touch it if there is a control edge.
      return nullptr;
    }
    if (IsRefType(e->src()->output_type(e->src_output()))) {
      // Don't touch it if the identity node is effectively de-reffing
      // a ref.
      return nullptr;
    }
    if (IsRecv(e->src()) || IsSwitch(e->src())) {
      // Don't touch it if the identity is introduced for control flow.
      // Recv disables all its successors if it receives a dead signal.
      // When Recv has an outgoing control edge, the current executor
      // would not disable the destination. The current solution (see
      // graph_partition.cc) is to add an identity after Recv and change
      // the control edge to be from this identity node. So the identity
      // can't be removed.
      return nullptr;
    }
    ret = e;
  }
  return ret;
}
}  // end namespace

bool RemoveIdentityNodes(Graph* g) {
  VLOG(2) << "Removing identity nodes";
  bool removed_any = false;
  gtl::InlinedVector<Node*, 8> matches;
  for (Node* n : g->nodes()) {
    if (!n->IsIdentity()) continue;
    if (!GetTheOnlyDataEdge(n->in_edges())) continue;

    // Some identity nodes are used as sink nodes to give names to output
    // tensors. These nodes are not going to be executed unless they are in the
    // fetch set. But if they are in the fetch set we don't want to remove them.
    if (n->out_edges().empty()) continue;

    matches.push_back(n);
  }
  if (!matches.empty()) {
    for (Node* n : matches) {
      const Edge* in = GetTheOnlyDataEdge(n->in_edges());
      for (const Edge* out : n->out_edges()) {
        if (out->IsControlEdge()) {
          g->AddControlEdge(in->src(), out->dst());
        } else {
          g->AddEdge(in->src(), in->src_output(), out->dst(), out->dst_input());
        }
      }
      VLOG(2) << "Remove Identity: " << n->DebugString();
      g->RemoveNode(n);
      removed_any = true;
    }
  }
  return removed_any;
}

bool RemoveListArrayConverter(Graph* g) {
  VLOG(2) << "Removing list array converter";
  gtl::InlinedVector<Node*, 8> matches;
  for (Node* n : g->nodes()) {
    if ((n->type_string() == "_ListToArray") ||
        (n->type_string() == "_ArrayToList")) {
      matches.push_back(n);
    }
  }
  bool removed_any = false;
  if (!matches.empty()) {
    for (Node* n : matches) {
      if (n->num_inputs() != n->num_outputs()) {
        continue;  // Not expected. Skip.
      }
      gtl::InlinedVector<Node*, 8> identity_nodes(n->num_inputs(), nullptr);

      // Process input edges first.
      Node* input_control_node = nullptr;
      for (const Edge* e : n->in_edges()) {
        if (e->IsControlEdge()) {
          if (input_control_node == nullptr) {
            // If node "n" has any control dependencies, adds a no-op
            // node (input_control_node) which the additional Identity
            // nodes depends on and the input_control_node depends on
            // the node "n"s control dependencies.
            input_control_node = AddNoOp(g);
          }
          g->AddControlEdge(e->src(), input_control_node);
        } else {
          const int index = e->dst_input();
          Node** id_node = &identity_nodes[index];
          if (*id_node != nullptr) {
            LOG(ERROR)
                << "RemoveListArrayConverter unexpected duplicated input: "
                << e->dst_input();
            return removed_any;
          }
          *id_node = AddIdentity(g, {e->src(), e->src_output()});
        }
      }

      // If node "n" has any control dependencies, the added identity
      // nodes should have control dependencies on input_control_node.
      if (input_control_node != nullptr) {
        for (Node* id : identity_nodes) {
          g->AddControlEdge(input_control_node, id);
        }
      }

      Node* output_control_node = nullptr;
      for (const Edge* e : n->out_edges()) {
        if (e->IsControlEdge()) {
          if (output_control_node == nullptr) {
            // If node "n" is control-depended upon by other nodes,
            // adds a no-op node (output_control_node) which those
            // nodes will depend on and output_control_node depends on
            // all Identity nodes.
            output_control_node = AddNoOp(g);
          }
          g->AddControlEdge(output_control_node, e->dst());
        } else {
          Node* id_node = identity_nodes[e->src_output()];
          if (id_node == nullptr) {
            LOG(ERROR) << "RemoveListArrayConverter unexpected missing input: "
                       << e->src_output();
            return removed_any;
          }
          CHECK(id_node);
          g->AddEdge(id_node, 0, e->dst(), e->dst_input());
        }
      }

      // If any nodes have control dependencies on node "n", those
      // nodes should have control dependencies on
      // output_control_node.
      if (output_control_node != nullptr) {
        for (Node* id : identity_nodes) {
          g->AddControlEdge(id, output_control_node);
        }
      }

      g->RemoveNode(n);
      removed_any = true;
    }
  }
  return removed_any;
}

// Returns true iff the function '*fbody' can be inlined at 'node'
// based on the type signature of 'node' and 'fbody'.
static bool ValidateInlining(const Node* node, const FunctionBody* fbody) {
  if (static_cast<size_t>(node->num_inputs()) != fbody->arg_types.size()) {
    return false;
  }
  if (static_cast<size_t>(node->num_inputs()) != fbody->arg_nodes.size()) {
    return false;
  }
  if (static_cast<size_t>(node->num_outputs()) != fbody->ret_types.size()) {
    return false;
  }
  if (static_cast<size_t>(node->num_outputs()) != fbody->ret_nodes.size()) {
    return false;
  }
  for (int i = 0; i < node->num_inputs(); ++i) {
    if (node->input_type(i) != fbody->arg_types[i]) return false;
  }
  for (int i = 0; i < node->num_outputs(); ++i) {
    if (node->output_type(i) != fbody->ret_types[i]) return false;
  }
  return true;
}

// Given a "caller" in graph "g", which is a function call of a function
// to "fbody". Replaces the "caller" with fbody->graph and connects
// edges properly. "override_device" specifies whether inlining should replace
// explicitly specified devices inside fbody with the callee's device.
void InlineFunctionBody(const FunctionLibraryDefinition& flib_def, Graph* g,
                        Node* caller, const FunctionBody* fbody,
                        bool override_device) {
  if (!ValidateInlining(caller, fbody)) {
    LOG(WARNING) << "Inlining mismatch: " << caller->DebugString() << " vs. "
                 << DebugString(fbody->graph);
    return;
  }

  // Input edges. For data edges coming into "caller", we first compute the
  // <src>:<src_output> for the i-th input in "inputs".
  // If "caller" has any input control dependencies, we add a NoOp
  // node "input_control_node", which depends on "caller"'s control inputs.
  std::vector<Endpoint> inputs(caller->num_inputs());
  Node* input_control_node = nullptr;
  for (const Edge* e : caller->in_edges()) {
    if (e->IsControlEdge()) {
      if (input_control_node == nullptr) {
        input_control_node = AddNoOp(g);
      }
      g->AddControlEdge(e->src(), input_control_node);
    } else {
      inputs[e->dst_input()] = {e->src(), e->src_output()};
    }
  }

  // Duplicate fbody->graph into 'g'.  First, we copy the nodes of
  // fbody->graph into 'g' except the source and sink nodes.  We copy
  // edges among nodes in 'fbody->graph'.
  //
  // If 'x' is a node in fbody->graph and its copy in 'g' is 'y', we
  // remember 'y' in node_map[x->id()].
  std::vector<Node*> node_map(fbody->graph->num_node_ids());
  Status s;
  for (Node* n : fbody->graph->op_nodes()) {
    NodeDef ndef = n->def();
    ndef.set_name(strings::StrCat(caller->name(), "/", ndef.name()));
    if (override_device || ndef.device().empty()) {
      ndef.set_device(caller->def().device());
    }
    Node* clone = g->AddNode(ndef, &s);
    TF_CHECK_OK(s);
    node_map[n->id()] = clone;

    // If there is an input control node, and one of:
    // a) the node has no data or control inputs, or
    // b) the node is a function call or SymbolicGradient,
    // then add a control edge from the input control node to the clone.
    //
    // We must not execute any nodes if the original function call would not
    // have executed. This is especially critical when the function call is
    // inside a control-flow construct like tf.cond(). Case (a) ensures that
    // such nodes do not run.
    //
    // The purpose of case (b) is to ensure that instances of case (a) created
    // by further inlining steps also receive the control dependency.
    if (input_control_node) {
      bool has_inputs = false;
      for (const Edge* e : n->in_edges()) {
        if (!e->src()->IsSource()) {
          has_inputs = true;
          break;
        }
      }
      if (!has_inputs || flib_def.Find(clone->type_string()) != nullptr ||
          clone->type_string() == "SymbolicGradient") {
        g->AddControlEdge(input_control_node, clone);
      }
    }
  }
  for (const Edge* e : fbody->graph->edges()) {
    if (e->src()->IsSource() || e->src()->IsSink() || e->dst()->IsSource() ||
        e->dst()->IsSink()) {
      continue;
    }
    Node* src_copy = node_map[e->src()->id()];
    Node* dst_copy = node_map[e->dst()->id()];
    g->AddEdge(src_copy, e->src_output(), dst_copy, e->dst_input());
  }

  // Connect input edges.
  //
  // We create one Identity node for each input. Then, we connect inputs[i] to
  // the i-th identity node added. The nodes that previously connected
  // to the j-th output of i-th arg node are reconnected to the i-th
  // identity node.
  //
  // The added identity nodes depend on "input_control_node".
  for (std::size_t i = 0; i < fbody->arg_nodes.size(); ++i) {
    Node* arg = node_map[fbody->arg_nodes[i]->id()];
    Node* n = AddIdentity(g, inputs[i]);
    if (input_control_node) {
      g->AddControlEdge(input_control_node, n);
    }
    for (const Edge* e : arg->out_edges()) {
      if (e->IsControlEdge()) {
        g->AddControlEdge(n, e->dst());
      } else {
        g->AddEdge(n, 0, e->dst(), e->dst_input());
      }
    }
    node_map[fbody->arg_nodes[i]->id()] = n;
    g->RemoveNode(arg);  // 'arg' is disconnected.
  }

  // Connect output edges.
  //
  // For i-th return node in fbody->graph, we add in "g" an identity
  // node (outputs[i-th]). We then reconnect every incoming edge into
  // the i-th return node to the added identity node.
  //
  // For every data edge coming out of "callee"s i-th output, we
  // reconnect it to the i-th identity added above.
  //
  // If "callee" is control-depended upon by any other nodes, we add a
  // NoOp node "output_control_node". "output_control_node" depends on
  // all identity nodes added above. And nodes previously depend on
  // "callee" is changed to depend on "output_control_node".
  std::vector<Node*> outputs(caller->num_outputs());
  for (std::size_t i = 0; i < fbody->ret_nodes.size(); ++i) {
    Node* ret = node_map[fbody->ret_nodes[i]->id()];
    Endpoint data;  // Data input for the ret node.
    for (const Edge* e : ret->in_edges()) {
      if (!e->IsControlEdge()) {
        data = {e->src(), e->src_output()};
        break;
      }
    }
    CHECK(data.node != nullptr);
    Node* n = AddIdentity(g, data);
    outputs[i] = n;
    for (const Edge* e : ret->in_edges()) {
      if (e->IsControlEdge()) {
        g->AddControlEdge(e->src(), n);
      }
    }
    g->RemoveNode(ret);  // 'ret' is disconnected.
  }
  Node* output_control_node = nullptr;
  for (const Edge* e : caller->out_edges()) {
    if (e->IsControlEdge()) {
      if (output_control_node == nullptr) {
        output_control_node = AddNoOp(g);
        for (Node* n : outputs) {
          g->AddControlEdge(n, output_control_node);
        }
      }
      g->AddControlEdge(output_control_node, e->dst());
    } else {
      g->AddEdge(outputs[e->src_output()], 0, e->dst(), e->dst_input());
    }
  }
  g->RemoveNode(caller);  // 'caller' is replaced with inlined nodes.
}

bool ExpandInlineFunctions(FunctionLibraryRuntime* lib, Graph* graph) {
  std::vector<std::pair<Node*, const FunctionBody*>> candidates;
  const FunctionLibraryDefinition* fld = lib->GetFunctionLibraryDefinition();
  for (Node* node : graph->nodes()) {
    VLOG(3) << "Expanding " << node->DebugString();
    bool noinline;
    if (fld->GetAttr(*node, kNoInlineAttr, &noinline).ok() && noinline) {
      VLOG(3) << "noinline: " << node->DebugString();
      continue;
    }
    FunctionLibraryRuntime::Handle handle;
    Status s = lib->Instantiate(node->type_string(), node->attrs(), &handle);
    if (!s.ok()) {
      // Either "node" is a primitive op, or the instantiation failed.
      if (errors::IsNotFound(s)) {
        VLOG(3) << "ExpandInlineFunctions " << s;
      } else {
        LOG(ERROR) << "ExpandInlineFunctions " << s;
      }
      continue;
    }
    const FunctionBody* fbody = lib->GetFunctionBody(handle);
    CHECK_NOTNULL(fbody);
    candidates.push_back({node, fbody});
  }
  for (const auto& p : candidates) {
    InlineFunctionBody(*fld, graph, p.first, p.second);
  }
  return !candidates.empty();
}

string NewName(const Node* n, bool pretty) {
  if (pretty) {
    return strings::StrCat(n->type_string(), n->id());
  } else {
    return strings::StrCat("n", n->id());
  }
}

// TODO(zhifengc): Maybe this should be the default Graph::AsGraphDef.
// and stash the original NodeDef name as an attr for documentation
// purpose.
void ToGraphDef(const Graph* g, GraphDef* gdef, bool pretty) {
  // We visit nodes in forward topological sort order, which is a
  // possible execution order of the graph.
  gtl::InlinedVector<const Edge*, 4> inputs;
  gdef->Clear();
  gdef->mutable_versions()->CopyFrom(g->versions());

  std::vector<Node*> start_nodes;
  for (Node* n : g->nodes()) {
    if (n->out_edges().empty()) {
      start_nodes.push_back(n);
    }
  }

  ReverseDFSFrom(*g, start_nodes, nullptr, [gdef, pretty, &inputs](Node* n) {
    if (!n->IsOp()) return;
    NodeDef* ndef = gdef->add_node();
    ndef->set_name(NewName(n, pretty));
    ndef->set_op(n->type_string());
    for (const auto& attr : n->attrs()) {
      (*ndef->mutable_attr())[attr.first] = attr.second;
    }
    inputs.clear();
    inputs.resize(n->num_inputs());
    for (const Edge* e : n->in_edges()) {
      if (e->IsControlEdge()) {
        inputs.push_back(e);
      } else {
        if (inputs[e->dst_input()] == nullptr) {
          inputs[e->dst_input()] = e;
        } else {
          LOG(WARNING) << "Malformed graph node. multiple input edges: "
                       << n->DebugString();
        }
      }
    }
    // node->name() is merely NodeDef::name, which are not guaranteed
    // to be unique and stable after optimization rewrites. Therefore,
    // we use "n<node id>" instead.
    for (const Edge* e : inputs) {
      if (e == nullptr) {
        ndef->add_input("unknown");
        continue;
      }
      const string srcname = NewName(e->src(), pretty);
      if (!e->src()->IsOp()) {
      } else if (e->IsControlEdge()) {
        ndef->add_input(strings::StrCat("^", srcname));
      } else if (e->src_output() == 0) {
        ndef->add_input(srcname);
      } else {
        ndef->add_input(strings::StrCat(srcname, ":", e->src_output()));
      }
    }
  });
}

string DebugString(const Graph* g) {
  GraphDef gdef;
  ToGraphDef(g, &gdef);
  return DebugString(gdef);
}

FunctionBody::FunctionBody(const FunctionDef& f, DataTypeSlice arg_t,
                           DataTypeSlice ret_t, Graph* g)
    : fdef(f),
      graph(g),
      arg_types(arg_t.begin(), arg_t.end()),
      ret_types(ret_t.begin(), ret_t.end()) {
  this->arg_nodes.resize(arg_types.size());
  this->ret_nodes.resize(ret_types.size());
  for (Node* n : this->graph->op_nodes()) {
    gtl::InlinedVector<Node*, 4>* node_vec;
    if (n->type_string() == kRetOp) {
      node_vec = &this->ret_nodes;
    } else if (n->type_string() == kArgOp) {
      node_vec = &this->arg_nodes;
    } else {
      continue;
    }
    int index;
    TF_CHECK_OK(GetNodeAttr(n->attrs(), "index", &index));
    CHECK_LE(0, index);
    CHECK_LT(index, node_vec->size());
    (*node_vec)[index] = n;
  }
}

FunctionBody::~FunctionBody() { delete this->graph; }

class SymbolicGradientHelper {
 public:
  explicit SymbolicGradientHelper(const FunctionBody& f) : fbody_(&f) {}

  ~SymbolicGradientHelper() { delete gbody_; }

  FunctionBody* Compute();

 private:
  const FunctionBody* fbody_;
  FunctionBody* gbody_ = nullptr;

  // Makes a copy of fbody_ in gbody_.
  void Copy();

  TF_DISALLOW_COPY_AND_ASSIGN(SymbolicGradientHelper);
};

void SymbolicGradientHelper::Copy() {
  const Graph& src = *(fbody_->graph);
  gbody_->graph = new Graph(src.op_registry());
  Graph* dst = gbody_->graph;

  std::vector<Node*> node_map(src.num_node_ids());

  // Copy the nodes.
  node_map[src.source_node()->id()] = dst->source_node();
  node_map[src.sink_node()->id()] = dst->sink_node();
  for (Node* n : src.op_nodes()) {
    node_map[n->id()] = dst->CopyNode(n);
  }

  // Copy the edges.
  for (const Edge* e : src.edges()) {
    Node* src_copy = node_map[e->src()->id()];
    Node* dst_copy = node_map[e->dst()->id()];
    dst->AddEdge(src_copy, e->src_output(), dst_copy, e->dst_input());
  }

  // Save inputs in copied graph.
  CHECK_EQ(fbody_->arg_types.size(), fbody_->arg_nodes.size());
  gbody_->arg_types = fbody_->arg_types;
  for (std::size_t i = 0; i < fbody_->arg_nodes.size(); ++i) {
    gbody_->arg_nodes.push_back(node_map[fbody_->arg_nodes[i]->id()]);
  }

  // Save outputs in copied graph.
  CHECK_EQ(fbody_->ret_types.size(), fbody_->ret_nodes.size());
  gbody_->ret_types = fbody_->ret_types;
  for (std::size_t i = 0; i < fbody_->ret_nodes.size(); ++i) {
    gbody_->ret_nodes.push_back(node_map[fbody_->ret_nodes[i]->id()]);
  }
}

FunctionBody* SymbolicGradientHelper::Compute() {
  CHECK(gbody_ == nullptr);
  gbody_ = new FunctionBody;

  // Copy fbody_ into gbody_.
  Copy();

  Graph* g = gbody_->graph;

  const int num_y = static_cast<int>(gbody_->ret_nodes.size());

  // Populate 'y_node_outputs_' with node function body outputs.
  // Populate 'y_grad_nodes' with initial gradient nodes for each return node of
  // the original function body (these will be 'arg' nodes in the function
  // gradient body).
  std::vector<NodeOut> y_node_outputs;
  y_node_outputs.reserve(num_y);
  std::vector<NodeOut> y_grad_node_outputs;
  y_grad_node_outputs.reserve(num_y);
  for (int i = 0; i < num_y; ++i) {
    Node* y = gbody_->ret_nodes[i];
    y_node_outputs.push_back({y, 0});
    DCHECK_EQ(y->type_string(), kRetOp);
    const DataType dtype = y->input_type(0);
    const int index = static_cast<int>(gbody_->arg_nodes.size());
    Node* dy = AddArg(g, dtype, index);
    gbody_->arg_types.push_back(dtype);
    gbody_->arg_nodes.push_back(dy);
    y_grad_node_outputs.push_back({dy, 0});
  }

  // Populate 'x_nodes' with function args (excluding 'y_grad_node_outputs').
  const size_t num_x = fbody_->arg_nodes.size();
  std::vector<NodeOut> x_node_outputs;
  x_node_outputs.reserve(num_x);
  for (size_t i = 0; i < fbody_->arg_nodes.size(); ++i) {
    x_node_outputs.push_back({gbody_->arg_nodes[i], 0});
  }

  // Call AddSymbolicGradients which will add nodes to graph 'g' that
  // compute the function gradient (adding an entry in 'x_grad_node_outputs' for
  // each node in 'x_node_outputs').
  std::vector<NodeOut> x_grad_node_outputs;
  TF_CHECK_OK(AddSymbolicGradients(y_node_outputs, x_node_outputs,
                                   y_grad_node_outputs, &x_grad_node_outputs,
                                   g));

  // Remove the old return nodes from the function body.
  for (Node* n : gbody_->ret_nodes) {
    g->RemoveNode(n);
  }
  gbody_->ret_types = fbody_->arg_types;
  // TODO(apassos): use the right dtype for gradients of  resource variables
  for (int i = 0; i < gbody_->ret_types.size(); ++i) {
    if (gbody_->ret_types[i] == DT_RESOURCE) {
      gbody_->ret_types[i] = DT_FLOAT;
    }
  }
  gbody_->ret_nodes.clear();
  // Add new return nodes to the function gradient body for each node
  // in 'x_grad_nodes'.
  const int arg_types_size = static_cast<int>(fbody_->arg_types.size());
  for (int i = 0; i < arg_types_size; ++i) {
    Endpoint grad = {x_grad_node_outputs[i].node, x_grad_node_outputs[i].index};
    Node* ret = AddRet(g, grad, i);
    gbody_->ret_nodes.push_back(ret);
  }

  auto ret = gbody_;
  gbody_ = nullptr;
  return ret;
}

FunctionBody* SymbolicGradient(const FunctionBody& f) {
  return SymbolicGradientHelper(f).Compute();
}

Status FunctionDefToBodyHelper(
    const FunctionDef& fdef, const AttrSlice& attrs,
    const FunctionLibraryDefinition* const lib_def,
    const std::function<Status(const string&, const OpDef**)>& get_func_sig,
    FunctionBody** fbody) {
  // Instantiates the function template into a graph def.
  InstantiationResult result;
  TF_RETURN_IF_ERROR(InstantiateFunction(fdef, attrs, get_func_sig, &result));

  std::unique_ptr<Graph> graph(new Graph(lib_def));
  GraphConstructorOptions opts;
  opts.allow_internal_ops = true;
  opts.expect_device_spec = false;
  TF_RETURN_IF_ERROR(ConvertNodeDefsToGraph(opts, result.nodes, graph.get()));

  // Call BuildControlFlowInfo to validate that this function body has
  // well-formed control flow.
  std::vector<ControlFlowInfo> dummy;
  TF_RETURN_IF_ERROR(BuildControlFlowInfo(graph.get(), &dummy));

  *fbody = new FunctionBody(fdef, result.arg_types, result.ret_types,
                            graph.release());
  return Status::OK();
}

}  // end namespace tensorflow


text:
<?xml version='1.0'?>
<methodResponse>
<params>
<param>
<value><struct>
<member>
<name>jobguid</name>
<value><nil/></value></member>
<member>
<name>result</name>
<value><array><data>
<value><string>8d6a8b91-22fc-4be4-863b-11dc4456b315</string></value>
</data></array></value>
</member>
</struct></value>
</param>
</params>
</methodResponse>

redpajama_set_name: RedPajamaGithub


text:
package org.assertj.core.error;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.error.ElementsShouldBe.elementsShouldBe;
import static org.assertj.core.util.Lists.newArrayList;

import org.assertj.core.api.TestCondition;
import org.assertj.core.description.Description;
import org.assertj.core.description.TextDescription;
import org.assertj.core.presentation.StandardRepresentation;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

/**
 * Tests for <code>{@link ElementsShouldBe#create(Description)}</code>.
 *
 * @author Nicolas François
 * @author Joel Costigliola
 */
public class ElementsShouldBe_create_Test {

  private ErrorMessageFactory factory;

  @BeforeEach
  public void setUp() {
    factory = elementsShouldBe(newArrayList("Yoda", "Luke", "Leia"), newArrayList("Leia"), new TestCondition<String>("a Jedi"));
  }

  @Test
  public void should_create_error_message() {
    String message = factory.create(new TextDescription("Test"), new StandardRepresentation());
    assertThat(message).isEqualTo(String.format(
        "[Test] %nExpecting elements:%n<[\"Leia\"]>%n of %n<[\"Yoda\", \"Luke\", \"Leia\"]>%n to be <a Jedi>"
    ));
  }

}


text:
#include <AP_HAL.h>
extern const AP_HAL::HAL& hal;

#include <GCS_MAVLink.h>
#include <GCS_Console.h>
#include "simplegcs.h"

extern mavlink_channel_t downstream_channel;

static volatile uint8_t lock = 0;

void simplegcs_send_console_async(uint32_t machtnichts) {
  if (lock) return;
  lock = 1;
  gcs_console_send(downstream_channel);
  lock = 0;
}

void simplegcs_send_heartbeat_async(uint32_t us) {
  uint32_t ms = us / 1000;
  static uint32_t last_ms = 0;
  if (ms - last_ms < 1000) return;
  if (lock) return;
  last_ms = ms;
  lock = 1;
  {
    uint8_t base_mode = MAV_MODE_FLAG_STABILIZE_ENABLED
                      | MAV_MODE_FLAG_GUIDED_ENABLED
                      | MAV_MODE_FLAG_SAFETY_ARMED
                      | MAV_MODE_FLAG_CUSTOM_MODE_ENABLED;
    uint8_t system_status = MAV_STATE_ACTIVE;
    uint32_t custom_mode = 5 ; /* Loiter is mode 5. */
    mavlink_msg_heartbeat_send(
            downstream_channel,
            MAV_TYPE_QUADROTOR,
            MAV_AUTOPILOT_ARDUPILOTMEGA,
            base_mode,
            custom_mode,
            system_status);
  }
  lock = 0;
}

void simplegcs_send_heartbeat(mavlink_channel_t chan) {
    uint8_t base_mode = MAV_MODE_FLAG_STABILIZE_ENABLED
                      | MAV_MODE_FLAG_GUIDED_ENABLED
                      | MAV_MODE_FLAG_SAFETY_ARMED
                      | MAV_MODE_FLAG_CUSTOM_MODE_ENABLED;
    uint8_t system_status = MAV_STATE_ACTIVE;
    uint32_t custom_mode = 5 ; /* Loiter is mode 5. */

    while(lock);
    lock = 1;
    mavlink_msg_heartbeat_send(
            chan,
            MAV_TYPE_QUADROTOR,
            MAV_AUTOPILOT_ARDUPILOTMEGA,
            base_mode,
            custom_mode,
            system_status);
    lock = 0;
}

bool simplegcs_try_send_statustext(mavlink_channel_t chan, const char *text, int len) {

    uint16_t txspace = comm_get_txspace(chan);
    if (payload_space < MAVLINK_MSG_ID_STATUSTEXT_LEN+MAVLINK_NUM_NON_PAYLOAD_BYTES) return false;

    char statustext[50] = { 0 };
    if (len < 50) {
        memcpy(statustext, text, len);
    }
    while(lock);
    lock = 1;
    mavlink_msg_statustext_send(
            chan,
            1, /* SEVERITY_LOW */
            statustext);
    lock = 0;
    return true;
}
// -----------------------------------------------------------------------

void simplegcs_update(mavlink_channel_t chan, simplegcs_handler_t handler) {
    mavlink_message_t msg;
    mavlink_status_t status;
    while(comm_get_available(chan)){
        uint8_t c = comm_receive_ch(chan);
        bool newmsg = mavlink_parse_char(chan, c, &msg, &status);
        if (newmsg) {
            handler(chan, &msg);
        }
    }
}


text:
from azure.identity import DefaultAzureCredential
from azure.mgmt.rdbms import MySQLManagementClient

"""
# PREREQUISITES
    pip install azure-identity
    pip install azure-mgmt-rdbms
# USAGE
    python create_a_replica_server.py

    Before run the sample, please set the values of the client ID, tenant ID and client secret
    of the AAD application as environment variables: AZURE_CLIENT_ID, AZURE_TENANT_ID,
    AZURE_CLIENT_SECRET. For more info about how to get the value, please see:
    https://docs.microsoft.com/azure/active-directory/develop/howto-create-service-principal-portal
"""

def main():
    client = MySQLManagementClient(
        credential=DefaultAzureCredential(),
        subscription_id="ffffffff-ffff-ffff-ffff-ffffffffffff",
    )

    response = client.servers.begin_create(
        resource_group_name="TargetResourceGroup",
        server_name="targetserver",
        parameters={
            "location": "westus",
            "properties": {
                "createMode": "Replica",
                "sourceServerId": "/subscriptions/ffffffff-ffff-ffff-ffff-ffffffffffff/resourceGroups/MasterResourceGroup/providers/Microsoft.DBforMySQL/servers/masterserver",
            },
        },
    ).result()
    print(response)

# x-ms-original-file: specification/mysql/resource-manager/Microsoft.DBforMySQL/stable/2017-12-01/examples/ServerCreateReplicaMode.json
if __name__ == "__main__":
    main()


text:
using System.Collections.Generic;
using System.IO;

namespace IniParser.Parser
{
    public sealed class StringBuffer
    {
        public struct Range
        {
            int _start, _size;

            public int start
            {
                get
                {
                    return _start;
                }

                set
                {
                    _start = value < 0 ? 0 : value;
                }
            }

            public int size
            {
                get
                {
                    return _size;
                }

                set
                {
                    _size = value < 0 ? 0 : value;
                }
            }

            public int end
            {
                get
                {
                    return _size <= 0 ? 0 : _start + (_size - 1);
                }
            }

            public bool IsEmpty { get { return _size == 0; } }

            public void Reset()
            {
                start = 0;
                size = 0;
            }

            public static Range FromIndexWithSize(int start, int size)
            {
                if (start < 0 || size <= 0) return new Range();

                return new Range { start = start, size = size };
            }

            public static Range WithIndexes(int start, int end)
            {
                if (start < 0 || end < 0 || end - start < 0)

                {
                    return new Range();
                }

                return new Range { start = start, size = end - start + 1 };
            }

            public override string ToString()
            {
                return string.Format("[start:{0}, end:{1} size: {2}]",
                                     start,
                                     end,
                                     size);
            }
        }

        readonly static int DefaultCapacity = 256;

        public StringBuffer()
            :this(StringBuffer.DefaultCapacity
                 )
        {

        }

        public StringBuffer(int capacity)
        {
            _buffer = new List<char>(capacity);
        }

        internal StringBuffer(List<char> buffer, Range bufferIndexes)
        {
            _buffer = buffer;
            _bufferIndexes = bufferIndexes;
        }

        public int Count { get { return _bufferIndexes.size; } }

        public bool IsEmpty
        {
            get { return _bufferIndexes.IsEmpty; }
        }

        public bool IsWhitespace
        {
            get
            {
                int startIdx = _bufferIndexes.start;
                while (startIdx <= _bufferIndexes.end
                    && char.IsWhiteSpace(_buffer[startIdx]))
                {
                    startIdx++;
                }

                return startIdx > _bufferIndexes.end;
            }
        }

        public char this[int idx]
        {
            get
            {
                return _buffer[idx + _bufferIndexes.start];
            }
        }

        public StringBuffer DiscardChanges()
        {
            _bufferIndexes = Range.FromIndexWithSize(0, _buffer.Count);
            return this;
        }

        public Range FindSubstring(string subString, int startingIndex = 0)
        {
            int subStringLength = subString.Length;

            if (subStringLength <= 0 || Count < subStringLength)
            {
                return new Range();
            }

            startingIndex += _bufferIndexes.start;

            // Search the first char of the substring
            for (int firstCharIdx = startingIndex; firstCharIdx <= _bufferIndexes.end; ++firstCharIdx)
            {
                if (_buffer[firstCharIdx] != subString[0])
                {
                    continue;
                }

                // Fail now if the substring can't fit given the size of the
                // buffer and the search start index
                if (firstCharIdx + subStringLength - 1 > _bufferIndexes.end)
                {
                    return new Range();
                }

                bool isSubstringMismatch = false;
                // Check if the substring matches starting at the index
                for (int currentIdx = 0; currentIdx < subStringLength; ++currentIdx)
                {
                    if (_buffer[firstCharIdx + currentIdx] != subString[currentIdx])
                    {
                        isSubstringMismatch = true;
                        break;
                    }
                }

                if (isSubstringMismatch)
                {
                    continue;
                }

                 return Range.FromIndexWithSize(firstCharIdx - _bufferIndexes.start, subStringLength);
            }

            return new Range();
        }

        public bool ReadLine()
        {
            if (_dataSource == null) return false;

            _buffer.Clear();
            int c = _dataSource.Read();

            // Read until new line ('\n') or EOF (-1)
            while (c != '\n' && c != -1)
            {
                if (c != '\r')
                {
                    _buffer.Add((char)c);
                }

                c = _dataSource.Read();
            }

            _bufferIndexes = Range.FromIndexWithSize(0, _buffer.Count);

            return _buffer.Count > 0 || c != -1;
        }

        public void Reset(TextReader dataSource)
        {
            _dataSource = dataSource;
            _bufferIndexes.Reset();
            _buffer.Clear();
        }

        public void Resize(Range range)
        {
            Resize(range.start, range.size);
        }

        public void Resize(int newSize)
        {
            Resize(0, newSize);
        }

        public void Resize(int startIdx, int size)
        {
            if (startIdx < 0 || size < 0) return;

            var internalStartIdx = _bufferIndexes.start + startIdx;
            var internalEndIdx = internalStartIdx + size - 1;

            if (internalEndIdx > _bufferIndexes.end) return;

            _bufferIndexes.start = internalStartIdx;
            _bufferIndexes.size = size;
        }

        public void ResizeBetweenIndexes(int startIdx, int endIdx)
        {
            Resize(startIdx, endIdx - startIdx + 1);
        }

        public StringBuffer Substring(Range range)
        {
            var copy = SwallowCopy();
            copy.Resize(range);
            return copy;
        }

        public StringBuffer SwallowCopy()
        {
            return new StringBuffer(_buffer, _bufferIndexes);
        }

        public void TrimStart()
        {
            if (IsEmpty) return;

            int startIdx = _bufferIndexes.start;
            while (startIdx <= _bufferIndexes.end
                && char.IsWhiteSpace(_buffer[startIdx]))
            {
                startIdx++;
            }

            // We need to make a copy of this value because _bufferIndexes.end
            // is a computed property, so it will change if we modify
            // _bufferIndexes.start or _bufferIndexes.size
            int endIdx = _bufferIndexes.end;

            _bufferIndexes.start = startIdx;
            _bufferIndexes.size = endIdx - startIdx + 1;
        }

        public void TrimEnd()
        {
            if (IsEmpty) return;

            int endIdx = _bufferIndexes.end;

            while (endIdx >= _bufferIndexes.start
                && char.IsWhiteSpace(_buffer[endIdx]))
            {
                endIdx--;
            }

            _bufferIndexes.size = endIdx - _bufferIndexes.start + 1;
        }
        public void Trim()
        {
            TrimEnd();
            TrimStart();
        }

        public bool StartsWith(string str)
        {
            if (string.IsNullOrEmpty(str)) return false;
            if (IsEmpty) return false;

            int strIdx = 0;
            int bufferIdx = _bufferIndexes.start;

            for (; strIdx < str.Length; ++strIdx, ++bufferIdx)
            {
                if (str[strIdx] != _buffer[bufferIdx]) return false;
            }

            return true;
        }

        public override string ToString()
        {
            return new string(_buffer.ToArray(),
                               _bufferIndexes.start,
                               _bufferIndexes.size);
        }

        public string ToString(Range range)
        {
            if (range.IsEmpty
             || range.start < 0
             || range.size > _bufferIndexes.size
             || range.start + _bufferIndexes.start > _bufferIndexes.end)
            {
                return string.Empty;
            }

            return new string(_buffer.ToArray(),
                              _bufferIndexes.start + range.start,
                              range.size);
        }

        TextReader _dataSource;
        List<char> _buffer;
        Range _bufferIndexes;
    }
}


text:
namespace Ra {
namespace Core {

namespace _internalIterator {
template <class T>
struct _reversed {
    T& t;
    explicit _reversed( T& _t ) : t( _t ) {}
};
template <class T>
struct _creversed {
    const T& t;
    explicit _creversed( const T& _t ) : t( _t ) {}
};
} // namespace _internalIterator

// Provide reverse iterators for range loops and std::containers:
// \code for (auto x: reversed(c)) ; \endcode
// source: https://stackoverflow.com/a/21510185
template <class T>
_internalIterator::_reversed<T> reversed( T& t ) {
    return _internalIterator::_reversed<T>( t );
}
template <class T>
_internalIterator::_reversed<T const> reversed( T const& t ) {
    return _internalIterator::_reversed<T const>( t );
}
} // namespace Core
} // namespace Ra

namespace std {
template <class T>
auto begin( Ra::Core::_internalIterator::_reversed<T>& r ) -> decltype( r.t.rbegin() ) {
    return r.t.rbegin();
}
template <class T>
auto end( Ra::Core::_internalIterator::_reversed<T>& r ) -> decltype( r.t.rend() ) {
    return r.t.rend();
}

template <class T>
auto begin( Ra::Core::_internalIterator::_creversed<T> const& cr ) -> decltype( cr.t.rbegin() ) {
    return cr.t.rbegin();
}
template <class T>
auto end( Ra::Core::_internalIterator::_creversed<T> const& cr ) -> decltype( cr.t.rend() ) {
    return cr.t.rend();
}
} // namespace std


text:
package org.fairdom.openseekapi;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import ch.ethz.sis.openbis.generic.asapi.v3.IApplicationServerApi;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.common.search.AbstractEntitySearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.common.search.SearchResult;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.dataset.DataSet;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.dataset.DataSetType;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.dataset.fetchoptions.DataSetFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.dataset.fetchoptions.DataSetTypeFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.dataset.search.DataSetSearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.dataset.search.DataSetTypeSearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.experiment.Experiment;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.experiment.ExperimentType;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.experiment.fetchoptions.ExperimentFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.experiment.fetchoptions.ExperimentTypeFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.experiment.search.ExperimentSearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.experiment.search.ExperimentTypeSearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.sample.Sample;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.sample.SampleType;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.sample.fetchoptions.SampleFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.sample.fetchoptions.SampleTypeFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.sample.search.SampleSearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.sample.search.SampleTypeSearchCriteria;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.space.Space;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.space.fetchoptions.SpaceFetchOptions;
import ch.ethz.sis.openbis.generic.asapi.v3.dto.space.search.SpaceSearchCriteria;
import ch.systemsx.cisd.common.spring.HttpInvokerUtils;
import java.util.Map;

import org.fairdom.openseekapi.facility.InvalidOptionException;
import org.fairdom.openseekapi.facility.SslCertificateHelper;
import org.fairdom.openseekapi.general.AuthenticationException;
import org.fairdom.openseekapi.general.QueryType;
import org.json.simple.JSONObject;

/**
 * @author Quyen Nguyen
 * @author Stuart Owen
 */
public class ApplicationServerQuery {
	private static IApplicationServerApi as;
	private static String endpoint;
	private static String sessionToken;

        static final int TIMEOUT = 500000;

	public static IApplicationServerApi as(String endpoint) {
            //SslCertificateHelper.trustAnyCertificate(endpoint);
            SslCertificateHelper.addTrustedUrl(endpoint);

            return HttpInvokerUtils.createServiceStub(IApplicationServerApi.class,
				endpoint + IApplicationServerApi.SERVICE_URL, TIMEOUT);

	}

	public ApplicationServerQuery(String startEndpoint, String startSessionToken) {
		endpoint = startEndpoint;
		sessionToken = startSessionToken;
		as = ApplicationServerQuery.as(endpoint);
	}

	public List<DataSet> allDatasets() throws InvalidOptionException {
            return dataSetsByAttribute("permId", "");
	}

	public List<DataSet> datasetsByAnyField(String searchTerm) {
		DataSetFetchOptions options = dataSetFetchOptions();

		DataSetSearchCriteria criterion = new DataSetSearchCriteria();
		criterion.withAnyField().thatContains(searchTerm);
		return as.searchDataSets(sessionToken, criterion, options).getObjects();
	}

	public List<DataSet> dataSetsByAttribute(String attribute, List<String> values) throws InvalidOptionException {
		DataSetFetchOptions options = dataSetFetchOptions();

		DataSetSearchCriteria criterion = new DataSetSearchCriteria();
		criterion.withOrOperator();
		updateCriterianForAttribute(criterion, attribute, values);

		return as.searchDataSets(sessionToken, criterion, options).getObjects();

	}

	public List<DataSet> dataSetsByAttribute(String attribute, String value) throws InvalidOptionException {
		List<String> values = Arrays.asList(new String[] { value });
		return dataSetsByAttribute(attribute, values);
	}

	public List<DataSet> dataSetsByProperty(String property, String propertyValue) {
		DataSetSearchCriteria criterion = new DataSetSearchCriteria();
		criterion.withProperty(property).thatContains(propertyValue);

		DataSetFetchOptions options = dataSetFetchOptions();

		return as.searchDataSets(sessionToken, criterion, options).getObjects();
	}

	protected List<DataSet> dataSetsByCriteria(DataSetSearchCriteria criteria, DataSetFetchOptions fetchOptions) {

		return as.searchDataSets(sessionToken, criteria, fetchOptions).getObjects();
	}

        @SuppressWarnings("unchecked")
        public List<DataSet> dataSetsByType(JSONObject query) throws InvalidOptionException {

            if (!query.containsKey("typeCode") && !query.containsKey("typeCodes"))
                throw new InvalidOptionException("Missing type code(s)");

            DataSetFetchOptions options = dataSetFetchOptions();
            DataSetSearchCriteria criterion = new DataSetSearchCriteria();

            if (query.containsKey("typeCodes")) {
                List<String> codes = Arrays.asList(query.get("typeCodes").toString().split(","));
                criterion.withType().withCodes().thatIn(codes);
            } else {
                String typeCode = (String)query.get("typeCode");
                criterion.withType().withCode().thatEquals(typeCode);
            }

            return as.searchDataSets(sessionToken, criterion, options).getObjects();
        }

	public List<Experiment> allExperiments() throws InvalidOptionException  {

            return experimentsByAttribute("permId", "");
	}

	public List<Experiment> experimentsByAnyField(String searchTerm) {
		ExperimentFetchOptions options = experimentFetchOptions();

		ExperimentSearchCriteria criterion = new ExperimentSearchCriteria();
		criterion.withAnyField().thatContains(searchTerm);
		return as.searchExperiments(sessionToken, criterion, options).getObjects();
	}

	public List<Experiment> experimentsByAttribute(String attribute, List<String> values)
			throws InvalidOptionException {
		ExperimentFetchOptions options = experimentFetchOptions();

		ExperimentSearchCriteria criterion = new ExperimentSearchCriteria();
		criterion.withOrOperator();
		updateCriterianForAttribute(criterion, attribute, values);

		return as.searchExperiments(sessionToken, criterion, options).getObjects();

	}

	public List<Experiment> experimentsByAttribute(String attribute, String value) throws InvalidOptionException {
		List<String> values = Arrays.asList(new String[] { value });
		return experimentsByAttribute(attribute, values);

	}

	public List<Experiment> experimentsByProperty(String property, String propertyValue) {
		ExperimentSearchCriteria criterion = new ExperimentSearchCriteria();
		criterion.withProperty(property).thatContains(propertyValue);

		ExperimentFetchOptions options = experimentFetchOptions();

		return as.searchExperiments(sessionToken, criterion, options).getObjects();
	}

        @SuppressWarnings("unchecked")
        public List<Experiment> experimentsByType(JSONObject query) throws InvalidOptionException {

            if (!query.containsKey("typeCode") && !query.containsKey("typeCodes"))
                throw new InvalidOptionException("Missing type code(s)");

            ExperimentFetchOptions options = experimentFetchOptions();
            ExperimentSearchCriteria criterion = new ExperimentSearchCriteria();

            if (query.containsKey("typeCodes")) {
                List<String> codes = Arrays.asList(query.get("typeCodes").toString().split(","));
                criterion.withType().withCodes().thatIn(codes);

            } else {
                String typeCode = (String)query.get("typeCode");
                criterion.withType().withCode().thatEquals(typeCode);
            }

            return as.searchExperiments(sessionToken, criterion, options).getObjects();
        }

        public List<ExperimentType> allExperimentTypes() {

            ExperimentTypeFetchOptions fetchOptions = new ExperimentTypeFetchOptions();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            ExperimentTypeSearchCriteria searchCriteria = new ExperimentTypeSearchCriteria();

            SearchResult<ExperimentType> types = as.searchExperimentTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();

        }

        public List<ExperimentType> experimentTypesByCodes(List<String> codes) {

            ExperimentTypeFetchOptions fetchOptions = new ExperimentTypeFetchOptions();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            ExperimentTypeSearchCriteria searchCriteria = new ExperimentTypeSearchCriteria();
            searchCriteria.withCodes().thatIn(codes);
            SearchResult<ExperimentType> types = as.searchExperimentTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();
        }

        public List<? extends Object> allEntities(String type) throws InvalidOptionException {

            switch (type) {
                case "Experiment":
                    return allExperiments();
                case "Sample":
                    return allSamples();
                case "DataSet":
                    return allDatasets();
                case "Space":
                    return allSpaces();
                case "SampleType":
                    return allSampleTypes();
                case "DataSetType":
                    return allDataSetTypes();
                case "ExperimentType":
                    return allExperimentTypes();
                default:
                    throw new InvalidOptionException("Unrecognised type: " + type);
            }
        }

	public List<? extends Object> query(String type, QueryType queryType, String key, List<String> values)
			throws InvalidOptionException {
		List<? extends Object> result = null;
		if (queryType == QueryType.ATTRIBUTE) {
                    switch (type) {
                        case "Experiment":
                            result = experimentsByAttribute(key, values);
                            break;
                        case "Sample":
                            result = samplesByAttribute(key, values);
                            break;
                        case "DataSet":
                            result = dataSetsByAttribute(key, values);
                            break;
                        case "Space":
                            result = spacesByAttribute(key, values);
                            break;
                        case "SampleType":
                            if (!"CODE".equals(key)) throw new InvalidOptionException("Unsupported attribute: " + key);
                            result = sampleTypesByCodes(values);
                            break;
                        case "DataSetType":
                            if (!"CODE".equals(key)) throw new InvalidOptionException("Unsupported attribute: " + key);
                            result = dataSetTypesByCode(values.get(0));
                            break;
                        case "ExperimentType":
                            if (!"CODE".equals(key)) throw new InvalidOptionException("Unsupported attribute: " + key);
                            result = experimentTypesByCodes(values);
                            break;
                        default:
                            throw new InvalidOptionException("Unrecognised type: " + type);
                    }
		} else {
			throw new InvalidOptionException("It is only possible to query by ATTRIBUTE when using an array of values");
		}

		return result;
	}

	public List<? extends Object> query(String type, QueryType queryType, String key, String value)
			throws InvalidOptionException {
		List<? extends Object> result = null;
		if (null == queryType) {
                    throw new InvalidOptionException("Unrecognised query type");
                } else switch (queryType) {
                case PROPERTY:
                    switch (type) {
                        case "Experiment":
                            result = experimentsByProperty(key, value);
                            break;
                        case "Sample":
                            result = samplesByProperty(key, value);
                            break;
                        case "DataSet":
                            result = dataSetsByProperty(key, value);
                            break;
                        default:
                            throw new InvalidOptionException("Unrecognised entity type: " + type);
                    }
                    break;
                case ATTRIBUTE:
                    List<String> values = new ArrayList<>();
                    values.add(value);
                    result = query(type, queryType, key, values);
                    break;
                default:
                    throw new InvalidOptionException("Unrecognised query type");
            }

		return result;
	}

	public List<Sample> samplesByAnyField(String searchTerm) {
		SampleFetchOptions options = sampleFetchOptions();

		SampleSearchCriteria criterion = new SampleSearchCriteria();
		criterion.withAnyField().thatContains(searchTerm);
		return as.searchSamples(sessionToken, criterion, options).getObjects();
	}

	public List<Sample> allSamples() throws InvalidOptionException {
		return samplesByAttribute("permId", "");
	}

	public List<Sample> samplesByAttribute(String attribute, List<String> values) throws InvalidOptionException {
		SampleFetchOptions options = sampleFetchOptions();

		SampleSearchCriteria criterion = new SampleSearchCriteria();

		criterion.withOrOperator();
		updateCriterianForAttribute(criterion, attribute, values);

		return as.searchSamples(sessionToken, criterion, options).getObjects();
	}

	public List<Sample> samplesByAttribute(String attribute, String value) throws InvalidOptionException {
		List<String> values = Arrays.asList(new String[] { value });
		return samplesByAttribute(attribute, values);
	}

	public List<Sample> samplesByProperty(String property, String propertyValue) {
		SampleSearchCriteria criterion = new SampleSearchCriteria();
		criterion.withProperty(property).thatContains(propertyValue);

		SampleFetchOptions options = sampleFetchOptions();

		return as.searchSamples(sessionToken, criterion, options).getObjects();
	}

        @SuppressWarnings("unchecked")
        public List<Sample> samplesByType(JSONObject query) throws InvalidOptionException {

            if (!query.containsKey("typeCode") && !query.containsKey("typeCodes"))
                throw new InvalidOptionException("Missing type code(s)");

            SampleFetchOptions options = sampleFetchOptions();
            SampleSearchCriteria criterion = new SampleSearchCriteria();

            if (query.containsKey("typeCodes")) {
                List<String> codes = Arrays.asList(query.get("typeCodes").toString().split(","));
                criterion.withType().withCodes().thatIn(codes);
            } else {
                String typeCode = (String)query.get("typeCode");
                criterion.withType().withCode().thatEquals(typeCode);
            }

            return as.searchSamples(sessionToken, criterion, options).getObjects();
        }

	public List<Space> allSpaces() throws InvalidOptionException  {
            return spacesByAttribute("permId", "");
	}

	public List<Space> spacesByAttribute(String attribute, List<String> values) throws InvalidOptionException {
		SpaceFetchOptions options = new SpaceFetchOptions();
		//options.withProjects().withExperiments().withDataSets();
		//options.withSamples().withDataSets();
		SpaceSearchCriteria criterion = new SpaceSearchCriteria();

		criterion.withOrOperator();
		for (String value : values) {
			criterion.withPermId().thatContains(value);
		}

		return as.searchSpaces(sessionToken, criterion, options).getObjects();
	}

	public List<Space> spacesByAttribute(String attribute, String value) throws InvalidOptionException {
		List<String> values = Arrays.asList(new String[] { value });
		return spacesByAttribute(attribute, values);
	}

        protected <K> boolean notNullAtKey(Map map,K key) {
            return map.containsKey(key) && ( map.get(key) != null);
        }

        public List<SampleType> allSampleTypes() {

            SampleTypeFetchOptions fetchOptions = new SampleTypeFetchOptions();
            //fetchOptions.withSemanticAnnotations();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            SampleTypeSearchCriteria searchCriteria = new SampleTypeSearchCriteria();

            SearchResult<SampleType> types = as.searchSampleTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();

        }

        public List<SampleType> sampleTypesByCodes(List<String> codes) {

            SampleTypeFetchOptions fetchOptions = new SampleTypeFetchOptions();
            //fetchOptions.withSemanticAnnotations();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            SampleTypeSearchCriteria searchCriteria = new SampleTypeSearchCriteria();

            searchCriteria.withCodes().thatIn(codes);

            SearchResult<SampleType> types = as.searchSampleTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();
        }

        public List<SampleType> sampleTypesByCode(String code) {

            SampleTypeFetchOptions fetchOptions = new SampleTypeFetchOptions();
            //fetchOptions.withSemanticAnnotations();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            SampleTypeSearchCriteria searchCriteria = new SampleTypeSearchCriteria();

            if (code.contains(",")) {
                List<String> codes = Arrays.asList(code.split(","));
                searchCriteria.withCodes().thatIn(codes);
            } else {
                searchCriteria.withCode().thatEquals(code);
            }

            SearchResult<SampleType> types = as.searchSampleTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();
        }

        public List<SampleType> sampleTypesBySemantic(JSONObject query) throws AuthenticationException {

            throw new UnsupportedOperationException("Semantic annotationa are not available in official release");
            /*
            SampleTypeFetchOptions fetchOptions = new SampleTypeFetchOptions();
            fetchOptions.withSemanticAnnotations();
            fetchOptions.withPropertyAssignments().withSemanticAnnotations();

            SampleTypeSearchCriteria searchCriteria = new SampleTypeSearchCriteria();

            SemanticAnnotationSearchCriteria semCriteria = searchCriteria.withSemanticAnnotations();

            if (notNullAtKey(query,"predicateOntologyId"))
                semCriteria.withPredicateOntologyId().thatEquals(query.get("predicateOntologyId").toString());

            if (notNullAtKey(query,"predicateOntologyVersion"))
                semCriteria.withPredicateOntologyVersion().thatEquals(query.get("predicateOntologyVersion").toString());

            if (notNullAtKey(query,"predicateAccessionId"))
                semCriteria.withPredicateAccessionId().thatEquals(query.get("predicateAccessionId").toString());

            if (notNullAtKey(query,"descriptorOntologyId"))
                semCriteria.withDescriptorOntologyId().thatEquals(query.get("descriptorOntologyId").toString());

            if (notNullAtKey(query,"descriptorOntologyVersion"))
                semCriteria.withDescriptorOntologyVersion().thatEquals(query.get("descriptorOntologyVersion").toString());

            if (notNullAtKey(query,"descriptorAccessionId"))
                semCriteria.withDescriptorAccessionId().thatEquals(query.get("descriptorAccessionId").toString());

            SearchResult<SampleType> types = as.searchSampleTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();
            */
        }

        public List<DataSetType> dataSetTypesByCode(String code) {

            DataSetTypeFetchOptions fetchOptions = new DataSetTypeFetchOptions();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            DataSetTypeSearchCriteria searchCriteria = new DataSetTypeSearchCriteria();

            searchCriteria.withCode().thatEquals(code);

            SearchResult<DataSetType> types = as.searchDataSetTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();
        }

        public List<DataSetType> allDataSetTypes() {

            DataSetTypeFetchOptions fetchOptions = new DataSetTypeFetchOptions();
            //fetchOptions.withPropertyAssignments().withSemanticAnnotations();
            fetchOptions.withPropertyAssignments();

            DataSetTypeSearchCriteria searchCriteria = new DataSetTypeSearchCriteria();

            SearchResult<DataSetType> types = as.searchDataSetTypes(sessionToken, searchCriteria, fetchOptions);
            return types.getObjects();

        }

	private DataSetFetchOptions dataSetFetchOptions() {
		DataSetFetchOptions options = new DataSetFetchOptions();
		options.withProperties();
		options.withSample();
		options.withExperiment();
		options.withModifier();
		options.withRegistrator();
		options.withTags();
		options.withType();
		return options;
	}

	private ExperimentFetchOptions experimentFetchOptions() {
		ExperimentFetchOptions options = new ExperimentFetchOptions();
		options.withProperties();
		options.withSamples();
		options.withDataSets();
		options.withProject();
		options.withModifier();
		options.withRegistrator();
		options.withTags();
		options.withType();
		return options;
	}

	private SampleFetchOptions sampleFetchOptions() {
		SampleFetchOptions options = new SampleFetchOptions();
		options.withProperties();
		options.withExperiment();
		options.withDataSets();
		options.withModifier();
		options.withRegistrator();
		options.withTags();
		options.withType();
		return options;
	}

	private void updateCriterianForAttribute(AbstractEntitySearchCriteria<?> criteria, String key, List<String> values)
			throws InvalidOptionException {
		if (key.equalsIgnoreCase("permid")) {
			for (String value : values) {
				criteria.withPermId().thatContains(value);
			}
		} else {
			throw new InvalidOptionException("Invalid attribute name:" + key);
		}
	}

}


text:
SELECT
    geometrie,
    beschreibung
FROM
    awjf_wildtierkorridore_v1.wildbarriere_linie
;


text:
module Api
  class TodoItemsController < Api::BaseController
    before_action :find_todo_list
    before_action :find_todo_item, only: %i[update destroy]

    # rubocop:disable Metrics/MethodLength
    def create
      item = @list.todo_items.new(item_params)
      if item.save
        render status: :ok, json: {
          message: 'Successfully created To-do Item.',
          todo_list: @list,
          todo_item: item
        }.to_json
      else
        render status: :unprocessable_entity, json: {
          message: 'To-do Item creation failed.',
          errors: item.errors
        }.to_json
      end
    end

    def update
      if @item.update(item_params)
        render status: :ok, json: {
          message: 'Successfully updated To-do Item.',
          todo_list: @list,
          todo_item: @item
        }.to_json
      else
        render status: :unprocessable_entity, json: {
          message: 'To-do Item update is failed.',
          errors: @item.errors
        }.to_json
      end
    end
    # rubocop:enable Metrics/MethodLength

    def destroy
      @item.destroy
      render status: :ok, json: {
        messsage: 'Successfully deleted To-do Item.'
      }.to_json
    end

    private

    def item_params
      params.require(:todo_item).permit(:content)
    end

    def find_todo_list
      @list = current_user.todo_lists.find(params[:todo_list_id])
    end

    def find_todo_item
      @item = @list.todo_items.find(params[:id])
    end
  end
end