ralnd_cb.c

非常经典的一个分布式系统

/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * Copyright (C) 2004 Cluster File Systems, Inc.
 *   Author: Eric Barton <[e-mail]>
 *
 *   This file is part of Lustre, http://www.lustre.org.
 *
 *   Lustre is free software; you can redistribute it and/or
 *   modify it under the terms of version 2 of the GNU General Public
 *   License as published by the Free Software Foundation.
 *
 *   Lustre 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 for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Lustre; if not, write to the Free Software
 *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 *
 */

#include "ralnd.h"

void
kranal_device_callback(RAP_INT32 devid, RAP_PVOID arg)
{
        kra_device_t *dev;
        int           i;
        unsigned long flags;

        CDEBUG(D_NET, "callback for device %d\n", devid);

        for (i = 0; i < kranal_data.kra_ndevs; i++) {

                dev = &kranal_data.kra_devices[i];
                if (dev->rad_id != devid)
                        continue;

                spin_lock_irqsave(&dev->rad_lock, flags);

                if (!dev->rad_ready) {
                        dev->rad_ready = 1;
                        wake_up(&dev->rad_waitq);
                }

                spin_unlock_irqrestore(&dev->rad_lock, flags);
                return;
        }

        CWARN("callback for unknown device %d\n", devid);
}

void
kranal_schedule_conn(kra_conn_t *conn)
{
        kra_device_t    *dev = conn->rac_device;
        unsigned long    flags;

        spin_lock_irqsave(&dev->rad_lock, flags);

        if (!conn->rac_scheduled) {
                kranal_conn_addref(conn);       /* +1 ref for scheduler */
                conn->rac_scheduled = 1;
                list_add_tail(&conn->rac_schedlist, &dev->rad_ready_conns);
                wake_up(&dev->rad_waitq);
        }

        spin_unlock_irqrestore(&dev->rad_lock, flags);
}

kra_tx_t *
kranal_get_idle_tx (void)
{
        unsigned long  flags;
        kra_tx_t      *tx;

        spin_lock_irqsave(&kranal_data.kra_tx_lock, flags);

        if (list_empty(&kranal_data.kra_idle_txs)) {
                spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
                return NULL;
        }

        tx = list_entry(kranal_data.kra_idle_txs.next, kra_tx_t, tx_list);
        list_del(&tx->tx_list);

        /* Allocate a new completion cookie.  It might not be needed, but we've
         * got a lock right now... */
        tx->tx_cookie = kranal_data.kra_next_tx_cookie++;

        spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);

        LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
        LASSERT (tx->tx_msg.ram_type == RANAL_MSG_NONE);
        LASSERT (tx->tx_conn == NULL);
        LASSERT (tx->tx_lntmsg[0] == NULL);
        LASSERT (tx->tx_lntmsg[1] == NULL);

        return tx;
}

void
kranal_init_msg(kra_msg_t *msg, int type)
{
        msg->ram_magic = RANAL_MSG_MAGIC;
        msg->ram_version = RANAL_MSG_VERSION;
        msg->ram_type = type;
        msg->ram_srcnid = kranal_data.kra_ni->ni_nid;
        /* ram_connstamp gets set when FMA is sent */
}

kra_tx_t *
kranal_new_tx_msg (int type)
{
        kra_tx_t *tx = kranal_get_idle_tx();

        if (tx != NULL)
                kranal_init_msg(&tx->tx_msg, type);

        return tx;
}

int
kranal_setup_immediate_buffer (kra_tx_t *tx, 
                               unsigned int niov, struct iovec *iov,
                               int offset, int nob)

