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