path: root/fluent-bit/lib/librdkafka-2.1.0/src/rdkafka_transport.c

/*
 * librdkafka - Apache Kafka C library
 *
 * Copyright (c) 2015, Magnus Edenhill
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * 1. Redistributions of source code must retain the above copyright notice,
 *    this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */
#ifdef _WIN32
#pragma comment(lib, "ws2_32.lib")
#endif

#define __need_IOV_MAX

#define _DARWIN_C_SOURCE /* MSG_DONTWAIT */

#include "rdkafka_int.h"
#include "rdaddr.h"
#include "rdkafka_transport.h"
#include "rdkafka_transport_int.h"
#include "rdkafka_broker.h"
#include "rdkafka_interceptor.h"

#include <errno.h>

/* AIX doesn't have MSG_DONTWAIT */
#ifndef MSG_DONTWAIT
#define MSG_DONTWAIT MSG_NONBLOCK
#endif

#if WITH_SSL
#include "rdkafka_ssl.h"
#endif

/**< Current thread's rd_kafka_transport_t instance.
 *   This pointer is set up when calling any OpenSSL APIs that might
 *   trigger SSL callbacks, and is used to retrieve the SSL object's
 *   corresponding rd_kafka_transport_t instance.
 *   There is an set/get_ex_data() API in OpenSSL, but it requires storing
 *   a unique index somewhere, which we can't do without having a singleton
 *   object, so instead we cut out the middle man and store the
 *   rd_kafka_transport_t pointer directly in the thread-local memory. */
RD_TLS rd_kafka_transport_t *rd_kafka_curr_transport;


static int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout);


/**
 * Low-level socket close
 */
static void rd_kafka_transport_close0(rd_kafka_t *rk, rd_socket_t s) {
        if (rk->rk_conf.closesocket_cb)
                rk->rk_conf.closesocket_cb((int)s, rk->rk_conf.opaque);
        else
                rd_socket_close(s);
}

/**
 * Close and destroy a transport handle
 */
void rd_kafka_transport_close(rd_kafka_transport_t *rktrans) {
#if WITH_SSL
        rd_kafka_curr_transport = rktrans;
        if (rktrans->rktrans_ssl)
                rd_kafka_transport_ssl_close(rktrans);
#endif

        rd_kafka_sasl_close(rktrans);

        if (rktrans->rktrans_recv_buf)
                rd_kafka_buf_destroy(rktrans->rktrans_recv_buf);

#ifdef _WIN32
        WSACloseEvent(rktrans->rktrans_wsaevent);
#endif

        if (rktrans->rktrans_s != -1)
                rd_kafka_transport_close0(rktrans->rktrans_rkb->rkb_rk,
                                          rktrans->rktrans_s);

        rd_free(rktrans);
}

/**
 * @brief shutdown(2) a transport's underlying socket.
 *
 * This will prohibit further sends and receives.
 * rd_kafka_transport_close() must still be called to close the socket.
 */
void rd_kafka_transport_shutdown(rd_kafka_transport_t *rktrans) {
        shutdown(rktrans->rktrans_s,
#ifdef _WIN32
                 SD_BOTH
#else
                 SHUT_RDWR
#endif
        );
}


#ifndef _WIN32
/**
 * @brief sendmsg() abstraction, converting a list of segments to iovecs.
 * @remark should only be called if the number of segments is > 1.
 */
static ssize_t rd_kafka_transport_socket_sendmsg(rd_kafka_transport_t *rktrans,
                                                 rd_slice_t *slice,
                                                 char *errstr,
                                                 size_t errstr_size) {
        struct iovec iov[IOV_MAX];
        struct msghdr msg = {.msg_iov = iov};
        size_t iovlen;
        ssize_t r;
        size_t r2;

        rd_slice_get_iov(slice, msg.msg_iov, &iovlen, IOV_MAX,
                         /* FIXME: Measure the effects of this */
                         rktrans->rktrans_sndbuf_size);
        msg.msg_iovlen = (int)iovlen;

#ifdef __sun
        /* See recvmsg() comment. Setting it here to be safe. */
        rd_socket_errno = EAGAIN;
#endif

        r = sendmsg(rktrans->rktrans_s, &msg,
                    MSG_DONTWAIT
#ifdef MSG_NOSIGNAL
                        | MSG_NOSIGNAL
#endif
        );

        if (r == -1) {
                if (rd_socket_errno == EAGAIN)
                        return 0;
                rd_snprintf(errstr, errstr_size, "%s", rd_strerror(errno));
                return -1;
        }

        /* Update buffer read position */
        r2 = rd_slice_read(slice, NULL, (size_t)r);
        rd_assert((size_t)r == r2 &&
                  *"BUG: wrote more bytes than available in slice");

        return r;
}
#endif


/**
 * @brief Plain send() abstraction
 */
