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_ha.h>
  28
  29 void ptlrpc_init_client(struct recovd_obd *recovd,
  30                         void (*recover)(struct ptlrpc_client *recover),
  31                         int req_portal,
  32                         int rep_portal, struct ptlrpc_client *cl)
  33 {
  34         memset(cl, 0, sizeof(*cl));
  35         cl->cli_recovd = recovd;
  36         cl->cli_recover = recover;
  37         if (recovd)
  38                 recovd_cli_manage(recovd, cl);
  39         cl->cli_obd = NULL;
  40         cl->cli_request_portal = req_portal;
  41         cl->cli_reply_portal = rep_portal;
  42         INIT_LIST_HEAD(&cl->cli_sending_head);
  43         INIT_LIST_HEAD(&cl->cli_sent_head);
  44         INIT_LIST_HEAD(&cl->cli_replied_head);
  45         INIT_LIST_HEAD(&cl->cli_replay_head);
  46         spin_lock_init(&cl->cli_lock);
  47         sema_init(&cl->cli_rpc_sem, 32);
  48 }
  49
  50 __u8 *ptlrpc_req_to_uuid(struct ptlrpc_request *req)
  51 {
  52         return req->rq_connection->c_remote_uuid;
  53 }
  54
  55 struct ptlrpc_connection *ptlrpc_uuid_to_connection(char *uuid)
  56 {
  57         struct ptlrpc_connection *c;
  58         struct lustre_peer peer;
  59         int err;
  60
  61         err = kportal_uuid_to_peer(uuid, &peer);
  62         if (err != 0) {
  63                 CERROR("cannot find peer %s!\n", uuid);
  64                 return NULL;
  65         }
  66
  67         c = ptlrpc_get_connection(&peer);
  68         if (c)
  69                 c->c_epoch++;
  70
  71         return c;
  72 }
  73
  74 struct ptlrpc_bulk_desc *ptlrpc_prep_bulk(struct ptlrpc_connection *conn)
  75 {
  76         struct ptlrpc_bulk_desc *bulk;
  77
  78         OBD_ALLOC(bulk, sizeof(*bulk));
  79         if (bulk != NULL) {
  80                 bulk->b_connection = ptlrpc_connection_addref(conn);
  81                 init_waitqueue_head(&bulk->b_waitq);
  82         }
  83
  84         return bulk;
  85 }
  86
  87 void ptlrpc_free_bulk(struct ptlrpc_bulk_desc *bulk)
  88 {
  89         ENTRY;
  90         if (bulk == NULL) {
  91                 EXIT;
  92                 return;
  93         }
  94
  95         ptlrpc_put_connection(bulk->b_connection);
  96
  97         OBD_FREE(bulk, sizeof(*bulk));
  98         EXIT;
  99 }
 100
 101 struct ptlrpc_request *ptlrpc_prep_req(struct ptlrpc_client *cl,
 102                                        struct ptlrpc_connection *conn,
 103                                        int opcode, int count, int *lengths,
 104                                        char **bufs)
 105 {
 106         struct ptlrpc_request *request;
 107         int rc;
 108         ENTRY;
 109
 110         OBD_ALLOC(request, sizeof(*request));
 111         if (!request) {
 112                 CERROR("request allocation out of memory\n");
 113                 RETURN(NULL);
 114         }
 115
 116         rc = lustre_pack_msg(count, lengths, bufs,
 117                              &request->rq_reqlen, &request->rq_reqmsg);
 118         if (rc) {
 119                 CERROR("cannot pack request %d\n", rc);
 120                 RETURN(NULL);
 121         }
 122
 123         request->rq_type = PTL_RPC_TYPE_REQUEST;
 124         request->rq_connection = ptlrpc_connection_addref(conn);
 125
 126         request->rq_reqmsg->conn = (__u64)(unsigned long)conn->c_remote_conn;
 127         request->rq_reqmsg->token = conn->c_remote_token;
 128         request->rq_reqmsg->opc = HTON__u32(opcode);
 129         request->rq_reqmsg->type = HTON__u32(PTL_RPC_MSG_REQUEST);
 130         INIT_LIST_HEAD(&request->rq_list);
 131
 132         /* this will be dec()d once in req_finished, once in free_committed */
 133         atomic_set(&request->rq_refcount, 2);
 134
 135         spin_lock(&conn->c_lock);
 136         request->rq_reqmsg->xid = HTON__u32(++conn->c_xid_out);
 137         spin_unlock(&conn->c_lock);
 138
 139         request->rq_client = cl;
 140
 141         RETURN(request);
 142 }
 143
 144 void ptlrpc_req_finished(struct ptlrpc_request *request)
 145 {
 146         if (request == NULL)
 147                 return;
 148
 149         if (request->rq_repmsg != NULL) {
 150                 OBD_FREE(request->rq_repmsg, request->rq_replen);
 151                 request->rq_repmsg = NULL;
 152         }
 153
 154         if (atomic_dec_and_test(&request->rq_refcount))
 155                 ptlrpc_free_req(request);
 156 }
 157
 158 void ptlrpc_free_req(struct ptlrpc_request *request)
 159 {
 160         if (request == NULL)
 161                 return;
 162
 163         if (request->rq_repmsg != NULL)
 164                 OBD_FREE(request->rq_repmsg, request->rq_replen);
 165         if (request->rq_reqmsg != NULL)
 166                 OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 167
 168         if (request->rq_client) {
 169                 spin_lock(&request->rq_client->cli_lock);
 170                 list_del(&request->rq_list);
 171                 spin_unlock(&request->rq_client->cli_lock);
 172         }
 173
 174         ptlrpc_put_connection(request->rq_connection);
 175
 176         OBD_FREE(request, sizeof(*request));
 177 }
 178
 179 static int ptlrpc_check_reply(struct ptlrpc_request *req)
 180 {
 181         int rc = 0;
 182
 183         if (req->rq_repmsg != NULL) {
 184                 req->rq_transno = NTOH__u64(req->rq_repmsg->transno);
 185                 req->rq_flags |= PTL_RPC_FL_REPLY;
 186                 GOTO(out, rc = 1);
 187         }
 188
 189         if (req->rq_flags & PTL_RPC_FL_RESEND) {
 190                 CERROR("-- RESEND --\n");
 191                 req->rq_status = -EAGAIN;
 192                 GOTO(out, rc = 1);
 193         }
 194
 195         if (CURRENT_TIME - req->rq_time >= req->rq_timeout) {
 196                 CERROR("-- REQ TIMEOUT --\n");
 197                 /* clear the timeout */
 198                 req->rq_timeout = 0;
 199                 req->rq_flags |= PTL_RPC_FL_TIMEOUT;
 200                 if (req->rq_client && req->rq_client->cli_recovd)
 201                         recovd_cli_fail(req->rq_client);
 202                 GOTO(out, rc = 0);
 203         }
 204
 205         if (req->rq_timeout) {
 206                 schedule_timeout(req->rq_timeout * HZ);
 207         }
 208
 209         if (sigismember(&(current->pending.signal), SIGKILL) ||
 210             sigismember(&(current->pending.signal), SIGTERM) ||
 211             sigismember(&(current->pending.signal), SIGINT)) {
 212                 req->rq_flags |= PTL_RPC_FL_INTR;
 213                 GOTO(out, rc = 1);
 214         }
 215
 216  out:
 217         return rc;
 218 }
 219
 220 int ptlrpc_check_status(struct ptlrpc_request *req, int err)
 221 {
 222         ENTRY;
 223
 224         if (err != 0) {
 225                 CERROR("err is %d\n", err);
 226                 RETURN(err);
 227         }
 228
 229         if (req == NULL) {
 230                 CERROR("req == NULL\n");
 231                 RETURN(-ENOMEM);
 232         }
 233
 234         if (req->rq_repmsg == NULL) {
 235                 CERROR("req->rq_repmsg == NULL\n");
 236                 RETURN(-ENOMEM);
 237         }
 238
 239         if (req->rq_repmsg->type == NTOH__u32(PTL_RPC_MSG_ERR)) {
 240                 CERROR("req->rq_repmsg->type == PTL_RPC_MSG_ERR\n");
 241                 RETURN(-EINVAL);
 242         }
 243
 244         if (req->rq_repmsg->status != 0) {
 245                 CERROR("req->rq_repmsg->status is %d\n",
 246                        req->rq_repmsg->status);
 247                 /* XXX: translate this error from net to host */
 248                 RETURN(req->rq_repmsg->status);
 249         }
 250
 251         RETURN(0);
 252 }
 253
 254 static void ptlrpc_cleanup_request_buf(struct ptlrpc_request *request)
 255 {
 256         OBD_FREE(request->rq_reqmsg, request->rq_reqlen);
 257         request->rq_reqmsg = NULL;
 258         request->rq_reqlen = 0;
 259 }
 260
 261 /* Abort this request and cleanup any resources associated with it. */
 262 static int ptlrpc_abort(struct ptlrpc_request *request)
 263 {
 264         /* First remove the ME for the reply; in theory, this means
 265          * that we can tear down the buffer safely. */
 266         PtlMEUnlink(request->rq_reply_me_h);
 267         OBD_FREE(request->rq_reply_md.start, request->rq_replen);
 268         request->rq_repmsg = NULL;
 269         request->rq_replen = 0;
 270         return 0;
 271 }
 272
 273 /* caller must lock cli */
 274 void ptlrpc_free_committed(struct ptlrpc_client *cli)
 275 {
 276         struct list_head *tmp, *saved;
 277         struct ptlrpc_request *req;
 278
 279         list_for_each_safe(tmp, saved, &cli->cli_replied_head) {
 280                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 281
 282                 /* not yet committed */
 283                 if (req->rq_transno > cli->cli_last_committed)
 284                         break;
 285
 286                 /* retain for replay if flagged */
 287                 if (req->rq_flags & PTL_RPC_FL_RETAIN) {
 288                         list_del(&req->rq_list);
 289                         list_add(&req->rq_list, &cli->cli_replay_head);
 290                 } else {
 291                         CDEBUG(D_INFO, "Marking request %p as committed ("
 292                                "transno=%Lu, last_committed=%Lu\n", req,
 293                                req->rq_transno, cli->cli_last_committed);
 294                         if (atomic_dec_and_test(&req->rq_refcount))
 295                                 ptlrpc_free_req(req);
 296                 }
 297         }
 298
 299         EXIT;
 300         return;
 301 }
 302
 303 void ptlrpc_cleanup_client(struct ptlrpc_client *cli)
 304 {
 305         struct list_head *tmp, *saved;
 306         struct ptlrpc_request *req;
 307         ENTRY;
 308
 309         spin_lock(&cli->cli_lock);
 310         list_for_each_safe(tmp, saved, &cli->cli_replied_head) {
 311                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 312                 /* We do this to prevent ptlrpc_free_req from taking cli_lock */
 313                 CDEBUG(D_INFO, "Cleaning req %p from replied head.\n", req);
 314                 list_del(&req->rq_list);
 315                 req->rq_client = NULL;
 316                 ptlrpc_free_req(req);
 317         }
 318         list_for_each_safe(tmp, saved, &cli->cli_sent_head) {
 319                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 320                 CDEBUG(D_INFO, "Cleaning req %p from sent head.\n", req);
 321                 list_del(&req->rq_list);
 322                 req->rq_client = NULL;
 323                 ptlrpc_free_req(req);
 324         }
 325         list_for_each_safe(tmp, saved, &cli->cli_replay_head) {
 326                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 327                 CERROR("Request %p is on the replay head at cleanup!\n", req);
 328                 list_del(&req->rq_list);
 329                 req->rq_client = NULL;
 330                 ptlrpc_free_req(req);
 331         }
 332         list_for_each_safe(tmp, saved, &cli->cli_sending_head) {
 333                 req = list_entry(tmp, struct ptlrpc_request, rq_list);
 334                 CDEBUG(D_INFO, "Cleaning req %p from sending head.\n", req);
 335                 list_del(&req->rq_list);
 336                 req->rq_client = NULL;
 337                 ptlrpc_free_req(req);
 338         }
 339         spin_unlock(&cli->cli_lock);
 340         EXIT;
 341         return;
 342 }
 343
 344 int ptlrpc_queue_wait(struct ptlrpc_request *req)
 345 {
 346         int rc = 0;
 347         ENTRY;
 348
 349         init_waitqueue_head(&req->rq_wait_for_rep);
 350  resend:
 351         req->rq_time = CURRENT_TIME;
 352         req->rq_timeout = 3;
 353         rc = ptl_send_rpc(req);
 354         if (rc) {
 355                 CERROR("error %d, opcode %d\n", rc, req->rq_reqmsg->opc);
 356                 ptlrpc_cleanup_request_buf(req);
 357                 up(&req->rq_client->cli_rpc_sem);
 358                 RETURN(-rc);
 359         }
 360
 361         CDEBUG(D_OTHER, "-- sleeping\n");
 362         wait_event_interruptible(req->rq_wait_for_rep, ptlrpc_check_reply(req));
 363         CDEBUG(D_OTHER, "-- done\n");
 364
 365         if (req->rq_flags & PTL_RPC_FL_RESEND) {
 366                 req->rq_flags &= ~PTL_RPC_FL_RESEND;
 367                 goto resend;
 368         }
 369
 370         //ptlrpc_cleanup_request_buf(req);
 371         up(&req->rq_client->cli_rpc_sem);
 372         if (req->rq_flags & PTL_RPC_FL_INTR) {
 373                 /* Clean up the dangling reply buffers */
 374                 ptlrpc_abort(req);
 375                 GOTO(out, rc = -EINTR);
 376         }
 377
 378         if (! (req->rq_flags & PTL_RPC_FL_REPLY)) {
 379                 CERROR("Unknown reason for wakeup\n");
 380                 /* XXX Phil - I end up here when I kill obdctl */
 381                 ptlrpc_abort(req);
 382                 GOTO(out, rc = -EINTR);
 383         }
 384
 385         rc = lustre_unpack_msg(req->rq_repmsg, req->rq_replen);
 386         if (rc) {
 387                 CERROR("unpack_rep failed: %d\n", rc);
 388                 GOTO(out, rc);
 389         }
 390         CDEBUG(D_NET, "got rep %d\n", req->rq_repmsg->xid);
 391         if (req->rq_repmsg->status == 0)
 392                 CDEBUG(D_NET, "--> buf %p len %d status %d\n", req->rq_repmsg,
 393                        req->rq_replen, req->rq_repmsg->status);
 394
 395         spin_lock(&req->rq_client->cli_lock);
 396         /* add to the tail of the replied head */
 397         list_del(&req->rq_list);
 398         list_add(&req->rq_list, req->rq_client->cli_replied_head.prev);
 399
 400         req->rq_client->cli_last_rcvd = req->rq_repmsg->last_rcvd;
 401         req->rq_client->cli_last_committed = req->rq_repmsg->last_committed;
 402         ptlrpc_free_committed(req->rq_client);
 403         spin_unlock(&req->rq_client->cli_lock);
 404
 405         EXIT;
 406  out:
 407         return rc;
 408 }