net/stream.c

   1 /*
   2  * QEMU System Emulator
   3  *
   4  * Copyright (c) 2003-2008 Fabrice Bellard
   5  * Copyright (c) 2022 Red Hat, Inc.
   6  *
   7  * Permission is hereby granted, free of charge, to any person obtaining a copy
   8  * of this software and associated documentation files (the "Software"), to deal
   9  * in the Software without restriction, including without limitation the rights
  10  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11  * copies of the Software, and to permit persons to whom the Software is
  12  * furnished to do so, subject to the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be included in
  15  * all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  20  * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  23  * THE SOFTWARE.
  24  */
  25
  26 #include "qemu/osdep.h"
  27
  28 #include "net/net.h"
  29 #include "clients.h"
  30 #include "monitor/monitor.h"
  31 #include "qapi/error.h"
  32 #include "qemu/error-report.h"
  33 #include "qemu/option.h"
  34 #include "qemu/sockets.h"
  35 #include "qemu/iov.h"
  36 #include "qemu/main-loop.h"
  37 #include "qemu/cutils.h"
  38 #include "io/channel.h"
  39 #include "io/channel-socket.h"
  40 #include "io/net-listener.h"
  41 #include "qapi/qapi-events-net.h"
  42
  43 typedef struct NetStreamState {
  44     NetClientState nc;
  45     QIOChannel *listen_ioc;
  46     QIONetListener *listener;
  47     QIOChannel *ioc;
  48     guint ioc_read_tag;
  49     guint ioc_write_tag;
  50     SocketReadState rs;
  51     unsigned int send_index;      /* number of bytes sent*/
  52 } NetStreamState;
  53
  54 static void net_stream_listen(QIONetListener *listener,
  55                               QIOChannelSocket *cioc,
  56                               void *opaque);
  57
  58 static gboolean net_stream_writable(QIOChannel *ioc,
  59                                     GIOCondition condition,
  60                                     gpointer data)
  61 {
  62     NetStreamState *s = data;
  63
  64     s->ioc_write_tag = 0;
  65
  66     qemu_flush_queued_packets(&s->nc);
  67
  68     return G_SOURCE_REMOVE;
  69 }
  70
  71 static ssize_t net_stream_receive(NetClientState *nc, const uint8_t *buf,
  72                                   size_t size)
  73 {
  74     NetStreamState *s = DO_UPCAST(NetStreamState, nc, nc);
  75     uint32_t len = htonl(size);
  76     struct iovec iov[] = {
  77         {
  78             .iov_base = &len,
  79             .iov_len  = sizeof(len),
  80         }, {
  81             .iov_base = (void *)buf,
  82             .iov_len  = size,
  83         },
  84     };
  85     struct iovec local_iov[2];
  86     unsigned int nlocal_iov;
  87     size_t remaining;
  88     ssize_t ret;
  89
  90     remaining = iov_size(iov, 2) - s->send_index;
  91     nlocal_iov = iov_copy(local_iov, 2, iov, 2, s->send_index, remaining);
  92     ret = qio_channel_writev(s->ioc, local_iov, nlocal_iov, NULL);
  93     if (ret == QIO_CHANNEL_ERR_BLOCK) {
  94         ret = 0; /* handled further down */
  95     }
  96     if (ret == -1) {
  97         s->send_index = 0;
  98         return -errno;
  99     }
 100     if (ret < (ssize_t)remaining) {
 101         s->send_index += ret;
 102         s->ioc_write_tag = qio_channel_add_watch(s->ioc, G_IO_OUT,
 103                                                  net_stream_writable, s, NULL);
 104         return 0;
 105     }
 106     s->send_index = 0;
 107     return size;
 108 }
 109
 110 static gboolean net_stream_send(QIOChannel *ioc,
 111                                 GIOCondition condition,
 112                                 gpointer data);
 113
 114 static void net_stream_send_completed(NetClientState *nc, ssize_t len)
 115 {
 116     NetStreamState *s = DO_UPCAST(NetStreamState, nc, nc);
 117
 118     if (!s->ioc_read_tag) {
 119         s->ioc_read_tag = qio_channel_add_watch(s->ioc, G_IO_IN,
 120                                                 net_stream_send, s, NULL);
 121     }
 122 }
 123
 124 static void net_stream_rs_finalize(SocketReadState *rs)
 125 {
 126     NetStreamState *s = container_of(rs, NetStreamState, rs);
 127
 128     if (qemu_send_packet_async(&s->nc, rs->buf,
 129                                rs->packet_len,
 130                                net_stream_send_completed) == 0) {
 131         if (s->ioc_read_tag) {
 132             g_source_remove(s->ioc_read_tag);
 133             s->ioc_read_tag = 0;
 134         }
 135     }
 136 }
 137
 138 static gboolean net_stream_send(QIOChannel *ioc,
 139                                 GIOCondition condition,
 140                                 gpointer data)
 141 {
 142     NetStreamState *s = data;
 143     int size;
 144     int ret;
 145     char buf1[NET_BUFSIZE];
 146     const char *buf;
 147
 148     size = qio_channel_read(s->ioc, buf1, sizeof(buf1), NULL);
 149     if (size < 0) {
 150         if (errno != EWOULDBLOCK) {
 151             goto eoc;
 152         }
 153     } else if (size == 0) {
 154         /* end of connection */
 155     eoc:
 156         s->ioc_read_tag = 0;
 157         if (s->ioc_write_tag) {
 158             g_source_remove(s->ioc_write_tag);
 159             s->ioc_write_tag = 0;
 160         }
 161         if (s->listener) {
 162             qio_net_listener_set_client_func(s->listener, net_stream_listen,
 163                                              s, NULL);
 164         }
 165         object_unref(OBJECT(s->ioc));
 166         s->ioc = NULL;
 167
 168         net_socket_rs_init(&s->rs, net_stream_rs_finalize, false);
 169         s->nc.link_down = true;
 170         qemu_set_info_str(&s->nc, "%s", "");
 171
 172         qapi_event_send_netdev_stream_disconnected(s->nc.name);
 173
 174         return G_SOURCE_REMOVE;
 175     }
 176     buf = buf1;
 177
 178     ret = net_fill_rstate(&s->rs, (const uint8_t *)buf, size);
 179
 180     if (ret == -1) {
 181         goto eoc;
 182     }
 183
 184     return G_SOURCE_CONTINUE;
 185 }
 186
 187 static void net_stream_cleanup(NetClientState *nc)
 188 {
 189     NetStreamState *s = DO_UPCAST(NetStreamState, nc, nc);
 190     if (s->ioc) {
 191         if (QIO_CHANNEL_SOCKET(s->ioc)->fd != -1) {
 192             if (s->ioc_read_tag) {
 193                 g_source_remove(s->ioc_read_tag);
 194                 s->ioc_read_tag = 0;
 195             }
 196             if (s->ioc_write_tag) {
 197                 g_source_remove(s->ioc_write_tag);
 198                 s->ioc_write_tag = 0;
 199             }
 200         }
 201         object_unref(OBJECT(s->ioc));
 202         s->ioc = NULL;
 203     }
 204     if (s->listen_ioc) {
 205         if (s->listener) {
 206             qio_net_listener_disconnect(s->listener);
 207             object_unref(OBJECT(s->listener));
 208             s->listener = NULL;
 209         }
 210         object_unref(OBJECT(s->listen_ioc));
 211         s->listen_ioc = NULL;
 212     }
 213 }
 214
 215 static NetClientInfo net_stream_info = {
 216     .type = NET_CLIENT_DRIVER_STREAM,
 217     .size = sizeof(NetStreamState),
 218     .receive = net_stream_receive,
 219     .cleanup = net_stream_cleanup,
 220 };
 221
 222 static void net_stream_listen(QIONetListener *listener,
 223                               QIOChannelSocket *cioc,
 224                               void *opaque)
 225 {
 226     NetStreamState *s = opaque;
 227     SocketAddress *addr;
 228     char *uri;
 229
 230     object_ref(OBJECT(cioc));
 231
 232     qio_net_listener_set_client_func(s->listener, NULL, s, NULL);
 233
 234     s->ioc = QIO_CHANNEL(cioc);
 235     qio_channel_set_name(s->ioc, "stream-server");
 236     s->nc.link_down = false;
 237
 238     s->ioc_read_tag = qio_channel_add_watch(s->ioc, G_IO_IN, net_stream_send,
 239                                             s, NULL);
 240
 241     if (cioc->localAddr.ss_family == AF_UNIX) {
 242         addr = qio_channel_socket_get_local_address(cioc, NULL);
 243     } else {
 244         addr = qio_channel_socket_get_remote_address(cioc, NULL);
 245     }
 246     g_assert(addr != NULL);
 247     uri = socket_uri(addr);
 248     qemu_set_info_str(&s->nc, "%s", uri);
 249     g_free(uri);
 250     qapi_event_send_netdev_stream_connected(s->nc.name, addr);
 251     qapi_free_SocketAddress(addr);
 252 }
 253
 254 static void net_stream_server_listening(QIOTask *task, gpointer opaque)
 255 {
 256     NetStreamState *s = opaque;
 257     QIOChannelSocket *listen_sioc = QIO_CHANNEL_SOCKET(s->listen_ioc);
 258     SocketAddress *addr;
 259     int ret;
 260
 261     if (listen_sioc->fd < 0) {
 262         qemu_set_info_str(&s->nc, "connection error");
 263         return;
 264     }
 265
 266     addr = qio_channel_socket_get_local_address(listen_sioc, NULL);
 267     g_assert(addr != NULL);
 268     ret = qemu_socket_try_set_nonblock(listen_sioc->fd);
 269     if (addr->type == SOCKET_ADDRESS_TYPE_FD && ret < 0) {
 270         qemu_set_info_str(&s->nc, "can't use file descriptor %s (errno %d)",
 271                           addr->u.fd.str, -ret);
 272         return;
 273     }
 274     g_assert(ret == 0);
 275     qapi_free_SocketAddress(addr);
 276
 277     s->nc.link_down = true;
 278     s->listener = qio_net_listener_new();
 279
 280     net_socket_rs_init(&s->rs, net_stream_rs_finalize, false);
 281     qio_net_listener_set_client_func(s->listener, net_stream_listen, s, NULL);
 282     qio_net_listener_add(s->listener, listen_sioc);
 283 }
 284
 285 static int net_stream_server_init(NetClientState *peer,
 286                                   const char *model,
 287                                   const char *name,
 288                                   SocketAddress *addr,
 289                                   Error **errp)
 290 {
 291     NetClientState *nc;
 292     NetStreamState *s;
 293     QIOChannelSocket *listen_sioc = qio_channel_socket_new();
 294
 295     nc = qemu_new_net_client(&net_stream_info, peer, model, name);
 296     s = DO_UPCAST(NetStreamState, nc, nc);
 297
 298     s->listen_ioc = QIO_CHANNEL(listen_sioc);
 299     qio_channel_socket_listen_async(listen_sioc, addr, 0,
 300                                     net_stream_server_listening, s,
 301                                     NULL, NULL);
 302
 303     return 0;
 304 }
 305
 306 static void net_stream_client_connected(QIOTask *task, gpointer opaque)
 307 {
 308     NetStreamState *s = opaque;
 309     QIOChannelSocket *sioc = QIO_CHANNEL_SOCKET(s->ioc);
 310     SocketAddress *addr;
 311     gchar *uri;
 312     int ret;
 313
 314     if (sioc->fd < 0) {
 315         qemu_set_info_str(&s->nc, "connection error");
 316         goto error;
 317     }
 318
 319     addr = qio_channel_socket_get_remote_address(sioc, NULL);
 320     g_assert(addr != NULL);
 321     uri = socket_uri(addr);
 322     qemu_set_info_str(&s->nc, "%s", uri);
 323     g_free(uri);
 324
 325     ret = qemu_socket_try_set_nonblock(sioc->fd);
 326     if (addr->type == SOCKET_ADDRESS_TYPE_FD && ret < 0) {
 327         qemu_set_info_str(&s->nc, "can't use file descriptor %s (errno %d)",
 328                           addr->u.fd.str, -ret);
 329         qapi_free_SocketAddress(addr);
 330         goto error;
 331     }
 332     g_assert(ret == 0);
 333
 334     net_socket_rs_init(&s->rs, net_stream_rs_finalize, false);
 335
 336     /* Disable Nagle algorithm on TCP sockets to reduce latency */
 337     qio_channel_set_delay(s->ioc, false);
 338
 339     s->ioc_read_tag = qio_channel_add_watch(s->ioc, G_IO_IN, net_stream_send,
 340                                             s, NULL);
 341     s->nc.link_down = false;
 342     qapi_event_send_netdev_stream_connected(s->nc.name, addr);
 343     qapi_free_SocketAddress(addr);
 344
 345     return;
 346 error:
 347     object_unref(OBJECT(s->ioc));
 348     s->ioc = NULL;
 349 }
 350
 351 static int net_stream_client_init(NetClientState *peer,
 352                                   const char *model,
 353                                   const char *name,
 354                                   SocketAddress *addr,
 355                                   Error **errp)
 356 {
 357     NetStreamState *s;
 358     NetClientState *nc;
 359     QIOChannelSocket *sioc = qio_channel_socket_new();
 360
 361     nc = qemu_new_net_client(&net_stream_info, peer, model, name);
 362     s = DO_UPCAST(NetStreamState, nc, nc);
 363
 364     s->ioc = QIO_CHANNEL(sioc);
 365     s->nc.link_down = true;
 366
 367     qio_channel_socket_connect_async(sioc, addr,
 368                                      net_stream_client_connected, s,
 369                                      NULL, NULL);
 370
 371     return 0;
 372 }
 373
 374 int net_init_stream(const Netdev *netdev, const char *name,
 375                     NetClientState *peer, Error **errp)
 376 {
 377     const NetdevStreamOptions *sock;
 378
 379     assert(netdev->type == NET_CLIENT_DRIVER_STREAM);
 380     sock = &netdev->u.stream;
 381
 382     if (!sock->has_server || !sock->server) {
 383         return net_stream_client_init(peer, "stream", name, sock->addr, errp);
 384     }
 385     return net_stream_server_init(peer, "stream", name, sock->addr, errp);
 386 }