static ssize_t rd_kafka_transport_socket_send0(rd_kafka_transport_t *rktrans,
                                               rd_slice_t *slice,
                                               char *errstr,
                                               size_t errstr_size) {
        ssize_t sum = 0;
        const void *p;
        size_t rlen;

        while ((rlen = rd_slice_peeker(slice, &p))) {
                ssize_t r;
                size_t r2;

                r = send(rktrans->rktrans_s, p,
#ifdef _WIN32
                         (int)rlen, (int)0
#else
                         rlen, 0
#endif
                );

#ifdef _WIN32
                if (unlikely(r == RD_SOCKET_ERROR)) {
                        if (sum > 0 || rd_socket_errno == WSAEWOULDBLOCK) {
                                rktrans->rktrans_blocked = rd_true;
                                return sum;
                        } else {
                                rd_snprintf(
                                    errstr, errstr_size, "%s",
                                    rd_socket_strerror(rd_socket_errno));
                                return -1;
                        }
                }

                rktrans->rktrans_blocked = rd_false;
#else
                if (unlikely(r <= 0)) {
                        if (r == 0 || rd_socket_errno == EAGAIN)
                                return 0;
                        rd_snprintf(errstr, errstr_size, "%s",
                                    rd_socket_strerror(rd_socket_errno));
                        return -1;
                }
#endif

                /* Update buffer read position */
                r2 = rd_slice_read(slice, NULL, (size_t)r);
                rd_assert((size_t)r == r2 &&
                          *"BUG: wrote more bytes than available in slice");


                sum += r;

                /* FIXME: remove this and try again immediately and let
                 *        the next write() call fail instead? */
                if ((size_t)r < rlen)
                        break;
        }

        return sum;
}


static ssize_t rd_kafka_transport_socket_send(rd_kafka_transport_t *rktrans,
                                              rd_slice_t *slice,
                                              char *errstr,
                                              size_t errstr_size) {
#ifndef _WIN32
        /* FIXME: Use sendmsg() with iovecs if there's more than one segment
         * remaining, otherwise (or if platform does not have sendmsg)
         * use plain send(). */
        return rd_kafka_transport_socket_sendmsg(rktrans, slice, errstr,
                                                 errstr_size);
#endif
        return rd_kafka_transport_socket_send0(rktrans, slice, errstr,
                                               errstr_size);
}



#ifndef _WIN32
/**
 * @brief recvmsg() abstraction, converting a list of segments to iovecs.
 * @remark should only be called if the number of segments is > 1.
 */
static ssize_t rd_kafka_transport_socket_recvmsg(rd_kafka_transport_t *rktrans,
                                                 rd_buf_t *rbuf,
                                                 char *errstr,
                                                 size_t errstr_size) {
        ssize_t r;
        struct iovec iov[IOV_MAX];
        struct msghdr msg = {.msg_iov = iov};
        size_t iovlen;

        rd_buf_get_write_iov(rbuf, msg.msg_iov, &iovlen, IOV_MAX,
                             /* FIXME: Measure the effects of this */
                             rktrans->rktrans_rcvbuf_size);
        msg.msg_iovlen = (int)iovlen;

#ifdef __sun
        /* SunOS doesn't seem to set errno when recvmsg() fails
         * due to no data and MSG_DONTWAIT is set. */
        rd_socket_errno = EAGAIN;
#endif
        r = recvmsg(rktrans->rktrans_s, &msg, MSG_DONTWAIT);
        if (unlikely(r <= 0)) {
                if (r == -1 && rd_socket_errno == EAGAIN)
                        return 0;
                else if (r == 0 || (r == -1 && rd_socket_errno == ECONNRESET)) {
                        /* Receive 0 after POLLIN event means
                         * connection closed. */
                        rd_snprintf(errstr, errstr_size, "Disconnected");
                        return -1;
                } else if (r == -1) {
                        rd_snprintf(errstr, errstr_size, "%s",
                                    rd_strerror(errno));
                        return -1;
                }
        }

        /* Update buffer write position */
        rd_buf_write(rbuf, NULL, (size_t)r);

        return r;
}
#endif


/**
 * @brief Plain recv()
 */
static ssize_t rd_kafka_transport_socket_recv0(rd_kafka_transport_t *rktrans,
                                               rd_buf_t *rbuf,
                                               char *errstr,
                                               size_t errstr_size) {
        ssize_t sum = 0;
        void *p;
        size_t len;

        while ((len = rd_buf_get_writable(rbuf, &p))) {
                ssize_t r;

                r = recv(rktrans->rktrans_s, p,
#ifdef _WIN32
                         (int)
#endif
                             len,
                         0);

                if (unlikely(r == RD_SOCKET_ERROR)) {
                        if (rd_socket_errno == EAGAIN
#ifdef _WIN32
                            || rd_socket_errno == WSAEWOULDBLOCK
#endif
                        )
                                return sum;
                        else {
                                rd_snprintf(
                                    errstr, errstr_size, "%s",
                                    rd_socket_strerror(rd_socket_errno));
                                return -1;
                        }
                } else if (unlikely(r == 0)) {
                        /* Receive 0 after POLLIN event means
                         * connection closed. */
                        rd_snprintf(errstr, errstr_size, "Disconnected");
                        return -1;
                }

                /* Update buffer write position */
                rd_buf_write(rbuf, NULL, (size_t)r);

                sum += r;

                /* FIXME: remove this and try again immediately and let
                 *        the next recv() call fail instead? */
                if ((size_t)r < len)
                        break;
        }
        return sum;
}


static ssize_t rd_kafka_transport_socket_recv(rd_kafka_transport_t *rktrans,
                                              rd_buf_t *buf,
                                              char *errstr,
                                              size_t errstr_size) {
#ifndef _WIN32
        return rd_kafka_transport_socket_recvmsg(rktrans, buf, errstr,
                                                 errstr_size);
#endif
        return rd_kafka_transport_socket_recv0(rktrans, buf, errstr,
                                               errstr_size);
}



/**
 * CONNECT state is failed (errstr!=NULL) or done (TCP is up, SSL is working..).
 * From this state we either hand control back to the broker code,
 * or if authentication is configured we ente the AUTH state.
 */
void rd_kafka_transport_connect_done(rd_kafka_transport_t *rktrans,
                                     char *errstr) {
        rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;

        rd_kafka_curr_transport = rktrans;

        rd_kafka_broker_connect_done(rkb, errstr);
}



