// If the request context is canceled, the stream will no longer process requests.
	// Requests sent cannot be used any further by the called.
	Requests chan<- []byte

	muxID uint64
	ctx   context.Context
}

// Send a payload to the remote server.
func (s *Stream) Send(b []byte) error {
	if s.Requests == nil {
		return errors.New("stream does not accept requests")
	}
	select {

}

// moduledata records information about the layout of the executable
// image. It is written by the linker. Any changes here must be
// matched changes to the code in cmd/link/internal/ld/symtab.go:symtab.
// moduledata is stored in statically allocated non-pointer memory;
// none of the pointers here are visible to the garbage collector.
type moduledata struct {
	sys.NotInHeap // Only in static data

    // on transfers, then we might want to rethink this property.
    // Also ideally this host callback should be on source stream rather than
    // destination stream, but when this function returns, the send requests
    // might not be enqueued to the stream yet, we put it on destination stream.
    TensorReference input_reference(*input);
    std::move(return_substream).release();

se::Stream* GetStream(OpKernelContext* ctx) {
  return ctx->op_device_context() ? ctx->op_device_context()->stream()
                                  : nullptr;
}

XlaComputationLaunchContext GetLaunchContext(
    const XlaPlatformInfo& platform_info, OpKernelContext* ctx,
    xla::LocalClient* client, se::DeviceMemoryAllocator* allocator) {
  se::Stream* stream = GetStream(ctx);

                                       bool* stream_was_changed) {
  if (!(*stream) || !(*stream)->ok()) {
    xla::StreamPool::Ptr ptr;
    TF_ASSIGN_OR_RETURN(ptr, backend->BorrowStream(device_ordinal_));
    *stream = std::shared_ptr<se::Stream>(std::move(ptr));
    VLOG(1) << "XlaDevice " << this << " new " << name
            << " stream=" << (*stream);

                                           &HostCallbackTrampoline, ctx);
  }
  void DeallocateStream(Stream* stream) override {
    static_cast<CStream*>(stream)->Destroy();
  }
  absl::Status BlockHostForEvent(Stream* stream, Event* event) {
    OwnedTFStatus c_status(TF_NewStatus());
    SP_Event event_handle = static_cast<CEvent*>(event)->Handle();
    stream_executor_->block_host_for_event(&device_, event_handle,

    status->status = absl::FailedPreconditionError(
        "Accessing device stream is not supported for a CPU device.");
    return nullptr;
  } else if (!cc_ctx->op_device_context()->IsPluggableDevice()) {
    status->status = absl::FailedPreconditionError(
        "Accessing device stream is only supported for pluggable devices.");
    return nullptr;
  } else {  // Is a PluggableDevice

            stream.received(bytes)
        }

        then:
        1 * dispatch.onOutput({ it instanceof ReadStdInEvent }) >> { instant.requested }
    }

    def "read byte returns when stream is closed"() {
        def dispatch = Mock(OutputEventListener)
        def stream = new StdInStream(dispatch)

        when:
        async {
            start {
                def b1 = stream.read()

  TF_ASSERT_OK(stream->WaitFor(event.get()));
  ASSERT_TRUE(wait_called);
}

TEST_F(StreamExecutorTest, MemcpyToHost) {
  se_.create_stream = [](const SP_Device* const device, SP_Stream* stream,
                         TF_Status* const status) -> void {
    *stream = new SP_Stream_st(14);
  };
  se_.destroy_stream = [](const SP_Device* const device,
                          SP_Stream stream) -> void { delete stream; };

        Stream<? extends Element> annotatedTypes = getSupportedAnnotations().stream()
            .flatMap(annotation -> roundEnv.getElementsAnnotatedWith(annotation).stream())
            .flatMap(element -> findActualTypesToVisit(element).stream())
            .sorted(Comparator.comparing(AbstractInstrumentationProcessor::elementQualifiedName));