/*      $NetBSD: isns_thread.c,v 1.2 2019/07/03 18:40:33 dholland Exp $ */

/*-
 * Copyright (c) 2004,2009 The NetBSD Foundation, Inc.
 * All rights reserved.
 *
 * This code is derived from software contributed to The NetBSD Foundation
 * by Wasabi Systems, Inc.
 *
 * 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 NETBSD FOUNDATION, INC. 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 FOUNDATION 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.
 */

#include <sys/cdefs.h>
__RCSID("$NetBSD: isns_thread.c,v 1.2 2019/07/03 18:40:33 dholland Exp $");


/*
 * isns_thread.c
 */

#include <sys/types.h>

#include <unistd.h>

#include "isns.h"
#include "isns_config.h"
#include "isns_defs.h"

static struct iovec read_buf[2 + (ISNS_MAX_PDU_PAYLOAD / ISNS_BUF_SIZE) +
    ((ISNS_MAX_PDU_PAYLOAD % ISNS_BUF_SIZE) != 0)];

static struct isns_task_s *isns_get_next_task(struct isns_config_s *); 

/*
 * isns_control_thread()
 */ 
void *
isns_control_thread(void *arg)
{
        struct isns_config_s *cfg_p = (struct isns_config_s *)arg;
        struct kevent evt_chgs[5], *evt_p;

        int n, nevents;
        isns_kevent_handler *evt_handler_p;
        int run_thread;

        run_thread = 1;

        while (run_thread) {
                /* if no task outstanding, check queue here and send PDU */
                while ((cfg_p->curtask_p == NULL)
                    && ((cfg_p->curtask_p = isns_get_next_task(cfg_p)) != NULL))
                        isns_run_task(cfg_p->curtask_p);

                nevents = kevent(cfg_p->kq, NULL, 0, evt_chgs,
                    ARRAY_ELEMS(evt_chgs), NULL);

                DBG("isns_control_thread: kevent() nevents=%d\n", nevents);

                for (n = 0, evt_p = evt_chgs; n < nevents; n++, evt_p++) {
                        DBG("event[%d] - data=%d\n", n, (int)evt_p->data);
                        evt_handler_p = (void *)evt_p->udata;
                        run_thread = (evt_handler_p(evt_p, cfg_p) == 0);
                }
        }

        return 0;
}

/*
 * isns_get_next_task()
 */
static struct isns_task_s *
isns_get_next_task(struct isns_config_s *cfg_p)
{
        struct isns_task_s *task_p = NULL;


        DBG("isns_get_next_task: entered\n");

        task_p = isns_taskq_remove(cfg_p);

        if (cfg_p->sd_connected)
                return task_p;
        else {
                if (task_p == NULL)
                        return NULL;
                else {
                        if (task_p->task_type != ISNS_TASK_INIT_SOCKET_IO) {
                                isns_taskq_insert_head(cfg_p, task_p);

                                task_p = isns_new_task(cfg_p,
                                    ISNS_TASK_RECONNECT_SERVER, 0);
                                task_p->var.reconnect_server.ai_p = cfg_p->ai_p;
                        }

                        return task_p;
                }
        }
}

/*
 * isns_kevent_pipe()
 */ 
int
isns_kevent_pipe(struct kevent* evt_p, struct isns_config_s *cfg_p)
{
        uint8_t cmd_type;
        int force_isns_stop;
        uint16_t trans_id;
        ssize_t rbytes;
        int pipe_nbytes;

        force_isns_stop = 0;
        pipe_nbytes = (int)evt_p->data;

        while (pipe_nbytes > 0) {
                rbytes = read(cfg_p->pipe_fds[0], &cmd_type,
                    sizeof(cmd_type));
                if (rbytes < 0) {
                        DBG("isns_kevent_pipe: error on wepe_sys_read\n");
                        /*?? should we break here? */
                        continue;
                }

                pipe_nbytes -= (int)rbytes;
                switch (cmd_type) {
                case ISNS_CMD_PROCESS_TASKQ:
                        DBG("isns_kevent_pipe: ISNS_CMD_PROCESS_TASKQ\n");
                        break;

                case ISNS_CMD_ABORT_TRANS:
                        DBG("isns_kevent_pipe: ISNS_CMD_ABORT_TRANS\n");
                        rbytes = read(cfg_p->pipe_fds[0], &trans_id,
                            sizeof(trans_id));
                        if (rbytes < 0)
                                DBG("isns_kevent_pipe: "
                                    "error reading trans id\n");
                        else if (rbytes != sizeof(trans_id))
                                DBG("isns_kevent_pipe: "
                                    "short read reading trans id\n");
                        else {
                                isns_abort_trans(cfg_p, trans_id);
                                pipe_nbytes -= (int)rbytes;
                        }
                        break;

                case ISNS_CMD_STOP:
                        DBG("isns_kevent_pipe: ISNS_CMD_STOP\n");
                        force_isns_stop = 1;
                        pipe_nbytes = 0;
                        break;

                default:
                        DBG("isns_kevent_pipe: unknown command (cmd=%d)\n",
                            cmd_type);
                        break;
                }
        }

        return (force_isns_stop ? 1 : 0);
}

