<commit_before>/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
 *
 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
 * CA 95054 USA or visit www.sun.com if you need additional information or
 * have any questions.
 *
 * GPL HEADER END
 */
/*
 * Copyright  2008 Sun Microsystems, Inc. All rights reserved
 * Use is subject to license terms.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 *
 * lnet/ulnds/socklnd/handlers.c
 *
 * Author: <NAME> <<EMAIL>>
 */

#include "usocklnd.h"
#include <unistd.h>
#include <sys/syscall.h>

#include "psc_util/iostats.h"

int
usocklnd_notifier_handler(int fd)
{
        int notification;
        return syscall(SYS_read, fd, &notification, sizeof(notification));
}

void
usocklnd_exception_handler(usock_conn_t *conn)
{
        pthread_mutex_lock(&conn->uc_lock);

        if (conn->uc_state == UC_CONNECTING ||
            conn->uc_state == UC_SENDING_HELLO)
                usocklnd_conn_kill_locked(conn);

        pthread_mutex_unlock(&conn->uc_lock);
}

int
usocklnd_read_handler(usock_conn_t *conn)
{
        int rc;
        int continue_reading;
        int state;

  read_again:
        rc = 0;
        pthread_mutex_lock(&conn->uc_lock);
        state = conn->uc_state;

        /* process special case: LNET calls lnd_recv() asyncronously */
        if (state == UC_READY && conn->uc_rx_state == UC_RX_PARSE) {
                /* still don't have usocklnd_recv() called */
                rc = usocklnd_add_pollrequest(conn, POLL_RX_SET_REQUEST, 0);
                if (rc == 0)
                        conn->uc_rx_state = UC_RX_PARSE_WAIT;
                else
                        usocklnd_conn_kill_locked(conn);

                pthread_mutex_unlock(&conn->uc_lock);
                return rc;
        }

        pthread_mutex_unlock(&conn->uc_lock);
        /* From here and below the conn cannot be changed
         * asyncronously, except:
         * 1) usocklnd_send() can work with uc_tx_list and uc_zcack_list,
         * 2) usocklnd_shutdown() can change uc_state to UC_DEAD */

        switch (state) {

        case UC_RECEIVING_HELLO:
        case UC_READY:
                if (conn->uc_rx_nob_wanted != 0) {
                        /* read from conn fd as much wanted data as possible */
                        rc = usocklnd_read_data(conn);
                        if (rc == 0) /* partial read */
                                break;
                        if (rc < 0) {/* error happened or EOF */
                                usocklnd_conn_kill(conn);
                                break;
                        }
                }

                /* process incoming data */
                if (state == UC_READY )
                        rc = usocklnd_read_msg(conn, &continue_reading);
                else /* state == UC_RECEIVING_HELLO */
                        rc = usocklnd_read_hello(conn, &continue_reading);

                if (rc < 0) {
                        usocklnd_conn_kill(conn);
                        break;
                }

                if (continue_reading)
                        goto read_again;

                break;

        case UC_DEAD:
                break;

        default:
                LBUG();
        }

        return rc;
}

/* Switch on rx_state.
 * Return 0 on success, 1 if whole packet is read, else return <0
 * Always set cont_flag: 1 if we're ready to continue reading, else 0
 * NB: If whole packet is read, cont_flag will be set to zero to take
 * care of fairess
 */
