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