lustre/portals/knals/gmnal/gmnal_cb.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (c) 2003 Los Alamos National Laboratory (LANL)
   5  *
   6  *   This file is part of Lustre, http://www.lustre.org/
   7  *
   8  *   Lustre is free software; you can redistribute it and/or
   9  *   modify it under the terms of version 2 of the GNU General Public
  10  *   License as published by the Free Software Foundation.
  11  *
  12  *   Lustre is distributed in the hope that it will be useful,
  13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *   GNU General Public License for more details.
  16  *
  17  *   You should have received a copy of the GNU General Public License
  18  *   along with Lustre; if not, write to the Free Software
  19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20  */
  21
  22
  23 /*
  24  *      This file implements the nal cb functions
  25  */
  26
  27
  28 #include "gmnal.h"
  29
  30 ptl_err_t gmnal_cb_recv(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
  31                    unsigned int niov, struct iovec *iov, size_t offset,
  32                    size_t mlen, size_t rlen)
  33 {
  34         gmnal_srxd_t    *srxd = (gmnal_srxd_t*)private;
  35         int             status = PTL_OK;
  36
  37
  38         CDEBUG(D_TRACE, "gmnal_cb_recv libnal [%p], private[%p], cookie[%p], "
  39                "niov[%d], iov [%p], offset["LPSZ"], mlen["LPSZ"], rlen["LPSZ"]\n",
  40                libnal, private, cookie, niov, iov, offset, mlen, rlen);
  41
  42         switch(srxd->type) {
  43         case(GMNAL_SMALL_MESSAGE):
  44                 CDEBUG(D_INFO, "gmnal_cb_recv got small message\n");
  45                 status = gmnal_small_rx(libnal, private, cookie, niov,
  46                                          iov, offset, mlen, rlen);
  47         break;
  48         case(GMNAL_LARGE_MESSAGE_INIT):
  49                 CDEBUG(D_INFO, "gmnal_cb_recv got large message init\n");
  50                 status = gmnal_large_rx(libnal, private, cookie, niov,
  51                                          iov, offset, mlen, rlen);
  52         }
  53
  54
  55         CDEBUG(D_INFO, "gmnal_cb_recv gmnal_return status [%d]\n", status);
  56         return(status);
  57 }
  58
  59 ptl_err_t gmnal_cb_recv_pages(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
  60                          unsigned int kniov, ptl_kiov_t *kiov, size_t offset,
  61                          size_t mlen, size_t rlen)
  62 {
  63         gmnal_srxd_t    *srxd = (gmnal_srxd_t*)private;
  64         int             status = PTL_OK;
  65         struct iovec    *iovec = NULL, *iovec_dup = NULL;
  66         int             i = 0;
  67         ptl_kiov_t      *kiov_dup = kiov;;
  68
  69
  70         CDEBUG(D_TRACE, "gmnal_cb_recv_pages libnal [%p],private[%p], "
  71                "cookie[%p], kniov[%d], kiov [%p], offset["LPSZ"], mlen["LPSZ"], rlen["LPSZ"]\n",
  72                libnal, private, cookie, kniov, kiov, offset, mlen, rlen);
  73
  74         if (srxd->type == GMNAL_SMALL_MESSAGE) {
  75                 PORTAL_ALLOC(iovec, sizeof(struct iovec)*kniov);
  76                 if (!iovec) {
  77                         CDEBUG(D_ERROR, "Can't malloc\n");
  78                         return(GMNAL_STATUS_FAIL);
  79                 }
  80                 iovec_dup = iovec;
  81
  82                 /*
  83                  *      map each page and create an iovec for it
  84                  */
  85                 for (i=0; i<kniov; i++) {
  86                         CDEBUG(D_INFO, "processing kniov [%d] [%p]\n", i, kiov);
  87                         CDEBUG(D_INFO, "kniov page [%p] len [%d] offset[%d]\n",
  88                                kiov->kiov_page, kiov->kiov_len,
  89                                kiov->kiov_offset);
  90                         iovec->iov_len = kiov->kiov_len;
  91                         CDEBUG(D_INFO, "Calling kmap[%p]", kiov->kiov_page);
  92
  93                         iovec->iov_base = kmap(kiov->kiov_page) +
  94                                                   kiov->kiov_offset;
  95
  96                         CDEBUG(D_INFO, "iov_base is [%p]\n", iovec->iov_base);
  97                         iovec++;
  98                         kiov++;
  99                 }
 100                 CDEBUG(D_INFO, "calling gmnal_small_rx\n");
 101                 status = gmnal_small_rx(libnal, private, cookie, kniov,
 102                                          iovec_dup, offset, mlen, rlen);
 103                 for (i=0; i<kniov; i++) {
 104                         kunmap(kiov_dup->kiov_page);
 105                         kiov_dup++;
 106                 }
 107                 PORTAL_FREE(iovec_dup, sizeof(struct iovec)*kniov);
 108         }
 109
 110
 111         CDEBUG(D_INFO, "gmnal_return status [%d]\n", status);
 112         return(status);
 113 }
 114
 115
 116 ptl_err_t gmnal_cb_send(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
 117                    ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
 118                    unsigned int niov, struct iovec *iov, size_t offset, size_t len)
 119 {
 120
 121         gmnal_data_t    *nal_data;
 122
 123
 124         CDEBUG(D_TRACE, "gmnal_cb_send niov[%d] offset["LPSZ"] len["LPSZ"] nid["LPU64"]\n",
 125                niov, offset, len, nid);
 126         nal_data = libnal->libnal_data;
 127
 128         if (GMNAL_IS_SMALL_MESSAGE(nal_data, niov, iov, len)) {
 129                 CDEBUG(D_INFO, "This is a small message send\n");
 130                 gmnal_small_tx(libnal, private, cookie, hdr, type, nid, pid,
 131                                 niov, iov, offset,  len);
 132         } else {
 133                 CDEBUG(D_ERROR, "Large message send it is not supported\n");
 134                 lib_finalize(libnal, private, cookie, PTL_FAIL);
 135                 return(PTL_FAIL);
 136                 gmnal_large_tx(libnal, private, cookie, hdr, type, nid, pid,
 137                                 niov, iov, offset, len);
 138         }
 139         return(PTL_OK);
 140 }
 141
 142 ptl_err_t gmnal_cb_send_pages(lib_nal_t *libnal, void *private, lib_msg_t *cookie,
 143                          ptl_hdr_t *hdr, int type, ptl_nid_t nid, ptl_pid_t pid,
 144                          unsigned int kniov, ptl_kiov_t *kiov, size_t offset, size_t len)
 145 {
 146
 147         int     i = 0;
 148         gmnal_data_t    *nal_data;
 149         struct  iovec   *iovec = NULL, *iovec_dup = NULL;
 150         ptl_kiov_t      *kiov_dup = kiov;
 151
 152         CDEBUG(D_TRACE, "gmnal_cb_send_pages nid ["LPU64"] niov[%d] offset["LPSZ"] len["LPSZ"]\n",
 153                nid, kniov, offset, len);
 154         nal_data = libnal->libnal_data;
 155         PORTAL_ALLOC(iovec, kniov*sizeof(struct iovec));
 156         iovec_dup = iovec;
 157         if (GMNAL_IS_SMALL_MESSAGE(nal_data, 0, NULL, len)) {
 158                 CDEBUG(D_INFO, "This is a small message send\n");
 159
 160                 for (i=0; i<kniov; i++) {
 161                         CDEBUG(D_INFO, "processing kniov [%d] [%p]\n", i, kiov);
 162                         CDEBUG(D_INFO, "kniov page [%p] len [%d] offset[%d]\n",
 163                                kiov->kiov_page, kiov->kiov_len,
 164                                kiov->kiov_offset);
 165
 166                         iovec->iov_base = kmap(kiov->kiov_page)
 167                                                 + kiov->kiov_offset;
 168
 169                         iovec->iov_len = kiov->kiov_len;
 170                         iovec++;
 171                         kiov++;
 172                 }
 173                 gmnal_small_tx(libnal, private, cookie, hdr, type, nid,
 174                                 pid, kniov, iovec_dup, offset, len);
 175         } else {
 176                 CDEBUG(D_ERROR, "Large message send it is not supported yet\n");
 177                 return(PTL_FAIL);
 178                 for (i=0; i<kniov; i++) {
 179                         CDEBUG(D_INFO, "processing kniov [%d] [%p]\n", i, kiov);
 180                         CDEBUG(D_INFO, "kniov page [%p] len [%d] offset[%d]\n",
 181                                kiov->kiov_page, kiov->kiov_len,
 182                                kiov->kiov_offset);
 183
 184                         iovec->iov_base = kmap(kiov->kiov_page)
 185                                                  + kiov->kiov_offset;
 186                         iovec->iov_len = kiov->kiov_len;
 187                         iovec++;
 188                         kiov++;
 189                 }
 190                 gmnal_large_tx(libnal, private, cookie, hdr, type, nid,
 191                                 pid, kniov, iovec, offset, len);
 192         }
 193         for (i=0; i<kniov; i++) {
 194                 kunmap(kiov_dup->kiov_page);
 195                 kiov_dup++;
 196         }
 197         PORTAL_FREE(iovec_dup, kniov*sizeof(struct iovec));
 198         return(PTL_OK);
 199 }
 200
 201 int gmnal_cb_dist(lib_nal_t *libnal, ptl_nid_t nid, unsigned long *dist)
 202 {
 203         CDEBUG(D_TRACE, "gmnal_cb_dist\n");
 204         if (dist)
 205                 *dist = 27;
 206         return(PTL_OK);
 207 }