int
usocklnd_read_msg(usock_conn_t *conn, int *cont_flag)
{
        int   rc = 0;
        __u64 cookie;

       *cont_flag = 0;

        /* smth. new emerged in RX part - let's process it */
        switch (conn->uc_rx_state) {
        case UC_RX_KSM_HEADER:
                if (conn->uc_flip) {
                        __swab32s(&conn->uc_rx_msg.ksm_type);
                        __swab32s(&conn->uc_rx_msg.ksm_csum);
                        __swab64s(&conn->uc_rx_msg.ksm_zc_req_cookie);
                        __swab64s(&conn->uc_rx_msg.ksm_zc_ack_cookie);
                }

                /* we never send packets for wich zc-acking is required */
                if (conn->uc_rx_msg.ksm_type != KSOCK_MSG_LNET ||
                    conn->uc_rx_msg.ksm_zc_ack_cookie != 0) {
                        conn->uc_errored = 1;
                        return -EPROTO;
                }

                /* zc_req will be processed later, when
                   lnet payload will be received */

                usocklnd_rx_lnethdr_state_transition(conn);
                *cont_flag = 1;
                break;

        case UC_RX_LNET_HEADER:
                if (the_lnet.ln_pid & LNET_PID_USERFLAG) {
                        /* replace dest_nid,pid (ksocknal sets its own) */
                        conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_nid =
                                cpu_to_le64(conn->uc_peer->up_ni->ni_nid);
                        conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_pid =
                                cpu_to_le32(the_lnet.ln_pid);

                } else if (conn->uc_peer->up_peerid.pid & LNET_PID_USERFLAG) {
                        /* Userspace peer */
                        lnet_process_id_t *id = &conn->uc_peer->up_peerid;
                        lnet_hdr_t        *lhdr = &conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr;

                        /* Substitute process ID assigned at connection time */
                        lhdr->src_pid = cpu_to_le32(id->pid);
                        lhdr->src_nid = cpu_to_le64(id->nid);
                }

                conn->uc_rx_state = UC_RX_PARSE;
                usocklnd_conn_addref(conn); /* ++ref while parsing */

		if (lnet_acceptor_masquerading())
			conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_nid =
			    conn->uc_peer->up_ni;

                rc = lnet_parse(conn->uc_peer->up_ni,
                                &conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr,
                                conn->uc_peerid.nid, conn, 0);

                if (rc < 0) {
                        /* I just received garbage: give up on this conn */
                        conn->uc_errored = 1;
                        usocklnd_conn_decref(conn);
                        return -EPROTO;
                }

                /* Race with usocklnd_recv() is possible */
                pthread_mutex_lock(&conn->uc_lock);
                LASSERT (conn->uc_rx_state == UC_RX_PARSE ||
                         conn->uc_rx_state == UC_RX_LNET_PAYLOAD);

                /* check whether usocklnd_recv() got called */
                if (conn->uc_rx_state == UC_RX_LNET_PAYLOAD)
                        *cont_flag = 1;
                pthread_mutex_unlock(&conn->uc_lock);
                break;

        case UC_RX_PARSE:
                LBUG(); /* it's error to be here, because this special
                         * case is handled by caller */
                break;

        case UC_RX_PARSE_WAIT:
                LBUG(); /* it's error to be here, because the conn
                         * shouldn't wait for POLLIN event in this<commit_msg>fix a boo boo<commit_after>/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 only,
 * as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License version 2 for more details (a copy is included
 * in the LICENSE file that accompanied this code).
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; If not, see
 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
 *
 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
 * CA 95054 USA or visit www.sun.com if you need additional information or
 * have any questions.
 *
 * GPL HEADER END
 */
/*
 * Copyright  2008 Sun Microsystems, Inc. All rights reserved
 * Use is subject to license terms.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 *
 * lnet/ulnds/socklnd/handlers.c
 *
 * Author: <NAME> <<EMAIL>>
 */

#include "usocklnd.h"
#include <unistd.h>
#include <sys/syscall.h>

#include "psc_util/iostats.h"

int
usocklnd_notifier_handler(int fd)
{
        int notification;
        return syscall(SYS_read, fd, &notification, sizeof(notification));
}

void
usocklnd_exception_handler(usock_conn_t *conn)
{
        pthread_mutex_lock(&conn->uc_lock);

        if (conn->uc_state == UC_CONNECTING ||
            conn->uc_state == UC_SENDING_HELLO)
                usocklnd_conn_kill_locked(conn);

        pthread_mutex_unlock(&conn->uc_lock);
}

int
usocklnd_read_handler(usock_conn_t *conn)
{
        int rc;
        int continue_reading;
        int state;

  read_again:
        rc = 0;
        pthread_mutex_lock(&conn->uc_lock);
        state = conn->uc_state;

        /* process special case: LNET calls lnd_recv() asyncronously */
        if (state == UC_READY && conn->uc_rx_state == UC_RX_PARSE) {
                /* still don't have usocklnd_recv() called */
                rc = usocklnd_add_pollrequest(conn, POLL_RX_SET_REQUEST, 0);
                if (rc == 0)
                        conn->uc_rx_state = UC_RX_PARSE_WAIT;
                else
                        usocklnd_conn_kill_locked(conn);

                pthread_mutex_unlock(&conn->uc_lock);
                return rc;
        }

        pthread_mutex_unlock(&conn->uc_lock);
        /* From here and below the conn cannot be changed
         * asyncronously, except:
         * 1) usocklnd_send() can work with uc_tx_list and uc_zcack_list,
         * 2) usocklnd_shutdown() can change uc_state to UC_DEAD */

        switch (state) {

        case UC_RECEIVING_HELLO:
        case UC_READY:
                if (conn->uc_rx_nob_wanted != 0) {
                        /* read from conn fd as much wanted data as possible */
                        rc = usocklnd_read_data(conn);
                        if (rc == 0) /* partial read */
                                break;
                        if (rc < 0) {/* error happened or EOF */
                                usocklnd_conn_kill(conn);
                                break;
                        }
                }

                /* process incoming data */
                if (state == UC_READY )
                        rc = usocklnd_read_msg(conn, &continue_reading);
                else /* state == UC_RECEIVING_HELLO */
                        rc = usocklnd_read_hello(conn, &continue_reading);

                if (rc < 0) {
                        usocklnd_conn_kill(conn);
                        break;
                }

                if (continue_reading)
                        goto read_again;

                break;

        case UC_DEAD:
                break;

        default:
                LBUG();
        }

        return rc;
}

/* Switch on rx_state.
 * Return 0 on success, 1 if whole packet is read, else return <0
 * Always set cont_flag: 1 if we're ready to continue reading, else 0
 * NB: If whole packet is read, cont_flag will be set to zero to take
 * care of fairess
 */
int
usocklnd_read_msg(usock_conn_t *conn, int *cont_flag)
{
        int   rc = 0;
        __u64 cookie;

       *cont_flag = 0;

        /* smth. new emerged in RX part - let's process it */
        switch (conn->uc_rx_state) {
        case UC_RX_KSM_HEADER:
                if (conn->uc_flip) {
                        __swab32s(&conn->uc_rx_msg.ksm_type);
                        __swab32s(&conn->uc_rx_msg.ksm_csum);
                        __swab64s(&conn->uc_rx_msg.ksm_zc_req_cookie);
                        __swab64s(&conn->uc_rx_msg.ksm_zc_ack_cookie);
                }

                /* we never send packets for wich zc-acking is required */
                if (conn->uc_rx_msg.ksm_type != KSOCK_MSG_LNET ||
                    conn->uc_rx_msg.ksm_zc_ack_cookie != 0) {
                        conn->uc_errored = 1;
                        return -EPROTO;
                }

                /* zc_req will be processed later, when
                   lnet payload will be received */

                usocklnd_rx_lnethdr_state_transition(conn);
                *cont_flag = 1;
                break;

        case UC_RX_LNET_HEADER:
                if (the_lnet.ln_pid & LNET_PID_USERFLAG) {
                        /* replace dest_nid,pid (ksocknal sets its own) */
                        conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_nid =
                                cpu_to_le64(conn->uc_peer->up_ni->ni_nid);
                        conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_pid =
                                cpu_to_le32(the_lnet.ln_pid);

                } else if (conn->uc_peer->up_peerid.pid & LNET_PID_USERFLAG) {
                        /* Userspace peer */
                        lnet_process_id_t *id = &conn->uc_peer->up_peerid;
                        lnet_hdr_t        *lhdr = &conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr;

                        /* Substitute process ID assigned at connection time */
                        lhdr->src_pid = cpu_to_le32(id->pid);
                        lhdr->src_nid = cpu_to_le64(id->nid);
                }

                conn->uc_rx_state = UC_RX_PARSE;
                usocklnd_conn_addref(conn); /* ++ref while parsing */

		if (lnet_acceptor_masquerading())
			conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_nid =
			    cpu_to_le64(conn->uc_peer->up_ni->ni_nid);

                rc = lnet_parse(conn->uc_peer->up_ni,
                                &conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr,
                                conn->uc_peerid.nid, conn, 0);

                if (rc < 0) {
                        /* I just received garbage: give up on this conn */
                        conn->uc_errored = 1;
                        usocklnd_conn_decref(conn);
                        return -EPROTO;
                }

                /* Race with usocklnd_recv() is possible */
                pthread_mutex_lock(&conn->uc_lock);
                LASSERT (conn->uc_rx_state == UC_RX_PARSE ||
                         conn->uc_rx_state == UC_RX_LNET_PAYLOAD);

                /* check whether usocklnd_recv() got called */
                if (conn->uc_rx_state == UC_RX_LNET_PAYLOAD)
                        *cont_flag = 1;
                pthread_mutex_unlock(&conn->uc_lock);
                break;

        case UC_RX_PARSE:
                LBUG(); /* it's error to be here, because this special
                         * case is handled by caller */
                break;

        case UC_RX_PARSE_WAIT:
                LBUG(); /* it's error to be here, because the conn
                         * shouldn't wait for POLLIN event in this
{'fraction_non_alphanumeric': 0.08908989529768911, 'fraction_numerical': 0.0076822997335976705, 'mean_word_length': 1.5553268956783284, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 110, 'https://': 0, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7229441', 'n_tokens_mistral': 5201, 'n_tokens_neox': 4731, 'n_words': 1196}