/*
 * isns_is_trans_complete()
 */
static int
isns_is_trans_complete(struct isns_trans_s *trans_p)
{
        struct isns_pdu_s *pdu_p;
        uint16_t count;

        pdu_p = trans_p->pdu_rsp_list;
        count = 0;
        while (pdu_p->next != NULL) {
                if (pdu_p->hdr.seq_id != count++) return 0;
                pdu_p = pdu_p->next;
        }
        if ((pdu_p->hdr.seq_id != count) ||
            !(pdu_p->hdr.flags & ISNS_FLAG_LAST_PDU))
                return 0;

        return 1;
}

/*
 * isns_is_valid_resp()
 */
static int
isns_is_valid_resp(struct isns_trans_s *trans_p, struct isns_pdu_s *pdu_p)
{
        struct isns_pdu_s *curpdu_p;

        if (pdu_p->hdr.trans_id != trans_p->id)
                return 0;
        if (pdu_p->hdr.func_id != (trans_p->func_id | 0x8000))
                return 0;
        curpdu_p = trans_p->pdu_rsp_list;
        while (curpdu_p != NULL) {
                if (curpdu_p->hdr.seq_id == pdu_p->hdr.seq_id) return 0;
                curpdu_p = curpdu_p->next;
        }

        return 1;
}

/*
 * isns_process_in_pdu()
 */
static void
isns_process_in_pdu(struct isns_config_s *cfg_p)
{
        struct isns_task_s *curtask_p;
        struct isns_trans_s *trans_p;
        
        DBG("isns_process_in_pdu: entered\n");

        if ((curtask_p = cfg_p->curtask_p) == NULL)
                isns_free_pdu(cfg_p->pdu_in_p);
        else if ((trans_p = curtask_p->var.send_pdu.trans_p) == NULL)
                isns_free_pdu(cfg_p->pdu_in_p);
        else if (!isns_is_valid_resp(trans_p, cfg_p->pdu_in_p))
                isns_free_pdu(cfg_p->pdu_in_p);
        else {
                isns_add_pdu_response(trans_p, cfg_p->pdu_in_p);

                if (isns_is_trans_complete(trans_p)) {
                        isns_complete_trans(trans_p);
                        isns_end_task(curtask_p);
                }
        }

        cfg_p->pdu_in_p = NULL;
}

/*
 * isns_kevent_socket()
 */
