Re: [Qemu-devel] [RFC PATCH v4 06/15] net: port socket to GSource

[liu ping fan]

On Thu, Apr 18, 2013 at 10:34 PM, Stefan Hajnoczi <address@hidden> wrote:
> On Wed, Apr 17, 2013 at 04:39:15PM +0800, Liu Ping Fan wrote:
>> @@ -160,7 +154,13 @@ static void net_socket_send(void *opaque)
>>          net_socket_read_poll(s, false);
>>          net_socket_write_poll(s, false);
>>          if (s->listen_fd != -1) {
>> -            qemu_set_fd_handler(s->listen_fd, net_socket_accept, NULL, s);
>> +            nsrc = s->nsrc;
>> +            new_nsrc = event_source_new(s->listen_fd, 
>> net_socket_listen_handler,
>> +                                s);
>> +            s->nsrc = new_nsrc;
>> +            new_nsrc->gfd.events = G_IO_IN;
>> +            g_source_destroy(&nsrc->source);
>> +            s->nc.info->bind_ctx(&s->nc, NULL);
>
> The following is equivalent:
>
> event_source_release(s->nsrc);
> s->nsrc = event_source_new(s->listen_fd, net_socket_listen_handler, s);
> s->nc.info->bind_ctx(&s->nc, NULL);
>
> Then new_nsrc/nsrc can be dropped and the nsrc memory leak is avoided.
>
Apply here and the following same issues.
> Note that gfd.events = G_IO_IN does not get used since prepare()
> overwrites gfd.events.  Please drop and make sure read_poll == true.
>
Apply,
> I'm a little worried that we're lacking G_IO_HUP | G_IO_ERR.  Perhaps
> disconnect and network errors will be ignored.
>
NetSocketState can do limited things about these situation, perhaps,
implement net_socket_can_receive() to export such message to frontend
?

Pingfan
>>          }
>>          closesocket(s->fd);
>>
>> @@ -331,6 +331,14 @@ static void net_socket_cleanup(NetClientState *nc)
>>          closesocket(s->listen_fd);
>>          s->listen_fd = -1;
>>      }
>> +    event_source_release(s->nsrc);
>> +}
>> +
>> +static void net_socket_bind_ctx(NetClientState *nc, GMainContext *ctx)
>> +{
>> +    NetSocketState *s = DO_UPCAST(NetSocketState, nc, nc);
>> +
>> +    g_source_attach(&s->nsrc->source, ctx);
>>  }
>>
>>  static NetClientInfo net_dgram_socket_info = {
>> @@ -338,8 +346,22 @@ static NetClientInfo net_dgram_socket_info = {
>>      .size = sizeof(NetSocketState),
>>      .receive = net_socket_receive_dgram,
>>      .cleanup = net_socket_cleanup,
>> +    .bind_ctx = net_socket_bind_ctx,
>>  };
>>
>> +static gboolean net_socket_dgram_handler(gpointer data)
>> +{
>> +    EventGSource *nsrc = (EventGSource *)data;
>> +    NetSocketState *s = nsrc->opaque;
>> +
>> +    if (nsrc->gfd.revents & G_IO_IN) {
>> +        net_socket_send_dgram(s);
>> +    } else {
>> +        net_socket_writable(s);
>> +    }
>> +    return true;
>> +}
>> +
>>  static NetSocketState *net_socket_fd_init_dgram(NetClientState *peer,
>>                                                  const char *model,
>>                                                  const char *name,
>> @@ -350,6 +372,7 @@ static NetSocketState 
>> *net_socket_fd_init_dgram(NetClientState *peer,
>>      socklen_t saddr_len;
>>      NetClientState *nc;
>>      NetSocketState *s;
>> +    EventGSource *nsrc;
>>
>>      /* fd passed: multicast: "learn" dgram_dst address from bound address 
>> and save it
>>       * Because this may be "shared" socket from a "master" process, 
>> datagrams would be recv()
>> @@ -393,7 +416,10 @@ static NetSocketState 
>> *net_socket_fd_init_dgram(NetClientState *peer,
>>
>>      s->fd = fd;
>>      s->listen_fd = -1;
>> -    s->send_fn = net_socket_send_dgram;
>> +    nsrc = event_source_new(fd, net_socket_dgram_handler, s);
>> +    s->nsrc = nsrc;
>> +    nsrc->gfd.events = G_IO_IN|G_IO_OUT;
>
> Please drop.
>
>> +    nc->info->bind_ctx(nc, NULL);
>>      net_socket_read_poll(s, true);
>>
>>      /* mcast: save bound address as dst */
>> @@ -408,20 +434,28 @@ err:
>>      return NULL;
>>  }
>>
>> -static void net_socket_connect(void *opaque)
>> -{
>> -    NetSocketState *s = opaque;
>> -    s->send_fn = net_socket_send;
>> -    net_socket_read_poll(s, true);
>> -}
>> -
>>  static NetClientInfo net_socket_info = {
>>      .type = NET_CLIENT_OPTIONS_KIND_SOCKET,
>>      .size = sizeof(NetSocketState),
>>      .receive = net_socket_receive,
>>      .cleanup = net_socket_cleanup,
>> +    .bind_ctx = net_socket_bind_ctx,
>>  };
>>
>> +static gboolean net_socket_connect_handler(gpointer data)
>> +{
>> +    EventGSource *new_nsrc, *nsrc = data;
>> +    NetSocketState *s = nsrc->opaque;
>> +
>> +    new_nsrc = event_source_new(s->fd, net_socket_establish_handler, s);
>> +    s->nsrc = new_nsrc;
>> +    new_nsrc->gfd.events = G_IO_IN|G_IO_OUT;
>
> Please drop.
>
>> +    g_source_destroy(&nsrc->source);
>> +    s->nc.info->bind_ctx(&s->nc, NULL);
>> +
>> +    return true;
>> +}
>> +
>>  static NetSocketState *net_socket_fd_init_stream(NetClientState *peer,
>>                                                   const char *model,
>>                                                   const char *name,
>> @@ -429,6 +463,7 @@ static NetSocketState 
>> *net_socket_fd_init_stream(NetClientState *peer,
>>  {
>>      NetClientState *nc;
>>      NetSocketState *s;
>> +    EventGSource *nsrc;
>>
>>      nc = qemu_new_net_client(&net_socket_info, peer, model, name);
>>
>> @@ -440,9 +475,16 @@ static NetSocketState 
>> *net_socket_fd_init_stream(NetClientState *peer,
>>      s->listen_fd = -1;
>>
>>      if (is_connected) {
>> -        net_socket_connect(s);
>> +        nsrc = event_source_new(fd, net_socket_establish_handler, s);
>> +        s->nsrc = nsrc;
>> +        nsrc->gfd.events = G_IO_IN|G_IO_OUT;
>
> Please drop.
>
>> +        nc->info->bind_ctx(nc, NULL);
>>      } else {
>> -        qemu_set_fd_handler(s->fd, NULL, net_socket_connect, s);
>> +        nsrc = event_source_new(fd, net_socket_connect_handler, s);
>> +        s->nsrc = nsrc;
>> +        nsrc->gfd.events = G_IO_IN;
>
> Please drop.
>
>> +        nc->info->bind_ctx(nc, NULL);
>> +
>>      }
>>      return s;
>>  }
>> @@ -473,30 +515,69 @@ static NetSocketState 
>> *net_socket_fd_init(NetClientState *peer,
>>      return NULL;
>>  }
>>
>> -static void net_socket_accept(void *opaque)
>> +static gboolean net_socket_establish_handler(gpointer data)
>> +{
>> +    EventGSource *nsrc = (EventGSource *)data;
>> +    NetSocketState *s = nsrc->opaque;
>> +
>> +    if (nsrc->gfd.revents & G_IO_IN) {
>> +        net_socket_send(s);
>> +    } else {
>> +        net_socket_writable(s);
>> +    }
>> +    return true;
>> +}
>> +
>> +static bool readable(void *opaque)
>>  {
>>      NetSocketState *s = opaque;
>> +
>> +    if (s->read_poll && net_socket_can_send(s)) {
>> +        return true;
>> +    }
>> +    return false;
>> +}
>> +
>> +static bool writable(void *opaque)
>> +{
>> +    NetSocketState *s = opaque;
>> +
>> +    if (s->write_poll) {
>> +        return true;
>> +    }
>> +    return false;
>> +}
>> +
>> +static gboolean net_socket_listen_handler(gpointer data)
>> +{
>> +    EventGSource *new_nsrc, *nsrc = data;
>> +    NetSocketState *s = nsrc->opaque;
>>      struct sockaddr_in saddr;
>>      socklen_t len;
>>      int fd;
>>
>> -    for(;;) {
>> -        len = sizeof(saddr);
>> -        fd = qemu_accept(s->listen_fd, (struct sockaddr *)&saddr, &len);
>> -        if (fd < 0 && errno != EINTR) {
>> -            return;
>> -        } else if (fd >= 0) {
>> -            qemu_set_fd_handler(s->listen_fd, NULL, NULL, NULL);
>> -            break;
>> -        }
>> +    len = sizeof(saddr);
>> +    fd = qemu_accept(s->listen_fd, (struct sockaddr *)&saddr, &len);
>> +    if (fd < 0 && errno != EINTR) {
>> +        return false;
>>      }
>>
>>      s->fd = fd;
>>      s->nc.link_down = false;
>> -    net_socket_connect(s);
>> +    new_nsrc = event_source_new(fd, net_socket_establish_handler, s);
>> +    s->nsrc = new_nsrc;
>> +    new_nsrc->gfd.events = G_IO_IN|G_IO_OUT;
>
> Please drop.
>
>> +    new_nsrc->readable = readable;
>> +    new_nsrc->writable = writable;
>> +    /* prevent more than one connect req */
>> +    g_source_destroy(&nsrc->source);
>> +    s->nc.info->bind_ctx(&s->nc, NULL);
>> +    net_socket_read_poll(s, true);
>>      snprintf(s->nc.info_str, sizeof(s->nc.info_str),
>>               "socket: connection from %s:%d",
>>               inet_ntoa(saddr.sin_addr), ntohs(saddr.sin_port));
>> +
>> +    return true;
>>  }
>>
>>  static int net_socket_listen_init(NetClientState *peer,
>> @@ -508,6 +589,7 @@ static int net_socket_listen_init(NetClientState *peer,
>>      NetSocketState *s;
>>      struct sockaddr_in saddr;
>>      int fd, val, ret;
>> +    EventGSource *nsrc;
>>
>>      if (parse_host_port(&saddr, host_str) < 0)
>>          return -1;
>> @@ -542,7 +624,11 @@ static int net_socket_listen_init(NetClientState *peer,
>>      s->listen_fd = fd;
>>      s->nc.link_down = true;
>>
>> -    qemu_set_fd_handler(s->listen_fd, net_socket_accept, NULL, s);
>> +    nsrc = event_source_new(fd, net_socket_listen_handler, s);
>> +    s->nsrc = nsrc;
>> +    nsrc->gfd.events = G_IO_IN;
>
> Please drop.