isc/unix/socket.c

	socket.c revision 4095b364155591b10bfe79c77e686031f6976852
/* $Id: socket.c,v 1.9 1998/11/15 11:48:17 explorer Exp $ */

#include "attribute.h"

#include <errno.h>
#include <stddef.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>

#include <sys/types.h>
#include <sys/socket.h>

#include <isc/assertions.h>
#include <isc/unexpect.h>
#include <isc/thread.h>
#include <isc/mutex.h>
#include <isc/condition.h>
#include <isc/socket.h>

#ifndef _WIN32
#define WINAPI /* we're not windows */
#endif

/*
 * We use macros instead of calling the routines directly because
 * the capital letters make the locking stand out.
 *
 * We INSIST that they succeed since there's no way for us to continue
 * if they fail.
 */

#define LOCK(lp) \
    INSIST(isc_mutex_lock((lp)) == ISC_R_SUCCESS);
#define UNLOCK(lp) \
    INSIST(isc_mutex_unlock((lp)) == ISC_R_SUCCESS);

/*
 * Debugging
 */
#if 1
#define XTRACE(a)   printf a
#define XENTER(a)   printf("ENTER %s\n", (a))
#define XEXIT(a)    printf("EXIT %s\n", (a))
#else
#define XTRACE(a)
#define XENTER(a)
#define XEXIT(a)
#endif

/*
 * internal event used to send readable/writable events to our internal
 * functions.
 */
typedef struct isc_socket_intev {
    struct isc_event        common;    /* Sender is the socket. */
    isc_task_t          task;      /* task to send these to */
    isc_socketevent_t       done_ev;   /* the done event to post */
    isc_boolean_t           partial;   /* partial i/o ok */
    isc_boolean_t           canceled;  /* I/O was canceled */
    LINK(struct isc_socket_intev)   link;
} *isc_socket_intev_t;

typedef struct isc_socket_ncintev {
    struct isc_event    common;
    isc_boolean_t       canceled;
    isc_task_t      task;
    isc_socket_newconnev_t  done;  /* the done event */
    LINK(struct isc_socket_ncintev) link;
} *isc_socket_ncintev_t;

#define SOCKET_MAGIC            0x494f696fU /* IOio */
#define VALID_SOCKET(t)         ((t) != NULL && \
                 (t)->magic == SOCKET_MAGIC)
struct isc_socket {
    /* Not locked. */
    unsigned int            magic;
    isc_socketmgr_t         manager;
    isc_mutex_t         lock;
    /* Locked by socket lock. */
    unsigned int            references;
    int             fd;
    LIST(struct isc_socket_intev)   read_list;
    LIST(struct isc_socket_intev)   write_list;
    LIST(struct isc_socket_ncintev) listen_list;
    isc_boolean_t           pending_read;
    isc_boolean_t           pending_write;
    isc_boolean_t           listener;  /* listener socket */
    isc_sockettype_t        type;
    isc_socket_intev_t      riev;
    isc_socket_intev_t      wiev;
    struct isc_sockaddr     address;
    unsigned int            addrlength;
};

#define SOCKET_MANAGER_MAGIC        0x494f6d67U /* IOmg */
#define VALID_MANAGER(m)        ((m) != NULL && \
                     (m)->magic == SOCKET_MANAGER_MAGIC)
struct isc_socketmgr {
    /* Not locked. */
    unsigned int            magic;
    isc_memctx_t            mctx;
    isc_mutex_t         lock;
    /* Locked by manager lock. */
    isc_boolean_t           done;
    unsigned int            nscheduled;
    unsigned int            nsockets;  /* sockets managed */
    isc_thread_t            thread;
    fd_set              read_fds;
    fd_set              write_fds;
    isc_socket_t            fds[FD_SETSIZE];
    int             maxfd;
    int             pipe_fds[2];
    sig_atomic_t            pipe_msgs;
};

#define SELECT_POKE_SHUTDOWN        (-1)
#define SELECT_POKE_NOTHING     (-2)

static void send_recvdone_event(isc_socket_t, isc_socket_intev_t *,
                isc_socketevent_t *, isc_result_t);
static void send_senddone_event(isc_socket_t, isc_socket_intev_t *,
                isc_socketevent_t *, isc_result_t);
static void rwdone_event_destroy(isc_event_t);
static void free_socket(isc_socket_t *);
static isc_result_t allocate_socket(isc_socketmgr_t, isc_sockettype_t,
                    isc_socket_t *);
static void destroy(isc_socket_t *);

/*
 * poke the select loop when there is something for us to do.  Manager must
 * be locked.
 */
static void
select_poke(isc_socketmgr_t mgr, int msg)
{
    int cc;

    XTRACE(("Poking watcher with data %d\n", msg));
    cc = write(mgr->pipe_fds[1], &msg, sizeof(int));
    if (cc < 0)
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "write() failed during watcher poke: %s",
                 strerror(errno));

    INSIST(cc == sizeof(int));
}

/*
 * read a message on the internal fd.
 */
static int
select_readmsg(isc_socketmgr_t mgr)
{
    int msg;
    int cc;

    cc = read(mgr->pipe_fds[0], &msg, sizeof(int));
    if (cc < 0) {
        if (errno == EWOULDBLOCK)
            return SELECT_POKE_NOTHING;

        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "read() failed during watcher poke: %s",
                 strerror(errno));
        return SELECT_POKE_NOTHING;  /* XXX */
    }

    INSIST(cc == sizeof(int));

    return msg;
}

/*
 * Make a fd non-blocking
 */
static isc_result_t
make_nonblock(int fd)
{
    int ret;
    int flags;

    flags = fcntl(fd, F_GETFL, 0);
    flags |= O_NONBLOCK;
    ret = fcntl(fd, F_SETFL, flags);

    if (ret == -1) {
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "fcntl(%d, F_SETFL, %d): %s",
                 fd, flags, strerror(errno));
        return ISC_R_UNEXPECTED;
    }

    return ISC_R_SUCCESS;
}

/*
 * Handle freeing a done event when needed.
 */
