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