int
isns_kevent_socket(struct kevent *evt_p, struct isns_config_s *cfg_p)
{
        struct iovec *iovp;
        struct isns_buffer_s *curbuf_p, *newbuf_p;
        struct isns_pdu_s *pdu_p;
        int64_t bavail; /* bytes available in socket buffer */
        uint32_t cur_len, buf_len, unread_len, rd_len, b_len;
        ssize_t rv;
        uint16_t payload_len;
        int iovcnt, more, transport_evt;


        DBG("isns_kevent_socket: entered\n");

        transport_evt = 0;
        bavail = evt_p->data;
        iovp = read_buf;

        more = (bavail > 0);
        while (more) {
                if (cfg_p->pdu_in_p == NULL) {
                        /*
                         * Try to form a valid pdu by starting with the hdr.
                         * If there isn't enough data in the socket buffer
                         * to form a full hdr, just return.
                         *
                         * Once we have read in our hdr, allocate all buffers
                         * needed.
                         */

                        if (bavail < (int64_t)sizeof(struct isns_pdu_hdr_s))
                                return 0;

                        /* Form a placeholder pdu */
                        pdu_p = isns_new_pdu(cfg_p, 0, 0, 0);

                        /* Read the header into our placeholder pdu */
                        read_buf[0].iov_base = &(pdu_p->hdr);
                        read_buf[0].iov_len = sizeof(struct isns_pdu_hdr_s);
                        iovcnt = 1;

                        iovp = read_buf;
                        rv = isns_socket_readv(cfg_p->sd, iovp, iovcnt);
                        if ((rv == 0) || (rv == -1)) {
                                DBG("isns_kevent_socket: isns_socket_readv(1) "
                                    "returned %d\n", rv);
                                transport_evt = 1;
                                break;
                        }

                        bavail -= sizeof(struct isns_pdu_hdr_s);
                        /*
                         * ToDo: read until sizeof(struct isns_pdu_hdr_s) has
                         *       been read in. This statement should be
                         *
                         *       bavail -= rv;
                         */

                        /* adjust byte order */
                        pdu_p->hdr.isnsp_version = isns_ntohs(pdu_p->hdr.
                            isnsp_version);
                        pdu_p->hdr.func_id = isns_ntohs(pdu_p->hdr.func_id);
                        pdu_p->hdr.payload_len = isns_ntohs(pdu_p->hdr.
                            payload_len);
                        pdu_p->hdr.flags = isns_ntohs(pdu_p->hdr.flags);
                        pdu_p->hdr.trans_id = isns_ntohs(pdu_p->hdr.trans_id);
                        pdu_p->hdr.seq_id = isns_ntohs(pdu_p->hdr.seq_id);
                        pdu_p->byteorder_host = 1;

                        /* Try to sense early whether we might have garbage */
                        if (pdu_p->hdr.isnsp_version != ISNSP_VERSION) {
                                DBG("isns_kevent_socket: pdu_p->hdr."
                                    "isnsp_version != ISNSP_VERSION\n");
                                isns_free_pdu(pdu_p);

                                transport_evt = 1;
                                break;
                        }

                        /* Allocate all the necessary payload buffers */ 
                        payload_len = pdu_p->hdr.payload_len;
                        curbuf_p = pdu_p->payload_p;
                        buf_len = 0;
                        while (buf_len + curbuf_p->alloc_len < payload_len) {
                                buf_len += curbuf_p->alloc_len;
                                newbuf_p = isns_new_buffer(0);
                                curbuf_p->next = newbuf_p;
                                curbuf_p = newbuf_p;
                        }
                        curbuf_p->next = NULL;

                        /* Hold on to our placeholder pdu */
                        cfg_p->pdu_in_p = pdu_p;
                        more = (bavail > 0) ? 1 : 0;
                } else if (bavail > 0) {
                        /*
                         * Fill in the pdu payload data.
                         *
                         * If we can fill it all in now
                         *     -AND- it corresponds to the active transaction
                         *           then add the pdu to the transaction's
                         *           pdu_rsp_list
                         *     -AND- it does not correspond to the active
                         *           transaction (or there is no active
                         *           transaction) then drop it on the floor.
                         * We may not be able to fill it all in now.
                         *     -EITHER WAY- fill in as much payload data now
                         *                  as we can.
                         */

                        /* Refer to our placeholder pdu */
                        pdu_p = cfg_p->pdu_in_p;

                        /* How much payload data has been filled in? */
                        cur_len = 0;
                        curbuf_p = pdu_p->payload_p;
                        while (curbuf_p->cur_len == curbuf_p->alloc_len) {
                                cur_len += curbuf_p->cur_len;
                                curbuf_p = curbuf_p->next;
                        }
                        cur_len += curbuf_p->cur_len;

                        /* How much payload data is left to be filled in? */
                        unread_len = pdu_p->hdr.payload_len - cur_len;

                        /* Read as much remaining payload data as possible */
                        iovcnt = 0;
                        while (curbuf_p->next != NULL) {
                                read_buf[iovcnt].iov_base = isns_buffer_data(
                                    curbuf_p, curbuf_p->cur_len); 
                                read_buf[iovcnt].iov_len = curbuf_p->alloc_len -
                                    curbuf_p->cur_len;
                                iovcnt++;
                                
                                curbuf_p = curbuf_p->next;
                        }
                        read_buf[iovcnt].iov_base = isns_buffer_data(curbuf_p,
                            curbuf_p->cur_len);
                        read_buf[iovcnt].iov_len = unread_len;
                        iovcnt++;

                        rv = isns_socket_readv(cfg_p->sd, iovp, iovcnt);
                        if ((rv == 0) || (rv == -1)) {
                                DBG("isns_kevent_socket: isns_socket_readv(2) "
                                    "returned %d\n",rv);
                                isns_free_pdu(cfg_p->pdu_in_p);
                                cfg_p->pdu_in_p = NULL;

                                transport_evt = 1;
                                break;
                        }

                        /* Update cur_len in buffers that newly have data */
                        curbuf_p = pdu_p->payload_p;
                        while (curbuf_p->cur_len == curbuf_p->alloc_len)
                                curbuf_p = curbuf_p->next;
                        
                        rd_len = (uint32_t)rv;
                        do {
                                b_len = curbuf_p->alloc_len - curbuf_p->cur_len;
                                if (rd_len > b_len) {
                                        curbuf_p->cur_len = curbuf_p->alloc_len;
                                        rd_len -= b_len;
                                } else {
                                        curbuf_p->cur_len += rd_len;
                                        break;
                                }

                                curbuf_p = curbuf_p->next;
                        } while (curbuf_p != NULL);

                        bavail -= rv;

                        if (rv == (int)unread_len)
                                isns_process_in_pdu(cfg_p);

                        more = (bavail > (int64_t)sizeof(struct isns_pdu_hdr_s)) ? 1 : 0;
                }
        }

        transport_evt |= (evt_p->flags & EV_EOF);
        if (transport_evt) {
                DBG("isns_kevent_socket: processing transport event\n");

                isns_socket_close(cfg_p->sd);
                cfg_p->sd_connected = 0;

                if (cfg_p->curtask_p != NULL)
                        isns_process_connection_loss(cfg_p);

                if (cfg_p->pdu_in_p != NULL) {
                        isns_free_pdu(cfg_p->pdu_in_p);
                        cfg_p->pdu_in_p = NULL;
                }
        }

        return 0;
}

