Node* recv = RecvAtHost(
        ops::NodeOut(key_constant, 0), "F1", "F1", "O1", {DT_FLOAT, DT_FLOAT},
        b2.opts().WithAttr(kXlaHasHostTransferAttrName, true));
    Node* e = Binary(ops::NodeOut(recv, 0), ops::NodeOut(recv, 1),
                     b2.opts()
                         .WithName("E")
                         .WithControlInputs({recv})

			}
			if !cas.recv.IsValid() {
				t.Fatalf("%s\nselected #%d but internal error: missing recv value", fmtSelect(info), i)
			}
			if recv.Interface() != cas.recv.Interface() || recvOK != !cas.closed {
				if recv.Interface() == cas.recv.Interface() && recvOK == !cas.closed {

	return v.recv(false)
}

// internal recv, possibly non-blocking (nb).
// v is known to be a channel.
func (v Value) recv(nb bool) (val Value, ok bool) {
	tt := (*chanType)(unsafe.Pointer(v.typ()))
	if ChanDir(tt.Dir)&RecvDir == 0 {
		panic("reflect: recv on send-only channel")
	}
	t := tt.Elem
	val = Value{t, nil, flag(t.Kind())}
	var p unsafe.Pointer
	if t.IfaceIndir() {
		p = unsafe_New(t)
		val.ptr = p

  let constructor = "TFTPU::CreateTPUParallelExecuteSinkResourceWritePass()";
}

def RewriteTPUEmbeddingOpsPass : Pass<"tf-rewrite-tpu-embedding-ops", "mlir::func::FuncOp"> {
  let summary = "Rewrites TPU embedding send/recv ops by adding TPU embedding "
           "deduplication data";

  let constructor = "TF::CreateRewriteTPUEmbeddingOpsPass()";
}

    // `stablehlo.custom_call` will be lowered to `stablehlo.send` and
    // `stablehlo.recv`.
    effects.emplace_back(MemoryEffects::Write::get(),
                         ResourceEffects::Send::get());
    effects.emplace_back(MemoryEffects::Write::get(),
                         ResourceEffects::Recv::get());
    effects.emplace_back(MemoryEffects::Write::get(),

  // Step 1: add key placeholder node.
  TF_ASSIGN_OR_RETURN(
      Node * key_placeholder,
      AddHostComputeKeyPlaceholder(xla_cluster_name, &host_graph));

  // Step 2: build XlaRecvAtHost node to recv predicate.
  NodeDefBuilder recv_pred_builder(
      absl::StrCat("recv_oc_if_pred_", if_node_name), "_XlaRecvAtHost");
  recv_pred_builder.Attr("Toutputs", std::vector<DataType>{DT_BOOL});

						// in this window, back off to give pp a chance to
						// schedule runnext. This will avoid thrashing gs
						// between different Ps.
						// A sync chan send/recv takes ~50ns as of time of
						// writing, so 3us gives ~50x overshoot.
						if !osHasLowResTimer {
							usleep(3)
						} else {
							// On some platforms system timer granularity is