conrpc.c

lustre 1.6.5 source code

                LASSERT (mksn_rep->mksn_status == EBUSY ||
                         mksn_rep->mksn_status == EINVAL);

                lstcon_sesop_stat_failure(stat, 1);
                errno = mksn_rep->mksn_status;
                break;

        case LST_TRANS_SESEND:
                rmsn_rep = &msg->msg_body.rmsn_reply;
                /* ESRCH is not an error for end session */
                if (rmsn_rep->rmsn_status == 0 ||
                    rmsn_rep->rmsn_status == ESRCH) {
                        lstcon_sesop_stat_success(stat, 1);
                        return;
                }

                LASSERT (rmsn_rep->rmsn_status == EBUSY ||
                         rmsn_rep->rmsn_status == EINVAL);

                lstcon_sesop_stat_failure(stat, 1);
                errno = rmsn_rep->rmsn_status;
                break;

        case LST_TRANS_SESQRY:
        case LST_TRANS_SESPING:
                dbg_rep = &msg->msg_body.dbg_reply;

                if (dbg_rep->dbg_status == ESRCH) {
                        lstcon_sesqry_stat_unknown(stat, 1);
                        return;
                } 

                LASSERT (dbg_rep->dbg_status == 0);

                if (lstcon_session_match(dbg_rep->dbg_sid))
                        lstcon_sesqry_stat_active(stat, 1);
                else
                        lstcon_sesqry_stat_busy(stat, 1);
                return;

        case LST_TRANS_TSBRUN:
        case LST_TRANS_TSBSTOP:
                bat_rep = &msg->msg_body.bat_reply;

                if (bat_rep->bar_status == 0) {
                        lstcon_tsbop_stat_success(stat, 1);
                        return;
                }

                if (bat_rep->bar_status == EPERM && 
                    transop == LST_TRANS_TSBSTOP) {
                        lstcon_tsbop_stat_success(stat, 1);
                        return;
                }

                lstcon_tsbop_stat_failure(stat, 1);
                errno = bat_rep->bar_status;
                break;

        case LST_TRANS_TSBCLIQRY:
        case LST_TRANS_TSBSRVQRY:
                bat_rep = &msg->msg_body.bat_reply;

                if (bat_rep->bar_active != 0) 
                        lstcon_tsbqry_stat_run(stat, 1);
                else
                        lstcon_tsbqry_stat_idle(stat, 1);

                if (bat_rep->bar_status == 0) 
                        return;

                lstcon_tsbqry_stat_failure(stat, 1);
                errno = bat_rep->bar_status;
                break;

        case LST_TRANS_TSBCLIADD:
        case LST_TRANS_TSBSRVADD:
                test_rep = &msg->msg_body.tes_reply;

                if (test_rep->tsr_status == 0) {
                        lstcon_tsbop_stat_success(stat, 1);
                        return;
                }

                lstcon_tsbop_stat_failure(stat, 1);
                errno = test_rep->tsr_status;
                break;

        case LST_TRANS_STATQRY:
                stat_rep = &msg->msg_body.stat_reply;

                if (stat_rep->str_status == 0) {
                        lstcon_statqry_stat_success(stat, 1);
                        return;
                }

                lstcon_statqry_stat_failure(stat, 1);
                errno = stat_rep->str_status;
                break;

        default:
                LBUG();
        }

        if (stat->trs_fwk_errno == 0)
                stat->trs_fwk_errno = errno;

        return;
}