static void
rwdone_event_destroy(isc_event_t ev)
{
    isc_socket_t sock = ev->sender;
    isc_boolean_t kill_socket = ISC_FALSE;

    /*
     * detach from the socket.  We would have already detached from the
     * task when we actually queue this event up.
     */
    LOCK(&sock->lock);

    REQUIRE(sock->references > 0);
    sock->references--;
    XTRACE(("rwdone_event_destroy: sock %p, ref cnt == %d\n",
        sock, sock->references));

    if (sock->references == 0)
        kill_socket = ISC_TRUE;
    UNLOCK(&sock->lock);

    if (kill_socket)
        destroy(&sock);
}

static void
ncdone_event_destroy(isc_event_t ev)
{
    isc_socket_t sock = ev->sender;
    isc_boolean_t kill_socket = ISC_FALSE;

    /*
     * detach from the socket.  We would have already detached from the
     * task when we actually queue this event up.
     */
    LOCK(&sock->lock);
    sock->references--;

    if (sock->references == 0)
        kill_socket = ISC_TRUE;
    UNLOCK(&sock->lock);

    if (kill_socket)
        destroy(&sock);
}

/*
 * Kill.
 *
 * Caller must ensure locking.
 */
static void
destroy(isc_socket_t *sockp)
{
    isc_socket_t sock = *sockp;
    isc_socketmgr_t manager = sock->manager;

    XTRACE(("destroy sockp = %p, sock = %p\n", sockp, sock));

    LOCK(&manager->lock);

    /*
     * Noone has this socket open, so the watcher doesn't have to be
     * poked, and the socket doesn't have to be locked.
     */
    manager->fds[sock->fd] = NULL;
    manager->nsockets--;

    /*
     * XXX should reset manager->maxfd here
     */

    UNLOCK(&manager->lock);

    free_socket(sockp);
}

static isc_result_t
allocate_socket(isc_socketmgr_t manager, isc_sockettype_t type,
        isc_socket_t *socketp)
{
    isc_socket_t sock;

    sock = isc_mem_get(manager->mctx, sizeof *sock);

    if (sock == NULL)
        return (NULL);

    sock->magic = SOCKET_MAGIC;
    sock->references = 1;

    sock->manager = manager;
    sock->type = type;

    /*
     * set up list of readers and writers to be initially empty
     */
    INIT_LIST(sock->read_list);
    INIT_LIST(sock->write_list);
    INIT_LIST(sock->listen_list);
    sock->pending_read = ISC_FALSE;
    sock->pending_write = ISC_FALSE;

    /*
     * initialize the lock
     */
    if (isc_mutex_init(&sock->lock) != ISC_R_SUCCESS) {
        sock->magic = 0;
        isc_mem_put(manager->mctx, sock, sizeof *sock);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "isc_mutex_init() failed");
        return (ISC_R_UNEXPECTED);
    }

    *socketp = sock;

    return (ISC_R_SUCCESS);
}

/*
 * This event requires that the various lists be empty, that the reference
 * count be 1, and that the magic number is valid.  The other socket bits,
 * like the lock, must be initialized as well.  The fd associated must be
 * marked as closed, by setting it to -1 on close, or this routine will
 * also close the socket.
 */
static void
free_socket(isc_socket_t *socketp)
{
    isc_socket_t sock = *socketp;

    REQUIRE(sock->references == 0);
    REQUIRE(VALID_SOCKET(sock));
    REQUIRE(!sock->listener);
    REQUIRE(!sock->pending_read);
    REQUIRE(!sock->pending_write);
    REQUIRE(EMPTY(sock->read_list));
    REQUIRE(EMPTY(sock->write_list));
    REQUIRE(EMPTY(sock->listen_list));

    sock->magic = 0;

    if (sock->fd != -1) {
        close(sock->fd);
        sock->fd = -1;
    }

    (void)isc_mutex_destroy(&sock->lock);

    isc_mem_put(sock->manager->mctx, sock, sizeof *sock);
}

/*
 * Create a new 'type' socket managed by 'manager'.  The sockets
 * parameters are specified by 'expires' and 'interval'.  Events
 * will be posted to 'task' and when dispatched 'action' will be
 * called with 'arg' as the arg value.  The new socket is returned
 * in 'socketp'.
 */
isc_result_t
isc_socket_create(isc_socketmgr_t manager, isc_sockettype_t type,
          isc_socket_t *socketp)
{
    isc_socket_t sock = NULL;
    isc_result_t ret;

    REQUIRE(VALID_MANAGER(manager));
    REQUIRE(socketp != NULL && *socketp == NULL);

    XENTER("isc_socket_create");


    ret = allocate_socket(manager, type, &sock);
    if (ret != ISC_R_SUCCESS)
        return (ret);

    /*
     * Create the associated socket XXX
     */
    switch (type) {
    case isc_socket_udp:
        sock->fd = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
        break;
    case isc_socket_tcp:
        sock->fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
        break;
    }
    if (sock->fd < 0) {
        free_socket(&sock);

        switch (errno) {
        case EMFILE:
        case ENFILE:
        case ENOBUFS:
            return (ISC_R_NORESOURCES);
            break;
        default:
            UNEXPECTED_ERROR(__FILE__, __LINE__,
                     "socket() failed: %s",
                     strerror(errno));
            return (ISC_R_UNEXPECTED);
            break;
        }
    }

    if (make_nonblock(sock->fd) != ISC_R_SUCCESS) {
        free_socket(&sock);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "make_nonblock(%d)", sock->fd);
        return (ISC_R_UNEXPECTED);
    }

    LOCK(&manager->lock);

    /*
     * Note we don't have to lock the socket like we normally would because
     * there are no external references to it yet.
     */

    manager->fds[sock->fd] = sock;
    manager->nsockets++;
    if (manager->maxfd < sock->fd)
        manager->maxfd = sock->fd;

    UNLOCK(&manager->lock);

    sock->references++;

    *socketp = sock;

    XEXIT("isc_socket_create");

    return (ISC_R_SUCCESS);
}

/*
 * Attach to a socket.  Caller must explicitly detach when it is done.
 */