{
        /* For now this is almost identical to kranal_setup_virt_buffer, but we
         * could "flatten" the payload into a single contiguous buffer ready
         * for sending direct over an FMA if we ever needed to. */

        LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
        LASSERT (nob >= 0);

        if (nob == 0) {
                tx->tx_buffer = NULL;
        } else {
                LASSERT (niov > 0);

                while (offset >= iov->iov_len) {
                        offset -= iov->iov_len;
                        niov--;
                        iov++;
                        LASSERT (niov > 0);
                }

                if (nob > iov->iov_len - offset) {
                        CERROR("Can't handle multiple vaddr fragments\n");
                        return -EMSGSIZE;
                }

                tx->tx_buffer = (void *)(((unsigned long)iov->iov_base) + offset);
        }

        tx->tx_buftype = RANAL_BUF_IMMEDIATE;
        tx->tx_nob = nob;
        return 0;
}

int
kranal_setup_virt_buffer (kra_tx_t *tx, 
                          unsigned int niov, struct iovec *iov,
                          int offset, int nob)

{
        LASSERT (nob > 0);
        LASSERT (niov > 0);
        LASSERT (tx->tx_buftype == RANAL_BUF_NONE);

        while (offset >= iov->iov_len) {
                offset -= iov->iov_len;
                niov--;
                iov++;
                LASSERT (niov > 0);
        }

        if (nob > iov->iov_len - offset) {
                CERROR("Can't handle multiple vaddr fragments\n");
                return -EMSGSIZE;
        }

        tx->tx_buftype = RANAL_BUF_VIRT_UNMAPPED;
        tx->tx_nob = nob;
        tx->tx_buffer = (void *)(((unsigned long)iov->iov_base) + offset);
        return 0;
}

int
kranal_setup_phys_buffer (kra_tx_t *tx, int nkiov, lnet_kiov_t *kiov,
                          int offset, int nob)
{
        RAP_PHYS_REGION *phys = tx->tx_phys;
        int              resid;

        CDEBUG(D_NET, "niov %d offset %d nob %d\n", nkiov, offset, nob);

        LASSERT (nob > 0);
        LASSERT (nkiov > 0);
        LASSERT (tx->tx_buftype == RANAL_BUF_NONE);

        while (offset >= kiov->kiov_len) {
                offset -= kiov->kiov_len;
                nkiov--;
                kiov++;
                LASSERT (nkiov > 0);
        }

        tx->tx_buftype = RANAL_BUF_PHYS_UNMAPPED;
        tx->tx_nob = nob;
        tx->tx_buffer = (void *)((unsigned long)(kiov->kiov_offset + offset));

        phys->Address = lnet_page2phys(kiov->kiov_page);
        phys++;

        resid = nob - (kiov->kiov_len - offset);
        while (resid > 0) {
                kiov++;
                nkiov--;
                LASSERT (nkiov > 0);

                if (kiov->kiov_offset != 0 ||
                    ((resid > PAGE_SIZE) &&
                     kiov->kiov_len < PAGE_SIZE)) {
                        /* Can't have gaps */
                        CERROR("Can't make payload contiguous in I/O VM:"
                               "page %d, offset %d, len %d \n",
                               (int)(phys - tx->tx_phys),
                               kiov->kiov_offset, kiov->kiov_len);
                        return -EINVAL;
                }

                if ((phys - tx->tx_phys) == LNET_MAX_IOV) {
                        CERROR ("payload too big (%d)\n", (int)(phys - tx->tx_phys));
                        return -EMSGSIZE;
                }

                phys->Address = lnet_page2phys(kiov->kiov_page);
                phys++;

                resid -= PAGE_SIZE;
        }

        tx->tx_phys_npages = phys - tx->tx_phys;
        return 0;
}

static inline int
kranal_setup_rdma_buffer (kra_tx_t *tx, unsigned int niov,
                          struct iovec *iov, lnet_kiov_t *kiov,
                          int offset, int nob)
{
        LASSERT ((iov == NULL) != (kiov == NULL));

        if (kiov != NULL)
                return kranal_setup_phys_buffer(tx, niov, kiov, offset, nob);

        return kranal_setup_virt_buffer(tx, niov, iov, offset, nob);
}

