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_ADDR_GID, 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                 BUG();
  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                 BUG();
  99                 return rc;
 100         }
 101
 102         remote_id.addr_kind = PTL_ADDR_NID;
 103         remote_id.nid = peer->peer_nid;
 104         remote_id.pid = 0;
 105
 106         CDEBUG(D_NET, "Sending %d bytes to portal %d, xid %d\n",
 107                request->rq_req_md.length, portal, request->rq_xid);
 108
 109         rc = PtlPut(md_h, ack, remote_id, portal, 0, request->rq_xid, 0, 0);
 110         if (rc != PTL_OK) {
 111                 CERROR("PtlPut(%d, %d, %d) failed: %d\n", remote_id.nid,
 112                        portal, request->rq_xid, rc);
 113                 /* FIXME: tear down md */
 114                 BUG();
 115         }
 116
 117         return rc;
 118 }
 119
 120 int ptlrpc_send_bulk(struct ptlrpc_bulk_desc *bulk, int portal)
 121 {
 122         int rc;
 123         ptl_process_id_t remote_id;
 124         ptl_handle_md_t md_h;
 125
 126         bulk->b_md.start = bulk->b_buf;
 127         bulk->b_md.length = bulk->b_buflen;
 128         bulk->b_md.eventq = bulk_source_eq;
 129         bulk->b_md.threshold = 2; /* SENT and ACK events */
 130         bulk->b_md.options = PTL_MD_OP_PUT;
 131         bulk->b_md.user_ptr = bulk;
 132
 133         rc = PtlMDBind(bulk->b_peer.peer_ni, bulk->b_md, &md_h);
 134         if (rc != 0) {
 135                 CERROR("PtlMDBind failed: %d\n", rc);
 136                 BUG();
 137                 return rc;
 138         }
 139
 140         remote_id.addr_kind = PTL_ADDR_NID;
 141         remote_id.nid = bulk->b_peer.peer_nid;
 142         remote_id.pid = 0;
 143
 144         CDEBUG(D_NET, "Sending %d bytes to portal %d, xid %d\n",
 145                bulk->b_md.length, portal, bulk->b_xid);
 146
 147         rc = PtlPut(md_h, PTL_ACK_REQ, remote_id, portal, 0, bulk->b_xid, 0, 0);
 148         if (rc != PTL_OK) {
 149                 CERROR("PtlPut(%d, %d, %d) failed: %d\n", remote_id.nid,
 150                        portal, bulk->b_xid, rc);
 151                 PtlMDUnlink(md_h);
 152                 BUG();
 153         }
 154
 155         return rc;
 156 }
 157
 158 int ptlrpc_register_bulk(struct ptlrpc_bulk_desc *bulk)
 159 {
 160         int rc;
 161
 162         ENTRY;
 163
 164         rc = PtlMEPrepend(bulk->b_peer.peer_ni, bulk->b_portal, local_id,
 165                           bulk->b_xid, 0, PTL_UNLINK, &bulk->b_me_h);
 166         if (rc != PTL_OK) {
 167                 CERROR("PtlMEAttach failed: %d\n", rc);
 168                 BUG();
 169                 EXIT;
 170                 goto cleanup1;
 171         }
 172
 173         bulk->b_md.start = bulk->b_buf;
 174         bulk->b_md.length = bulk->b_buflen;
 175         bulk->b_md.threshold = 1;
 176         bulk->b_md.options = PTL_MD_OP_PUT;
 177         bulk->b_md.user_ptr = bulk;
 178         bulk->b_md.eventq = bulk_sink_eq;
 179
 180         rc = PtlMDAttach(bulk->b_me_h, bulk->b_md, PTL_UNLINK, &bulk->b_md_h);
 181         //CERROR("MDAttach (bulk sink): %Lu\n", (__u64)bulk->b_md_h);
 182         if (rc != PTL_OK) {
 183                 CERROR("PtlMDAttach failed: %d\n", rc);
 184                 BUG();
 185                 EXIT;
 186                 goto cleanup2;
 187         }
 188
 189         CDEBUG(D_NET, "Setup bulk sink buffer: %u bytes, xid %u, portal %u\n",
 190                bulk->b_buflen, bulk->b_xid, bulk->b_portal);
 191         EXIT;
 192         return 0;
 193
 194         // XXX Confirm that this is safe!
 195  cleanup2:
 196         PtlMDUnlink(bulk->b_md_h);
 197  cleanup1:
 198         PtlMEUnlink(bulk->b_me_h);
 199         return rc;
 200 }
 201
 202 int ptlrpc_reply(struct obd_device *obddev, struct ptlrpc_service *svc,
 203                  struct ptlrpc_request *req)
 204 {
 205         struct ptlrpc_request *clnt_req = req->rq_reply_handle;
 206         ENTRY;
 207
 208         if (req->rq_reply_handle == NULL) {
 209                 /* This is a request that came from the network via portals. */
 210
 211                 /* FIXME: we need to increment the count of handled events */
 212                 req->rq_type = PTL_RPC_REPLY;
 213                 req->rq_reqhdr->xid = req->rq_reqhdr->xid;
 214                 ptl_send_buf(req, &req->rq_peer, svc->srv_rep_portal);
 215         } else {
 216                 /* This is a local request that came from another thread. */
 217
 218                 /* move the reply to the client */
 219                 clnt_req->rq_replen = req->rq_replen;
 220                 clnt_req->rq_repbuf = req->rq_repbuf;
 221                 req->rq_repbuf = NULL;
 222                 req->rq_replen = 0;
 223
 224                 /* free the request buffer */
 225                 OBD_FREE(req->rq_reqbuf, req->rq_reqlen);
 226                 req->rq_reqbuf = NULL;
 227
 228                 /* wake up the client */
 229                 wake_up_interruptible(&clnt_req->rq_wait_for_rep);
 230         }
 231
 232         EXIT;
 233         return 0;
 234 }
 235
 236 int ptlrpc_error(struct obd_device *obddev, struct ptlrpc_service *svc,
 237                  struct ptlrpc_request *req)
 238 {
 239         struct ptlrep_hdr *hdr;
 240
 241         ENTRY;
 242
 243         OBD_ALLOC(hdr, sizeof(*hdr));
 244         if (!hdr) {
 245                 EXIT;
 246                 return -ENOMEM;
 247         }
 248
 249         memset(hdr, 0, sizeof(*hdr));
 250
 251         hdr->xid = req->rq_reqhdr->xid;
 252         hdr->status = req->rq_status;
 253         hdr->type = OST_TYPE_ERR;
 254
 255         if (req->rq_repbuf) {
 256                 CERROR("req has repbuf\n");
 257                 BUG();
 258         }
 259
 260         req->rq_repbuf = (char *)hdr;
 261         req->rq_replen = sizeof(*hdr);
 262
 263         EXIT;
 264         return ptlrpc_reply(obddev, svc, req);
 265 }
 266
 267 int ptl_send_rpc(struct ptlrpc_request *request, struct lustre_peer *peer)
 268 {
 269         ptl_process_id_t local_id;
 270         struct ptlreq_hdr *hdr;
 271         int rc;
 272         char *repbuf;
 273
 274         ENTRY;
 275
 276         hdr = (struct ptlreq_hdr *)request->rq_reqbuf;
 277         if (NTOH__u32(hdr->type) != OST_TYPE_REQ) {
 278                 CERROR("lustre_ost: wrong packet type sent %d\n",
 279                        NTOH__u32(hdr->type));
 280                 BUG();
 281         }
 282         if (request->rq_replen == 0) {
 283                 CERROR("request->rq_replen is 0!\n");
 284                 EXIT;
 285                 return -EINVAL;
 286         }
 287
 288         /* request->rq_repbuf is set only when the reply comes in, in
 289          * client_packet_callback() */
 290         OBD_ALLOC(repbuf, request->rq_replen);
 291         if (!repbuf) {
 292                 EXIT;
 293                 return -ENOMEM;
 294         }
 295
 296         local_id.addr_kind = PTL_ADDR_GID;
 297         local_id.gid = PTL_ID_ANY;
 298         local_id.rid = PTL_ID_ANY;
 299
 300         //CERROR("sending req %d\n", request->rq_xid);
 301         rc = PtlMEPrepend(peer->peer_ni, request->rq_reply_portal, local_id,
 302                           request->rq_xid, 0, PTL_UNLINK,
 303                           &request->rq_reply_me_h);
 304         if (rc != PTL_OK) {
 305                 CERROR("PtlMEAttach failed: %d\n", rc);
 306                 BUG();
 307                 EXIT;
 308                 goto cleanup;
 309         }
 310
 311         request->rq_type = PTL_RPC_REQUEST;
 312         request->rq_reply_md.start = repbuf;
 313         request->rq_reply_md.length = request->rq_replen;
 314         request->rq_reply_md.threshold = 1;
 315         request->rq_reply_md.options = PTL_MD_OP_PUT;
 316         request->rq_reply_md.user_ptr = request;
 317         request->rq_reply_md.eventq = rcvd_rep_eq;
 318
 319         rc = PtlMDAttach(request->rq_reply_me_h, request->rq_reply_md,
 320                          PTL_UNLINK, &request->rq_reply_md_h);
 321         //CERROR("MDAttach (send RPC): %Lu\n", (__u64)request->rq_reply_md_h);
 322         if (rc != PTL_OK) {
 323                 CERROR("PtlMDAttach failed: %d\n", rc);
 324                 BUG();
 325                 EXIT;
 326                 goto cleanup2;
 327         }
 328
 329         CDEBUG(D_NET, "Setup reply buffer: %u bytes, xid %u, portal %u\n",
 330                request->rq_replen, request->rq_xid, request->rq_reply_portal);
 331
 332         return ptl_send_buf(request, peer, request->rq_req_portal);
 333
 334  cleanup2:
 335         PtlMEUnlink(request->rq_reply_me_h);
 336  cleanup:
 337         OBD_FREE(repbuf, request->rq_replen);
 338
 339         return rc;
 340 }
 341
 342 /* ptl_handled_rpc() should be called by the sleeping process once
 343  * it finishes processing an event.  This ensures the ref count is
 344  * decremented and that the rpc ring buffer cycles properly.
 345  */
 346 int ptl_handled_rpc(struct ptlrpc_service *service, void *start)
 347 {
 348         int rc, index = 0;
 349
 350         spin_lock(&service->srv_lock);
 351
 352         while (index < service->srv_ring_length) {
 353                 if ( service->srv_md[index].start == start)
 354                         break;
 355                 index++;
 356         }
 357         if (index == service->srv_ring_length)
 358                 BUG();
 359
 360         CDEBUG(D_INFO, "MD index=%d Ref Count=%d\n", index,
 361                service->srv_ref_count[index]);
 362         service->srv_ref_count[index]--;
 363
 364         if (service->srv_ref_count[index] < 0)
 365                 BUG();
 366
 367         if (service->srv_ref_count[index] == 0 &&
 368             service->srv_me_h[index] == 0) {
 369
 370                 /* Replace the unlinked ME and MD */
 371                 rc = PtlMEInsert(service->srv_me_h[service->srv_me_tail],
 372                                  service->srv_id, 0, ~0, PTL_RETAIN,
 373                                  PTL_INS_AFTER, &(service->srv_me_h[index]));
 374                 if (rc != PTL_OK) {
 375                         CERROR("PtlMEInsert failed: %d\n", rc);
 376                         BUG();
 377                         spin_unlock(&service->srv_lock);
 378                         return rc;
 379                 }
 380                 CDEBUG(D_NET, "Inserting new ME and MD in ring, rc %d\n", rc);
 381
 382                 service->srv_me_tail = index;
 383
 384                 service->srv_md[index].start        = service->srv_buf[index];
 385                 service->srv_md[index].length       = service->srv_buf_size;
 386                 service->srv_md[index].threshold    = PTL_MD_THRESH_INF;
 387                 service->srv_md[index].options      = PTL_MD_OP_PUT;
 388                 service->srv_md[index].user_ptr     = service;
 389                 service->srv_md[index].eventq       = service->srv_eq_h;
 390
 391                 rc = PtlMDAttach(service->srv_me_h[index],
 392                                  service->srv_md[index],
 393                                  PTL_RETAIN, &(service->srv_md_h[index]));
 394                 //CERROR("MDAttach (request MDs): %Lu\n",
 395                 //(__u64)(service->srv_md_h[index]));
 396
 397                 CDEBUG(D_INFO, "Attach MD in ring, rc %d\n", rc);
 398                 if (rc != PTL_OK) {
 399                         /* XXX cleanup */
 400                         CERROR("PtlMDAttach failed: %d\n", rc);
 401                         BUG();
 402                         spin_unlock(&service->srv_lock);
 403                         return rc;
 404                 }
 405         }
 406
 407         spin_unlock(&service->srv_lock);
 408         return 0;
 409 }