ssize_t rd_kafka_transport_send(rd_kafka_transport_t *rktrans,
                                rd_slice_t *slice,
                                char *errstr,
                                size_t errstr_size) {
        ssize_t r;
#if WITH_SSL
        if (rktrans->rktrans_ssl) {
                rd_kafka_curr_transport = rktrans;
                r = rd_kafka_transport_ssl_send(rktrans, slice, errstr,
                                                errstr_size);
        } else
#endif
                r = rd_kafka_transport_socket_send(rktrans, slice, errstr,
                                                   errstr_size);

        return r;
}


ssize_t rd_kafka_transport_recv(rd_kafka_transport_t *rktrans,
                                rd_buf_t *rbuf,
                                char *errstr,
                                size_t errstr_size) {
        ssize_t r;

#if WITH_SSL
        if (rktrans->rktrans_ssl) {
                rd_kafka_curr_transport = rktrans;
                r = rd_kafka_transport_ssl_recv(rktrans, rbuf, errstr,
                                                errstr_size);
        } else
#endif
                r = rd_kafka_transport_socket_recv(rktrans, rbuf, errstr,
                                                   errstr_size);

        return r;
}



/**
 * @brief Notify transport layer of full request sent.
 */
void rd_kafka_transport_request_sent(rd_kafka_broker_t *rkb,
                                     rd_kafka_buf_t *rkbuf) {
        rd_kafka_transport_t *rktrans = rkb->rkb_transport;

        /* Call on_request_sent interceptors */
        rd_kafka_interceptors_on_request_sent(
            rkb->rkb_rk, (int)rktrans->rktrans_s, rkb->rkb_name,
            rkb->rkb_nodeid, rkbuf->rkbuf_reqhdr.ApiKey,
            rkbuf->rkbuf_reqhdr.ApiVersion, rkbuf->rkbuf_corrid,
            rd_slice_size(&rkbuf->rkbuf_reader));
}



/**
 * Length framed receive handling.
 * Currently only supports a the following framing:
 *     [int32_t:big_endian_length_of_payload][payload]
 *
 * To be used on POLLIN event, will return:
 *   -1: on fatal error (errstr will be updated, *rkbufp remains unset)
 *    0: still waiting for data (*rkbufp remains unset)
 *    1: data complete, (buffer returned in *rkbufp)
 */
int rd_kafka_transport_framed_recv(rd_kafka_transport_t *rktrans,
                                   rd_kafka_buf_t **rkbufp,
                                   char *errstr,
                                   size_t errstr_size) {
        rd_kafka_buf_t *rkbuf = rktrans->rktrans_recv_buf;
        ssize_t r;
        const int log_decode_errors = LOG_ERR;

        /* States:
         *   !rktrans_recv_buf: initial state; set up buf to receive header.
         *    rkbuf_totlen == 0:   awaiting header
         *    rkbuf_totlen > 0:    awaiting payload
         */

        if (!rkbuf) {
                rkbuf = rd_kafka_buf_new(1, 4 /*length field's length*/);
                /* Set up buffer reader for the length field */
                rd_buf_write_ensure(&rkbuf->rkbuf_buf, 4, 4);
                rktrans->rktrans_recv_buf = rkbuf;
        }


        r = rd_kafka_transport_recv(rktrans, &rkbuf->rkbuf_buf, errstr,
                                    errstr_size);
        if (r == 0)
                return 0;
        else if (r == -1)
                return -1;

        if (rkbuf->rkbuf_totlen == 0) {
                /* Frame length not known yet. */
                int32_t frame_len;

                if (rd_buf_write_pos(&rkbuf->rkbuf_buf) < sizeof(frame_len)) {
                        /* Wait for entire frame header. */
                        return 0;
                }

                /* Initialize reader */
                rd_slice_init(&rkbuf->rkbuf_reader, &rkbuf->rkbuf_buf, 0, 4);

                /* Reader header: payload length */
                rd_kafka_buf_read_i32(rkbuf, &frame_len);

                if (frame_len < 0 ||
                    frame_len > rktrans->rktrans_rkb->rkb_rk->rk_conf
                                    .recv_max_msg_size) {
                        rd_snprintf(errstr, errstr_size,
                                    "Invalid frame size %" PRId32, frame_len);
                        return -1;
                }

                rkbuf->rkbuf_totlen = 4 + frame_len;
                if (frame_len == 0) {
                        /* Payload is empty, we're done. */
                        rktrans->rktrans_recv_buf = NULL;
                        *rkbufp                   = rkbuf;
                        return 1;
                }

                /* Allocate memory to hold entire frame payload in contigious
                 * memory. */
                rd_buf_write_ensure_contig(&rkbuf->rkbuf_buf, frame_len);

                /* Try reading directly, there is probably more data available*/
                return rd_kafka_transport_framed_recv(rktrans, rkbufp, errstr,
                                                      errstr_size);
        }

        if (rd_buf_write_pos(&rkbuf->rkbuf_buf) == rkbuf->rkbuf_totlen) {
                /* Payload is complete. */
                rktrans->rktrans_recv_buf = NULL;
                *rkbufp                   = rkbuf;
                return 1;
        }

        /* Wait for more data */
        return 0;

err_parse:
        rd_snprintf(errstr, errstr_size, "Frame header parsing failed: %s",
                    rd_kafka_err2str(rkbuf->rkbuf_err));
        return -1;
}


/**
 * @brief Final socket setup after a connection has been established
 */
void rd_kafka_transport_post_connect_setup(rd_kafka_transport_t *rktrans) {
        rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
        unsigned int slen;

        /* Set socket send & receive buffer sizes if configuerd */
        if (rkb->rkb_rk->rk_conf.socket_sndbuf_size != 0) {
                if (setsockopt(
                        rktrans->rktrans_s, SOL_SOCKET, SO_SNDBUF,
                        (void *)&rkb->rkb_rk->rk_conf.socket_sndbuf_size,
                        sizeof(rkb->rkb_rk->rk_conf.socket_sndbuf_size)) ==
                    RD_SOCKET_ERROR)
                        rd_rkb_log(rkb, LOG_WARNING, "SNDBUF",
                                   "Failed to set socket send "
                                   "buffer size to %i: %s",
                                   rkb->rkb_rk->rk_conf.socket_sndbuf_size,
                                   rd_socket_strerror(rd_socket_errno));
        }

        if (rkb->rkb_rk->rk_conf.socket_rcvbuf_size != 0) {
                if (setsockopt(
                        rktrans->rktrans_s, SOL_SOCKET, SO_RCVBUF,
                        (void *)&rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
                        sizeof(rkb->rkb_rk->rk_conf.socket_rcvbuf_size)) ==
                    RD_SOCKET_ERROR)
                        rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                                   "Failed to set socket receive "
                                   "buffer size to %i: %s",
                                   rkb->rkb_rk->rk_conf.socket_rcvbuf_size,
                                   rd_socket_strerror(rd_socket_errno));
        }

        /* Get send and receive buffer sizes to allow limiting
         * the total number of bytes passed with iovecs to sendmsg()
         * and recvmsg(). */
        slen = sizeof(rktrans->rktrans_rcvbuf_size);
        if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_RCVBUF,
                       (void *)&rktrans->rktrans_rcvbuf_size,
                       &slen) == RD_SOCKET_ERROR) {
                rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                           "Failed to get socket receive "
                           "buffer size: %s: assuming 1MB",
                           rd_socket_strerror(rd_socket_errno));
                rktrans->rktrans_rcvbuf_size = 1024 * 1024;
        } else if (rktrans->rktrans_rcvbuf_size < 1024 * 64)
                rktrans->rktrans_rcvbuf_size =
                    1024 * 64; /* Use at least 64KB */

        slen = sizeof(rktrans->rktrans_sndbuf_size);
        if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_SNDBUF,
                       (void *)&rktrans->rktrans_sndbuf_size,
                       &slen) == RD_SOCKET_ERROR) {
                rd_rkb_log(rkb, LOG_WARNING, "RCVBUF",
                           "Failed to get socket send "
                           "buffer size: %s: assuming 1MB",
                           rd_socket_strerror(rd_socket_errno));
                rktrans->rktrans_sndbuf_size = 1024 * 1024;
        } else if (rktrans->rktrans_sndbuf_size < 1024 * 64)
                rktrans->rktrans_sndbuf_size =
                    1024 * 64; /* Use at least 64KB */


#ifdef TCP_NODELAY
        if (rkb->rkb_rk->rk_conf.socket_nagle_disable) {
                int one = 1;
                if (setsockopt(rktrans->rktrans_s, IPPROTO_TCP, TCP_NODELAY,
                               (void *)&one, sizeof(one)) == RD_SOCKET_ERROR)
                        rd_rkb_log(rkb, LOG_WARNING, "NAGLE",
                                   "Failed to disable Nagle (TCP_NODELAY) "
                                   "on socket: %s",
                                   rd_socket_strerror(rd_socket_errno));
        }
#endif
}


/**
 * TCP connection established.
 * Set up socket options, SSL, etc.
 *
 * Locality: broker thread
 */
static void rd_kafka_transport_connected(rd_kafka_transport_t *rktrans) {
        rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;

        rd_rkb_dbg(
            rkb, BROKER, "CONNECT", "Connected to %s",
            rd_sockaddr2str(rkb->rkb_addr_last,
                            RD_SOCKADDR2STR_F_PORT | RD_SOCKADDR2STR_F_FAMILY));

        rd_kafka_transport_post_connect_setup(rktrans);

#if WITH_SSL
        if (rkb->rkb_proto == RD_KAFKA_PROTO_SSL ||
            rkb->rkb_proto == RD_KAFKA_PROTO_SASL_SSL) {
                char errstr[512];

                rd_kafka_broker_lock(rkb);
                rd_kafka_broker_set_state(rkb,
                                          RD_KAFKA_BROKER_STATE_SSL_HANDSHAKE);
                rd_kafka_broker_unlock(rkb);

                /* Set up SSL connection.
                 * This is also an asynchronous operation so dont
                 * propagate to broker_connect_done() just yet. */
                if (rd_kafka_transport_ssl_connect(rkb, rktrans, errstr,
                                                   sizeof(errstr)) == -1) {
                        rd_kafka_transport_connect_done(rktrans, errstr);
                        return;
                }
                return;
        }
#endif

        /* Propagate connect success */
        rd_kafka_transport_connect_done(rktrans, NULL);
}



/**
 * @brief the kernel SO_ERROR in \p errp for the given transport.
 * @returns 0 if getsockopt() was succesful (and \p and errp can be trusted),
 * else -1 in which case \p errp 's value is undefined.
 */
static int rd_kafka_transport_get_socket_error(rd_kafka_transport_t *rktrans,
                                               int *errp) {
        socklen_t intlen = sizeof(*errp);

        if (getsockopt(rktrans->rktrans_s, SOL_SOCKET, SO_ERROR, (void *)errp,
                       &intlen) == -1) {
                rd_rkb_dbg(rktrans->rktrans_rkb, BROKER, "SO_ERROR",
                           "Failed to get socket error: %s",
                           rd_socket_strerror(rd_socket_errno));
                return -1;
        }

        return 0;
}


/**
 * IO event handler.
 *
 * @param socket_errstr Is an optional (else NULL) error string from the
 *                      socket layer.
 *
 * Locality: broker thread
 */
static void rd_kafka_transport_io_event(rd_kafka_transport_t *rktrans,
                                        int events,
                                        const char *socket_errstr) {
        char errstr[512];
        int r;
        rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;

        switch (rkb->rkb_state) {
        case RD_KAFKA_BROKER_STATE_CONNECT:
                /* Asynchronous connect finished, read status. */
                if (!(events & (POLLOUT | POLLERR | POLLHUP)))
                        return;

                if (socket_errstr)
                        rd_kafka_broker_fail(
                            rkb, LOG_ERR, RD_KAFKA_RESP_ERR__TRANSPORT,
                            "Connect to %s failed: %s",
                            rd_sockaddr2str(rkb->rkb_addr_last,
                                            RD_SOCKADDR2STR_F_PORT |
                                                RD_SOCKADDR2STR_F_FAMILY),
                            socket_errstr);
                else if (rd_kafka_transport_get_socket_error(rktrans, &r) ==
                         -1) {
                        rd_kafka_broker_fail(
                            rkb, LOG_ERR, RD_KAFKA_RESP_ERR__TRANSPORT,
                            "Connect to %s failed: "
                            "unable to get status from "
                            "socket %d: %s",
                            rd_sockaddr2str(rkb->rkb_addr_last,
                                            RD_SOCKADDR2STR_F_PORT |
                                                RD_SOCKADDR2STR_F_FAMILY),
                            rktrans->rktrans_s, rd_strerror(rd_socket_errno));
                } else if (r != 0) {
                        /* Connect failed */
                        rd_snprintf(
                            errstr, sizeof(errstr), "Connect to %s failed: %s",
                            rd_sockaddr2str(rkb->rkb_addr_last,
                                            RD_SOCKADDR2STR_F_PORT |
                                                RD_SOCKADDR2STR_F_FAMILY),
                            rd_strerror(r));

                        rd_kafka_transport_connect_done(rktrans, errstr);
                } else {
                        /* Connect succeeded */
                        rd_kafka_transport_connected(rktrans);
                }
                break;

        case RD_KAFKA_BROKER_STATE_SSL_HANDSHAKE:
#if WITH_SSL
                rd_assert(rktrans->rktrans_ssl);

                /* Currently setting up SSL connection:
                 * perform handshake. */
                r = rd_kafka_transport_ssl_handshake(rktrans);

                if (r == 0 /* handshake still in progress */ &&
                    (events & POLLHUP)) {
                        rd_kafka_broker_conn_closed(
                            rkb, RD_KAFKA_RESP_ERR__TRANSPORT, "Disconnected");
                        return;
                }

#else
                RD_NOTREACHED();
#endif
                break;

        case RD_KAFKA_BROKER_STATE_AUTH_LEGACY:
                /* SASL authentication.
                 * Prior to broker version v1.0.0 this is performed
                 * directly on the socket without Kafka framing. */
                if (rd_kafka_sasl_io_event(rktrans, events, errstr,
                                           sizeof(errstr)) == -1) {
                        rd_kafka_broker_fail(
                            rkb, LOG_ERR, RD_KAFKA_RESP_ERR__AUTHENTICATION,
                            "SASL authentication failure: %s", errstr);
                        return;
                }

                if (events & POLLHUP) {
                        rd_kafka_broker_fail(rkb, LOG_ERR,
                                             RD_KAFKA_RESP_ERR__AUTHENTICATION,
                                             "Disconnected");

                        return;
                }

                break;

        case RD_KAFKA_BROKER_STATE_APIVERSION_QUERY:
        case RD_KAFKA_BROKER_STATE_AUTH_HANDSHAKE:
        case RD_KAFKA_BROKER_STATE_AUTH_REQ:
        case RD_KAFKA_BROKER_STATE_UP:
        case RD_KAFKA_BROKER_STATE_UPDATE:

                if (events & POLLIN) {
                        while (rkb->rkb_state >= RD_KAFKA_BROKER_STATE_UP &&
                               rd_kafka_recv(rkb) > 0)
                                ;

                        /* If connection went down: bail out early */
                        if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_DOWN)
                                return;
                }

                if (events & POLLHUP) {
                        rd_kafka_broker_conn_closed(
                            rkb, RD_KAFKA_RESP_ERR__TRANSPORT, "Disconnected");
                        return;
                }

                if (events & POLLOUT) {
                        while (rd_kafka_send(rkb) > 0)
                                ;
                }
                break;

        case RD_KAFKA_BROKER_STATE_INIT:
        case RD_KAFKA_BROKER_STATE_DOWN:
        case RD_KAFKA_BROKER_STATE_TRY_CONNECT:
                rd_kafka_assert(rkb->rkb_rk, !*"bad state");
        }
}



#ifdef _WIN32
/**
 * @brief Convert WSA FD_.. events to POLL.. events.
 */
static RD_INLINE int rd_kafka_transport_wsa2events(long wevents) {
        int events = 0;

        if (unlikely(wevents == 0))
                return 0;

        if (wevents & FD_READ)
                events |= POLLIN;
        if (wevents & (FD_WRITE | FD_CONNECT))
                events |= POLLOUT;
        if (wevents & FD_CLOSE)
                events |= POLLHUP;

        rd_dassert(events != 0);

        return events;
}

/**
 * @brief Convert POLL.. events to WSA FD_.. events.
 */
