lustre/ptlrpc/client.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 #include <linux/list.h>
  29
  30 #define DEBUG_SUBSYSTEM S_RPC
  31
  32 #include <linux/obd_support.h>
  33 #include <linux/obd_class.h>
  34 #include <linux/lustre_net.h>
  35
  36 void ptlrpc_init_client(int dev, int req_portal, int rep_portal,
  37                         struct ptlrpc_client *cl)
  38 {
  39         memset(cl, 0, sizeof(*cl));
  40         spin_lock_init(&cl->cli_lock);
  41         cl->cli_xid = 1;
  42         cl->cli_generation = 1;
  43         cl->cli_epoch = 1;
  44         cl->cli_bootcount = 0;
  45         cl->cli_obd = NULL;
  46         cl->cli_request_portal = req_portal;
  47         cl->cli_reply_portal = rep_portal;
  48         INIT_LIST_HEAD(&cl->cli_sending_head);
  49         INIT_LIST_HEAD(&cl->cli_sent_head);
  50         sema_init(&cl->cli_rpc_sem, 32);
  51 }
  52
  53 int ptlrpc_connect_client(char *uuid, struct ptlrpc_client *cl,
  54                           struct lustre_peer *peer)
  55 {
  56         int err;
  57
  58         cl->cli_epoch++;
  59         err = kportal_uuid_to_peer(uuid, peer);
  60         if (err != 0)
  61                 CERROR("cannot find peer %s!\n", uuid);
  62
  63         return err;
  64 }
  65
  66 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk(struct lustre_peer *peer)
  67 {
  68         struct ptlrpc_bulk_desc *bulk;
  69
  70         OBD_ALLOC(bulk, sizeof(*bulk));
  71         if (bulk != NULL) {
  72                 memcpy(&bulk->b_peer, peer, sizeof(*peer));
  73                 init_waitqueue_head(&bulk->b_waitq);
  74         }
  75
  76         return bulk;
  77 }
  78
  79 struct ptlrpc_request *ptlrpc_prep_req(struct ptlrpc_client *cl,
  80                                        struct lustre_peer *peer, int opcode,
  81                                        int count, int *lengths, char **bufs)
  82 {
  83         struct ptlrpc_request *request;
  84         int rc;
  85         ENTRY;
  86
  87         OBD_ALLOC(request, sizeof(*request));
  88         if (!request) {
  89                 CERROR("request allocation out of memory\n");
  90                 RETURN(NULL);
  91         }
  92
  93         spin_lock(&cl->cli_lock);
  94         request->rq_xid = cl->cli_xid++;
  95         spin_unlock(&cl->cli_lock);
  96
  97         rc = lustre_pack_msg(count, lengths, bufs,
  98                              &request->rq_reqlen, &request->rq_reqbuf);
  99         if (rc) {
 100                 CERROR("cannot pack request %d\n", rc);
 101                 RETURN(NULL);
 102         }
 103         request->rq_type = PTL_RPC_REQUEST;
 104         memcpy(&request->rq_peer, peer, sizeof(*peer));
 105         request->rq_reqmsg = (struct lustre_msg *)request->rq_reqbuf;
 106         request->rq_reqmsg->opc = HTON__u32(opcode);
 107         request->rq_reqmsg->xid = HTON__u32(request->rq_xid);
 108         request->rq_reqmsg->type = HTON__u32(request->rq_type);
 109
 110         RETURN(request);
 111 }
 112
 113 void ptlrpc_free_req(struct ptlrpc_request *request)
 114 {
 115         if (request == NULL)
 116                 return;
 117
 118         if (request->rq_repbuf != NULL)
 119                 OBD_FREE(request->rq_repbuf, request->rq_replen);
 120         OBD_FREE(request, sizeof(*request));
 121 }
 122
 123 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 124 {
 125         int rc = 0;
 126
 127         if (req->rq_repbuf != NULL) {
 128                 req->rq_flags = PTL_RPC_REPLY;
 129                 GOTO(out, rc = 1);
 130         }
 131
 132         if (sigismember(&(current->pending.signal), SIGKILL) ||
 133             sigismember(&(current->pending.signal), SIGTERM) ||
 134             sigismember(&(current->pending.signal), SIGINT)) {
 135                 req->rq_flags = PTL_RPC_INTR;
 136                 GOTO(out, rc = 1);
 137         }
 138
 139  out:
 140         return rc;
 141 }
 142
 143 int ptlrpc_check_status(struct ptlrpc_request *req, int err)
 144 {
 145         ENTRY;
 146
 147         if (err != 0) {
 148                 CERROR("err is %d\n", err);
 149                 RETURN(err);
 150         }
 151
 152         if (req == NULL) {
 153                 CERROR("req == NULL\n");
 154                 RETURN(-ENOMEM);
 155         }
 156
 157         if (req->rq_repmsg == NULL) {
 158                 CERROR("req->rq_repmsg == NULL\n");
 159                 RETURN(-ENOMEM);
 160         }
 161
 162         if (req->rq_repmsg->status != 0) {
 163                 CERROR("req->rq_repmsg->status is %d\n",
 164                        req->rq_repmsg->status);
 165                 /* XXX: translate this error from net to host */
 166                 RETURN(req->rq_repmsg->status);
 167         }
 168
 169         RETURN(0);
 170 }
 171
 172 static void ptlrpc_cleanup_request_buf(struct ptlrpc_request *request)
 173 {
 174         OBD_FREE(request->rq_reqbuf, request->rq_reqlen);
 175         request->rq_reqbuf = NULL;
 176         request->rq_reqlen = 0;
 177 }
 178
 179 /* Abort this request and cleanup any resources associated with it. */
 180 static int ptlrpc_abort(struct ptlrpc_request *request)
 181 {
 182         /* First remove the ME for the reply; in theory, this means
 183          * that we can tear down the buffer safely. */
 184         PtlMEUnlink(request->rq_reply_me_h);
 185         OBD_FREE(request->rq_reply_md.start, request->rq_replen);
 186         request->rq_repbuf = NULL;
 187         request->rq_replen = 0;
 188         return 0;
 189 }
 190
 191 int ptlrpc_queue_wait(struct ptlrpc_client *cl, struct ptlrpc_request *req)
 192 {
 193         int rc = 0;
 194         ENTRY;
 195
 196         init_waitqueue_head(&req->rq_wait_for_rep);
 197
 198         req->rq_client = cl;
 199         req->rq_req_portal = cl->cli_request_portal;
 200         req->rq_reply_portal = cl->cli_reply_portal;
 201         rc = ptl_send_rpc(req, cl);
 202         if (rc) {
 203                 CERROR("error %d, opcode %d\n", rc, req->rq_reqmsg->opc);
 204                 ptlrpc_cleanup_request_buf(req);
 205                 up(&cl->cli_rpc_sem);
 206                 RETURN(-rc);
 207         }
 208
 209         CDEBUG(D_OTHER, "-- sleeping\n");
 210         wait_event_interruptible(req->rq_wait_for_rep, ptlrpc_check_reply(req));
 211         CDEBUG(D_OTHER, "-- done\n");
 212         ptlrpc_cleanup_request_buf(req);
 213         up(&cl->cli_rpc_sem);
 214         if (req->rq_flags == PTL_RPC_INTR) {
 215                 /* Clean up the dangling reply buffers */
 216                 ptlrpc_abort(req);
 217                 GOTO(out, rc = -EINTR);
 218         }
 219
 220         if (req->rq_flags != PTL_RPC_REPLY) {
 221                 CERROR("Unknown reason for wakeup\n");
 222                 /* XXX Phil - I end up here when I kill obdctl */
 223                 ptlrpc_abort(req);
 224                 GOTO(out, rc = -EINTR);
 225         }
 226
 227         rc = lustre_unpack_msg(req->rq_repbuf, req->rq_replen);
 228         req->rq_repmsg = (struct lustre_msg *)req->rq_repbuf;
 229         if (rc) {
 230                 CERROR("unpack_rep failed: %d\n", rc);
 231                 GOTO(out, rc);
 232         }
 233         CDEBUG(D_NET, "got rep %d\n", req->rq_repmsg->xid);
 234
 235         if (req->rq_repmsg->status == 0)
 236                 CDEBUG(D_NET, "--> buf %p len %d status %d\n", req->rq_repbuf,
 237                        req->rq_replen, req->rq_repmsg->status);
 238
 239         EXIT;
 240  out:
 241         return rc;
 242 }