lustre/ptlrpc/niobuf.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 #define DEBUG_SUBSYSTEM S_RPC
  30
  31 #include <linux/obd_support.h>
  32 #include <linux/obd_class.h>
  33 #include <linux/lustre_net.h>
  34
  35 extern ptl_handle_eq_t bulk_source_eq, sent_pkt_eq, rcvd_rep_eq, bulk_sink_eq;
  36 static ptl_process_id_t local_id = {PTL_ID_ANY, PTL_ID_ANY};
  37
  38
  39 int ptlrpc_check_bulk_sent(struct ptlrpc_bulk_desc *bulk)
  40 {
  41         if (bulk->b_flags == PTL_BULK_SENT) {
  42                 EXIT;
  43                 return 1;
  44         }
  45
  46         if (sigismember(&(current->pending.signal), SIGKILL) ||
  47             sigismember(&(current->pending.signal), SIGINT)) {
  48                 bulk->b_flags = PTL_RPC_INTR;
  49                 EXIT;
  50                 return 1;
  51         }
  52
  53         CDEBUG(D_NET, "no event yet\n");
  54         return 0;
  55 }
  56
  57 int ptl_send_buf(struct ptlrpc_request *request, struct lustre_peer *peer,
  58                  int portal)
  59 {
  60         int rc;
  61         ptl_process_id_t remote_id;
  62         ptl_handle_md_t md_h;
  63         ptl_ack_req_t ack;
  64
  65         switch (request->rq_type) {
  66         case PTL_RPC_BULK:
  67                 request->rq_req_md.start = request->rq_bulkbuf;
  68                 request->rq_req_md.length = request->rq_bulklen;
  69                 request->rq_req_md.eventq = bulk_source_eq;
  70                 request->rq_req_md.threshold = 2; /* SENT and ACK events */
  71                 ack = PTL_ACK_REQ;
  72                 break;
  73         case PTL_RPC_REQUEST:
  74                 request->rq_req_md.start = request->rq_reqbuf;
  75                 request->rq_req_md.length = request->rq_reqlen;
  76                 request->rq_req_md.eventq = sent_pkt_eq;
  77                 request->rq_req_md.threshold = 1;
  78                 ack = PTL_NOACK_REQ;
  79                 break;
  80         case PTL_RPC_REPLY:
  81                 request->rq_req_md.start = request->rq_repbuf;
  82                 request->rq_req_md.length = request->rq_replen;
  83                 request->rq_req_md.eventq = sent_pkt_eq;
  84                 request->rq_req_md.threshold = 1;
  85                 ack = PTL_NOACK_REQ;
  86                 break;
  87         default:
  88                 LBUG();
  89                 return -1; /* notreached */
  90         }
  91         request->rq_req_md.options = PTL_MD_OP_PUT;
  92         request->rq_req_md.user_ptr = request;
  93
  94         rc = PtlMDBind(peer->peer_ni, request->rq_req_md, &md_h);
  95         //CERROR("MDBind (outgoing req/rep/bulk): %Lu\n", (__u64)md_h);
  96         if (rc != 0) {
  97                 CERROR("PtlMDBind failed: %d\n", rc);
  98                 LBUG();
  99                 return rc;
 100         }
 101
 102         remote_id.nid = peer->peer_nid;
 103         remote_id.pid = 0;
 104
 105         CDEBUG(D_NET, "Sending %d bytes to portal %d, xid %d\n",
 106                request->rq_req_md.length, portal, request->rq_xid);
 107
 108         rc = PtlPut(md_h, ack, remote_id, portal, 0, request->rq_xid, 0, 0);
 109         if (rc != PTL_OK) {
 110                 CERROR("PtlPut(%d, %d, %d) failed: %d\n", remote_id.nid,
 111                        portal, request->rq_xid, rc);
 112                 PtlMDUnlink(md_h);
 113         }
 114
 115         return rc;
 116 }
 117
 118 int ptlrpc_send_bulk(struct ptlrpc_bulk_desc *bulk, int portal)
 119 {
 120         int rc;
 121         ptl_process_id_t remote_id;
 122         ptl_handle_md_t md_h;
 123
 124         bulk->b_md.start = bulk->b_buf;
 125         bulk->b_md.length = bulk->b_buflen;
 126         bulk->b_md.eventq = bulk_source_eq;
 127         bulk->b_md.threshold = 2; /* SENT and ACK events */
 128         bulk->b_md.options = PTL_MD_OP_PUT;
 129         bulk->b_md.user_ptr = bulk;
 130
 131         rc = PtlMDBind(bulk->b_peer.peer_ni, bulk->b_md, &md_h);
 132         if (rc != 0) {
 133                 CERROR("PtlMDBind failed: %d\n", rc);
 134                 LBUG();
 135                 return rc;
 136         }
 137
 138         remote_id.nid = bulk->b_peer.peer_nid;
 139         remote_id.pid = 0;
 140
 141         CDEBUG(D_NET, "Sending %d bytes to portal %d, xid %d\n",
 142                bulk->b_md.length, portal, bulk->b_xid);
 143
 144         rc = PtlPut(md_h, PTL_ACK_REQ, remote_id, portal, 0, bulk->b_xid, 0, 0);
 145         if (rc != PTL_OK) {
 146                 CERROR("PtlPut(%d, %d, %d) failed: %d\n", remote_id.nid,
 147                        portal, bulk->b_xid, rc);
 148                 PtlMDUnlink(md_h);
 149                 LBUG();
 150         }
 151
 152         return rc;
 153 }
 154
 155 int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *bulk)
 156 {
 157         int rc;
 158
 159         ENTRY;
 160
 161         rc = PtlMEAttach(bulk->b_peer.peer_ni, bulk->b_portal, local_id,
 162                           bulk->b_xid, 0, PTL_UNLINK, PTL_INS_AFTER,
 163                          &bulk->b_me_h);
 164         if (rc != PTL_OK) {
 165                 CERROR("PtlMEAttach failed: %d\n", rc);
 166                 LBUG();
 167                 EXIT;
 168                 goto cleanup1;
 169         }
 170
 171         bulk->b_md.start = bulk->b_buf;
 172         bulk->b_md.length = bulk->b_buflen;
 173         bulk->b_md.threshold = 1;
 174         bulk->b_md.options = PTL_MD_OP_PUT;
 175         bulk->b_md.user_ptr = bulk;
 176         bulk->b_md.eventq = bulk_sink_eq;
 177
 178         rc = PtlMDAttach(bulk->b_me_h, bulk->b_md, PTL_UNLINK, &bulk->b_md_h);
 179         //CERROR("MDAttach (bulk sink): %Lu\n", (__u64)bulk->b_md_h);
 180         if (rc != PTL_OK) {
 181                 CERROR("PtlMDAttach failed: %d\n", rc);
 182                 LBUG();
 183                 EXIT;
 184                 goto cleanup2;
 185         }
 186
 187         CDEBUG(D_NET, "Setup bulk sink buffer: %u bytes, xid %u, portal %u\n",
 188                bulk->b_buflen, bulk->b_xid, bulk->b_portal);
 189         EXIT;
 190         return 0;
 191
 192         // XXX Confirm that this is safe!
 193  cleanup2:
 194         PtlMDUnlink(bulk->b_md_h);
 195  cleanup1:
 196         PtlMEUnlink(bulk->b_me_h);
 197         return rc;
 198 }
 199
 200 int ptlrpc_reply(struct obd_device *obddev, struct ptlrpc_service *svc,
 201                  struct ptlrpc_request *req)
 202 {
 203         struct ptlrpc_request *clnt_req = req->rq_reply_handle;
 204         ENTRY;
 205
 206         if (req->rq_reply_handle == NULL) {
 207                 /* This is a request that came from the network via portals. */
 208
 209                 /* FIXME: we need to increment the count of handled events */
 210                 req->rq_type = PTL_RPC_REPLY;
 211                 req->rq_reqhdr->xid = req->rq_reqhdr->xid;
 212                 ptl_send_buf(req, &req->rq_peer, svc->srv_rep_portal);
 213         } else {
 214                 /* This is a local request that came from another thread. */
 215
 216                 /* move the reply to the client */
 217                 clnt_req->rq_replen = req->rq_replen;
 218                 clnt_req->rq_repbuf = req->rq_repbuf;
 219                 req->rq_repbuf = NULL;
 220                 req->rq_replen = 0;
 221
 222                 /* free the request buffer */
 223                 OBD_FREE(req->rq_reqbuf, req->rq_reqlen);
 224                 req->rq_reqbuf = NULL;
 225
 226                 /* wake up the client */
 227                 wake_up_interruptible(&clnt_req->rq_wait_for_rep);
 228         }
 229
 230         EXIT;
 231         return 0;
 232 }
 233
 234 int ptlrpc_error(struct obd_device *obddev, struct ptlrpc_service *svc,
 235                  struct ptlrpc_request *req)
 236 {
 237         struct ptlrep_hdr *hdr;
 238
 239         ENTRY;
 240
 241         OBD_ALLOC(hdr, sizeof(*hdr));
 242         if (!hdr) {
 243                 EXIT;
 244                 return -ENOMEM;
 245         }
 246
 247         memset(hdr, 0, sizeof(*hdr));
 248
 249         hdr->xid = req->rq_reqhdr->xid;
 250         hdr->status = req->rq_status;
 251         hdr->type = PTL_RPC_ERR;
 252
 253         if (req->rq_repbuf) {
 254                 CERROR("req has repbuf\n");
 255                 LBUG();
 256         }
 257
 258         req->rq_repbuf = (char *)hdr;
 259         req->rq_replen = sizeof(*hdr);
 260
 261         EXIT;
 262         return ptlrpc_reply(obddev, svc, req);
 263 }
 264
 265 int ptl_send_rpc(struct ptlrpc_request *request, struct lustre_peer *peer)
 266 {
 267         ptl_process_id_t local_id;
 268         struct ptlreq_hdr *hdr;
 269         int rc;
 270         char *repbuf;
 271
 272         ENTRY;
 273
 274         hdr = (struct ptlreq_hdr *)request->rq_reqbuf;
 275         if (NTOH__u32(hdr->type) != PTL_RPC_REQUEST) {
 276                 CERROR("wrong packet type sent %d\n", NTOH__u32(hdr->type));
 277                 LBUG();
 278         }
 279         if (request->rq_replen == 0) {
 280                 CERROR("request->rq_replen is 0!\n");
 281                 EXIT;
 282                 return -EINVAL;
 283         }
 284
 285         /* request->rq_repbuf is set only when the reply comes in, in
 286          * client_packet_callback() */
 287         OBD_ALLOC(repbuf, request->rq_replen);
 288         if (!repbuf) {
 289                 EXIT;
 290                 return -ENOMEM;
 291         }
 292
 293         local_id.nid = PTL_ID_ANY;
 294         local_id.pid = PTL_ID_ANY;
 295
 296         rc = PtlMEAttach(peer->peer_ni, request->rq_reply_portal, local_id,
 297                           request->rq_xid, 0, PTL_UNLINK, PTL_INS_AFTER,
 298                           &request->rq_reply_me_h);
 299         if (rc != PTL_OK) {
 300                 CERROR("PtlMEAttach failed: %d\n", rc);
 301                 LBUG();
 302                 EXIT;
 303                 goto cleanup;
 304         }
 305
 306         request->rq_type = PTL_RPC_REQUEST;
 307         request->rq_reply_md.start = repbuf;
 308         request->rq_reply_md.length = request->rq_replen;
 309         request->rq_reply_md.threshold = 1;
 310         request->rq_reply_md.options = PTL_MD_OP_PUT;
 311         request->rq_reply_md.user_ptr = request;
 312         request->rq_reply_md.eventq = rcvd_rep_eq;
 313
 314         rc = PtlMDAttach(request->rq_reply_me_h, request->rq_reply_md,
 315                          PTL_UNLINK, &request->rq_reply_md_h);
 316         if (rc != PTL_OK) {
 317                 CERROR("PtlMDAttach failed: %d\n", rc);
 318                 LBUG();
 319                 EXIT;
 320                 goto cleanup2;
 321         }
 322
 323         CDEBUG(D_NET, "Setup reply buffer: %u bytes, xid %u, portal %u\n",
 324                request->rq_replen, request->rq_xid, request->rq_reply_portal);
 325
 326         return ptl_send_buf(request, peer, request->rq_req_portal);
 327
 328  cleanup2:
 329         PtlMEUnlink(request->rq_reply_me_h);
 330  cleanup:
 331         OBD_FREE(repbuf, request->rq_replen);
 332
 333         return rc;
 334 }
 335
 336 void ptlrpc_link_svc_me(struct ptlrpc_service *service, int i)
 337 {
 338         int rc;
 339         ptl_md_t dummy;
 340         ptl_handle_md_t md_h;
 341
 342         /* Attach the leading ME on which we build the ring */
 343         rc = PtlMEAttach(service->srv_self.peer_ni, service->srv_req_portal,
 344                          service->srv_id, 0, ~0, PTL_RETAIN, PTL_INS_BEFORE,
 345                          &(service->srv_me_h[i]));
 346         if (rc != PTL_OK) {
 347                 CERROR("PtlMEAttach failed: %d\n", rc);
 348                 LBUG();
 349         }
 350
 351         if (service->srv_ref_count[i])
 352                 LBUG();
 353
 354         dummy.start         = service->srv_buf[i];
 355         dummy.length        = service->srv_buf_size;
 356         dummy.max_offset    = service->srv_buf_size;
 357         dummy.threshold     = PTL_MD_THRESH_INF;
 358         dummy.options       = PTL_MD_OP_PUT | PTL_MD_AUTO_UNLINK;
 359         dummy.user_ptr      = service;
 360         dummy.eventq        = service->srv_eq_h;
 361         dummy.max_offset    = service->srv_buf_size;
 362
 363         rc = PtlMDAttach(service->srv_me_h[i], dummy, PTL_UNLINK, &md_h);
 364         if (rc != PTL_OK) {
 365                 /* cleanup */
 366                 CERROR("PtlMDAttach failed: %d\n", rc);
 367                 LBUG();
 368         }
 369 }
 370
 371 /* ptl_handled_rpc() should be called by the sleeping process once
 372  * it finishes processing an event.  This ensures the ref count is
 373  * decremented and that the rpc ring buffer cycles properly.
 374  */
 375 int ptl_handled_rpc(struct ptlrpc_service *service, void *start)
 376 {
 377         int index;
 378
 379         spin_lock(&service->srv_lock);
 380
 381         for (index = 0; index < service->srv_ring_length; index++)
 382                 if (service->srv_buf[index] == start)
 383                         break;
 384
 385         if (index == service->srv_ring_length)
 386                 LBUG();
 387
 388         CDEBUG(D_INFO, "MD index=%d Ref Count=%d\n", index,
 389                service->srv_ref_count[index]);
 390         service->srv_ref_count[index]--;
 391
 392         if (service->srv_ref_count[index] < 0)
 393                 LBUG();
 394
 395         if (service->srv_ref_count[index] == 0 &&
 396             service->srv_me_h[index] == 0)  {
 397                 CDEBUG(D_NET, "relinking %d\n", index);
 398                 ptlrpc_link_svc_me(service, index);
 399         }
 400
 401         spin_unlock(&service->srv_lock);
 402         return 0;
 403 }