void
isc_socket_attach(isc_socket_t sock, isc_socket_t *socketp)
{
    REQUIRE(VALID_SOCKET(sock));
    REQUIRE(socketp != NULL && *socketp == NULL);

    LOCK(&sock->lock);
    sock->references++;
    UNLOCK(&sock->lock);

    *socketp = sock;
}

/*
 * Dereference a socket.  If this is the last reference to it, clean things
 * up by destroying the socket.
 */
void
isc_socket_detach(isc_socket_t *socketp)
{
    isc_socket_t sock;
    isc_boolean_t kill_socket = ISC_FALSE;

    REQUIRE(socketp != NULL);
    sock = *socketp;
    REQUIRE(VALID_SOCKET(sock));

    XENTER("isc_socket_detach");

    LOCK(&sock->lock);
    REQUIRE(sock->references > 0);
    sock->references--;
    if (sock->references == 0)
        kill_socket = ISC_TRUE;
    UNLOCK(&sock->lock);

    if (kill_socket)
        destroy(&sock);

    XEXIT("isc_socket_detach");

    *socketp = NULL;
}

/*
 * I/O is possible on a given socket.  Schedule an event to this task that
 * will call an internal function to do the I/O.  This will charge the
 * task with the I/O operation and let our select loop handler get back
 * to doing something real as fast as possible.
 *
 * The socket and manager must be locked before calling this function.
 */
static void
dispatch_read(isc_socket_t sock)
{
    isc_socket_intev_t iev;
    isc_event_t ev;

    iev = HEAD(sock->read_list);
    ev = (isc_event_t)iev;

    INSIST(!sock->pending_read);

    sock->pending_read = ISC_TRUE;

    XTRACE(("dispatch_read:  posted event %p to task %p\n",
        ev, iev->task));

    INSIST(isc_task_send(iev->task, &ev) == ISC_R_SUCCESS);
}

static void
dispatch_write(isc_socket_t sock)
{
    isc_socket_intev_t iev;
    isc_event_t ev;

    iev = HEAD(sock->write_list);
    ev = (isc_event_t)iev;

    INSIST(!sock->pending_write);
    sock->pending_write = ISC_TRUE;

    isc_task_send(iev->task, &ev);
}

static void
dispatch_listen(isc_socket_t sock)
{
    isc_socket_ncintev_t iev;
    isc_event_t ev;

    iev = HEAD(sock->listen_list);
    ev = (isc_event_t)iev;

    INSIST(!sock->pending_read);

    sock->pending_read = ISC_TRUE;

    isc_task_send(iev->task, &ev);
}

/*
 * Dequeue an item off the given socket's read queue, set the result code
 * in the done event to the one provided, and send it to the task it was
 * destined for.
 *
 * Caller must have the socket locked.
 */
static void
send_recvdone_event(isc_socket_t sock, isc_socket_intev_t *iev,
            isc_socketevent_t *dev, isc_result_t resultcode)
{
    REQUIRE(!EMPTY(sock->read_list));
    REQUIRE(iev != NULL);
    REQUIRE(*iev != NULL);
    REQUIRE(dev != NULL);
    REQUIRE(*dev != NULL);

    DEQUEUE(sock->read_list, *iev, link);
    (*dev)->result = resultcode;
    isc_task_send((*iev)->task, (isc_event_t *)dev);
    (*iev)->done_ev = NULL;
    isc_event_free((isc_event_t *)iev);
}
static void
send_senddone_event(isc_socket_t sock, isc_socket_intev_t *iev,
            isc_socketevent_t *dev, isc_result_t resultcode)
{
    REQUIRE(!EMPTY(sock->write_list));
    REQUIRE(iev != NULL);
    REQUIRE(*iev != NULL);
    REQUIRE(dev != NULL);
    REQUIRE(*dev != NULL);

    DEQUEUE(sock->write_list, *iev, link);
    (*dev)->result = resultcode;
    isc_task_send((*iev)->task, (isc_event_t *)dev);
    (*iev)->done_ev = NULL;
    isc_event_free((isc_event_t *)iev);
}

static void
send_ncdone_event(isc_socket_t sock, isc_socket_ncintev_t *iev,
          isc_socket_newconnev_t *dev, isc_result_t resultcode)
{
    REQUIRE(!EMPTY(sock->listen_list));
    REQUIRE(iev != NULL);
    REQUIRE(*iev != NULL);
    REQUIRE(dev != NULL);
    REQUIRE(*dev != NULL);

    DEQUEUE(sock->listen_list, *iev, link);
    (*dev)->result = resultcode;
    isc_task_send((*iev)->task, (isc_event_t *)dev);
    (*iev)->done = NULL;

    isc_event_free((isc_event_t *)iev);
}

/*
 * Call accept() on a socket, to get the new file descriptor.  The listen
 * socket is used as a prototype to create a new isc_socket_t.  The new
 * socket is referenced twice (one for the task which is receiving this
 * message, and once for the message itself) so the task does not need to
 * attach to the socket again.  The task is not attached at all.
 */
