lustre/ost/ost_handler.c

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