From nobody Sat Nov 15 12:47:47 2025 Delivered-To: importer@patchew.org Authentication-Results: mx.zohomail.com; dkim=pass; spf=pass (zohomail.com: domain of gnu.org designates 209.51.188.17 as permitted sender) smtp.mailfrom=qemu-devel-bounces+importer=patchew.org@nongnu.org; dmarc=pass(p=quarantine dis=none) header.from=redhat.com ARC-Seal: i=1; a=rsa-sha256; t=1752554264; cv=none; d=zohomail.com; s=zohoarc; b=RF4fkJxBcN7BUZAQaExpMsqMZ9xFhB6pKB3TxEYj+ljiNK/hvmNC0lv+mWrZXqkM9oaGBdp/mKpwxx/pMmeUNI/jc4fLpulNU4YsOaYXVpYEXeR1WyNOt6/dNLUJZqBh/wDypvvihaEs0YtRJ3WVWit399YQQYB9V3y5UGUMRE0= ARC-Message-Signature: i=1; a=rsa-sha256; c=relaxed/relaxed; d=zohomail.com; s=zohoarc; t=1752554264; h=Content-Type:Content-Transfer-Encoding:Cc:Cc:Date:Date:From:From:In-Reply-To:List-Subscribe:List-Post:List-Id:List-Archive:List-Help:List-Unsubscribe:MIME-Version:Message-ID:References:Sender:Subject:Subject:To:To:Message-Id:Reply-To; bh=01NUvw0ZJlHXFkftpgQ5s2vtgM+qyaS0xCjJIBcuPOE=; b=D7vaVhf7K5PirCqwxG0gOYBF1z9sLle3dFihfYiZXK5ZN5xxlVzxTMZBoH1xp4J/uUAUOuRkqm5VwMc5DOhLym4s4LLfP9VGbX51S0UduohOGS6iCIl82e5IRGY1rDlYMiXAVFgILaPepwiIGM85Lq9+uYUVxkD62+nHYF8a+ZQ= ARC-Authentication-Results: i=1; mx.zohomail.com; dkim=pass; spf=pass (zohomail.com: domain of gnu.org designates 209.51.188.17 as permitted sender) smtp.mailfrom=qemu-devel-bounces+importer=patchew.org@nongnu.org; dmarc=pass header.from= (p=quarantine dis=none) Return-Path: Received: from lists.gnu.org (lists.gnu.org [209.51.188.17]) by mx.zohomail.com with SMTPS id 1752554264235466.22706751023384; Mon, 14 Jul 2025 21:37:44 -0700 (PDT) Received: from localhost ([::1] helo=lists1p.gnu.org) by lists.gnu.org with esmtp (Exim 4.90_1) (envelope-from ) id 1ubXPK-0001s9-1i; Tue, 15 Jul 2025 00:36:46 -0400 Received: from eggs.gnu.org ([2001:470:142:3::10]) by lists.gnu.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.90_1) (envelope-from ) id 1ubXOO-0001MB-IM for qemu-devel@nongnu.org; Tue, 15 Jul 2025 00:35:52 -0400 Received: from us-smtp-delivery-124.mimecast.com ([170.10.133.124]) by eggs.gnu.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.90_1) (envelope-from ) id 1ubXOL-0005zn-8t for qemu-devel@nongnu.org; Tue, 15 Jul 2025 00:35:48 -0400 Received: from mx-prod-mc-03.mail-002.prod.us-west-2.aws.redhat.com (ec2-54-186-198-63.us-west-2.compute.amazonaws.com [54.186.198.63]) by relay.mimecast.com with ESMTP with STARTTLS (version=TLSv1.3, cipher=TLS_AES_256_GCM_SHA384) id us-mta-370-a9Nux4TDMiauThWqUJw3Cg-1; Tue, 15 Jul 2025 00:35:41 -0400 Received: from mx-prod-int-05.mail-002.prod.us-west-2.aws.redhat.com (mx-prod-int-05.mail-002.prod.us-west-2.aws.redhat.com [10.30.177.17]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (2048 bits) server-digest SHA256) (No client certificate requested) by mx-prod-mc-03.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTPS id 6A6CF19560B6 for ; Tue, 15 Jul 2025 04:35:40 +0000 (UTC) Received: from localhost.localdomain (unknown [10.72.112.42]) by mx-prod-int-05.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTP id 6E12A1977000; Tue, 15 Jul 2025 04:35:37 +0000 (UTC) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=redhat.com; s=mimecast20190719; t=1752554144; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding: in-reply-to:in-reply-to:references:references; bh=01NUvw0ZJlHXFkftpgQ5s2vtgM+qyaS0xCjJIBcuPOE=; b=ZLQJyzeW/p0Mx73pWd09iZxjDvoh7pq5IxE8ZndPZ6jKcekyp2u2s+B1JKrFn9tVkfkGTR RsnT+YB0X9mfHG1kaqb0422iGQ2YpfzSNorL2a4MivWc1XaLVmrZ3Mva5Vqjqc9L3vxYiG q0OKfFgMjqzhHvICatalxaWuXVf9qEw= X-MC-Unique: a9Nux4TDMiauThWqUJw3Cg-1 X-Mimecast-MFC-AGG-ID: a9Nux4TDMiauThWqUJw3Cg_1752554140 From: Jason Wang To: qemu-devel@nongnu.org Cc: Laurent Vivier , Jason Wang Subject: [PULL V2 03/16] net: Refactor stream logic for reuse in '-net passt' Date: Tue, 15 Jul 2025 12:35:11 +0800 Message-ID: <20250715043524.21719-4-jasowang@redhat.com> In-Reply-To: <20250715043524.21719-1-jasowang@redhat.com> References: <20250715043524.21719-1-jasowang@redhat.com> MIME-Version: 1.0 Content-Transfer-Encoding: quoted-printable X-Scanned-By: MIMEDefang 3.0 on 10.30.177.17 Received-SPF: pass (zohomail.com: domain of gnu.org designates 209.51.188.17 as permitted sender) client-ip=209.51.188.17; envelope-from=qemu-devel-bounces+importer=patchew.org@nongnu.org; helo=lists.gnu.org; Received-SPF: pass client-ip=170.10.133.124; envelope-from=jasowang@redhat.com; helo=us-smtp-delivery-124.mimecast.com X-Spam_score_int: -20 X-Spam_score: -2.1 X-Spam_bar: -- X-Spam_report: (-2.1 / 5.0 requ) BAYES_00=-1.9, DKIMWL_WL_HIGH=-0.001, DKIM_SIGNED=0.1, DKIM_VALID=-0.1, DKIM_VALID_AU=-0.1, DKIM_VALID_EF=-0.1, RCVD_IN_DNSWL_NONE=-0.0001, RCVD_IN_MSPIKE_H5=0.001, RCVD_IN_MSPIKE_WL=0.001, RCVD_IN_VALIDITY_CERTIFIED_BLOCKED=0.001, RCVD_IN_VALIDITY_RPBL_BLOCKED=0.001, SPF_HELO_NONE=0.001, SPF_PASS=-0.001 autolearn=ham autolearn_force=no X-Spam_action: no action X-BeenThere: qemu-devel@nongnu.org X-Mailman-Version: 2.1.29 Precedence: list List-Id: List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Errors-To: qemu-devel-bounces+importer=patchew.org@nongnu.org Sender: qemu-devel-bounces+importer=patchew.org@nongnu.org X-ZohoMail-DKIM: pass (identity @redhat.com) X-ZM-MESSAGEID: 1752554265980116600 Content-Type: text/plain; charset="utf-8" From: Laurent Vivier To prepare for the implementation of '-net passt', this patch moves the generic stream handling functions from net/stream.c into new net/stream_data.c and net/stream_data.h files. This refactoring introduces a NetStreamData struct that encapsulates the generic fields and logic previously in NetStreamState. The NetStreamState now embeds NetStreamData and delegates the core stream operations to the new generic functions. To maintain flexibility for different users of this generic code, callbacks for send and listen operations are now passed via function pointers within the NetStreamData struct. This allows callers to provide their own specific implementations while reusing the common connection and data transfer logic. Signed-off-by: Laurent Vivier Signed-off-by: Jason Wang --- net/meson.build | 3 +- net/stream.c | 282 +++++++++++----------------------------------- net/stream_data.c | 193 +++++++++++++++++++++++++++++++ net/stream_data.h | 31 +++++ 4 files changed, 290 insertions(+), 219 deletions(-) create mode 100644 net/stream_data.c create mode 100644 net/stream_data.h diff --git a/net/meson.build b/net/meson.build index bb97b4dcbe..bb3c011e5a 100644 --- a/net/meson.build +++ b/net/meson.build @@ -1,6 +1,7 @@ system_ss.add(files( 'announce.c', 'checksum.c', + 'dgram.c', 'dump.c', 'eth.c', 'filter-buffer.c', @@ -12,7 +13,7 @@ system_ss.add(files( 'queue.c', 'socket.c', 'stream.c', - 'dgram.c', + 'stream_data.c', 'util.c', )) =20 diff --git a/net/stream.c b/net/stream.c index 6152d2a05e..d893f02cab 100644 --- a/net/stream.c +++ b/net/stream.c @@ -27,173 +27,50 @@ =20 #include "net/net.h" #include "clients.h" -#include "monitor/monitor.h" #include "qapi/error.h" -#include "qemu/error-report.h" -#include "qemu/option.h" -#include "qemu/sockets.h" -#include "qemu/iov.h" -#include "qemu/main-loop.h" -#include "qemu/cutils.h" -#include "io/channel.h" -#include "io/channel-socket.h" #include "io/net-listener.h" #include "qapi/qapi-events-net.h" #include "qapi/qapi-visit-sockets.h" #include "qapi/clone-visitor.h" =20 +#include "stream_data.h" + typedef struct NetStreamState { - NetClientState nc; - QIOChannel *listen_ioc; - QIONetListener *listener; - QIOChannel *ioc; - guint ioc_read_tag; - guint ioc_write_tag; - SocketReadState rs; - unsigned int send_index; /* number of bytes sent*/ + NetStreamData data; uint32_t reconnect_ms; guint timer_tag; SocketAddress *addr; } NetStreamState; =20 -static void net_stream_listen(QIONetListener *listener, - QIOChannelSocket *cioc, - void *opaque); static void net_stream_arm_reconnect(NetStreamState *s); =20 -static gboolean net_stream_writable(QIOChannel *ioc, - GIOCondition condition, - gpointer data) -{ - NetStreamState *s =3D data; - - s->ioc_write_tag =3D 0; - - qemu_flush_queued_packets(&s->nc); - - return G_SOURCE_REMOVE; -} - static ssize_t net_stream_receive(NetClientState *nc, const uint8_t *buf, size_t size) { - NetStreamState *s =3D DO_UPCAST(NetStreamState, nc, nc); - uint32_t len =3D htonl(size); - struct iovec iov[] =3D { - { - .iov_base =3D &len, - .iov_len =3D sizeof(len), - }, { - .iov_base =3D (void *)buf, - .iov_len =3D size, - }, - }; - struct iovec local_iov[2]; - unsigned int nlocal_iov; - size_t remaining; - ssize_t ret; - - remaining =3D iov_size(iov, 2) - s->send_index; - nlocal_iov =3D iov_copy(local_iov, 2, iov, 2, s->send_index, remaining= ); - ret =3D qio_channel_writev(s->ioc, local_iov, nlocal_iov, NULL); - if (ret =3D=3D QIO_CHANNEL_ERR_BLOCK) { - ret =3D 0; /* handled further down */ - } - if (ret =3D=3D -1) { - s->send_index =3D 0; - return -errno; - } - if (ret < (ssize_t)remaining) { - s->send_index +=3D ret; - s->ioc_write_tag =3D qio_channel_add_watch(s->ioc, G_IO_OUT, - net_stream_writable, s, N= ULL); - return 0; - } - s->send_index =3D 0; - return size; -} - -static gboolean net_stream_send(QIOChannel *ioc, - GIOCondition condition, - gpointer data); - -static void net_stream_send_completed(NetClientState *nc, ssize_t len) -{ - NetStreamState *s =3D DO_UPCAST(NetStreamState, nc, nc); - - if (!s->ioc_read_tag) { - s->ioc_read_tag =3D qio_channel_add_watch(s->ioc, G_IO_IN, - net_stream_send, s, NULL); - } -} + NetStreamData *d =3D DO_UPCAST(NetStreamData, nc, nc); =20 -static void net_stream_rs_finalize(SocketReadState *rs) -{ - NetStreamState *s =3D container_of(rs, NetStreamState, rs); - - if (qemu_send_packet_async(&s->nc, rs->buf, - rs->packet_len, - net_stream_send_completed) =3D=3D 0) { - if (s->ioc_read_tag) { - g_source_remove(s->ioc_read_tag); - s->ioc_read_tag =3D 0; - } - } + return net_stream_data_receive(d, buf, size); } =20 static gboolean net_stream_send(QIOChannel *ioc, GIOCondition condition, gpointer data) { - NetStreamState *s =3D data; - int size; - int ret; - QEMU_UNINITIALIZED char buf1[NET_BUFSIZE]; - const char *buf; - - size =3D qio_channel_read(s->ioc, buf1, sizeof(buf1), NULL); - if (size < 0) { - if (errno !=3D EWOULDBLOCK) { - goto eoc; - } - } else if (size =3D=3D 0) { - /* end of connection */ - eoc: - s->ioc_read_tag =3D 0; - if (s->ioc_write_tag) { - g_source_remove(s->ioc_write_tag); - s->ioc_write_tag =3D 0; - } - if (s->listener) { - qemu_set_info_str(&s->nc, "listening"); - qio_net_listener_set_client_func(s->listener, net_stream_liste= n, - s, NULL); - } - object_unref(OBJECT(s->ioc)); - s->ioc =3D NULL; - - net_socket_rs_init(&s->rs, net_stream_rs_finalize, false); - s->nc.link_down =3D true; + if (net_stream_data_send(ioc, condition, data) =3D=3D G_SOURCE_REMOVE)= { + NetStreamState *s =3D DO_UPCAST(NetStreamState, data, data); =20 - qapi_event_send_netdev_stream_disconnected(s->nc.name); + qapi_event_send_netdev_stream_disconnected(s->data.nc.name); net_stream_arm_reconnect(s); =20 return G_SOURCE_REMOVE; } - buf =3D buf1; - - ret =3D net_fill_rstate(&s->rs, (const uint8_t *)buf, size); - - if (ret =3D=3D -1) { - goto eoc; - } =20 return G_SOURCE_CONTINUE; } =20 static void net_stream_cleanup(NetClientState *nc) { - NetStreamState *s =3D DO_UPCAST(NetStreamState, nc, nc); + NetStreamState *s =3D DO_UPCAST(NetStreamState, data.nc, nc); if (s->timer_tag) { g_source_remove(s->timer_tag); s->timer_tag =3D 0; @@ -202,28 +79,28 @@ static void net_stream_cleanup(NetClientState *nc) qapi_free_SocketAddress(s->addr); s->addr =3D NULL; } - if (s->ioc) { - if (QIO_CHANNEL_SOCKET(s->ioc)->fd !=3D -1) { - if (s->ioc_read_tag) { - g_source_remove(s->ioc_read_tag); - s->ioc_read_tag =3D 0; + if (s->data.ioc) { + if (QIO_CHANNEL_SOCKET(s->data.ioc)->fd !=3D -1) { + if (s->data.ioc_read_tag) { + g_source_remove(s->data.ioc_read_tag); + s->data.ioc_read_tag =3D 0; } - if (s->ioc_write_tag) { - g_source_remove(s->ioc_write_tag); - s->ioc_write_tag =3D 0; + if (s->data.ioc_write_tag) { + g_source_remove(s->data.ioc_write_tag); + s->data.ioc_write_tag =3D 0; } } - object_unref(OBJECT(s->ioc)); - s->ioc =3D NULL; + object_unref(OBJECT(s->data.ioc)); + s->data.ioc =3D NULL; } - if (s->listen_ioc) { - if (s->listener) { - qio_net_listener_disconnect(s->listener); - object_unref(OBJECT(s->listener)); - s->listener =3D NULL; + if (s->data.listen_ioc) { + if (s->data.listener) { + qio_net_listener_disconnect(s->data.listener); + object_unref(OBJECT(s->data.listener)); + s->data.listener =3D NULL; } - object_unref(OBJECT(s->listen_ioc)); - s->listen_ioc =3D NULL; + object_unref(OBJECT(s->data.listen_ioc)); + s->data.listen_ioc =3D NULL; } } =20 @@ -235,23 +112,13 @@ static NetClientInfo net_stream_info =3D { }; =20 static void net_stream_listen(QIONetListener *listener, - QIOChannelSocket *cioc, - void *opaque) + QIOChannelSocket *cioc, gpointer data) { - NetStreamState *s =3D opaque; + NetStreamData *d =3D data; SocketAddress *addr; char *uri; =20 - object_ref(OBJECT(cioc)); - - qio_net_listener_set_client_func(s->listener, NULL, s, NULL); - - s->ioc =3D QIO_CHANNEL(cioc); - qio_channel_set_name(s->ioc, "stream-server"); - s->nc.link_down =3D false; - - s->ioc_read_tag =3D qio_channel_add_watch(s->ioc, G_IO_IN, net_stream_= send, - s, NULL); + net_stream_data_listen(listener, cioc, data); =20 if (cioc->localAddr.ss_family =3D=3D AF_UNIX) { addr =3D qio_channel_socket_get_local_address(cioc, NULL); @@ -260,22 +127,22 @@ static void net_stream_listen(QIONetListener *listene= r, } g_assert(addr !=3D NULL); uri =3D socket_uri(addr); - qemu_set_info_str(&s->nc, "%s", uri); + qemu_set_info_str(&d->nc, "%s", uri); g_free(uri); - qapi_event_send_netdev_stream_connected(s->nc.name, addr); + qapi_event_send_netdev_stream_connected(d->nc.name, addr); qapi_free_SocketAddress(addr); } =20 static void net_stream_server_listening(QIOTask *task, gpointer opaque) { - NetStreamState *s =3D opaque; - QIOChannelSocket *listen_sioc =3D QIO_CHANNEL_SOCKET(s->listen_ioc); + NetStreamData *d =3D opaque; + QIOChannelSocket *listen_sioc =3D QIO_CHANNEL_SOCKET(d->listen_ioc); SocketAddress *addr; int ret; Error *err =3D NULL; =20 if (qio_task_propagate_error(task, &err)) { - qemu_set_info_str(&s->nc, "error: %s", error_get_pretty(err)); + qemu_set_info_str(&d->nc, "error: %s", error_get_pretty(err)); error_free(err); return; } @@ -284,20 +151,21 @@ static void net_stream_server_listening(QIOTask *task= , gpointer opaque) g_assert(addr !=3D NULL); ret =3D qemu_socket_try_set_nonblock(listen_sioc->fd); if (addr->type =3D=3D SOCKET_ADDRESS_TYPE_FD && ret < 0) { - qemu_set_info_str(&s->nc, "can't use file descriptor %s (errno %d)= ", + qemu_set_info_str(&d->nc, "can't use file descriptor %s (errno %d)= ", addr->u.fd.str, -ret); return; } g_assert(ret =3D=3D 0); qapi_free_SocketAddress(addr); =20 - s->nc.link_down =3D true; - s->listener =3D qio_net_listener_new(); + d->nc.link_down =3D true; + d->listener =3D qio_net_listener_new(); =20 - qemu_set_info_str(&s->nc, "listening"); - net_socket_rs_init(&s->rs, net_stream_rs_finalize, false); - qio_net_listener_set_client_func(s->listener, net_stream_listen, s, NU= LL); - qio_net_listener_add(s->listener, listen_sioc); + qemu_set_info_str(&d->nc, "listening"); + net_socket_rs_init(&d->rs, net_stream_data_rs_finalize, false); + qio_net_listener_set_client_func(d->listener, d->listen, d, + NULL); + qio_net_listener_add(d->listener, listen_sioc); } =20 static int net_stream_server_init(NetClientState *peer, @@ -307,16 +175,18 @@ static int net_stream_server_init(NetClientState *pee= r, Error **errp) { NetClientState *nc; - NetStreamState *s; + NetStreamData *d; QIOChannelSocket *listen_sioc =3D qio_channel_socket_new(); =20 nc =3D qemu_new_net_client(&net_stream_info, peer, model, name); - s =3D DO_UPCAST(NetStreamState, nc, nc); - qemu_set_info_str(&s->nc, "initializing"); + d =3D DO_UPCAST(NetStreamData, nc, nc); + d->send =3D net_stream_send; + d->listen =3D net_stream_listen; + qemu_set_info_str(&d->nc, "initializing"); =20 - s->listen_ioc =3D QIO_CHANNEL(listen_sioc); + d->listen_ioc =3D QIO_CHANNEL(listen_sioc); qio_channel_socket_listen_async(listen_sioc, addr, 0, - net_stream_server_listening, s, + net_stream_server_listening, d, NULL, NULL); =20 return 0; @@ -325,49 +195,23 @@ static int net_stream_server_init(NetClientState *pee= r, static void net_stream_client_connected(QIOTask *task, gpointer opaque) { NetStreamState *s =3D opaque; - QIOChannelSocket *sioc =3D QIO_CHANNEL_SOCKET(s->ioc); + NetStreamData *d =3D &s->data; + QIOChannelSocket *sioc =3D QIO_CHANNEL_SOCKET(d->ioc); SocketAddress *addr; gchar *uri; - int ret; - Error *err =3D NULL; =20 - if (qio_task_propagate_error(task, &err)) { - qemu_set_info_str(&s->nc, "error: %s", error_get_pretty(err)); - error_free(err); - goto error; + if (net_stream_data_client_connected(task, d) =3D=3D -1) { + net_stream_arm_reconnect(s); + return; } =20 addr =3D qio_channel_socket_get_remote_address(sioc, NULL); g_assert(addr !=3D NULL); uri =3D socket_uri(addr); - qemu_set_info_str(&s->nc, "%s", uri); + qemu_set_info_str(&d->nc, "%s", uri); g_free(uri); - - ret =3D qemu_socket_try_set_nonblock(sioc->fd); - if (addr->type =3D=3D SOCKET_ADDRESS_TYPE_FD && ret < 0) { - qemu_set_info_str(&s->nc, "can't use file descriptor %s (errno %d)= ", - addr->u.fd.str, -ret); - qapi_free_SocketAddress(addr); - goto error; - } - g_assert(ret =3D=3D 0); - - net_socket_rs_init(&s->rs, net_stream_rs_finalize, false); - - /* Disable Nagle algorithm on TCP sockets to reduce latency */ - qio_channel_set_delay(s->ioc, false); - - s->ioc_read_tag =3D qio_channel_add_watch(s->ioc, G_IO_IN, net_stream_= send, - s, NULL); - s->nc.link_down =3D false; - qapi_event_send_netdev_stream_connected(s->nc.name, addr); + qapi_event_send_netdev_stream_connected(d->nc.name, addr); qapi_free_SocketAddress(addr); - - return; -error: - object_unref(OBJECT(s->ioc)); - s->ioc =3D NULL; - net_stream_arm_reconnect(s); } =20 static gboolean net_stream_reconnect(gpointer data) @@ -378,7 +222,7 @@ static gboolean net_stream_reconnect(gpointer data) s->timer_tag =3D 0; =20 sioc =3D qio_channel_socket_new(); - s->ioc =3D QIO_CHANNEL(sioc); + s->data.ioc =3D QIO_CHANNEL(sioc); qio_channel_socket_connect_async(sioc, s->addr, net_stream_client_connected, s, NULL, NULL); @@ -388,7 +232,7 @@ static gboolean net_stream_reconnect(gpointer data) static void net_stream_arm_reconnect(NetStreamState *s) { if (s->reconnect_ms && s->timer_tag =3D=3D 0) { - qemu_set_info_str(&s->nc, "connecting"); + qemu_set_info_str(&s->data.nc, "connecting"); s->timer_tag =3D g_timeout_add(s->reconnect_ms, net_stream_reconne= ct, s); } } @@ -405,11 +249,13 @@ static int net_stream_client_init(NetClientState *pee= r, QIOChannelSocket *sioc =3D qio_channel_socket_new(); =20 nc =3D qemu_new_net_client(&net_stream_info, peer, model, name); - s =3D DO_UPCAST(NetStreamState, nc, nc); - qemu_set_info_str(&s->nc, "connecting"); + s =3D DO_UPCAST(NetStreamState, data.nc, nc); + qemu_set_info_str(&s->data.nc, "connecting"); =20 - s->ioc =3D QIO_CHANNEL(sioc); - s->nc.link_down =3D true; + s->data.ioc =3D QIO_CHANNEL(sioc); + s->data.nc.link_down =3D true; + s->data.send =3D net_stream_send; + s->data.listen =3D net_stream_listen; =20 s->reconnect_ms =3D reconnect_ms; if (reconnect_ms) { diff --git a/net/stream_data.c b/net/stream_data.c new file mode 100644 index 0000000000..5af27e0d1d --- /dev/null +++ b/net/stream_data.c @@ -0,0 +1,193 @@ +/* + * net stream generic functions + * + * Copyright Red Hat + * + * SPDX-License-Identifier: GPL-2.0-or-later + */ + +#include "qemu/osdep.h" +#include "qemu/iov.h" +#include "qapi/error.h" +#include "net/net.h" +#include "io/channel.h" +#include "io/net-listener.h" + +#include "stream_data.h" + +static gboolean net_stream_data_writable(QIOChannel *ioc, + GIOCondition condition, gpointer = data) +{ + NetStreamData *d =3D data; + + d->ioc_write_tag =3D 0; + + qemu_flush_queued_packets(&d->nc); + + return G_SOURCE_REMOVE; +} + +ssize_t net_stream_data_receive(NetStreamData *d, const uint8_t *buf, + size_t size) +{ + uint32_t len =3D htonl(size); + struct iovec iov[] =3D { + { + .iov_base =3D &len, + .iov_len =3D sizeof(len), + }, { + .iov_base =3D (void *)buf, + .iov_len =3D size, + }, + }; + struct iovec local_iov[2]; + unsigned int nlocal_iov; + size_t remaining; + ssize_t ret; + + remaining =3D iov_size(iov, 2) - d->send_index; + nlocal_iov =3D iov_copy(local_iov, 2, iov, 2, d->send_index, remaining= ); + ret =3D qio_channel_writev(d->ioc, local_iov, nlocal_iov, NULL); + if (ret =3D=3D QIO_CHANNEL_ERR_BLOCK) { + ret =3D 0; /* handled further down */ + } + if (ret =3D=3D -1) { + d->send_index =3D 0; + return -errno; + } + if (ret < (ssize_t)remaining) { + d->send_index +=3D ret; + d->ioc_write_tag =3D qio_channel_add_watch(d->ioc, G_IO_OUT, + net_stream_data_writable,= d, + NULL); + return 0; + } + d->send_index =3D 0; + return size; +} + +static void net_stream_data_send_completed(NetClientState *nc, ssize_t len) +{ + NetStreamData *d =3D DO_UPCAST(NetStreamData, nc, nc); + + if (!d->ioc_read_tag) { + d->ioc_read_tag =3D qio_channel_add_watch(d->ioc, G_IO_IN, d->send= , d, + NULL); + } +} + +void net_stream_data_rs_finalize(SocketReadState *rs) +{ + NetStreamData *d =3D container_of(rs, NetStreamData, rs); + + if (qemu_send_packet_async(&d->nc, rs->buf, + rs->packet_len, + net_stream_data_send_completed) =3D=3D 0) { + if (d->ioc_read_tag) { + g_source_remove(d->ioc_read_tag); + d->ioc_read_tag =3D 0; + } + } +} + +gboolean net_stream_data_send(QIOChannel *ioc, GIOCondition condition, + NetStreamData *d) +{ + int size; + int ret; + QEMU_UNINITIALIZED char buf1[NET_BUFSIZE]; + const char *buf; + + size =3D qio_channel_read(d->ioc, buf1, sizeof(buf1), NULL); + if (size < 0) { + if (errno !=3D EWOULDBLOCK) { + goto eoc; + } + } else if (size =3D=3D 0) { + /* end of connection */ + eoc: + d->ioc_read_tag =3D 0; + if (d->ioc_write_tag) { + g_source_remove(d->ioc_write_tag); + d->ioc_write_tag =3D 0; + } + if (d->listener) { + qemu_set_info_str(&d->nc, "listening"); + qio_net_listener_set_client_func(d->listener, + d->listen, d, NULL); + } + object_unref(OBJECT(d->ioc)); + d->ioc =3D NULL; + + net_socket_rs_init(&d->rs, net_stream_data_rs_finalize, false); + d->nc.link_down =3D true; + + return G_SOURCE_REMOVE; + } + buf =3D buf1; + + ret =3D net_fill_rstate(&d->rs, (const uint8_t *)buf, size); + + if (ret =3D=3D -1) { + goto eoc; + } + + return G_SOURCE_CONTINUE; +} + +void net_stream_data_listen(QIONetListener *listener, + QIOChannelSocket *cioc, + NetStreamData *d) +{ + object_ref(OBJECT(cioc)); + + qio_net_listener_set_client_func(d->listener, NULL, d, NULL); + + d->ioc =3D QIO_CHANNEL(cioc); + qio_channel_set_name(d->ioc, "stream-server"); + d->nc.link_down =3D false; + + d->ioc_read_tag =3D qio_channel_add_watch(d->ioc, G_IO_IN, d->send, d,= NULL); +} + +int net_stream_data_client_connected(QIOTask *task, NetStreamData *d) +{ + QIOChannelSocket *sioc =3D QIO_CHANNEL_SOCKET(d->ioc); + SocketAddress *addr; + int ret; + Error *err =3D NULL; + + if (qio_task_propagate_error(task, &err)) { + qemu_set_info_str(&d->nc, "error: %s", error_get_pretty(err)); + error_free(err); + goto error; + } + + addr =3D qio_channel_socket_get_remote_address(sioc, NULL); + g_assert(addr !=3D NULL); + + ret =3D qemu_socket_try_set_nonblock(sioc->fd); + if (addr->type =3D=3D SOCKET_ADDRESS_TYPE_FD && ret < 0) { + qemu_set_info_str(&d->nc, "can't use file descriptor %s (errno %d)= ", + addr->u.fd.str, -ret); + qapi_free_SocketAddress(addr); + goto error; + } + g_assert(ret =3D=3D 0); + qapi_free_SocketAddress(addr); + + net_socket_rs_init(&d->rs, net_stream_data_rs_finalize, false); + + /* Disable Nagle algorithm on TCP sockets to reduce latency */ + qio_channel_set_delay(d->ioc, false); + + d->ioc_read_tag =3D qio_channel_add_watch(d->ioc, G_IO_IN, d->send, d,= NULL); + d->nc.link_down =3D false; + + return 0; +error: + object_unref(OBJECT(d->ioc)); + d->ioc =3D NULL; + + return -1; +} diff --git a/net/stream_data.h b/net/stream_data.h new file mode 100644 index 0000000000..b868625665 --- /dev/null +++ b/net/stream_data.h @@ -0,0 +1,31 @@ +/* + * net stream generic functions + * + * Copyright Red Hat + * + * SPDX-License-Identifier: GPL-2.0-or-later + */ + +typedef struct NetStreamData { + NetClientState nc; + QIOChannel *ioc; + guint ioc_read_tag; + guint ioc_write_tag; + SocketReadState rs; + unsigned int send_index; /* number of bytes sent*/ + QIOChannelFunc send; + /* server data */ + QIOChannel *listen_ioc; + QIONetListener *listener; + QIONetListenerClientFunc listen; +} NetStreamData; + +ssize_t net_stream_data_receive(NetStreamData *d, const uint8_t *buf, + size_t size); +void net_stream_data_rs_finalize(SocketReadState *rs); +gboolean net_stream_data_send(QIOChannel *ioc, GIOCondition condition, + NetStreamData *d); +int net_stream_data_client_connected(QIOTask *task, NetStreamData *d); +void net_stream_data_listen(QIONetListener *listener, + QIOChannelSocket *cioc, + NetStreamData *d); --=20 2.42.0