lustre/osp/osp_sync.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/osp/osp_sync.c
  33  *
  34  * Lustre OST Proxy Device
  35  *
  36  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  37  * Author: Mikhail Pershin <mike.pershin@intel.com>
  38  */
  39
  40 #define DEBUG_SUBSYSTEM S_MDS
  41
  42 #include <linux/kthread.h>
  43 #include <linux/delay.h>
  44 #include <lustre_log.h>
  45 #include <lustre_update.h>
  46 #include "osp_internal.h"
  47
  48 /*
  49  * this is a components of OSP implementing synchronization between MDS and OST
  50  * it llogs all interesting changes (currently it's uig/gid change and object
  51  * destroy) atomically, then makes sure changes hit OST storage
  52  *
  53  * we have 4 queues of work:
  54  *
  55  * the first queue is llog itself, once read a change is stored in 2nd queue
  56  * in form of RPC (but RPC isn't fired yet).
  57  *
  58  * the second queue (opd_sync_waiting_for_commit) holds changes awaiting local
  59  * commit. once change is committed locally it migrates onto 3rd queue.
  60  *
  61  * the third queue (opd_sync_committed_here) holds changes committed locally,
  62  * but not sent to OST (as the pipe can be full). once pipe becomes non-full
  63  * we take a change from the queue and fire corresponded RPC.
  64  *
  65  * once RPC is reported committed by OST (using regular last_committed mech.)
  66  * the change jumps into 4th queue (opd_sync_committed_there), now we can
  67  * cancel corresponded llog record and release RPC
  68  *
  69  * opd_sync_changes is a number of unread llog records (to be processed).
  70  * notice this number doesn't include llog records from previous boots.
  71  * with OSP_SYNC_THRESHOLD we try to batch processing a bit (TO BE IMPLEMENTED)
  72  *
  73  * opd_sync_rpcs_in_progress is total number of requests in above 2-4 queues.
  74  * we control this with OSP_MAX_RPCS_IN_PROGRESS so that OSP don't consume
  75  * too much memory -- how to deal with 1000th OSTs ? batching could help?
  76  *
  77  * opd_sync_rpcs_in_flight is a number of RPC in flight.
  78  * we control this with OSP_MAX_RPCS_IN_FLIGHT
  79  */
  80
  81 /* XXX: do math to learn reasonable threshold
  82  * should it be ~ number of changes fitting bulk? */
  83
  84 #define OSP_SYNC_THRESHOLD              10
  85 #define OSP_MAX_RPCS_IN_FLIGHT          8
  86 #define OSP_MAX_RPCS_IN_PROGRESS        4096
  87
  88 #define OSP_JOB_MAGIC           0x26112005
  89
  90 struct osp_job_req_args {
  91         /** bytes reserved for ptlrpc_replay_req() */
  92         struct ptlrpc_replay_async_args jra_raa;
  93         struct list_head                jra_committed_link;
  94         struct list_head                jra_in_flight_link;
  95         struct llog_cookie              jra_lcookie;
  96         __u32                           jra_magic;
  97 };
  98
  99 static int osp_sync_add_commit_cb(const struct lu_env *env,
 100                                   struct osp_device *d, struct thandle *th);
 101
 102 static inline int osp_sync_running(struct osp_device *d)
 103 {
 104         return !!(d->opd_sync_thread.t_flags & SVC_RUNNING);
 105 }
 106
 107 /**
 108  * Check status: whether OSP thread has stopped
 109  *
 110  * \param[in] d         OSP device
 111  *
 112  * \retval 0            still running
 113  * \retval 1            stopped
 114  */
 115 static inline int osp_sync_stopped(struct osp_device *d)
 116 {
 117         return !!(d->opd_sync_thread.t_flags & SVC_STOPPED);
 118 }
 119
 120 /*
 121  ** Check for new changes to sync
 122  *
 123  * \param[in] d         OSP device
 124  *
 125  * \retval 1            there are changes
 126  * \retval 0            there are no changes
 127  */
 128 static inline int osp_sync_has_new_job(struct osp_device *d)
 129 {
 130         return atomic_read(&d->opd_sync_changes) > 0 ||
 131                 d->opd_sync_prev_done == 0;
 132 }
 133
 134 static inline int osp_sync_in_flight_conflict(struct osp_device *d,
 135                                              struct llog_rec_hdr *h)
 136 {
 137         struct osp_job_req_args *jra;
 138         struct ost_id            ostid;
 139         int                      conflict = 0;
 140
 141         if (h == NULL || h->lrh_type == LLOG_GEN_REC ||
 142             list_empty(&d->opd_sync_in_flight_list))
 143                 return conflict;
 144
 145         memset(&ostid, 0, sizeof(ostid));
 146         switch (h->lrh_type) {
 147         case MDS_UNLINK_REC: {
 148                 struct llog_unlink_rec *unlink = (struct llog_unlink_rec *)h;
 149
 150                 ostid_set_seq(&ostid, unlink->lur_oseq);
 151                 if (ostid_set_id(&ostid, unlink->lur_oid)) {
 152                         CERROR("Bad %llu to set " DOSTID "\n",
 153                                (unsigned long long)(unlink->lur_oid),
 154                                POSTID(&ostid));
 155                         return 1;
 156                 }
 157                 }
 158                 break;
 159         case MDS_UNLINK64_REC:
 160                 fid_to_ostid(&((struct llog_unlink64_rec *)h)->lur_fid, &ostid);
 161                 break;
 162         case MDS_SETATTR64_REC:
 163                 ostid = ((struct llog_setattr64_rec *)h)->lsr_oi;
 164                 break;
 165         default:
 166                 LBUG();
 167         }
 168
 169         spin_lock(&d->opd_sync_lock);
 170         list_for_each_entry(jra, &d->opd_sync_in_flight_list,
 171                             jra_in_flight_link) {
 172                 struct ptlrpc_request   *req;
 173                 struct ost_body         *body;
 174
 175                 LASSERT(jra->jra_magic == OSP_JOB_MAGIC);
 176
 177                 req = container_of((void *)jra, struct ptlrpc_request,
 178                                    rq_async_args);
 179                 body = req_capsule_client_get(&req->rq_pill,
 180                                               &RMF_OST_BODY);
 181                 LASSERT(body);
 182
 183                 if (memcmp(&ostid, &body->oa.o_oi, sizeof(ostid)) == 0) {
 184                         conflict = 1;
 185                         break;
 186                 }
 187         }
 188         spin_unlock(&d->opd_sync_lock);
 189
 190         return conflict;
 191 }
 192
 193 static inline int osp_sync_rpcs_in_progress_low(struct osp_device *d)
 194 {
 195         return atomic_read(&d->opd_sync_rpcs_in_progress) <
 196                 d->opd_sync_max_rpcs_in_progress;
 197 }
 198
 199 /**
 200  * Check for room in the network pipe to OST
 201  *
 202  * \param[in] d         OSP device
 203  *
 204  * \retval 1            there is room
 205  * \retval 0            no room, the pipe is full
 206  */
 207 static inline int osp_sync_rpcs_in_flight_low(struct osp_device *d)
 208 {
 209         return atomic_read(&d->opd_sync_rpcs_in_flight) <
 210                 d->opd_sync_max_rpcs_in_flight;
 211 }
 212
 213 /**
 214  * Wake up check for the main sync thread
 215  *
 216  * \param[in] d         OSP device
 217  *
 218  * \retval 1            time to wake up
 219  * \retval 0            no need to wake up
 220  */
 221 static inline int osp_sync_has_work(struct osp_device *osp)
 222 {
 223         /* has new/old changes and low in-progress? */
 224         if (osp_sync_has_new_job(osp) && osp_sync_rpcs_in_progress_low(osp) &&
 225             osp_sync_rpcs_in_flight_low(osp) && osp->opd_imp_connected)
 226                 return 1;
 227
 228         /* has remotely committed? */
 229         if (!list_empty(&osp->opd_sync_committed_there))
 230                 return 1;
 231
 232         return 0;
 233 }
 234
 235 void osp_sync_check_for_work(struct osp_device *osp)
 236 {
 237         if (osp_sync_has_work(osp))
 238                 wake_up(&osp->opd_sync_waitq);
 239 }
 240
 241 static inline __u64 osp_sync_correct_id(struct osp_device *d,
 242                                         struct llog_rec_hdr *rec)
 243 {
 244         /*
 245          * llog use cyclic store with 32 bit lrh_id
 246          * so overflow lrh_id is possible. Range between
 247          * last_processed and last_committed is less than
 248          * 64745 ^ 2 and less than 2^32 - 1
 249          */
 250         __u64 correct_id = d->opd_sync_last_committed_id;
 251
 252         if ((correct_id & 0xffffffffULL) < rec->lrh_id)
 253                 correct_id -= 0x100000000ULL;
 254
 255         correct_id &= ~0xffffffffULL;
 256         correct_id |= rec->lrh_id;
 257
 258         return correct_id;
 259 }
 260
 261 /**
 262  * Check and return ready-for-new status.
 263  *
 264  * The thread processing llog record uses this function to check whether
 265  * it's time to take another record and process it. The number of conditions
 266  * must be met: the connection should be ready, RPCs in flight not exceeding
 267  * the limit, the record is committed locally, etc (see the lines below).
 268  *
 269  * \param[in] d         OSP device
 270  * \param[in] rec       next llog record to process
 271  *
 272  * \retval 0            not ready
 273  * \retval 1            ready
 274  */
 275 static inline int osp_sync_can_process_new(struct osp_device *d,
 276                                            struct llog_rec_hdr *rec)
 277 {
 278         LASSERT(d);
 279
 280         if (unlikely(atomic_read(&d->opd_sync_barrier) > 0))
 281                 return 0;
 282         if (unlikely(osp_sync_in_flight_conflict(d, rec)))
 283                 return 0;
 284         if (!osp_sync_rpcs_in_progress_low(d))
 285                 return 0;
 286         if (!osp_sync_rpcs_in_flight_low(d))
 287                 return 0;
 288         if (!d->opd_imp_connected)
 289                 return 0;
 290         if (d->opd_sync_prev_done == 0)
 291                 return 1;
 292         if (atomic_read(&d->opd_sync_changes) == 0)
 293                 return 0;
 294         if (rec == NULL)
 295                 return 1;
 296         /* notice "<" not "<=" */
 297         if (osp_sync_correct_id(d, rec) < d->opd_sync_last_committed_id)
 298                 return 1;
 299         return 0;
 300 }
 301
 302 /**
 303  * Declare intention to add a new change.
 304  *
 305  * With regard to OSD API, we have to declare any changes ahead. In this
 306  * case we declare an intention to add a llog record representing the
 307  * change on the local storage.
 308  *
 309  * \param[in] env       LU environment provided by the caller
 310  * \param[in] o         OSP object
 311  * \param[in] type      type of change: MDS_UNLINK64_REC or MDS_SETATTR64_REC
 312  * \param[in] th        transaction handle (local)
 313  *
 314  * \retval 0            on success
 315  * \retval negative     negated errno on error
 316  */
 317 int osp_sync_declare_add(const struct lu_env *env, struct osp_object *o,
 318                          enum llog_op_type type, struct thandle *th)
 319 {
 320         struct osp_thread_info  *osi = osp_env_info(env);
 321         struct osp_device       *d = lu2osp_dev(o->opo_obj.do_lu.lo_dev);
 322         struct llog_ctxt        *ctxt;
 323         struct thandle          *storage_th;
 324         int                      rc;
 325
 326         ENTRY;
 327
 328         /* it's a layering violation, to access internals of th,
 329          * but we can do this as a sanity check, for a while */
 330         LASSERT(th->th_top != NULL);
 331         storage_th = thandle_get_sub_by_dt(env, th->th_top, d->opd_storage);
 332         if (IS_ERR(storage_th))
 333                 RETURN(PTR_ERR(storage_th));
 334
 335         switch (type) {
 336         case MDS_UNLINK64_REC:
 337                 osi->osi_hdr.lrh_len = sizeof(struct llog_unlink64_rec);
 338                 break;
 339         case MDS_SETATTR64_REC:
 340                 osi->osi_hdr.lrh_len = sizeof(struct llog_setattr64_rec_v2);
 341                 break;
 342         default:
 343                 LBUG();
 344         }
 345
 346         ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
 347         if (!ctxt) {
 348                 /* for a reason OSP wasn't able to open llog,
 349                  * just skip logging this operation and hope
 350                  * LFSCK will fix it eventually */
 351                 CERROR("logging isn't available, run LFSCK\n");
 352                 RETURN(0);
 353         }
 354
 355         rc = llog_declare_add(env, ctxt->loc_handle, &osi->osi_hdr,
 356                               storage_th);
 357         llog_ctxt_put(ctxt);
 358
 359         RETURN(rc);
 360 }
 361
 362 /**
 363  * Generate a llog record for a given change.
 364  *
 365  * Generates a llog record for the change passed. The change can be of two
 366  * types: unlink and setattr. The record gets an ID which later will be
 367  * used to track commit status of the change. For unlink changes, the caller
 368  * can supply a starting FID and the count of the objects to destroy. For
 369  * setattr the caller should apply attributes to apply.
 370  *
 371  *
 372  * \param[in] env       LU environment provided by the caller
 373  * \param[in] d         OSP device
 374  * \param[in] fid       fid of the object the change should be applied to
 375  * \param[in] type      type of change: MDS_UNLINK64_REC or MDS_SETATTR64_REC
 376  * \param[in] count     count of objects to destroy
 377  * \param[in] th        transaction handle (local)
 378  * \param[in] attr      attributes for setattr
 379  *
 380  * \retval 0            on success
 381  * \retval negative     negated errno on error
 382  */
 383 static int osp_sync_add_rec(const struct lu_env *env, struct osp_device *d,
 384                             const struct lu_fid *fid, enum llog_op_type type,
 385                             int count, struct thandle *th,
 386                             const struct lu_attr *attr)
 387 {
 388         struct osp_thread_info  *osi = osp_env_info(env);
 389         struct llog_ctxt        *ctxt;
 390         struct thandle          *storage_th;
 391         bool                     immediate_commit_cb = false;
 392         int                      rc;
 393
 394         ENTRY;
 395
 396         /* it's a layering violation, to access internals of th,
 397          * but we can do this as a sanity check, for a while */
 398         LASSERT(th->th_top != NULL);
 399         storage_th = thandle_get_sub_by_dt(env, th->th_top, d->opd_storage);
 400         if (IS_ERR(storage_th))
 401                 RETURN(PTR_ERR(storage_th));
 402
 403         switch (type) {
 404         case MDS_UNLINK64_REC:
 405                 osi->osi_hdr.lrh_len = sizeof(osi->osi_unlink);
 406                 osi->osi_hdr.lrh_type = MDS_UNLINK64_REC;
 407                 osi->osi_unlink.lur_fid  = *fid;
 408                 osi->osi_unlink.lur_count = count;
 409                 break;
 410         case MDS_SETATTR64_REC:
 411                 rc = fid_to_ostid(fid, &osi->osi_oi);
 412                 LASSERT(rc == 0);
 413                 osi->osi_hdr.lrh_len = sizeof(osi->osi_setattr);
 414                 osi->osi_hdr.lrh_type = MDS_SETATTR64_REC;
 415                 osi->osi_setattr.lsr_oi  = osi->osi_oi;
 416                 LASSERT(attr);
 417                 osi->osi_setattr.lsr_uid = attr->la_uid;
 418                 osi->osi_setattr.lsr_gid = attr->la_gid;
 419                 osi->osi_setattr.lsr_layout_version = attr->la_layout_version;
 420                 osi->osi_setattr.lsr_projid = attr->la_projid;
 421                 osi->osi_setattr.lsr_valid =
 422                         ((attr->la_valid & LA_UID) ? OBD_MD_FLUID : 0) |
 423                         ((attr->la_valid & LA_GID) ? OBD_MD_FLGID : 0) |
 424                         ((attr->la_valid & LA_PROJID) ? OBD_MD_FLPROJID : 0);
 425                 if (attr->la_valid & LA_LAYOUT_VERSION) {
 426                         osi->osi_setattr.lsr_valid |= OBD_MD_LAYOUT_VERSION;
 427
 428                         /* FLR: the layout version has to be transferred to
 429                          * OST objects ASAP, otherwise clients will have to
 430                          * experience delay to be able to write OST objects. */
 431                         immediate_commit_cb = true;
 432                 }
 433                 break;
 434         default:
 435                 LBUG();
 436         }
 437
 438         /* we keep the same id, but increment it when the callback
 439          * is registered, so that all records upto the one taken
 440          * by the callback are subject to processing */
 441         spin_lock(&d->opd_sync_lock);
 442         osi->osi_hdr.lrh_id = d->opd_sync_last_used_id;
 443         spin_unlock(&d->opd_sync_lock);
 444
 445         ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
 446         if (ctxt == NULL) {
 447                 /* see comment in osp_sync_declare_add() */
 448                 RETURN(0);
 449         }
 450
 451         rc = llog_add(env, ctxt->loc_handle, &osi->osi_hdr, &osi->osi_cookie,
 452                       storage_th);
 453         llog_ctxt_put(ctxt);
 454
 455         if (likely(rc >= 0)) {
 456                 CDEBUG(D_OTHER, "%s: new record "DFID":%x.%u: rc = %d\n",
 457                        d->opd_obd->obd_name,
 458                        PFID(&osi->osi_cookie.lgc_lgl.lgl_oi.oi_fid),
 459                        osi->osi_cookie.lgc_lgl.lgl_ogen,
 460                        osi->osi_cookie.lgc_index, rc);
 461                 atomic_inc(&d->opd_sync_changes);
 462         }
 463
 464         if (immediate_commit_cb)
 465                 rc = osp_sync_add_commit_cb(env, d, th);
 466         else
 467                 rc = osp_sync_add_commit_cb_1s(env, d, th);
 468
 469         /* return 0 always here, error case just cause no llog record */
 470         RETURN(0);
 471 }
 472
 473 int osp_sync_add(const struct lu_env *env, struct osp_object *o,
 474                  enum llog_op_type type, struct thandle *th,
 475                  const struct lu_attr *attr)
 476 {
 477         return osp_sync_add_rec(env, lu2osp_dev(o->opo_obj.do_lu.lo_dev),
 478                                 lu_object_fid(&o->opo_obj.do_lu), type, 1,
 479                                 th, attr);
 480 }
 481
 482 int osp_sync_gap(const struct lu_env *env, struct osp_device *d,
 483                         struct lu_fid *fid, int lost, struct thandle *th)
 484 {
 485         return osp_sync_add_rec(env, d, fid, MDS_UNLINK64_REC, lost, th, NULL);
 486 }
 487
 488 /*
 489  * it's quite obvious we can't maintain all the structures in the memory:
 490  * while OST is down, MDS can be processing thousands and thousands of unlinks
 491  * filling persistent llogs and in-core respresentation
 492  *
 493  * this doesn't scale at all. so we need basically the following:
 494  * a) destroy/setattr append llog records
 495  * b) once llog has grown to X records, we process first Y committed records
 496  *
 497  *  once record R is found via llog_process(), it becomes committed after any
 498  *  subsequent commit callback (at the most)
 499  */
 500
 501 /**
 502  * ptlrpc commit callback.
 503  *
 504  * The callback is called by PTLRPC when a RPC is reported committed by the
 505  * target (OST). We register the callback for the every RPC applying a change
 506  * from the llog. This way we know then the llog records can be cancelled.
 507  * Notice the callback can be called when OSP is finishing. We can detect this
 508  * checking that actual transno in the request is less or equal of known
 509  * committed transno (see osp_sync_process_committed() for the details).
 510  * XXX: this is pretty expensive and can be improved later using batching.
 511  *
 512  * \param[in] req       request
 513  */
 514 static void osp_sync_request_commit_cb(struct ptlrpc_request *req)
 515 {
 516         struct osp_device *d = req->rq_cb_data;
 517         struct osp_job_req_args *jra;
 518
 519         CDEBUG(D_HA, "commit req %p, transno %llu\n", req, req->rq_transno);
 520
 521         if (unlikely(req->rq_transno == 0))
 522                 return;
 523
 524         /* do not do any opd_sync_rpcs_* accounting here
 525          * it's done in osp_sync_interpret sooner or later */
 526         LASSERT(d);
 527
 528         jra = ptlrpc_req_async_args(jra, req);
 529         LASSERT(jra->jra_magic == OSP_JOB_MAGIC);
 530         LASSERT(list_empty(&jra->jra_committed_link));
 531
 532         ptlrpc_request_addref(req);
 533
 534         spin_lock(&d->opd_sync_lock);
 535         list_add(&jra->jra_committed_link, &d->opd_sync_committed_there);
 536         spin_unlock(&d->opd_sync_lock);
 537
 538         /* XXX: some batching wouldn't hurt */
 539         wake_up(&d->opd_sync_waitq);
 540 }
 541
 542 /**
 543  * RPC interpretation callback.
 544  *
 545  * The callback is called by ptlrpc when RPC is replied. Now we have to decide
 546  * whether we should:
 547  *  - put request on a special list to wait until it's committed by the target,
 548  *    if the request is successful
 549  *  - schedule llog record cancel if no target object is found
 550  *  - try later (essentially after reboot) in case of unexpected error
 551  *
 552  * \param[in] env       LU environment provided by the caller
 553  * \param[in] req       request replied
 554  * \param[in] aa        callback data
 555  * \param[in] rc        result of RPC
 556  *
 557  * \retval 0            always
 558  */
 559 static int osp_sync_interpret(const struct lu_env *env,
 560                               struct ptlrpc_request *req, void *args, int rc)
 561 {
 562         struct osp_job_req_args *jra = args;
 563         struct osp_device *d = req->rq_cb_data;
 564
 565         if (jra->jra_magic != OSP_JOB_MAGIC) {
 566                 DEBUG_REQ(D_ERROR, req, "bad magic %u", jra->jra_magic);
 567                 LBUG();
 568         }
 569         LASSERT(d);
 570
 571         CDEBUG(D_HA, "reply req %p/%d, rc %d, transno %u\n", req,
 572                atomic_read(&req->rq_refcount),
 573                rc, (unsigned) req->rq_transno);
 574
 575         if (rc == -ENOENT) {
 576                 /*
 577                  * we tried to destroy object or update attributes,
 578                  * but object doesn't exist anymore - cancell llog record
 579                  */
 580                 LASSERT(req->rq_transno == 0);
 581                 LASSERT(list_empty(&jra->jra_committed_link));
 582
 583                 ptlrpc_request_addref(req);
 584
 585                 spin_lock(&d->opd_sync_lock);
 586                 list_add(&jra->jra_committed_link,
 587                          &d->opd_sync_committed_there);
 588                 spin_unlock(&d->opd_sync_lock);
 589
 590                 wake_up(&d->opd_sync_waitq);
 591         } else if (rc) {
 592                 struct obd_import *imp = req->rq_import;
 593                 /*
 594                  * error happened, we'll try to repeat on next boot ?
 595                  */
 596                 LASSERTF(req->rq_transno == 0 || rc == -EIO || rc == -EROFS ||
 597                          req->rq_import_generation < imp->imp_generation,
 598                          "transno %llu, rc %d, gen: req %d, imp %d\n",
 599                          req->rq_transno, rc, req->rq_import_generation,
 600                          imp->imp_generation);
 601                 if (req->rq_transno == 0) {
 602                         /* this is the last time we see the request
 603                          * if transno is not zero, then commit cb
 604                          * will be called at some point */
 605                         LASSERT(atomic_read(&d->opd_sync_rpcs_in_progress) > 0);
 606                         atomic_dec(&d->opd_sync_rpcs_in_progress);
 607                 }
 608
 609                 wake_up(&d->opd_sync_waitq);
 610         } else if (d->opd_pre != NULL &&
 611                    unlikely(d->opd_pre_status == -ENOSPC)) {
 612                 /*
 613                  * if current status is -ENOSPC (lack of free space on OST)
 614                  * then we should poll OST immediately once object destroy
 615                  * is replied
 616                  */
 617                 osp_statfs_need_now(d);
 618         }
 619
 620         spin_lock(&d->opd_sync_lock);
 621         list_del_init(&jra->jra_in_flight_link);
 622         spin_unlock(&d->opd_sync_lock);
 623         LASSERT(atomic_read(&d->opd_sync_rpcs_in_flight) > 0);
 624         atomic_dec(&d->opd_sync_rpcs_in_flight);
 625         if (unlikely(atomic_read(&d->opd_sync_barrier) > 0))
 626                 wake_up(&d->opd_sync_barrier_waitq);
 627         CDEBUG(D_OTHER, "%s: %d in flight, %d in progress\n",
 628                d->opd_obd->obd_name, atomic_read(&d->opd_sync_rpcs_in_flight),
 629                atomic_read(&d->opd_sync_rpcs_in_progress));
 630
 631         osp_sync_check_for_work(d);
 632
 633         return 0;
 634 }
 635
 636 /*
 637  ** Add request to ptlrpc queue.
 638  *
 639  * This is just a tiny helper function to put the request on the sending list
 640  *
 641  * \param[in] d         OSP device
 642  * \param[in] llh       llog handle where the record is stored
 643  * \param[in] h         llog record
 644  * \param[in] req       request
 645  */
 646 static void osp_sync_send_new_rpc(struct osp_device *d,
 647                                   struct llog_handle *llh,
 648                                   struct llog_rec_hdr *h,
 649                                   struct ptlrpc_request *req)
 650 {
 651         struct osp_job_req_args *jra;
 652
 653         LASSERT(atomic_read(&d->opd_sync_rpcs_in_flight) <=
 654                 d->opd_sync_max_rpcs_in_flight);
 655
 656         jra = ptlrpc_req_async_args(jra, req);
 657         jra->jra_magic = OSP_JOB_MAGIC;
 658         jra->jra_lcookie.lgc_lgl = llh->lgh_id;
 659         jra->jra_lcookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
 660         jra->jra_lcookie.lgc_index = h->lrh_index;
 661         INIT_LIST_HEAD(&jra->jra_committed_link);
 662         spin_lock(&d->opd_sync_lock);
 663         list_add_tail(&jra->jra_in_flight_link, &d->opd_sync_in_flight_list);
 664         spin_unlock(&d->opd_sync_lock);
 665
 666         ptlrpcd_add_req(req);
 667 }
 668
 669
 670 /**
 671  * Allocate and prepare RPC for a new change.
 672  *
 673  * The function allocates and initializes an RPC which will be sent soon to
 674  * apply the change to the target OST. The request is initialized from the
 675  * llog record passed. Notice only the fields common to all type of changes
 676  * are initialized.
 677  *
 678  * \param[in] d         OSP device
 679  * \param[in] op        type of the change
 680  * \param[in] format    request format to be used
 681  *
 682  * \retval pointer              new request on success
 683  * \retval ERR_PTR(errno)       on error
 684  */
 685 static struct ptlrpc_request *osp_sync_new_job(struct osp_device *d,
 686                                                enum ost_cmd op,
 687                                                const struct req_format *format)
 688 {
 689         struct ptlrpc_request   *req;
 690         struct obd_import       *imp;
 691         int                      rc;
 692
 693         /* Prepare the request */
 694         imp = d->opd_obd->u.cli.cl_import;
 695         LASSERT(imp);
 696
 697         if (OBD_FAIL_CHECK(OBD_FAIL_OSP_CHECK_ENOMEM))
 698                 RETURN(ERR_PTR(-ENOMEM));
 699
 700         req = ptlrpc_request_alloc(imp, format);
 701         if (req == NULL)
 702                 RETURN(ERR_PTR(-ENOMEM));
 703
 704         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, op);
 705         if (rc) {
 706                 ptlrpc_req_finished(req);
 707                 return ERR_PTR(rc);
 708         }
 709
 710         req->rq_interpret_reply = osp_sync_interpret;
 711         req->rq_commit_cb = osp_sync_request_commit_cb;
 712         req->rq_cb_data = d;
 713
 714         ptlrpc_request_set_replen(req);
 715
 716         return req;
 717 }
 718
 719 /**
 720  * Generate a request for setattr change.
 721  *
 722  * The function prepares a new RPC, initializes it with setattr specific
 723  * bits and send the RPC.
 724  *
 725  * \param[in] d         OSP device
 726  * \param[in] llh       llog handle where the record is stored
 727  * \param[in] h         llog record
 728  *
 729  * \retval 0            on success
 730  * \retval 1            on invalid record
 731  * \retval negative     negated errno on error
 732  */
 733 static int osp_sync_new_setattr_job(struct osp_device *d,
 734                                     struct llog_handle *llh,
 735                                     struct llog_rec_hdr *h)
 736 {
 737         struct llog_setattr64_rec       *rec = (struct llog_setattr64_rec *)h;
 738         struct ptlrpc_request           *req;
 739         struct ost_body                 *body;
 740
 741         ENTRY;
 742         LASSERT(h->lrh_type == MDS_SETATTR64_REC);
 743
 744         if (OBD_FAIL_CHECK(OBD_FAIL_OSP_CHECK_INVALID_REC))
 745                 RETURN(1);
 746
 747         /* lsr_valid can only be 0 or HAVE OBD_MD_{FLUID, FLGID, FLPROJID} set,
 748          * so no bits other than these should be set. */
 749         if ((rec->lsr_valid & ~(OBD_MD_FLUID | OBD_MD_FLGID |
 750             OBD_MD_FLPROJID | OBD_MD_LAYOUT_VERSION)) != 0) {
 751                 CERROR("%s: invalid setattr record, lsr_valid:%llu\n",
 752                         d->opd_obd->obd_name, rec->lsr_valid);
 753                 /* return 1 on invalid record */
 754                 RETURN(1);
 755         }
 756
 757         req = osp_sync_new_job(d, OST_SETATTR, &RQF_OST_SETATTR);
 758         if (IS_ERR(req))
 759                 RETURN(PTR_ERR(req));
 760
 761         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 762         LASSERT(body);
 763         body->oa.o_oi = rec->lsr_oi;
 764         body->oa.o_uid = rec->lsr_uid;
 765         body->oa.o_gid = rec->lsr_gid;
 766         body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID;
 767         if (h->lrh_len > sizeof(struct llog_setattr64_rec)) {
 768                 struct llog_setattr64_rec_v2 *rec_v2 = (typeof(rec_v2))rec;
 769                 body->oa.o_projid = rec_v2->lsr_projid;
 770                 body->oa.o_layout_version = rec_v2->lsr_layout_version;
 771         }
 772
 773         /* old setattr record (prior 2.6.0) doesn't have 'valid' stored,
 774          * we assume that both UID and GID are valid in that case. */
 775         if (rec->lsr_valid == 0)
 776                 body->oa.o_valid |= (OBD_MD_FLUID | OBD_MD_FLGID);
 777         else
 778                 body->oa.o_valid |= rec->lsr_valid;
 779
 780         if (body->oa.o_valid & OBD_MD_LAYOUT_VERSION) {
 781                 OBD_FAIL_TIMEOUT(OBD_FAIL_FLR_LV_DELAY, cfs_fail_val);
 782                 if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_FLR_LV_INC)))
 783                         ++body->oa.o_layout_version;
 784         }
 785
 786         osp_sync_send_new_rpc(d, llh, h, req);
 787         RETURN(0);
 788 }
 789
 790 /**
 791  * Generate a request for unlink change.
 792  *
 793  * The function prepares a new RPC, initializes it with unlink(destroy)
 794  * specific bits and sends the RPC. The function is used to handle
 795  * llog_unlink_rec which were used in the older versions of Lustre.
 796  * Current version uses llog_unlink_rec64.
 797  *
 798  * \param[in] d         OSP device
 799  * \param[in] llh       llog handle where the record is stored
 800  * \param[in] h         llog record
 801  *
 802  * \retval 0            on success
 803  * \retval negative     negated errno on error
 804  */
 805 static int osp_sync_new_unlink_job(struct osp_device *d,
 806                                    struct llog_handle *llh,
 807                                    struct llog_rec_hdr *h)
 808 {
 809         struct llog_unlink_rec  *rec = (struct llog_unlink_rec *)h;
 810         struct ptlrpc_request   *req;
 811         struct ost_body         *body;
 812         int rc;
 813
 814         ENTRY;
 815         LASSERT(h->lrh_type == MDS_UNLINK_REC);
 816
 817         req = osp_sync_new_job(d, OST_DESTROY, &RQF_OST_DESTROY);
 818         if (IS_ERR(req))
 819                 RETURN(PTR_ERR(req));
 820
 821         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 822         LASSERT(body);
 823         ostid_set_seq(&body->oa.o_oi, rec->lur_oseq);
 824         rc = ostid_set_id(&body->oa.o_oi, rec->lur_oid);
 825         if (rc)
 826                 return rc;
 827         body->oa.o_misc = rec->lur_count;
 828         body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID;
 829         if (rec->lur_count)
 830                 body->oa.o_valid |= OBD_MD_FLOBJCOUNT;
 831
 832         osp_sync_send_new_rpc(d, llh, h, req);
 833         RETURN(0);
 834 }
 835
 836 /**
 837  * Generate a request for unlink change.
 838  *
 839  * The function prepares a new RPC, initializes it with unlink(destroy)
 840  * specific bits and sends the RPC. Depending on the target (MDT or OST)
 841  * two different protocols are used. For MDT we use OUT (basically OSD API
 842  * updates transferred via a network). For OST we still use the old
 843  * protocol (OBD?), originally for compatibility. Later we can start to
 844  * use OUT for OST as well, this will allow batching and better code
 845  * unification.
 846  *
 847  * \param[in] d         OSP device
 848  * \param[in] llh       llog handle where the record is stored
 849  * \param[in] h         llog record
 850  *
 851  * \retval 0            on success
 852  * \retval negative     negated errno on error
 853  */
 854 static int osp_sync_new_unlink64_job(struct osp_device *d,
 855                                      struct llog_handle *llh,
 856                                      struct llog_rec_hdr *h)
 857 {
 858         struct llog_unlink64_rec        *rec = (struct llog_unlink64_rec *)h;
 859         struct ptlrpc_request           *req = NULL;
 860         struct ost_body                 *body;
 861         int                              rc;
 862
 863         ENTRY;
 864         LASSERT(h->lrh_type == MDS_UNLINK64_REC);
 865         req = osp_sync_new_job(d, OST_DESTROY, &RQF_OST_DESTROY);
 866         if (IS_ERR(req))
 867                 RETURN(PTR_ERR(req));
 868
 869         body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
 870         if (body == NULL)
 871                 RETURN(-EFAULT);
 872         rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
 873         if (rc < 0)
 874                 RETURN(rc);
 875         body->oa.o_misc = rec->lur_count;
 876         body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
 877                            OBD_MD_FLOBJCOUNT;
 878         osp_sync_send_new_rpc(d, llh, h, req);
 879         RETURN(0);
 880 }
 881
 882 /**
 883  * Process llog records.
 884  *
 885  * This function is called to process the llog records committed locally.
 886  * In the recovery model used by OSP we can apply a change to a remote
 887  * target once corresponding transaction (like posix unlink) is committed
 888  * locally so can't revert.
 889  * Depending on the llog record type, a given handler is called that is
 890  * responsible for preparing and sending the RPC to apply the change.
 891  * Special record type LLOG_GEN_REC marking a reboot is cancelled right away.
 892  *
 893  * \param[in] env       LU environment provided by the caller
 894  * \param[in] d         OSP device
 895  * \param[in] llh       llog handle where the record is stored
 896  * \param[in] rec       llog record
 897  */
 898 static void osp_sync_process_record(const struct lu_env *env,
 899                                     struct osp_device *d,
 900                                     struct llog_handle *llh,
 901                                     struct llog_rec_hdr *rec)
 902 {
 903         struct llog_handle      *cathandle = llh->u.phd.phd_cat_handle;
 904         struct llog_cookie       cookie;
 905         int                      rc = 0;
 906
 907         ENTRY;
 908
 909         cookie.lgc_lgl = llh->lgh_id;
 910         cookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
 911         cookie.lgc_index = rec->lrh_index;
 912
 913         d->opd_sync_last_catalog_idx = llh->lgh_hdr->llh_cat_idx;
 914
 915         if (unlikely(rec->lrh_type == LLOG_GEN_REC)) {
 916                 struct llog_gen_rec *gen = (struct llog_gen_rec *)rec;
 917
 918                 /* we're waiting for the record generated by this instance */
 919                 LASSERT(d->opd_sync_prev_done == 0);
 920                 if (!memcmp(&d->opd_sync_generation, &gen->lgr_gen,
 921                             sizeof(gen->lgr_gen))) {
 922                         CDEBUG(D_HA, "processed all old entries\n");
 923                         d->opd_sync_prev_done = 1;
 924                 }
 925
 926                 /* cancel any generation record */
 927                 rc = llog_cat_cancel_records(env, cathandle, 1, &cookie);
 928
 929                 /* flush all pending records ASAP */
 930                 osp_sync_force(env, d);
 931
 932                 RETURN_EXIT;
 933         }
 934
 935         /*
 936          * now we prepare and fill requests to OST, put them on the queue
 937          * and fire after next commit callback
 938          */
 939
 940         /* notice we increment counters before sending RPC, to be consistent
 941          * in RPC interpret callback which may happen very quickly */
 942         atomic_inc(&d->opd_sync_rpcs_in_flight);
 943         atomic_inc(&d->opd_sync_rpcs_in_progress);
 944
 945         switch (rec->lrh_type) {
 946         /* case MDS_UNLINK_REC is kept for compatibility */
 947         case MDS_UNLINK_REC:
 948                 rc = osp_sync_new_unlink_job(d, llh, rec);
 949                 break;
 950         case MDS_UNLINK64_REC:
 951                 rc = osp_sync_new_unlink64_job(d, llh, rec);
 952                 break;
 953         case MDS_SETATTR64_REC:
 954                 rc = osp_sync_new_setattr_job(d, llh, rec);
 955                 break;
 956         default:
 957                 CERROR("%s: unknown record type: %x\n", d->opd_obd->obd_name,
 958                        rec->lrh_type);
 959                 /* treat "unknown record type" as "invalid" */
 960                 rc = 1;
 961                 break;
 962         }
 963
 964         /* For all kinds of records, not matter successful or not,
 965          * we should decrease changes and bump last_processed_id.
 966          */
 967         if (d->opd_sync_prev_done) {
 968                 LASSERT(atomic_read(&d->opd_sync_changes) > 0);
 969                 atomic_dec(&d->opd_sync_changes);
 970                 wake_up(&d->opd_sync_barrier_waitq);
 971         }
 972         atomic64_inc(&d->opd_sync_processed_recs);
 973         if (rc != 0) {
 974                 atomic_dec(&d->opd_sync_rpcs_in_flight);
 975                 atomic_dec(&d->opd_sync_rpcs_in_progress);
 976         }
 977
 978         CDEBUG(D_OTHER, "%s: %d in flight, %d in progress\n",
 979                d->opd_obd->obd_name, atomic_read(&d->opd_sync_rpcs_in_flight),
 980                atomic_read(&d->opd_sync_rpcs_in_progress));
 981
 982         /* Delete the invalid record */
 983         if (rc == 1) {
 984                 rc = llog_cat_cancel_records(env, cathandle, 1, &cookie);
 985                 if (rc != 0)
 986                         CERROR("%s: can't delete invalid record: "
 987                                "fid = "DFID", rec_id = %u, rc = %d\n",
 988                                d->opd_obd->obd_name,
 989                                PFID(lu_object_fid(&cathandle->lgh_obj->do_lu)),
 990                                rec->lrh_id, rc);
 991         }
 992
 993         CDEBUG(D_HA, "found record %x, %d, idx %u, id %u\n",
 994                rec->lrh_type, rec->lrh_len, rec->lrh_index, rec->lrh_id);
 995
 996         RETURN_EXIT;
 997 }
 998
 999 /**
1000  * Cancel llog records for the committed changes.
1001  *
1002  * The function walks through the list of the committed RPCs and cancels
1003  * corresponding llog records. see osp_sync_request_commit_cb() for the
1004  * details.
1005  *
1006  * \param[in] env       LU environment provided by the caller
1007  * \param[in] d         OSP device
1008  */
1009 static void osp_sync_process_committed(const struct lu_env *env,
1010                                        struct osp_device *d)
1011 {
1012         struct obd_device       *obd = d->opd_obd;
1013         struct obd_import       *imp = obd->u.cli.cl_import;
1014         struct ost_body         *body;
1015         struct ptlrpc_request   *req;
1016         struct llog_ctxt        *ctxt;
1017         struct llog_handle      *llh;
1018         int                     *arr;
1019         LIST_HEAD(list);
1020         struct list_head         *le;
1021         struct llog_logid        lgid;
1022         int                      rc, i, count = 0, done = 0;
1023
1024         ENTRY;
1025
1026         if (list_empty(&d->opd_sync_committed_there))
1027                 return;
1028
1029         /*
1030          * if current status is -ENOSPC (lack of free space on OST)
1031          * then we should poll OST immediately once object destroy
1032          * is committed.
1033          * notice: we do this upon commit as well because some backends
1034          * (like DMU) do not release space right away.
1035          */
1036         if (d->opd_pre != NULL && unlikely(d->opd_pre_status == -ENOSPC))
1037                 osp_statfs_need_now(d);
1038
1039         /*
1040          * now cancel them all
1041          * XXX: can we improve this using some batching?
1042          *      with batch RPC that'll happen automatically?
1043          * XXX: can we store ctxt in lod_device and save few cycles ?
1044          */
1045         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
1046         LASSERT(ctxt);
1047
1048         llh = ctxt->loc_handle;
1049         LASSERT(llh);
1050
1051         spin_lock(&d->opd_sync_lock);
1052         list_splice(&d->opd_sync_committed_there, &list);
1053         INIT_LIST_HEAD(&d->opd_sync_committed_there);
1054         spin_unlock(&d->opd_sync_lock);
1055
1056         list_for_each(le, &list)
1057                 count++;
1058         if (count > 2)
1059                 OBD_ALLOC_WAIT(arr, sizeof(int) * count);
1060         else
1061                 arr = NULL;
1062         i = 0;
1063         while (!list_empty(&list)) {
1064                 struct osp_job_req_args *jra;
1065
1066                 jra = list_entry(list.next, struct osp_job_req_args,
1067                                  jra_committed_link);
1068                 LASSERT(jra->jra_magic == OSP_JOB_MAGIC);
1069                 list_del_init(&jra->jra_committed_link);
1070
1071                 req = container_of((void *)jra, struct ptlrpc_request,
1072                                    rq_async_args);
1073                 body = req_capsule_client_get(&req->rq_pill,
1074                                               &RMF_OST_BODY);
1075                 LASSERT(body);
1076                 /* import can be closing, thus all commit cb's are
1077                  * called we can check committness directly */
1078                 if (req->rq_import_generation == imp->imp_generation) {
1079                         if (arr && (!i ||
1080                                     !memcmp(&jra->jra_lcookie.lgc_lgl, &lgid,
1081                                            sizeof(lgid)))) {
1082                                 if (unlikely(!i))
1083                                         lgid = jra->jra_lcookie.lgc_lgl;
1084
1085                                 arr[i++] = jra->jra_lcookie.lgc_index;
1086                         } else {
1087                                 rc = llog_cat_cancel_records(env, llh, 1,
1088                                                              &jra->jra_lcookie);
1089                                 if (rc)
1090                                         CERROR("%s: can't cancel record: %d\n",
1091                                                obd->obd_name, rc);
1092                         }
1093                 } else {
1094                         DEBUG_REQ(D_OTHER, req, "imp_committed = %llu",
1095                                   imp->imp_peer_committed_transno);
1096                 }
1097                 ptlrpc_req_finished(req);
1098                 done++;
1099         }
1100         if (arr && i > 0) {
1101                 rc = llog_cat_cancel_arr_rec(env, llh, &lgid, i, arr);
1102
1103                 if (rc)
1104                         CERROR("%s: can't cancel %d records rc: %d\n",
1105                                obd->obd_name, i, rc);
1106                 else
1107                         CDEBUG(D_OTHER, "%s: massive records cancel id "DFID\
1108                                " num %d\n", obd->obd_name,
1109                                PFID(&lgid.lgl_oi.oi_fid), i);
1110         }
1111         if (arr)
1112                 OBD_FREE(arr, sizeof(int) * count);
1113
1114         llog_ctxt_put(ctxt);
1115
1116         LASSERT(atomic_read(&d->opd_sync_rpcs_in_progress) >= done);
1117         atomic_sub(done, &d->opd_sync_rpcs_in_progress);
1118         CDEBUG(D_OTHER, "%s: %d in flight, %d in progress, done %d\n",
1119                d->opd_obd->obd_name, atomic_read(&d->opd_sync_rpcs_in_flight),
1120                atomic_read(&d->opd_sync_rpcs_in_progress), done);
1121
1122         osp_sync_check_for_work(d);
1123
1124         /* wake up the thread if requested to stop:
1125          * it might be waiting for in-progress to complete */
1126         if (unlikely(osp_sync_running(d) == 0))
1127                 wake_up(&d->opd_sync_waitq);
1128
1129         EXIT;
1130 }
1131
1132 /**
1133  * The core of the syncing mechanism.
1134  *
1135  * This is a callback called by the llog processing function. Essentially it
1136  * suspends llog processing until there is a record to process (it's supposed
1137  * to be committed locally). The function handles RPCs committed by the target
1138  * and cancels corresponding llog records.
1139  *
1140  * \param[in] env       LU environment provided by the caller
1141  * \param[in] llh       llog handle we're processing
1142  * \param[in] rec       current llog record
1143  * \param[in] data      callback data containing a pointer to the device
1144  *
1145  * \retval 0                    to ask the caller (llog_process()) to continue
1146  * \retval LLOG_PROC_BREAK      to ask the caller to break
1147  */
1148 static int osp_sync_process_queues(const struct lu_env *env,
1149                                    struct llog_handle *llh,
1150                                    struct llog_rec_hdr *rec,
1151                                    void *data)
1152 {
1153         struct osp_device       *d = data;
1154
1155         do {
1156                 if (!osp_sync_running(d)) {
1157                         CDEBUG(D_HA, "stop llog processing\n");
1158                         return LLOG_PROC_BREAK;
1159                 }
1160
1161                 /* process requests committed by OST */
1162                 osp_sync_process_committed(env, d);
1163
1164                 /* if we there are changes to be processed and we have
1165                  * resources for this ... do now */
1166                 if (osp_sync_can_process_new(d, rec)) {
1167                         if (llh == NULL) {
1168                                 /* ask llog for another record */
1169                                 CDEBUG(D_HA, "%u changes, %u in progress,"
1170                                      " %u in flight\n",
1171                                      atomic_read(&d->opd_sync_changes),
1172                                      atomic_read(&d->opd_sync_rpcs_in_progress),
1173                                      atomic_read(&d->opd_sync_rpcs_in_flight));
1174                                 return 0;
1175                         }
1176                         osp_sync_process_record(env, d, llh, rec);
1177                         llh = NULL;
1178                         rec = NULL;
1179                 }
1180                 if (OBD_FAIL_PRECHECK(OBD_FAIL_CATALOG_FULL_CHECK) &&
1181                             cfs_fail_val != 1)
1182                         msleep(1 * MSEC_PER_SEC);
1183
1184                 wait_event_idle(d->opd_sync_waitq,
1185                                 !osp_sync_running(d) ||
1186                                 osp_sync_can_process_new(d, rec) ||
1187                                 !list_empty(&d->opd_sync_committed_there));
1188         } while (1);
1189 }
1190
1191 /**
1192  * OSP sync thread.
1193  *
1194  * This thread runs llog_cat_process() scanner calling our callback
1195  * to process llog records. in the callback we implement tricky
1196  * state machine as we don't want to start scanning of the llog again
1197  * and again, also we don't want to process too many records and send
1198  * too many RPCs a time. so, depending on current load (num of changes
1199  * being synced to OST) the callback can suspend awaiting for some
1200  * new conditions, like syncs completed.
1201  *
1202  * In order to process llog records left by previous boots and to allow
1203  * llog_process_thread() to find something (otherwise it'd just exit
1204  * immediately) we add a special GENERATATION record on each boot.
1205  *
1206  * \param[in] _arg      a pointer to thread's arguments
1207  *
1208  * \retval 0            on success
1209  * \retval negative     negated errno on error
1210  */
1211 static int osp_sync_thread(void *_arg)
1212 {
1213         struct osp_device       *d = _arg;
1214         struct ptlrpc_thread    *thread = &d->opd_sync_thread;
1215         struct llog_ctxt        *ctxt;
1216         struct obd_device       *obd = d->opd_obd;
1217         struct llog_handle      *llh;
1218         struct lu_env            env;
1219         int                      rc, count;
1220         bool                     wrapped;
1221
1222         ENTRY;
1223
1224         rc = lu_env_init(&env, LCT_LOCAL);
1225         if (rc) {
1226                 CERROR("%s: can't initialize env: rc = %d\n",
1227                        obd->obd_name, rc);
1228
1229                 spin_lock(&d->opd_sync_lock);
1230                 thread->t_flags = SVC_STOPPED;
1231                 spin_unlock(&d->opd_sync_lock);
1232                 wake_up(&thread->t_ctl_waitq);
1233
1234                 RETURN(rc);
1235         }
1236
1237         spin_lock(&d->opd_sync_lock);
1238         thread->t_flags = SVC_RUNNING;
1239         spin_unlock(&d->opd_sync_lock);
1240         wake_up(&thread->t_ctl_waitq);
1241
1242 again:
1243         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
1244         if (ctxt == NULL) {
1245                 CERROR("can't get appropriate context\n");
1246                 GOTO(out, rc = -EINVAL);
1247         }
1248
1249         llh = ctxt->loc_handle;
1250         if (llh == NULL) {
1251                 CERROR("can't get llh\n");
1252                 llog_ctxt_put(ctxt);
1253                 GOTO(out, rc = -EINVAL);
1254         }
1255
1256         /*
1257          * Catalog processing stops when it processed last catalog record
1258          * with index equal to the end of catalog bitmap. Or if it is wrapped,
1259          * processing stops with index equal to the lgh_last_idx. We need to
1260          * continue processing.
1261          */
1262         d->opd_sync_last_catalog_idx = 0;
1263         do {
1264                 int     size;
1265
1266                 wrapped = (llh->lgh_hdr->llh_cat_idx >= llh->lgh_last_idx &&
1267                            llh->lgh_hdr->llh_count > 1);
1268
1269                 if (OBD_FAIL_CHECK(OBD_FAIL_OSP_CANT_PROCESS_LLOG)) {
1270                         rc = -EINPROGRESS;
1271                         goto next;
1272                 }
1273                 rc = llog_cat_process(&env, llh, osp_sync_process_queues, d,
1274                                       d->opd_sync_last_catalog_idx, 0);
1275
1276 next:
1277                 size = OBD_FAIL_PRECHECK(OBD_FAIL_CAT_RECORDS) ?
1278                        cfs_fail_val : (LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - 1);
1279                 /* processing reaches catalog bottom */
1280                 if (d->opd_sync_last_catalog_idx == size)
1281                         d->opd_sync_last_catalog_idx = LLOG_CAT_FIRST;
1282                 /* If catalog is wrapped we can`t predict last index of
1283                  * processing because lgh_last_idx could be changed.
1284                  * Starting form the next one. Index would be increased
1285                  * at llog_process_thread
1286                  */
1287         } while (rc == 0 && (wrapped ||
1288                              d->opd_sync_last_catalog_idx == LLOG_CAT_FIRST));
1289
1290         if (rc < 0) {
1291                 if (rc == -EINPROGRESS) {
1292                         /* can't access the llog now - OI scrub is trying to fix
1293                          * underlying issue. let's wait and try again */
1294                         llog_cat_close(&env, llh);
1295                         rc = llog_cleanup(&env, ctxt);
1296                         if (rc)
1297                                 GOTO(out, rc);
1298                         schedule_timeout_interruptible(cfs_time_seconds(5));
1299                         goto again;
1300                 }
1301
1302                 CERROR("%s: llog process with osp_sync_process_queues "
1303                        "failed: %d\n", d->opd_obd->obd_name, rc);
1304                 GOTO(close, rc);
1305         }
1306         LASSERTF(rc == 0 || rc == LLOG_PROC_BREAK,
1307                  "%u changes, %u in progress, %u in flight: %d\n",
1308                  atomic_read(&d->opd_sync_changes),
1309                  atomic_read(&d->opd_sync_rpcs_in_progress),
1310                  atomic_read(&d->opd_sync_rpcs_in_flight), rc);
1311
1312         /* we don't expect llog_process_thread() to exit till umount */
1313         LASSERTF(thread->t_flags != SVC_RUNNING,
1314                  "%u changes, %u in progress, %u in flight\n",
1315                  atomic_read(&d->opd_sync_changes),
1316                  atomic_read(&d->opd_sync_rpcs_in_progress),
1317                  atomic_read(&d->opd_sync_rpcs_in_flight));
1318
1319         /* wait till all the requests are completed */
1320         count = 0;
1321         while (atomic_read(&d->opd_sync_rpcs_in_progress) > 0) {
1322                 osp_sync_process_committed(&env, d);
1323
1324                 rc = wait_event_idle_timeout(
1325                         d->opd_sync_waitq,
1326                         atomic_read(&d->opd_sync_rpcs_in_progress) == 0,
1327                         cfs_time_seconds(5));
1328                 if (rc == 0)
1329                         count++;
1330                 LASSERTF(count < 10, "%s: %d %d %sempty\n",
1331                          d->opd_obd->obd_name,
1332                          atomic_read(&d->opd_sync_rpcs_in_progress),
1333                          atomic_read(&d->opd_sync_rpcs_in_flight),
1334                          list_empty(&d->opd_sync_committed_there) ? "" : "!");
1335
1336         }
1337
1338 close:
1339         llog_cat_close(&env, llh);
1340         rc = llog_cleanup(&env, ctxt);
1341         if (rc)
1342                 CERROR("can't cleanup llog: %d\n", rc);
1343 out:
1344         LASSERTF(atomic_read(&d->opd_sync_rpcs_in_progress) == 0,
1345                  "%s: %d %d %sempty\n", d->opd_obd->obd_name,
1346                  atomic_read(&d->opd_sync_rpcs_in_progress),
1347                  atomic_read(&d->opd_sync_rpcs_in_flight),
1348                  list_empty(&d->opd_sync_committed_there) ? "" : "!");
1349
1350         thread->t_flags = SVC_STOPPED;
1351
1352         wake_up(&thread->t_ctl_waitq);
1353
1354         lu_env_fini(&env);
1355
1356         RETURN(0);
1357 }
1358
1359 /**
1360  * Initialize llog.
1361  *
1362  * Initializes the llog. Specific llog to be used depends on the type of the
1363  * target OSP represents (OST or MDT). The function adds appends a new llog
1364  * record to mark the place where the records associated with this boot
1365  * start.
1366  *
1367  * \param[in] env       LU environment provided by the caller
1368  * \param[in] d         OSP device
1369  *
1370  * \retval 0            on success
1371  * \retval negative     negated errno on error
1372  */
1373 static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
1374 {
1375         struct osp_thread_info  *osi = osp_env_info(env);
1376         struct lu_fid           *fid = &osi->osi_fid;
1377         struct llog_handle      *lgh = NULL;
1378         struct obd_device       *obd = d->opd_obd;
1379         struct llog_ctxt        *ctxt;
1380         int                     rc;
1381
1382         ENTRY;
1383
1384         LASSERT(obd);
1385
1386         /*
1387          * open llog corresponding to our OST
1388          */
1389         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1390         obd->obd_lvfs_ctxt.dt = d->opd_storage;
1391
1392         lu_local_obj_fid(fid, LLOG_CATALOGS_OID);
1393
1394         rc = llog_osd_get_cat_list(env, d->opd_storage, d->opd_index, 1,
1395                                    &osi->osi_cid, fid);
1396         if (rc < 0) {
1397                 if (rc != -EFAULT) {
1398                         CERROR("%s: can't get id from catalogs: rc = %d\n",
1399                                obd->obd_name, rc);
1400                         RETURN(rc);
1401                 }
1402
1403                 /* After sparse OST indices is supported, the CATALOG file
1404                  * may become a sparse file that results in failure on
1405                  * reading. Skip this error as the llog will be created
1406                  * later */
1407                 memset(&osi->osi_cid, 0, sizeof(osi->osi_cid));
1408                 rc = 0;
1409         }
1410
1411         CDEBUG(D_INFO, "%s: Init llog for %d - catid "DFID":%x\n",
1412                obd->obd_name, d->opd_index,
1413                PFID(&osi->osi_cid.lci_logid.lgl_oi.oi_fid),
1414                osi->osi_cid.lci_logid.lgl_ogen);
1415
1416         rc = llog_setup(env, obd, &obd->obd_olg, LLOG_MDS_OST_ORIG_CTXT,
1417                         d->opd_storage->dd_lu_dev.ld_obd,
1418                         &llog_common_cat_ops);
1419         if (rc)
1420                 RETURN(rc);
1421
1422         ctxt = llog_get_context(obd, LLOG_MDS_OST_ORIG_CTXT);
1423         LASSERT(ctxt);
1424
1425         if (likely(logid_id(&osi->osi_cid.lci_logid) != 0)) {
1426                 struct lu_fid fid_temp;
1427
1428                 if (CFS_FAIL_CHECK(OBD_FAIL_OSP_INVALID_LOGID)) {
1429                         memset(&osi->osi_cid, 0, sizeof(osi->osi_cid));
1430                         logid_set_id(&osi->osi_cid.lci_logid, cfs_fail_val);
1431                 }
1432
1433                 logid_to_fid(&osi->osi_cid.lci_logid, &fid_temp);
1434                 if (fid_is_sane(&fid_temp)) {
1435                         rc = llog_open(env, ctxt, &lgh, &osi->osi_cid.lci_logid,
1436                                        NULL, LLOG_OPEN_EXISTS);
1437
1438                         /* re-create llog if it is missing */
1439                         if (rc == -ENOENT)
1440                                 logid_set_id(&osi->osi_cid.lci_logid, 0);
1441                         else if (rc < 0)
1442                                 GOTO(out_cleanup, rc);
1443                 } else {
1444                         CERROR("%s: the catid "DFID" for init llog %d is bad\n",
1445                                obd->obd_name, PFID(&fid_temp), d->opd_index);
1446
1447                         /* it will be recreated later */
1448                         logid_set_id(&osi->osi_cid.lci_logid, 0);
1449                 }
1450         }
1451
1452         if (unlikely(logid_id(&osi->osi_cid.lci_logid) == 0)) {
1453                 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
1454                 if (rc < 0)
1455                         GOTO(out_cleanup, rc);
1456                 osi->osi_cid.lci_logid = lgh->lgh_id;
1457         }
1458
1459         LASSERT(lgh != NULL);
1460         ctxt->loc_handle = lgh;
1461
1462         rc = llog_init_handle(env, lgh, LLOG_F_IS_CAT, NULL);
1463         if (rc)
1464                 GOTO(out_close, rc);
1465
1466         rc = llog_osd_put_cat_list(env, d->opd_storage, d->opd_index, 1,
1467                                    &osi->osi_cid, fid);
1468         if (rc)
1469                 GOTO(out_close, rc);
1470
1471         /*
1472          * put a mark in the llog till which we'll be processing
1473          * old records restless
1474          */
1475         d->opd_sync_generation.mnt_cnt = ktime_get_ns();
1476         d->opd_sync_generation.conn_cnt = ktime_get_ns();
1477
1478         osi->osi_hdr.lrh_type = LLOG_GEN_REC;
1479         osi->osi_hdr.lrh_len = sizeof(osi->osi_gen);
1480
1481         memcpy(&osi->osi_gen.lgr_gen, &d->opd_sync_generation,
1482                sizeof(osi->osi_gen.lgr_gen));
1483
1484         rc = llog_cat_add(env, lgh, &osi->osi_gen.lgr_hdr, &osi->osi_cookie);
1485         if (rc < 0)
1486                 GOTO(out_close, rc);
1487         llog_ctxt_put(ctxt);
1488         RETURN(0);
1489 out_close:
1490         llog_cat_close(env, lgh);
1491 out_cleanup:
1492         llog_cleanup(env, ctxt);
1493         RETURN(rc);
1494 }
1495
1496 /**
1497  * Cleanup llog used for syncing.
1498  *
1499  * Closes and cleanups the llog. The function is called when the device is
1500  * shutting down.
1501  *
1502  * \param[in] env       LU environment provided by the caller
1503  * \param[in] d         OSP device
1504  */
1505 static void osp_sync_llog_fini(const struct lu_env *env, struct osp_device *d)
1506 {
1507         struct llog_ctxt *ctxt;
1508
1509         ctxt = llog_get_context(d->opd_obd, LLOG_MDS_OST_ORIG_CTXT);
1510         if (ctxt) {
1511                 llog_cat_close(env, ctxt->loc_handle);
1512                 llog_cleanup(env, ctxt);
1513         }
1514 }
1515
1516 /**
1517  * Initialization of the sync component of OSP.
1518  *
1519  * Initializes the llog and starts a new thread to handle the changes to
1520  * the remote target (OST or MDT).
1521  *
1522  * \param[in] env       LU environment provided by the caller
1523  * \param[in] d         OSP device
1524  *
1525  * \retval 0            on success
1526  * \retval negative     negated errno on error
1527  */
1528 int osp_sync_init(const struct lu_env *env, struct osp_device *d)
1529 {
1530         struct task_struct      *task;
1531         int                      rc;
1532
1533         ENTRY;
1534
1535         d->opd_sync_max_rpcs_in_flight = OSP_MAX_RPCS_IN_FLIGHT;
1536         d->opd_sync_max_rpcs_in_progress = OSP_MAX_RPCS_IN_PROGRESS;
1537         spin_lock_init(&d->opd_sync_lock);
1538         init_waitqueue_head(&d->opd_sync_waitq);
1539         init_waitqueue_head(&d->opd_sync_barrier_waitq);
1540         thread_set_flags(&d->opd_sync_thread, SVC_INIT);
1541         init_waitqueue_head(&d->opd_sync_thread.t_ctl_waitq);
1542         INIT_LIST_HEAD(&d->opd_sync_in_flight_list);
1543         INIT_LIST_HEAD(&d->opd_sync_committed_there);
1544
1545         if (d->opd_storage->dd_rdonly)
1546                 RETURN(0);
1547
1548         /*
1549          * initialize llog storing changes
1550          */
1551         rc = osp_sync_llog_init(env, d);
1552         if (rc) {
1553                 CERROR("%s: can't initialize llog: rc = %d\n",
1554                        d->opd_obd->obd_name, rc);
1555                 GOTO(err_id, rc);
1556         }
1557
1558         /*
1559          * Start synchronization thread
1560          */
1561         task = kthread_run(osp_sync_thread, d, "osp-syn-%u-%u",
1562                            d->opd_index, d->opd_group);
1563         if (IS_ERR(task)) {
1564                 rc = PTR_ERR(task);
1565                 CERROR("%s: cannot start sync thread: rc = %d\n",
1566                        d->opd_obd->obd_name, rc);
1567                 GOTO(err_llog, rc);
1568         }
1569
1570         wait_event_idle(d->opd_sync_thread.t_ctl_waitq,
1571                         osp_sync_running(d) || osp_sync_stopped(d));
1572
1573         RETURN(0);
1574 err_llog:
1575         osp_sync_llog_fini(env, d);
1576 err_id:
1577         return rc;
1578 }
1579
1580 /**
1581  * Stop the syncing thread.
1582  *
1583  * Asks the syncing thread to stop and wait until it's stopped.
1584  *
1585  * \param[in] d         OSP device
1586  *
1587  * \retval              0
1588  */
1589 int osp_sync_fini(struct osp_device *d)
1590 {
1591         struct ptlrpc_thread *thread = &d->opd_sync_thread;
1592
1593         ENTRY;
1594
1595         if (!thread_is_init(thread) && !thread_is_stopped(thread)) {
1596                 thread->t_flags = SVC_STOPPING;
1597                 wake_up(&d->opd_sync_waitq);
1598                 wait_event(thread->t_ctl_waitq, thread_is_stopped(thread));
1599         }
1600
1601         RETURN(0);
1602 }
1603
1604 struct osp_last_committed_cb {
1605         struct dt_txn_commit_cb  ospc_cb;
1606         struct osp_device       *ospc_dev;
1607         __u64                    ospc_transno;
1608 };
1609
1610 void osp_sync_local_commit_cb(struct lu_env *env, struct thandle *th,
1611                               struct dt_txn_commit_cb *dcb, int err)
1612 {
1613         struct osp_last_committed_cb    *cb;
1614         struct osp_device               *d;
1615
1616         cb = container_of0(dcb, struct osp_last_committed_cb, ospc_cb);
1617         d = cb->ospc_dev;
1618
1619         CDEBUG(D_HA, "%s: %llu committed\n", d->opd_obd->obd_name,
1620                cb->ospc_transno);
1621
1622         spin_lock(&d->opd_sync_lock);
1623         if (cb->ospc_transno > d->opd_sync_last_committed_id)
1624                 d->opd_sync_last_committed_id = cb->ospc_transno;
1625         spin_unlock(&d->opd_sync_lock);
1626
1627         osp_sync_check_for_work(d);
1628         lu_device_put(osp2lu_dev(d));
1629         if (atomic_dec_and_test(&d->opd_commits_registered))
1630                 wake_up(&d->opd_sync_waitq);
1631
1632         OBD_FREE_PTR(cb);
1633 }
1634
1635 static int osp_sync_add_commit_cb(const struct lu_env *env,
1636                                   struct osp_device *d, struct thandle *th)
1637 {
1638         struct osp_last_committed_cb    *cb;
1639         struct dt_txn_commit_cb         *dcb;
1640         int                              rc = 0;
1641
1642         OBD_ALLOC_PTR(cb);
1643         if (cb == NULL)
1644                 return -ENOMEM;
1645         cb->ospc_dev = d;
1646         dcb = &cb->ospc_cb;
1647         dcb->dcb_func = osp_sync_local_commit_cb;
1648         spin_lock(&d->opd_sync_lock);
1649         cb->ospc_transno = ++d->opd_sync_last_used_id;
1650         spin_unlock(&d->opd_sync_lock);
1651
1652         rc = dt_trans_cb_add(th, dcb);
1653         CDEBUG(D_HA, "%s: add commit cb at %lluns, next at %lluns, rc = %d\n",
1654                d->opd_obd->obd_name, ktime_get_ns(),
1655                ktime_to_ns(d->opd_sync_next_commit_cb), rc);
1656
1657         if (likely(rc == 0)) {
1658                 lu_device_get(osp2lu_dev(d));
1659                 atomic_inc(&d->opd_commits_registered);
1660         } else
1661                 OBD_FREE_PTR(cb);
1662
1663         return rc;
1664 }
1665
1666 /* add the commit callback every second */
1667 int osp_sync_add_commit_cb_1s(const struct lu_env *env, struct osp_device *d,
1668                               struct thandle *th)
1669 {
1670         ktime_t now = ktime_get();
1671         bool add = false;
1672
1673         /* fast path */
1674         if (ktime_before(now, d->opd_sync_next_commit_cb))
1675                 return 0;
1676
1677         spin_lock(&d->opd_sync_lock);
1678         if (ktime_before(d->opd_sync_next_commit_cb, now)) {
1679                 add = true;
1680                 d->opd_sync_next_commit_cb = ktime_add_ns(now, NSEC_PER_SEC);
1681         }
1682         spin_unlock(&d->opd_sync_lock);
1683
1684         if (!add)
1685                 return 0;
1686
1687         return osp_sync_add_commit_cb(env, d, th);
1688 }
1689
1690 /*
1691  * generate an empty transaction and hook the commit callback in
1692  * then force transaction commit
1693  */
1694 void osp_sync_force(const struct lu_env *env, struct osp_device *d)
1695 {
1696         struct thandle *th;
1697         int rc;
1698
1699         th = dt_trans_create(env, d->opd_storage);
1700         if (IS_ERR(th)) {
1701                 CERROR("%s: can't sync\n", d->opd_obd->obd_name);
1702                 return;
1703         }
1704         rc = dt_trans_start_local(env, d->opd_storage, th);
1705         if (rc == 0) {
1706                 CDEBUG(D_OTHER, "%s: sync forced, %d changes\n",
1707                        d->opd_obd->obd_name,
1708                        atomic_read(&d->opd_sync_changes));
1709                 rc = osp_sync_add_commit_cb(env, d, th);
1710                 dt_trans_stop(env, d->opd_storage, th);
1711         }
1712
1713         dt_commit_async(env, d->opd_storage);
1714 }