static isc_boolean_t
internal_accept(isc_task_t task, isc_event_t ev)
{
    isc_socket_t sock;
    isc_socket_newconnev_t dev;
    isc_socket_ncintev_t iev;
    struct sockaddr addr;
    u_int addrlen;
    int fd;

    sock = ev->sender;
    iev = (isc_socket_ncintev_t)ev;

    LOCK(&sock->lock);
    XTRACE(("internal_accept called, locked parent sock %p\n", sock));

    REQUIRE(sock->pending_read);
    REQUIRE(sock->listener);
    REQUIRE(!EMPTY(sock->listen_list));
    REQUIRE(iev->task == task);

    sock->pending_read = ISC_FALSE;

    /*
     * Has this event been canceled?
     */
    if (iev->canceled) {
        DEQUEUE(sock->listen_list, iev, link);
        isc_event_free((isc_event_t *)iev);
        if (!EMPTY(sock->listen_list))
            select_poke(sock->manager, sock->fd);

        UNLOCK(&sock->lock);

        return (0);
    }

    /*
     * Try to accept the new connection.  If the accept fails with
     * EWOULDBLOCK, simply poke the watcher to watch this socket
     * again.
     */
    addrlen = sizeof(addr);
    fd = accept(sock->fd, &addr, &addrlen);
    if (fd < 0) {
        if (errno == EWOULDBLOCK) {
            XTRACE(("internal_accept: ewouldblock\n"));
            sock->pending_read = ISC_FALSE;
            select_poke(sock->manager, sock->fd);
            UNLOCK(&sock->lock);
            return (0);
        }

        /*
         * If some other error, ignore it as well and hope
         * for the best, but log it.  XXX This will have to be
         * changed, thanks to broken OSs trying to overload what
         * accept does.
         */
        XTRACE(("internal_accept: accept returned %s\n",
            strerror(errno)));
        sock->pending_read = ISC_FALSE;
        select_poke(sock->manager, sock->fd);
        UNLOCK(&sock->lock);
        return (0);
    }

    /*
     * The accept succeeded.  Pull off the done event and set the
     * fd and other information in the socket descriptor here.  These
     * were preallocated for us.
     */
    dev = iev->done;
    iev->done = NULL;

    dev->newsocket->fd = fd;

    XTRACE(("internal_accept: newsock %p, fd %d\n",
        dev->newsocket, fd));

    sock->addrlength = addrlen;
    memcpy(&sock->address, &addr, addrlen);
    dev->addrlength = addrlen;
    memcpy(&dev->address, &addr, addrlen);

    UNLOCK(&sock->lock);

    /*
     * It's safe to do this, since the done event's free routine will
     * detach from the socket, so sock can't disappear out from under
     * us.
     */
    LOCK(&sock->manager->lock);
    sock->manager->fds[fd] = dev->newsocket;
    if (sock->manager->maxfd < fd)
        sock->manager->maxfd = fd;
    UNLOCK(&sock->manager->lock);

    send_ncdone_event(sock, &iev, &dev, ISC_R_SUCCESS);

    return (0);
}

static isc_boolean_t
internal_read(isc_task_t task, isc_event_t ev)
{
    isc_socket_intev_t iev;
    isc_socketevent_t dev;
    isc_socket_t sock;
    int cc;
    size_t read_count;
    struct sockaddr addr;
    u_int addrlen;

    /*
     * Find out what socket this is and lock it.
     */
    sock = (isc_socket_t)ev->sender;
    LOCK(&sock->lock);

    INSIST(sock->pending_read == ISC_TRUE);
    sock->pending_read = ISC_FALSE;

    XTRACE(("internal_read: sock %p, fd %d\n", sock, sock->fd));

    /*
     * Pull the first entry off the list, and look at it.  If it is
     * NULL, or not ours, something bad happened.
     */
    iev = HEAD(sock->read_list);
    INSIST(iev != NULL);
    INSIST(iev->task == task);

    /*
     * Try to do as much I/O as possible on this socket.  There are no
     * limits here, currently.  If some sort of quantum read count is
     * desired before giving up control, make certain to process markers
     * regardless of quantum.
     */
    do {
        iev = HEAD(sock->read_list);
        dev = iev->done_ev;

        /*
         * check for canceled I/O
         */
        if (iev->canceled) {
            DEQUEUE(sock->read_list, iev, link);
            isc_event_free((isc_event_t *)&iev);
            continue;
        }

        /*
         * If this is a marker event, post its completion and
         * continue the loop.
         */
        if (dev->common.type == ISC_SOCKEVENT_RECVMARK) {
            send_recvdone_event(sock, &iev, &dev, ISC_R_SUCCESS);
            continue;
        }

        /*
         * It must be a read request.  Try to satisfy it as best
         * we can.
         */
        read_count = dev->region.length - dev->n;
        if (sock->type == isc_socket_tcp) {
            cc = recv(sock->fd, dev->region.base + dev->n,
                  read_count, 0);
            memcpy(&dev->address, &sock->address,
                   sock->addrlength);
            dev->addrlength = sock->addrlength;
        } else {
            addrlen = sizeof(addr);
            cc = recvfrom(sock->fd, dev->region.base + dev->n,
                      read_count, 0,
                      (struct sockaddr *)&addr,
                      &addrlen);
            memcpy(&dev->address, &addr, addrlen);
            dev->addrlength = addrlen;
        }

        XTRACE(("internal_read:  read(%d) %d\n", sock->fd, cc));

        /*
         * check for error or block condition
         */
        if (cc < 0) {
            if (cc == EWOULDBLOCK)
                goto poke;
            UNEXPECTED_ERROR(__FILE__, __LINE__,
                     "internal read: %s",
                     strerror(errno));
            INSIST(cc >= 0);
        }
        /*
         * read of 0 means the remote end was closed.  Run through
         * the event queue and dispatch all the events with an EOF
         * result code.  This will set the EOF flag in markers as
         * well, but that's really ok.
         */
        if (cc == 0) {
            do {
                send_recvdone_event(sock, &iev, &dev,
                            ISC_R_EOF);
                iev = HEAD(sock->read_list);
            } while (iev != NULL);

            goto poke;
        }

        /*
         * if we read less than we expected, update counters,
         * poke.
         */
        if ((size_t)cc < read_count) {
            dev->n += cc;

            /*
             * If partial reads are allowed, we return whatever
             * was read with a success result, and continue
             * the loop.
             */
            if (iev->partial) {
                send_recvdone_event(sock, &iev, &dev,
                            ISC_R_SUCCESS);
                continue;
            }

            /*
             * Partials not ok.  Exit the loop and notify the
             * watcher to wait for more reads
             */
            goto poke;
        }

        /*
         * Exactly what we wanted to read.  We're done with this
         * entry.  Post its completion event.
         */
        if ((size_t)cc == read_count) {
            dev->n += read_count;
            send_recvdone_event(sock, &iev, &dev, ISC_R_SUCCESS);
        }

    } while (!EMPTY(sock->read_list));

 poke:
    if (!EMPTY(sock->read_list))
        select_poke(sock->manager, sock->fd);

    UNLOCK(&sock->lock);

    return (0);
}

