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                 sptlrpc_svc_unwrap_bulk(req, desc);
1117
1118         repbody = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF,
1119                                  sizeof(*repbody));
1120         memcpy(&repbody->oa, &body->oa, sizeof(repbody->oa));
1121
1122         if (unlikely(client_cksum != 0 && rc == 0)) {
1123                 static int cksum_counter;
1124                 server_cksum = ost_checksum_bulk(desc, OST_WRITE);
1125                 repbody->oa.o_valid |= OBD_MD_FLCKSUM;
1126                 repbody->oa.o_cksum = server_cksum;
1127                 cksum_counter++;
1128                 if (unlikely(client_cksum != server_cksum)) {
1129                         CERROR("client csum %x, server csum %x\n",
1130                                client_cksum, server_cksum);
1131                         cksum_counter = 0;
1132                 } else if ((cksum_counter & (-cksum_counter)) == cksum_counter){
1133                         CDEBUG(D_INFO, "Checksum %u from %s OK: %x\n",
1134                                cksum_counter, libcfs_id2str(req->rq_peer),
1135                                server_cksum);
1136                 }
1137         }
1138
1139         /* Must commit after prep above in all cases */
1140         rc = obd_commitrw(OBD_BRW_WRITE, exp, &repbody->oa,
1141                            objcount, ioo, npages, local_nb, oti, rc);
1142
1143         if (unlikely(client_cksum != server_cksum && rc == 0)) {
1144                 int   new_cksum = ost_checksum_bulk(desc, OST_WRITE);
1145                 char *msg;
1146                 char *via;
1147                 char *router;
1148
1149                 if (new_cksum == server_cksum)
1150                         msg = "changed in transit before arrival at OST";
1151                 else if (new_cksum == client_cksum)
1152                         msg = "initial checksum before message complete";
1153                 else
1154                         msg = "changed in transit AND after initial checksum";
1155
1156                 if (req->rq_peer.nid == desc->bd_sender) {
1157                         via = router = "";
1158                 } else {
1159                         via = " via ";
1160                         router = libcfs_nid2str(desc->bd_sender);
1161                 }
1162
1163                 LCONSOLE_ERROR_MSG(0x168, "%s: BAD WRITE CHECKSUM: %s from "
1164                                    "%s%s%s inum "LPU64"/"LPU64" object "
1165                                    LPU64"/"LPU64" extent ["LPU64"-"LPU64"]\n",
1166                                    exp->exp_obd->obd_name, msg,
1167                                    libcfs_id2str(req->rq_peer),
1168                                    via, router,
1169                                    body->oa.o_valid & OBD_MD_FLFID ?
1170                                                 body->oa.o_fid : (__u64)0,
1171                                    body->oa.o_valid & OBD_MD_FLFID ?
1172                                                 body->oa.o_generation :(__u64)0,
1173                                    body->oa.o_id,
1174                                    body->oa.o_valid & OBD_MD_FLGROUP ?
1175                                                 body->oa.o_gr : (__u64)0,
1176                                    pp_rnb[0].offset,
1177                                    pp_rnb[npages-1].offset+pp_rnb[npages-1].len
1178                                    - 1 );
1179                 CERROR("client csum %x, original server csum %x, "
1180                        "server csum now %x\n",
1181                        client_cksum, server_cksum, new_cksum);
1182         }
1183
1184         ost_nio_pages_put(req, local_nb, npages);
1185
1186         if (rc == 0) {
1187                 /* set per-requested niobuf return codes */
1188                 for (i = j = 0; i < niocount; i++) {
1189                         int nob = remote_nb[i].len;
1190
1191                         rcs[i] = 0;
1192                         do {
1193                                 LASSERT(j < npages);
1194                                 if (local_nb[j].rc < 0)
1195                                         rcs[i] = local_nb[j].rc;
1196                                 nob -= pp_rnb[j].len;
1197                                 j++;
1198                         } while (nob > 0);
1199                         LASSERT(nob == 0);
1200                 }
1201                 LASSERT(j == npages);
1202         }
1203
1204 out_lock:
1205         ost_brw_lock_put(LCK_PW, ioo, pp_rnb, &lockh);
1206 out_bulk:
1207         ptlrpc_free_bulk(desc);
1208 out:
1209         if (rc == 0) {
1210                 oti_to_request(oti, req);
1211                 target_committed_to_req(req);
1212                 rc = ptlrpc_reply(req);
1213         } else if (!no_reply) {
1214                 /* Only reply if there was no comms problem with bulk */
1215                 target_committed_to_req(req);
1216                 req->rq_status = rc;
1217                 ptlrpc_error(req);
1218         } else {
1219                 if (req->rq_reply_state != NULL) {
1220                         /* reply out callback would free */
1221                         ptlrpc_rs_decref(req->rq_reply_state);
1222                         req->rq_reply_state = NULL;
1223                 }
1224                 CWARN("%s: ignoring bulk IO comm error with %s@%s id %s - "
1225                       "client will retry\n",
1226                       exp->exp_obd->obd_name,
1227                       exp->exp_client_uuid.uuid,
1228                       exp->exp_connection->c_remote_uuid.uuid,
1229                       libcfs_id2str(req->rq_peer));
1230         }
1231         RETURN(rc);
1232 }
1233
1234 static int ost_set_info(struct obd_export *exp, struct ptlrpc_request *req)
1235 {
1236         char *key, *val = NULL;
1237         int keylen, vallen, rc = 0;
1238         ENTRY;
1239
1240         key = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, 1);
1241         if (key == NULL) {
1242                 DEBUG_REQ(D_HA, req, "no set_info key");
1243                 RETURN(-EFAULT);
1244         }
1245         keylen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF);
1246
1247         rc = lustre_pack_reply(req, 1, NULL, NULL);
1248         if (rc)
1249                 RETURN(rc);
1250
1251         vallen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF + 1);
1252         if (vallen)
1253                 val = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF + 1, 0);
1254
1255         if (KEY_IS("evict_by_nid")) {
1256                 if (val && vallen)
1257                         obd_export_evict_by_nid(exp->exp_obd, val);
1258
1259                 GOTO(out, rc = 0);
1260         }
1261
1262         rc = obd_set_info_async(exp, keylen, key, vallen, val, NULL);
1263 out:
1264         lustre_msg_set_status(req->rq_repmsg, 0);
1265         RETURN(rc);
1266 }
1267
1268 static int ost_get_info(struct obd_export *exp, struct ptlrpc_request *req)
1269 {
1270         void *key, *reply;
1271         int keylen, rc = 0;
1272         int size[2] = { sizeof(struct ptlrpc_body), 0 };
1273         ENTRY;
1274
1275         key = lustre_msg_buf(req->rq_reqmsg, REQ_REC_OFF, 1);
1276         if (key == NULL) {
1277                 DEBUG_REQ(D_HA, req, "no get_info key");
1278                 RETURN(-EFAULT);
1279         }
1280         keylen = lustre_msg_buflen(req->rq_reqmsg, REQ_REC_OFF);
1281
1282         /* call once to get the size to allocate the reply buffer */
1283         rc = obd_get_info(exp, keylen, key, &size[1], NULL);
1284         if (rc)
1285                 RETURN(rc);
1286
1287         rc = lustre_pack_reply(req, 2, size, NULL);
1288         if (rc)
1289                 RETURN(rc);
1290
1291         reply = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*reply));
1292         /* call again to fill in the reply buffer */
1293         rc = obd_get_info(exp, keylen, key, size, reply);
1294         lustre_msg_set_status(req->rq_repmsg, 0);
1295
1296         RETURN(rc);
1297 }
1298
1299 static int ost_handle_quotactl(struct ptlrpc_request *req)
1300 {
1301         struct obd_quotactl *oqctl, *repoqc;
1302         int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*repoqc) };
1303         ENTRY;
1304
1305         oqctl = lustre_swab_reqbuf(req, REQ_REC_OFF, sizeof(*oqctl),
1306                                    lustre_swab_obd_quotactl);
1307         if (oqctl == NULL)
1308                 GOTO(out, rc = -EPROTO);
1309
1310         rc = lustre_pack_reply(req, 2, size, NULL);
1311         if (rc)
1312                 GOTO(out, rc);
1313
1314         repoqc = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*repoqc));
1315
1316         req->rq_status = obd_quotactl(req->rq_export, oqctl);
1317         *repoqc = *oqctl;
1318 out:
1319         RETURN(rc);
1320 }
1321
1322 static int ost_handle_quotacheck(struct ptlrpc_request *req)
1323 {
1324         struct obd_quotactl *oqctl;
1325         int rc;
1326         ENTRY;
1327
1328         oqctl = req_capsule_client_get(&req->rq_pill, &RMF_OBD_QUOTACTL);
1329         if (oqctl == NULL)
1330                 RETURN(-EPROTO);
1331
1332         rc = req_capsule_server_pack(&req->rq_pill);
1333         if (rc) {
1334                 CERROR("ost: out of memory while packing quotacheck reply\n");
1335                 RETURN(-ENOMEM);
1336         }
1337
1338         req->rq_status = obd_quotacheck(req->rq_export, oqctl);
1339         RETURN(0);
1340 }
1341
1342 static int ost_llog_handle_connect(struct obd_export *exp,
1343                                    struct ptlrpc_request *req)
1344 {
1345         struct llogd_conn_body *body;
1346         int rc;
1347         ENTRY;
1348
1349         body = lustre_msg_buf(req->rq_reqmsg, 1, sizeof(*body));
1350         rc = obd_llog_connect(exp, body);
1351         RETURN(rc);
1352 }
1353
1354 static int filter_export_check_flavor(struct filter_obd *filter,
1355                                       struct obd_export *exp,
1356                                       struct ptlrpc_request *req)
1357 {
1358         int     rc = 0;
1359
1360         /* FIXME
1361          * this should be done in filter_connect()/filter_reconnect(), but
1362          * we can't obtain information like NID, which stored in incoming
1363          * request, thus can't decide what flavor to use. so we do it here.
1364          *
1365          * This hack should be removed after the OST stack be rewritten, just
1366          * like what we are doing in mdt_obd_connect()/mdt_obd_reconnect().
1367          */
1368         if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_INVALID)
1369                 return 0;
1370
1371         CDEBUG(D_SEC, "from %s\n", sptlrpc_part2name(req->rq_sp_from));
1372         spin_lock(&exp->exp_lock);
1373         exp->exp_sp_peer = req->rq_sp_from;
1374
1375         read_lock(&filter->fo_sptlrpc_lock);
1376         sptlrpc_rule_set_choose(&filter->fo_sptlrpc_rset, exp->exp_sp_peer,
1377                                 req->rq_peer.nid, &exp->exp_flvr);
1378         read_unlock(&filter->fo_sptlrpc_lock);
1379
1380         if (exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
1381                 CERROR("invalid rpc flavor %x, expect %x, from %s\n",
1382                        req->rq_flvr.sf_rpc, exp->exp_flvr.sf_rpc,
1383                        libcfs_nid2str(req->rq_peer.nid));
1384                 exp->exp_flvr.sf_rpc = SPTLRPC_FLVR_INVALID;
1385                 rc = -EACCES;
1386         }
1387
1388         spin_unlock(&exp->exp_lock);
1389
1390         return rc;
1391 }
1392
1393 static int ost_filter_recovery_request(struct ptlrpc_request *req,
1394                                        struct obd_device *obd, int *process)
1395 {
1396         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1397         case OST_CONNECT: /* This will never get here, but for completeness. */
1398         case OST_DISCONNECT:
1399                *process = 1;
1400                RETURN(0);
1401
1402         case OBD_PING:
1403         case OST_CREATE:
1404         case OST_DESTROY:
1405         case OST_PUNCH:
1406         case OST_SETATTR:
1407         case OST_SYNC:
1408         case OST_WRITE:
1409         case OBD_LOG_CANCEL:
1410         case LDLM_ENQUEUE:
1411                 *process = target_queue_recovery_request(req, obd);
1412                 RETURN(0);
1413
1414         default:
1415                 DEBUG_REQ(D_ERROR, req, "not permitted during recovery");
1416                 *process = -EAGAIN;
1417                 RETURN(0);
1418         }
1419 }
1420
1421 int ost_msg_check_version(struct lustre_msg *msg)
1422 {
1423         int rc;
1424
1425         switch(lustre_msg_get_opc(msg)) {
1426         case OST_CONNECT:
1427         case OST_DISCONNECT:
1428         case OBD_PING:
1429         case SEC_CTX_INIT:
1430         case SEC_CTX_INIT_CONT:
1431         case SEC_CTX_FINI:
1432                 rc = lustre_msg_check_version(msg, LUSTRE_OBD_VERSION);
1433                 if (rc)
1434                         CERROR("bad opc %u version %08x, expecting %08x\n",
1435                                lustre_msg_get_opc(msg),
1436                                lustre_msg_get_version(msg),
1437                                LUSTRE_OBD_VERSION);
1438                 break;
1439         case OST_CREATE:
1440         case OST_DESTROY:
1441         case OST_GETATTR:
1442         case OST_SETATTR:
1443         case OST_WRITE:
1444         case OST_READ:
1445         case OST_PUNCH:
1446         case OST_STATFS:
1447         case OST_SYNC:
1448         case OST_SET_INFO:
1449         case OST_GET_INFO:
1450         case OST_QUOTACHECK:
1451         case OST_QUOTACTL:
1452                 rc = lustre_msg_check_version(msg, LUSTRE_OST_VERSION);
1453                 if (rc)
1454                         CERROR("bad opc %u version %08x, expecting %08x\n",
1455                                lustre_msg_get_opc(msg),
1456                                lustre_msg_get_version(msg),
1457                                LUSTRE_OST_VERSION);
1458                 break;
1459         case LDLM_ENQUEUE:
1460         case LDLM_CONVERT:
1461         case LDLM_CANCEL:
1462         case LDLM_BL_CALLBACK:
1463         case LDLM_CP_CALLBACK:
1464                 rc = lustre_msg_check_version(msg, LUSTRE_DLM_VERSION);
1465                 if (rc)
1466                         CERROR("bad opc %u version %08x, expecting %08x\n",
1467                                lustre_msg_get_opc(msg),
1468                                lustre_msg_get_version(msg),
1469                                LUSTRE_DLM_VERSION);
1470                 break;
1471         case LLOG_ORIGIN_CONNECT:
1472         case OBD_LOG_CANCEL:
1473                 rc = lustre_msg_check_version(msg, LUSTRE_LOG_VERSION);
1474                 if (rc)
1475                         CERROR("bad opc %u version %08x, expecting %08x\n",
1476                                lustre_msg_get_opc(msg),
1477                                lustre_msg_get_version(msg),
1478                                LUSTRE_LOG_VERSION);
1479                 break;
1480         default:
1481                 CERROR("Unexpected opcode %d\n", lustre_msg_get_opc(msg));
1482                 rc = -ENOTSUPP;
1483         }
1484         return rc;
1485 }
1486
1487 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
1488 int ost_handle(struct ptlrpc_request *req)
1489 {
1490         struct obd_trans_info trans_info = { 0, };
1491         struct obd_trans_info *oti = &trans_info;
1492         int should_process, fail = OBD_FAIL_OST_ALL_REPLY_NET, rc = 0;
1493         struct obd_device *obd = NULL;
1494         ENTRY;
1495
1496         LASSERT(current->journal_info == NULL);
1497
1498         /* primordial rpcs don't affect server recovery */
1499         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1500         case SEC_CTX_INIT:
1501         case SEC_CTX_INIT_CONT:
1502         case SEC_CTX_FINI:
1503                 GOTO(out, rc = 0);
1504         }
1505
1506         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
1507
1508         /* XXX identical to MDS */
1509         if (lustre_msg_get_opc(req->rq_reqmsg) != OST_CONNECT) {
1510                 int recovering;
1511
1512                 if (req->rq_export == NULL) {
1513                         CDEBUG(D_HA,"operation %d on unconnected OST from %s\n",
1514                                lustre_msg_get_opc(req->rq_reqmsg),
1515                                libcfs_id2str(req->rq_peer));
1516                         req->rq_status = -ENOTCONN;
1517                         GOTO(out, rc = -ENOTCONN);
1518                 }
1519
1520                 obd = req->rq_export->exp_obd;
1521
1522                 /* Check for aborted recovery. */
1523                 spin_lock_bh(&obd->obd_processing_task_lock);
1524                 recovering = obd->obd_recovering;
1525                 spin_unlock_bh(&obd->obd_processing_task_lock);
1526                 if (recovering) {
1527                         rc = ost_filter_recovery_request(req, obd,
1528                                                          &should_process);
1529                         if (rc || !should_process)
1530                                 RETURN(rc);
1531                         else if (should_process < 0) {
1532                                 req->rq_status = should_process;
1533                                 rc = ptlrpc_error(req);
1534                                 RETURN(rc);
1535                         }
1536                 }
1537         }
1538
1539         oti_init(oti, req);
1540
1541         rc = ost_msg_check_version(req->rq_reqmsg);
1542         if (rc)
1543                 RETURN(rc);
1544
1545         switch (lustre_msg_get_opc(req->rq_reqmsg)) {
1546         case OST_CONNECT: {
1547                 CDEBUG(D_INODE, "connect\n");
1548                 req_capsule_set(&req->rq_pill, &RQF_OST_CONNECT);
1549                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CONNECT_NET))
1550                         RETURN(0);
1551                 rc = target_handle_connect(req);
1552                 if (!rc) {
1553                         struct obd_export *exp = req->rq_export;
1554
1555                         obd = exp->exp_obd;
1556
1557                         rc = filter_export_check_flavor(&obd->u.filter,
1558                                                         exp, req);
1559                 }
1560                 break;
1561         }
1562         case OST_DISCONNECT:
1563                 CDEBUG(D_INODE, "disconnect\n");
1564                 req_capsule_set(&req->rq_pill, &RQF_OST_DISCONNECT);
1565                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_DISCONNECT_NET))
1566                         RETURN(0);
1567                 rc = target_handle_disconnect(req);
1568                 break;
1569         case OST_CREATE:
1570                 CDEBUG(D_INODE, "create\n");
1571                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_CREATE_NET))
1572                         RETURN(0);
1573                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
1574                         GOTO(out, rc = -ENOSPC);
1575                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1576                         GOTO(out, rc = -EROFS);
1577                 rc = ost_create(req->rq_export, req, oti);
1578                 break;
1579         case OST_DESTROY:
1580                 CDEBUG(D_INODE, "destroy\n");
1581                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_DESTROY_NET))
1582                         RETURN(0);
1583                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1584                         GOTO(out, rc = -EROFS);
1585                 rc = ost_destroy(req->rq_export, req, oti);
1586                 break;
1587         case OST_GETATTR:
1588                 CDEBUG(D_INODE, "getattr\n");
1589                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_GETATTR_NET))
1590                         RETURN(0);
1591                 rc = ost_getattr(req->rq_export, req);
1592                 break;
1593         case OST_SETATTR:
1594                 CDEBUG(D_INODE, "setattr\n");
1595                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_SETATTR_NET))
1596                         RETURN(0);
1597                 rc = ost_setattr(req->rq_export, req, oti);
1598                 break;
1599         case OST_WRITE:
1600                 CDEBUG(D_INODE, "write\n");
1601                 /* req->rq_request_portal would be nice, if it was set */
1602                 if (req->rq_rqbd->rqbd_service->srv_req_portal !=OST_IO_PORTAL){
1603                         CERROR("%s: deny write request from %s to portal %u\n",
1604                                req->rq_export->exp_obd->obd_name,
1605                                obd_export_nid2str(req->rq_export),
1606                                req->rq_rqbd->rqbd_service->srv_req_portal);
1607                         GOTO(out, rc = -EPROTO);
1608                 }
1609                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_NET))
1610                         RETURN(0);
1611                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOSPC))
1612                         GOTO(out, rc = -ENOSPC);
1613                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1614                         GOTO(out, rc = -EROFS);
1615                 rc = ost_brw_write(req, oti);
1616                 LASSERT(current->journal_info == NULL);
1617                 /* ost_brw_write sends its own replies */
1618                 RETURN(rc);
1619         case OST_READ:
1620                 CDEBUG(D_INODE, "read\n");
1621                 /* req->rq_request_portal would be nice, if it was set */
1622                 if (req->rq_rqbd->rqbd_service->srv_req_portal !=OST_IO_PORTAL){
1623                         CERROR("%s: deny read request from %s to portal %u\n",
1624                                req->rq_export->exp_obd->obd_name,
1625                                obd_export_nid2str(req->rq_export),
1626                                req->rq_rqbd->rqbd_service->srv_req_portal);
1627                         GOTO(out, rc = -EPROTO);
1628                 }
1629                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_BRW_NET))
1630                         RETURN(0);
1631                 rc = ost_brw_read(req, oti);
1632                 LASSERT(current->journal_info == NULL);
1633                 /* ost_brw_read sends its own replies */
1634                 RETURN(rc);
1635         case OST_PUNCH:
1636                 CDEBUG(D_INODE, "punch\n");
1637                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_PUNCH_NET))
1638                         RETURN(0);
1639                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_EROFS))
1640                         GOTO(out, rc = -EROFS);
1641                 rc = ost_punch(req->rq_export, req, oti);
1642                 break;
1643         case OST_STATFS:
1644                 CDEBUG(D_INODE, "statfs\n");
1645                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_STATFS_NET))
1646                         RETURN(0);
1647                 rc = ost_statfs(req);
1648                 break;
1649         case OST_SYNC:
1650                 CDEBUG(D_INODE, "sync\n");
1651                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_SYNC_NET))
1652                         RETURN(0);
1653                 rc = ost_sync(req->rq_export, req);
1654                 break;
1655         case OST_SET_INFO:
1656                 DEBUG_REQ(D_INODE, req, "set_info");
1657                 rc = ost_set_info(req->rq_export, req);
1658                 break;
1659         case OST_GET_INFO:
1660                 DEBUG_REQ(D_INODE, req, "get_info");
1661                 rc = ost_get_info(req->rq_export, req);
1662                 break;
1663         case OST_QUOTACHECK:
1664                 CDEBUG(D_INODE, "quotacheck\n");
1665                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTACHECK);
1666                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_QUOTACHECK_NET))
1667                         RETURN(0);
1668                 rc = ost_handle_quotacheck(req);
1669                 break;
1670         case OST_QUOTACTL:
1671                 CDEBUG(D_INODE, "quotactl\n");
1672                 req_capsule_set(&req->rq_pill, &RQF_OST_QUOTACTL);
1673                 if (OBD_FAIL_CHECK(OBD_FAIL_OST_QUOTACTL_NET))
1674                         RETURN(0);
1675                 rc = ost_handle_quotactl(req);
1676                 break;
1677         case OBD_PING:
1678                 DEBUG_REQ(D_INODE, req, "ping");
1679                 req_capsule_set(&req->rq_pill, &RQF_OBD_PING);
1680                 rc = target_handle_ping(req);
1681                 break;
1682         /* FIXME - just reply status */
1683         case LLOG_ORIGIN_CONNECT:
1684                 DEBUG_REQ(D_INODE, req, "log connect");
1685                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_CONNECT);
1686                 rc = ost_llog_handle_connect(req->rq_export, req);
1687                 req->rq_status = rc;
1688                 rc = req_capsule_server_pack(&req->rq_pill);
1689                 if (rc)
1690                         RETURN(rc);
1691                 RETURN(ptlrpc_reply(req));
1692         case OBD_LOG_CANCEL:
1693                 CDEBUG(D_INODE, "log cancel\n");
1694                 req_capsule_set(&req->rq_pill, &RQF_LOG_CANCEL);
1695                 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_LOG_CANCEL_NET))
1696                         RETURN(0);
1697                 rc = llog_origin_handle_cancel(req);
1698                 req->rq_status = rc;
1699                 rc = req_capsule_server_pack(&req->rq_pill);
1700                 if (rc)
1701                         RETURN(rc);
1702                 RETURN(ptlrpc_reply(req));
1703         case LDLM_ENQUEUE:
1704                 CDEBUG(D_INODE, "enqueue\n");
1705                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);
1706                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_ENQUEUE))
1707                         RETURN(0);
1708                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
1709                                          ldlm_server_blocking_ast,
1710                                          ldlm_server_glimpse_ast);
1711                 fail = OBD_FAIL_OST_LDLM_REPLY_NET;
1712                 break;
1713         case LDLM_CONVERT:
1714                 CDEBUG(D_INODE, "convert\n");
1715                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CONVERT);
1716                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CONVERT))
1717                         RETURN(0);
1718                 rc = ldlm_handle_convert(req);
1719                 break;
1720         case LDLM_CANCEL:
1721                 CDEBUG(D_INODE, "cancel\n");
1722                 req_capsule_set(&req->rq_pill, &RQF_LDLM_CANCEL);
1723                 if (OBD_FAIL_CHECK(OBD_FAIL_LDLM_CANCEL))
1724                         RETURN(0);
1725                 rc = ldlm_handle_cancel(req);
1726                 break;
1727         case LDLM_BL_CALLBACK:
1728         case LDLM_CP_CALLBACK:
1729                 CDEBUG(D_INODE, "callback\n");
1730                 CERROR("callbacks should not happen on OST\n");
1731                 /* fall through */
1732         default:
1733                 CERROR("Unexpected opcode %d\n",
1734                        lustre_msg_get_opc(req->rq_reqmsg));
1735                 req->rq_status = -ENOTSUPP;
1736                 rc = ptlrpc_error(req);
1737                 RETURN(rc);
1738         }
1739
1740         LASSERT(current->journal_info == NULL);
1741
1742         EXIT;
1743         /* If we're DISCONNECTing, the export_data is already freed */
1744         if (!rc && lustre_msg_get_opc(req->rq_reqmsg) != OST_DISCONNECT)
1745                 target_committed_to_req(req);
1746
1747 out:
1748         if (!rc)
1749                 oti_to_request(oti, req);
1750
1751         target_send_reply(req, rc, fail);
1752         return 0;
1753 }
1754 EXPORT_SYMBOL(ost_handle);
1755 /*
1756  * free per-thread pool created by ost_thread_init().
1757  */
1758 static void ost_thread_done(struct ptlrpc_thread *thread)
1759 {
1760         int i;
1761         struct ost_thread_local_cache *tls; /* TLS stands for Thread-Local
1762                                              * Storage */
1763
1764         ENTRY;
1765
1766         LASSERT(thread != NULL);
1767
1768         /*
1769          * be prepared to handle partially-initialized pools (because this is
1770          * called from ost_thread_init() for cleanup.
1771          */
1772         tls = thread->t_data;
1773         if (tls != NULL) {
1774                 for (i = 0; i < OST_THREAD_POOL_SIZE; ++ i) {
1775                         if (tls->page[i] != NULL)
1776                                 OBD_PAGE_FREE(tls->page[i]);
1777                 }
1778                 OBD_FREE_PTR(tls);
1779                 thread->t_data = NULL;
1780         }
1781         EXIT;
1782 }
1783
1784 /*
1785  * initialize per-thread page pool (bug 5137).
1786  */
1787 static int ost_thread_init(struct ptlrpc_thread *thread)
1788 {
1789         int result;
1790         int i;
1791         struct ost_thread_local_cache *tls;
1792
1793         ENTRY;
1794
1795         LASSERT(thread != NULL);
1796         LASSERT(thread->t_data == NULL);
1797         LASSERTF(thread->t_id <= OSS_THREADS_MAX, "%u\n", thread->t_id);
1798
1799         OBD_ALLOC_PTR(tls);
1800         if (tls != NULL) {
1801                 result = 0;
1802                 thread->t_data = tls;
1803                 /*
1804                  * populate pool
1805                  */
1806                 for (i = 0; i < OST_THREAD_POOL_SIZE; ++ i) {
1807                         OBD_PAGE_ALLOC(tls->page[i], OST_THREAD_POOL_GFP);
1808                         if (tls->page[i] == NULL) {
1809                                 ost_thread_done(thread);
1810                                 result = -ENOMEM;
1811                                 break;
1812                         }
1813                 }
1814         } else
1815                 result = -ENOMEM;
1816         RETURN(result);
1817 }
1818
1819 #define OST_WATCHDOG_TIMEOUT (obd_timeout * 1000)
1820
1821 /* Sigh - really, this is an OSS, the _server_, not the _target_ */
1822 static int ost_setup(struct obd_device *obd, struct lustre_cfg* lcfg)
1823 {
1824         struct ost_obd *ost = &obd->u.ost;
1825         struct lprocfs_static_vars lvars;
1826         int oss_min_threads;
1827         int oss_max_threads;
1828         int oss_min_create_threads;
1829         int oss_max_create_threads;
1830         int rc;
1831         ENTRY;
1832
1833         rc = cleanup_group_info();
1834         if (rc)
1835                 RETURN(rc);
1836
1837         rc = llog_start_commit_thread();
1838         if (rc < 0)
1839                 RETURN(rc);
1840
1841         lprocfs_ost_init_vars(&lvars);
1842         lprocfs_obd_setup(obd, lvars.obd_vars);
1843
1844         sema_init(&ost->ost_health_sem, 1);
1845
1846         if (oss_num_threads) {
1847                 /* If oss_num_threads is set, it is the min and the max. */
1848                 if (oss_num_threads > OSS_THREADS_MAX)
1849                         oss_num_threads = OSS_THREADS_MAX;
1850                 if (oss_num_threads < OSS_THREADS_MIN)
1851                         oss_num_threads = OSS_THREADS_MIN;
1852                 oss_max_threads = oss_min_threads = oss_num_threads;
1853         } else {
1854                 /* Base min threads on memory and cpus */
1855                 oss_min_threads = num_possible_cpus() * num_physpages >>
1856                         (27 - CFS_PAGE_SHIFT);
1857                 if (oss_min_threads < OSS_THREADS_MIN)
1858                         oss_min_threads = OSS_THREADS_MIN;
1859                 /* Insure a 4x range for dynamic threads */
1860                 if (oss_min_threads > OSS_THREADS_MAX / 4)
1861                         oss_min_threads = OSS_THREADS_MAX / 4;
1862                 oss_max_threads = min(OSS_THREADS_MAX, oss_min_threads * 4);
1863         }
1864
1865         ost->ost_service =
1866                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
1867                                 OST_MAXREPSIZE, OST_REQUEST_PORTAL,
1868                                 OSC_REPLY_PORTAL,
1869                                 OST_WATCHDOG_TIMEOUT, ost_handle,
1870                                 LUSTRE_OSS_NAME, obd->obd_proc_entry,
1871                                 ost_print_req, oss_min_threads,
1872                                 oss_max_threads, "ll_ost",
1873                                 LCT_DT_THREAD);
1874         if (ost->ost_service == NULL) {
1875                 CERROR("failed to start service\n");
1876                 GOTO(out_lprocfs, rc = -ENOMEM);
1877         }
1878
1879         rc = ptlrpc_start_threads(obd, ost->ost_service);
1880         if (rc)
1881                 GOTO(out_service, rc = -EINVAL);
1882
1883         if (oss_num_create_threads) {
1884                 if (oss_num_create_threads > OSS_MAX_CREATE_THREADS)
1885                         oss_num_create_threads = OSS_MAX_CREATE_THREADS;
1886                 if (oss_num_create_threads < OSS_DEF_CREATE_THREADS)
1887                         oss_num_create_threads = OSS_DEF_CREATE_THREADS;
1888                 oss_min_create_threads = oss_max_create_threads =
1889                         oss_num_create_threads;
1890         } else {
1891                 oss_min_create_threads = OSS_DEF_CREATE_THREADS;
1892                 oss_max_create_threads = OSS_MAX_CREATE_THREADS;
1893         }
1894
1895         ost->ost_create_service =
1896                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
1897                                 OST_MAXREPSIZE, OST_CREATE_PORTAL,
1898                                 OSC_REPLY_PORTAL,
1899                                 OST_WATCHDOG_TIMEOUT, ost_handle, "ost_create",
1900                                 obd->obd_proc_entry, ost_print_req,
1901                                 oss_min_create_threads,
1902                                 oss_max_create_threads,
1903                                 "ll_ost_creat", LCT_DT_THREAD);
1904         if (ost->ost_create_service == NULL) {
1905                 CERROR("failed to start OST create service\n");
1906                 GOTO(out_service, rc = -ENOMEM);
1907         }
1908
1909         rc = ptlrpc_start_threads(obd, ost->ost_create_service);
1910         if (rc)
1911                 GOTO(out_create, rc = -EINVAL);
1912
1913         ost->ost_io_service =
1914                 ptlrpc_init_svc(OST_NBUFS, OST_BUFSIZE, OST_MAXREQSIZE,
1915                                 OST_MAXREPSIZE, OST_IO_PORTAL,
1916                                 OSC_REPLY_PORTAL,
1917                                 OST_WATCHDOG_TIMEOUT, ost_handle, "ost_io",
1918                                 obd->obd_proc_entry, ost_print_req,
1919                                 oss_min_threads, oss_max_threads,
1920                                 "ll_ost_io", LCT_DT_THREAD);
1921         if (ost->ost_io_service == NULL) {
1922                 CERROR("failed to start OST I/O service\n");
1923                 GOTO(out_create, rc = -ENOMEM);
1924         }
1925
1926         ost->ost_io_service->srv_init = ost_thread_init;
1927         ost->ost_io_service->srv_done = ost_thread_done;
1928         ost->ost_io_service->srv_cpu_affinity = 1;
1929         rc = ptlrpc_start_threads(obd, ost->ost_io_service);
1930         if (rc)
1931                 GOTO(out_io, rc = -EINVAL);
1932
1933         ping_evictor_start();
1934
1935         RETURN(0);
1936
1937 out_io:
1938         ptlrpc_unregister_service(ost->ost_io_service);
1939         ost->ost_io_service = NULL;
1940 out_create:
1941         ptlrpc_unregister_service(ost->ost_create_service);
1942         ost->ost_create_service = NULL;
1943 out_service:
1944         ptlrpc_unregister_service(ost->ost_service);
1945         ost->ost_service = NULL;
1946 out_lprocfs:
1947         lprocfs_obd_cleanup(obd);
1948         RETURN(rc);
1949 }
1950
1951 static int ost_cleanup(struct obd_device *obd)
1952 {
1953         struct ost_obd *ost = &obd->u.ost;
1954         int err = 0;
1955         ENTRY;
1956
1957         ping_evictor_stop();
1958
1959         spin_lock_bh(&obd->obd_processing_task_lock);
1960         if (obd->obd_recovering) {
1961                 target_cancel_recovery_timer(obd);
1962                 obd->obd_recovering = 0;
1963         }
1964         spin_unlock_bh(&obd->obd_processing_task_lock);
1965
1966         down(&ost->ost_health_sem);
1967         ptlrpc_unregister_service(ost->ost_service);
1968         ptlrpc_unregister_service(ost->ost_create_service);
1969         ptlrpc_unregister_service(ost->ost_io_service);
1970         ost->ost_service = NULL;
1971         ost->ost_create_service = NULL;
1972         up(&ost->ost_health_sem);
1973
1974         lprocfs_obd_cleanup(obd);
1975
1976         RETURN(err);
1977 }
1978
1979 static int ost_health_check(struct obd_device *obd)
1980 {
1981         struct ost_obd *ost = &obd->u.ost;
1982         int rc = 0;
1983
1984         down(&ost->ost_health_sem);
1985         rc |= ptlrpc_service_health_check(ost->ost_service);
1986         rc |= ptlrpc_service_health_check(ost->ost_create_service);
1987         rc |= ptlrpc_service_health_check(ost->ost_io_service);
1988         up(&ost->ost_health_sem);
1989
1990         /*
1991          * health_check to return 0 on healthy
1992          * and 1 on unhealthy.
1993          */
1994         if( rc != 0)
1995                 rc = 1;
1996
1997         return rc;
1998 }
1999
2000 struct ost_thread_local_cache *ost_tls(struct ptlrpc_request *r)
2001 {
2002         return (struct ost_thread_local_cache *)(r->rq_svc_thread->t_data);
2003 }
2004
2005 /* use obd ops to offer management infrastructure */
2006 static struct obd_ops ost_obd_ops = {
2007         .o_owner        = THIS_MODULE,
2008         .o_setup        = ost_setup,
2009         .o_cleanup      = ost_cleanup,
2010         .o_health_check = ost_health_check,
2011 };
2012
2013
2014 static int __init ost_init(void)
2015 {
2016         struct lprocfs_static_vars lvars;
2017         int rc;
2018         ENTRY;
2019
2020         lprocfs_ost_init_vars(&lvars);
2021         rc = class_register_type(&ost_obd_ops, NULL, lvars.module_vars,
2022                                  LUSTRE_OSS_NAME, NULL);
2023
2024         if (ost_num_threads != 0 && oss_num_threads == 0) {
2025                 LCONSOLE_INFO("ost_num_threads module parameter is deprecated, "
2026                               "use oss_num_threads instead or unset both for "
2027                               "dynamic thread startup\n");
2028                 oss_num_threads = ost_num_threads;
2029         }
2030
2031         RETURN(rc);
2032 }
2033
2034 static void /*__exit*/ ost_exit(void)
2035 {
2036         class_unregister_type(LUSTRE_OSS_NAME);
2037 }
2038
2039 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2040 MODULE_DESCRIPTION("Lustre Object Storage Target (OST) v0.01");
2041 MODULE_LICENSE("GPL");
2042
2043 module_init(ost_init);
2044 module_exit(ost_exit);