/* ARGSUSED */
/*
 * isns_kevent_timer_recon()
 */
int
isns_kevent_timer_recon(struct kevent *evt_p, struct isns_config_s *cfg_p)
{
        int rv;


        DBG("isns_kevent_timer_recon: entered\n");

        rv = isns_socket_create(&(cfg_p->sd), cfg_p->ai_p->ai_family,
                cfg_p->ai_p->ai_socktype);
        if (rv != 0)
                return 0;

        rv = isns_socket_connect(cfg_p->sd, cfg_p->ai_p->ai_addr,
            cfg_p->ai_p->ai_addrlen);
        if (rv == 0) {
                /* Remove ISNS_EVT_TIMER_RECON from kqueue */
                rv = isns_change_kevent_list(cfg_p,
                    (uintptr_t)ISNS_EVT_TIMER_RECON, EVFILT_TIMER, EV_DELETE,
                    (int64_t)0, (intptr_t)0);
                if (rv == -1)
                        DBG("isns_kevent_timer_recon: error on "
                            "isns_change_kevent_list(1)\n");

                cfg_p->sd_connected = 1;

                /* Add cfg_p->sd to kqueue */
                rv = isns_change_kevent_list(cfg_p, (uintptr_t)cfg_p->sd,
                    EVFILT_READ, EV_ADD | EV_CLEAR, (int64_t)0,
                    (intptr_t)isns_kevent_socket);
                if (rv == -1)
                        DBG("isns_kevent_timer_recon: error on "
                            "isns_change_kevent_list(2)\n");

                isns_end_task(cfg_p->curtask_p);
        }

        return 0;
}


/* ARGSUSED */
/*
 * isns_kevent_timer_refresh
 */
int
isns_kevent_timer_refresh(struct kevent* evt_p, struct isns_config_s *cfg_p)
{
        struct isns_refresh_s *ref_p;
        ISNS_TRANS trans;
        uint32_t status;
        int rval;

        DBG("isns_kevent_timer_refresh: entered\n");

        /* If refresh info pointer NULL, or no name assigned, just return. */
        ref_p = cfg_p->refresh_p;
        if ((ref_p == NULL) || (ref_p->node[0] == '\0'))
                return 0;

        if (ref_p->trans_p != NULL) {
                /* If the previous refresh trans is not complete, return. */
                rval = isns_get_pdu_response_status(ref_p->trans_p, &status);
                if (rval == EPERM) {
                        DBG("isns_kevent_timer_refresh: "
                            "prev refresh trans not complete\n");
                        return 0;
                }
                /* Free previous refresh trans. */
                isns_free_trans(ref_p->trans_p);
                ref_p->trans_p = NULL;
        }

        /* Build new refresh transaction and send it. */
        trans = isns_new_trans((ISNS_HANDLE)cfg_p, isnsp_DevAttrQry, 0);
        if (trans == ISNS_INVALID_TRANS) {
                DBG("isns_kevent_timer_refresh: error on isns_new_trans()\n");
                return 0;
        }

        ref_p->trans_p = (struct isns_trans_s *)trans;
        /* First we add our source attribute */
        isns_add_string(trans, isnst_iSCSIName, ref_p->node);
        /* Now add our message attribute */
        isns_add_string(trans, isnst_iSCSIName, ref_p->node);
        isns_add_tlv(trans, isnst_Delimiter, 0, NULL);
        /* and finally the operating attributes */
        isns_add_tlv(trans, isnst_EID, 0, NULL);
        isns_send_trans(trans, NULL, NULL);

        return 0;
}