kern/48725: recvmmsg erroneous EAGAIN on nonblocking socket when all msgs aren't filled

[seanb%netbsd.org@localhost]

>Number:         48725
>Category:       kern
>Synopsis:       recvmmsg erroneous EAGAIN on nonblocking socket when all msgs 
>aren't filled
>Confidential:   no
>Severity:       serious
>Priority:       medium
>Responsible:    kern-bug-people
>State:          open
>Class:          sw-bug
>Submitter-Id:   net
>Arrival-Date:   Mon Apr 07 15:15:00 +0000 2014
>Originator:     Sean Boudreau
>Release:        6.99.39
>Organization:
NetBSD
>Environment:
NetBSD  6.99.39 NetBSD 6.99.39 (seanb) #4: Wed Apr  2 11:39:05 EDT 2014  
seanb@:/home/seanb/cvs_netbsd/HEAD/src/sys/arch/i386/compile/obj/seanb i386
>Description:
An erroneous EAGAIN can be returned when receiving on a non-blocking socket 
with recvmmsg with a mmsg array containing multiple members.  This occurs when 
some, but not all of the mmsg array members are filled with data.
>How-To-Repeat:
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <pthread.h>
#include <string.h>
#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <time.h>
#include <poll.h>
#include <err.h>

#define MAX_PACKETS 2
#define BUFSIZE 2048

void *
runRecvThread(void *arg)
{
        struct pollfd pfd;
        int ret, fd;
        struct mmsghdr  msgs[MAX_PACKETS];
        struct iovec    iovecs[MAX_PACKETS];
        char            bufs[MAX_PACKETS][BUFSIZE+1];
        int             retval, i;

        fd = *(int *)arg;
        pfd.fd = fd;
        pfd.events = POLLIN;
        pfd.revents = 0;

        memset(msgs, 0, sizeof(msgs));

        for (i = 0; i < MAX_PACKETS; i++){
                iovecs[i].iov_base = bufs[i];
                iovecs[i].iov_len = BUFSIZE;
                msgs[i].msg_hdr.msg_name = NULL;
                msgs[i].msg_hdr.msg_namelen = 0;
                msgs[i].msg_hdr.msg_control = NULL;
                msgs[i].msg_hdr.msg_controllen = 0;
                msgs[i].msg_hdr.msg_flags = 0;
                msgs[i].msg_hdr.msg_iovlen = 1;
                msgs[i].msg_hdr.msg_iov = &iovecs[i];
                msgs[i].msg_hdr.msg_iovlen = 1;
        }

        for (;;) {
                ret = poll(&pfd, 1, -1);
                if (ret == 1 && (pfd.revents & POLLIN)) {
                        retval = recvmmsg(fd, msgs, MAX_PACKETS, 0, 0);
                        if (retval == -1)
                                err(1, "recvmmsg");
                        printf("got data %d\n", retval);
                        for (i = 0; i < retval; i++)
                                printf("\t%d %s\n", i, bufs[i]);
                } else {
                        errx(1, "unexpected return from poll: %d\n", ret);
                }
        }
}

int
main(int argc, char *argv[])
{
        int fds[2];
        pthread_t recvT;

        if (socketpair(AF_LOCAL, SOCK_DGRAM, 0, fds) == -1)
                err(1, "socketpair");

        fcntl(fds[1], F_SETFL, fcntl(fds[1], F_GETFL, 0) | O_NONBLOCK);
        pthread_create(&recvT, NULL, &runRecvThread, &fds[1]);
        for (;;) {
                sleep(1);
                send(fds[0], "data\n", sizeof("data\n"), 0);
        }
        pthread_join(recvT, NULL);
        return 0;
}
>Fix: