libecp/common/ecp_impl.c

/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.
 */

#include "ecp_impl.h"

static __thread ecp_instance_t *my_ecp_instp;
static int  __ecp_logging = 0;

static int ecp_send_impl(void *, char *, int, boolean_t);

void
ecp_logmsg(int pri, const char *fmt, ...)
{
    va_list ap;

    if (__ecp_logging == 0) {
        return;
    }
    va_start(ap, fmt);

    if (__ecp_logging == ECP_SYS_LOG)
        vsyslog(pri, fmt, ap);
    else
        (void) vfprintf(stderr, fmt, ap);
    va_end(ap);
}

static void
ecp_print_hdr(char *buf, int len)
{
    ecp_hdr_t *ecp_hdrp;
    uint16_t ecp_ver_op_stype;

    if (len < sizeof (ecp_hdr_t)) {
        return;
    }

    ecp_hdrp = (ecp_hdr_t *)buf;

    logerr("SRC:        %s\n",
        ether_ntoa(&ecp_hdrp->ecp_ether_header.ether_shost));

    logerr("DST:        %s\n",
        ether_ntoa(&ecp_hdrp->ecp_ether_header.ether_dhost));

    ecp_ver_op_stype = ntohs(ecp_hdrp->ecp_ver_op_stype);

    logerr("Version %d\n",
        ((ecp_ver_op_stype >> ECP_VER_SHIFT) & ECP_VER_MASK));
    logerr("OP      %d\n",
        ((ecp_ver_op_stype >> ECP_OP_SHIFT) & ECP_OP_MASK));
    logerr("Type:       %d\n",
        ((ecp_ver_op_stype >> ECP_SUBTYPE_SHIFT) & ECP_SUBTYPE_MASK));
    logerr("Seq:        %d\n",
        ntohs(ecp_hdrp->ecp_seq));
}

/* Only VDP is defined as a consumer of ECP */
static boolean_t
ecp_validate_subtype(uint16_t subtype)
{
    switch (subtype) {
    case ECP_VDP:
        return (B_TRUE);
    default:
        return (B_FALSE);
    }
}

static void
ecp_sighandler()
{
    logtrace("ecp_sighandler\n");
    pthread_exit(0);
}

static void
ecp_setup_sig_handler()
{
    struct sigaction actions;

    (void) memset(&actions, 0, sizeof (actions));
    (void) sigemptyset(&actions.sa_mask);
    actions.sa_flags = SA_NODEFER;
    actions.sa_handler = ecp_sighandler;

    sigaction(SIGQUIT, &actions, NULL);
}

/*
 * Parse an incoming ECP packet. If this is an ACK for the last sequence
 * we sent, we are done. If this is a request, we ACK it. Additionally, if
 * we are seeing a request from the peer for the first time, we initialize the
 * peer's sequence number.
 */
static void
ecp_parse_rx_packet(char *buf, int len, ecp_instance_t *ecp_instp)
{
    ecp_hdr_t   *ecpp;
    uint16_t    ecp_ver_op_stype;
    uint8_t     ecp_ver;
    uint8_t     ecp_op;
    uint16_t    ecp_type;

    if (len < sizeof (ecp_hdr_t)) {
        ECP_STAT_UPDATE(ecp_instp, ierrors, 1);
        return;
    }

    ecpp = (ecp_hdr_t *)buf;
    assert(ntohs(ecpp->ecp_ether_header.ether_type) == ETHERTYPE_ECP);
    /* Ignore packets sent by us */
    if (bcmp(ecp_instp->ecp_src_addr.ether_addr_octet,
        ecpp->ecp_ether_shost, ETHERADDRL) == 0) {
        return;
    }
    ecp_ver_op_stype = ntohs(ecpp->ecp_ver_op_stype);
    ecp_ver = ((ecp_ver_op_stype >> ECP_VER_SHIFT) & ECP_VER_MASK);
    ecp_op = ((ecp_ver_op_stype >> ECP_OP_SHIFT) & ECP_OP_MASK);
    ecp_type = ((ecp_ver_op_stype >> ECP_SUBTYPE_SHIFT) & ECP_SUBTYPE_MASK);
    if (ecp_ver != ecp_instp->ecp_version) {
        logerr("ecp_parse_rx_packet: unsupported version %d\n",
            ecp_ver);
        return;
    }
    assert((ecp_op == ECP_REQ) || (ecp_op == ECP_ACK));
    if (ecp_op == ECP_ACK) {
        (void) pthread_mutex_lock(&ecp_instp->ecp_tx_lock);
        if (ntohs(ecpp->ecp_seq) == ecp_instp->ecp_ack_seq) {
            ecp_instp->ecp_last_ack_seq = ntohs(ecpp->ecp_seq);
            ecp_instp->ecp_tx_acked = B_TRUE;
            (void) pthread_cond_signal(&ecp_instp->ecp_tx_cv);
        } else {
            logerr("ecp rcvd ack %d expected %d \n",
                ntohs(ecpp->ecp_seq), ecp_instp->ecp_ack_seq);
        }
        (void) pthread_mutex_unlock(&ecp_instp->ecp_tx_lock);
    } else if (ecp_instp->ecp_subtype == ecp_type) {
        if (ecp_instp->ecp_init_req_seq) {
            ecp_instp->ecp_init_req_seq = B_FALSE;
            ecp_instp->ecp_req_seq = ntohs(ecpp->ecp_seq);
        } else if (ntohs(ecpp->ecp_seq) == ecp_instp->ecp_req_seq) {
            /*
             * Resend REQ ack
             */
            (void) ecp_send_impl(ecp_instp, NULL, 0, B_FALSE);
            return;
        }
        ecp_instp->ecp_req_seq = ntohs(ecpp->ecp_seq);
        /*
         * ACK the request
         */
        (void) ecp_send_impl(ecp_instp, NULL, 0, B_FALSE);
        /*
         * Pass it on to the ULP
         */
        len -= sizeof (ecp_hdr_t);
        ecp_instp->ecp_ulp_cb(ecp_instp->ecp_ulp_cb_arg,
            ecp_instp->ecp_ack_seq, ECP_REQ, (char *)++ecpp, len);
    } else {
        logerr("ecp_parse_rx_packet not for us - subtype %d\n",
            ecp_type);
        ECP_STAT_UPDATE(ecp_instp, ierrors, 1);
        ecp_print_hdr(buf, len);
    }
}

/* Thread for receiving ECP packets */
static void *
ecp_inst_thr_process(void *arg)
{
    char buf[ETHERMTU];
    int ret;

    ecp_setup_sig_handler();
    my_ecp_instp = arg;

    /* LINTED */
    while (B_TRUE) {
        errno = 0;
        ret = recv(my_ecp_instp->ecp_sockfd, buf, sizeof (buf), 0);
        if (ret <= 0) {
            logerr("ecp thread exiting, recv errno %d\n", errno);
            break;
        } else {
            /* Let's look at the packet */
            ECP_STAT_UPDATE(my_ecp_instp, ipkts, 1);
            ecp_parse_rx_packet(buf, ret, my_ecp_instp);
        }
    }
    return (NULL);
}

/*
 * Initialize an ECP instance including the callback function to be invoked.
 */
void *
ecp_init(struct ether_addr *dst_addr,
    struct ether_addr *src_addr, uint16_t subtype, int sockfd,
    int num_of_tries, ecp_ulp_cb_t ecp_ulp_cb, void *arg, int log,
    int *errorp)
{
    ecp_instance_t *ecp_instp;

    if (!ecp_validate_subtype(subtype)) {
        *errorp = EINVAL;
        return (NULL);
    }

    ecp_instp = calloc(sizeof (ecp_instance_t), 1);
    if (ecp_instp == NULL) {
        *errorp = ENOMEM;
        return (NULL);
    }

    ecp_instp->ecp_subtype = subtype;
    ecp_instp->ecp_version = ECP_VERSION;
    /* LINTED */
    ecp_instp->ecp_req_seq = -1;
    ecp_instp->ecp_ack_seq = rand() % 0xffff;
    ecp_instp->ecp_ether_type = ETHERTYPE_ECP;
    ecp_instp->ecp_sockfd = sockfd;
    ecp_instp->ecp_max_tx_attempts = num_of_tries;
    ecp_instp->ecp_ulp_cb = ecp_ulp_cb;
    ecp_instp->ecp_ulp_cb_arg = arg;
    ecp_instp->ecp_init_req_seq = B_TRUE;
    (void) pthread_mutex_init(&ecp_instp->ecp_tx_lock, NULL);
    (void) pthread_cond_init(&ecp_instp->ecp_tx_cv, NULL);
    ecp_instp->ecp_tx_acked = B_FALSE;
    /*
     * Need a progrmatic way to find this.
     */
    bcopy(dst_addr, ecp_instp->ecp_dst_addr.ether_addr_octet, ETHERADDRL);
    bcopy(src_addr, ecp_instp->ecp_src_addr.ether_addr_octet, ETHERADDRL);
    (void) pthread_create(&ecp_instp->ecp_process_thread, NULL,
        ecp_inst_thr_process, (void *)ecp_instp);

    __ecp_logging = log;
    return (ecp_instp);
}

void
ecp_fini(void *instp)
{
    ecp_instance_t *ecp_instp = instp;

    logtrace("ecp_fini: calling pthread_kill \n");
    (void) pthread_kill(ecp_instp->ecp_process_thread, SIGQUIT);
    (void) pthread_join(ecp_instp->ecp_process_thread, NULL);
    free(instp);
}

int
ecp_offset()
{
    return (sizeof (ecp_hdr_t));
}

/*
 * Prepare and transmit an ECP packet after prepending the ECP header, if not
 * a retransmit.
 */
static int
ecp_send_impl(void *instp, char *buf, int len, boolean_t resend)
{
    ecp_hdr_t ecp_hdr;
    ecp_hdr_t *ecp_hdrp;
    ecp_instance_t *ecp_instp = instp;
    int ret;

    if (buf == NULL) {
        ecp_hdrp = &ecp_hdr;
        len = sizeof (ecp_hdr_t);
    } else {
        assert(len >= sizeof (ecp_hdr_t));
        ecp_hdrp = (ecp_hdr_t *)buf;
    }

    if (!resend) {
        bcopy(&ecp_instp->ecp_dst_addr,
            &ecp_hdrp->ecp_ether_header.ether_dhost, ETHERADDRL);
        bcopy(&ecp_instp->ecp_src_addr,
            &ecp_hdrp->ecp_ether_header.ether_shost, ETHERADDRL);
        ecp_hdrp->ecp_ether_header.ether_type =
            htons(ecp_instp->ecp_ether_type);
        ecp_hdrp->ecp_ver_op_stype =
            (ecp_instp->ecp_version & ECP_VER_MASK) << ECP_VER_SHIFT;
        ecp_hdrp->ecp_ver_op_stype |=
            (ecp_instp->ecp_subtype & ECP_SUBTYPE_MASK)
            << ECP_SUBTYPE_SHIFT;
    }
    if (buf == NULL) {
        /*
         * ACK request received
         * OR resend the last req ACK
         */
        ecp_hdrp->ecp_ver_op_stype |=
            (ECP_ACK & ECP_OP_MASK) << ECP_OP_SHIFT;
        ecp_hdrp->ecp_seq = htons(ecp_instp->ecp_req_seq);
        ecp_hdrp->ecp_ver_op_stype = htons(ecp_hdrp->ecp_ver_op_stype);
    } else if (!resend) {
        ecp_hdrp->ecp_seq = htons(++ecp_instp->ecp_ack_seq);
        /* LINTED */
        ecp_hdrp->ecp_ver_op_stype |=
            (ECP_REQ & ECP_OP_MASK) << ECP_OP_SHIFT;
        ecp_hdrp->ecp_ver_op_stype = htons(ecp_hdrp->ecp_ver_op_stype);
    }
    errno = 0;
    ret = send(ecp_instp->ecp_sockfd, (void *)ecp_hdrp, len, 0);
    if (ret != len) {
        ret = errno;
        ECP_STAT_UPDATE(ecp_instp, oerrors, 1);
        logerr("ecp send error %s \n", strerror(errno));
        return (ret);
    }
    if (resend)
        ECP_STAT_UPDATE(ecp_instp, rexmits, 1);

    ECP_STAT_UPDATE(ecp_instp, opkts, 1);
    return (0);
}

/*
 * Send an ECP request and wait for the ACK. Retransmit if not ack'd
 * upto the max retries time.
 */
boolean_t
ecp_send(void *instp, char *buf, int len)
{
    ecp_instance_t *ecp_instp = instp;
    boolean_t ret = B_FALSE;
    int retry_count = 0;
    struct timespec next_time;

    (void) pthread_mutex_lock(&ecp_instp->ecp_tx_lock);
    ecp_instp->ecp_tx_acked = B_FALSE;
    while (retry_count < ecp_instp->ecp_max_tx_attempts) {
        (void) ecp_send_impl(instp, buf, len, retry_count > 0);
        (void) clock_gettime(CLOCK_REALTIME, &next_time);
        next_time.tv_sec += (next_time.tv_nsec +
            (ECP_ACK_TIME_OUT * 1000000)) / NANOSEC;
        next_time.tv_nsec = (next_time.tv_nsec +
            (ECP_ACK_TIME_OUT * 1000000)) % NANOSEC;
        (void) pthread_cond_timedwait(&ecp_instp->ecp_tx_cv,
            &ecp_instp->ecp_tx_lock, &next_time);
        /*
         * Check if ECP was acked
         */
        if (ecp_instp->ecp_tx_acked)
            break;
        retry_count++;
    }
    if (!ecp_instp->ecp_tx_acked)
        ECP_STAT_UPDATE(ecp_instp, timeouts, 1);
    ret = ecp_instp->ecp_tx_acked;
    (void) pthread_mutex_unlock(&ecp_instp->ecp_tx_lock);
    return (ret);
}

void
ecp_stats(void *instp, ecp_stat_t *es)
{
    ecp_instance_t *ecp_instp = instp;

    *es = ecp_instp->ecp_stat;
}

void
ecp_timers(ecp_timers_t *et)
{
    et->et_r = ECP_MAX_TRIES;
    et->et_rte = ECP_RTE_EXPONENT;
}

void
ecp_get_config(void *instp, ecp_config_t *ecpc)
{
    ecp_instance_t *ecp_instp = instp;

    ecpc->ecpc_req_seq = ecp_instp->ecp_req_seq;
    ecpc->ecpc_ack_seq = ecp_instp->ecp_ack_seq;
    ecpc->ecpc_last_ack_seq = ecp_instp->ecp_last_ack_seq;
    ecpc->ecpc_max_tx_attempts = ecp_instp->ecp_max_tx_attempts;
}