static isc_boolean_t
internal_write(isc_task_t task, isc_event_t ev)
{
    isc_socket_intev_t iev;
    isc_socketevent_t dev;
    isc_socket_t sock;
    int cc;
    size_t write_count;

    /*
     * Find out what socket this is and lock it.
     */
    sock = (isc_socket_t)ev->sender;
    LOCK(&sock->lock);

    INSIST(sock->pending_write == ISC_TRUE);
    sock->pending_write = ISC_FALSE;

    XTRACE(("internal_write: sock %p, fd %d\n", sock, sock->fd));

    /*
     * Pull the first entry off the list, and look at it.  If it is
     * NULL, or not ours, something bad happened.
     */
    iev = HEAD(sock->write_list);
    INSIST(iev != NULL);
    INSIST(iev->task == task);

    /*
     * Try to do as much I/O as possible on this socket.  There are no
     * limits here, currently.  If some sort of quantum write count is
     * desired before giving up control, make certain to process markers
     * regardless of quantum.
     */
    do {
        iev = HEAD(sock->write_list);
        dev = iev->done_ev;

        /*
         * check for canceled I/O
         */
        if (iev->canceled) {
            DEQUEUE(sock->write_list, iev, link);
            isc_event_free((isc_event_t *)&iev);
            continue;
        }

        /*
         * If this is a marker event, post its completion and
         * continue the loop.
         */
        if (dev->common.type == ISC_SOCKEVENT_SENDMARK) {
            send_senddone_event(sock, &iev, &dev, ISC_R_SUCCESS);
            continue;
        }

        /*
         * It must be a write request.  Try to satisfy it as best
         * we can.
         */
        write_count = dev->region.length - dev->n;
        if (sock->type == isc_socket_udp)
            cc = sendto(sock->fd, dev->region.base + dev->n,
                    write_count, 0,
                    (struct sockaddr *)&dev->address,
                    dev->addrlength);
        else
            cc = send(sock->fd, dev->region.base + dev->n,
                  write_count, 0);

        XTRACE(("internal_write:  send(%d) %d\n", sock->fd, cc));

        /*
         * check for error or block condition
         */
        if (cc < 0) {
            if (cc == EWOULDBLOCK)
                goto poke;
            UNEXPECTED_ERROR(__FILE__, __LINE__,
                     "internal_write: %s",
                     strerror(errno));
            INSIST(cc >= 0);
        }
        /*
         * write of 0 means the remote end was closed.  Run through
         * the event queue and dispatch all the events with an EOF
         * result code.  This will set the EOF flag in markers as
         * well, but that's really ok.
         */
        if (cc == 0) {
            do {
                send_senddone_event(sock, &iev, &dev,
                          ISC_R_EOF);
                iev = HEAD(sock->write_list);
            } while (iev != NULL);

            goto poke;
        }

        /*
         * if we write less than we expected, update counters,
         * poke.
         */
        if ((size_t)cc < write_count) {
            dev->n += cc;

            goto poke;
        }

        /*
         * Exactly what we wanted to write.  We're done with this
         * entry.  Post its completion event.
         */
        if ((size_t)cc == write_count) {
            dev->n += write_count;
            send_senddone_event(sock, &iev, &dev, ISC_R_SUCCESS);
        }

    } while (!EMPTY(sock->write_list));

 poke:
    if (!EMPTY(sock->write_list))
        select_poke(sock->manager, sock->fd);

    UNLOCK(&sock->lock);

    return (0);
}

/*
 * This is the thread that will loop forever, always in a select or poll
 * call.
 *
 * When select returns something to do, track down what thread gets to do
 * this I/O and post the event to it.
 */
static isc_threadresult_t
WINAPI
watcher(void *uap)
{
    isc_socketmgr_t manager = uap;
    isc_socket_t sock;
    isc_boolean_t done;
    int ctlfd;
    int cc;
    fd_set readfds;
    fd_set writefds;
    int msg;
    isc_boolean_t unlock_sock;
    int i;
    isc_socket_intev_t  iev;
    isc_socket_ncintev_t    nciev;
    int maxfd;

    /*
     * Get the control fd here.  This will never change.
     */
    LOCK(&manager->lock);
    ctlfd = manager->pipe_fds[0];

    done = ISC_FALSE;
    while (!done) {
        do {
            readfds = manager->read_fds;
            writefds = manager->write_fds;
            maxfd = manager->maxfd + 1;

            UNLOCK(&manager->lock);

            cc = select(maxfd, &readfds, &writefds, NULL,
                    NULL);
            XTRACE(("select(%d, ...) == %d, errno %d\n",
                maxfd, cc, errno));
            if (cc < 0) {
                if (errno != EINTR)
                    UNEXPECTED_ERROR(__FILE__, __LINE__,
                             "select failed: %s",
                             strerror(errno));
            }

            LOCK(&manager->lock);
        } while (cc < 0);


        XTRACE(("watcher got manager lock\n"));

        /*
         * Process reads on internal, control fd.
         */
        if (FD_ISSET(ctlfd, &readfds)) {
            while (1) {
                msg = select_readmsg(manager);

                XTRACE(("watcher got message %d\n", msg));

                /*
                 * Nothing to read?
                 */
                if (msg == SELECT_POKE_NOTHING)
                    break;

                /*
                 * handle shutdown message.  We really should
                 * jump out of this loop right away, but
                 * it doesn't matter if we have to do a little
                 * more work first.
                 */
                if (msg == SELECT_POKE_SHUTDOWN)
                    done = ISC_TRUE;

                /*
                 * This is a wakeup on a socket.  Look
                 * at the event queue for both read and write,
                 * and decide if we need to watch on it now
                 * or not.
                 */
                if (msg >= 0) {
                    INSIST(msg < FD_SETSIZE);

                    sock = manager->fds[msg];
                    LOCK(&sock->lock);
                    XTRACE(("watcher locked socket %p\n",
                        sock));

                    /*
                     * If there are no events, or there
                     * is an event but we have already
                     * queued up the internal event on a
                     * task's queue, clear the bit.
                     * Otherwise, set it.
                     */
                    iev = HEAD(sock->read_list);
                    nciev = HEAD(sock->listen_list);
                    if ((iev == NULL && nciev == NULL)
                        || sock->pending_read) {
                        FD_CLR(sock->fd,
                               &manager->read_fds);
                        XTRACE(("watch cleared r\n"));
                    } else {
                        FD_SET(sock->fd,
                               &manager->read_fds);
                        XTRACE(("watch set r\n"));
                    }

                    iev = HEAD(sock->write_list);
                    if (iev == NULL || sock->pending_write) {
                        FD_CLR(sock->fd,
                               &manager->write_fds);
                        XTRACE(("watch cleared w\n"));
                    } else {
                        FD_SET(sock->fd,
                               &manager->write_fds);
                        XTRACE(("watch set w\n"));
                    }

                    UNLOCK(&sock->lock);
                }
            }
        }

        /*
         * Process read/writes on other fds here.  Avoid locking
         * and unlocking twice if both reads and writes are possible.
         */
        for (i = 0 ; i < maxfd ; i++) {
            if (manager->fds[i] != NULL) {
                sock = manager->fds[i];
                unlock_sock = ISC_FALSE;
                if (FD_ISSET(i, &readfds)) {
                    XTRACE(("watcher r on %d, sock %p\n",
                        i, manager->fds[i]));
                    unlock_sock = ISC_TRUE;
                    LOCK(&sock->lock);
                    if (sock->listener)
                        dispatch_listen(sock);
                    else
                        dispatch_read(sock);
                    FD_CLR(i, &manager->read_fds);
                }
                if (FD_ISSET(i, &writefds)) {
                    XTRACE(("watcher w on %d, sock %p\n",
                        i, manager->fds[i]));
                    if (!unlock_sock) {
                        unlock_sock = ISC_TRUE;
                        LOCK(&sock->lock);
                    }
                    dispatch_write(sock);
                    FD_CLR(i, &manager->write_fds);
                }
                if (unlock_sock)
                    UNLOCK(&sock->lock);
            }
        }
    }

    UNLOCK(&manager->lock);
    return ((isc_threadresult_t)0);
}

