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