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