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 void oti_to_request(struct obd_trans_info *oti, struct ptlrpc_request *req)
  73 {
  74         struct oti_req_ack_lock *ack_lock;
  75         int i;
  76
  77         if (oti == NULL)
  78                 return;
  79
  80         if (req->rq_repmsg)
  81                 lustre_msg_set_transno(req->rq_repmsg, oti->oti_transno);
  82         req->rq_transno = oti->oti_transno;
  83
  84         /* XXX 4 == entries in oti_ack_locks??? */
  85         for (ack_lock = oti->oti_ack_locks, i = 0; i < 4; i++, ack_lock++) {
  86                 if (!ack_lock->mode)
  87                         break;
  88                 /* XXX not even calling target_send_reply in some cases... */
  89                 ptlrpc_save_lock (req, &ack_lock->lock, ack_lock->mode, 0);
  90         }
  91 }
  92
  93 static int ost_destroy(struct obd_export *exp, struct ptlrpc_request *req,
  94                        struct obd_trans_info *oti)
  95 {
  96         struct ost_body *body, *repbody;
  97         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
  98         int rc;
  99         ENTRY;
 100
 101         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 102                                   lustre_swab_ost_body);
 103         if (body == NULL)
 104                 RETURN(-EFAULT);
 105
 106         if (body->oa.o_id == 0)
 107                 RETURN(-EPROTO);
 108
 109         if (lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF + 1)) {
 110                 struct ldlm_request *dlm;
 111                 dlm = lustre_swab_reqbuf(req, REQ_REC_OFF + 1, sizeof(*dlm),
 112                                          lustre_swab_ldlm_request);
 113                 if (dlm == NULL)
 114                         RETURN (-EFAULT);
 115                 ldlm_request_cancel(req, dlm, 0);
 116         }
 117
 118         rc = lustre_pack_reply(req, 2, size, NULL);
 119         if (rc)
 120                 RETURN(rc);
 121
 122         if (body->oa.o_valid & OBD_MD_FLCOOKIE)
 123                 oti->oti_logcookies = &body->oa.o_lcookie;
 124         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 125                                  sizeof(*repbody));
 126         memcpy(&repbody->oa, &body->oa, sizeof(body->oa));
 127         req->rq_status = obd_destroy(exp, &body->oa, NULL, oti, NULL);
 128         RETURN(0);
 129 }
 130
 131 static int ost_getattr(struct obd_export *exp, struct ptlrpc_request *req)
 132 {
 133         struct ost_body *body, *repbody;
 134         struct obd_info oinfo = { { { 0 } } };
 135         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 136         int rc;
 137         ENTRY;
 138
 139         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 140                                   lustre_swab_ost_body);
 141         if (body == NULL)
 142                 RETURN(-EFAULT);
 143
 144         rc = lustre_pack_reply(req, 2, size, NULL);
 145         if (rc)
 146                 RETURN(rc);
 147
 148         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 149                                  sizeof(*repbody));
 150         repbody->oa = body->oa;
 151
 152         oinfo.oi_oa = &repbody->oa;
 153         if (oinfo.oi_oa->o_valid & OBD_MD_FLOSSCAPA)
 154                 oinfo.oi_capa = lustre_unpack_capa(req->rq_reqmsg,
 155                                                    REQ_REC_OFF + 1);
 156         req->rq_status = obd_getattr(exp, &oinfo);
 157         RETURN(0);
 158 }
 159
 160 static int ost_statfs(struct ptlrpc_request *req)
 161 {
 162         struct obd_statfs *osfs;
 163         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*osfs) };
 164         int rc;
 165         ENTRY;
 166
 167         rc = lustre_pack_reply(req, 2, size, NULL);
 168         if (rc)
 169                 RETURN(rc);
 170
 171         osfs = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*osfs));
 172
 173         req->rq_status = obd_statfs(req->rq_export->exp_obd, osfs,
 174                                     cfs_time_current_64() - HZ, 0);
 175         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
 176                 osfs->os_bfree = osfs->os_bavail = 64;
 177         if (req->rq_status != 0)
 178                 CERROR("ost: statfs failed: rc %d\n", req->rq_status);
 179
 180         RETURN(0);
 181 }
 182
 183 static int ost_create(struct obd_export *exp, struct ptlrpc_request *req,
 184                       struct obd_trans_info *oti)
 185 {
 186         struct ost_body *body, *repbody;
 187         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*repbody) };
 188         int rc;
 189         ENTRY;
 190
 191         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 192                                   lustre_swab_ost_body);
 193         if (body == NULL)
 194                 RETURN(-EFAULT);
 195
 196         rc = lustre_pack_reply(req, 2, size, NULL);
 197         if (rc)
 198                 RETURN(rc);
 199
 200         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 201                                  sizeof(*repbody));
 202         memcpy(&repbody->oa, &body->oa, sizeof(body->oa));
 203         oti->oti_logcookies = &repbody->oa.o_lcookie;
 204         req->rq_status = obd_create(exp, &repbody->oa, NULL, oti);
 205         //obd_log_cancel(conn, NULL, 1, oti->oti_logcookies, 0);
 206         RETURN(0);
 207 }
 208
 209 /*
 210  * Helper function for ost_punch(): if asked by client, acquire [size, EOF]
 211  * lock on the file being truncated.
 212  */
 213 static int ost_punch_lock_get(struct obd_export *exp, struct obdo *oa,
 214                               struct lustre_handle *lh)
 215 {
 216         int flags;
 217         struct ldlm_res_id res_id;
 218         ldlm_policy_data_t policy;
 219         __u64 start;
 220         __u64 finis;
 221
 222         ENTRY;
 223
 224         osc_build_res_name(oa->o_id, oa->o_gr, &res_id);
 225         LASSERT(!lustre_handle_is_used(lh));
 226
 227         if (!(oa->o_valid & OBD_MD_FLFLAGS) ||
 228             !(oa->o_flags & OBD_FL_TRUNCLOCK))
 229                 RETURN(0);
 230
 231         CDEBUG(D_INODE, "OST-side truncate lock.\n");
 232
 233         start = oa->o_size;
 234         finis = start + oa->o_blocks;
 235
 236         /*
 237          * standard truncate optimization: if file body is completely
 238          * destroyed, don't send data back to the server.
 239          */
 240         flags = (start == 0) ? LDLM_AST_DISCARD_DATA : 0;
 241
 242         policy.l_extent.start = start & CFS_PAGE_MASK;
 243
 244         /*
 245          * If ->o_blocks is EOF it means "lock till the end of the
 246          * file". Otherwise, it's size of a hole being punched (in bytes)
 247          */
 248         if (oa->o_blocks == OBD_OBJECT_EOF || finis < start)
 249                 policy.l_extent.end = OBD_OBJECT_EOF;
 250         else
 251                 policy.l_extent.end = finis | ~CFS_PAGE_MASK;
 252
 253         RETURN(ldlm_cli_enqueue_local(exp->exp_obd->obd_namespace, &res_id,
 254                                       LDLM_EXTENT, &policy, LCK_PW, &flags,
 255                                       ldlm_blocking_ast, ldlm_completion_ast,
 256                                       ldlm_glimpse_ast, NULL, 0, NULL,
 257                                       NULL, lh));
 258 }
 259
 260 /*
 261  * Helper function for ost_punch(): release lock acquired by
 262  * ost_punch_lock_get(), if any.
 263  */
 264 static void ost_punch_lock_put(struct obd_export *exp, struct obdo *oa,
 265                                struct lustre_handle *lh)
 266 {
 267         ENTRY;
 268         if (lustre_handle_is_used(lh))
 269                 ldlm_lock_decref(lh, LCK_PW);
 270         EXIT;
 271 }
 272
 273 static int ost_punch(struct obd_export *exp, struct ptlrpc_request *req,
 274                      struct obd_trans_info *oti)
 275 {
 276         struct obd_info oinfo = { { { 0 } } };
 277         struct ost_body *body, *repbody;
 278         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*repbody) };
 279         int rc;
 280         struct lustre_handle lh = {0,};
 281         ENTRY;
 282
 283         /* check that we do support OBD_CONNECT_TRUNCLOCK. */
 284         CLASSERT(OST_CONNECT_SUPPORTED & OBD_CONNECT_TRUNCLOCK);
 285
 286         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 287                                   lustre_swab_ost_body);
 288         if (body == NULL)
 289                 RETURN(-EFAULT);
 290
 291         oinfo.oi_oa = &body->oa;
 292         oinfo.oi_policy.l_extent.start = oinfo.oi_oa->o_size;
 293         oinfo.oi_policy.l_extent.end = oinfo.oi_oa->o_blocks;
 294
 295         if ((oinfo.oi_oa->o_valid & (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS)) !=
 296             (OBD_MD_FLSIZE | OBD_MD_FLBLOCKS))
 297                 RETURN(-EPROTO);
 298
 299         rc = lustre_pack_reply(req, 2, size, NULL);
 300         if (rc)
 301                 RETURN(rc);
 302
 303         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 304                                  sizeof(*repbody));
 305         rc = ost_punch_lock_get(exp, oinfo.oi_oa, &lh);
 306         if (rc == 0) {
 307                 if (oinfo.oi_oa->o_valid & OBD_MD_FLFLAGS &&
 308                     oinfo.oi_oa->o_flags == OBD_FL_TRUNCLOCK)
 309                         /*
 310                          * If OBD_FL_TRUNCLOCK is the only bit set in
 311                          * ->o_flags, clear OBD_MD_FLFLAGS to avoid falling
 312                          * through filter_setattr() to filter_iocontrol().
 313                          */
 314                         oinfo.oi_oa->o_valid &= ~OBD_MD_FLFLAGS;
 315
 316                 if (oinfo.oi_oa->o_valid & OBD_MD_FLOSSCAPA)
 317                         oinfo.oi_capa = lustre_unpack_capa(req->rq_reqmsg,
 318                                                            REQ_REC_OFF + 1);
 319                 req->rq_status = obd_punch(exp, &oinfo, oti, NULL);
 320                 ost_punch_lock_put(exp, oinfo.oi_oa, &lh);
 321         }
 322         repbody->oa = *oinfo.oi_oa;
 323         RETURN(rc);
 324 }
 325
 326 static int ost_sync(struct obd_export *exp, struct ptlrpc_request *req)
 327 {
 328         struct ost_body *body, *repbody;
 329         struct lustre_capa *capa = NULL;
 330         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*repbody) };
 331         int rc;
 332         ENTRY;
 333
 334         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 335                                   lustre_swab_ost_body);
 336         if (body == NULL)
 337                 RETURN(-EFAULT);
 338
 339         if (body->oa.o_valid & OBD_MD_FLOSSCAPA)
 340                 capa = lustre_unpack_capa(req->rq_reqmsg, REQ_REC_OFF + 1);
 341
 342         rc = lustre_pack_reply(req, 2, size, NULL);
 343         if (rc)
 344                 RETURN(rc);
 345
 346         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 347                                  sizeof(*repbody));
 348         memcpy(&repbody->oa, &body->oa, sizeof(body->oa));
 349         req->rq_status = obd_sync(exp, &repbody->oa, NULL, repbody->oa.o_size,
 350                                   repbody->oa.o_blocks, capa);
 351         RETURN(0);
 352 }
 353
 354 static int ost_setattr(struct obd_export *exp, struct ptlrpc_request *req,
 355                        struct obd_trans_info *oti)
 356 {
 357         struct ost_body *body, *repbody;
 358         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*repbody) };
 359         int rc;
 360         struct obd_info oinfo = { { { 0 } } };
 361         ENTRY;
 362
 363         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 364                                   lustre_swab_ost_body);
 365         if (body == NULL)
 366                 RETURN(-EFAULT);
 367
 368         rc = lustre_pack_reply(req, 2, size, NULL);
 369         if (rc)
 370                 RETURN(rc);
 371
 372         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 373                                  sizeof(*repbody));
 374         repbody->oa = body->oa;
 375
 376         oinfo.oi_oa = &repbody->oa;
 377         if (oinfo.oi_oa->o_valid & OBD_MD_FLOSSCAPA)
 378                 oinfo.oi_capa = lustre_unpack_capa(req->rq_reqmsg,
 379                                                    REQ_REC_OFF + 1);
 380         req->rq_status = obd_setattr(exp, &oinfo, oti);
 381         RETURN(0);
 382 }
 383
 384 static int ost_bulk_timeout(void *data)
 385 {
 386         ENTRY;
 387         /* We don't fail the connection here, because having the export
 388          * killed makes the (vital) call to commitrw very sad.
 389          */
 390         RETURN(1);
 391 }
 392
 393 static __u32 ost_checksum_bulk(struct ptlrpc_bulk_desc *desc, int opc,
 394                                cksum_type_t cksum_type)
 395 {
 396         __u32 cksum;
 397         int i;
 398
 399         cksum = init_checksum(cksum_type);
 400         for (i = 0; i < desc->bd_iov_count; i++) {
 401                 struct page *page = desc->bd_iov[i].kiov_page;
 402                 int off = desc->bd_iov[i].kiov_offset & ~CFS_PAGE_MASK;
 403                 char *ptr = kmap(page) + off;
 404                 int len = desc->bd_iov[i].kiov_len;
 405
 406                 /* corrupt the data before we compute the checksum, to
 407                  * simulate a client->OST data error */
 408                 if (i == 0 && opc == OST_WRITE &&
 409                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_RECEIVE))
 410                         memcpy(ptr, "bad3", min(4, len));
 411                 cksum = compute_checksum(cksum, ptr, len, cksum_type);
 412                 /* corrupt the data after we compute the checksum, to
 413                  * simulate an OST->client data error */
 414                 if (i == 0 && opc == OST_READ &&
 415                     OBD_FAIL_CHECK(OBD_FAIL_OST_CHECKSUM_SEND)) {
 416                         memcpy(ptr, "bad4", min(4, len));
 417                         /* nobody should use corrupted page again */
 418                         ClearPageUptodate(page);
 419                 }
 420                 kunmap(page);
 421         }
 422
 423         return cksum;
 424 }
 425
 426 static int ost_brw_lock_get(int mode, struct obd_export *exp,
 427                             struct obd_ioobj *obj, struct niobuf_remote *nb,
 428                             struct lustre_handle *lh)
 429 {
 430         int flags                 = 0;
 431         int nrbufs                = obj->ioo_bufcnt;
 432         struct ldlm_res_id res_id;
 433         ldlm_policy_data_t policy;
 434         int i;
 435         ENTRY;
 436
 437         osc_build_res_name(obj->ioo_id, obj->ioo_gr, &res_id);
 438         LASSERT(mode == LCK_PR || mode == LCK_PW);
 439         LASSERT(!lustre_handle_is_used(lh));
 440
 441         if (nrbufs == 0 || !(nb[0].flags & OBD_BRW_SRVLOCK))
 442                 RETURN(0);
 443
 444         /* EXPENSIVE ASSERTION */
 445         for (i = 1; i < nrbufs; i ++)
 446                 LASSERT((nb[0].flags & OBD_BRW_SRVLOCK) ==
 447                         (nb[i].flags & OBD_BRW_SRVLOCK));
 448
 449         policy.l_extent.start = nb[0].offset & CFS_PAGE_MASK;
 450         policy.l_extent.end   = (nb[nrbufs - 1].offset +
 451                                  nb[nrbufs - 1].len - 1) | ~CFS_PAGE_MASK;
 452
 453         RETURN(ldlm_cli_enqueue_local(exp->exp_obd->obd_namespace, &res_id,
 454                                       LDLM_EXTENT, &policy, mode, &flags,
 455                                       ldlm_blocking_ast, ldlm_completion_ast,
 456                                       ldlm_glimpse_ast, NULL, 0, NULL,
 457                                       NULL, lh));
 458 }
 459
 460 static void ost_brw_lock_put(int mode,
 461                              struct obd_ioobj *obj, struct niobuf_remote *niob,
 462                              struct lustre_handle *lh)
 463 {
 464         ENTRY;
 465         LASSERT(mode == LCK_PR || mode == LCK_PW);
 466         LASSERT((obj->ioo_bufcnt > 0 && (niob[0].flags & OBD_BRW_SRVLOCK)) ==
 467                 lustre_handle_is_used(lh));
 468         if (lustre_handle_is_used(lh))
 469                 ldlm_lock_decref(lh, mode);
 470         EXIT;
 471 }
 472
 473 struct ost_prolong_data {
 474         struct obd_export *opd_exp;
 475         ldlm_policy_data_t opd_policy;
 476         ldlm_mode_t opd_mode;
 477 };
 478
 479 static int ost_prolong_locks_iter(struct ldlm_lock *lock, void *data)
 480 {
 481         struct ost_prolong_data *opd = data;
 482
 483         LASSERT(lock->l_resource->lr_type == LDLM_EXTENT);
 484
 485         if (lock->l_req_mode != lock->l_granted_mode) {
 486                 /* scan granted locks only */
 487                 return LDLM_ITER_STOP;
 488         }
 489
 490         if (lock->l_export != opd->opd_exp) {
 491                 /* prolong locks only for given client */
 492                 return LDLM_ITER_CONTINUE;
 493         }
 494
 495         if (!(lock->l_granted_mode & opd->opd_mode)) {
 496                 /* we aren't interesting in all type of locks */
 497                 return LDLM_ITER_CONTINUE;
 498         }
 499
 500         if (lock->l_policy_data.l_extent.end < opd->opd_policy.l_extent.start ||
 501             lock->l_policy_data.l_extent.start > opd->opd_policy.l_extent.end) {
 502                 /* the request doesn't cross the lock, skip it */
 503                 return LDLM_ITER_CONTINUE;
 504         }
 505
 506         if (!(lock->l_flags & LDLM_FL_AST_SENT)) {
 507                 /* ignore locks not being cancelled */
 508                 return LDLM_ITER_CONTINUE;
 509         }
 510
 511         /* OK. this is a possible lock the user holds doing I/O
 512          * let's refresh eviction timer for it */
 513         ldlm_refresh_waiting_lock(lock);
 514
 515         return LDLM_ITER_CONTINUE;
 516 }
 517
 518 static void ost_prolong_locks(struct obd_export *exp, struct obd_ioobj *obj,
 519                               struct niobuf_remote *nb, struct obdo *oa,
 520                               ldlm_mode_t mode)
 521 {
 522         struct ldlm_res_id res_id;
 523         int nrbufs = obj->ioo_bufcnt;
 524         struct ost_prolong_data opd;
 525         ENTRY;
 526
 527         osc_build_res_name(obj->ioo_id, obj->ioo_gr, &res_id);
 528
 529         opd.opd_mode = mode;
 530         opd.opd_exp = exp;
 531         opd.opd_policy.l_extent.start = nb[0].offset & CFS_PAGE_MASK;
 532         opd.opd_policy.l_extent.end = (nb[nrbufs - 1].offset +
 533                                        nb[nrbufs - 1].len - 1) | ~CFS_PAGE_MASK;
 534
 535         CDEBUG(D_DLMTRACE,"refresh locks: "LPU64"/"LPU64" ("LPU64"->"LPU64")\n",
 536                res_id.name[0], res_id.name[1], opd.opd_policy.l_extent.start,
 537                opd.opd_policy.l_extent.end);
 538
 539         if (oa->o_valid & OBD_MD_FLHANDLE) {
 540                 struct ldlm_lock *lock;
 541
 542                 lock = ldlm_handle2lock(&oa->o_handle);
 543                 if (lock != NULL) {
 544                         ost_prolong_locks_iter(lock, &opd);
 545                         LDLM_LOCK_PUT(lock);
 546                         EXIT;
 547                         return;
 548                 }
 549         }
 550
 551         ldlm_resource_iterate(exp->exp_obd->obd_namespace, &res_id,
 552                               ost_prolong_locks_iter, &opd);
 553
 554         EXIT;
 555 }
 556
 557 static int ost_brw_read(struct ptlrpc_request *req, struct obd_trans_info *oti)
 558 {
 559         struct ptlrpc_bulk_desc *desc = NULL;
 560         struct obd_export *exp = req->rq_export;
 561         struct niobuf_remote *remote_nb;
 562         struct niobuf_local *local_nb;
 563         struct obd_ioobj *ioo;
 564         struct ost_body *body, *repbody;
 565         struct lustre_capa *capa = NULL;
 566         struct l_wait_info lwi;
 567         struct lustre_handle lockh = { 0 };
 568         __u32  size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 569         int objcount, niocount, npages, nob = 0, rc, i;
 570         int no_reply = 0;
 571         ENTRY;
 572
 573         req->rq_bulk_read = 1;
 574
 575         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_READ_BULK))
 576                 GOTO(out, rc = -EIO);
 577
 578         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, (obd_timeout + 1) / 4);
 579
 580         /* Check if there is eviction in progress, and if so, wait for it to
 581          * finish */
 582         if (unlikely(atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
 583                 lwi = LWI_INTR(NULL, NULL); // We do not care how long it takes
 584                 rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
 585                         !atomic_read(&exp->exp_obd->obd_evict_inprogress),
 586                         &lwi);
 587         }
 588         if (exp->exp_failed)
 589                 GOTO(out, rc = -ENOTCONN);
 590
 591         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 592                                   lustre_swab_ost_body);
 593         if (body == NULL) {
 594                 CERROR("Missing/short ost_body\n");
 595                 GOTO(out, rc = -EFAULT);
 596         }
 597
 598         objcount = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF + 1) /
 599                    sizeof(*ioo);
 600         if (objcount == 0) {
 601                 CERROR("Missing/short ioobj\n");
 602                 GOTO(out, rc = -EFAULT);
 603         }
 604         if (objcount > 1) {
 605                 CERROR("too many ioobjs (%d)\n", objcount);
 606                 GOTO(out, rc = -EFAULT);
 607         }
 608
 609         ioo = lustre_swab_reqbuf(req, REQ_REC_OFF + 1, sizeof(*ioo),
 610                                  lustre_swab_obd_ioobj);
 611         if (ioo == NULL) {
 612                 CERROR("Missing/short ioobj\n");
 613                 GOTO(out, rc = -EFAULT);
 614         }
 615
 616         niocount = ioo->ioo_bufcnt;
 617         if (niocount > PTLRPC_MAX_BRW_PAGES) {
 618                 DEBUG_REQ(D_ERROR, req, "bulk has too many pages (%d)",
 619                           niocount);
 620                 GOTO(out, rc = -EFAULT);
 621         }
 622
 623         remote_nb = lustre_swab_reqbuf(req, REQ_REC_OFF + 2,
 624                                        niocount * sizeof(*remote_nb),
 625                                        lustre_swab_niobuf_remote);
 626         if (remote_nb == NULL) {
 627                 CERROR("Missing/short niobuf\n");
 628                 GOTO(out, rc = -EFAULT);
 629         }
 630
 631         if (lustre_req_need_swab(req)) { /* swab remaining niobufs */
 632                 for (i = 1; i < niocount; i++)
 633                         lustre_swab_niobuf_remote (&remote_nb[i]);
 634         }
 635
 636         if (body->oa.o_valid & OBD_MD_FLOSSCAPA)
 637                 capa = lustre_unpack_capa(req->rq_reqmsg, REQ_REC_OFF + 3);
 638
 639         rc = lustre_pack_reply(req, 2, size, NULL);
 640         if (rc)
 641                 GOTO(out, rc);
 642
 643         /*
 644          * Per-thread array of struct niobuf_{local,remote}'s was allocated by
 645          * ost_thread_init().
 646          */
 647         local_nb = ost_tls(req)->local;
 648
 649         rc = ost_brw_lock_get(LCK_PR, exp, ioo, remote_nb, &lockh);
 650         if (rc != 0)
 651                 GOTO(out_bulk, rc);
 652
 653         /*
 654          * If getting the lock took more time than
 655          * client was willing to wait, drop it. b=11330
 656          */
 657         if (cfs_time_current_sec() > req->rq_deadline ||
 658             OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
 659                 no_reply = 1;
 660                 CERROR("Dropping timed-out read from %s because locking"
 661                        "object "LPX64" took %ld seconds (limit was %ld).\n",
 662                        libcfs_id2str(req->rq_peer), ioo->ioo_id,
 663                        cfs_time_current_sec() - req->rq_arrival_time.tv_sec,
 664                        req->rq_deadline - req->rq_arrival_time.tv_sec);
 665                 GOTO(out_lock, rc = -ETIMEDOUT);
 666         }
 667
 668         npages = OST_THREAD_POOL_SIZE;
 669         rc = obd_preprw(OBD_BRW_READ, exp, &body->oa, 1, ioo,
 670                         remote_nb, &npages, local_nb, oti, capa);
 671         if (rc != 0)
 672                 GOTO(out_lock, rc);
 673
 674         desc = ptlrpc_prep_bulk_exp(req, npages,
 675                                      BULK_PUT_SOURCE, OST_BULK_PORTAL);
 676         if (desc == NULL) /* XXX: check all cleanup stuff */
 677                 GOTO(out, rc = -ENOMEM);
 678
 679         ost_prolong_locks(exp, ioo, remote_nb, &body->oa, LCK_PW | LCK_PR);
 680
 681         nob = 0;
 682         for (i = 0; i < npages; i++) {
 683                 int page_rc = local_nb[i].rc;
 684
 685                 if (page_rc < 0) {              /* error */
 686                         rc = page_rc;
 687                         break;
 688                 }
 689
 690                 nob += page_rc;
 691                 if (page_rc != 0) {             /* some data! */
 692                         LASSERT (local_nb[i].page != NULL);
 693                         ptlrpc_prep_bulk_page(desc, local_nb[i].page,
 694                                               local_nb[i].offset & ~CFS_PAGE_MASK,
 695                                               page_rc);
 696                 }
 697
 698                 if (page_rc != local_nb[i].len) { /* short read */
 699                         /* All subsequent pages should be 0 */
 700                         while(++i < npages)
 701                                 LASSERT(local_nb[i].rc == 0);
 702                         break;
 703                 }
 704         }
 705
 706         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
 707                 cksum_type_t cksum_type = OBD_CKSUM_CRC32;
 708
 709                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
 710                         cksum_type = cksum_type_unpack(body->oa.o_flags);
 711                 body->oa.o_flags = cksum_type_pack(cksum_type);
 712                 body->oa.o_valid = OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
 713                 body->oa.o_cksum = ost_checksum_bulk(desc, OST_READ, cksum_type);
 714                 CDEBUG(D_PAGE,"checksum at read origin: %x\n",body->oa.o_cksum);
 715         } else {
 716                 body->oa.o_valid = 0;
 717         }
 718         /* We're finishing using body->oa as an input variable */
 719
 720         /* Check if client was evicted while we were doing i/o before touching
 721            network */
 722         if (rc == 0) {
 723                 /* Check if there is eviction in progress, and if so, wait for
 724                  * it to finish */
 725                 if (unlikely(atomic_read(&exp->exp_obd->
 726                                                 obd_evict_inprogress))) {
 727                         lwi = LWI_INTR(NULL, NULL);
 728                         rc = l_wait_event(exp->exp_obd->
 729                                                 obd_evict_inprogress_waitq,
 730                                           !atomic_read(&exp->exp_obd->
 731                                                         obd_evict_inprogress),
 732                                           &lwi);
 733                 }
 734                 if (exp->exp_failed)
 735                         rc = -ENOTCONN;
 736                 else {
 737                         sptlrpc_svc_wrap_bulk(req, desc);
 738
 739                         rc = ptlrpc_start_bulk_transfer(desc);
 740                 }
 741
 742                 if (rc == 0) {
 743                         time_t start = cfs_time_current_sec();
 744                         do {
 745                                 long timeoutl = req->rq_deadline -
 746                                         cfs_time_current_sec();
 747                                 cfs_duration_t timeout = (timeoutl <= 0 || rc) ?
 748                                         CFS_TICK : cfs_time_seconds(timeoutl);
 749                                 lwi = LWI_TIMEOUT_INTERVAL(timeout,
 750                                                            cfs_time_seconds(1),
 751                                                            ost_bulk_timeout,
 752                                                            desc);
 753                                 rc = l_wait_event(desc->bd_waitq,
 754                                                   !ptlrpc_bulk_active(desc) ||
 755                                                   exp->exp_failed, &lwi);
 756                                 LASSERT(rc == 0 || rc == -ETIMEDOUT);
 757                                 /* Wait again if we changed deadline */
 758                         } while ((rc == -ETIMEDOUT) &&
 759                                  (req->rq_deadline > cfs_time_current_sec()));
 760
 761                         if (rc == -ETIMEDOUT) {
 762                                 DEBUG_REQ(D_ERROR, req,
 763                                           "timeout on bulk PUT after %ld%+lds",
 764                                           req->rq_deadline - start,
 765                                           cfs_time_current_sec() -
 766                                           req->rq_deadline);
 767                                 ptlrpc_abort_bulk(desc);
 768                         } else if (exp->exp_failed) {
 769                                 DEBUG_REQ(D_ERROR, req, "Eviction on bulk PUT");
 770                                 rc = -ENOTCONN;
 771                                 ptlrpc_abort_bulk(desc);
 772                         } else if (!desc->bd_success ||
 773                                    desc->bd_nob_transferred != desc->bd_nob) {
 774                                 DEBUG_REQ(D_ERROR, req, "%s bulk PUT %d(%d)",
 775                                           desc->bd_success ?
 776                                           "truncated" : "network error on",
 777                                           desc->bd_nob_transferred,
 778                                           desc->bd_nob);
 779                                 /* XXX should this be a different errno? */
 780                                 rc = -ETIMEDOUT;
 781                         }
 782                 } else {
 783                         DEBUG_REQ(D_ERROR, req, "bulk PUT failed: rc %d", rc);
 784                 }
 785                 no_reply = rc != 0;
 786         }
 787
 788         /* Must commit after prep above in all cases */
 789         rc = obd_commitrw(OBD_BRW_READ, exp, &body->oa, 1, ioo,
 790                           remote_nb, npages, local_nb, oti, rc);
 791
 792         if (rc == 0) {
 793                 repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
 794                                          sizeof(*repbody));
 795                 memcpy(&repbody->oa, &body->oa, sizeof(repbody->oa));
 796         }
 797
 798 out_lock:
 799         ost_brw_lock_put(LCK_PR, ioo, remote_nb, &lockh);
 800 out_bulk:
 801         if (desc)
 802                 ptlrpc_free_bulk(desc);
 803 out:
 804         LASSERT(rc <= 0);
 805         if (rc == 0) {
 806                 req->rq_status = nob;
 807                 ptlrpc_lprocfs_brw(req, nob);
 808                 target_committed_to_req(req);
 809                 ptlrpc_reply(req);
 810         } else if (!no_reply) {
 811                 /* Only reply if there was no comms problem with bulk */
 812                 target_committed_to_req(req);
 813                 req->rq_status = rc;
 814                 ptlrpc_error(req);
 815         } else {
 816                 /* reply out callback would free */
 817                 ptlrpc_req_drop_rs(req);
 818                 CWARN("%s: ignoring bulk IO comm error with %s@%s id %s - "
 819                       "client will retry\n",
 820                       exp->exp_obd->obd_name,
 821                       exp->exp_client_uuid.uuid,
 822                       exp->exp_connection->c_remote_uuid.uuid,
 823                       libcfs_id2str(req->rq_peer));
 824         }
 825
 826         RETURN(rc);
 827 }
 828
 829 static int ost_brw_write(struct ptlrpc_request *req, struct obd_trans_info *oti)
 830 {
 831         struct ptlrpc_bulk_desc *desc = NULL;
 832         struct obd_export       *exp = req->rq_export;
 833         struct niobuf_remote    *remote_nb;
 834         struct niobuf_local     *local_nb;
 835         struct obd_ioobj        *ioo;
 836         struct ost_body         *body, *repbody;
 837         struct l_wait_info       lwi;
 838         struct lustre_handle     lockh = {0};
 839         struct lustre_capa      *capa = NULL;
 840         __u32                   *rcs;
 841         __u32 size[3] = { sizeof(struct ptlrpc_body), sizeof(*body) };
 842         int objcount, niocount, npages;
 843         int rc, swab, i, j;
 844         obd_count                client_cksum = 0, server_cksum = 0;
 845         cksum_type_t             cksum_type = OBD_CKSUM_CRC32;
 846         int                      no_reply = 0;
 847         ENTRY;
 848
 849         req->rq_bulk_write = 1;
 850
 851         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_WRITE_BULK))
 852                 GOTO(out, rc = -EIO);
 853         if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_WRITE_BULK2))
 854                 GOTO(out, rc = -EFAULT);
 855
 856         /* pause before transaction has been started */
 857         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_BULK, (obd_timeout + 1) / 4);
 858
 859         /* Check if there is eviction in progress, and if so, wait for it to
 860          * finish */
 861         if (unlikely(atomic_read(&exp->exp_obd->obd_evict_inprogress))) {
 862                 lwi = LWI_INTR(NULL, NULL); // We do not care how long it takes
 863                 rc = l_wait_event(exp->exp_obd->obd_evict_inprogress_waitq,
 864                         !atomic_read(&exp->exp_obd->obd_evict_inprogress),
 865                         &lwi);
 866         }
 867         if (exp->exp_failed)
 868                 GOTO(out, rc = -ENOTCONN);
 869
 870         swab = lustre_req_need_swab(req);
 871         body = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*body),
 872                                   lustre_swab_ost_body);
 873         if (body == NULL) {
 874                 CERROR("Missing/short ost_body\n");
 875                 GOTO(out, rc = -EFAULT);
 876         }
 877
 878         lustre_set_req_swabbed(req, REQ_REC_OFF + 1);
 879         objcount = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF + 1) /
 880                    sizeof(*ioo);
 881         if (objcount == 0) {
 882                 CERROR("Missing/short ioobj\n");
 883                 GOTO(out, rc = -EFAULT);
 884         }
 885         if (objcount > 1) {
 886                 CERROR("too many ioobjs (%d)\n", objcount);
 887                 GOTO(out, rc = -EFAULT);
 888         }
 889
 890         ioo = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 1,
 891                              objcount * sizeof(*ioo));
 892         LASSERT (ioo != NULL);
 893         for (niocount = i = 0; i < objcount; i++) {
 894                 if (swab)
 895                         lustre_swab_obd_ioobj(&ioo[i]);
 896                 if (ioo[i].ioo_bufcnt == 0) {
 897                         CERROR("ioo[%d] has zero bufcnt\n", i);
 898                         GOTO(out, rc = -EFAULT);
 899                 }
 900                 niocount += ioo[i].ioo_bufcnt;
 901         }
 902
 903         if (niocount > PTLRPC_MAX_BRW_PAGES) {
 904                 DEBUG_REQ(D_ERROR, req, "bulk has too many pages (%d)",
 905                           niocount);
 906                 GOTO(out, rc = -EFAULT);
 907         }
 908
 909         remote_nb = lustre_swab_reqbuf(req, REQ_REC_OFF + 2,
 910                                        niocount * sizeof(*remote_nb),
 911                                        lustre_swab_niobuf_remote);
 912         if (remote_nb == NULL) {
 913                 CERROR("Missing/short niobuf\n");
 914                 GOTO(out, rc = -EFAULT);
 915         }
 916         if (swab) {                             /* swab the remaining niobufs */
 917                 for (i = 1; i < niocount; i++)
 918                         lustre_swab_niobuf_remote (&remote_nb[i]);
 919         }
 920
 921         if (body->oa.o_valid & OBD_MD_FLOSSCAPA)
 922                 capa = lustre_unpack_capa(req->rq_reqmsg, REQ_REC_OFF + 3);
 923
 924         size[REPLY_REC_OFF + 1] = niocount * sizeof(*rcs);
 925         rc = lustre_pack_reply(req, 3, size, NULL);
 926         if (rc != 0)
 927                 GOTO(out, rc);
 928         OBD_FAIL_TIMEOUT(OBD_FAIL_OST_BRW_PAUSE_PACK, obd_fail_val);
 929         rcs = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF + 1,
 930                              niocount * sizeof(*rcs));
 931
 932         /*
 933          * Per-thread array of struct niobuf_{local,remote}'s was allocated by
 934          * ost_thread_init().
 935          */
 936         local_nb = ost_tls(req)->local;
 937
 938         rc = ost_brw_lock_get(LCK_PW, exp, ioo, remote_nb, &lockh);
 939         if (rc != 0)
 940                 GOTO(out_bulk, rc);
 941
 942         /*
 943          * If getting the lock took more time than
 944          * client was willing to wait, drop it. b=11330
 945          */
 946         if (cfs_time_current_sec() > req->rq_deadline ||
 947             OBD_FAIL_CHECK(OBD_FAIL_OST_DROP_REQ)) {
 948                 no_reply = 1;
 949                 CERROR("Dropping timed-out write from %s because locking "
 950                        "object "LPX64" took %ld seconds (limit was %ld).\n",
 951                        libcfs_id2str(req->rq_peer), ioo->ioo_id,
 952                        cfs_time_current_sec() - req->rq_arrival_time.tv_sec,
 953                        req->rq_deadline - req->rq_arrival_time.tv_sec);
 954                 GOTO(out_lock, rc = -ETIMEDOUT);
 955         }
 956
 957         ost_prolong_locks(exp, ioo, remote_nb,&body->oa,  LCK_PW);
 958
 959         /* obd_preprw clobbers oa->valid, so save what we need */
 960         if (body->oa.o_valid & OBD_MD_FLCKSUM) {
 961                 client_cksum = body->oa.o_cksum;
 962                 if (body->oa.o_valid & OBD_MD_FLFLAGS)
 963                         cksum_type = cksum_type_unpack(body->oa.o_flags);
 964         }
 965
 966         /* Because we already sync grant info with client when reconnect,
 967          * grant info will be cleared for resent req, then fed_grant and
 968          * total_grant will not be modified in following preprw_write */
 969         if (lustre_msg_get_flags(req->rq_reqmsg) & (MSG_RESENT | MSG_REPLAY)) {
 970                 DEBUG_REQ(D_CACHE, req, "clear resent/replay req grant info");
 971                 body->oa.o_valid &= ~OBD_MD_FLGRANT;
 972         }
 973
 974         npages = OST_THREAD_POOL_SIZE;
 975         rc = obd_preprw(OBD_BRW_WRITE, exp, &body->oa, objcount,
 976                         ioo, remote_nb, &npages, local_nb, oti, capa);
 977         if (rc != 0)
 978                 GOTO(out_lock, rc);
 979
 980         desc = ptlrpc_prep_bulk_exp(req, npages,
 981                                      BULK_GET_SINK, OST_BULK_PORTAL);
 982         if (desc == NULL)
 983                 GOTO(out, rc = -ENOMEM);
 984
 985         /* NB Having prepped, we must commit... */
 986
 987         for (i = 0; i < npages; i++)
 988                 ptlrpc_prep_bulk_page(desc, local_nb[i].page,
 989                                       local_nb[i].offset & ~CFS_PAGE_MASK,
 990                                       local_nb[i].len);
 991
 992         /* Check if client was evicted while we were doing i/o before touching
 993            network */
 994         if (desc->bd_export->exp_failed)
 995                 rc = -ENOTCONN;
 996         else
 997                 rc = ptlrpc_start_bulk_transfer (desc);
 998         if (rc == 0) {
 999                 time_t start = cfs_time_current_sec();
1000                 do {
1001                         long timeoutl = req->rq_deadline -
1002                                 cfs_time_current_sec();
1003                         cfs_duration_t timeout = (timeoutl <= 0 || rc) ?
1004                                 CFS_TICK : cfs_time_seconds(timeoutl);
1005                         lwi = LWI_TIMEOUT_INTERVAL(timeout, cfs_time_seconds(1),
1006                                                    ost_bulk_timeout, desc);
1007                         rc = l_wait_event(desc->bd_waitq,
1008                                           !ptlrpc_bulk_active(desc) ||
1009                                           desc->bd_export->exp_failed, &lwi);
1010                         LASSERT(rc == 0 || rc == -ETIMEDOUT);
1011                         /* Wait again if we changed deadline */
1012                 } while ((rc == -ETIMEDOUT) &&
1013                          (req->rq_deadline > cfs_time_current_sec()));
1014
1015                 if (rc == -ETIMEDOUT) {
1016                         DEBUG_REQ(D_ERROR, req,
1017                                   "timeout on bulk GET after %ld%+lds",
1018                                   req->rq_deadline - start,
1019                                   cfs_time_current_sec() -
1020                                   req->rq_deadline);
1021                         ptlrpc_abort_bulk(desc);
1022                 } else if (desc->bd_export->exp_failed) {
1023                         DEBUG_REQ(D_ERROR, req, "Eviction on bulk GET");
1024                         rc = -ENOTCONN;
1025                         ptlrpc_abort_bulk(desc);
1026                 } else if (!desc->bd_success ||
1027                            desc->bd_nob_transferred != desc->bd_nob) {
1028                         DEBUG_REQ(D_ERROR, req, "%s bulk GET %d(%d)",
1029                                   desc->bd_success ?
1030                                   "truncated" : "network error on",
1031                                   desc->bd_nob_transferred, desc->bd_nob);
1032                         /* XXX should this be a different errno? */
1033                         rc = -ETIMEDOUT;
1034                 }
1035         } else {
1036                 DEBUG_REQ(D_ERROR, req, "ptlrpc_bulk_get failed: rc %d", rc);
1037         }
1038         no_reply = rc != 0;
1039
1040         if (rc == 0)
1041                 sptlrpc_svc_unwrap_bulk(req, desc);
1042
1043         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
1044                                  sizeof(*repbody));
1045         memcpy(&repbody->oa, &body->oa, sizeof(repbody->oa));
1046
1047         if (unlikely(client_cksum != 0 && rc == 0)) {
1048                 static int cksum_counter;
1049                 repbody->oa.o_valid |= OBD_MD_FLCKSUM | OBD_MD_FLFLAGS;
1050                 repbody->oa.o_flags &= ~OBD_FL_CKSUM_ALL;
1051                 repbody->oa.o_flags |= cksum_type_pack(cksum_type);
1052                 server_cksum = ost_checksum_bulk(desc, OST_WRITE, cksum_type);
1053                 repbody->oa.o_cksum = server_cksum;
1054                 cksum_counter++;
1055                 if (unlikely(client_cksum != server_cksum)) {
1056                         CERROR("client csum %x, server csum %x\n",
1057                                client_cksum, server_cksum);
1058                         cksum_counter = 0;
1059                 } else if ((cksum_counter & (-cksum_counter)) == cksum_counter){
1060                         CDEBUG(D_INFO, "Checksum %u from %s OK: %x\n",
1061                                cksum_counter, libcfs_id2str(req->rq_peer),
1062                                server_cksum);
1063                 }
1064         }
1065
1066         /* Must commit after prep above in all cases */
1067         rc = obd_commitrw(OBD_BRW_WRITE, exp, &repbody->oa, objcount, ioo,
1068                           remote_nb, npages, local_nb, oti, rc);
1069
1070         if (unlikely(client_cksum != server_cksum && rc == 0)) {
1071                 int  new_cksum = ost_checksum_bulk(desc, OST_WRITE, cksum_type);
1072                 char *msg;
1073                 char *via;
1074                 char *router;
1075
1076                 if (new_cksum == server_cksum)
1077                         msg = "changed in transit before arrival at OST";
1078                 else if (new_cksum == client_cksum)
1079                         msg = "initial checksum before message complete";
1080                 else
1081                         msg = "changed in transit AND after initial checksum";
1082
1083                 if (req->rq_peer.nid == desc->bd_sender) {
1084                         via = router = "";
1085                 } else {
1086                         via = " via ";
1087                         router = libcfs_nid2str(desc->bd_sender);
1088                 }
1089
1090                 LCONSOLE_ERROR_MSG(0x168, "%s: BAD WRITE CHECKSUM: %s from "
1091                                    "%s%s%s inum "LPU64"/"LPU64" object "
1092                                    LPU64"/"LPU64" extent ["LPU64"-"LPU64"]\n",
1093                                    exp->exp_obd->obd_name, msg,
1094                                    libcfs_id2str(req->rq_peer),
1095                                    via, router,
1096                                    body->oa.o_valid & OBD_MD_FLFID ?
1097                                                 body->oa.o_fid : (__u64)0,
1098                                    body->oa.o_valid & OBD_MD_FLFID ?
1099                                                 body->oa.o_generation :(__u64)0,
1100                                    body->oa.o_id,
1101                                    body->oa.o_valid & OBD_MD_FLGROUP ?
1102                                                 body->oa.o_gr : (__u64)0,
1103                                    local_nb[0].offset,
1104                                    local_nb[npages-1].offset +
1105                                    local_nb[npages-1].len - 1 );
1106                 CERROR("client csum %x, original server csum %x, "
1107                        "server csum now %x\n",
1108                        client_cksum, server_cksum, new_cksum);
1109         }
1110
1111         if (rc == 0) {
1112                 int nob = 0;
1113
1114                 /* set per-requested niobuf return codes */
1115                 for (i = j = 0; i < niocount; i++) {
1116                         int len = remote_nb[i].len;
1117
1118                         nob += len;
1119                         rcs[i] = 0;
1120                         do {
1121                                 LASSERT(j < npages);
1122                                 if (local_nb[j].rc < 0)
1123                                         rcs[i] = local_nb[j].rc;
1124                                 len -= local_nb[j].len;
1125                                 j++;
1126                         } while (len > 0);
1127                         LASSERT(len == 0);
1128                 }
1129                 LASSERT(j == npages);
1130                 ptlrpc_lprocfs_brw(req, nob);
1131         }
1132
1133 out_lock:
1134         ost_brw_lock_put(LCK_PW, ioo, remote_nb, &lockh);
1135 out_bulk:
1136         if (desc)
1137                 ptlrpc_free_bulk(desc);
1138 out:
1139         if (rc == 0) {
1140                 oti_to_request(oti, req);
1141                 target_committed_to_req(req);
1142                 rc = ptlrpc_reply(req);
1143         } else if (!no_reply) {
1144                 /* Only reply if there was no comms problem with bulk */
1145                 target_committed_to_req(req);
1146                 req->rq_status = rc;
1147                 ptlrpc_error(req);
1148         } else {
1149                 /* reply out callback would free */
1150                 ptlrpc_req_drop_rs(req);
1151                 CWARN("%s: ignoring bulk IO comm error with %s@%s id %s - "
1152                       "client will retry\n",
1153                       exp->exp_obd->obd_name,
1154                       exp->exp_client_uuid.uuid,
1155                       exp->exp_connection->c_remote_uuid.uuid,
1156                       libcfs_id2str(req->rq_peer));
1157         }
1158         RETURN(rc);
1159 }
1160
1161 static int ost_set_info(struct obd_export *exp, struct ptlrpc_request *req)
1162 {
1163         char *key, *val = NULL;
1164         int keylen, vallen, rc = 0;
1165         ENTRY;
1166
1167         key = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, 1);
1168         if (key == NULL) {
1169                 DEBUG_REQ(D_HA, req, "no set_info key");
1170                 RETURN(-EFAULT);
1171         }
1172         keylen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF);
1173
1174         rc = lustre_pack_reply(req, 1, NULL, NULL);
1175         if (rc)
1176                 RETURN(rc);
1177
1178         vallen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF + 1);
1179         if (vallen)
1180                 val = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 1, 0);
1181
1182         if (KEY_IS(KEY_EVICT_BY_NID)) {
1183                 if (val && vallen)
1184                         obd_export_evict_by_nid(exp->exp_obd, val);
1185
1186                 GOTO(out, rc = 0);
1187         }
1188
1189         rc = obd_set_info_async(exp, keylen, key, vallen, val, NULL);
1190 out:
1191         lustre_msg_set_status(req->rq_repmsg, 0);
1192         RETURN(rc);
1193 }
1194
1195 static int ost_get_info(struct obd_export *exp, struct ptlrpc_request *req)
1196 {
1197         void *key, *reply;
1198         int keylen, replylen, rc = 0;
1199         struct req_capsule *pill = &req->rq_pill;
1200         ENTRY;
1201
1202         req_capsule_set(&req->rq_pill, &RQF_OST_GET_INFO_GENERIC);
1203
1204         /* this common part for get_info rpc */
1205         key = req_capsule_client_get(pill, &RMF_SETINFO_KEY);
1206         if (key == NULL) {
1207                 DEBUG_REQ(D_HA, req, "no get_info key");
1208                 RETURN(-EFAULT);
1209         }
1210         keylen = req_capsule_get_size(pill, &RMF_SETINFO_KEY, RCL_CLIENT);
1211
1212         rc = obd_get_info(exp, keylen, key, &replylen, NULL, NULL);
1213         if (rc)
1214                 RETURN(rc);
1215
1216         req_capsule_set_size(pill, &RMF_GENERIC_DATA,
1217                              RCL_SERVER, replylen);
1218
1219         rc = req_capsule_server_pack(pill);
1220         if (rc)
1221                 RETURN(rc);
1222
1223         reply = req_capsule_server_get(pill, &RMF_GENERIC_DATA);
1224         if (reply == NULL)
1225                 RETURN(-ENOMEM);
1226
1227         /* call again to fill in the reply buffer */
1228         rc = obd_get_info(exp, keylen, key, &replylen, reply, NULL);
1229
1230         lustre_msg_set_status(req->rq_repmsg, 0);
1231         RETURN(rc);
1232 }
1233
1234 static int ost_handle_quotactl(struct ptlrpc_request *req)
1235 {
1236         struct obd_quotactl *oqctl, *repoqc;
1237         __u32 size[2] = { sizeof(struct ptlrpc_body), sizeof(*repoqc) };
1238         int rc;
1239         ENTRY;
1240
1241         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1242                                    lustre_swab_obd_quotactl);
1243         if (oqctl == NULL)
1244                 GOTO(out, rc = -EPROTO);
1245
1246         rc = lustre_pack_reply(req, 2, size, NULL);
1247         if (rc)
1248                 GOTO(out, rc);
1249
1250         repoqc = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*repoqc));
1251
1252         req->rq_status = obd_quotactl(req->rq_export, oqctl);
1253         *repoqc = *oqctl;
1254 out:
1255         RETURN(rc);
1256 }
1257
1258 static int ost_handle_quotacheck(struct ptlrpc_request *req)
1259 {
1260         struct obd_quotactl *oqctl;
1261         int rc;
1262         ENTRY;
1263
1264         oqctl = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1265         if (oqctl == NULL)
1266                 RETURN(-EPROTO);
1267
1268         rc = req_capsule_server_pack(&req->rq_pill);
1269         if (rc) {
1270                 CERROR("ost: out of memory while packing quotacheck reply\n");
1271                 RETURN(-ENOMEM);
1272         }
1273
1274         req->rq_status = obd_quotacheck(req->rq_export, oqctl);
1275         RETURN(0);
1276 }
1277
1278 static int ost_llog_handle_connect(struct obd_export *exp,
1279                                    struct ptlrpc_request *req)
1280 {
1281         struct llogd_conn_body *body;
1282         int rc;
1283         ENTRY;
1284
1285         body = lustre_msg_buf(req->rq_reqmsg, 1, sizeof(*body));
1286         rc = obd_llog_connect(exp, body);
1287         RETURN(rc);
1288 }
1289
1290 static int filter_export_check_flavor(struct filter_obd *filter,
1291                                       struct obd_export *exp,
1292                                       struct ptlrpc_request *req)
1293 {
1294         int     rc = 0;
1295
1296         /* FIXME
1297          * this should be done in filter_connect()/filter_reconnect(), but
1298          * we can't obtain information like NID, which stored in incoming
1299          * request, thus can't decide what flavor to use. so we do it here.
1300          *
1301          * This hack should be removed after the OST stack be rewritten, just
1302          * like what we are doing in mdt_obd_connect()/mdt_obd_reconnect().
1303          */
1304         if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_INVALID)
1305                 return 0;
1306
1307         CDEBUG(D_SEC, "from %s\n", sptlrpc_part2name(req->rq_sp_from));
1308         spin_lock(&exp->exp_lock);
1309         exp->exp_sp_peer = req->rq_sp_from;
1310
1311         read_lock(&filter->fo_sptlrpc_lock);
1312         sptlrpc_rule_set_choose(&filter->fo_sptlrpc_rset, exp->exp_sp_peer,
1313                                 req->rq_peer.nid, &exp->exp_flvr);
1314         read_unlock(&filter->fo_sptlrpc_lock);
1315
1316         if (exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
1317                 CERROR("invalid rpc flavor %x, expect %x, from %s\n",
1318                        req->rq_flvr.sf_rpc, exp->exp_flvr.sf_rpc,
1319                        libcfs_nid2str(req->rq_peer.nid));
1320                 exp->exp_flvr.sf_rpc = SPTLRPC_FLVR_INVALID;
1321                 rc = -EACCES;
1322         }
1323
1324         spin_unlock(&exp->exp_lock);
1325
1326         return rc;
1327 }
1328
1329 static int ost_filter_recovery_request(struct ptlrpc_request *req,
1330                                        struct obd_device *obd, int *process)
1331 {
1332         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1333         case OST_CONNECT: /* This will never get here, but for completeness. */
1334         case OST_DISCONNECT:
1335                *process = 1;
1336                RETURN(0);
1337
1338         case OBD_PING:
1339         case OST_CREATE:
1340         case OST_DESTROY:
1341         case OST_PUNCH:
1342         case OST_SETATTR:
1343         case OST_SYNC:
1344         case OST_WRITE:
1345         case OBD_LOG_CANCEL:
1346         case LDLM_ENQUEUE:
1347                 *process = target_queue_recovery_request(req, obd);
1348                 RETURN(0);
1349
1350         default:
1351                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1352                 *process = -EAGAIN;
1353                 RETURN(0);
1354         }
1355 }
1356
1357 int ost_msg_check_version(struct lustre_msg *msg)
1358 {
1359         int rc;
1360
1361         switch(lustre_msg_get_opc(msg)) {
1362         case OST_CONNECT:
1363         case OST_DISCONNECT:
1364         case OBD_PING:
1365         case SEC_CTX_INIT:
1366         case SEC_CTX_INIT_CONT:
1367         case SEC_CTX_FINI:
1368                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
1369                 if (rc)
1370                         CERROR("bad opc %u version %08x, expecting %08x\n",
1371                                lustre_msg_get_opc(msg),
1372                                lustre_msg_get_version(msg),
1373                                LUSTRE_OBD_VERSION);
1374                 break;
1375         case OST_CREATE:
1376         case OST_DESTROY:
1377         case OST_GETATTR:
1378         case OST_SETATTR:
1379         case OST_WRITE:
1380         case OST_READ:
1381         case OST_PUNCH:
1382         case OST_STATFS:
1383         case OST_SYNC:
1384         case OST_SET_INFO:
1385         case OST_GET_INFO:
1386         case OST_QUOTACHECK:
1387         case OST_QUOTACTL:
1388                 rc = lustre_msg_check_version(msg, LUSTRE_OST_VERSION);
1389                 if (rc)
1390                         CERROR("bad opc %u version %08x, expecting %08x\n",
1391                                lustre_msg_get_opc(msg),
1392                                lustre_msg_get_version(msg),
1393                                LUSTRE_OST_VERSION);
1394                 break;
1395         case LDLM_ENQUEUE:
1396         case LDLM_CONVERT:
1397         case LDLM_CANCEL:
1398         case LDLM_BL_CALLBACK:
1399         case LDLM_CP_CALLBACK:
1400                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
1401                 if (rc)
1402                         CERROR("bad opc %u version %08x, expecting %08x\n",
1403                                lustre_msg_get_opc(msg),
1404                                lustre_msg_get_version(msg),
1405                                LUSTRE_DLM_VERSION);
1406                 break;
1407         case LLOG_ORIGIN_CONNECT:
1408         case OBD_LOG_CANCEL:
1409                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
1410                 if (rc)
1411                         CERROR("bad opc %u version %08x, expecting %08x\n",
1412                                lustre_msg_get_opc(msg),
1413                                lustre_msg_get_version(msg),
1414                                LUSTRE_LOG_VERSION);
1415                 break;
1416         default:
1417                 CERROR("Unexpected opcode %d\n", lustre_msg_get_opc(msg));
1418                 rc = -ENOTSUPP;
1419         }
1420         return rc;
1421 }
1422
1423 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
1424 int ost_handle(struct ptlrpc_request *req)
1425 {
1426         struct obd_trans_info trans_info = { 0, };
1427         struct obd_trans_info *oti = &trans_info;
1428         int should_process, fail = OBD_FAIL_OST_ALL_REPLY_NET, rc = 0;
1429         struct obd_device *obd = NULL;
1430         ENTRY;
1431
1432         LASSERT(current->journal_info == NULL);
1433
1434         /* primordial rpcs don't affect server recovery */
1435         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1436         case SEC_CTX_INIT:
1437         case SEC_CTX_INIT_CONT:
1438         case SEC_CTX_FINI:
1439                 GOTO(out, rc = 0);
1440         }
1441
1442         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
1443
1444         /* XXX identical to MDS */
1445         if (lustre_msg_get_opc(req->rq_reqmsg) != OST_CONNECT) {
1446                 int recovering;
1447
1448                 if (req->rq_export == NULL) {
1449                         CDEBUG(D_HA,"operation %d on unconnected OST from %s\n",
1450                                lustre_msg_get_opc(req->rq_reqmsg),
1451                                libcfs_id2str(req->rq_peer));
1452                         req->rq_status = -ENOTCONN;
1453                         GOTO(out, rc = -ENOTCONN);
1454                 }
1455
1456                 obd = req->rq_export->exp_obd;
1457
1458                 /* Check for aborted recovery. */
1459                 spin_lock_bh(&obd->obd_processing_task_lock);
1460                 recovering = obd->obd_recovering;
1461                 spin_unlock_bh(&obd->obd_processing_task_lock);
1462                 if (recovering) {
1463                         rc = ost_filter_recovery_request(req, obd,
1464                                                          &should_process);
1465                         if (rc || !should_process)
1466                                 RETURN(rc);
1467                         else if (should_process < 0) {
1468                                 req->rq_status = should_process;
1469                                 rc = ptlrpc_error(req);
1470                                 RETURN(rc);
1471                         }
1472                 }
1473         }
1474
1475         oti_init(oti, req);
1476
1477         rc = ost_msg_check_version(req->rq_reqmsg);
1478         if (rc)
1479                 RETURN(rc);
1480
1481         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1482         case OST_CONNECT: {
1483                 CDEBUG(D_INODE, "connect\n");
1484                 req_capsule_set(&req->rq_pill, &RQF_OST_CONNECT);
1485                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CONNECT_NET))
1486                         RETURN(0);
1487                 rc = target_handle_connect(req);
1488                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CONNECT_NET2))
1489                         RETURN(0);
1490                 if (!rc) {
1491                         struct obd_export *exp = req->rq_export;
1492
1493                         obd = exp->exp_obd;
1494
1495                         rc = filter_export_check_flavor(&obd->u.filter,
1496                                                         exp, req);
1497                 }
1498                 break;
1499         }
1500         case OST_DISCONNECT:
1501                 CDEBUG(D_INODE, "disconnect\n");
1502                 req_capsule_set(&req->rq_pill, &RQF_OST_DISCONNECT);
1503                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_DISCONNECT_NET))
1504                         RETURN(0);
1505                 rc = target_handle_disconnect(req);
1506                 break;
1507         case OST_CREATE:
1508                 CDEBUG(D_INODE, "create\n");
1509                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CREATE_NET))
1510                         RETURN(0);
1511                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
1512                         GOTO(out, rc = -ENOSPC);
1513                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1514                         GOTO(out, rc = -EROFS);
1515                 rc = ost_create(req->rq_export, req, oti);
1516                 break;
1517         case OST_DESTROY:
1518                 CDEBUG(D_INODE, "destroy\n");
1519                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_DESTROY_NET))
1520                         RETURN(0);
1521                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1522                         GOTO(out, rc = -EROFS);
1523                 rc = ost_destroy(req->rq_export, req, oti);
1524                 break;
1525         case OST_GETATTR:
1526                 CDEBUG(D_INODE, "getattr\n");
1527                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_GETATTR_NET))
1528                         RETURN(0);
1529                 rc = ost_getattr(req->rq_export, req);
1530                 break;
1531         case OST_SETATTR:
1532                 CDEBUG(D_INODE, "setattr\n");
1533                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_NET))
1534                         RETURN(0);
1535                 rc = ost_setattr(req->rq_export, req, oti);
1536                 break;
1537         case OST_WRITE:
1538                 CDEBUG(D_INODE, "write\n");
1539                 /* req->rq_request_portal would be nice, if it was set */
1540                 if (req->rq_rqbd->rqbd_service->srv_req_portal !=OST_IO_PORTAL){
1541                         CERROR("%s: deny write request from %s to portal %u\n",
1542                                req->rq_export->exp_obd->obd_name,
1543                                obd_export_nid2str(req->rq_export),
1544                                req->rq_rqbd->rqbd_service->srv_req_portal);
1545                         GOTO(out, rc = -EPROTO);
1546                 }
1547                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_NET))
1548                         RETURN(0);
1549                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
1550                         GOTO(out, rc = -ENOSPC);
1551                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1552                         GOTO(out, rc = -EROFS);
1553                 rc = ost_brw_write(req, oti);
1554                 LASSERT(current->journal_info == NULL);
1555                 /* ost_brw_write sends its own replies */
1556                 RETURN(rc);
1557         case OST_READ:
1558                 CDEBUG(D_INODE, "read\n");
1559                 /* req->rq_request_portal would be nice, if it was set */
1560                 if (req->rq_rqbd->rqbd_service->srv_req_portal !=OST_IO_PORTAL){
1561                         CERROR("%s: deny read request from %s to portal %u\n",
1562                                req->rq_export->exp_obd->obd_name,
1563                                obd_export_nid2str(req->rq_export),
1564                                req->rq_rqbd->rqbd_service->srv_req_portal);
1565                         GOTO(out, rc = -EPROTO);
1566                 }
1567                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_NET))
1568                         RETURN(0);
1569                 rc = ost_brw_read(req, oti);
1570                 LASSERT(current->journal_info == NULL);
1571                 /* ost_brw_read sends its own replies */
1572                 RETURN(rc);
1573         case OST_PUNCH:
1574                 CDEBUG(D_INODE, "punch\n");
1575                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_PUNCH_NET))
1576                         RETURN(0);
1577                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1578                         GOTO(out, rc = -EROFS);
1579                 rc = ost_punch(req->rq_export, req, oti);
1580                 break;
1581         case OST_STATFS:
1582                 CDEBUG(D_INODE, "statfs\n");
1583                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_STATFS_NET))
1584                         RETURN(0);
1585                 rc = ost_statfs(req);
1586                 break;
1587         case OST_SYNC:
1588                 CDEBUG(D_INODE, "sync\n");
1589                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_SYNC_NET))
1590                         RETURN(0);
1591                 rc = ost_sync(req->rq_export, req);
1592                 break;
1593         case OST_SET_INFO:
1594                 DEBUG_REQ(D_INODE, req, "set_info");
1595                 req_capsule_set(&req->rq_pill, &RQF_OST_SET_INFO);
1596                 rc = ost_set_info(req->rq_export, req);
1597                 break;
1598         case OST_GET_INFO:
1599                 DEBUG_REQ(D_INODE, req, "get_info");
1600                 rc = ost_get_info(req->rq_export, req);
1601                 break;
1602         case OST_QUOTACHECK:
1603                 CDEBUG(D_INODE, "quotacheck\n");
1604                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTACHECK);
1605                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_QUOTACHECK_NET))
1606                         RETURN(0);
1607                 rc = ost_handle_quotacheck(req);
1608                 break;
1609         case OST_QUOTACTL:
1610                 CDEBUG(D_INODE, "quotactl\n");
1611                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTACTL);
1612                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_QUOTACTL_NET))
1613                         RETURN(0);
1614                 rc = ost_handle_quotactl(req);
1615                 break;
1616         case OBD_PING:
1617                 DEBUG_REQ(D_INODE, req, "ping");
1618                 req_capsule_set(&req->rq_pill, &RQF_OBD_PING);
1619                 rc = target_handle_ping(req);
1620                 break;
1621         /* FIXME - just reply status */
1622         case LLOG_ORIGIN_CONNECT:
1623                 DEBUG_REQ(D_INODE, req, "log connect");
1624                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_CONNECT);
1625                 rc = ost_llog_handle_connect(req->rq_export, req);
1626                 req->rq_status = rc;
1627                 rc = req_capsule_server_pack(&req->rq_pill);
1628                 if (rc)
1629                         RETURN(rc);
1630                 RETURN(ptlrpc_reply(req));
1631         case OBD_LOG_CANCEL:
1632                 CDEBUG(D_INODE, "log cancel\n");
1633                 req_capsule_set(&req->rq_pill, &RQF_LOG_CANCEL);
1634                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
1635                         RETURN(0);
1636                 rc = llog_origin_handle_cancel(req);
1637                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_REP))
1638                         RETURN(0);
1639                 req->rq_status = rc;
1640                 rc = req_capsule_server_pack(&req->rq_pill);
1641                 if (rc)
1642                         RETURN(rc);
1643                 RETURN(ptlrpc_reply(req));
1644         case LDLM_ENQUEUE:
1645                 CDEBUG(D_INODE, "enqueue\n");
1646                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);
1647                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE))
1648                         RETURN(0);
1649                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
1650                                          ldlm_server_blocking_ast,
1651                                          ldlm_server_glimpse_ast);
1652                 fail = OBD_FAIL_OST_LDLM_REPLY_NET;
1653                 break;
1654         case LDLM_CONVERT:
1655                 CDEBUG(D_INODE, "convert\n");
1656                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CONVERT);
1657                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CONVERT))
1658                         RETURN(0);
1659                 rc = ldlm_handle_convert(req);
1660                 break;
1661         case LDLM_CANCEL:
1662                 CDEBUG(D_INODE, "cancel\n");
1663                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
1664                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL))
1665                         RETURN(0);
1666                 rc = ldlm_handle_cancel(req);
1667                 break;
1668         case LDLM_BL_CALLBACK:
1669         case LDLM_CP_CALLBACK:
1670                 CDEBUG(D_INODE, "callback\n");
1671                 CERROR("callbacks should not happen on OST\n");
1672                 /* fall through */
1673         default:
1674                 CERROR("Unexpected opcode %d\n",
1675                        lustre_msg_get_opc(req->rq_reqmsg));
1676                 req->rq_status = -ENOTSUPP;
1677                 rc = ptlrpc_error(req);
1678                 RETURN(rc);
1679         }
1680
1681         LASSERT(current->journal_info == NULL);
1682
1683         EXIT;
1684         /* If we're DISCONNECTing, the export_data is already freed */
1685         if (!rc && lustre_msg_get_opc(req->rq_reqmsg) != OST_DISCONNECT)
1686                 target_committed_to_req(req);
1687
1688 out:
1689         if (!rc)
1690                 oti_to_request(oti, req);
1691
1692         target_send_reply(req, rc, fail);
1693         return 0;
1694 }
1695 EXPORT_SYMBOL(ost_handle);
1696 /*
1697  * free per-thread pool created by ost_thread_init().
1698  */
1699 static void ost_thread_done(struct ptlrpc_thread *thread)
1700 {
1701         struct ost_thread_local_cache *tls; /* TLS stands for Thread-Local
1702                                              * Storage */
1703
1704         ENTRY;
1705
1706         LASSERT(thread != NULL);
1707
1708         /*
1709          * be prepared to handle partially-initialized pools (because this is
1710          * called from ost_thread_init() for cleanup.
1711          */
1712         tls = thread->t_data;
1713         if (tls != NULL) {
1714                 OBD_FREE_PTR(tls);
1715                 thread->t_data = NULL;
1716         }
1717         EXIT;
1718 }
1719
1720 /*
1721  * initialize per-thread page pool (bug 5137).
1722  */
1723 static int ost_thread_init(struct ptlrpc_thread *thread)
1724 {
1725         struct ost_thread_local_cache *tls;
1726
1727         ENTRY;
1728
1729         LASSERT(thread != NULL);
1730         LASSERT(thread->t_data == NULL);
1731         LASSERTF(thread->t_id <= OSS_THREADS_MAX, "%u\n", thread->t_id);
1732
1733         OBD_ALLOC_PTR(tls);
1734         if (tls == NULL)
1735                 RETURN(-ENOMEM);
1736         thread->t_data = tls;
1737         RETURN(0);
1738 }
1739
1740 #define OST_WATCHDOG_TIMEOUT (obd_timeout * 1000)
1741
1742 /* Sigh - really, this is an OSS, the _server_, not the _target_ */
1743 static int ost_setup(struct obd_device *obd, struct lustre_cfg* lcfg)
1744 {
1745         struct ost_obd *ost = &obd->u.ost;
1746         struct lprocfs_static_vars lvars;
1747         int oss_min_threads;
1748         int oss_max_threads;
1749         int oss_min_create_threads;
1750         int oss_max_create_threads;
1751         int rc;
1752         ENTRY;
1753
1754         rc = cleanup_group_info();
1755         if (rc)
1756                 RETURN(rc);
1757
1758         lprocfs_ost_init_vars(&lvars);
1759         lprocfs_obd_setup(obd, lvars.obd_vars);
1760
1761         sema_init(&ost->ost_health_sem, 1);
1762
1763         if (oss_num_threads) {
1764                 /* If oss_num_threads is set, it is the min and the max. */
1765                 if (oss_num_threads > OSS_THREADS_MAX)
1766                         oss_num_threads = OSS_THREADS_MAX;
1767                 if (oss_num_threads < OSS_THREADS_MIN)
1768                         oss_num_threads = OSS_THREADS_MIN;
1769                 oss_max_threads = oss_min_threads = oss_num_threads;
1770         } else {
1771                 /* Base min threads on memory and cpus */
1772                 oss_min_threads = num_possible_cpus() * num_physpages >>
1773                         (27 - CFS_PAGE_SHIFT);
1774                 if (oss_min_threads < OSS_THREADS_MIN)
1775                         oss_min_threads = OSS_THREADS_MIN;
1776                 /* Insure a 4x range for dynamic threads */
1777                 if (oss_min_threads > OSS_THREADS_MAX / 4)
1778                         oss_min_threads = OSS_THREADS_MAX / 4;
1779                 oss_max_threads = min(OSS_THREADS_MAX, oss_min_threads * 4);
1780         }
1781
1782         ost->ost_service =
1783                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
1784                                 OST_MAXREPSIZE, OST_REQUEST_PORTAL,
1785                                 OSC_REPLY_PORTAL, OSS_SERVICE_WATCHDOG_FACTOR,
1786                                 ost_handle, LUSTRE_OSS_NAME,
1787                                 obd->obd_proc_entry, target_print_req,
1788                                 oss_min_threads, oss_max_threads,
1789                                 "ll_ost", LCT_DT_THREAD);
1790         if (ost->ost_service == NULL) {
1791                 CERROR("failed to start service\n");
1792                 GOTO(out_lprocfs, rc = -ENOMEM);
1793         }
1794
1795         rc = ptlrpc_start_threads(obd, ost->ost_service);
1796         if (rc)
1797                 GOTO(out_service, rc = -EINVAL);
1798
1799         if (oss_num_create_threads) {
1800                 if (oss_num_create_threads > OSS_MAX_CREATE_THREADS)
1801                         oss_num_create_threads = OSS_MAX_CREATE_THREADS;
1802                 if (oss_num_create_threads < OSS_MIN_CREATE_THREADS)
1803                         oss_num_create_threads = OSS_MIN_CREATE_THREADS;
1804                 oss_min_create_threads = oss_max_create_threads =
1805                         oss_num_create_threads;
1806         } else {
1807                 oss_min_create_threads = OSS_MIN_CREATE_THREADS;
1808                 oss_max_create_threads = OSS_MAX_CREATE_THREADS;
1809         }
1810
1811         ost->ost_create_service =
1812                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
1813                                 OST_MAXREPSIZE, OST_CREATE_PORTAL,
1814                                 OSC_REPLY_PORTAL, OSS_SERVICE_WATCHDOG_FACTOR,
1815                                 ost_handle, "ost_create",
1816                                 obd->obd_proc_entry, target_print_req,
1817                                 oss_min_create_threads, oss_max_create_threads,
1818                                 "ll_ost_creat", LCT_DT_THREAD);
1819         if (ost->ost_create_service == NULL) {
1820                 CERROR("failed to start OST create service\n");
1821                 GOTO(out_service, rc = -ENOMEM);
1822         }
1823
1824         rc = ptlrpc_start_threads(obd, ost->ost_create_service);
1825         if (rc)
1826                 GOTO(out_create, rc = -EINVAL);
1827
1828         ost->ost_io_service =
1829                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
1830                                 OST_MAXREPSIZE, OST_IO_PORTAL,
1831                                 OSC_REPLY_PORTAL, OSS_SERVICE_WATCHDOG_FACTOR,
1832                                 ost_handle, "ost_io",
1833                                 obd->obd_proc_entry, target_print_req,
1834                                 oss_min_threads, oss_max_threads,
1835                                 "ll_ost_io", LCT_DT_THREAD);
1836         if (ost->ost_io_service == NULL) {
1837                 CERROR("failed to start OST I/O service\n");
1838                 GOTO(out_create, rc = -ENOMEM);
1839         }
1840
1841         ost->ost_io_service->srv_init = ost_thread_init;
1842         ost->ost_io_service->srv_done = ost_thread_done;
1843         ost->ost_io_service->srv_cpu_affinity = 1;
1844         rc = ptlrpc_start_threads(obd, ost->ost_io_service);
1845         if (rc)
1846                 GOTO(out_io, rc = -EINVAL);
1847
1848         ping_evictor_start();
1849
1850         RETURN(0);
1851
1852 out_io:
1853         ptlrpc_unregister_service(ost->ost_io_service);
1854         ost->ost_io_service = NULL;
1855 out_create:
1856         ptlrpc_unregister_service(ost->ost_create_service);
1857         ost->ost_create_service = NULL;
1858 out_service:
1859         ptlrpc_unregister_service(ost->ost_service);
1860         ost->ost_service = NULL;
1861 out_lprocfs:
1862         lprocfs_obd_cleanup(obd);
1863         RETURN(rc);
1864 }
1865
1866 static int ost_cleanup(struct obd_device *obd)
1867 {
1868         struct ost_obd *ost = &obd->u.ost;
1869         int err = 0;
1870         ENTRY;
1871
1872         ping_evictor_stop();
1873
1874         spin_lock_bh(&obd->obd_processing_task_lock);
1875         if (obd->obd_recovering) {
1876                 target_cancel_recovery_timer(obd);
1877                 obd->obd_recovering = 0;
1878         }
1879         spin_unlock_bh(&obd->obd_processing_task_lock);
1880
1881         down(&ost->ost_health_sem);
1882         ptlrpc_unregister_service(ost->ost_service);
1883         ptlrpc_unregister_service(ost->ost_create_service);
1884         ptlrpc_unregister_service(ost->ost_io_service);
1885         ost->ost_service = NULL;
1886         ost->ost_create_service = NULL;
1887         up(&ost->ost_health_sem);
1888
1889         lprocfs_obd_cleanup(obd);
1890
1891         RETURN(err);
1892 }
1893
1894 static int ost_health_check(struct obd_device *obd)
1895 {
1896         struct ost_obd *ost = &obd->u.ost;
1897         int rc = 0;
1898
1899         down(&ost->ost_health_sem);
1900         rc |= ptlrpc_service_health_check(ost->ost_service);
1901         rc |= ptlrpc_service_health_check(ost->ost_create_service);
1902         rc |= ptlrpc_service_health_check(ost->ost_io_service);
1903         up(&ost->ost_health_sem);
1904
1905         /*
1906          * health_check to return 0 on healthy
1907          * and 1 on unhealthy.
1908          */
1909         if( rc != 0)
1910                 rc = 1;
1911
1912         return rc;
1913 }
1914
1915 struct ost_thread_local_cache *ost_tls(struct ptlrpc_request *r)
1916 {
1917         return (struct ost_thread_local_cache *)(r->rq_svc_thread->t_data);
1918 }
1919
1920 /* use obd ops to offer management infrastructure */
1921 static struct obd_ops ost_obd_ops = {
1922         .o_owner        = THIS_MODULE,
1923         .o_setup        = ost_setup,
1924         .o_cleanup      = ost_cleanup,
1925         .o_health_check = ost_health_check,
1926 };
1927
1928
1929 static int __init ost_init(void)
1930 {
1931         struct lprocfs_static_vars lvars;
1932         int rc;
1933         ENTRY;
1934
1935         lprocfs_ost_init_vars(&lvars);
1936         rc = class_register_type(&ost_obd_ops, NULL, lvars.module_vars,
1937                                  LUSTRE_OSS_NAME, NULL);
1938
1939         if (ost_num_threads != 0 && oss_num_threads == 0) {
1940                 LCONSOLE_INFO("ost_num_threads module parameter is deprecated, "
1941                               "use oss_num_threads instead or unset both for "
1942                               "dynamic thread startup\n");
1943                 oss_num_threads = ost_num_threads;
1944         }
1945
1946         RETURN(rc);
1947 }
1948
1949 static void /*__exit*/ ost_exit(void)
1950 {
1951         class_unregister_type(LUSTRE_OSS_NAME);
1952 }
1953
1954 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1955 MODULE_DESCRIPTION("Lustre Object Storage Target (OST) v0.01");
1956 MODULE_LICENSE("GPL");
1957
1958 module_init(ost_init);
1959 module_exit(ost_exit);