int
lstcon_rpc_trans_ndlist(struct list_head *ndlist,
                        struct list_head *translist, int transop,
                        void *arg, lstcon_rpc_cond_func_t condition,
                        lstcon_rpc_trans_t **transpp)
{
        lstcon_rpc_trans_t *trans;
        lstcon_ndlink_t    *ndl;
        lstcon_node_t      *nd;
        lstcon_rpc_t       *rpc;
        int                 rc;

        /* Creating session RPG for list of nodes */

        rc = lstcon_rpc_trans_prep(translist, transop, &trans);
        if (rc != 0) {
                CERROR("Can't create transaction %d: %d\n", transop, rc);
                return rc;
        }

        list_for_each_entry(ndl, ndlist, ndl_link) {
                rc = condition == NULL ? 1 :
                     condition(transop, ndl->ndl_node, arg);

                if (rc == 0)
                        continue;

                if (rc < 0) {
                        CDEBUG(D_NET, "Condition error while creating RPC "
                                      " for transaction %d: %d\n", transop, rc);
                        break;
                }

                nd = ndl->ndl_node;

                switch (transop) {
                case LST_TRANS_SESNEW:
                case LST_TRANS_SESEND:
                        rc = lstcon_sesrpc_prep(nd, transop, &rpc);
                        break;
                case LST_TRANS_SESQRY:
                case LST_TRANS_SESPING:
                        rc = lstcon_dbgrpc_prep(nd, &rpc);
                        break;
                case LST_TRANS_TSBCLIADD:
                case LST_TRANS_TSBSRVADD:
                        rc = lstcon_testrpc_prep(nd, transop,
                                                 (lstcon_test_t *)arg, &rpc);
                        break;
                case LST_TRANS_TSBRUN:
                case LST_TRANS_TSBSTOP:
                case LST_TRANS_TSBCLIQRY:
                case LST_TRANS_TSBSRVQRY:
                        rc = lstcon_batrpc_prep(nd, transop,
                                                (lstcon_tsb_hdr_t *)arg, &rpc);
                        break;
                case LST_TRANS_STATQRY:
                        rc = lstcon_statrpc_prep(nd, &rpc);
                        break;
                default:
                        rc = -EINVAL;
                        break;
                }

                if (rc != 0) {
                        CERROR("Failed to create RPC for transaction %s: %d\n",
                               lstcon_rpc_trans_name(transop), rc);
                        break;
                }
                                
                lstcon_rpc_trans_addreq(trans, rpc);
        }

        if (rc == 0) {
                *transpp = trans;
                return 0;
        }

        lstcon_rpc_trans_destroy(trans);

        return rc;
}

void
lstcon_rpc_pinger(void *arg)
{
        stt_timer_t        *ptimer = (stt_timer_t *)arg;
        lstcon_rpc_trans_t *trans;
        lstcon_rpc_t       *crpc;
        srpc_msg_t         *rep;
        srpc_debug_reqst_t *drq;
        lstcon_ndlink_t    *ndl;
        lstcon_node_t      *nd;
        time_t              intv;
        int                 count = 0;
        int                 rc;

        /* RPC pinger is a special case of transaction,
         * it's called by timer at 8 seconds interval.
         */
        mutex_down(&console_session.ses_mutex);

        if (console_session.ses_shutdown || console_session.ses_expired) {
                mutex_up(&console_session.ses_mutex);
                return;
        }

        if (!console_session.ses_expired &&
            cfs_time_current_sec() - console_session.ses_laststamp >
            console_session.ses_timeout)
                console_session.ses_expired = 1;

        trans = console_session.ses_ping;

        LASSERT (trans != NULL);

        list_for_each_entry(ndl, &console_session.ses_ndl_list, ndl_link) {
                nd = ndl->ndl_node;

                if (console_session.ses_expired) {
                        /* idle console, end session on all nodes */
                        if (nd->nd_state != LST_NODE_ACTIVE)
                                continue;

                        rc = lstcon_sesrpc_prep(nd, LST_TRANS_SESEND, &crpc);
                        if (rc != 0) {
                                CERROR("Out of memory\n");
                                break;
                        }

                        lstcon_rpc_trans_addreq(trans, crpc);
                        lstcon_rpc_post(crpc);

                        continue;
                }

                crpc = &nd->nd_ping;

                if (crpc->crp_rpc != NULL) {
                        LASSERT (crpc->crp_trans == trans);
                        LASSERT (!list_empty(&crpc->crp_link));

                        spin_lock(&crpc->crp_rpc->crpc_lock);

                        LASSERT (crpc->crp_posted);

                        if (!crpc->crp_finished) {
                                /* in flight */
                                spin_unlock(&crpc->crp_rpc->crpc_lock);
                                continue;
                        }

                        spin_unlock(&crpc->crp_rpc->crpc_lock);

                        lstcon_rpc_get_reply(crpc, &rep);

                        list_del_init(&crpc->crp_link);
                
                        lstcon_rpc_put(crpc);
                }

                if (nd->nd_state != LST_NODE_ACTIVE)
                        continue;

                intv = cfs_duration_sec(cfs_time_sub(cfs_time_current(),
                                                     nd->nd_stamp));
                if (intv < nd->nd_timeout / 2)
                        continue;

                rc = lstcon_rpc_init(nd, SRPC_SERVICE_DEBUG, 0, 0, crpc);
                if (rc != 0) {
                        CERROR("Out of memory\n");
                        break;
                }

                drq = &crpc->crp_rpc->crpc_reqstmsg.msg_body.dbg_reqst;

                drq->dbg_sid   = console_session.ses_id;
                drq->dbg_flags = 0;

                lstcon_rpc_trans_addreq(trans, crpc);
                lstcon_rpc_post(crpc);

                count ++;
        }

        if (console_session.ses_expired) {
                mutex_up(&console_session.ses_mutex);
                return;
        }

        CDEBUG(D_NET, "Ping %d nodes in session\n", count);

        ptimer->stt_expires = cfs_time_current_sec() + LST_PING_INTERVAL;
        stt_add_timer(ptimer);

        mutex_up(&console_session.ses_mutex);
}

