lustre/ost/ost_handler.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2001, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/ost/ost_handler.c
  37  *
  38  * Author: Peter J. Braam <braam@clusterfs.com>
  39  * Author: Phil Schwan <phil@clusterfs.com>
  40  */
  41
  42 #ifndef EXPORT_SYMTAB
  43 # define EXPORT_SYMTAB
  44 #endif
  45 #define DEBUG_SUBSYSTEM S_OST
  46
  47 #include <linux/module.h>
  48 #include <obd_cksum.h>
  49 #include <obd_ost.h>
  50 #include <lustre_net.h>
  51 #include <lustre_dlm.h>
  52 #include <lustre_export.h>
  53 #include <lustre_debug.h>
  54 #include <linux/init.h>
  55 #include <lprocfs_status.h>
  56 #include <libcfs/list.h>
  57 #include <lustre_quota.h>
  58 #include "ost_internal.h"
  59
  60 static int oss_num_threads;
  61 CFS_MODULE_PARM(oss_num_threads, "i", int, 0444,
  62                 "number of OSS service threads to start");
  63
  64 static int ost_num_threads;
  65 CFS_MODULE_PARM(ost_num_threads, "i", int, 0444,
  66                 "number of OST service threads to start (deprecated)");
  67
  68 static int oss_num_create_threads;
  69 CFS_MODULE_PARM(oss_num_create_threads, "i", int, 0444,
  70                 "number of OSS create threads to start");
  71
  72 /**
  73  * Do not return server-side uid/gid to remote client
  74  */
  75 static void ost_drop_id(struct obd_export *exp, struct obdo *oa)
  76 {
  77         if (exp_connect_rmtclient(exp)) {
  78                 oa->o_uid = -1;
  79                 oa->o_gid = -1;
  80                 oa->o_valid &= ~(OBD_MD_FLUID | OBD_MD_FLGID);
  81         }
  82 }
  83
  84 /**
  85  * Validate oa from client.
  86  * If the request comes from 2.0 clients, currently only RSVD seq and IDIF
  87  * req are valid.
  88  *    a. for single MDS  seq = FID_SEQ_OST_MDT0,
  89  *    b. for CMD, seq = FID_SEQ_OST_MDT0, FID_SEQ_OST_MDT1 - FID_SEQ_OST_MAX
  90  */
  91 static int ost_validate_obdo(struct obd_export *exp, struct obdo *oa,
  92                              struct obd_ioobj *ioobj)
  93 {
  94         if (oa != NULL && !(oa->o_valid & OBD_MD_FLGROUP)) {
  95                 oa->o_seq = FID_SEQ_OST_MDT0;
  96                 if (ioobj)
  97                         ioobj->ioo_seq = FID_SEQ_OST_MDT0;
  98         /* remove fid_seq_is_rsvd() after FID-on-OST allows SEQ > 9 */
  99         } else if (oa == NULL || !(fid_seq_is_rsvd(oa->o_seq) ||
 100                                    fid_seq_is_mdt0(oa->o_seq))) {
 101                 CERROR("%s: client %s sent invalid object "POSTID"\n",
 102                        exp->exp_obd->obd_name, obd_export_nid2str(exp),
 103                        oa ? oa->o_id : -1, oa ? oa->o_seq : -1);
 104                 return -EPROTO;
 105         }
 106         obdo_from_ostid(oa, &oa->o_oi);
 107         if (ioobj)
 108                 ioobj_from_obdo(ioobj, oa);
 109         return 0;
 110 }
 111
 112 void oti_to_request(struct obd_trans_info *oti, struct ptlrpc_request *req)
 113 {
 114         struct oti_req_ack_lock *ack_lock;
 115         int i;
 116
 117         if (oti == NULL)
 118                 return;
 119
 120         if (req->rq_repmsg) {
 121                 __u64 versions[PTLRPC_NUM_VERSIONS] = { 0 };
 122                 lustre_msg_set_transno(req->rq_repmsg, oti->oti_transno);
 123                 versions[0] = oti->oti_pre_version;
 124                 lustre_msg_set_versions(req->rq_repmsg, versions);
 125         }
 126         req->rq_transno = oti->oti_transno;
 127
 128         /* XXX 4 == entries in oti_ack_locks??? */
 129         for (ack_lock = oti->oti_ack_locks, i = 0; i < 4; i++, ack_lock++) {
 130                 if (!ack_lock->mode)
 131                         break;
 132                 /* XXX not even calling target_send_reply in some cases... */
 133                 ptlrpc_save_lock (req, &ack_lock->lock, ack_lock->mode, 0);
 134         }
 135 }
 136
 137 static int ost_destroy(struct obd_export *exp, struct ptlrpc_request *req,
 138                        struct obd_trans_info *oti)
 139 {
 140         struct ost_body *body, *repbody;
 141         struct lustre_capa *capa = NULL;
 142         int rc;
 143         ENTRY;
 144
 145         /* Get the request body */
 146         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 147         if (body == NULL)
 148                 RETURN(-EFAULT);
 149
 150         if (body->oa.o_id == 0)
 151                 RETURN(-EPROTO);
 152
 153         rc = ost_validate_obdo(exp, &body->oa, NULL);
 154         if (rc)
 155                 RETURN(rc);
 156
 157         /* If there's a DLM request, cancel the locks mentioned in it*/
 158         if (req_capsule_field_present(&req->rq_pill, &RMF_DLM_REQ, RCL_CLIENT)) {
 159                 struct ldlm_request *dlm;
 160
 161                 dlm = req_capsule_client_get(&req->rq_pill, &RMF_DLM_REQ);
 162                 if (dlm == NULL)
 163                         RETURN (-EFAULT);
 164                 ldlm_request_cancel(req, dlm, 0);
 165         }
 166
 167         /* If there's a capability, get it */
 168         if (body->oa.o_valid & OBD_MD_FLOSSCAPA) {
 169                 capa = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 170                 if (capa == NULL) {
 171                         CERROR("Missing capability for OST DESTROY");
 172                         RETURN (-EFAULT);
 173                 }
 174         }
 175
 176         /* Prepare the reply */
 177         rc = req_capsule_server_pack(&req->rq_pill);
 178         if (rc)
 179                 RETURN(rc);
 180
 181         /* Get the log cancellation cookie */
 182         if (body->oa.o_valid & OBD_MD_FLCOOKIE)
 183                 oti->oti_logcookies = &body->oa.o_lcookie;
 184
 185         /* Finish the reply */
 186         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 187         memcpy(&repbody->oa, &body->oa, sizeof(body->oa));
 188
 189         /* Do the destroy and set the reply status accordingly  */
 190         req->rq_status = obd_destroy(exp, &repbody->oa, NULL, oti, NULL, capa);
 191         RETURN(0);
 192 }
 193
 194 /**
 195  * Helper function for getting server side [start, start+count] DLM lock
 196  * if asked by client.
 197  */
 198 static int ost_lock_get(struct obd_export *exp, struct obdo *oa,
 199                         __u64 start, __u64 count, struct lustre_handle *lh,
 200                         int mode, int flags)
 201 {
 202         struct ldlm_res_id res_id;
 203         ldlm_policy_data_t policy;
 204         __u64 end = start + count;
 205
 206         ENTRY;
 207
 208         LASSERT(!lustre_handle_is_used(lh));
 209         /* o_id and o_gr are used for localizing resource, if client miss to set
 210          * them, do not trigger ASSERTION. */
 211         if (unlikely((oa->o_valid & (OBD_MD_FLID | OBD_MD_FLGROUP)) !=
 212                      (OBD_MD_FLID | OBD_MD_FLGROUP)))
 213                 RETURN(-EPROTO);
 214
 215         if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
 216             !(oa->o_flags & OBD_FL_SRVLOCK))
 217                 RETURN(0);
 218
 219         osc_build_res_name(oa->o_id, oa->o_seq, &res_id);
 220         CDEBUG(D_INODE, "OST-side extent lock.\n");
 221
 222         policy.l_extent.start = start & CFS_PAGE_MASK;
 223
 224         /* If ->o_blocks is EOF it means "lock till the end of the
 225          * file". Otherwise, it's size of a hole being punched (in bytes) */
 226         if (count == OBD_OBJECT_EOF || end < start)
 227                 policy.l_extent.end = OBD_OBJECT_EOF;
 228         else
 229                 policy.l_extent.end = end | ~CFS_PAGE_MASK;
 230
 231         RETURN(ldlm_cli_enqueue_local(exp->exp_obd->obd_namespace, &res_id,
 232                                       LDLM_EXTENT, &policy, mode, &flags,
 233                                       ldlm_blocking_ast, ldlm_completion_ast,
 234                                       ldlm_glimpse_ast, NULL, 0, NULL, lh));
 235 }
 236
 237 /* Helper function: release lock, if any. */
 238 static void ost_lock_put(struct obd_export *exp,
 239                          struct lustre_handle *lh, int mode)
 240 {
 241         ENTRY;
 242         if (lustre_handle_is_used(lh))
 243                 ldlm_lock_decref(lh, mode);
 244         EXIT;
 245 }
 246
 247 static int ost_getattr(struct obd_export *exp, struct ptlrpc_request *req)
 248 {
 249         struct ost_body *body, *repbody;
 250         struct obd_info *oinfo;
 251         struct lustre_handle lh = { 0 };
 252         struct lustre_capa *capa = NULL;
 253         int rc;
 254         ENTRY;
 255
 256         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 257         if (body == NULL)
 258                 RETURN(-EFAULT);
 259
 260         rc = ost_validate_obdo(exp, &body->oa, NULL);
 261         if (rc)
 262                 RETURN(rc);
 263
 264         if (body->oa.o_valid & OBD_MD_FLOSSCAPA) {
 265                 capa = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 266                 if (capa == NULL) {
 267                         CERROR("Missing capability for OST GETATTR");
 268                         RETURN(-EFAULT);
 269                 }
 270         }
 271
 272         rc = req_capsule_server_pack(&req->rq_pill);
 273         if (rc)
 274                 RETURN(rc);
 275
 276         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 277         repbody->oa = body->oa;
 278
 279         rc = ost_lock_get(exp, &repbody->oa, 0, OBD_OBJECT_EOF, &lh, LCK_PR, 0);
 280         if (rc)
 281                 RETURN(rc);
 282
 283         OBD_ALLOC_PTR(oinfo);
 284         if (!oinfo)
 285                 GOTO(unlock, rc = -ENOMEM);
 286         oinfo->oi_oa = &repbody->oa;
 287         oinfo->oi_capa = capa;
 288
 289         req->rq_status = obd_getattr(exp, oinfo);
 290
 291         OBD_FREE_PTR(oinfo);
 292
 293         ost_drop_id(exp, &repbody->oa);
 294
 295 unlock:
 296         ost_lock_put(exp, &lh, LCK_PR);
 297         RETURN(rc);
 298 }
 299
 300 static int ost_statfs(struct ptlrpc_request *req)
 301 {
 302         struct obd_statfs *osfs;
 303         int rc;
 304         ENTRY;
 305
 306         rc = req_capsule_server_pack(&req->rq_pill);
 307         if (rc)
 308                 RETURN(rc);
 309
 310         osfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
 311
 312         req->rq_status = obd_statfs(req->rq_export->exp_obd, osfs,
 313                                     cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS),
 314                                     0);
 315         if (req->rq_status != 0)
 316                 CERROR("ost: statfs failed: rc %d\n", req->rq_status);
 317
 318         RETURN(0);
 319 }
 320
 321 static int ost_create(struct obd_export *exp, struct ptlrpc_request *req,
 322                       struct obd_trans_info *oti)
 323 {
 324         struct ost_body *body, *repbody;
 325         int rc;
 326         ENTRY;
 327
 328         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 329         if (body == NULL)
 330                 RETURN(-EFAULT);
 331
 332         rc = ost_validate_obdo(req->rq_export, &body->oa, NULL);
 333         if (rc)
 334                 RETURN(rc);
 335
 336         rc = req_capsule_server_pack(&req->rq_pill);
 337         if (rc)
 338                 RETURN(rc);
 339
 340         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 341         repbody->oa = body->oa;
 342         oti->oti_logcookies = &body->oa.o_lcookie;
 343
 344         req->rq_status = obd_create(exp, &repbody->oa, NULL, oti);
 345         //obd_log_cancel(conn, NULL, 1, oti->oti_logcookies, 0);
 346         RETURN(0);
 347 }
 348
 349 static int ost_punch(struct obd_export *exp, struct ptlrpc_request *req,
 350                      struct obd_trans_info *oti)
 351 {
 352         struct ost_body *body, *repbody;
 353         int rc, flags = 0;
 354         struct lustre_handle lh = {0,};
 355         ENTRY;
 356
 357         /* check that we do support OBD_CONNECT_TRUNCLOCK. */
 358         CLASSERT(OST_CONNECT_SUPPORTED & OBD_CONNECT_TRUNCLOCK);
 359
 360         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 361         if (body == NULL)
 362                 RETURN(-EFAULT);
 363
 364         rc = ost_validate_obdo(exp, &body->oa, NULL);
 365         if (rc)
 366                 RETURN(rc);
 367
 368         if ((body->oa.o_valid & (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS)) !=
 369             (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS))
 370                 RETURN(-EPROTO);
 371
 372         rc = req_capsule_server_pack(&req->rq_pill);
 373         if (rc)
 374                 RETURN(rc);
 375
 376         /* standard truncate optimization: if file body is completely
 377          * destroyed, don't send data back to the server. */
 378         if (body->oa.o_size == 0)
 379                 flags |= LDLM_AST_DISCARD_DATA;
 380
 381         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 382         repbody->oa = body->oa;
 383
 384         rc = ost_lock_get(exp, &repbody->oa, repbody->oa.o_size,
 385                           repbody->oa.o_blocks, &lh, LCK_PW, flags);
 386         if (rc == 0) {
 387                 struct obd_info *oinfo;
 388                 struct lustre_capa *capa = NULL;
 389
 390                 if (repbody->oa.o_valid & OBD_MD_FLFLAGS &&
 391                     repbody->oa.o_flags == OBD_FL_SRVLOCK)
 392                         /*
 393                          * If OBD_FL_SRVLOCK is the only bit set in
 394                          * ->o_flags, clear OBD_MD_FLFLAGS to avoid falling
 395                          * through filter_setattr() to filter_iocontrol().
 396                          */
 397                         repbody->oa.o_valid &= ~OBD_MD_FLFLAGS;
 398
 399                 if (repbody->oa.o_valid & OBD_MD_FLOSSCAPA) {
 400                         capa = req_capsule_client_get(&req->rq_pill,
 401                                                       &RMF_CAPA1);
 402                         if (capa == NULL) {
 403                                 CERROR("Missing capability for OST PUNCH");
 404                                 GOTO(unlock, rc = -EFAULT);
 405                         }
 406                 }
 407
 408                 OBD_ALLOC_PTR(oinfo);
 409                 if (!oinfo)
 410                         GOTO(unlock, rc = -ENOMEM);
 411                 oinfo->oi_oa = &repbody->oa;
 412                 oinfo->oi_policy.l_extent.start = oinfo->oi_oa->o_size;
 413                 oinfo->oi_policy.l_extent.end = oinfo->oi_oa->o_blocks;
 414                 oinfo->oi_capa = capa;
 415                 oinfo->oi_flags = OBD_FL_PUNCH;
 416
 417                 req->rq_status = obd_punch(exp, oinfo, oti, NULL);
 418                 OBD_FREE_PTR(oinfo);
 419 unlock:
 420                 ost_lock_put(exp, &lh, LCK_PW);
 421         }
 422
 423         ost_drop_id(exp, &repbody->oa);
 424         RETURN(rc);
 425 }
 426
 427 static int ost_sync(struct obd_export *exp, struct ptlrpc_request *req)
 428 {
 429         struct ost_body *body, *repbody;
 430         struct obd_info *oinfo;
 431         struct lustre_capa *capa = NULL;
 432         int rc;
 433         ENTRY;
 434
 435         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 436         if (body == NULL)
 437                 RETURN(-EFAULT);
 438
 439         rc = ost_validate_obdo(exp, &body->oa, NULL);
 440         if (rc)
 441                 RETURN(rc);
 442
 443         if (body->oa.o_valid & OBD_MD_FLOSSCAPA) {
 444                 capa = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 445                 if (capa == NULL) {
 446                         CERROR("Missing capability for OST SYNC");
 447                         RETURN (-EFAULT);
 448                 }
 449         }
 450
 451         rc = req_capsule_server_pack(&req->rq_pill);
 452         if (rc)
 453                 RETURN(rc);
 454
 455         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 456         repbody->oa = body->oa;
 457
 458         OBD_ALLOC_PTR(oinfo);
 459         if (!oinfo)
 460                 RETURN(-ENOMEM);
 461
 462         oinfo->oi_oa = &repbody->oa;
 463         oinfo->oi_capa = capa;
 464         req->rq_status = obd_sync(exp, oinfo, repbody->oa.o_size,
 465                                   repbody->oa.o_blocks, NULL);
 466         OBD_FREE_PTR(oinfo);
 467
 468         ost_drop_id(exp, &repbody->oa);
 469         RETURN(0);
 470 }
 471
 472 static int ost_setattr(struct obd_export *exp, struct ptlrpc_request *req,
 473                        struct obd_trans_info *oti)
 474 {
 475         struct ost_body *body, *repbody;
 476         struct obd_info *oinfo;
 477         struct lustre_capa *capa = NULL;
 478         int rc;
 479         ENTRY;
 480
 481         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 482         if (body == NULL)
 483                 RETURN(-EFAULT);
 484
 485         rc = ost_validate_obdo(req->rq_export, &body->oa, NULL);
 486         if (rc)
 487                 RETURN(rc);
 488
 489         rc = req_capsule_server_pack(&req->rq_pill);
 490         if (rc)
 491                 RETURN(rc);
 492
 493         if (body->oa.o_valid & OBD_MD_FLOSSCAPA) {
 494                 capa = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 495                 if (capa == NULL) {
 496                         CERROR("Missing capability for OST SETATTR");
 497                         RETURN (-EFAULT);
 498                 }
 499         }
 500
 501         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 502         repbody->oa = body->oa;
 503
 504         OBD_ALLOC_PTR(oinfo);
 505         if (!oinfo)
 506                 RETURN(-ENOMEM);
 507         oinfo->oi_oa = &repbody->oa;
 508         oinfo->oi_capa = capa;
 509
 510         req->rq_status = obd_setattr(exp, oinfo, oti);
 511
 512         OBD_FREE_PTR(oinfo);
 513
 514         ost_drop_id(exp, &repbody->oa);
 515         RETURN(0);
 516 }
 517
 518 static __u32 ost_checksum_bulk(struct ptlrpc_bulk_desc *desc, int opc,
 519                                cksum_type_t cksum_type)
 520 {
 521         __u32 cksum;
 522         int i;
 523
 524         cksum = init_checksum(cksum_type);
 525         for (i = 0; i < desc->bd_iov_count; i++) {
 526                 struct page *page = desc->bd_iov[i].kiov_page;
 527                 int off = desc->bd_iov[i].kiov_offset & ~CFS_PAGE_MASK;
 528                 char *ptr = kmap(page) + off;
 529                 int len = desc->bd_iov[i].kiov_len;
 530
 531                 /* corrupt the data before we compute the checksum, to
 532                  * simulate a client->OST data error */
 533                 if (i == 0 && opc == OST_WRITE &&
 534                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_RECEIVE))
 535                         memcpy(ptr, "bad3", min(4, len));
 536                 cksum = compute_checksum(cksum, ptr, len, cksum_type);
 537                 /* corrupt the data after we compute the checksum, to
 538                  * simulate an OST->client data error */
 539                 if (i == 0 && opc == OST_READ &&
 540                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_SEND)) {
 541                         memcpy(ptr, "bad4", min(4, len));
 542                         /* nobody should use corrupted page again */
 543                         ClearPageUptodate(page);
 544                 }
 545                 kunmap(page);
 546         }
 547
 548         return fini_checksum(cksum, cksum_type);
 549 }
 550
 551 static int ost_brw_lock_get(int mode, struct obd_export *exp,
 552                             struct obd_ioobj *obj, struct niobuf_remote *nb,
 553                             struct lustre_handle *lh)
 554 {
 555         int flags                 = 0;
 556         int nrbufs                = obj->ioo_bufcnt;
 557         struct ldlm_res_id res_id;
 558         ldlm_policy_data_t policy;
 559         int i;
 560         ENTRY;
 561
 562         osc_build_res_name(obj->ioo_id, obj->ioo_seq, &res_id);
 563         LASSERT(mode == LCK_PR || mode == LCK_PW);
 564         LASSERT(!lustre_handle_is_used(lh));
 565
 566         if (nrbufs == 0 || !(nb[0].flags & OBD_BRW_SRVLOCK))
 567                 RETURN(0);
 568
 569         for (i = 1; i < nrbufs; i ++)
 570                 if ((nb[0].flags & OBD_BRW_SRVLOCK) !=
 571                     (nb[i].flags & OBD_BRW_SRVLOCK))
 572                         RETURN(-EFAULT);
 573
 574         policy.l_extent.start = nb[0].offset & CFS_PAGE_MASK;
 575         policy.l_extent.end   = (nb[nrbufs - 1].offset +
 576                                  nb[nrbufs - 1].len - 1) | ~CFS_PAGE_MASK;
 577
 578         RETURN(ldlm_cli_enqueue_local(exp->exp_obd->obd_namespace, &res_id,
 579                                       LDLM_EXTENT, &policy, mode, &flags,
 580                                       ldlm_blocking_ast, ldlm_completion_ast,
 581                                       ldlm_glimpse_ast, NULL, 0, NULL, lh));
 582 }
 583
 584 static void ost_brw_lock_put(int mode,
 585                              struct obd_ioobj *obj, struct niobuf_remote *niob,
 586                              struct lustre_handle *lh)
 587 {
 588         ENTRY;
 589         LASSERT(mode == LCK_PR || mode == LCK_PW);
 590         LASSERT((obj->ioo_bufcnt > 0 && (niob[0].flags & OBD_BRW_SRVLOCK)) ==
 591                 lustre_handle_is_used(lh));
 592         if (lustre_handle_is_used(lh))
 593                 ldlm_lock_decref(lh, mode);
 594         EXIT;
 595 }
 596
 597 /* Allocate thread local buffers if needed */
 598 static struct ost_thread_local_cache *ost_tls_get(struct ptlrpc_request *r)
 599 {
 600         struct ost_thread_local_cache *tls =
 601                 (struct ost_thread_local_cache *)(r->rq_svc_thread->t_data);
 602
 603         /* In normal mode of operation an I/O request is serviced only
 604          * by ll_ost_io threads each of them has own tls buffers allocated by
 605          * ost_thread_init().
 606          * During recovery, an I/O request may be queued until any of the ost
 607          * service threads process it. Not necessary it should be one of
 608          * ll_ost_io threads. In that case we dynamically allocating tls
 609          * buffers for the request service time. */
 610         if (unlikely(tls == NULL)) {
 611                 LASSERT(r->rq_export->exp_in_recovery);
 612                 OBD_ALLOC_PTR(tls);
 613                 if (tls != NULL) {
 614                         tls->temporary = 1;
 615                         r->rq_svc_thread->t_data = tls;
 616                 }
 617         }
 618         return  tls;
 619 }
 620
 621 /* Free thread local buffers if they were allocated only for servicing
 622  * this one request */
 623 static void ost_tls_put(struct ptlrpc_request *r)
 624 {
 625         struct ost_thread_local_cache *tls =
 626                 (struct ost_thread_local_cache *)(r->rq_svc_thread->t_data);
 627
 628         if (unlikely(tls->temporary)) {
 629                 OBD_FREE_PTR(tls);
 630                 r->rq_svc_thread->t_data = NULL;
 631         }
 632 }
 633
 634 static int ost_brw_read(struct ptlrpc_request *req, struct obd_trans_info *oti)
 635 {
 636         struct ptlrpc_bulk_desc *desc = NULL;
 637         struct obd_export *exp = req->rq_export;
 638         struct niobuf_remote *remote_nb;
 639         struct niobuf_local *local_nb;
 640         struct obd_ioobj *ioo;
 641         struct ost_body *body, *repbody;
 642         struct lustre_capa *capa = NULL;
 643         struct l_wait_info lwi;
 644         struct lustre_handle lockh = { 0 };
 645         int niocount, npages, nob = 0, rc, i;
 646         int no_reply = 0;
 647         struct ost_thread_local_cache *tls;
 648         ENTRY;
 649
 650         req->rq_bulk_read = 1;
 651
 652         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_READ_BULK))
 653                 GOTO(out, rc = -EIO);
 654
 655         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, (obd_timeout + 1) / 4);
 656
 657         /* Check if there is eviction in progress, and if so, wait for it to
 658          * finish */
 659         if (unlikely(cfs_atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
 660                 lwi = LWI_INTR(NULL, NULL); // We do not care how long it takes
 661                 rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
 662                         !cfs_atomic_read(&exp->exp_obd->obd_evict_inprogress),
 663                         &lwi);
 664         }
 665         if (exp->exp_failed)
 666                 GOTO(out, rc = -ENOTCONN);
 667
 668         /* ost_body, ioobj & noibuf_remote are verified and swabbed in
 669          * ost_rw_hpreq_check(). */
 670         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 671         if (body == NULL)
 672                 GOTO(out, rc = -EFAULT);
 673
 674         /*
 675          * A req_capsule_X_get_array(pill, field, ptr_to_element_count) function
 676          * would be useful here and wherever we get &RMF_OBD_IOOBJ and
 677          * &RMF_NIOBUF_REMOTE.
 678          */
 679         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
 680         if (ioo == NULL)
 681                 GOTO(out, rc = -EFAULT);
 682
 683         rc = ost_validate_obdo(exp, &body->oa, ioo);
 684         if (rc)
 685                 RETURN(rc);
 686
 687         niocount = ioo->ioo_bufcnt;
 688         remote_nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
 689         if (remote_nb == NULL)
 690                 GOTO(out, rc = -EFAULT);
 691
 692         if (body->oa.o_valid & OBD_MD_FLOSSCAPA) {
 693                 capa = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 694                 if (capa == NULL) {
 695                         CERROR("Missing capability for OST BRW READ");
 696                         GOTO(out, rc = -EFAULT);
 697                 }
 698         }
 699
 700         rc = req_capsule_server_pack(&req->rq_pill);
 701         if (rc)
 702                 GOTO(out, rc);
 703
 704         tls = ost_tls_get(req);
 705         if (tls == NULL)
 706                 GOTO(out_bulk, rc = -ENOMEM);
 707         local_nb = tls->local;
 708
 709         rc = ost_brw_lock_get(LCK_PR, exp, ioo, remote_nb, &lockh);
 710         if (rc != 0)
 711                 GOTO(out_tls, rc);
 712
 713         /*
 714          * If getting the lock took more time than
 715          * client was willing to wait, drop it. b=11330
 716          */
 717         if (cfs_time_current_sec() > req->rq_deadline ||
 718             OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
 719                 no_reply = 1;
 720                 CERROR("Dropping timed-out read from %s because locking"
 721                        "object "LPX64" took %ld seconds (limit was %ld).\n",
 722                        libcfs_id2str(req->rq_peer), ioo->ioo_id,
 723                        cfs_time_current_sec() - req->rq_arrival_time.tv_sec,
 724                        req->rq_deadline - req->rq_arrival_time.tv_sec);
 725                 GOTO(out_lock, rc = -ETIMEDOUT);
 726         }
 727
 728         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 729         memcpy(&repbody->oa, &body->oa, sizeof(repbody->oa));
 730
 731         npages = OST_THREAD_POOL_SIZE;
 732         rc = obd_preprw(OBD_BRW_READ, exp, &repbody->oa, 1, ioo,
 733                         remote_nb, &npages, local_nb, oti, capa);
 734         if (rc != 0)
 735                 GOTO(out_lock, rc);
 736
 737         desc = ptlrpc_prep_bulk_exp(req, npages,
 738                                      BULK_PUT_SOURCE, OST_BULK_PORTAL);
 739         if (desc == NULL)
 740                 GOTO(out_commitrw, rc = -ENOMEM);
 741
 742         nob = 0;
 743         for (i = 0; i < npages; i++) {
 744                 int page_rc = local_nb[i].rc;
 745
 746                 if (page_rc < 0) {              /* error */
 747                         rc = page_rc;
 748                         break;
 749                 }
 750
 751                 nob += page_rc;
 752                 if (page_rc != 0) {             /* some data! */
 753                         LASSERT (local_nb[i].page != NULL);
 754                         ptlrpc_prep_bulk_page(desc, local_nb[i].page,
 755                                               local_nb[i].offset & ~CFS_PAGE_MASK,
 756                                               page_rc);
 757                 }
 758
 759                 if (page_rc != local_nb[i].len) { /* short read */
 760                         /* All subsequent pages should be 0 */
 761                         while(++i < npages)
 762                                 LASSERT(local_nb[i].rc == 0);
 763                         break;
 764                 }
 765         }
 766
 767         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
 768                 cksum_type_t cksum_type =
 769                         cksum_type_unpack(repbody->oa.o_valid & OBD_MD_FLFLAGS ?
 770                                           repbody->oa.o_flags : 0);
 771                 repbody->oa.o_flags = cksum_type_pack(cksum_type);
 772                 repbody->oa.o_valid = OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
 773                 repbody->oa.o_cksum = ost_checksum_bulk(desc, OST_READ,cksum_type);
 774                 CDEBUG(D_PAGE, "checksum at read origin: %x\n",
 775                        repbody->oa.o_cksum);
 776         } else {
 777                 repbody->oa.o_valid = 0;
 778         }
 779         /* We're finishing using body->oa as an input variable */
 780
 781         /* Check if client was evicted while we were doing i/o before touching
 782            network */
 783         if (rc == 0) {
 784                 if (likely(!CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2)))
 785                         rc = target_bulk_io(exp, desc, &lwi);
 786                 no_reply = rc != 0;
 787         }
 788
 789 out_commitrw:
 790         /* Must commit after prep above in all cases */
 791         rc = obd_commitrw(OBD_BRW_READ, exp, &repbody->oa, 1, ioo,
 792                           remote_nb, npages, local_nb, oti, rc);
 793
 794         if (rc == 0)
 795                 ost_drop_id(exp, &repbody->oa);
 796
 797 out_lock:
 798         ost_brw_lock_put(LCK_PR, ioo, remote_nb, &lockh);
 799 out_tls:
 800         ost_tls_put(req);
 801 out_bulk:
 802         if (desc && !CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))
 803                 ptlrpc_free_bulk(desc);
 804 out:
 805         LASSERT(rc <= 0);
 806         if (rc == 0) {
 807                 req->rq_status = nob;
 808                 ptlrpc_lprocfs_brw(req, nob);
 809                 target_committed_to_req(req);
 810                 ptlrpc_reply(req);
 811         } else if (!no_reply) {
 812                 /* Only reply if there was no comms problem with bulk */
 813                 target_committed_to_req(req);
 814                 req->rq_status = rc;
 815                 ptlrpc_error(req);
 816         } else {
 817                 /* reply out callback would free */
 818                 ptlrpc_req_drop_rs(req);
 819                 LCONSOLE_WARN("%s: Bulk IO read error with %s (at %s), "
 820                               "client will retry: rc %d\n",
 821                               exp->exp_obd->obd_name,
 822                               obd_uuid2str(&exp->exp_client_uuid),
 823                               obd_export_nid2str(exp), rc);
 824         }
 825         /* send a bulk after reply to simulate a network delay or reordering
 826          * by a router */
 827         if (unlikely(CFS_FAIL_PRECHECK(OBD_FAIL_PTLRPC_CLIENT_BULK_CB2))) {
 828                 cfs_waitq_t              waitq;
 829                 struct l_wait_info       lwi1;
 830
 831                 CDEBUG(D_INFO, "reorder BULK\n");
 832                 cfs_waitq_init(&waitq);
 833
 834                 lwi1 = LWI_TIMEOUT_INTR(cfs_time_seconds(3), NULL, NULL, NULL);
 835                 l_wait_event(waitq, 0, &lwi1);
 836                 rc = target_bulk_io(exp, desc, &lwi);
 837                 ptlrpc_free_bulk(desc);
 838         }
 839
 840         RETURN(rc);
 841 }
 842
 843 static int ost_brw_write(struct ptlrpc_request *req, struct obd_trans_info *oti)
 844 {
 845         struct ptlrpc_bulk_desc *desc = NULL;
 846         struct obd_export       *exp = req->rq_export;
 847         struct niobuf_remote    *remote_nb;
 848         struct niobuf_local     *local_nb;
 849         struct obd_ioobj        *ioo;
 850         struct ost_body         *body, *repbody;
 851         struct l_wait_info       lwi;
 852         struct lustre_handle     lockh = {0};
 853         struct lustre_capa      *capa = NULL;
 854         __u32                   *rcs;
 855         int objcount, niocount, npages;
 856         int rc, i, j;
 857         obd_count                client_cksum = 0, server_cksum = 0;
 858         cksum_type_t             cksum_type = OBD_CKSUM_CRC32;
 859         int                      no_reply = 0, mmap = 0;
 860         __u32                    o_uid = 0, o_gid = 0;
 861         struct ost_thread_local_cache *tls;
 862         ENTRY;
 863
 864         req->rq_bulk_write = 1;
 865
 866         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_WRITE_BULK))
 867                 GOTO(out, rc = -EIO);
 868         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_WRITE_BULK2))
 869                 GOTO(out, rc = -EFAULT);
 870
 871         /* pause before transaction has been started */
 872         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, (obd_timeout + 1) / 4);
 873
 874         /* ost_body, ioobj & noibuf_remote are verified and swabbed in
 875          * ost_rw_hpreq_check(). */
 876         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 877         if (body == NULL)
 878                 GOTO(out, rc = -EFAULT);
 879
 880         objcount = req_capsule_get_size(&req->rq_pill, &RMF_OBD_IOOBJ,
 881                                         RCL_CLIENT) / sizeof(*ioo);
 882         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
 883         if (ioo == NULL)
 884                 GOTO(out, rc = -EFAULT);
 885
 886         rc = ost_validate_obdo(exp, &body->oa, ioo);
 887         if (rc)
 888                 RETURN(rc);
 889
 890         for (niocount = i = 0; i < objcount; i++)
 891                 niocount += ioo[i].ioo_bufcnt;
 892
 893         /*
 894          * It'd be nice to have a capsule function to indicate how many elements
 895          * there were in a buffer for an RMF that's declared to be an array.
 896          * It's easy enough to compute the number of elements here though.
 897          */
 898         remote_nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
 899         if (remote_nb == NULL || niocount != (req_capsule_get_size(&req->rq_pill,
 900             &RMF_NIOBUF_REMOTE, RCL_CLIENT) / sizeof(*remote_nb)))
 901                 GOTO(out, rc = -EFAULT);
 902
 903         if ((remote_nb[0].flags & OBD_BRW_MEMALLOC) &&
 904             (exp->exp_connection->c_peer.nid == exp->exp_connection->c_self))
 905                 cfs_memory_pressure_set();
 906
 907         if (body->oa.o_valid & OBD_MD_FLOSSCAPA) {
 908                 capa = req_capsule_client_get(&req->rq_pill, &RMF_CAPA1);
 909                 if (capa == NULL) {
 910                         CERROR("Missing capability for OST BRW WRITE");
 911                         GOTO(out, rc = -EFAULT);
 912                 }
 913         }
 914
 915         req_capsule_set_size(&req->rq_pill, &RMF_RCS, RCL_SERVER,
 916                              niocount * sizeof(*rcs));
 917         rc = req_capsule_server_pack(&req->rq_pill);
 918         if (rc != 0)
 919                 GOTO(out, rc);
 920         CFS_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_PACK, cfs_fail_val);
 921         rcs = req_capsule_server_get(&req->rq_pill, &RMF_RCS);
 922
 923         tls = ost_tls_get(req);
 924         if (tls == NULL)
 925                 GOTO(out_bulk, rc = -ENOMEM);
 926         local_nb = tls->local;
 927
 928         rc = ost_brw_lock_get(LCK_PW, exp, ioo, remote_nb, &lockh);
 929         if (rc != 0)
 930                 GOTO(out_tls, rc);
 931
 932         /*
 933          * If getting the lock took more time than
 934          * client was willing to wait, drop it. b=11330
 935          */
 936         if (cfs_time_current_sec() > req->rq_deadline ||
 937             OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
 938                 no_reply = 1;
 939                 CERROR("Dropping timed-out write from %s because locking "
 940                        "object "LPX64" took %ld seconds (limit was %ld).\n",
 941                        libcfs_id2str(req->rq_peer), ioo->ioo_id,
 942                        cfs_time_current_sec() - req->rq_arrival_time.tv_sec,
 943                        req->rq_deadline - req->rq_arrival_time.tv_sec);
 944                 GOTO(out_lock, rc = -ETIMEDOUT);
 945         }
 946
 947         /* obd_preprw clobbers oa->valid, so save what we need */
 948         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
 949                 client_cksum = body->oa.o_cksum;
 950                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
 951                         cksum_type = cksum_type_unpack(body->oa.o_flags);
 952         }
 953         if (body->oa.o_valid & OBD_MD_FLFLAGS && body->oa.o_flags & OBD_FL_MMAP)
 954                 mmap = 1;
 955
 956         /* Because we already sync grant info with client when reconnect,
 957          * grant info will be cleared for resent req, then fed_grant and
 958          * total_grant will not be modified in following preprw_write */
 959         if (lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT | MSG_REPLAY)) {
 960                 DEBUG_REQ(D_CACHE, req, "clear resent/replay req grant info");
 961                 body->oa.o_valid &= ~OBD_MD_FLGRANT;
 962         }
 963
 964         if (exp_connect_rmtclient(exp)) {
 965                 o_uid = body->oa.o_uid;
 966                 o_gid = body->oa.o_gid;
 967         }
 968
 969         repbody = req_capsule_server_get(&req->rq_pill, &RMF_OST_BODY);
 970         memcpy(&repbody->oa, &body->oa, sizeof(repbody->oa));
 971
 972         npages = OST_THREAD_POOL_SIZE;
 973         rc = obd_preprw(OBD_BRW_WRITE, exp, &repbody->oa, objcount,
 974                         ioo, remote_nb, &npages, local_nb, oti, capa);
 975         if (rc != 0)
 976                 GOTO(out_lock, rc);
 977
 978         desc = ptlrpc_prep_bulk_exp(req, npages,
 979                                      BULK_GET_SINK, OST_BULK_PORTAL);
 980         if (desc == NULL)
 981                 GOTO(skip_transfer, rc = -ENOMEM);
 982
 983         /* NB Having prepped, we must commit... */
 984
 985         for (i = 0; i < npages; i++)
 986                 ptlrpc_prep_bulk_page(desc, local_nb[i].page,
 987                                       local_nb[i].offset & ~CFS_PAGE_MASK,
 988                                       local_nb[i].len);
 989
 990         rc = sptlrpc_svc_prep_bulk(req, desc);
 991         if (rc != 0)
 992                 GOTO(out_lock, rc);
 993
 994         rc = target_bulk_io(exp, desc, &lwi);
 995         no_reply = rc != 0;
 996
 997 skip_transfer:
 998         if (client_cksum != 0 && rc == 0) {
 999                 static int cksum_counter;
1000                 repbody->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1001                 repbody->oa.o_flags &= ~OBD_FL_CKSUM_ALL;
1002                 repbody->oa.o_flags |= cksum_type_pack(cksum_type);
1003                 server_cksum = ost_checksum_bulk(desc, OST_WRITE, cksum_type);
1004                 repbody->oa.o_cksum = server_cksum;
1005                 cksum_counter++;
1006                 if (unlikely(client_cksum != server_cksum)) {
1007                         CDEBUG_LIMIT(mmap ? D_INFO : D_ERROR,
1008                                      "client csum %x, server csum %x\n",
1009                                      client_cksum, server_cksum);
1010                         cksum_counter = 0;
1011                 } else if ((cksum_counter & (-cksum_counter)) == cksum_counter){
1012                         CDEBUG(D_INFO, "Checksum %u from %s OK: %x\n",
1013                                cksum_counter, libcfs_id2str(req->rq_peer),
1014                                server_cksum);
1015                 }
1016         }
1017
1018         /* Must commit after prep above in all cases */
1019         rc = obd_commitrw(OBD_BRW_WRITE, exp, &repbody->oa, objcount, ioo,
1020                           remote_nb, npages, local_nb, oti, rc);
1021         if (rc == -ENOTCONN)
1022                 /* quota acquire process has been given up because
1023                  * either the client has been evicted or the client
1024                  * has timed out the request already */
1025                 no_reply = 1;
1026
1027         if (exp_connect_rmtclient(exp)) {
1028                 repbody->oa.o_uid = o_uid;
1029                 repbody->oa.o_gid = o_gid;
1030         }
1031
1032         /*
1033          * Disable sending mtime back to the client. If the client locked the
1034          * whole object, then it has already updated the mtime on its side,
1035          * otherwise it will have to glimpse anyway (see bug 21489, comment 32)
1036          */
1037         repbody->oa.o_valid &= ~(OBD_MD_FLMTIME | OBD_MD_FLATIME);
1038
1039         if (unlikely(client_cksum != server_cksum && rc == 0 && !mmap)) {
1040                 int  new_cksum = ost_checksum_bulk(desc, OST_WRITE, cksum_type);
1041                 char *msg;
1042                 char *via;
1043                 char *router;
1044
1045                 if (new_cksum == server_cksum)
1046                         msg = "changed in transit before arrival at OST";
1047                 else if (new_cksum == client_cksum)
1048                         msg = "initial checksum before message complete";
1049                 else
1050                         msg = "changed in transit AND after initial checksum";
1051
1052                 if (req->rq_peer.nid == desc->bd_sender) {
1053                         via = router = "";
1054                 } else {
1055                         via = " via ";
1056                         router = libcfs_nid2str(desc->bd_sender);
1057                 }
1058
1059                 LCONSOLE_ERROR_MSG(0x168, "%s: BAD WRITE CHECKSUM: %s from "
1060                                    "%s%s%s inode "DFID" object "
1061                                    LPU64"/"LPU64" extent ["LPU64"-"LPU64"]\n",
1062                                    exp->exp_obd->obd_name, msg,
1063                                    libcfs_id2str(req->rq_peer),
1064                                    via, router,
1065                                    body->oa.o_valid & OBD_MD_FLFID ?
1066                                                 body->oa.o_parent_seq : (__u64)0,
1067                                    body->oa.o_valid & OBD_MD_FLFID ?
1068                                                 body->oa.o_parent_oid : 0,
1069                                    body->oa.o_valid & OBD_MD_FLFID ?
1070                                                 body->oa.o_parent_ver : 0,
1071                                    body->oa.o_id,
1072                                    body->oa.o_valid & OBD_MD_FLGROUP ?
1073                                                 body->oa.o_seq : (__u64)0,
1074                                    local_nb[0].offset,
1075                                    local_nb[npages-1].offset +
1076                                    local_nb[npages-1].len - 1 );
1077                 CERROR("client csum %x, original server csum %x, "
1078                        "server csum now %x\n",
1079                        client_cksum, server_cksum, new_cksum);
1080         }
1081
1082         if (rc == 0) {
1083                 int nob = 0;
1084
1085                 /* set per-requested niobuf return codes */
1086                 for (i = j = 0; i < niocount; i++) {
1087                         int len = remote_nb[i].len;
1088
1089                         nob += len;
1090                         rcs[i] = 0;
1091                         do {
1092                                 LASSERT(j < npages);
1093                                 if (local_nb[j].rc < 0)
1094                                         rcs[i] = local_nb[j].rc;
1095                                 len -= local_nb[j].len;
1096                                 j++;
1097                         } while (len > 0);
1098                         LASSERT(len == 0);
1099                 }
1100                 LASSERT(j == npages);
1101                 ptlrpc_lprocfs_brw(req, nob);
1102         }
1103
1104 out_lock:
1105         ost_brw_lock_put(LCK_PW, ioo, remote_nb, &lockh);
1106 out_tls:
1107         ost_tls_put(req);
1108 out_bulk:
1109         if (desc)
1110                 ptlrpc_free_bulk(desc);
1111 out:
1112         if (rc == 0) {
1113                 oti_to_request(oti, req);
1114                 target_committed_to_req(req);
1115                 rc = ptlrpc_reply(req);
1116         } else if (!no_reply) {
1117                 /* Only reply if there was no comms problem with bulk */
1118                 target_committed_to_req(req);
1119                 req->rq_status = rc;
1120                 ptlrpc_error(req);
1121         } else {
1122                 /* reply out callback would free */
1123                 ptlrpc_req_drop_rs(req);
1124                 LCONSOLE_WARN("%s: Bulk IO write error with %s (at %s), "
1125                               "client will retry: rc %d\n",
1126                               exp->exp_obd->obd_name,
1127                               obd_uuid2str(&exp->exp_client_uuid),
1128                               obd_export_nid2str(exp), rc);
1129         }
1130         cfs_memory_pressure_clr();
1131         RETURN(rc);
1132 }
1133
1134 /**
1135  * Implementation of OST_SET_INFO.
1136  *
1137  * OST_SET_INFO is like ioctl(): heavily overloaded.  Specifically, it takes a
1138  * "key" and a value RPC buffers as arguments, with the value's contents
1139  * interpreted according to the key.
1140  *
1141  * Value types that need swabbing have swabbing done explicitly, either here or
1142  * in functions called from here.  This should be corrected: all swabbing should
1143  * be done in the capsule abstraction, as that will then allow us to move
1144  * swabbing exclusively to the client without having to modify server code
1145  * outside the capsule abstraction's implementation itself.  To correct this
1146  * will require minor changes to the capsule abstraction; see the comments for
1147  * req_capsule_extend() in layout.c.
1148  */
1149 static int ost_set_info(struct obd_export *exp, struct ptlrpc_request *req)
1150 {
1151         struct ost_body *body = NULL, *repbody;
1152         char *key, *val = NULL;
1153         int keylen, vallen, rc = 0;
1154         int is_grant_shrink = 0;
1155         ENTRY;
1156
1157         key = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
1158         if (key == NULL) {
1159                 DEBUG_REQ(D_HA, req, "no set_info key");
1160                 RETURN(-EFAULT);
1161         }
1162         keylen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_KEY,
1163                                       RCL_CLIENT);
1164
1165         vallen = req_capsule_get_size(&req->rq_pill, &RMF_SETINFO_VAL,
1166                                       RCL_CLIENT);
1167
1168         if ((is_grant_shrink = KEY_IS(KEY_GRANT_SHRINK)))
1169                 /* In this case the value is actually an RMF_OST_BODY, so we
1170                  * transmutate the type of this PTLRPC */
1171                 req_capsule_extend(&req->rq_pill, &RQF_OST_SET_GRANT_INFO);
1172
1173         rc = req_capsule_server_pack(&req->rq_pill);
1174         if (rc)
1175                 RETURN(rc);
1176
1177         if (vallen) {
1178                 if (is_grant_shrink) {
1179                         body = req_capsule_client_get(&req->rq_pill,
1180                                                       &RMF_OST_BODY);
1181                         if (!body)
1182                                 RETURN(-EFAULT);
1183
1184                         repbody = req_capsule_server_get(&req->rq_pill,
1185                                                          &RMF_OST_BODY);
1186                         memcpy(repbody, body, sizeof(*body));
1187                         val = (char*)repbody;
1188                 } else {
1189                         val = req_capsule_client_get(&req->rq_pill,
1190                                                      &RMF_SETINFO_VAL);
1191                 }
1192         }
1193
1194         if (KEY_IS(KEY_EVICT_BY_NID)) {
1195                 if (val && vallen)
1196                         obd_export_evict_by_nid(exp->exp_obd, val);
1197                 GOTO(out, rc = 0);
1198         } else if (KEY_IS(KEY_MDS_CONN) && ptlrpc_req_need_swab(req)) {
1199                 if (vallen < sizeof(__u32))
1200                         RETURN(-EFAULT);
1201                 __swab32s((__u32 *)val);
1202         }
1203
1204         /* OBD will also check if KEY_IS(KEY_GRANT_SHRINK), and will cast val to
1205          * a struct ost_body * value */
1206         rc = obd_set_info_async(exp, keylen, key, vallen, val, NULL);
1207 out:
1208         lustre_msg_set_status(req->rq_repmsg, 0);
1209         RETURN(rc);
1210 }
1211
1212 static int ost_get_info(struct obd_export *exp, struct ptlrpc_request *req)
1213 {
1214         void *key, *reply;
1215         int keylen, replylen, rc = 0;
1216         struct req_capsule *pill = &req->rq_pill;
1217         ENTRY;
1218
1219         /* this common part for get_info rpc */
1220         key = req_capsule_client_get(pill, &RMF_SETINFO_KEY);
1221         if (key == NULL) {
1222                 DEBUG_REQ(D_HA, req, "no get_info key");
1223                 RETURN(-EFAULT);
1224         }
1225         keylen = req_capsule_get_size(pill, &RMF_SETINFO_KEY, RCL_CLIENT);
1226
1227         if (KEY_IS(KEY_FIEMAP)) {
1228                 struct ll_fiemap_info_key *fm_key = key;
1229                 int rc;
1230
1231                 rc = ost_validate_obdo(exp, &fm_key->oa, NULL);
1232                 if (rc)
1233                         RETURN(rc);
1234         }
1235
1236         rc = obd_get_info(exp, keylen, key, &replylen, NULL, NULL);
1237         if (rc)
1238                 RETURN(rc);
1239
1240         req_capsule_set_size(pill, &RMF_GENERIC_DATA,
1241                              RCL_SERVER, replylen);
1242
1243         rc = req_capsule_server_pack(pill);
1244         if (rc)
1245                 RETURN(rc);
1246
1247         reply = req_capsule_server_get(pill, &RMF_GENERIC_DATA);
1248         if (reply == NULL)
1249                 RETURN(-ENOMEM);
1250
1251         /* call again to fill in the reply buffer */
1252         rc = obd_get_info(exp, keylen, key, &replylen, reply, NULL);
1253
1254         lustre_msg_set_status(req->rq_repmsg, 0);
1255         RETURN(rc);
1256 }
1257
1258 #ifdef HAVE_QUOTA_SUPPORT
1259 static int ost_handle_quotactl(struct ptlrpc_request *req)
1260 {
1261         struct obd_quotactl *oqctl, *repoqc;
1262         int rc;
1263         ENTRY;
1264
1265         oqctl = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1266         if (oqctl == NULL)
1267                 GOTO(out, rc = -EPROTO);
1268
1269         rc = req_capsule_server_pack(&req->rq_pill);
1270         if (rc)
1271                 GOTO(out, rc);
1272
1273         repoqc = req_capsule_server_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1274         req->rq_status = obd_quotactl(req->rq_export, oqctl);
1275         *repoqc = *oqctl;
1276
1277 out:
1278         RETURN(rc);
1279 }
1280
1281 static int ost_handle_quotacheck(struct ptlrpc_request *req)
1282 {
1283         struct obd_quotactl *oqctl;
1284         int rc;
1285         ENTRY;
1286
1287         oqctl = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1288         if (oqctl == NULL)
1289                 RETURN(-EPROTO);
1290
1291         rc = req_capsule_server_pack(&req->rq_pill);
1292         if (rc)
1293                 RETURN(-ENOMEM);
1294
1295         req->rq_status = obd_quotacheck(req->rq_export, oqctl);
1296         RETURN(0);
1297 }
1298
1299 static int ost_handle_quota_adjust_qunit(struct ptlrpc_request *req)
1300 {
1301         struct quota_adjust_qunit *oqaq, *repoqa;
1302         struct lustre_quota_ctxt *qctxt;
1303         int rc;
1304         ENTRY;
1305
1306         qctxt = &req->rq_export->exp_obd->u.obt.obt_qctxt;
1307         oqaq = req_capsule_client_get(&req->rq_pill, &RMF_QUOTA_ADJUST_QUNIT);
1308         if (oqaq == NULL)
1309                 GOTO(out, rc = -EPROTO);
1310
1311         rc = req_capsule_server_pack(&req->rq_pill);
1312         if (rc)
1313                 GOTO(out, rc);
1314
1315         repoqa = req_capsule_server_get(&req->rq_pill, &RMF_QUOTA_ADJUST_QUNIT);
1316         req->rq_status = obd_quota_adjust_qunit(req->rq_export, oqaq, qctxt, NULL);
1317         *repoqa = *oqaq;
1318
1319  out:
1320         RETURN(rc);
1321 }
1322 #endif
1323
1324 static int ost_llog_handle_connect(struct obd_export *exp,
1325                                    struct ptlrpc_request *req)
1326 {
1327         struct llogd_conn_body *body;
1328         int rc;
1329         ENTRY;
1330
1331         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_CONN_BODY);
1332         rc = obd_llog_connect(exp, body);
1333         RETURN(rc);
1334 }
1335
1336 #define ost_init_sec_none(reply, exp)                                   \
1337 do {                                                                    \
1338         reply->ocd_connect_flags &= ~(OBD_CONNECT_RMT_CLIENT |          \
1339                                       OBD_CONNECT_RMT_CLIENT_FORCE |    \
1340                                       OBD_CONNECT_OSS_CAPA);            \
1341         cfs_spin_lock(&exp->exp_lock);                                  \
1342         exp->exp_connect_flags = reply->ocd_connect_flags;              \
1343         cfs_spin_unlock(&exp->exp_lock);                                \
1344 } while (0)
1345
1346 static int ost_init_sec_level(struct ptlrpc_request *req)
1347 {
1348         struct obd_export *exp = req->rq_export;
1349         struct req_capsule *pill = &req->rq_pill;
1350         struct obd_device *obd = exp->exp_obd;
1351         struct filter_obd *filter = &obd->u.filter;
1352         char *client = libcfs_nid2str(req->rq_peer.nid);
1353         struct obd_connect_data *data, *reply;
1354         int rc = 0, remote;
1355         ENTRY;
1356
1357         data = req_capsule_client_get(pill, &RMF_CONNECT_DATA);
1358         reply = req_capsule_server_get(pill, &RMF_CONNECT_DATA);
1359         if (data == NULL || reply == NULL)
1360                 RETURN(-EFAULT);
1361
1362         /* connection from MDT is always trusted */
1363         if (req->rq_auth_usr_mdt) {
1364                 ost_init_sec_none(reply, exp);
1365                 RETURN(0);
1366         }
1367
1368         /* no GSS support case */
1369         if (!req->rq_auth_gss) {
1370                 if (filter->fo_sec_level > LUSTRE_SEC_NONE) {
1371                         CWARN("client %s -> target %s does not user GSS, "
1372                               "can not run under security level %d.\n",
1373                               client, obd->obd_name, filter->fo_sec_level);
1374                         RETURN(-EACCES);
1375                 } else {
1376                         ost_init_sec_none(reply, exp);
1377                         RETURN(0);
1378                 }
1379         }
1380
1381         /* old version case */
1382         if (unlikely(!(data->ocd_connect_flags & OBD_CONNECT_RMT_CLIENT) ||
1383                      !(data->ocd_connect_flags & OBD_CONNECT_OSS_CAPA))) {
1384                 if (filter->fo_sec_level > LUSTRE_SEC_NONE) {
1385                         CWARN("client %s -> target %s uses old version, "
1386                               "can not run under security level %d.\n",
1387                               client, obd->obd_name, filter->fo_sec_level);
1388                         RETURN(-EACCES);
1389                 } else {
1390                         CWARN("client %s -> target %s uses old version, "
1391                               "run under security level %d.\n",
1392                               client, obd->obd_name, filter->fo_sec_level);
1393                         ost_init_sec_none(reply, exp);
1394                         RETURN(0);
1395                 }
1396         }
1397
1398         remote = data->ocd_connect_flags & OBD_CONNECT_RMT_CLIENT_FORCE;
1399         if (remote) {
1400                 if (!req->rq_auth_remote)
1401                         CDEBUG(D_SEC, "client (local realm) %s -> target %s "
1402                                "asked to be remote.\n", client, obd->obd_name);
1403         } else if (req->rq_auth_remote) {
1404                 remote = 1;
1405                 CDEBUG(D_SEC, "client (remote realm) %s -> target %s is set "
1406                        "as remote by default.\n", client, obd->obd_name);
1407         }
1408
1409         if (remote) {
1410                 if (!filter->fo_fl_oss_capa) {
1411                         CDEBUG(D_SEC, "client %s -> target %s is set as remote,"
1412                                " but OSS capabilities are not enabled: %d.\n",
1413                                client, obd->obd_name, filter->fo_fl_oss_capa);
1414                         RETURN(-EACCES);
1415                 }
1416         }
1417
1418         switch (filter->fo_sec_level) {
1419         case LUSTRE_SEC_NONE:
1420                 if (!remote) {
1421                         ost_init_sec_none(reply, exp);
1422                         break;
1423                 } else {
1424                         CDEBUG(D_SEC, "client %s -> target %s is set as remote, "
1425                                "can not run under security level %d.\n",
1426                                client, obd->obd_name, filter->fo_sec_level);
1427                         RETURN(-EACCES);
1428                 }
1429         case LUSTRE_SEC_REMOTE:
1430                 if (!remote)
1431                         ost_init_sec_none(reply, exp);
1432                 break;
1433         case LUSTRE_SEC_ALL:
1434                 if (!remote) {
1435                         reply->ocd_connect_flags &= ~(OBD_CONNECT_RMT_CLIENT |
1436                                                       OBD_CONNECT_RMT_CLIENT_FORCE);
1437                         if (!filter->fo_fl_oss_capa)
1438                                 reply->ocd_connect_flags &= ~OBD_CONNECT_OSS_CAPA;
1439
1440                         cfs_spin_lock(&exp->exp_lock);
1441                         exp->exp_connect_flags = reply->ocd_connect_flags;
1442                         cfs_spin_unlock(&exp->exp_lock);
1443                 }
1444                 break;
1445         default:
1446                 RETURN(-EINVAL);
1447         }
1448
1449         RETURN(rc);
1450 }
1451
1452 /*
1453  * FIXME
1454  * this should be done in filter_connect()/filter_reconnect(), but
1455  * we can't obtain information like NID, which stored in incoming
1456  * request, thus can't decide what flavor to use. so we do it here.
1457  *
1458  * This hack should be removed after the OST stack be rewritten, just
1459  * like what we are doing in mdt_obd_connect()/mdt_obd_reconnect().
1460  */
1461 static int ost_connect_check_sptlrpc(struct ptlrpc_request *req)
1462 {
1463         struct obd_export     *exp = req->rq_export;
1464         struct filter_obd     *filter = &exp->exp_obd->u.filter;
1465         struct sptlrpc_flavor  flvr;
1466         int                    rc = 0;
1467
1468         if (unlikely(strcmp(exp->exp_obd->obd_type->typ_name,
1469                             LUSTRE_ECHO_NAME) == 0)) {
1470                 exp->exp_flvr.sf_rpc = SPTLRPC_FLVR_ANY;
1471                 return 0;
1472         }
1473
1474         if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) {
1475                 cfs_read_lock(&filter->fo_sptlrpc_lock);
1476                 sptlrpc_target_choose_flavor(&filter->fo_sptlrpc_rset,
1477                                              req->rq_sp_from,
1478                                              req->rq_peer.nid,
1479                                              &flvr);
1480                 cfs_read_unlock(&filter->fo_sptlrpc_lock);
1481
1482                 cfs_spin_lock(&exp->exp_lock);
1483
1484                 exp->exp_sp_peer = req->rq_sp_from;
1485                 exp->exp_flvr = flvr;
1486
1487                 if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
1488                     exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
1489                         CERROR("unauthorized rpc flavor %x from %s, "
1490                                "expect %x\n", req->rq_flvr.sf_rpc,
1491                                libcfs_nid2str(req->rq_peer.nid),
1492                                exp->exp_flvr.sf_rpc);
1493                         rc = -EACCES;
1494                 }
1495
1496                 cfs_spin_unlock(&exp->exp_lock);
1497         } else {
1498                 if (exp->exp_sp_peer != req->rq_sp_from) {
1499                         CERROR("RPC source %s doesn't match %s\n",
1500                                sptlrpc_part2name(req->rq_sp_from),
1501                                sptlrpc_part2name(exp->exp_sp_peer));
1502                         rc = -EACCES;
1503                 } else {
1504                         rc = sptlrpc_target_export_check(exp, req);
1505                 }
1506         }
1507
1508         return rc;
1509 }
1510
1511 /* Ensure that data and metadata are synced to the disk when lock is cancelled
1512  * (if requested) */
1513 int ost_blocking_ast(struct ldlm_lock *lock,
1514                              struct ldlm_lock_desc *desc,
1515                              void *data, int flag)
1516 {
1517         __u32 sync_lock_cancel = 0;
1518         __u32 len = sizeof(sync_lock_cancel);
1519         int rc = 0;
1520         ENTRY;
1521
1522         rc = obd_get_info(lock->l_export, sizeof(KEY_SYNC_LOCK_CANCEL),
1523                           KEY_SYNC_LOCK_CANCEL, &len, &sync_lock_cancel, NULL);
1524
1525         if (!rc && flag == LDLM_CB_CANCELING &&
1526             (lock->l_granted_mode & (LCK_PW|LCK_GROUP)) &&
1527             (sync_lock_cancel == ALWAYS_SYNC_ON_CANCEL ||
1528              (sync_lock_cancel == BLOCKING_SYNC_ON_CANCEL &&
1529               lock->l_flags & LDLM_FL_CBPENDING))) {
1530                 struct obd_info *oinfo;
1531                 struct obdo *oa;
1532                 int rc;
1533
1534                 OBD_ALLOC_PTR(oinfo);
1535                 if (!oinfo)
1536                         RETURN(-ENOMEM);
1537                 OBDO_ALLOC(oa);
1538                 if (!oa) {
1539                         OBD_FREE_PTR(oinfo);
1540                         RETURN(-ENOMEM);
1541                 }
1542                 oa->o_id = lock->l_resource->lr_name.name[0];
1543                 oa->o_seq = lock->l_resource->lr_name.name[1];
1544                 oa->o_valid = OBD_MD_FLID|OBD_MD_FLGROUP;
1545                 oinfo->oi_oa = oa;
1546
1547                 rc = obd_sync(lock->l_export, oinfo,
1548                               lock->l_policy_data.l_extent.start,
1549                               lock->l_policy_data.l_extent.end, NULL);
1550                 if (rc)
1551                         CERROR("Error %d syncing data on lock cancel\n", rc);
1552
1553                 OBDO_FREE(oa);
1554                 OBD_FREE_PTR(oinfo);
1555         }
1556
1557         rc = ldlm_server_blocking_ast(lock, desc, data, flag);
1558         RETURN(rc);
1559 }
1560
1561 static int ost_filter_recovery_request(struct ptlrpc_request *req,
1562                                        struct obd_device *obd, int *process)
1563 {
1564         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1565         case OST_CONNECT: /* This will never get here, but for completeness. */
1566         case OST_DISCONNECT:
1567                *process = 1;
1568                RETURN(0);
1569
1570         case OBD_PING:
1571         case OST_CREATE:
1572         case OST_DESTROY:
1573         case OST_PUNCH:
1574         case OST_SETATTR:
1575         case OST_SYNC:
1576         case OST_WRITE:
1577         case OBD_LOG_CANCEL:
1578         case LDLM_ENQUEUE:
1579                 *process = target_queue_recovery_request(req, obd);
1580                 RETURN(0);
1581
1582         default:
1583                 DEBUG_REQ(D_WARNING, req, "not permitted during recovery");
1584                 *process = -EAGAIN;
1585                 RETURN(0);
1586         }
1587 }
1588
1589 int ost_msg_check_version(struct lustre_msg *msg)
1590 {
1591         int rc;
1592
1593         switch(lustre_msg_get_opc(msg)) {
1594         case OST_CONNECT:
1595         case OST_DISCONNECT:
1596         case OBD_PING:
1597         case SEC_CTX_INIT:
1598         case SEC_CTX_INIT_CONT:
1599         case SEC_CTX_FINI:
1600                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
1601                 if (rc)
1602                         CERROR("bad opc %u version %08x, expecting %08x\n",
1603                                lustre_msg_get_opc(msg),
1604                                lustre_msg_get_version(msg),
1605                                LUSTRE_OBD_VERSION);
1606                 break;
1607         case OST_CREATE:
1608         case OST_DESTROY:
1609         case OST_GETATTR:
1610         case OST_SETATTR:
1611         case OST_WRITE:
1612         case OST_READ:
1613         case OST_PUNCH:
1614         case OST_STATFS:
1615         case OST_SYNC:
1616         case OST_SET_INFO:
1617         case OST_GET_INFO:
1618 #ifdef HAVE_QUOTA_SUPPORT
1619         case OST_QUOTACHECK:
1620         case OST_QUOTACTL:
1621         case OST_QUOTA_ADJUST_QUNIT:
1622 #endif
1623                 rc = lustre_msg_check_version(msg, LUSTRE_OST_VERSION);
1624                 if (rc)
1625                         CERROR("bad opc %u version %08x, expecting %08x\n",
1626                                lustre_msg_get_opc(msg),
1627                                lustre_msg_get_version(msg),
1628                                LUSTRE_OST_VERSION);
1629                 break;
1630         case LDLM_ENQUEUE:
1631         case LDLM_CONVERT:
1632         case LDLM_CANCEL:
1633         case LDLM_BL_CALLBACK:
1634         case LDLM_CP_CALLBACK:
1635                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
1636                 if (rc)
1637                         CERROR("bad opc %u version %08x, expecting %08x\n",
1638                                lustre_msg_get_opc(msg),
1639                                lustre_msg_get_version(msg),
1640                                LUSTRE_DLM_VERSION);
1641                 break;
1642         case LLOG_ORIGIN_CONNECT:
1643         case OBD_LOG_CANCEL:
1644                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
1645                 if (rc)
1646                         CERROR("bad opc %u version %08x, expecting %08x\n",
1647                                lustre_msg_get_opc(msg),
1648                                lustre_msg_get_version(msg),
1649                                LUSTRE_LOG_VERSION);
1650                 break;
1651         default:
1652                 CERROR("Unexpected opcode %d\n", lustre_msg_get_opc(msg));
1653                 rc = -ENOTSUPP;
1654         }
1655         return rc;
1656 }
1657
1658 struct ost_prolong_data {
1659         struct ptlrpc_request *opd_req;
1660         struct obd_export     *opd_exp;
1661         struct obdo           *opd_oa;
1662         struct ldlm_res_id     opd_resid;
1663         struct ldlm_extent     opd_extent;
1664         ldlm_mode_t            opd_mode;
1665         unsigned int           opd_locks;
1666         int                    opd_timeout;
1667 };
1668
1669 /* prolong locks for the current service time of the corresponding
1670  * portal (= OST_IO_PORTAL)
1671  */
1672 static inline int prolong_timeout(struct ptlrpc_request *req)
1673 {
1674         struct ptlrpc_service *svc = req->rq_rqbd->rqbd_service;
1675
1676         if (AT_OFF)
1677                 return obd_timeout / 2;
1678
1679         return max(at_est2timeout(at_get(&svc->srv_at_estimate)), ldlm_timeout);
1680 }
1681
1682 static void ost_prolong_lock_one(struct ost_prolong_data *opd,
1683                                  struct ldlm_lock *lock)
1684 {
1685         LASSERT(lock->l_req_mode == lock->l_granted_mode);
1686         LASSERT(lock->l_export == opd->opd_exp);
1687
1688         /* XXX: never try to grab resource lock here because we're inside
1689          * exp_bl_list_lock; in ldlm_lockd.c to handle waiting list we take
1690          * res lock and then exp_bl_list_lock. */
1691
1692         if (!(lock->l_flags & LDLM_FL_AST_SENT))
1693                 /* ignore locks not being cancelled */
1694                 return;
1695
1696         LDLM_DEBUG(lock,
1697                    "refreshed for req x"LPU64" ext("LPU64"->"LPU64") to %ds.\n",
1698                    opd->opd_req->rq_xid, opd->opd_extent.start,
1699                    opd->opd_extent.end, opd->opd_timeout);
1700
1701         /* OK. this is a possible lock the user holds doing I/O
1702          * let's refresh eviction timer for it */
1703         ldlm_refresh_waiting_lock(lock, opd->opd_timeout);
1704         ++opd->opd_locks;
1705 }
1706
1707 static void ost_prolong_locks(struct ost_prolong_data *data)
1708 {
1709         struct obd_export *exp = data->opd_exp;
1710         struct obdo       *oa  = data->opd_oa;
1711         struct ldlm_lock  *lock;
1712         ENTRY;
1713
1714         if (oa->o_valid & OBD_MD_FLHANDLE) {
1715                 /* mostly a request should be covered by only one lock, try
1716                  * fast path. */
1717                 lock = ldlm_handle2lock(&oa->o_handle);
1718                 if (lock != NULL) {
1719                         /* Fast path to check if the lock covers the whole IO
1720                          * region exclusively. */
1721                         if (lock->l_granted_mode == LCK_PW &&
1722                             ldlm_extent_contain(&lock->l_policy_data.l_extent,
1723                                                 &data->opd_extent)) {
1724                                 /* bingo */
1725                                 ost_prolong_lock_one(data, lock);
1726                                 LDLM_LOCK_PUT(lock);
1727                                 RETURN_EXIT;
1728                         }
1729                         LDLM_LOCK_PUT(lock);
1730                 }
1731         }
1732
1733
1734         cfs_spin_lock_bh(&exp->exp_bl_list_lock);
1735         cfs_list_for_each_entry(lock, &exp->exp_bl_list, l_exp_list) {
1736                 LASSERT(lock->l_flags & LDLM_FL_AST_SENT);
1737                 LASSERT(lock->l_resource->lr_type == LDLM_EXTENT);
1738
1739                 if (!ldlm_res_eq(&data->opd_resid, &lock->l_resource->lr_name))
1740                         continue;
1741
1742                 if (!ldlm_extent_overlap(&lock->l_policy_data.l_extent,
1743                                          &data->opd_extent))
1744                         continue;
1745
1746                 ost_prolong_lock_one(data, lock);
1747         }
1748         cfs_spin_unlock_bh(&exp->exp_bl_list_lock);
1749
1750         EXIT;
1751 }
1752
1753 /**
1754  * Returns 1 if the given PTLRPC matches the given LDLM locks, or 0 if it does
1755  * not.
1756  */
1757 static int ost_rw_hpreq_lock_match(struct ptlrpc_request *req,
1758                                    struct ldlm_lock *lock)
1759 {
1760         struct niobuf_remote *nb;
1761         struct obd_ioobj *ioo;
1762         int mode, opc;
1763         struct ldlm_extent ext;
1764         ENTRY;
1765
1766         opc = lustre_msg_get_opc(req->rq_reqmsg);
1767         LASSERT(opc == OST_READ || opc == OST_WRITE);
1768
1769         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
1770         LASSERT(ioo != NULL);
1771
1772         nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
1773         LASSERT(nb != NULL);
1774
1775         ext.start = nb->offset;
1776         nb += ioo->ioo_bufcnt - 1;
1777         ext.end = nb->offset + nb->len - 1;
1778
1779         LASSERT(lock->l_resource != NULL);
1780         if (!osc_res_name_eq(ioo->ioo_id, ioo->ioo_seq,
1781                              &lock->l_resource->lr_name))
1782                 RETURN(0);
1783
1784         mode = LCK_PW;
1785         if (opc == OST_READ)
1786                 mode |= LCK_PR;
1787         if (!(lock->l_granted_mode & mode))
1788                 RETURN(0);
1789
1790         RETURN(ldlm_extent_overlap(&lock->l_policy_data.l_extent, &ext));
1791 }
1792
1793 /**
1794  * High-priority queue request check for whether the given PTLRPC request (\a
1795  * req) is blocking an LDLM lock cancel.
1796  *
1797  * Returns 1 if the given given PTLRPC request (\a req) is blocking an LDLM lock
1798  * cancel, 0 if it is not, and -EFAULT if the request is malformed.
1799  *
1800  * Only OST_READs, OST_WRITEs and OST_PUNCHes go on the h-p RPC queue.  This
1801  * function looks only at OST_READs and OST_WRITEs.
1802  */
1803 static int ost_rw_hpreq_check(struct ptlrpc_request *req)
1804 {
1805         struct obd_device *obd = req->rq_export->exp_obd;
1806         struct ost_body *body;
1807         struct obd_ioobj *ioo;
1808         struct niobuf_remote *nb;
1809         struct ost_prolong_data opd = { 0 };
1810         int mode, opc;
1811         ENTRY;
1812
1813         /*
1814          * Use LASSERT to do sanity check because malformed RPCs should have
1815          * been filtered out in ost_hpreq_handler().
1816          */
1817         opc = lustre_msg_get_opc(req->rq_reqmsg);
1818         LASSERT(opc == OST_READ || opc == OST_WRITE);
1819
1820         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
1821         LASSERT(body != NULL);
1822
1823         ioo = req_capsule_client_get(&req->rq_pill, &RMF_OBD_IOOBJ);
1824         LASSERT(ioo != NULL);
1825
1826         nb = req_capsule_client_get(&req->rq_pill, &RMF_NIOBUF_REMOTE);
1827         LASSERT(nb != NULL);
1828         LASSERT(!(nb->flags & OBD_BRW_SRVLOCK));
1829
1830         osc_build_res_name(ioo->ioo_id, ioo->ioo_seq, &opd.opd_resid);
1831
1832         opd.opd_req = req;
1833         mode = LCK_PW;
1834         if (opc == OST_READ)
1835                 mode |= LCK_PR;
1836         opd.opd_mode = mode;
1837         opd.opd_exp = req->rq_export;
1838         opd.opd_oa  = &body->oa;
1839         opd.opd_extent.start = nb->offset;
1840         nb += ioo->ioo_bufcnt - 1;
1841         opd.opd_extent.end = nb->offset + nb->len - 1;
1842         opd.opd_timeout = prolong_timeout(req);
1843
1844         DEBUG_REQ(D_RPCTRACE, req,
1845                "%s %s: refresh rw locks: " LPU64"/"LPU64" ("LPU64"->"LPU64")\n",
1846                obd->obd_name, cfs_current()->comm,
1847                opd.opd_resid.name[0], opd.opd_resid.name[1],
1848                opd.opd_extent.start, opd.opd_extent.end);
1849
1850         ost_prolong_locks(&opd);
1851
1852         CDEBUG(D_DLMTRACE, "%s: refreshed %u locks timeout for req %p.\n",
1853                obd->obd_name, opd.opd_locks, req);
1854
1855         RETURN(opd.opd_locks);
1856 }
1857
1858 static void ost_rw_hpreq_fini(struct ptlrpc_request *req)
1859 {
1860         (void)ost_rw_hpreq_check(req);
1861 }
1862
1863 /**
1864  * Like ost_rw_hpreq_lock_match(), but for OST_PUNCH RPCs.
1865  */
1866 static int ost_punch_hpreq_lock_match(struct ptlrpc_request *req,
1867                                       struct ldlm_lock *lock)
1868 {
1869         struct ost_body *body;
1870         ENTRY;
1871
1872         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
1873         LASSERT(body != NULL);
1874
1875         if (body->oa.o_valid & OBD_MD_FLHANDLE &&
1876             body->oa.o_handle.cookie == lock->l_handle.h_cookie)
1877                 RETURN(1);
1878
1879         RETURN(0);
1880 }
1881
1882 /**
1883  * Like ost_rw_hpreq_check(), but for OST_PUNCH RPCs.
1884  */
1885 static int ost_punch_hpreq_check(struct ptlrpc_request *req)
1886 {
1887         struct obd_device *obd = req->rq_export->exp_obd;
1888         struct ost_body *body;
1889         struct obdo *oa;
1890         struct ost_prolong_data opd = { 0 };
1891         __u64 start, end;
1892         ENTRY;
1893
1894         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
1895         LASSERT(body != NULL);
1896
1897         oa = &body->oa;
1898         LASSERT(!(oa->o_valid & OBD_MD_FLFLAGS) ||
1899                 !(oa->o_flags & OBD_FL_SRVLOCK));
1900
1901         start = oa->o_size;
1902         end = start + oa->o_blocks;
1903
1904         opd.opd_req = req;
1905         opd.opd_mode = LCK_PW;
1906         opd.opd_exp = req->rq_export;
1907         opd.opd_oa  = oa;
1908         opd.opd_extent.start = start;
1909         opd.opd_extent.end   = end;
1910         if (oa->o_blocks == OBD_OBJECT_EOF)
1911                 opd.opd_extent.end = OBD_OBJECT_EOF;
1912         opd.opd_timeout = prolong_timeout(req);
1913
1914         osc_build_res_name(oa->o_id, oa->o_seq, &opd.opd_resid);
1915
1916         CDEBUG(D_DLMTRACE,
1917                "%s: refresh locks: "LPU64"/"LPU64" ("LPU64"->"LPU64")\n",
1918                obd->obd_name,
1919                opd.opd_resid.name[0], opd.opd_resid.name[1],
1920                opd.opd_extent.start, opd.opd_extent.end);
1921
1922         ost_prolong_locks(&opd);
1923
1924         CDEBUG(D_DLMTRACE, "%s: refreshed %u locks timeout for req %p.\n",
1925                obd->obd_name, opd.opd_locks, req);
1926
1927         RETURN(opd.opd_locks > 0);
1928 }
1929
1930 static void ost_punch_hpreq_fini(struct ptlrpc_request *req)
1931 {
1932         (void)ost_punch_hpreq_check(req);
1933 }
1934
1935 struct ptlrpc_hpreq_ops ost_hpreq_rw = {
1936         .hpreq_lock_match = ost_rw_hpreq_lock_match,
1937         .hpreq_check      = ost_rw_hpreq_check,
1938         .hpreq_fini       = ost_rw_hpreq_fini
1939 };
1940
1941 struct ptlrpc_hpreq_ops ost_hpreq_punch = {
1942         .hpreq_lock_match = ost_punch_hpreq_lock_match,
1943         .hpreq_check      = ost_punch_hpreq_check,
1944         .hpreq_fini       = ost_punch_hpreq_fini
1945 };
1946
1947 /** Assign high priority operations to the request if needed. */
1948 static int ost_hpreq_handler(struct ptlrpc_request *req)
1949 {
1950         ENTRY;
1951         if (req->rq_export) {
1952                 int opc = lustre_msg_get_opc(req->rq_reqmsg);
1953                 struct ost_body *body;
1954
1955                 if (opc == OST_READ || opc == OST_WRITE) {
1956                         struct niobuf_remote *nb;
1957                         struct obd_ioobj *ioo;
1958                         int objcount, niocount;
1959                         int rc;
1960                         int i;
1961
1962                         /* RPCs on the H-P queue can be inspected before
1963                          * ost_handler() initializes their pills, so we
1964                          * initialize that here.  Capsule initialization is
1965                          * idempotent, as is setting the pill's format (provided
1966                          * it doesn't change).
1967                          */
1968                         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
1969                         if (opc == OST_READ)
1970                                 req_capsule_set(&req->rq_pill,
1971                                                 &RQF_OST_BRW_READ);
1972                         else
1973                                 req_capsule_set(&req->rq_pill,
1974                                                 &RQF_OST_BRW_WRITE);
1975
1976                         body = req_capsule_client_get(&req->rq_pill,
1977                                                       &RMF_OST_BODY);
1978                         if (body == NULL) {
1979                                 CERROR("Missing/short ost_body\n");
1980                                 RETURN(-EFAULT);
1981                         }
1982
1983                         objcount = req_capsule_get_size(&req->rq_pill,
1984                                                         &RMF_OBD_IOOBJ,
1985                                                         RCL_CLIENT) /
1986                                                         sizeof(*ioo);
1987                         if (objcount == 0) {
1988                                 CERROR("Missing/short ioobj\n");
1989                                 RETURN(-EFAULT);
1990                         }
1991                         if (objcount > 1) {
1992                                 CERROR("too many ioobjs (%d)\n", objcount);
1993                                 RETURN(-EFAULT);
1994                         }
1995
1996                         ioo = req_capsule_client_get(&req->rq_pill,
1997                                                      &RMF_OBD_IOOBJ);
1998                         if (ioo == NULL) {
1999                                 CERROR("Missing/short ioobj\n");
2000                                 RETURN(-EFAULT);
2001                         }
2002
2003                         rc = ost_validate_obdo(req->rq_export, &body->oa, ioo);
2004                         if (rc) {
2005                                 CERROR("invalid object ids\n");
2006                                 RETURN(rc);
2007                         }
2008
2009                         for (niocount = i = 0; i < objcount; i++) {
2010                                 if (ioo[i].ioo_bufcnt == 0) {
2011                                         CERROR("ioo[%d] has zero bufcnt\n", i);
2012                                         RETURN(-EFAULT);
2013                                 }
2014                                 niocount += ioo[i].ioo_bufcnt;
2015                         }
2016                         if (niocount > PTLRPC_MAX_BRW_PAGES) {
2017                                 DEBUG_REQ(D_RPCTRACE, req,
2018                                           "bulk has too many pages (%d)",
2019                                           niocount);
2020                                 RETURN(-EFAULT);
2021                         }
2022
2023                         nb = req_capsule_client_get(&req->rq_pill,
2024                                                     &RMF_NIOBUF_REMOTE);
2025                         if (nb == NULL) {
2026                                 CERROR("Missing/short niobuf\n");
2027                                 RETURN(-EFAULT);
2028                         }
2029
2030                         if (niocount == 0 || !(nb[0].flags & OBD_BRW_SRVLOCK))
2031                                 req->rq_ops = &ost_hpreq_rw;
2032                 } else if (opc == OST_PUNCH) {
2033                         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2034                         req_capsule_set(&req->rq_pill, &RQF_OST_PUNCH);
2035
2036                         body = req_capsule_client_get(&req->rq_pill,
2037                                                       &RMF_OST_BODY);
2038                         if (body == NULL) {
2039                                 CERROR("Missing/short ost_body\n");
2040                                 RETURN(-EFAULT);
2041                         }
2042
2043                         if (!(body->oa.o_valid & OBD_MD_FLFLAGS) ||
2044                             !(body->oa.o_flags & OBD_FL_SRVLOCK))
2045                                 req->rq_ops = &ost_hpreq_punch;
2046                 }
2047         }
2048         RETURN(0);
2049 }
2050
2051 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
2052 int ost_handle(struct ptlrpc_request *req)
2053 {
2054         struct obd_trans_info trans_info = { 0, };
2055         struct obd_trans_info *oti = &trans_info;
2056         int should_process, fail = OBD_FAIL_OST_ALL_REPLY_NET, rc = 0;
2057         struct obd_device *obd = NULL;
2058         ENTRY;
2059
2060         /* OST module is kept between remounts, but the last reference
2061          * to specific module (say, osd or ofd) kills all related keys
2062          * from the environment. so we have to refill it until the root
2063          * cause is fixed properly */
2064         lu_env_refill(req->rq_svc_thread->t_env);
2065
2066         LASSERT(current->journal_info == NULL);
2067
2068         /* primordial rpcs don't affect server recovery */
2069         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2070         case SEC_CTX_INIT:
2071         case SEC_CTX_INIT_CONT:
2072         case SEC_CTX_FINI:
2073                 GOTO(out, rc = 0);
2074         }
2075
2076         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
2077
2078         if (lustre_msg_get_opc(req->rq_reqmsg) != OST_CONNECT) {
2079                 if (!class_connected_export(req->rq_export)) {
2080                         CDEBUG(D_HA,"operation %d on unconnected OST from %s\n",
2081                                lustre_msg_get_opc(req->rq_reqmsg),
2082                                libcfs_id2str(req->rq_peer));
2083                         req->rq_status = -ENOTCONN;
2084                         GOTO(out, rc = -ENOTCONN);
2085                 }
2086
2087                 obd = req->rq_export->exp_obd;
2088
2089                 /* Check for aborted recovery. */
2090                 if (obd->obd_recovering) {
2091                         rc = ost_filter_recovery_request(req, obd,
2092                                                          &should_process);
2093                         if (rc || !should_process)
2094                                 RETURN(rc);
2095                         else if (should_process < 0) {
2096                                 req->rq_status = should_process;
2097                                 rc = ptlrpc_error(req);
2098                                 RETURN(rc);
2099                         }
2100                 }
2101         }
2102
2103         oti_init(oti, req);
2104
2105         rc = ost_msg_check_version(req->rq_reqmsg);
2106         if (rc)
2107                 RETURN(rc);
2108
2109         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
2110         case OST_CONNECT: {
2111                 CDEBUG(D_INODE, "connect\n");
2112                 req_capsule_set(&req->rq_pill, &RQF_OST_CONNECT);
2113                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CONNECT_NET))
2114                         RETURN(0);
2115                 rc = target_handle_connect(req);
2116                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CONNECT_NET2))
2117                         RETURN(0);
2118                 if (!rc) {
2119                         rc = ost_init_sec_level(req);
2120                         if (!rc)
2121                                 rc = ost_connect_check_sptlrpc(req);
2122                 }
2123                 break;
2124         }
2125         case OST_DISCONNECT:
2126                 CDEBUG(D_INODE, "disconnect\n");
2127                 req_capsule_set(&req->rq_pill, &RQF_OST_DISCONNECT);
2128                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_DISCONNECT_NET))
2129                         RETURN(0);
2130                 rc = target_handle_disconnect(req);
2131                 break;
2132         case OST_CREATE:
2133                 CDEBUG(D_INODE, "create\n");
2134                 req_capsule_set(&req->rq_pill, &RQF_OST_CREATE);
2135                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CREATE_NET))
2136                         RETURN(0);
2137                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
2138                         GOTO(out, rc = -EROFS);
2139                 rc = ost_create(req->rq_export, req, oti);
2140                 break;
2141         case OST_DESTROY:
2142                 CDEBUG(D_INODE, "destroy\n");
2143                 req_capsule_set(&req->rq_pill, &RQF_OST_DESTROY);
2144                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_DESTROY_NET))
2145                         RETURN(0);
2146                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
2147                         GOTO(out, rc = -EROFS);
2148                 rc = ost_destroy(req->rq_export, req, oti);
2149                 break;
2150         case OST_GETATTR:
2151                 CDEBUG(D_INODE, "getattr\n");
2152                 req_capsule_set(&req->rq_pill, &RQF_OST_GETATTR);
2153                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_GETATTR_NET))
2154                         RETURN(0);
2155                 rc = ost_getattr(req->rq_export, req);
2156                 break;
2157         case OST_SETATTR:
2158                 CDEBUG(D_INODE, "setattr\n");
2159                 req_capsule_set(&req->rq_pill, &RQF_OST_SETATTR);
2160                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_NET))
2161                         RETURN(0);
2162                 rc = ost_setattr(req->rq_export, req, oti);
2163                 break;
2164         case OST_WRITE:
2165                 req_capsule_set(&req->rq_pill, &RQF_OST_BRW_WRITE);
2166                 CDEBUG(D_INODE, "write\n");
2167                 /* req->rq_request_portal would be nice, if it was set */
2168                 if (req->rq_rqbd->rqbd_service->srv_req_portal !=OST_IO_PORTAL){
2169                         CERROR("%s: deny write request from %s to portal %u\n",
2170                                req->rq_export->exp_obd->obd_name,
2171                                obd_export_nid2str(req->rq_export),
2172                                req->rq_rqbd->rqbd_service->srv_req_portal);
2173                         GOTO(out, rc = -EPROTO);
2174                 }
2175                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_NET))
2176                         RETURN(0);
2177                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
2178                         GOTO(out, rc = -ENOSPC);
2179                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
2180                         GOTO(out, rc = -EROFS);
2181                 rc = ost_brw_write(req, oti);
2182                 LASSERT(current->journal_info == NULL);
2183                 /* ost_brw_write sends its own replies */
2184                 RETURN(rc);
2185         case OST_READ:
2186                 req_capsule_set(&req->rq_pill, &RQF_OST_BRW_READ);
2187                 CDEBUG(D_INODE, "read\n");
2188                 /* req->rq_request_portal would be nice, if it was set */
2189                 if (req->rq_rqbd->rqbd_service->srv_req_portal !=OST_IO_PORTAL){
2190                         CERROR("%s: deny read request from %s to portal %u\n",
2191                                req->rq_export->exp_obd->obd_name,
2192                                obd_export_nid2str(req->rq_export),
2193                                req->rq_rqbd->rqbd_service->srv_req_portal);
2194                         GOTO(out, rc = -EPROTO);
2195                 }
2196                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_NET))
2197                         RETURN(0);
2198                 rc = ost_brw_read(req, oti);
2199                 LASSERT(current->journal_info == NULL);
2200                 /* ost_brw_read sends its own replies */
2201                 RETURN(rc);
2202         case OST_PUNCH:
2203                 CDEBUG(D_INODE, "punch\n");
2204                 req_capsule_set(&req->rq_pill, &RQF_OST_PUNCH);
2205                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_PUNCH_NET))
2206                         RETURN(0);
2207                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
2208                         GOTO(out, rc = -EROFS);
2209                 rc = ost_punch(req->rq_export, req, oti);
2210                 break;
2211         case OST_STATFS:
2212                 CDEBUG(D_INODE, "statfs\n");
2213                 req_capsule_set(&req->rq_pill, &RQF_OST_STATFS);
2214                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_STATFS_NET))
2215                         RETURN(0);
2216                 rc = ost_statfs(req);
2217                 break;
2218         case OST_SYNC:
2219                 CDEBUG(D_INODE, "sync\n");
2220                 req_capsule_set(&req->rq_pill, &RQF_OST_SYNC);
2221                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_SYNC_NET))
2222                         RETURN(0);
2223                 rc = ost_sync(req->rq_export, req);
2224                 break;
2225         case OST_SET_INFO:
2226                 DEBUG_REQ(D_INODE, req, "set_info");
2227                 req_capsule_set(&req->rq_pill, &RQF_OBD_SET_INFO);
2228                 rc = ost_set_info(req->rq_export, req);
2229                 break;
2230         case OST_GET_INFO:
2231                 DEBUG_REQ(D_INODE, req, "get_info");
2232                 req_capsule_set(&req->rq_pill, &RQF_OST_GET_INFO_GENERIC);
2233                 rc = ost_get_info(req->rq_export, req);
2234                 break;
2235 #ifdef HAVE_QUOTA_SUPPORT
2236         case OST_QUOTACHECK:
2237                 CDEBUG(D_INODE, "quotacheck\n");
2238                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTACHECK);
2239                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_QUOTACHECK_NET))
2240                         RETURN(0);
2241                 rc = ost_handle_quotacheck(req);
2242                 break;
2243         case OST_QUOTACTL:
2244                 CDEBUG(D_INODE, "quotactl\n");
2245                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTACTL);
2246                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_QUOTACTL_NET))
2247                         RETURN(0);
2248                 rc = ost_handle_quotactl(req);
2249                 break;
2250         case OST_QUOTA_ADJUST_QUNIT:
2251                 CDEBUG(D_INODE, "quota_adjust_qunit\n");
2252                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTA_ADJUST_QUNIT);
2253                 rc = ost_handle_quota_adjust_qunit(req);
2254                 break;
2255 #endif
2256         case OBD_PING:
2257                 DEBUG_REQ(D_INODE, req, "ping");
2258                 req_capsule_set(&req->rq_pill, &RQF_OBD_PING);
2259                 rc = target_handle_ping(req);
2260                 break;
2261         /* FIXME - just reply status */
2262         case LLOG_ORIGIN_CONNECT:
2263                 DEBUG_REQ(D_INODE, req, "log connect");
2264                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_CONNECT);
2265                 rc = ost_llog_handle_connect(req->rq_export, req);
2266                 req->rq_status = rc;
2267                 rc = req_capsule_server_pack(&req->rq_pill);
2268                 if (rc)
2269                         RETURN(rc);
2270                 RETURN(ptlrpc_reply(req));
2271         case OBD_LOG_CANCEL:
2272                 CDEBUG(D_INODE, "log cancel\n");
2273                 req_capsule_set(&req->rq_pill, &RQF_LOG_CANCEL);
2274                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
2275                         RETURN(0);
2276                 rc = llog_origin_handle_cancel(req);
2277                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_REP))
2278                         RETURN(0);
2279                 req->rq_status = rc;
2280                 rc = req_capsule_server_pack(&req->rq_pill);
2281                 if (rc)
2282                         RETURN(rc);
2283                 RETURN(ptlrpc_reply(req));
2284         case LDLM_ENQUEUE:
2285                 CDEBUG(D_INODE, "enqueue\n");
2286                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);
2287                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE))
2288                         RETURN(0);
2289                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
2290                                          ost_blocking_ast,
2291                                          ldlm_server_glimpse_ast);
2292                 fail = OBD_FAIL_OST_LDLM_REPLY_NET;
2293                 break;
2294         case LDLM_CONVERT:
2295                 CDEBUG(D_INODE, "convert\n");
2296                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CONVERT);
2297                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CONVERT))
2298                         RETURN(0);
2299                 rc = ldlm_handle_convert(req);
2300                 break;
2301         case LDLM_CANCEL:
2302                 CDEBUG(D_INODE, "cancel\n");
2303                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
2304                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL))
2305                         RETURN(0);
2306                 rc = ldlm_handle_cancel(req);
2307                 break;
2308         case LDLM_BL_CALLBACK:
2309         case LDLM_CP_CALLBACK:
2310                 CDEBUG(D_INODE, "callback\n");
2311                 CERROR("callbacks should not happen on OST\n");
2312                 /* fall through */
2313         default:
2314                 CERROR("Unexpected opcode %d\n",
2315                        lustre_msg_get_opc(req->rq_reqmsg));
2316                 req->rq_status = -ENOTSUPP;
2317                 rc = ptlrpc_error(req);
2318                 RETURN(rc);
2319         }
2320
2321         LASSERT(current->journal_info == NULL);
2322
2323         EXIT;
2324         /* If we're DISCONNECTing, the export_data is already freed */
2325         if (!rc && lustre_msg_get_opc(req->rq_reqmsg) != OST_DISCONNECT)
2326                 target_committed_to_req(req);
2327
2328 out:
2329         if (!rc)
2330                 oti_to_request(oti, req);
2331
2332         target_send_reply(req, rc, fail);
2333         return 0;
2334 }
2335 EXPORT_SYMBOL(ost_handle);
2336 /*
2337  * free per-thread pool created by ost_thread_init().
2338  */
2339 static void ost_thread_done(struct ptlrpc_thread *thread)
2340 {
2341         struct ost_thread_local_cache *tls; /* TLS stands for Thread-Local
2342                                              * Storage */
2343
2344         ENTRY;
2345
2346         LASSERT(thread != NULL);
2347
2348         /*
2349          * be prepared to handle partially-initialized pools (because this is
2350          * called from ost_thread_init() for cleanup.
2351          */
2352         tls = thread->t_data;
2353         if (tls != NULL) {
2354                 OBD_FREE_PTR(tls);
2355                 thread->t_data = NULL;
2356         }
2357         EXIT;
2358 }
2359
2360 /*
2361  * initialize per-thread page pool (bug 5137).
2362  */
2363 static int ost_thread_init(struct ptlrpc_thread *thread)
2364 {
2365         struct ost_thread_local_cache *tls;
2366
2367         ENTRY;
2368
2369         LASSERT(thread != NULL);
2370         LASSERT(thread->t_data == NULL);
2371         LASSERTF(thread->t_id <= OSS_THREADS_MAX, "%u\n", thread->t_id);
2372
2373         OBD_ALLOC_PTR(tls);
2374         if (tls == NULL)
2375                 RETURN(-ENOMEM);
2376         thread->t_data = tls;
2377         RETURN(0);
2378 }
2379
2380 #define OST_WATCHDOG_TIMEOUT (obd_timeout * 1000)
2381
2382 /* Sigh - really, this is an OSS, the _server_, not the _target_ */
2383 static int ost_setup(struct obd_device *obd, struct lustre_cfg* lcfg)
2384 {
2385         struct ost_obd *ost = &obd->u.ost;
2386         struct lprocfs_static_vars lvars;
2387         int oss_min_threads;
2388         int oss_max_threads;
2389         int oss_min_create_threads;
2390         int oss_max_create_threads;
2391         int rc;
2392         ENTRY;
2393
2394         rc = cfs_cleanup_group_info();
2395         if (rc)
2396                 RETURN(rc);
2397
2398         lprocfs_ost_init_vars(&lvars);
2399         lprocfs_obd_setup(obd, lvars.obd_vars);
2400
2401         cfs_mutex_init(&ost->ost_health_mutex);
2402
2403         if (oss_num_threads) {
2404                 /* If oss_num_threads is set, it is the min and the max. */
2405                 if (oss_num_threads > OSS_THREADS_MAX)
2406                         oss_num_threads = OSS_THREADS_MAX;
2407                 if (oss_num_threads < OSS_THREADS_MIN)
2408                         oss_num_threads = OSS_THREADS_MIN;
2409                 oss_max_threads = oss_min_threads = oss_num_threads;
2410         } else {
2411                 /* Base min threads on memory and cpus */
2412                 oss_min_threads =
2413                         cfs_num_possible_cpus() * CFS_NUM_CACHEPAGES >>
2414                         (27 - CFS_PAGE_SHIFT);
2415                 if (oss_min_threads < OSS_THREADS_MIN)
2416                         oss_min_threads = OSS_THREADS_MIN;
2417                 /* Insure a 4x range for dynamic threads */
2418                 if (oss_min_threads > OSS_THREADS_MAX / 4)
2419                         oss_min_threads = OSS_THREADS_MAX / 4;
2420                 oss_max_threads = min(OSS_THREADS_MAX, oss_min_threads * 4 + 1);
2421         }
2422
2423         ost->ost_service =
2424                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
2425                                 OST_MAXREPSIZE, OST_REQUEST_PORTAL,
2426                                 OSC_REPLY_PORTAL, OSS_SERVICE_WATCHDOG_FACTOR,
2427                                 ost_handle, LUSTRE_OSS_NAME,
2428                                 obd->obd_proc_entry, target_print_req,
2429                                 oss_min_threads, oss_max_threads,
2430                                 "ll_ost", LCT_DT_THREAD, NULL);
2431         if (ost->ost_service == NULL) {
2432                 CERROR("failed to start service\n");
2433                 GOTO(out_lprocfs, rc = -ENOMEM);
2434         }
2435
2436         rc = ptlrpc_start_threads(ost->ost_service);
2437         if (rc)
2438                 GOTO(out_service, rc = -EINVAL);
2439
2440         if (oss_num_create_threads) {
2441                 if (oss_num_create_threads > OSS_MAX_CREATE_THREADS)
2442                         oss_num_create_threads = OSS_MAX_CREATE_THREADS;
2443                 if (oss_num_create_threads < OSS_MIN_CREATE_THREADS)
2444                         oss_num_create_threads = OSS_MIN_CREATE_THREADS;
2445                 oss_min_create_threads = oss_max_create_threads =
2446                         oss_num_create_threads;
2447         } else {
2448                 oss_min_create_threads = OSS_MIN_CREATE_THREADS;
2449                 oss_max_create_threads = OSS_MAX_CREATE_THREADS;
2450         }
2451
2452         ost->ost_create_service =
2453                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
2454                                 OST_MAXREPSIZE, OST_CREATE_PORTAL,
2455                                 OSC_REPLY_PORTAL, OSS_SERVICE_WATCHDOG_FACTOR,
2456                                 ost_handle, "ost_create",
2457                                 obd->obd_proc_entry, target_print_req,
2458                                 oss_min_create_threads, oss_max_create_threads,
2459                                 "ll_ost_creat", LCT_DT_THREAD, NULL);
2460         if (ost->ost_create_service == NULL) {
2461                 CERROR("failed to start OST create service\n");
2462                 GOTO(out_service, rc = -ENOMEM);
2463         }
2464
2465         rc = ptlrpc_start_threads(ost->ost_create_service);
2466         if (rc)
2467                 GOTO(out_create, rc = -EINVAL);
2468
2469         ost->ost_io_service =
2470                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
2471                                 OST_MAXREPSIZE, OST_IO_PORTAL,
2472                                 OSC_REPLY_PORTAL, OSS_SERVICE_WATCHDOG_FACTOR,
2473                                 ost_handle, "ost_io",
2474                                 obd->obd_proc_entry, target_print_req,
2475                                 oss_min_threads, oss_max_threads,
2476                                 "ll_ost_io", LCT_DT_THREAD, ost_hpreq_handler);
2477         if (ost->ost_io_service == NULL) {
2478                 CERROR("failed to start OST I/O service\n");
2479                 GOTO(out_create, rc = -ENOMEM);
2480         }
2481
2482         ost->ost_io_service->srv_init = ost_thread_init;
2483         ost->ost_io_service->srv_done = ost_thread_done;
2484         ost->ost_io_service->srv_cpu_affinity = 1;
2485         rc = ptlrpc_start_threads(ost->ost_io_service);
2486         if (rc)
2487                 GOTO(out_io, rc = -EINVAL);
2488
2489         ping_evictor_start();
2490
2491         RETURN(0);
2492
2493 out_io:
2494         ptlrpc_unregister_service(ost->ost_io_service);
2495         ost->ost_io_service = NULL;
2496 out_create:
2497         ptlrpc_unregister_service(ost->ost_create_service);
2498         ost->ost_create_service = NULL;
2499 out_service:
2500         ptlrpc_unregister_service(ost->ost_service);
2501         ost->ost_service = NULL;
2502 out_lprocfs:
2503         lprocfs_obd_cleanup(obd);
2504         RETURN(rc);
2505 }
2506
2507 static int ost_cleanup(struct obd_device *obd)
2508 {
2509         struct ost_obd *ost = &obd->u.ost;
2510         int err = 0;
2511         ENTRY;
2512
2513         ping_evictor_stop();
2514
2515         /* there is no recovery for OST OBD, all recovery is controlled by
2516          * obdfilter OBD */
2517         LASSERT(obd->obd_recovering == 0);
2518         cfs_mutex_lock(&ost->ost_health_mutex);
2519         ptlrpc_unregister_service(ost->ost_service);
2520         ptlrpc_unregister_service(ost->ost_create_service);
2521         ptlrpc_unregister_service(ost->ost_io_service);
2522         ost->ost_service = NULL;
2523         ost->ost_create_service = NULL;
2524         cfs_mutex_unlock(&ost->ost_health_mutex);
2525
2526         lprocfs_obd_cleanup(obd);
2527
2528         RETURN(err);
2529 }
2530
2531 static int ost_health_check(struct obd_device *obd)
2532 {
2533         struct ost_obd *ost = &obd->u.ost;
2534         int rc = 0;
2535
2536         cfs_mutex_lock(&ost->ost_health_mutex);
2537         rc |= ptlrpc_service_health_check(ost->ost_service);
2538         rc |= ptlrpc_service_health_check(ost->ost_create_service);
2539         rc |= ptlrpc_service_health_check(ost->ost_io_service);
2540         cfs_mutex_unlock(&ost->ost_health_mutex);
2541
2542         /*
2543          * health_check to return 0 on healthy
2544          * and 1 on unhealthy.
2545          */
2546         if( rc != 0)
2547                 rc = 1;
2548
2549         return rc;
2550 }
2551
2552 struct ost_thread_local_cache *ost_tls(struct ptlrpc_request *r)
2553 {
2554         return (struct ost_thread_local_cache *)(r->rq_svc_thread->t_data);
2555 }
2556
2557 /* use obd ops to offer management infrastructure */
2558 static struct obd_ops ost_obd_ops = {
2559         .o_owner        = THIS_MODULE,
2560         .o_setup        = ost_setup,
2561         .o_cleanup      = ost_cleanup,
2562         .o_health_check = ost_health_check,
2563 };
2564
2565
2566 static int __init ost_init(void)
2567 {
2568         struct lprocfs_static_vars lvars;
2569         int rc;
2570         ENTRY;
2571
2572         lprocfs_ost_init_vars(&lvars);
2573         rc = class_register_type(&ost_obd_ops, NULL, lvars.module_vars,
2574                                  LUSTRE_OSS_NAME, NULL);
2575
2576         if (ost_num_threads != 0 && oss_num_threads == 0) {
2577                 LCONSOLE_INFO("ost_num_threads module parameter is deprecated, "
2578                               "use oss_num_threads instead or unset both for "
2579                               "dynamic thread startup\n");
2580                 oss_num_threads = ost_num_threads;
2581         }
2582
2583         RETURN(rc);
2584 }
2585
2586 static void /*__exit*/ ost_exit(void)
2587 {
2588         class_unregister_type(LUSTRE_OSS_NAME);
2589 }
2590
2591 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
2592 MODULE_DESCRIPTION("Lustre Object Storage Target (OST) v0.01");
2593 MODULE_LICENSE("GPL");
2594
2595 module_init(ost_init);
2596 module_exit(ost_exit);