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         out_destroy_update_req(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_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                 out_destroy_update_req(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 = out_create_update_req(&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,
 300                              int op, struct osp_object *obj, int count,
 301                              int *lens, const char **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         rc = out_insert_update(env, update, op, lu_object_fid(osp2lu_obj(obj)),
 320                                count, lens, bufs);
 321         /* The queue is full. */
 322         if (rc == -E2BIG) {
 323                 osp->opd_async_requests = NULL;
 324                 mutex_unlock(&osp->opd_async_requests_mutex);
 325
 326                 rc = osp_unplug_async_request(env, osp, update);
 327                 mutex_lock(&osp->opd_async_requests_mutex);
 328                 if (rc != 0)
 329                         GOTO(out, rc);
 330
 331                 update = osp_find_or_create_async_update_request(osp);
 332                 if (IS_ERR(update))
 333                         GOTO(out, rc = PTR_ERR(update));
 334
 335                 goto again;
 336         }
 337
 338         if (rc == 0)
 339                 list_add_tail(&oar->oar_list, &update->dur_cb_items);
 340
 341         GOTO(out, rc);
 342
 343 out:
 344         if (rc != 0)
 345                 osp_async_request_fini(env, oar);
 346
 347         return rc;
 348 }
 349
 350 /**
 351  * The OSP layer dt_device_operations::dt_trans_create() interface
 352  * to create a transaction.
 353  *
 354  * There are two kinds of transactions that will involve OSP:
 355  *
 356  * 1) If the transaction only contains the updates on remote server
 357  *    (MDT or OST), such as re-generating the lost OST-object for
 358  *    LFSCK, then it is a remote transaction. For remote transaction,
 359  *    the upper layer caller (such as the LFSCK engine) will call the
 360  *    dt_trans_create() (with the OSP dt_device as the parameter),
 361  *    then the call will be directed to the osp_trans_create() that
 362  *    creates the transaction handler and returns it to the caller.
 363  *
 364  * 2) If the transcation contains both local and remote updates,
 365  *    such as cross MDTs create under DNE mode, then the upper layer
 366  *    caller will not trigger osp_trans_create(). Instead, it will
 367  *    call dt_trans_create() on other dt_device, such as LOD that
 368  *    will generate the transaction handler. Such handler will be
 369  *    used by the whole transaction in subsequent sub-operations.
 370  *
 371  * \param[in] env       pointer to the thread context
 372  * \param[in] d         pointer to the OSP dt_device
 373  *
 374  * \retval              pointer to the transaction handler
 375  * \retval              negative error number on failure
 376  */
 377 struct thandle *osp_trans_create(const struct lu_env *env, struct dt_device *d)
 378 {
 379         struct thandle          *th = NULL;
 380         struct thandle_update   *tu = NULL;
 381         int                      rc = 0;
 382
 383         OBD_ALLOC_PTR(th);
 384         if (unlikely(th == NULL))
 385                 GOTO(out, rc = -ENOMEM);
 386
 387         th->th_dev = d;
 388         th->th_tags = LCT_TX_HANDLE;
 389         atomic_set(&th->th_refc, 1);
 390         th->th_alloc_size = sizeof(*th);
 391
 392         OBD_ALLOC_PTR(tu);
 393         if (tu == NULL)
 394                 GOTO(out, rc = -ENOMEM);
 395
 396         INIT_LIST_HEAD(&tu->tu_remote_update_list);
 397         tu->tu_only_remote_trans = 1;
 398         th->th_update = tu;
 399
 400 out:
 401         if (rc != 0) {
 402                 if (tu != NULL)
 403                         OBD_FREE_PTR(tu);
 404                 if (th != NULL)
 405                         OBD_FREE_PTR(th);
 406                 th = ERR_PTR(rc);
 407         }
 408
 409         return th;
 410 }
 411
 412 /**
 413  * Trigger the request for remote updates.
 414  *
 415  * If the transaction is a remote transaction, then related remote updates
 416  * will be sent asynchronously; otherwise, the cross MDTs transaction will
 417  * be synchronized.
 418  *
 419  * Please refer to osp_trans_create() for transaction type.
 420  *
 421  * \param[in] env               pointer to the thread context
 422  * \param[in] osp               pointer to the OSP device
 423  * \param[in] dt_update         pointer to the dt_update_request
 424  * \param[in] th                pointer to the transaction handler
 425  * \param[in] flow_control      whether need to control the flow
 426  *
 427  * \retval                      0 for success
 428  * \retval                      negative error number on failure
 429  */
 430 static int osp_trans_trigger(const struct lu_env *env, struct osp_device *osp,
 431                              struct dt_update_request *dt_update,
 432                              struct thandle *th, bool flow_control)
 433 {
 434         struct thandle_update   *tu = th->th_update;
 435         int                      rc = 0;
 436
 437         LASSERT(tu != NULL);
 438
 439         if (is_only_remote_trans(th)) {
 440                 struct osp_async_update_args    *args;
 441                 struct ptlrpc_request           *req;
 442
 443                 list_del_init(&dt_update->dur_list);
 444                 rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
 445                                          dt_update->dur_req, &req);
 446                 if (rc == 0) {
 447                         args = ptlrpc_req_async_args(req);
 448                         args->oaua_update = dt_update;
 449                         args->oaua_flow_control = flow_control;
 450                         req->rq_interpret_reply =
 451                                 osp_async_update_interpret;
 452                         ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
 453                 } else {
 454                         out_destroy_update_req(dt_update);
 455                 }
 456         } else {
 457                 th->th_sync = 1;
 458                 rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import,
 459                                      dt_update, NULL);
 460         }
 461
 462         return rc;
 463 }
 464
 465 /**
 466  * The OSP layer dt_device_operations::dt_trans_start() interface
 467  * to start the transaction.
 468  *
 469  * If the transaction is a remote transaction, then related remote
 470  * updates will be triggered in the osp_trans_stop(); otherwise the
 471  * transaction contains both local and remote update(s), then when
 472  * the OUT RPC will be triggered depends on the operation, and is
 473  * indicated by the dt_device::tu_sent_after_local_trans, for example:
 474  *
 475  * 1) If it is remote create, it will send the remote req after local
 476  * transaction. i.e. create the object locally first, then insert the
 477  * remote name entry.
 478  *
 479  * 2) If it is remote unlink, it will send the remote req before the
 480  * local transaction, i.e. delete the name entry remotely first, then
 481  * destroy the local object.
 482  *
 483  * Please refer to osp_trans_create() for transaction type.
 484  *
 485  * \param[in] env               pointer to the thread context
 486  * \param[in] dt                pointer to the OSP dt_device
 487  * \param[in] th                pointer to the transaction handler
 488  *
 489  * \retval                      0 for success
 490  * \retval                      negative error number on failure
 491  */
 492 int osp_trans_start(const struct lu_env *env, struct dt_device *dt,
 493                     struct thandle *th)
 494 {
 495         struct thandle_update           *tu = th->th_update;
 496         struct dt_update_request        *dt_update;
 497         int                              rc = 0;
 498
 499         if (tu == NULL)
 500                 return rc;
 501
 502         /* Check whether there are updates related with this OSP */
 503         dt_update = out_find_update(tu, dt);
 504         if (dt_update == NULL)
 505                 return rc;
 506
 507         if (!is_only_remote_trans(th) && !tu->tu_sent_after_local_trans)
 508                 rc = osp_trans_trigger(env, dt2osp_dev(dt), dt_update, th,
 509                                        false);
 510
 511         return rc;
 512 }
 513
 514 /**
 515  * The OSP layer dt_device_operations::dt_trans_stop() interface
 516  * to stop the transaction.
 517  *
 518  * If the transaction is a remote transaction, or the update handler
 519  * is marked as 'tu_sent_after_local_trans', then related remote
 520  * updates will be triggered here via osp_trans_trigger().
 521  *
 522  * For synchronous mode update or any failed update, the request
 523  * will be destroyed explicitly when the osp_trans_stop().
 524  *
 525  * Please refer to osp_trans_create() for transaction type.
 526  *
 527  * \param[in] env               pointer to the thread context
 528  * \param[in] dt                pointer to the OSP dt_device
 529  * \param[in] th                pointer to the transaction handler
 530  *
 531  * \retval                      0 for success
 532  * \retval                      negative error number on failure
 533  */
 534 int osp_trans_stop(const struct lu_env *env, struct dt_device *dt,
 535                    struct thandle *th)
 536 {
 537         struct thandle_update           *tu = th->th_update;
 538         struct dt_update_request        *dt_update;
 539         int                              rc = 0;
 540
 541         LASSERT(tu != NULL);
 542         LASSERT(tu != LP_POISON);
 543
 544         /* Check whether there are updates related with this OSP */
 545         dt_update = out_find_update(tu, dt);
 546         if (dt_update == NULL) {
 547                 if (!is_only_remote_trans(th))
 548                         return rc;
 549                 goto put;
 550         }
 551
 552         if (dt_update->dur_req->ourq_count == 0) {
 553                 out_destroy_update_req(dt_update);
 554                 goto put;
 555         }
 556
 557         if (is_only_remote_trans(th)) {
 558                 if (th->th_result == 0) {
 559                         struct osp_device *osp = dt2osp_dev(th->th_dev);
 560                         struct client_obd *cli = &osp->opd_obd->u.cli;
 561
 562                         rc = obd_get_request_slot(cli);
 563                         if (!osp->opd_imp_active || osp->opd_got_disconnected) {
 564                                 if (rc == 0)
 565                                         obd_put_request_slot(cli);
 566
 567                                 rc = -ENOTCONN;
 568                         }
 569
 570                         if (rc != 0) {
 571                                 out_destroy_update_req(dt_update);
 572                                 goto put;
 573                         }
 574
 575                         rc = osp_trans_trigger(env, dt2osp_dev(dt),
 576                                                dt_update, th, true);
 577                         if (rc != 0)
 578                                 obd_put_request_slot(cli);
 579                 } else {
 580                         rc = th->th_result;
 581                         out_destroy_update_req(dt_update);
 582                 }
 583         } else {
 584                 if (tu->tu_sent_after_local_trans)
 585                         rc = osp_trans_trigger(env, dt2osp_dev(dt),
 586                                                dt_update, th, false);
 587                 rc = dt_update->dur_rc;
 588                 out_destroy_update_req(dt_update);
 589         }
 590
 591 put:
 592         thandle_put(th);
 593         return rc;
 594 }