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 ptlrpc_service *svc, struct ptlrpc_request *req)
 213 {
 214         /* FIXME: we need to increment the count of handled events */
 215         req->rq_type = PTL_RPC_REPLY;
 216         req->rq_repmsg->xid = HTON__u32(req->rq_reqmsg->xid);
 217         req->rq_repmsg->status = HTON__u32(req->rq_status);
 218         req->rq_reqmsg->type = HTON__u32(req->rq_type);
 219         return ptl_send_buf(req, &req->rq_peer, svc->srv_rep_portal);
 220 }
 221
 222 int ptlrpc_error(struct ptlrpc_service *svc, struct ptlrpc_request *req)
 223 {
 224         int rc;
 225         ENTRY;
 226
 227         if (req->rq_repbuf) {
 228                 CERROR("req has repbuf\n");
 229                 LBUG();
 230         }
 231
 232         rc = lustre_pack_msg(0, NULL, NULL, &req->rq_replen, &req->rq_repbuf);
 233         req->rq_repmsg = (struct lustre_msg *)req->rq_repbuf;
 234         if (rc)
 235                 RETURN(rc);
 236
 237         req->rq_repmsg->type = HTON__u32(PTL_RPC_ERR);
 238
 239         rc = ptlrpc_reply(svc, req);
 240         RETURN(rc);
 241 }
 242
 243 int ptl_send_rpc(struct ptlrpc_request *request, struct ptlrpc_client *cl)
 244 {
 245         ptl_process_id_t local_id;
 246         int rc;
 247         char *repbuf;
 248
 249         ENTRY;
 250
 251         if (NTOH__u32(request->rq_reqmsg->type) != PTL_RPC_REQUEST) {
 252                 CERROR("wrong packet type sent %d\n",
 253                        NTOH__u32(request->rq_reqmsg->type));
 254                 LBUG();
 255                 RETURN(-EINVAL);
 256         }
 257         if (request->rq_replen == 0) {
 258                 CERROR("request->rq_replen is 0!\n");
 259                 RETURN(-EINVAL);
 260         }
 261
 262         /* request->rq_repbuf is set only when the reply comes in, in
 263          * client_packet_callback() */
 264         OBD_ALLOC(repbuf, request->rq_replen);
 265         if (!repbuf)
 266                 RETURN(-ENOMEM);
 267
 268         local_id.nid = PTL_ID_ANY;
 269         local_id.pid = PTL_ID_ANY;
 270
 271         down(&cl->cli_rpc_sem);
 272
 273         rc = PtlMEAttach(request->rq_peer.peer_ni, request->rq_reply_portal,
 274                          local_id, request->rq_xid, 0, PTL_UNLINK,
 275                          PTL_INS_AFTER, &request->rq_reply_me_h);
 276         if (rc != PTL_OK) {
 277                 CERROR("PtlMEAttach failed: %d\n", rc);
 278                 LBUG();
 279                 GOTO(cleanup, rc);
 280         }
 281
 282         request->rq_type = PTL_RPC_REQUEST;
 283         request->rq_reply_md.start = repbuf;
 284         request->rq_reply_md.length = request->rq_replen;
 285         request->rq_reply_md.threshold = 1;
 286         request->rq_reply_md.options = PTL_MD_OP_PUT;
 287         request->rq_reply_md.user_ptr = request;
 288         request->rq_reply_md.eventq = reply_in_eq;
 289
 290         rc = PtlMDAttach(request->rq_reply_me_h, request->rq_reply_md,
 291                          PTL_UNLINK, &request->rq_reply_md_h);
 292         if (rc != PTL_OK) {
 293                 CERROR("PtlMDAttach failed: %d\n", rc);
 294                 LBUG();
 295                 GOTO(cleanup2, rc);
 296         }
 297
 298         CDEBUG(D_NET, "Setup reply buffer: %u bytes, xid %u, portal %u\n",
 299                request->rq_replen, request->rq_xid, request->rq_reply_portal);
 300
 301         list_add(&request->rq_list, &cl->cli_sending_head);
 302         rc = ptl_send_buf(request, &request->rq_peer, request->rq_req_portal);
 303         RETURN(rc);
 304
 305  cleanup2:
 306         PtlMEUnlink(request->rq_reply_me_h);
 307  cleanup:
 308         OBD_FREE(repbuf, request->rq_replen);
 309         up(&cl->cli_rpc_sem);
 310
 311         return rc;
 312 }
 313
 314 void ptlrpc_link_svc_me(struct ptlrpc_service *service, int i)
 315 {
 316         int rc;
 317         ptl_md_t dummy;
 318         ptl_handle_md_t md_h;
 319
 320         /* Attach the leading ME on which we build the ring */
 321         rc = PtlMEAttach(service->srv_self.peer_ni, service->srv_req_portal,
 322                          service->srv_id, 0, ~0, PTL_RETAIN, PTL_INS_BEFORE,
 323                          &(service->srv_me_h[i]));
 324         if (rc != PTL_OK) {
 325                 CERROR("PtlMEAttach failed: %d\n", rc);
 326                 LBUG();
 327         }
 328
 329         if (service->srv_ref_count[i])
 330                 LBUG();
 331
 332         dummy.start         = service->srv_buf[i];
 333         dummy.length        = service->srv_buf_size;
 334         dummy.max_offset    = service->srv_buf_size;
 335         dummy.threshold     = PTL_MD_THRESH_INF;
 336         dummy.options       = PTL_MD_OP_PUT | PTL_MD_AUTO_UNLINK;
 337         dummy.user_ptr      = service;
 338         dummy.eventq        = service->srv_eq_h;
 339         dummy.max_offset    = service->srv_buf_size;
 340
 341         rc = PtlMDAttach(service->srv_me_h[i], dummy, PTL_UNLINK, &md_h);
 342         if (rc != PTL_OK) {
 343                 /* cleanup */
 344                 CERROR("PtlMDAttach failed: %d\n", rc);
 345                 LBUG();
 346         }
 347 }
 348
 349 /* ptl_handled_rpc() should be called by the sleeping process once
 350  * it finishes processing an event.  This ensures the ref count is
 351  * decremented and that the rpc ring buffer cycles properly.
 352  */
 353 int ptl_handled_rpc(struct ptlrpc_service *service, void *start)
 354 {
 355         int index;
 356
 357         spin_lock(&service->srv_lock);
 358
 359         for (index = 0; index < service->srv_ring_length; index++)
 360                 if (service->srv_buf[index] == start)
 361                         break;
 362
 363         if (index == service->srv_ring_length)
 364                 LBUG();
 365
 366         CDEBUG(D_INFO, "MD index=%d Ref Count=%d\n", index,
 367                service->srv_ref_count[index]);
 368         service->srv_ref_count[index]--;
 369
 370         if (service->srv_ref_count[index] < 0)
 371                 LBUG();
 372
 373         if (service->srv_ref_count[index] == 0 &&
 374             !ptl_is_valid_handle(&(service->srv_me_h[index]))) {
 375                 CDEBUG(D_NET, "relinking %d\n", index);
 376                 ptlrpc_link_svc_me(service, index);
 377         }
 378
 379         spin_unlock(&service->srv_lock);
 380         return 0;
 381 }