static RD_INLINE int rd_kafka_transport_events2wsa(int events,
                                                   rd_bool_t is_connecting) {
        long wevents = FD_CLOSE;

        if (unlikely(is_connecting))
                return wevents | FD_CONNECT;

        if (events & POLLIN)
                wevents |= FD_READ;
        if (events & POLLOUT)
                wevents |= FD_WRITE;

        return wevents;
}


/**
 * @returns the WinSocket events (as POLL.. events) for the broker socket.
 */
static int rd_kafka_transport_get_wsa_events(rd_kafka_transport_t *rktrans) {
        const int try_bits[4 * 2] = {FD_READ_BIT,  POLLIN,         FD_WRITE_BIT,
                                     POLLOUT,      FD_CONNECT_BIT, POLLOUT,
                                     FD_CLOSE_BIT, POLLHUP};
        int r, i;
        WSANETWORKEVENTS netevents;
        int events                = 0;
        const char *socket_errstr = NULL;
        rd_kafka_broker_t *rkb    = rktrans->rktrans_rkb;

        /* Get Socket event */
        r = WSAEnumNetworkEvents(rktrans->rktrans_s, rktrans->rktrans_wsaevent,
                                 &netevents);
        if (unlikely(r == SOCKET_ERROR)) {
                rd_rkb_log(rkb, LOG_ERR, "WSAWAIT",
                           "WSAEnumNetworkEvents() failed: %s",
                           rd_socket_strerror(rd_socket_errno));
                socket_errstr = rd_socket_strerror(rd_socket_errno);
                return POLLHUP | POLLERR;
        }

        /* Get fired events and errors for each event type */
        for (i = 0; i < RD_ARRAYSIZE(try_bits); i += 2) {
                const int bit   = try_bits[i];
                const int event = try_bits[i + 1];

                if (!(netevents.lNetworkEvents & (1 << bit)))
                        continue;

                if (unlikely(netevents.iErrorCode[bit])) {
                        socket_errstr =
                            rd_socket_strerror(netevents.iErrorCode[bit]);
                        events |= POLLHUP;
                } else {
                        events |= event;

                        if (bit == FD_WRITE_BIT) {
                                /* Writing no longer blocked */
                                rktrans->rktrans_blocked = rd_false;
                        }
                }
        }

        return events;
}


/**
 * @brief Win32: Poll transport and \p rkq cond events.
 *
 * @returns the transport socket POLL.. event bits.
 */
static int rd_kafka_transport_io_serve_win32(rd_kafka_transport_t *rktrans,
                                             rd_kafka_q_t *rkq,
                                             int timeout_ms) {
        const DWORD wsaevent_cnt = 3;
        WSAEVENT wsaevents[3]    = {
            rkq->rkq_cond.mEvents[0],  /* rkq: cnd_signal */
            rkq->rkq_cond.mEvents[1],  /* rkq: cnd_broadcast */
            rktrans->rktrans_wsaevent, /* socket */
        };
        DWORD r;
        int events               = 0;
        rd_kafka_broker_t *rkb   = rktrans->rktrans_rkb;
        rd_bool_t set_pollout    = rd_false;
        rd_bool_t cnd_is_waiting = rd_false;

        /* WSA only sets FD_WRITE (e.g., POLLOUT) when the socket was
         * previously blocked, unlike BSD sockets that set POLLOUT as long as
         * the socket isn't blocked. So we need to imitate the BSD behaviour
         * here and cut the timeout short if a write is wanted and the socket
         * is not currently blocked. */
        if (rktrans->rktrans_rkb->rkb_state != RD_KAFKA_BROKER_STATE_CONNECT &&
            !rktrans->rktrans_blocked &&
            (rktrans->rktrans_pfd[0].events & POLLOUT)) {
                timeout_ms  = 0;
                set_pollout = rd_true;
        } else {
                /* Check if the queue already has ops enqueued in which case we
                 * cut the timeout short. Else add this thread as waiting on the
                 * queue's condvar so that cnd_signal() (et.al.) will perform
                 * SetEvent() and thus wake up this thread in case a new op is
                 * added to the queue. */
                mtx_lock(&rkq->rkq_lock);
                if (rkq->rkq_qlen > 0) {
                        timeout_ms = 0;
                } else {
                        cnd_is_waiting = rd_true;
                        cnd_wait_enter(&rkq->rkq_cond);
                }
                mtx_unlock(&rkq->rkq_lock);
        }

        /* Wait for IO and queue events */
        r = WSAWaitForMultipleEvents(wsaevent_cnt, wsaevents, FALSE, timeout_ms,
                                     FALSE);

        if (cnd_is_waiting) {
                mtx_lock(&rkq->rkq_lock);
                cnd_wait_exit(&rkq->rkq_cond);
                mtx_unlock(&rkq->rkq_lock);
        }

        if (unlikely(r == WSA_WAIT_FAILED)) {
                rd_rkb_log(rkb, LOG_CRIT, "WSAWAIT",
                           "WSAWaitForMultipleEvents failed: %s",
                           rd_socket_strerror(rd_socket_errno));
                return POLLERR;
        } else if (r != WSA_WAIT_TIMEOUT) {
                r -= WSA_WAIT_EVENT_0;

                /* Reset the cond events if any of them were triggered */
                if (r < 2) {
                        ResetEvent(rkq->rkq_cond.mEvents[0]);
                        ResetEvent(rkq->rkq_cond.mEvents[1]);
                }

                /* Get the socket events. */
                events = rd_kafka_transport_get_wsa_events(rktrans);
        }

        /* As explained above we need to set the POLLOUT flag
         * in case it is wanted but not triggered by Winsocket so that
         * io_event() knows it can attempt to send more data. */
        if (likely(set_pollout && !(events & (POLLHUP | POLLERR | POLLOUT))))
                events |= POLLOUT;

        return events;
}
#endif


