lustre/osp/osp_trans.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) 2014, Intel Corporation.
  24  */
  25 /*
  26  * lustre/osp/osp_trans.c
  27  *
  28  *
  29  * 1. OSP (Object Storage Proxy) transaction methods
  30  *
  31  * Implement OSP layer transaction related interfaces for the dt_device API
  32  * dt_device_operations.
  33  *
  34  *
  35  * 2. Handle asynchronous idempotent operations
  36  *
  37  * The OSP uses OUT (Object Unified Target) RPC to talk with other server
  38  * (MDT or OST) for kinds of operations, such as create, unlink, insert,
  39  * delete, lookup, set_(x)attr, get_(x)attr, and etc. To reduce the number
  40  * of RPCs, we allow multiple operations to be packaged together in single
  41  * OUT RPC.
  42  *
  43  * For the asynchronous idempotent operations, such as get_(x)attr, related
  44  * RPCs will be inserted into a osp_device based shared asynchronous request
  45  * queue - osp_device::opd_async_requests. When the queue is full, all the
  46  * requests in the queue will be packaged into a single OUT RPC and given to
  47  * the ptlrpcd daemon (for sending), then the queue is purged and other new
  48  * requests can be inserted into it.
  49  *
  50  * When the asynchronous idempotent operation inserts the request into the
  51  * shared queue, it will register an interpreter. When the packaged OUT RPC
  52  * is replied (or failed to be sent out), all the registered interpreters
  53  * will be called one by one to handle each own result.
  54  *
  55  *
  56  * Author: Di Wang <di.wang@intel.com>
  57  * Author: Fan, Yong <fan.yong@intel.com>
  58  */
  59
  60 #define DEBUG_SUBSYSTEM S_MDS
  61
  62 #include "osp_internal.h"
  63
  64 struct osp_async_update_args {
  65         struct dt_update_request *oaua_update;
  66         bool                      oaua_flow_control;
  67 };
  68
  69 struct osp_async_request {
  70         /* list in the dt_update_request::dur_cb_items */
  71         struct list_head                 oar_list;
  72
  73         /* The target of the async update request. */
  74         struct osp_object               *oar_obj;
  75
  76         /* The data used by oar_interpreter. */
  77         void                            *oar_data;
  78
  79         /* The interpreter function called after the async request handled. */
  80         osp_async_request_interpreter_t  oar_interpreter;
  81 };
  82
  83 /**
  84  * Allocate an asynchronous request and initialize it with the given parameters.
  85  *
  86  * \param[in] obj               pointer to the operation target
  87  * \param[in] data              pointer to the data used by the interpreter
  88  * \param[in] interpreter       pointer to the interpreter function
  89  *
  90  * \retval                      pointer to the asychronous request
  91  * \retval                      NULL if the allocation failed
  92  */
  93 static struct osp_async_request *
  94 osp_async_request_init(struct osp_object *obj, void *data,
  95                        osp_async_request_interpreter_t interpreter)
  96 {
  97         struct osp_async_request *oar;
  98
  99         OBD_ALLOC_PTR(oar);
 100         if (oar == NULL)
 101                 return NULL;
 102
 103         lu_object_get(osp2lu_obj(obj));
 104         INIT_LIST_HEAD(&oar->oar_list);
 105         oar->oar_obj = obj;
 106         oar->oar_data = data;
 107         oar->oar_interpreter = interpreter;
 108
 109         return oar;
 110 }
 111
 112 /**
 113  * Destroy the asychronous request.
 114  *
 115  * \param[in] env       pointer to the thread context
 116  * \param[in] oar       pointer to asychronous request
 117  */
 118 static void osp_async_request_fini(const struct lu_env *env,
 119                                    struct osp_async_request *oar)
 120 {
 121         LASSERT(list_empty(&oar->oar_list));
 122
 123         lu_object_put(env, osp2lu_obj(oar->oar_obj));
 124         OBD_FREE_PTR(oar);
 125 }
 126
 127 /**
 128  * Interpret the packaged OUT RPC results.
 129  *
 130  * For every packaged sub-request, call its registered interpreter function.
 131  * Then destroy the sub-request.
 132  *
 133  * \param[in] env       pointer to the thread context
 134  * \param[in] req       pointer to the RPC
 135  * \param[in] arg       pointer to data used by the interpreter
 136  * \param[in] rc        the RPC return value
 137  *
 138  * \retval              0 for success
 139  * \retval              negative error number on failure
 140  */
 141 static int osp_async_update_interpret(const struct lu_env *env,
 142                                       struct ptlrpc_request *req,
 143                                       void *arg, int rc)
 144 {
 145         struct object_update_reply      *reply  = NULL;
 146         struct osp_async_update_args    *oaua   = arg;
 147         struct dt_update_request        *dt_update = oaua->oaua_update;
 148         struct osp_async_request        *oar;
 149         struct osp_async_request        *next;
 150         int                              count  = 0;
 151         int                              index  = 0;
 152         int                              rc1    = 0;
 153
 154         if (oaua->oaua_flow_control)
 155                 obd_put_request_slot(
 156                                 &dt2osp_dev(dt_update->dur_dt)->opd_obd->u.cli);
 157
 158         /* Unpack the results from the reply message. */
 159         if (req->rq_repmsg != NULL) {
 160                 reply = req_capsule_server_sized_get(&req->rq_pill,
 161                                                      &RMF_OUT_UPDATE_REPLY,
 162                                                      OUT_UPDATE_REPLY_SIZE);
 163                 if (reply == NULL || reply->ourp_magic != UPDATE_REPLY_MAGIC)
 164                         rc1 = -EPROTO;
 165                 else
 166                         count = reply->ourp_count;
 167         } else {
 168                 rc1 = rc;
 169         }
 170
 171         list_for_each_entry_safe(oar, next, &dt_update->dur_cb_items,
 172                                  oar_list) {
 173                 list_del_init(&oar->oar_list);
 174
 175                 /* The peer may only have handled some requests (indicated
 176                  * by the 'count') in the packaged OUT RPC, we can only get
 177                  * results for the handled part. */
 178                 if (index < count && reply->ourp_lens[index] > 0) {
 179                         struct object_update_result *result;
 180
 181                         result = object_update_result_get(reply, index, NULL);
 182                         if (result == NULL)
 183                                 rc1 = -EPROTO;
 184                         else
 185                                 rc1 = result->our_rc;
 186                 } else {
 187                         rc1 = rc;
 188                         if (unlikely(rc1 == 0))
 189                                 rc1 = -EINVAL;
 190                 }
 191
 192                 oar->oar_interpreter(env, reply, req, oar->oar_obj,
 193                                        oar->oar_data, index, rc1);
 194                 osp_async_request_fini(env, oar);
 195                 index++;
 196         }
 197
 198         dt_update_request_destroy(dt_update);
 199
 200         return 0;
 201 }
 202
 203 /**
 204  * Pack all the requests in the shared asynchronous idempotent request queue
 205  * into a single OUT RPC that will be given to the background ptlrpcd daemon.
 206  *
 207  * \param[in] env       pointer to the thread context
 208  * \param[in] osp       pointer to the OSP device
 209  * \param[in] update    pointer to the shared queue
 210  *
 211  * \retval              0 for success
 212  * \retval              negative error number on failure
 213  */
 214 int osp_unplug_async_request(const struct lu_env *env,
 215                              struct osp_device *osp,
 216                              struct dt_update_request *update)
 217 {
 218         struct osp_async_update_args    *args;
 219         struct ptlrpc_request           *req = NULL;
 220         int                              rc;
 221
 222         rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
 223                                  update->dur_buf.ub_req, &req);
 224         if (rc != 0) {
 225                 struct osp_async_request *oar;
 226                 struct osp_async_request *next;
 227
 228                 list_for_each_entry_safe(oar, next,
 229                                          &update->dur_cb_items, oar_list) {
 230                         list_del_init(&oar->oar_list);
 231                         oar->oar_interpreter(env, NULL, NULL, oar->oar_obj,
 232                                                oar->oar_data, 0, rc);
 233                         osp_async_request_fini(env, oar);
 234                 }
 235                 dt_update_request_destroy(update);
 236         } else {
 237                 LASSERT(list_empty(&update->dur_list));
 238
 239                 args = ptlrpc_req_async_args(req);
 240                 args->oaua_update = update;
 241                 req->rq_interpret_reply = osp_async_update_interpret;
 242                 ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
 243         }
 244
 245         return rc;
 246 }
 247
 248 /**
 249  * Find or create (if NOT exist or purged) the shared asynchronous idempotent
 250  * request queue - osp_device::opd_async_requests.
 251  *
 252  * If the osp_device::opd_async_requests is not NULL, then return it directly;
 253  * otherwise create new dt_update_request and attach it to opd_async_requests.
 254  *
 255  * \param[in] osp       pointer to the OSP device
 256  *
 257  * \retval              pointer to the shared queue
 258  * \retval              negative error number on failure
 259  */
 260 static struct dt_update_request *
 261 osp_find_or_create_async_update_request(struct osp_device *osp)
 262 {
 263         struct dt_update_request *update = osp->opd_async_requests;
 264
 265         if (update != NULL)
 266                 return update;
 267
 268         update = dt_update_request_create(&osp->opd_dt_dev);
 269         if (!IS_ERR(update))
 270                 osp->opd_async_requests = update;
 271
 272         return update;
 273 }
 274
 275 /**
 276  * Insert an asynchronous idempotent request to the shared request queue that
 277  * is attached to the osp_device.
 278  *
 279  * This function generates a new osp_async_request with the given parameters,
 280  * then tries to insert the request into the osp_device-based shared request
 281  * queue. If the queue is full, then triggers the packaged OUT RPC to purge
 282  * the shared queue firstly, and then re-tries.
 283  *
 284  * NOTE: must hold the osp::opd_async_requests_mutex to serialize concurrent
 285  *       osp_insert_async_request call from others.
 286  *
 287  * \param[in] env               pointer to the thread context
 288  * \param[in] op                operation type, see 'enum update_type'
 289  * \param[in] obj               pointer to the operation target
 290  * \param[in] count             array size of the subsequent @lens and @bufs
 291  * \param[in] lens              buffer length array for the subsequent @bufs
 292  * \param[in] bufs              the buffers to compose the request
 293  * \param[in] data              pointer to the data used by the interpreter
 294  * \param[in] interpreter       pointer to the interpreter function
 295  *
 296  * \retval                      0 for success
 297  * \retval                      negative error number on failure
 298  */
 299 int osp_insert_async_request(const struct lu_env *env, enum update_type op,
 300                              struct osp_object *obj, int count,
 301                              __u16 *lens, const void **bufs, void *data,
 302                              osp_async_request_interpreter_t interpreter)
 303 {
 304         struct osp_async_request     *oar;
 305         struct osp_device            *osp = lu2osp_dev(osp2lu_obj(obj)->lo_dev);
 306         struct dt_update_request     *update;
 307         int                           rc  = 0;
 308         ENTRY;
 309
 310         oar = osp_async_request_init(obj, data, interpreter);
 311         if (oar == NULL)
 312                 RETURN(-ENOMEM);
 313
 314         update = osp_find_or_create_async_update_request(osp);
 315         if (IS_ERR(update))
 316                 GOTO(out, rc = PTR_ERR(update));
 317
 318 again:
 319         /* The queue is full. */
 320         rc = out_update_pack(env, &update->dur_buf, op,
 321                              lu_object_fid(osp2lu_obj(obj)), count, lens, bufs,
 322                              0);
 323         if (rc == -E2BIG) {
 324                 osp->opd_async_requests = NULL;
 325                 mutex_unlock(&osp->opd_async_requests_mutex);
 326
 327                 rc = osp_unplug_async_request(env, osp, update);
 328                 mutex_lock(&osp->opd_async_requests_mutex);
 329                 if (rc != 0)
 330                         GOTO(out, rc);
 331
 332                 update = osp_find_or_create_async_update_request(osp);
 333                 if (IS_ERR(update))
 334                         GOTO(out, rc = PTR_ERR(update));
 335
 336                 goto again;
 337         }
 338
 339         if (rc == 0)
 340                 list_add_tail(&oar->oar_list, &update->dur_cb_items);
 341
 342         GOTO(out, rc);
 343
 344 out:
 345         if (rc != 0)
 346                 osp_async_request_fini(env, oar);
 347
 348         return rc;
 349 }
 350
 351 /**
 352  * The OSP layer dt_device_operations::dt_trans_create() interface
 353  * to create a transaction.
 354  *
 355  * There are two kinds of transactions that will involve OSP:
 356  *
 357  * 1) If the transaction only contains the updates on remote server
 358  *    (MDT or OST), such as re-generating the lost OST-object for
 359  *    LFSCK, then it is a remote transaction. For remote transaction,
 360  *    the upper layer caller (such as the LFSCK engine) will call the
 361  *    dt_trans_create() (with the OSP dt_device as the parameter),
 362  *    then the call will be directed to the osp_trans_create() that
 363  *    creates the transaction handler and returns it to the caller.
 364  *
 365  * 2) If the transcation contains both local and remote updates,
 366  *    such as cross MDTs create under DNE mode, then the upper layer
 367  *    caller will not trigger osp_trans_create(). Instead, it will
 368  *    call dt_trans_create() on other dt_device, such as LOD that
 369  *    will generate the transaction handler. Such handler will be
 370  *    used by the whole transaction in subsequent sub-operations.
 371  *
 372  * \param[in] env       pointer to the thread context
 373  * \param[in] d         pointer to the OSP dt_device
 374  *
 375  * \retval              pointer to the transaction handler
 376  * \retval              negative error number on failure
 377  */
 378 struct thandle *osp_trans_create(const struct lu_env *env, struct dt_device *d)
 379 {
 380         struct thandle          *th = NULL;
 381         struct thandle_update   *tu = NULL;
 382         int                      rc = 0;
 383
 384         OBD_ALLOC_PTR(th);
 385         if (unlikely(th == NULL))
 386                 GOTO(out, rc = -ENOMEM);
 387
 388         th->th_dev = d;
 389         th->th_tags = LCT_TX_HANDLE;
 390         atomic_set(&th->th_refc, 1);
 391         th->th_alloc_size = sizeof(*th);
 392
 393         OBD_ALLOC_PTR(tu);
 394         if (tu == NULL)
 395                 GOTO(out, rc = -ENOMEM);
 396
 397         INIT_LIST_HEAD(&tu->tu_remote_update_list);
 398         tu->tu_only_remote_trans = 1;
 399         th->th_update = tu;
 400
 401 out:
 402         if (rc != 0) {
 403                 if (tu != NULL)
 404                         OBD_FREE_PTR(tu);
 405                 if (th != NULL)
 406                         OBD_FREE_PTR(th);
 407                 th = ERR_PTR(rc);
 408         }
 409
 410         return th;
 411 }
 412
 413 /**
 414  * Trigger the request for remote updates.
 415  *
 416  * If the transaction is a remote transaction, then related remote updates
 417  * will be sent asynchronously; otherwise, the cross MDTs transaction will
 418  * be synchronized.
 419  *
 420  * Please refer to osp_trans_create() for transaction type.
 421  *
 422  * \param[in] env               pointer to the thread context
 423  * \param[in] osp               pointer to the OSP device
 424  * \param[in] dt_update         pointer to the dt_update_request
 425  * \param[in] th                pointer to the transaction handler
 426  * \param[in] flow_control      whether need to control the flow
 427  *
 428  * \retval                      0 for success
 429  * \retval                      negative error number on failure
 430  */
 431 static int osp_trans_trigger(const struct lu_env *env, struct osp_device *osp,
 432                              struct dt_update_request *dt_update,
 433                              struct thandle *th, bool flow_control)
 434 {
 435         struct thandle_update   *tu = th->th_update;
 436         int                      rc = 0;
 437
 438         LASSERT(tu != NULL);
 439
 440         if (is_only_remote_trans(th)) {
 441                 struct osp_async_update_args    *args;
 442                 struct ptlrpc_request           *req;
 443
 444                 list_del_init(&dt_update->dur_list);
 445                 rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
 446                                          dt_update->dur_buf.ub_req, &req);
 447                 if (rc == 0) {
 448                         args = ptlrpc_req_async_args(req);
 449                         args->oaua_update = dt_update;
 450                         args->oaua_flow_control = flow_control;
 451                         req->rq_interpret_reply =
 452                                 osp_async_update_interpret;
 453                         ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
 454                 } else {
 455                         dt_update_request_destroy(dt_update);
 456                 }
 457         } else {
 458                 th->th_sync = 1;
 459                 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import,
 460                                      dt_update, NULL);
 461         }
 462
 463         return rc;
 464 }
 465
 466 /**
 467  * The OSP layer dt_device_operations::dt_trans_start() interface
 468  * to start the transaction.
 469  *
 470  * If the transaction is a remote transaction, then related remote
 471  * updates will be triggered in the osp_trans_stop(); otherwise the
 472  * transaction contains both local and remote update(s), then when
 473  * the OUT RPC will be triggered depends on the operation, and is
 474  * indicated by the dt_device::tu_sent_after_local_trans, for example:
 475  *
 476  * 1) If it is remote create, it will send the remote req after local
 477  * transaction. i.e. create the object locally first, then insert the
 478  * remote name entry.
 479  *
 480  * 2) If it is remote unlink, it will send the remote req before the
 481  * local transaction, i.e. delete the name entry remotely first, then
 482  * destroy the local object.
 483  *
 484  * Please refer to osp_trans_create() for transaction type.
 485  *
 486  * \param[in] env               pointer to the thread context
 487  * \param[in] dt                pointer to the OSP dt_device
 488  * \param[in] th                pointer to the transaction handler
 489  *
 490  * \retval                      0 for success
 491  * \retval                      negative error number on failure
 492  */
 493 int osp_trans_start(const struct lu_env *env, struct dt_device *dt,
 494                     struct thandle *th)
 495 {
 496         struct thandle_update           *tu = th->th_update;
 497         struct dt_update_request        *dt_update;
 498         int                              rc = 0;
 499
 500         if (tu == NULL)
 501                 return rc;
 502
 503         /* Check whether there are updates related with this OSP */
 504         dt_update = out_find_update(tu, dt);
 505         if (dt_update == NULL)
 506                 return rc;
 507
 508         if (!is_only_remote_trans(th) && !tu->tu_sent_after_local_trans)
 509                 rc = osp_trans_trigger(env, dt2osp_dev(dt), dt_update, th,
 510                                        false);
 511
 512         return rc;
 513 }
 514
 515 /**
 516  * The OSP layer dt_device_operations::dt_trans_stop() interface
 517  * to stop the transaction.
 518  *
 519  * If the transaction is a remote transaction, or the update handler
 520  * is marked as 'tu_sent_after_local_trans', then related remote
 521  * updates will be triggered here via osp_trans_trigger().
 522  *
 523  * For synchronous mode update or any failed update, the request
 524  * will be destroyed explicitly when the osp_trans_stop().
 525  *
 526  * Please refer to osp_trans_create() for transaction type.
 527  *
 528  * \param[in] env               pointer to the thread context
 529  * \param[in] dt                pointer to the OSP dt_device
 530  * \param[in] th                pointer to the transaction handler
 531  *
 532  * \retval                      0 for success
 533  * \retval                      negative error number on failure
 534  */
 535 int osp_trans_stop(const struct lu_env *env, struct dt_device *dt,
 536                    struct thandle *th)
 537 {
 538         struct thandle_update           *tu = th->th_update;
 539         struct dt_update_request        *dt_update;
 540         int                              rc = 0;
 541
 542         LASSERT(tu != NULL);
 543         LASSERT(tu != LP_POISON);
 544
 545         /* Check whether there are updates related with this OSP */
 546         dt_update = out_find_update(tu, dt);
 547         if (dt_update == NULL) {
 548                 if (!is_only_remote_trans(th))
 549                         return rc;
 550                 goto put;
 551         }
 552
 553         if (dt_update->dur_buf.ub_req == NULL ||
 554             dt_update->dur_buf.ub_req->ourq_count == 0) {
 555                 dt_update_request_destroy(dt_update);
 556                 goto put;
 557         }
 558
 559         if (is_only_remote_trans(th)) {
 560                 if (th->th_result == 0) {
 561                         struct osp_device *osp = dt2osp_dev(th->th_dev);
 562                         struct client_obd *cli = &osp->opd_obd->u.cli;
 563
 564                         rc = obd_get_request_slot(cli);
 565                         if (!osp->opd_imp_active || osp->opd_got_disconnected) {
 566                                 if (rc == 0)
 567                                         obd_put_request_slot(cli);
 568
 569                                 rc = -ENOTCONN;
 570                         }
 571
 572                         if (rc != 0) {
 573                                 dt_update_request_destroy(dt_update);
 574                                 goto put;
 575                         }
 576
 577                         rc = osp_trans_trigger(env, dt2osp_dev(dt),
 578                                                dt_update, th, true);
 579                         if (rc != 0)
 580                                 obd_put_request_slot(cli);
 581                 } else {
 582                         rc = th->th_result;
 583                         dt_update_request_destroy(dt_update);
 584                 }
 585         } else {
 586                 if (tu->tu_sent_after_local_trans)
 587                         rc = osp_trans_trigger(env, dt2osp_dev(dt),
 588                                                dt_update, th, false);
 589                 rc = dt_update->dur_rc;
 590                 dt_update_request_destroy(dt_update);
 591         }
 592
 593 put:
 594         thandle_put(th);
 595         return rc;
 596 }