lustre/ost/ost_handler.c

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