if err != nil {
		b.Fatal("error dialing:", err)
	}
	defer client.Close()

	// Asynchronous calls
	args := &Args{7, 8}
	procs := 4 * runtime.GOMAXPROCS(-1)
	send := int32(b.N)
	recv := int32(b.N)
	var wg sync.WaitGroup
	wg.Add(procs)
	gate := make(chan bool, MaxConcurrentCalls)
	res := make(chan *Call, MaxConcurrentCalls)
	b.ResetTimer()

	for p := 0; p < procs; p++ {
		go func() {

				Children: make(keySet),
			}
		}
		c.initialWatches = append(c.initialWatches, key)
	}
}

func (c *Client) handleRecv() {
	for {
		deltaLog.Infof("Start Recv for node %v", c.nodeID)
		msg, err := c.xdsClient.Recv()
		if err != nil {
			deltaLog.Infof("Connection closed for node %v with err: %v", c.nodeID, err)
			select {
			case c.errChan <- err:
			default:
			}

				return true
			}, true, retry.Timeout(time.Second), retry.Delay(time.Millisecond))

			if !cmp.Equal(tt.inClient.lastReceived, tt.expectedDeltaResources.lastReceived, protocmp.Transform()) {
				t.Errorf("%s: expected recv %v got %v", tt.desc, tt.expectedDeltaResources.lastReceived, tt.inClient.lastReceived)
			}

			tree := tt.inClient.dumpTree()
			if diff := cmp.Diff(tt.expectedTree, tree); diff != "" {

			case *syntax.ExprStmt:
				rhs = s.X
			}

			// if present, rhs must be a receive operation
			if rhs != nil {
				if x, _ := syntax.Unparen(rhs).(*syntax.Operation); x != nil && x.Y == nil && x.Op == syntax.Recv {
					valid = true
				}
			}

			if !valid {
				check.error(clause.Comm, InvalidSelectCase, "select case must be send or receive (possibly with assignment)")
				continue
			}
			end := s.Rbrace

	SizeofBpfStat    = 0x8
	SizeofBpfProgram = 0x10
	SizeofBpfInsn    = 0x8
	SizeofBpfHdr     = 0x14
)

type BpfVersion struct {
	Major uint16
	Minor uint16
}

type BpfStat struct {
	Recv uint32
	Drop uint32
}

type BpfProgram struct {
	Len   uint32
	Insns *BpfInsn
}

type BpfInsn struct {
	Code uint16
	Jt   uint8
	Jf   uint8
	K    uint32
}

	SizeofBpfStat    = 0x8
	SizeofBpfProgram = 0x10
	SizeofBpfInsn    = 0x8
	SizeofBpfHdr     = 0x14
)

type BpfVersion struct {
	Major uint16
	Minor uint16
}

type BpfStat struct {
	Recv uint32
	Drop uint32
}

type BpfProgram struct {
	Len   uint32
	Insns *BpfInsn
}

type BpfInsn struct {
	Code uint16
	Jt   uint8
	Jf   uint8
	K    uint32
}

    rewriter.eraseOp(assign_group);
    return success();
  }
};

void LegalizeTFCollective::runOnOperation() {
  // FIXME(b/226139061): Figure out a way to share the channel_id with
  // send/recv Ops. For now, start with a different range to avoid collision.
  int64_t channel_id = 10000;
  auto module = getOperation();
  MLIRContext* context = module->getContext();

  RewritePatternSet patterns(context);

	// to tell the receiving goroutines that all data have been sent."

	// Block until either a request is received or a push is triggered.
	// We need 2 go routines because 'read' blocks in Recv().
	go s.receiveDelta(con, ids)

	// Wait for the proxy to be fully initialized before we start serving traffic. Because
	// initialization doesn't have dependencies that will block, there is no need to add any timeout

        args.device_context = new XlaHostSendDeviceContext(
            stream, device_memory_base, shape, done_event);

        Tensor device_tensor;
        bool is_dead;
        TF_RETURN_IF_ERROR(ctx->rendezvous()->Recv(
            parsed_key, args, &device_tensor, /*is_dead=*/&is_dead));

        return std::move(done_event);
      };
}

absl::StatusOr<xla::ExecutionOutput> RunExecutable(

			return nil
		}
		return &ast.MapType{
			Key:   key,
			Value: value,
		}
	case *types.Chan:
		dir := ast.ChanDir(t.Dir())
		if t.Dir() == types.SendRecv {
			dir = ast.SEND | ast.RECV
		}
		value := TypeExpr(f, pkg, t.Elem())
		if value == nil {
			return nil
		}
		return &ast.ChanType{
			Dir:   dir,
			Value: value,
		}
	case *types.Signature:
		var params []*ast.Field