lustre/ptlrpc/rpc.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  Copyright (C) 2002 Cluster File Systems, Inc.
   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 #define EXPORT_SYMTAB
  24
  25 #include <linux/config.h>
  26 #include <linux/module.h>
  27 #include <linux/kernel.h>
  28
  29 #include <linux/obd_support.h>
  30 #include <linux/lustre_net.h>
  31
  32 static ptl_handle_eq_t req_eq, bulk_source_eq, bulk_sink_eq;
  33
  34 /* This callback performs two functions:
  35  *
  36  * 1. Free the request buffer after it has gone out on the wire
  37  * 2. Wake up the thread waiting for the reply once it comes in.
  38  */
  39 static int request_callback(ptl_event_t *ev, void *data)
  40 {
  41         struct ptlrpc_request *rpc = ev->mem_desc.user_ptr;
  42
  43         ENTRY;
  44
  45         if (ev->type == PTL_EVENT_SENT) {
  46                 kfree(ev->mem_desc.start);
  47         } else if (ev->type == PTL_EVENT_PUT) {
  48                 rpc->rq_repbuf = ev->mem_desc.start + ev->offset;
  49                 wake_up_interruptible(&rpc->rq_wait_for_rep);
  50         }
  51
  52         EXIT;
  53         return 1;
  54 }
  55
  56 static int incoming_callback(ptl_event_t *ev, void *data)
  57 {
  58         struct ptlrpc_service *service = data;
  59
  60         ENTRY;
  61
  62         if (ev->type == PTL_EVENT_PUT) {
  63                 wake_up(service->srv_wait_queue);
  64         } else {
  65                 printk("Unexpected event type: %d\n", ev->type);
  66         }
  67
  68         EXIT;
  69         return 0;
  70 }
  71
  72 static int bulk_source_callback(ptl_event_t *ev, void *data)
  73 {
  74         struct ptlrpc_request *rpc = ev->mem_desc.user_ptr;
  75
  76         ENTRY;
  77
  78         if (ev->type == PTL_EVENT_SENT) {
  79                 ;
  80         } else if (ev->type == PTL_EVENT_ACK) {
  81                 wake_up_interruptible(&rpc->rq_wait_for_bulk);
  82         } else {
  83                 printk("Unexpected event type in " __FUNCTION__ "!\n");
  84         }
  85
  86         EXIT;
  87         return 1;
  88 }
  89
  90 static int bulk_sink_callback(ptl_event_t *ev, void *data)
  91 {
  92         struct ptlrpc_request *rpc = ev->mem_desc.user_ptr;
  93
  94         ENTRY;
  95
  96         if (ev->type == PTL_EVENT_PUT) {
  97                 if (rpc->rq_bulkbuf != ev->mem_desc.start + ev->offset)
  98                         printk(__FUNCTION__ ": bulkbuf != mem_desc -- why?\n");
  99                 wake_up_interruptible(&rpc->rq_wait_for_bulk);
 100         } else {
 101                 printk("Unexpected event type in " __FUNCTION__ "!\n");
 102         }
 103
 104         EXIT;
 105         return 1;
 106 }
 107
 108 int ptl_send_buf(struct ptlrpc_request *request, struct lustre_peer *peer,
 109                  int portal, int is_request)
 110 {
 111         int rc;
 112         ptl_process_id_t remote_id;
 113         ptl_handle_md_t md_h;
 114
 115         /* FIXME: This is bad. */
 116         if (request->rq_bulklen) {
 117                 request->rq_req_md.start = request->rq_bulkbuf;
 118                 request->rq_req_md.length = request->rq_bulklen;
 119                 request->rq_req_md.eventq = bulk_source_eq;
 120         } else if (is_request) {
 121                 request->rq_req_md.start = request->rq_reqbuf;
 122                 request->rq_req_md.length = request->rq_reqlen;
 123                 request->rq_req_md.eventq = req_eq;
 124         } else {
 125                 request->rq_req_md.start = request->rq_repbuf;
 126                 request->rq_req_md.length = request->rq_replen;
 127                 request->rq_req_md.eventq = req_eq;
 128         }
 129         request->rq_req_md.threshold = 1;
 130         request->rq_req_md.options = PTL_MD_OP_PUT;
 131         request->rq_req_md.user_ptr = request;
 132
 133         rc = PtlMDBind(peer->peer_ni, request->rq_req_md, &md_h);
 134         if (rc != 0) {
 135                 printk(__FUNCTION__ ": PtlMDBind failed: %d\n", rc);
 136                 return rc;
 137         }
 138
 139         remote_id.addr_kind = PTL_ADDR_NID;
 140         remote_id.nid = peer->peer_nid;
 141         remote_id.pid = 0;
 142
 143         if (request->rq_bulklen) {
 144                 rc = PtlPut(md_h, PTL_ACK_REQ, remote_id, portal, 0,
 145                             request->rq_xid, 0, 0);
 146         } else {
 147                 rc = PtlPut(md_h, PTL_NOACK_REQ, remote_id, portal, 0,
 148                             request->rq_xid, 0, 0);
 149         }
 150         if (rc != PTL_OK) {
 151                 printk(__FUNCTION__ ": PtlPut failed: %d\n", rc);
 152                 /* FIXME: tear down md */
 153         }
 154
 155         return rc;
 156 }
 157
 158 int ptl_send_rpc(struct ptlrpc_request *request, struct lustre_peer *peer)
 159 {
 160         ptl_handle_me_t me_h, bulk_me_h;
 161         ptl_process_id_t local_id;
 162         int rc;
 163
 164         ENTRY;
 165
 166         request->rq_repbuf = kmalloc(request->rq_replen, GFP_KERNEL);
 167         if (!request->rq_repbuf) {
 168                 EXIT;
 169                 return -ENOMEM;
 170         }
 171
 172         local_id.addr_kind = PTL_ADDR_GID;
 173         local_id.gid = PTL_ID_ANY;
 174         local_id.rid = PTL_ID_ANY;
 175
 176         rc = PtlMEAttach(peer->peer_ni, request->rq_reply_portal, local_id,
 177                          request->rq_xid, 0, PTL_UNLINK, &me_h);
 178         if (rc != PTL_OK) {
 179                 EXIT;
 180                 /* FIXME: tear down EQ, free reqbuf */
 181                 return rc;
 182         }
 183
 184         request->rq_reply_md.start = request->rq_repbuf;
 185         request->rq_reply_md.length = request->rq_replen;
 186         request->rq_reply_md.threshold = 1;
 187         request->rq_reply_md.options = PTL_MD_OP_PUT;
 188         request->rq_reply_md.user_ptr = request;
 189         request->rq_reply_md.eventq = req_eq;
 190
 191         rc = PtlMDAttach(me_h, request->rq_reply_md, PTL_UNLINK,
 192                          &request->rq_reply_md_h);
 193         if (rc != PTL_OK) {
 194                 EXIT;
 195                 return rc;
 196         }
 197
 198         if (request->rq_bulklen != 0) {
 199                 rc = PtlMEAttach(peer->peer_ni, request->rq_bulk_portal,
 200                                  local_id, request->rq_xid, 0, PTL_UNLINK,
 201                                  &bulk_me_h);
 202                 if (rc != PTL_OK) {
 203                         EXIT;
 204                         return rc;
 205                 }
 206
 207                 request->rq_bulk_md.start = request->rq_bulkbuf;
 208                 request->rq_bulk_md.length = request->rq_bulklen;
 209                 request->rq_bulk_md.threshold = 1;
 210                 request->rq_bulk_md.options = PTL_MD_OP_PUT;
 211                 request->rq_bulk_md.user_ptr = request;
 212                 request->rq_bulk_md.eventq = bulk_sink_eq;
 213
 214                 rc = PtlMDAttach(bulk_me_h, request->rq_bulk_md, PTL_UNLINK,
 215                                  &request->rq_bulk_md_h);
 216                 if (rc != PTL_OK) {
 217                         EXIT;
 218                         return rc;
 219                 }
 220         }
 221
 222         return ptl_send_buf(request, peer, request->rq_req_portal, 1);
 223 }
 224
 225 int rpc_register_service(struct ptlrpc_service *service, char *uuid)
 226 {
 227         struct lustre_peer peer;
 228         int rc;
 229
 230         rc = kportal_uuid_to_peer(uuid, &peer);
 231         if (rc != 0) {
 232                 printk("Invalid uuid \"%s\"\n", uuid);
 233                 return -EINVAL;
 234         }
 235
 236         service->srv_buf = kmalloc(service->srv_buf_size, GFP_KERNEL);
 237         if (service->srv_buf == NULL) {
 238                 printk(__FUNCTION__ ": no memory\n");
 239                 return -ENOMEM;
 240         }
 241
 242         service->srv_id.addr_kind = PTL_ADDR_GID;
 243         service->srv_id.gid = PTL_ID_ANY;
 244         service->srv_id.rid = PTL_ID_ANY;
 245
 246         rc = PtlMEAttach(peer.peer_ni, service->srv_portal, service->srv_id,
 247                          0, ~0, PTL_RETAIN, &service->srv_me_h);
 248         if (rc != PTL_OK) {
 249                 printk("PtlMEAttach failed: %d\n", rc);
 250                 return rc;
 251         }
 252
 253         rc = PtlEQAlloc(peer.peer_ni, 128, incoming_callback, service,
 254                         &service->srv_eq_h);
 255         if (rc != PTL_OK) {
 256                 printk("PtlEQAlloc failed: %d\n", rc);
 257                 return rc;
 258         }
 259
 260         /* FIXME: Build an auto-unlinking MD and build a ring. */
 261         /* FIXME: Make sure that these are reachable by DMA on well-known
 262          * addresses. */
 263         service->srv_md.start           = service->srv_buf;
 264         service->srv_md.length          = service->srv_buf_size;
 265         service->srv_md.threshold       = PTL_MD_THRESH_INF;
 266         service->srv_md.options         = PTL_MD_OP_PUT;
 267         service->srv_md.user_ptr        = service;
 268         service->srv_md.eventq          = service->srv_eq_h;
 269
 270         rc = PtlMDAttach(service->srv_me_h, service->srv_md,
 271                          PTL_RETAIN, &service->srv_md_h);
 272         if (rc != PTL_OK) {
 273                 printk("PtlMDAttach failed: %d\n", rc);
 274                 /* FIXME: wow, we need to clean up. */
 275                 return rc;
 276         }
 277
 278         return 0;
 279 }
 280
 281 int rpc_unregister_service(struct ptlrpc_service *service)
 282 {
 283         int rc;
 284
 285         rc = PtlMDUnlink(service->srv_md_h);
 286         if (rc)
 287                 printk(__FUNCTION__ ": PtlMDUnlink failed: %d\n", rc);
 288         rc = PtlEQFree(service->srv_eq_h);
 289         if (rc)
 290                 printk(__FUNCTION__ ": PtlEQFree failed: %d\n", rc);
 291         rc = PtlMEUnlink(service->srv_me_h);
 292         if (rc)
 293                 printk(__FUNCTION__ ": PtlMEUnlink failed: %d\n", rc);
 294
 295         kfree(service->srv_buf);
 296 }
 297
 298 static int req_init_portals(void)
 299 {
 300         int rc;
 301         const ptl_handle_ni_t *nip;
 302         ptl_handle_ni_t ni;
 303
 304         nip = inter_module_get_request(LUSTRE_NAL "_ni", LUSTRE_NAL);
 305         if (nip == NULL) {
 306                 printk("get_ni failed: is the NAL module loaded?\n");
 307                 return -EIO;
 308         }
 309         ni = *nip;
 310
 311         rc = PtlEQAlloc(ni, 128, request_callback, NULL, &req_eq);
 312         if (rc != PTL_OK)
 313                 printk("PtlEQAlloc failed: %d\n", rc);
 314
 315         rc = PtlEQAlloc(ni, 128, bulk_source_callback, NULL, &bulk_source_eq);
 316         if (rc != PTL_OK)
 317                 printk("PtlEQAlloc failed: %d\n", rc);
 318
 319         rc = PtlEQAlloc(ni, 128, bulk_sink_callback, NULL, &bulk_sink_eq);
 320         if (rc != PTL_OK)
 321                 printk("PtlEQAlloc failed: %d\n", rc);
 322
 323         return rc;
 324 }
 325
 326 static int  init_module(void)
 327 {
 328         return req_init_portals();
 329 }
 330
 331 static void cleanup_module(void)
 332 {
 333         PtlEQFree(req_eq);
 334
 335         inter_module_put(LUSTRE_NAL "_ni");
 336
 337         return;
 338 }
 339
 340 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
 341 MODULE_DESCRIPTION("Lustre Request Processor v1.0");
 342 MODULE_LICENSE("GPL");
 343