/*
 * Create a new socket manager.
 */
isc_result_t
isc_socketmgr_create(isc_memctx_t mctx, isc_socketmgr_t *managerp)
{
    isc_socketmgr_t manager;

    REQUIRE(managerp != NULL && *managerp == NULL);

    XENTER("isc_socketmgr_create");

    manager = isc_mem_get(mctx, sizeof *manager);
    if (manager == NULL)
        return (ISC_R_NOMEMORY);

    manager->magic = SOCKET_MANAGER_MAGIC;
    manager->mctx = mctx;
    manager->done = ISC_FALSE;
    memset(manager->fds, 0, sizeof(manager->fds));
    manager->nsockets = 0;
    manager->nscheduled = 0;
    if (isc_mutex_init(&manager->lock) != ISC_R_SUCCESS) {
        isc_mem_put(mctx, manager, sizeof *manager);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "isc_mutex_init() failed");
        return (ISC_R_UNEXPECTED);
    }

    /*
     * Create the special fds that will be used to wake up the
     * select/poll loop when something internal needs to be done.
     */
    if (pipe(manager->pipe_fds) != 0) {
        (void)isc_mutex_destroy(&manager->lock);
        isc_mem_put(mctx, manager, sizeof *manager);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "pipe() failed: %s",
                 strerror(errno)); /* XXX */

        return (ISC_R_UNEXPECTED);
    }

    INSIST(make_nonblock(manager->pipe_fds[0]) == ISC_R_SUCCESS);
    INSIST(make_nonblock(manager->pipe_fds[1]) == ISC_R_SUCCESS);

    /*
     * Set up initial state for the select loop
     */
    FD_ZERO(&manager->read_fds);
    FD_ZERO(&manager->write_fds);
    FD_SET(manager->pipe_fds[0], &manager->read_fds);
    manager->maxfd = manager->pipe_fds[0];

    /*
     * Start up the select/poll thread.
     */
    if (isc_thread_create(watcher, manager, &manager->thread) !=
        ISC_R_SUCCESS) {
        (void)isc_mutex_destroy(&manager->lock);
        isc_mem_put(mctx, manager, sizeof *manager);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "isc_thread_create() failed");
        return (ISC_R_UNEXPECTED);
    }

    *managerp = manager;

    XEXIT("isc_socketmgr_create (normal)");
    return (ISC_R_SUCCESS);
}

void
isc_socketmgr_destroy(isc_socketmgr_t *managerp)
{
    isc_socketmgr_t manager;

    /*
     * Destroy a socket manager.
     */

    REQUIRE(managerp != NULL);
    manager = *managerp;
    REQUIRE(VALID_MANAGER(manager));

    LOCK(&manager->lock);
    REQUIRE(manager->nsockets == 0);
    manager->done = ISC_TRUE;
    UNLOCK(&manager->lock);

    /*
     * Here, poke our select/poll thread.  Do this by closing the write
     * half of the pipe, which will send EOF to the read half.
     */
    select_poke(manager, SELECT_POKE_SHUTDOWN);

    /*
     * Wait for thread to exit.
     */
    if (isc_thread_join(manager->thread, NULL) != ISC_R_SUCCESS)
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "isc_thread_join() failed");

    /*
     * Clean up.
     */
    close(manager->pipe_fds[0]);
    close(manager->pipe_fds[1]);
    (void)isc_mutex_destroy(&manager->lock);
    manager->magic = 0;
    isc_mem_put(manager->mctx, manager, sizeof *manager);

    *managerp = NULL;
}