/**
 * @brief Poll and serve IOs
 *
 * @returns 0 if \p rkq may need additional blocking/timeout polling, else 1.
 *
 * @locality broker thread
 */
int rd_kafka_transport_io_serve(rd_kafka_transport_t *rktrans,
                                rd_kafka_q_t *rkq,
                                int timeout_ms) {
        rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
        int events;

        rd_kafka_curr_transport = rktrans;

        if (
#ifndef _WIN32
            /* BSD sockets use POLLOUT to indicate success to connect.
             * Windows has its own flag for this (FD_CONNECT). */
            rkb->rkb_state == RD_KAFKA_BROKER_STATE_CONNECT ||
#endif
            (rkb->rkb_state > RD_KAFKA_BROKER_STATE_SSL_HANDSHAKE &&
             rd_kafka_bufq_cnt(&rkb->rkb_waitresps) < rkb->rkb_max_inflight &&
             rd_kafka_bufq_cnt(&rkb->rkb_outbufs) > 0))
                rd_kafka_transport_poll_set(rkb->rkb_transport, POLLOUT);

#ifdef _WIN32
        /* BSD sockets use POLLIN and a following recv() returning 0 to
         * to indicate connection close.
         * Windows has its own flag for this (FD_CLOSE). */
        if (rd_kafka_bufq_cnt(&rkb->rkb_waitresps) > 0)
#endif
                rd_kafka_transport_poll_set(rkb->rkb_transport, POLLIN);

                /* On Windows we can wait for both IO and condvars (rkq)
                 * simultaneously.
                 *
                 * On *nix/BSD sockets we use a local pipe (pfd[1]) to wake
                 * up the rkq. */
#ifdef _WIN32
        events = rd_kafka_transport_io_serve_win32(rktrans, rkq, timeout_ms);

#else
        if (rd_kafka_transport_poll(rktrans, timeout_ms) < 1)
                return 0; /* No events, caller can block on \p rkq poll */

        /* Broker socket events */
        events = rktrans->rktrans_pfd[0].revents;
#endif

        if (events) {
                rd_kafka_transport_poll_clear(rktrans, POLLOUT | POLLIN);

                rd_kafka_transport_io_event(rktrans, events, NULL);
        }

        return 1;
}


/**
 * @brief Create a new transport object using existing socket \p s.
 */
rd_kafka_transport_t *rd_kafka_transport_new(rd_kafka_broker_t *rkb,
                                             rd_socket_t s,
                                             char *errstr,
                                             size_t errstr_size) {
        rd_kafka_transport_t *rktrans;
        int on = 1;
        int r;

#ifdef SO_NOSIGPIPE
        /* Disable SIGPIPE signalling for this socket on OSX */
        if (setsockopt(s, SOL_SOCKET, SO_NOSIGPIPE, &on, sizeof(on)) == -1)
                rd_rkb_dbg(rkb, BROKER, "SOCKET",
                           "Failed to set SO_NOSIGPIPE: %s",
                           rd_socket_strerror(rd_socket_errno));
#endif

#ifdef SO_KEEPALIVE
        /* Enable TCP keep-alives, if configured. */
        if (rkb->rkb_rk->rk_conf.socket_keepalive) {
                if (setsockopt(s, SOL_SOCKET, SO_KEEPALIVE, (void *)&on,
                               sizeof(on)) == RD_SOCKET_ERROR)
                        rd_rkb_dbg(rkb, BROKER, "SOCKET",
                                   "Failed to set SO_KEEPALIVE: %s",
                                   rd_socket_strerror(rd_socket_errno));
        }
#endif

        /* Set the socket to non-blocking */
        if ((r = rd_fd_set_nonblocking(s))) {
                rd_snprintf(errstr, errstr_size,
                            "Failed to set socket non-blocking: %s",
                            rd_socket_strerror(r));
                return NULL;
        }


        rktrans              = rd_calloc(1, sizeof(*rktrans));
        rktrans->rktrans_rkb = rkb;
        rktrans->rktrans_s   = s;

#ifdef _WIN32
        rktrans->rktrans_wsaevent = WSACreateEvent();
        rd_assert(rktrans->rktrans_wsaevent != NULL);
#endif

        return rktrans;
}


/**
 * Initiate asynchronous connection attempt.
 *
 * Locality: broker thread
 */
