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