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  cleanup2:
 195         PtlMEUnlink(bulk->b_me_h);
 196  cleanup1:
 197         PtlMDUnlink(bulk->b_md_h);
 198         return rc;
 199 }
 200
 201 int ptlrpc_reply(struct obd_device *obddev, struct ptlrpc_service *svc,
 202                  struct ptlrpc_request *req)
 203 {
 204         struct ptlrpc_request *clnt_req = req->rq_reply_handle;
 205         ENTRY;
 206
 207         if (req->rq_reply_handle == NULL) {
 208                 /* This is a request that came from the network via portals. */
 209
 210                 /* FIXME: we need to increment the count of handled events */
 211                 req->rq_type = PTL_RPC_REPLY;
 212                 req->rq_reqhdr->xid = req->rq_reqhdr->xid;
 213                 ptl_send_buf(req, &req->rq_peer, svc->srv_rep_portal);
 214         } else {
 215                 /* This is a local request that came from another thread. */
 216
 217                 /* move the reply to the client */
 218                 clnt_req->rq_replen = req->rq_replen;
 219                 clnt_req->rq_repbuf = req->rq_repbuf;
 220                 req->rq_repbuf = NULL;
 221                 req->rq_replen = 0;
 222
 223                 /* free the request buffer */
 224                 OBD_FREE(req->rq_reqbuf, req->rq_reqlen);
 225                 req->rq_reqbuf = NULL;
 226
 227                 /* wake up the client */
 228                 wake_up_interruptible(&clnt_req->rq_wait_for_rep);
 229         }
 230
 231         EXIT;
 232         return 0;
 233 }
 234
 235 int ptlrpc_error(struct obd_device *obddev, struct ptlrpc_service *svc,
 236                  struct ptlrpc_request *req)
 237 {
 238         struct ptlrep_hdr *hdr;
 239
 240         ENTRY;
 241
 242         OBD_ALLOC(hdr, sizeof(*hdr));
 243         if (!hdr) {
 244                 EXIT;
 245                 return -ENOMEM;
 246         }
 247
 248         memset(hdr, 0, sizeof(*hdr));
 249
 250         hdr->xid = req->rq_reqhdr->xid;
 251         hdr->status = req->rq_status;
 252         hdr->type = OST_TYPE_ERR;
 253
 254         if (req->rq_repbuf) {
 255                 CERROR("req has repbuf\n");
 256                 BUG();
 257         }
 258
 259         req->rq_repbuf = (char *)hdr;
 260         req->rq_replen = sizeof(*hdr);
 261
 262         EXIT;
 263         return ptlrpc_reply(obddev, svc, req);
 264 }
 265
 266 int ptl_send_rpc(struct ptlrpc_request *request, struct lustre_peer *peer)
 267 {
 268         ptl_process_id_t local_id;
 269         struct ptlreq_hdr *hdr;
 270         int rc;
 271         char *repbuf;
 272
 273         ENTRY;
 274
 275         hdr = (struct ptlreq_hdr *)request->rq_reqbuf;
 276         if (NTOH__u32(hdr->type) != OST_TYPE_REQ) {
 277                 CERROR("lustre_ost: wrong packet type sent %d\n",
 278                        NTOH__u32(hdr->type));
 279                 BUG();
 280         }
 281         if (request->rq_replen == 0) {
 282                 CERROR("request->rq_replen is 0!\n");
 283                 EXIT;
 284                 return -EINVAL;
 285         }
 286
 287         /* request->rq_repbuf is set only when the reply comes in, in
 288          * client_packet_callback() */
 289         OBD_ALLOC(repbuf, request->rq_replen);
 290         if (!repbuf) {
 291                 EXIT;
 292                 return -ENOMEM;
 293         }
 294
 295         local_id.addr_kind = PTL_ADDR_GID;
 296         local_id.gid = PTL_ID_ANY;
 297         local_id.rid = PTL_ID_ANY;
 298
 299         //CERROR("sending req %d\n", request->rq_xid);
 300         rc = PtlMEPrepend(peer->peer_ni, request->rq_reply_portal, local_id,
 301                           request->rq_xid, 0, PTL_UNLINK,
 302                           &request->rq_reply_me_h);
 303         if (rc != PTL_OK) {
 304                 CERROR("PtlMEAttach failed: %d\n", rc);
 305                 BUG();
 306                 EXIT;
 307                 goto cleanup;
 308         }
 309
 310         request->rq_type = PTL_RPC_REQUEST;
 311         request->rq_reply_md.start = repbuf;
 312         request->rq_reply_md.length = request->rq_replen;
 313         request->rq_reply_md.threshold = 1;
 314         request->rq_reply_md.options = PTL_MD_OP_PUT;
 315         request->rq_reply_md.user_ptr = request;
 316         request->rq_reply_md.eventq = rcvd_rep_eq;
 317
 318         rc = PtlMDAttach(request->rq_reply_me_h, request->rq_reply_md,
 319                          PTL_UNLINK, &request->rq_reply_md_h);
 320         CERROR("MDAttach (send RPC): %Lu\n", (__u64)request->rq_reply_md_h);
 321         if (rc != PTL_OK) {
 322                 CERROR("PtlMDAttach failed: %d\n", rc);
 323                 BUG();
 324                 EXIT;
 325                 goto cleanup2;
 326         }
 327
 328         CDEBUG(D_NET, "Setup reply buffer: %u bytes, xid %u, portal %u\n",
 329                request->rq_replen, request->rq_xid, request->rq_reply_portal);
 330
 331         return ptl_send_buf(request, peer, request->rq_req_portal);
 332
 333  cleanup2:
 334         PtlMEUnlink(request->rq_reply_me_h);
 335  cleanup:
 336         OBD_FREE(repbuf, request->rq_replen);
 337
 338         return rc;
 339 }
 340
 341 /* ptl_handled_rpc() should be called by the sleeping process once
 342  * it finishes processing an event.  This ensures the ref count is
 343  * decremented and that the rpc ring buffer cycles properly.
 344  */
 345 int ptl_handled_rpc(struct ptlrpc_service *service, void *start)
 346 {
 347         int rc, index = 0;
 348
 349         spin_lock(&service->srv_lock);
 350
 351         while (index < service->srv_ring_length) {
 352                 if ( service->srv_md[index].start == start)
 353                         break;
 354                 index++;
 355         }
 356         if (index == service->srv_ring_length)
 357                 BUG();
 358
 359         CDEBUG(D_INFO, "MD index=%d Ref Count=%d\n", index,
 360                service->srv_ref_count[index]);
 361         service->srv_ref_count[index]--;
 362
 363         if (service->srv_ref_count[index] < 0)
 364                 BUG();
 365
 366         if (service->srv_ref_count[index] == 0 &&
 367             service->srv_me_h[index] == 0) {
 368
 369                 /* Replace the unlinked ME and MD */
 370                 rc = PtlMEInsert(service->srv_me_h[service->srv_me_tail],
 371                                  service->srv_id, 0, ~0, PTL_RETAIN,
 372                                  PTL_INS_AFTER, &(service->srv_me_h[index]));
 373                 if (rc != PTL_OK) {
 374                         CERROR("PtlMEInsert failed: %d\n", rc);
 375                         BUG();
 376                         spin_unlock(&service->srv_lock);
 377                         return rc;
 378                 }
 379                 CDEBUG(D_NET, "Inserting new ME and MD in ring, rc %d\n", rc);
 380
 381                 service->srv_me_tail = index;
 382
 383                 service->srv_md[index].start        = service->srv_buf[index];
 384                 service->srv_md[index].length       = service->srv_buf_size;
 385                 service->srv_md[index].threshold    = PTL_MD_THRESH_INF;
 386                 service->srv_md[index].options      = PTL_MD_OP_PUT;
 387                 service->srv_md[index].user_ptr     = service;
 388                 service->srv_md[index].eventq       = service->srv_eq_h;
 389
 390                 rc = PtlMDAttach(service->srv_me_h[index],
 391                                  service->srv_md[index],
 392                                  PTL_RETAIN, &(service->srv_md_h[index]));
 393                 CERROR("MDAttach (request MDs): %Lu\n",
 394                        (__u64)(service->srv_md_h[index]));
 395
 396                 CDEBUG(D_INFO, "Attach MD in ring, rc %d\n", rc);
 397                 if (rc != PTL_OK) {
 398                         /* XXX cleanup */
 399                         CERROR("PtlMDAttach failed: %d\n", rc);
 400                         BUG();
 401                         spin_unlock(&service->srv_lock);
 402                         return rc;
 403                 }
 404         }
 405
 406         spin_unlock(&service->srv_lock);
 407         return 0;
 408 }