rd_kafka_transport_t *rd_kafka_transport_connect(rd_kafka_broker_t *rkb,
                                                 const rd_sockaddr_inx_t *sinx,
                                                 char *errstr,
                                                 size_t errstr_size) {
        rd_kafka_transport_t *rktrans;
        int s = -1;
        int r;

        rkb->rkb_addr_last = sinx;

        s = rkb->rkb_rk->rk_conf.socket_cb(sinx->in.sin_family, SOCK_STREAM,
                                           IPPROTO_TCP,
                                           rkb->rkb_rk->rk_conf.opaque);
        if (s == -1) {
                rd_snprintf(errstr, errstr_size, "Failed to create socket: %s",
                            rd_socket_strerror(rd_socket_errno));
                return NULL;
        }

        rktrans = rd_kafka_transport_new(rkb, s, errstr, errstr_size);
        if (!rktrans) {
                rd_kafka_transport_close0(rkb->rkb_rk, s);
                return NULL;
        }

        rd_rkb_dbg(rkb, BROKER, "CONNECT",
                   "Connecting to %s (%s) "
                   "with socket %i",
                   rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_FAMILY |
                                             RD_SOCKADDR2STR_F_PORT),
                   rd_kafka_secproto_names[rkb->rkb_proto], s);

        /* Connect to broker */
        if (rkb->rkb_rk->rk_conf.connect_cb) {
                rd_kafka_broker_lock(rkb); /* for rkb_nodename */
                r = rkb->rkb_rk->rk_conf.connect_cb(
                    s, (struct sockaddr *)sinx, RD_SOCKADDR_INX_LEN(sinx),
                    rkb->rkb_nodename, rkb->rkb_rk->rk_conf.opaque);
                rd_kafka_broker_unlock(rkb);
        } else {
                if (connect(s, (struct sockaddr *)sinx,
                            RD_SOCKADDR_INX_LEN(sinx)) == RD_SOCKET_ERROR &&
                    (rd_socket_errno != EINPROGRESS
#ifdef _WIN32
                     && rd_socket_errno != WSAEWOULDBLOCK
#endif
                     ))
                        r = rd_socket_errno;
                else
                        r = 0;
        }

        if (r != 0) {
                rd_rkb_dbg(rkb, BROKER, "CONNECT",
                           "Couldn't connect to %s: %s (%i)",
                           rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_PORT |
                                                     RD_SOCKADDR2STR_F_FAMILY),
                           rd_socket_strerror(r), r);
                rd_snprintf(errstr, errstr_size,
                            "Failed to connect to broker at %s: %s",
                            rd_sockaddr2str(sinx, RD_SOCKADDR2STR_F_NICE),
                            rd_socket_strerror(r));

                rd_kafka_transport_close(rktrans);
                return NULL;
        }

        /* Set up transport handle */
        rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt++].fd = s;
        if (rkb->rkb_wakeup_fd[0] != -1) {
                rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt].events = POLLIN;
                rktrans->rktrans_pfd[rktrans->rktrans_pfd_cnt++].fd =
                    rkb->rkb_wakeup_fd[0];
        }


        /* Poll writability to trigger on connection success/failure. */
        rd_kafka_transport_poll_set(rktrans, POLLOUT);

        return rktrans;
}


#ifdef _WIN32
/**
 * @brief Set the WinSocket event poll bit to \p events.
 */
static void rd_kafka_transport_poll_set_wsa(rd_kafka_transport_t *rktrans,
                                            int events) {
        int r;
        r = WSAEventSelect(
            rktrans->rktrans_s, rktrans->rktrans_wsaevent,
            rd_kafka_transport_events2wsa(rktrans->rktrans_pfd[0].events,
                                          rktrans->rktrans_rkb->rkb_state ==
                                              RD_KAFKA_BROKER_STATE_CONNECT));
        if (unlikely(r != 0)) {
                rd_rkb_log(rktrans->rktrans_rkb, LOG_CRIT, "WSAEVENT",
                           "WSAEventSelect() failed: %s",
                           rd_socket_strerror(rd_socket_errno));
        }
}
#endif

void rd_kafka_transport_poll_set(rd_kafka_transport_t *rktrans, int event) {
        if ((rktrans->rktrans_pfd[0].events & event) == event)
                return;

        rktrans->rktrans_pfd[0].events |= event;

#ifdef _WIN32
        rd_kafka_transport_poll_set_wsa(rktrans,
                                        rktrans->rktrans_pfd[0].events);
#endif
}

void rd_kafka_transport_poll_clear(rd_kafka_transport_t *rktrans, int event) {
        if (!(rktrans->rktrans_pfd[0].events & event))
                return;

        rktrans->rktrans_pfd[0].events &= ~event;

#ifdef _WIN32
        rd_kafka_transport_poll_set_wsa(rktrans,
                                        rktrans->rktrans_pfd[0].events);
#endif
}

#ifndef _WIN32
/**
 * @brief Poll transport fds.
 *
 * @returns 1 if an event was raised, else 0, or -1 on error.
 */
static int rd_kafka_transport_poll(rd_kafka_transport_t *rktrans, int tmout) {
        int r;

        r = poll(rktrans->rktrans_pfd, rktrans->rktrans_pfd_cnt, tmout);
        if (r <= 0)
                return r;

        if (rktrans->rktrans_pfd[1].revents & POLLIN) {
                /* Read wake-up fd data and throw away, just used for wake-ups*/
                char buf[1024];
                while (rd_socket_read((int)rktrans->rktrans_pfd[1].fd, buf,
                                      sizeof(buf)) > 0)
                        ; /* Read all buffered signalling bytes */
        }

        return 1;
}
#endif

#ifdef _WIN32
/**
 * @brief A socket write operation would block, flag the socket
 *        as blocked so that POLLOUT events are handled correctly.
 *
 * This is really only used on Windows where POLLOUT (FD_WRITE) is
 * edge-triggered rather than level-triggered.
 */
void rd_kafka_transport_set_blocked(rd_kafka_transport_t *rktrans,
                                    rd_bool_t blocked) {
        rktrans->rktrans_blocked = blocked;
}
#endif


#if 0
/**
 * Global cleanup.
 * This is dangerous and SHOULD NOT be called since it will rip
 * the rug from under the application if it uses any of this functionality
 * in its own code. This means we might leak some memory on exit.
 */
void rd_kafka_transport_term (void) {
#ifdef _WIN32
        (void)WSACleanup(); /* FIXME: dangerous */
#endif
}
#endif

void rd_kafka_transport_init(void) {
#ifdef _WIN32
        WSADATA d;
        (void)WSAStartup(MAKEWORD(2, 2), &d);
#endif
}