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