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         if (request->rq_replen == 0) {
 167                 printk(__FUNCTION__ ": request->rq_replen is 0!\n");
 168                 EXIT;
 169                 return -EINVAL;
 170         }
 171
 172         request->rq_repbuf = kmalloc(request->rq_replen, GFP_KERNEL);
 173         if (!request->rq_repbuf) {
 174                 EXIT;
 175                 return -ENOMEM;
 176         }
 177
 178         local_id.addr_kind = PTL_ADDR_GID;
 179         local_id.gid = PTL_ID_ANY;
 180         local_id.rid = PTL_ID_ANY;
 181
 182         rc = PtlMEAttach(peer->peer_ni, request->rq_reply_portal, local_id,
 183                          request->rq_xid, 0, PTL_UNLINK, &me_h);
 184         if (rc != PTL_OK) {
 185                 EXIT;
 186                 /* FIXME: tear down EQ, free reqbuf */
 187                 return rc;
 188         }
 189
 190         request->rq_reply_md.start = request->rq_repbuf;
 191         request->rq_reply_md.length = request->rq_replen;
 192         request->rq_reply_md.threshold = 1;
 193         request->rq_reply_md.options = PTL_MD_OP_PUT;
 194         request->rq_reply_md.user_ptr = request;
 195         request->rq_reply_md.eventq = req_eq;
 196
 197         rc = PtlMDAttach(me_h, request->rq_reply_md, PTL_UNLINK,
 198                          &request->rq_reply_md_h);
 199         if (rc != PTL_OK) {
 200                 EXIT;
 201                 return rc;
 202         }
 203
 204         if (request->rq_bulklen != 0) {
 205                 rc = PtlMEAttach(peer->peer_ni, request->rq_bulk_portal,
 206                                  local_id, request->rq_xid, 0, PTL_UNLINK,
 207                                  &bulk_me_h);
 208                 if (rc != PTL_OK) {
 209                         EXIT;
 210                         return rc;
 211                 }
 212
 213                 request->rq_bulk_md.start = request->rq_bulkbuf;
 214                 request->rq_bulk_md.length = request->rq_bulklen;
 215                 request->rq_bulk_md.threshold = 1;
 216                 request->rq_bulk_md.options = PTL_MD_OP_PUT;
 217                 request->rq_bulk_md.user_ptr = request;
 218                 request->rq_bulk_md.eventq = bulk_sink_eq;
 219
 220                 rc = PtlMDAttach(bulk_me_h, request->rq_bulk_md, PTL_UNLINK,
 221                                  &request->rq_bulk_md_h);
 222                 if (rc != PTL_OK) {
 223                         EXIT;
 224                         return rc;
 225                 }
 226         }
 227
 228         return ptl_send_buf(request, peer, request->rq_req_portal, 1);
 229 }
 230
 231 int rpc_register_service(struct ptlrpc_service *service, char *uuid)
 232 {
 233         struct lustre_peer peer;
 234         int rc;
 235
 236         rc = kportal_uuid_to_peer(uuid, &peer);
 237         if (rc != 0) {
 238                 printk("Invalid uuid \"%s\"\n", uuid);
 239                 return -EINVAL;
 240         }
 241
 242         service->srv_buf = kmalloc(service->srv_buf_size, GFP_KERNEL);
 243         if (service->srv_buf == NULL) {
 244                 printk(__FUNCTION__ ": no memory\n");
 245                 return -ENOMEM;
 246         }
 247
 248         service->srv_id.addr_kind = PTL_ADDR_GID;
 249         service->srv_id.gid = PTL_ID_ANY;
 250         service->srv_id.rid = PTL_ID_ANY;
 251
 252         rc = PtlMEAttach(peer.peer_ni, service->srv_portal, service->srv_id,
 253                          0, ~0, PTL_RETAIN, &service->srv_me_h);
 254         if (rc != PTL_OK) {
 255                 printk("PtlMEAttach failed: %d\n", rc);
 256                 return rc;
 257         }
 258
 259         rc = PtlEQAlloc(peer.peer_ni, 128, incoming_callback, service,
 260                         &service->srv_eq_h);
 261         if (rc != PTL_OK) {
 262                 printk("PtlEQAlloc failed: %d\n", rc);
 263                 return rc;
 264         }
 265
 266         /* FIXME: Build an auto-unlinking MD and build a ring. */
 267         /* FIXME: Make sure that these are reachable by DMA on well-known
 268          * addresses. */
 269         service->srv_md.start           = service->srv_buf;
 270         service->srv_md.length          = service->srv_buf_size;
 271         service->srv_md.threshold       = PTL_MD_THRESH_INF;
 272         service->srv_md.options         = PTL_MD_OP_PUT;
 273         service->srv_md.user_ptr        = service;
 274         service->srv_md.eventq          = service->srv_eq_h;
 275
 276         rc = PtlMDAttach(service->srv_me_h, service->srv_md,
 277                          PTL_RETAIN, &service->srv_md_h);
 278         if (rc != PTL_OK) {
 279                 printk("PtlMDAttach failed: %d\n", rc);
 280                 /* FIXME: wow, we need to clean up. */
 281                 return rc;
 282         }
 283
 284         return 0;
 285 }
 286
 287 int rpc_unregister_service(struct ptlrpc_service *service)
 288 {
 289         int rc;
 290
 291         rc = PtlMDUnlink(service->srv_md_h);
 292         if (rc)
 293                 printk(__FUNCTION__ ": PtlMDUnlink failed: %d\n", rc);
 294         rc = PtlEQFree(service->srv_eq_h);
 295         if (rc)
 296                 printk(__FUNCTION__ ": PtlEQFree failed: %d\n", rc);
 297         rc = PtlMEUnlink(service->srv_me_h);
 298         if (rc)
 299                 printk(__FUNCTION__ ": PtlMEUnlink failed: %d\n", rc);
 300
 301         kfree(service->srv_buf);
 302         return 0;
 303 }
 304
 305 static int req_init_portals(void)
 306 {
 307         int rc;
 308         const ptl_handle_ni_t *nip;
 309         ptl_handle_ni_t ni;
 310
 311         nip = inter_module_get_request(LUSTRE_NAL "_ni", LUSTRE_NAL);
 312         if (nip == NULL) {
 313                 printk("get_ni failed: is the NAL module loaded?\n");
 314                 return -EIO;
 315         }
 316         ni = *nip;
 317
 318         rc = PtlEQAlloc(ni, 128, request_callback, NULL, &req_eq);
 319         if (rc != PTL_OK)
 320                 printk("PtlEQAlloc failed: %d\n", rc);
 321
 322         rc = PtlEQAlloc(ni, 128, bulk_source_callback, NULL, &bulk_source_eq);
 323         if (rc != PTL_OK)
 324                 printk("PtlEQAlloc failed: %d\n", rc);
 325
 326         rc = PtlEQAlloc(ni, 128, bulk_sink_callback, NULL, &bulk_sink_eq);
 327         if (rc != PTL_OK)
 328                 printk("PtlEQAlloc failed: %d\n", rc);
 329
 330         return rc;
 331 }
 332
 333 static int __init ptlrpc_init(void)
 334 {
 335         return req_init_portals();
 336 }
 337
 338 static void __exit ptlrpc_exit(void)
 339 {
 340         PtlEQFree(req_eq);
 341
 342         inter_module_put(LUSTRE_NAL "_ni");
 343
 344         return;
 345 }
 346
 347 MODULE_AUTHOR("Peter J. Braam <braam@clusterfs.com>");
 348 MODULE_DESCRIPTION("Lustre Request Processor v1.0");
 349 MODULE_LICENSE("GPL");
 350
 351 module_init(ptlrpc_init);
 352 module_exit(ptlrpc_exit);
 353