isc_result_t
isc_socket_recv(isc_socket_t sock, isc_region_t region,
        isc_boolean_t partial, isc_task_t task,
        isc_taskaction_t action, void *arg)
{
    isc_socketevent_t ev;
    isc_socket_intev_t iev;
    isc_socketmgr_t manager;
    isc_task_t ntask = NULL;
    int cc;

    manager = sock->manager;

    ev = (isc_socketevent_t)isc_event_allocate(manager->mctx, sock,
                           ISC_SOCKEVENT_RECVDONE,
                           action, arg, sizeof(*ev));
    if (ev == NULL)
        return (ISC_R_NOMEMORY);

    LOCK(&sock->lock);

    if (sock->riev == NULL) {
        iev = (isc_socket_intev_t)isc_event_allocate(manager->mctx,
                                 sock,
                                 ISC_SOCKEVENT_INTIO,
                                 internal_read,
                                 sock,
                                 sizeof(*iev));
        if (iev == NULL) {
            /* no special free routine yet */
            isc_event_free((isc_event_t *)&ev);
            return (ISC_R_NOMEMORY);
        }

        INIT_LINK(iev, link);

        sock->riev = iev;
        iev = NULL;  /* just in case */
    }

    sock->references++;  /* attach to socket in cheap way */

    /*
     * UDP sockets are always partial read
     */
    if (sock->type == isc_socket_udp)
        partial = ISC_TRUE;

    /*
     * Remember that we need to detach on event free
     */
    ev->common.destroy = rwdone_event_destroy;

    ev->region = *region;
    ev->n = 0;
    ev->result = ISC_R_SUCCESS;

    /*
     * If the read queue is empty, try to do the I/O right now.
     */
    if (EMPTY(sock->read_list)) {
        if (sock->type == isc_socket_udp) {
            cc = recvfrom(sock->fd, ev->region.base,
                      ev->region.length, 0,
                      (struct sockaddr *)&ev->address,
                      &ev->addrlength);
        } else {
            cc = recv(sock->fd, ev->region.base,
                  ev->region.length, 0);
            ev->address = sock->address;
            ev->addrlength = sock->addrlength;
        }

        if (cc < 0) {
            if (cc == EWOULDBLOCK)
                goto queue;
            UNEXPECTED_ERROR(__FILE__, __LINE__,
                     "isc_socket_recv: %s",
                     strerror(errno));
            INSIST(cc >= 0);
        }

        if (cc == 0) {
            ev->result = ISC_R_EOF;
            isc_task_send(task, (isc_event_t *)&ev);

            UNLOCK(&sock->lock);
            return (ISC_R_SUCCESS);
        }

        ev->n = cc;

        /*
         * Partial reads need to be queued
         */
        if ((size_t)cc != ev->region.length && !partial)
            goto queue;

        /*
         * full reads are posted, or partials if partials are ok.
         */
        isc_task_send(task, (isc_event_t *)&ev);

        UNLOCK(&sock->lock);

        return (ISC_R_SUCCESS);
    }

    /*
     * We couldn't read all or part of the request right now, so queue
     * it.
     */
 queue:
    iev = sock->riev;
    sock->riev = NULL;

    isc_task_attach(task, &ntask);

    iev->done_ev = ev;
    iev->task = ntask;
    iev->partial = partial;

    /*
     * Enqueue the request.  If the socket was previously not being
     * watched, poke the watcher to start paying attention to it.
     */
    if (EMPTY(sock->read_list)) {
        ENQUEUE(sock->read_list, iev, link);
        select_poke(sock->manager, sock->fd);
    } else {
        ENQUEUE(sock->read_list, iev, link);
    }

    XTRACE(("isc_socket_recv: posted ievent %p, dev %p, task %p\n",
        iev, iev->done_ev, task));

    UNLOCK(&sock->lock);

    return (ISC_R_SUCCESS);
}

isc_result_t
isc_socket_send(isc_socket_t sock, isc_region_t region,
        isc_task_t task, isc_taskaction_t action, void *arg)
{
    return isc_socket_sendto(sock, region, task, action, arg, NULL, 0);
}

isc_result_t
isc_socket_sendto(isc_socket_t sock, isc_region_t region,
          isc_task_t task, isc_taskaction_t action, void *arg,
          isc_sockaddr_t address, unsigned int addrlength)
{
    isc_socketevent_t ev;
    isc_socket_intev_t iev;
    isc_socketmgr_t manager;
    isc_task_t ntask = NULL;
    int cc;

    manager = sock->manager;

    ev = (isc_socketevent_t)isc_event_allocate(manager->mctx, sock,
                           ISC_SOCKEVENT_SENDDONE,
                           action, arg, sizeof(*ev));
    if (ev == NULL)
        return (ISC_R_NOMEMORY);

    LOCK(&sock->lock);

    if (sock->wiev == NULL) {
        iev = (isc_socket_intev_t)isc_event_allocate(manager->mctx,
                                 sock,
                                 ISC_SOCKEVENT_INTIO,
                                 internal_write,
                                 sock,
                                 sizeof(*iev));
        if (iev == NULL) {
            /* no special free routine yet */
            isc_event_free((isc_event_t *)&ev);
            return (ISC_R_NOMEMORY);
        }

        INIT_LINK(iev, link);

        sock->wiev = iev;
        iev = NULL;  /* just in case */
    }

    sock->references++;  /* attach to socket in cheap way */

    /*
     * Remember that we need to detach on event free
     */
    ev->common.destroy = rwdone_event_destroy;

    ev->region = *region;
    ev->n = 0;
    ev->result = ISC_R_SUCCESS;

    /*
     * If the write queue is empty, try to do the I/O right now.
     */
    if (sock->type == isc_socket_udp) {
        INSIST(addrlength > 0 || sock->addrlength > 0);
        if (addrlength > 0) {
            ev->address = *address;
            ev->addrlength = addrlength;
        } else if (sock->addrlength > 0) {
            ev->address = sock->address;
            ev->addrlength = sock->addrlength;
        }
    } else if (sock->type == isc_socket_tcp) {
        INSIST(address == NULL);
        INSIST(addrlength == 0);
        ev->address = sock->address;
        ev->addrlength = sock->addrlength;
    }
    if (EMPTY(sock->write_list)) {
        if (sock->type == isc_socket_udp)
            cc = sendto(sock->fd, ev->region.base,
                    ev->region.length, 0,
                    (struct sockaddr *)&ev->address,
                    ev->addrlength);
        else if (sock->type == isc_socket_tcp)
            cc = send(sock->fd, ev->region.base,
                  ev->region.length, 0);
        else
            cc = -1;  /* XXX */

        if (cc < 0) {
            if (cc == EWOULDBLOCK)
                goto queue;
            UNEXPECTED_ERROR(__FILE__, __LINE__,
                     "isc_socket_send: %s",
                     strerror(errno));
            INSIST(cc >= 0);
        }

        if (cc == 0) {
            ev->result = ISC_R_EOF;
            isc_task_send(task, (isc_event_t *)&ev);

            UNLOCK(&sock->lock);
            return (ISC_R_SUCCESS);
        }

        ev->n = cc;

        /*
         * Partial writes need to be queued
         */
        if ((size_t)cc != ev->region.length)
            goto queue;

        /*
         * full writes are posted.
         */
        isc_task_send(task, (isc_event_t *)&ev);

        UNLOCK(&sock->lock);

        return (ISC_R_SUCCESS);
    }

    /*
     * We couldn't send all or part of the request right now, so queue
     * it.
     */
 queue:
    iev = sock->wiev;
    sock->wiev = NULL;

    isc_task_attach(task, &ntask);

    iev->done_ev = ev;
    iev->task = ntask;
    iev->partial = ISC_FALSE; /* doesn't matter */

    /*
     * Enqueue the request.  If the socket was previously not being
     * watched, poke the watcher to start paying attention to it.
     */
    if (EMPTY(sock->write_list)) {
        ENQUEUE(sock->write_list, iev, link);
        select_poke(sock->manager, sock->fd);
    } else {
        ENQUEUE(sock->write_list, iev, link);
    }

    XTRACE(("isc_socket_send: posted ievent %p, dev %p, task %p\n",
        iev, iev->done_ev, task));

    UNLOCK(&sock->lock);

    return (ISC_R_SUCCESS);
}