int
lstcon_rpc_pinger_start(void)
{
        stt_timer_t    *ptimer;
        int             rc;

        LASSERT (list_empty(&console_session.ses_rpc_freelist));
        LASSERT (atomic_read(&console_session.ses_rpc_counter) == 0);

        rc = lstcon_rpc_trans_prep(NULL, LST_TRANS_SESPING,
                                   &console_session.ses_ping);
        if (rc != 0) {
                CERROR("Failed to create console pinger\n");
                return rc;
        }

        ptimer = &console_session.ses_ping_timer;
        ptimer->stt_expires = cfs_time_current_sec() + LST_PING_INTERVAL;

        stt_add_timer(ptimer);

        return 0;
}

void
lstcon_rpc_pinger_stop(void)
{
        LASSERT (console_session.ses_shutdown);

        stt_del_timer(&console_session.ses_ping_timer);

        lstcon_rpc_trans_abort(console_session.ses_ping, -ESHUTDOWN);
        lstcon_rpc_trans_stat(console_session.ses_ping, lstcon_trans_stat());
        lstcon_rpc_trans_destroy(console_session.ses_ping);

        memset(lstcon_trans_stat(), 0, sizeof(lstcon_trans_stat_t));

        console_session.ses_ping = NULL;
}

void
lstcon_rpc_cleanup_wait(void)
{
        lstcon_rpc_trans_t *trans;
        lstcon_rpc_t       *crpc;
        struct list_head   *pacer;
        struct list_head    zlist;

        /* Called with hold of global mutex */

        LASSERT (console_session.ses_shutdown);

        while (!list_empty(&console_session.ses_trans_list)) { 
                list_for_each(pacer, &console_session.ses_trans_list) {
                        trans = list_entry(pacer, lstcon_rpc_trans_t, tas_link);

                        CDEBUG(D_NET, "Session closed, wakeup transaction %s\n",
                               lstcon_rpc_trans_name(trans->tas_opc));

                        cfs_waitq_signal(&trans->tas_waitq);
                }

                mutex_up(&console_session.ses_mutex);

                CWARN("Session is shutting down, "
                      "waiting for termination of transactions\n");
                cfs_pause(cfs_time_seconds(1));

                mutex_down(&console_session.ses_mutex);
        }

        spin_lock(&console_session.ses_rpc_lock);

        lst_wait_until((atomic_read(&console_session.ses_rpc_counter) == 0),
                       console_session.ses_rpc_lock,
                       "Network is not accessable or target is down, "
                       "waiting for %d console RPCs to being recycled\n",
                       atomic_read(&console_session.ses_rpc_counter));

        list_add(&zlist, &console_session.ses_rpc_freelist);
        list_del_init(&console_session.ses_rpc_freelist);

        spin_unlock(&console_session.ses_rpc_lock);

        while (!list_empty(&zlist)) {
                crpc = list_entry(zlist.next, lstcon_rpc_t, crp_link);

                list_del(&crpc->crp_link);
                LIBCFS_FREE(crpc, sizeof(lstcon_rpc_t));
        }
}

int
lstcon_rpc_module_init(void)
{
        CFS_INIT_LIST_HEAD(&console_session.ses_ping_timer.stt_list);
        console_session.ses_ping_timer.stt_func = lstcon_rpc_pinger;
        console_session.ses_ping_timer.stt_data = &console_session.ses_ping_timer;

        console_session.ses_ping = NULL;

        spin_lock_init(&console_session.ses_rpc_lock);
        atomic_set(&console_session.ses_rpc_counter, 0);
        CFS_INIT_LIST_HEAD(&console_session.ses_rpc_freelist);

        return 0;
}

void
lstcon_rpc_module_fini(void)
{
        LASSERT (list_empty(&console_session.ses_rpc_freelist));
        LASSERT (atomic_read(&console_session.ses_rpc_counter) == 0);
}

#endif