int
kranal_map_buffer (kra_tx_t *tx)
{
        kra_conn_t     *conn = tx->tx_conn;
        kra_device_t   *dev = conn->rac_device;
        RAP_RETURN      rrc;

        LASSERT (current == dev->rad_scheduler);

        switch (tx->tx_buftype) {
        default:
                LBUG();

        case RANAL_BUF_NONE:
        case RANAL_BUF_IMMEDIATE:
        case RANAL_BUF_PHYS_MAPPED:
        case RANAL_BUF_VIRT_MAPPED:
                return 0;

        case RANAL_BUF_PHYS_UNMAPPED:
                rrc = RapkRegisterPhys(dev->rad_handle,
                                       tx->tx_phys, tx->tx_phys_npages,
                                       &tx->tx_map_key);
                if (rrc != RAP_SUCCESS) {
                        CERROR ("Can't map %d pages: dev %d "
                                "phys %u pp %u, virt %u nob %lu\n",
                                tx->tx_phys_npages, dev->rad_id, 
                                dev->rad_nphysmap, dev->rad_nppphysmap,
                                dev->rad_nvirtmap, dev->rad_nobvirtmap);
                        return -ENOMEM; /* assume insufficient resources */
                }

                dev->rad_nphysmap++;
                dev->rad_nppphysmap += tx->tx_phys_npages;

                tx->tx_buftype = RANAL_BUF_PHYS_MAPPED;
                return 0;

        case RANAL_BUF_VIRT_UNMAPPED:
                rrc = RapkRegisterMemory(dev->rad_handle,
                                         tx->tx_buffer, tx->tx_nob,
                                         &tx->tx_map_key);
                if (rrc != RAP_SUCCESS) {
                        CERROR ("Can't map %d bytes: dev %d "
                                "phys %u pp %u, virt %u nob %lu\n",
                                tx->tx_nob, dev->rad_id, 
                                dev->rad_nphysmap, dev->rad_nppphysmap,
                                dev->rad_nvirtmap, dev->rad_nobvirtmap);
                        return -ENOMEM; /* assume insufficient resources */
                }

                dev->rad_nvirtmap++;
                dev->rad_nobvirtmap += tx->tx_nob;

                tx->tx_buftype = RANAL_BUF_VIRT_MAPPED;
                return 0;
        }
}

void
kranal_unmap_buffer (kra_tx_t *tx)
{
        kra_device_t   *dev;
        RAP_RETURN      rrc;

        switch (tx->tx_buftype) {
        default:
                LBUG();

        case RANAL_BUF_NONE:
        case RANAL_BUF_IMMEDIATE:
        case RANAL_BUF_PHYS_UNMAPPED:
        case RANAL_BUF_VIRT_UNMAPPED:
                break;

        case RANAL_BUF_PHYS_MAPPED:
                LASSERT (tx->tx_conn != NULL);
                dev = tx->tx_conn->rac_device;
                LASSERT (current == dev->rad_scheduler);
                rrc = RapkDeregisterMemory(dev->rad_handle, NULL,
                                           &tx->tx_map_key);
                LASSERT (rrc == RAP_SUCCESS);

                dev->rad_nphysmap--;
                dev->rad_nppphysmap -= tx->tx_phys_npages;

                tx->tx_buftype = RANAL_BUF_PHYS_UNMAPPED;
                break;

        case RANAL_BUF_VIRT_MAPPED:
                LASSERT (tx->tx_conn != NULL);
                dev = tx->tx_conn->rac_device;
                LASSERT (current == dev->rad_scheduler);
                rrc = RapkDeregisterMemory(dev->rad_handle, tx->tx_buffer,
                                           &tx->tx_map_key);
                LASSERT (rrc == RAP_SUCCESS);

                dev->rad_nvirtmap--;
                dev->rad_nobvirtmap -= tx->tx_nob;

                tx->tx_buftype = RANAL_BUF_VIRT_UNMAPPED;
                break;
        }
}

void
kranal_tx_done (kra_tx_t *tx, int completion)
{
        lnet_msg_t      *lnetmsg[2];
        unsigned long    flags;
        int              i;

        LASSERT (!in_interrupt());