isc_result_t
isc_socket_bind(isc_socket_t sock, struct isc_sockaddr *sockaddr,
        int addrlen)
{
    LOCK(&sock->lock);

    if (bind(sock->fd, (struct sockaddr *)sockaddr, addrlen) < 0) {
        UNLOCK(&sock->lock);
        switch (errno) {
        case EACCES:
            return (ISC_R_NOPERM);
            break;
        case EADDRNOTAVAIL:
            return (ISC_R_ADDRNOTAVAIL);
            break;
        case EADDRINUSE:
            return (ISC_R_ADDRINUSE);
            break;
        case EINVAL:
            return (ISC_R_BOUND);
            break;
        default:
            UNEXPECTED_ERROR(__FILE__, __LINE__,
                     "bind: %s", strerror(errno));
            return (ISC_R_UNEXPECTED);
            break;
        }
    }

    UNLOCK(&sock->lock);

    return (ISC_R_SUCCESS);
}

/*
 * set up to listen on a given socket.  We do this by creating an internal
 * event that will be dispatched when the socket has read activity.  The
 * watcher will send the internal event to the task when there is a new
 * connection.
 *
 * Unlike in read, we don't preallocate a done event here.  Every time there
 * is a new connection we'll have to allocate a new one anyway, so we might
 * as well keep things simple rather than having to track them.
 */
isc_result_t
isc_socket_listen(isc_socket_t sock, int backlog)
{
    REQUIRE(VALID_SOCKET(sock));
    REQUIRE(backlog >= 0);

    LOCK(&sock->lock);

    if (sock->type != isc_socket_tcp) {
        UNLOCK(&sock->lock);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "Socket is not isc_socket_tcp");
        return (ISC_R_UNEXPECTED);
    }

    if (sock->listener) {
        UNLOCK(&sock->lock);
        UNEXPECTED_ERROR(__FILE__, __LINE__,
                 "Socket already listener");
        return (ISC_R_UNEXPECTED);
    }

    if (backlog == 0)
        backlog = SOMAXCONN;

    if (listen(sock->fd, backlog) < 0) {
        UNLOCK(&sock->lock);
        UNEXPECTED_ERROR(__FILE__, __LINE__, "listen: %s",
                 strerror(errno));

        return (ISC_R_UNEXPECTED);
    }

    sock->listener = ISC_TRUE;
    UNLOCK(&sock->lock);

    return (ISC_R_SUCCESS);
}

isc_result_t
isc_socket_accept(isc_socket_t sock,
          isc_task_t task, isc_taskaction_t action, void *arg)
{
    isc_socket_ncintev_t iev;
    isc_socket_newconnev_t dev;
    isc_task_t ntask = NULL;
    isc_socketmgr_t manager;
    isc_socket_t nsock;
    isc_result_t ret;

    XENTER("isc_socket_accept");
    REQUIRE(VALID_SOCKET(sock));
    manager = sock->manager;
    REQUIRE(VALID_MANAGER(manager));

    LOCK(&sock->lock);

    REQUIRE(sock->listener);

    iev = (isc_socket_ncintev_t)isc_event_allocate(manager->mctx,
                               sock,
                               ISC_SOCKEVENT_INTCONN,
                               internal_accept,
                               sock,
                               sizeof(*iev));
    if (iev == NULL) {
        UNLOCK(&sock->lock);
        return (ISC_R_NOMEMORY);
    }

    dev = (isc_socket_newconnev_t)isc_event_allocate(manager->mctx,
                             sock,
                             ISC_SOCKEVENT_NEWCONN,
                             action,
                             arg,
                             sizeof (*dev));
    if (dev == NULL) {
        UNLOCK(&sock->lock);
        isc_event_free((isc_event_t *)&iev);
        return (ISC_R_NOMEMORY);
    }


    ret = allocate_socket(manager, sock->type, &nsock);
    if (ret != ISC_R_SUCCESS) {
        UNLOCK(&sock->lock);
        isc_event_free((isc_event_t *)&iev);
        isc_event_free((isc_event_t *)&dev);
        return (ret);
    }

    INIT_LINK(iev, link);

    /*
     * Attach to socket and to task
     */
    isc_task_attach(task, &ntask);
    sock->references++;

    sock->listener = ISC_TRUE;

    iev->task = ntask;
    iev->done = dev;
    iev->canceled = ISC_FALSE;
    dev->common.destroy = ncdone_event_destroy;
    dev->newsocket = nsock;

    /*
     * poke watcher here.  We still have the socket locked, so there
     * is no race condition.  We will keep the lock for such a short
     * bit of time waking it up now or later won't matter all that much.
     */
    if (EMPTY(sock->listen_list))
        select_poke(manager, sock->fd);

    ENQUEUE(sock->listen_list, iev, link);

    UNLOCK(&sock->lock);

    return (ISC_R_SUCCESS);
}