lustre/mdt/mdt_coordinator.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,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License version 2 for more details.  A copy is
  14  * included in the COPYING file that accompanied this code.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2011, 2012 Commissariat a l'energie atomique et aux energies
  24  *                          alternatives
  25  *
  26  * Copyright (c) 2013, 2014, Intel Corporation.
  27  * Use is subject to license terms.
  28  */
  29 /*
  30  * lustre/mdt/mdt_coordinator.c
  31  *
  32  * Lustre HSM Coordinator
  33  *
  34  * Author: Jacques-Charles Lafoucriere <jacques-charles.lafoucriere@cea.fr>
  35  * Author: Aurelien Degremont <aurelien.degremont@cea.fr>
  36  * Author: Thomas Leibovici <thomas.leibovici@cea.fr>
  37  */
  38
  39 #define DEBUG_SUBSYSTEM S_MDS
  40
  41 #include <linux/kthread.h>
  42 #include <obd_support.h>
  43 #include <lustre_net.h>
  44 #include <lustre_export.h>
  45 #include <obd.h>
  46 #include <lprocfs_status.h>
  47 #include <lustre_log.h>
  48 #include <lustre_kernelcomm.h>
  49 #include "mdt_internal.h"
  50
  51 static struct lprocfs_vars lprocfs_mdt_hsm_vars[];
  52
  53 /**
  54  * get obj and HSM attributes on a fid
  55  * \param mti [IN] context
  56  * \param fid [IN] object fid
  57  * \param hsm [OUT] HSM meta data
  58  * \retval obj or error (-ENOENT if not found)
  59  */
  60 struct mdt_object *mdt_hsm_get_md_hsm(struct mdt_thread_info *mti,
  61                                       const struct lu_fid *fid,
  62                                       struct md_hsm *hsm)
  63 {
  64         struct md_attr          *ma;
  65         struct mdt_object       *obj;
  66         int                      rc;
  67         ENTRY;
  68
  69         ma = &mti->mti_attr;
  70         ma->ma_need = MA_HSM;
  71         ma->ma_valid = 0;
  72
  73         /* find object by FID */
  74         obj = mdt_object_find(mti->mti_env, mti->mti_mdt, fid);
  75         if (IS_ERR(obj))
  76                 RETURN(obj);
  77
  78         if (!mdt_object_exists(obj)) {
  79                 /* no more object */
  80                 mdt_object_put(mti->mti_env, obj);
  81                 RETURN(ERR_PTR(-ENOENT));
  82         }
  83
  84         rc = mdt_attr_get_complex(mti, obj, ma);
  85         if (rc) {
  86                 mdt_object_put(mti->mti_env, obj);
  87                 RETURN(ERR_PTR(rc));
  88         }
  89
  90         if (ma->ma_valid & MA_HSM)
  91                 *hsm = ma->ma_hsm;
  92         else
  93                 memset(hsm, 0, sizeof(*hsm));
  94         ma->ma_valid = 0;
  95         RETURN(obj);
  96 }
  97
  98 void mdt_hsm_dump_hal(int level, const char *prefix,
  99                       struct hsm_action_list *hal)
 100 {
 101         int                      i, sz;
 102         struct hsm_action_item  *hai;
 103         char                     buf[12];
 104
 105         CDEBUG(level, "%s: HAL header: version %X count %d compound "LPX64
 106                       " archive_id %d flags "LPX64"\n",
 107                prefix, hal->hal_version, hal->hal_count,
 108                hal->hal_compound_id, hal->hal_archive_id, hal->hal_flags);
 109
 110         hai = hai_first(hal);
 111         for (i = 0; i < hal->hal_count; i++) {
 112                 sz = hai->hai_len - sizeof(*hai);
 113                 CDEBUG(level, "%s %d: fid="DFID" dfid="DFID
 114                        " compound/cookie="LPX64"/"LPX64
 115                        " action=%s extent="LPX64"-"LPX64" gid="LPX64
 116                        " datalen=%d data=[%s]\n",
 117                        prefix, i,
 118                        PFID(&hai->hai_fid), PFID(&hai->hai_dfid),
 119                        hal->hal_compound_id, hai->hai_cookie,
 120                        hsm_copytool_action2name(hai->hai_action),
 121                        hai->hai_extent.offset,
 122                        hai->hai_extent.length,
 123                        hai->hai_gid, sz,
 124                        hai_dump_data_field(hai, buf, sizeof(buf)));
 125                 hai = hai_next(hai);
 126         }
 127 }
 128
 129 /**
 130  * data passed to llog_cat_process() callback
 131  * to scan requests and take actions
 132  */
 133 struct hsm_scan_request {
 134         int                      hal_sz;
 135         int                      hal_used_sz;
 136         struct hsm_action_list  *hal;
 137 };
 138
 139 struct hsm_scan_data {
 140         struct mdt_thread_info          *mti;
 141         char                             fs_name[MTI_NAME_MAXLEN+1];
 142         /* request to be send to agents */
 143         int                              max_requests;  /** vector size */
 144         int                              request_cnt;   /** used count */
 145         struct hsm_scan_request         *request;
 146 };
 147
 148 /**
 149  *  llog_cat_process() callback, used to:
 150  *  - find waiting request and start action
 151  *  - purge canceled and done requests
 152  * \param env [IN] environment
 153  * \param llh [IN] llog handle
 154  * \param hdr [IN] llog record
 155  * \param data [IN/OUT] cb data = struct hsm_scan_data
 156  * \retval 0 success
 157  * \retval -ve failure
 158  */
 159 static int mdt_coordinator_cb(const struct lu_env *env,
 160                               struct llog_handle *llh,
 161                               struct llog_rec_hdr *hdr,
 162                               void *data)
 163 {
 164         struct llog_agent_req_rec       *larr;
 165         struct hsm_scan_data            *hsd;
 166         struct hsm_action_item          *hai;
 167         struct mdt_device               *mdt;
 168         struct coordinator              *cdt;
 169         int                              rc;
 170         ENTRY;
 171
 172         hsd = data;
 173         mdt = hsd->mti->mti_mdt;
 174         cdt = &mdt->mdt_coordinator;
 175
 176         larr = (struct llog_agent_req_rec *)hdr;
 177         dump_llog_agent_req_rec("mdt_coordinator_cb(): ", larr);
 178         switch (larr->arr_status) {
 179         case ARS_WAITING: {
 180                 int i;
 181                 struct hsm_scan_request *request;
 182
 183                 /* Are agents full? */
 184                 if (atomic_read(&cdt->cdt_request_count) >=
 185                     cdt->cdt_max_requests)
 186                         break;
 187
 188                 /* first search whether the request is found in the
 189                  * list we have built. */
 190                 request = NULL;
 191                 for (i = 0; i < hsd->request_cnt; i++) {
 192                         if (hsd->request[i].hal->hal_compound_id ==
 193                             larr->arr_compound_id) {
 194                                 request = &hsd->request[i];
 195                                 break;
 196                         }
 197                 }
 198
 199                 if (!request) {
 200                         struct hsm_action_list *hal;
 201
 202                         if (hsd->request_cnt == hsd->max_requests)
 203                                 /* Unknown request and no more room
 204                                  * for a new request. Continue to scan
 205                                  * to find other entries for already
 206                                  * existing requests.
 207                                  */
 208                                 RETURN(0);
 209
 210                         request = &hsd->request[hsd->request_cnt];
 211
 212                         /* allocates hai vector size just needs to be large
 213                          * enough */
 214                         request->hal_sz =
 215                                 sizeof(*request->hal) +
 216                                 cfs_size_round(MTI_NAME_MAXLEN+1) +
 217                                 2 * cfs_size_round(larr->arr_hai.hai_len);
 218                         OBD_ALLOC(hal, request->hal_sz);
 219                         if (!hal) {
 220                                 CERROR("%s: Cannot allocate memory (%d o)"
 221                                        "for compound "LPX64"\n",
 222                                        mdt_obd_name(mdt),
 223                                        request->hal_sz,
 224                                        larr->arr_compound_id);
 225                                 RETURN(-ENOMEM);
 226                         }
 227                         hal->hal_version = HAL_VERSION;
 228                         strlcpy(hal->hal_fsname, hsd->fs_name,
 229                                 MTI_NAME_MAXLEN + 1);
 230                         hal->hal_compound_id = larr->arr_compound_id;
 231                         hal->hal_archive_id = larr->arr_archive_id;
 232                         hal->hal_flags = larr->arr_flags;
 233                         hal->hal_count = 0;
 234                         request->hal_used_sz = hal_size(hal);
 235                         request->hal = hal;
 236                         hsd->request_cnt++;
 237                         hai = hai_first(hal);
 238                 } else {
 239                         /* request is known */
 240                         /* we check if record archive num is the same as the
 241                          * known request, if not we will serve it in multiple
 242                          * time because we do not know if the agent can serve
 243                          * multiple backend
 244                          * a use case is a compound made of multiple restore
 245                          * where the files are not archived in the same backend
 246                          */
 247                         if (larr->arr_archive_id !=
 248                             request->hal->hal_archive_id)
 249                                 RETURN(0);
 250
 251                         if (request->hal_sz <
 252                             request->hal_used_sz +
 253                             cfs_size_round(larr->arr_hai.hai_len)) {
 254                                 /* Not enough room, need an extension */
 255                                 void *hal_buffer;
 256                                 int sz;
 257
 258                                 sz = 2 * request->hal_sz;
 259                                 OBD_ALLOC(hal_buffer, sz);
 260                                 if (!hal_buffer) {
 261                                         CERROR("%s: Cannot allocate memory "
 262                                                "(%d o) for compound "LPX64"\n",
 263                                                mdt_obd_name(mdt), sz,
 264                                                larr->arr_compound_id);
 265                                         RETURN(-ENOMEM);
 266                                 }
 267                                 memcpy(hal_buffer, request->hal,
 268                                        request->hal_used_sz);
 269                                 OBD_FREE(request->hal,
 270                                          request->hal_sz);
 271                                 request->hal = hal_buffer;
 272                                 request->hal_sz = sz;
 273                         }
 274                         hai = hai_first(request->hal);
 275                         for (i = 0; i < request->hal->hal_count; i++)
 276                                 hai = hai_next(hai);
 277                 }
 278                 memcpy(hai, &larr->arr_hai, larr->arr_hai.hai_len);
 279                 hai->hai_cookie = larr->arr_hai.hai_cookie;
 280                 hai->hai_gid = larr->arr_hai.hai_gid;
 281
 282                 request->hal_used_sz += cfs_size_round(hai->hai_len);
 283                 request->hal->hal_count++;
 284                 break;
 285         }
 286         case ARS_STARTED: {
 287                 struct hsm_progress_kernel pgs;
 288                 struct cdt_agent_req *car;
 289                 cfs_time_t now = cfs_time_current_sec();
 290                 cfs_time_t last;
 291
 292                 /* we search for a running request
 293                  * error may happen if coordinator crashes or stopped
 294                  * with running request
 295                  */
 296                 car = mdt_cdt_find_request(cdt, larr->arr_hai.hai_cookie, NULL);
 297                 if (car == NULL) {
 298                         last = larr->arr_req_create;
 299                 } else {
 300                         last = car->car_req_update;
 301                         mdt_cdt_put_request(car);
 302                 }
 303
 304                 /* test if request too long, if yes cancel it
 305                  * the same way the copy tool acknowledge a cancel request */
 306                 if (now <= last + cdt->cdt_active_req_timeout)
 307                         RETURN(0);
 308
 309                 dump_llog_agent_req_rec("request timed out, start cleaning",
 310                                         larr);
 311                 /* a too old cancel request just needs to be removed
 312                  * this can happen, if copy tool does not support
 313                  * cancel for other requests, we have to remove the
 314                  * running request and notify the copytool */
 315                 pgs.hpk_fid = larr->arr_hai.hai_fid;
 316                 pgs.hpk_cookie = larr->arr_hai.hai_cookie;
 317                 pgs.hpk_extent = larr->arr_hai.hai_extent;
 318                 pgs.hpk_flags = HP_FLAG_COMPLETED;
 319                 pgs.hpk_errval = ENOSYS;
 320                 pgs.hpk_data_version = 0;
 321
 322                 /* update request state, but do not record in llog, to
 323                  * avoid deadlock on cdt_llog_lock */
 324                 rc = mdt_hsm_update_request_state(hsd->mti, &pgs, 0);
 325                 if (rc)
 326                         CERROR("%s: cannot cleanup timed out request: "
 327                                DFID" for cookie "LPX64" action=%s\n",
 328                                mdt_obd_name(mdt),
 329                                PFID(&pgs.hpk_fid), pgs.hpk_cookie,
 330                                hsm_copytool_action2name(
 331                                        larr->arr_hai.hai_action));
 332
 333                 if (rc == -ENOENT) {
 334                         /* The request no longer exists, forget
 335                          * about it, and do not send a cancel request
 336                          * to the client, for which an error will be
 337                          * sent back, leading to an endless cycle of
 338                          * cancellation. */
 339                         RETURN(LLOG_DEL_RECORD);
 340                 }
 341
 342                 /* XXX A cancel request cannot be cancelled. */
 343                 if (larr->arr_hai.hai_action == HSMA_CANCEL)
 344                         RETURN(0);
 345
 346                 larr->arr_status = ARS_CANCELED;
 347                 larr->arr_req_change = now;
 348                 rc = llog_write(hsd->mti->mti_env, llh, hdr, hdr->lrh_index);
 349                 if (rc < 0)
 350                         CERROR("%s: cannot update agent log: rc = %d\n",
 351                                mdt_obd_name(mdt), rc);
 352                 break;
 353         }
 354         case ARS_FAILED:
 355         case ARS_CANCELED:
 356         case ARS_SUCCEED:
 357                 if ((larr->arr_req_change + cdt->cdt_grace_delay) <
 358                     cfs_time_current_sec())
 359                         RETURN(LLOG_DEL_RECORD);
 360                 break;
 361         }
 362         RETURN(0);
 363 }
 364
 365 /**
 366  * create /proc entries for coordinator
 367  * \param mdt [IN]
 368  * \retval 0 success
 369  * \retval -ve failure
 370  */
 371 int hsm_cdt_procfs_init(struct mdt_device *mdt)
 372 {
 373         struct coordinator      *cdt = &mdt->mdt_coordinator;
 374         int                      rc = 0;
 375         ENTRY;
 376
 377         /* init /proc entries, failure is not critical */
 378         cdt->cdt_proc_dir = lprocfs_register("hsm",
 379                                              mdt2obd_dev(mdt)->obd_proc_entry,
 380                                              lprocfs_mdt_hsm_vars, mdt);
 381         if (IS_ERR(cdt->cdt_proc_dir)) {
 382                 rc = PTR_ERR(cdt->cdt_proc_dir);
 383                 CERROR("%s: Cannot create 'hsm' directory in mdt proc dir,"
 384                        " rc=%d\n", mdt_obd_name(mdt), rc);
 385                 cdt->cdt_proc_dir = NULL;
 386                 RETURN(rc);
 387         }
 388
 389         RETURN(0);
 390 }
 391
 392 /**
 393  * remove /proc entries for coordinator
 394  * \param mdt [IN]
 395  */
 396 void  hsm_cdt_procfs_fini(struct mdt_device *mdt)
 397 {
 398         struct coordinator      *cdt = &mdt->mdt_coordinator;
 399
 400         LASSERT(cdt->cdt_state == CDT_STOPPED);
 401         if (cdt->cdt_proc_dir != NULL)
 402                 lprocfs_remove(&cdt->cdt_proc_dir);
 403 }
 404
 405 /**
 406  * get vector of hsm cdt /proc vars
 407  * \param none
 408  * \retval var vector
 409  */
 410 struct lprocfs_vars *hsm_cdt_get_proc_vars(void)
 411 {
 412         return lprocfs_mdt_hsm_vars;
 413 }
 414
 415 /**
 416  * coordinator thread
 417  * \param data [IN] obd device
 418  * \retval 0 success
 419  * \retval -ve failure
 420  */
 421 static int mdt_coordinator(void *data)
 422 {
 423         struct mdt_thread_info  *mti = data;
 424         struct mdt_device       *mdt = mti->mti_mdt;
 425         struct coordinator      *cdt = &mdt->mdt_coordinator;
 426         struct hsm_scan_data     hsd = { NULL };
 427         int                      rc = 0;
 428         int                      request_sz;
 429         ENTRY;
 430
 431         cdt->cdt_thread.t_flags = SVC_RUNNING;
 432         wake_up(&cdt->cdt_thread.t_ctl_waitq);
 433
 434         CDEBUG(D_HSM, "%s: coordinator thread starting, pid=%d\n",
 435                mdt_obd_name(mdt), current_pid());
 436
 437         /* we use a copy of cdt_max_requests in the cb, so if cdt_max_requests
 438          * increases due to a change from /proc we do not overflow the
 439          * hsd.request[] vector
 440          */
 441         hsd.max_requests = cdt->cdt_max_requests;
 442         request_sz = hsd.max_requests * sizeof(*hsd.request);
 443         OBD_ALLOC(hsd.request, request_sz);
 444         if (!hsd.request)
 445                 GOTO(out, rc = -ENOMEM);
 446
 447         hsd.mti = mti;
 448         obd_uuid2fsname(hsd.fs_name, mdt_obd_name(mdt), MTI_NAME_MAXLEN);
 449
 450         while (1) {
 451                 struct l_wait_info lwi;
 452                 int i;
 453
 454                 lwi = LWI_TIMEOUT(cfs_time_seconds(cdt->cdt_loop_period),
 455                                   NULL, NULL);
 456                 l_wait_event(cdt->cdt_thread.t_ctl_waitq,
 457                              (cdt->cdt_thread.t_flags &
 458                               (SVC_STOPPING|SVC_EVENT)),
 459                              &lwi);
 460
 461                 CDEBUG(D_HSM, "coordinator resumes\n");
 462
 463                 if (cdt->cdt_thread.t_flags & SVC_STOPPING ||
 464                     cdt->cdt_state == CDT_STOPPING) {
 465                         cdt->cdt_thread.t_flags &= ~SVC_STOPPING;
 466                         rc = 0;
 467                         break;
 468                 }
 469
 470                 /* wake up before timeout, new work arrives */
 471                 if (cdt->cdt_thread.t_flags & SVC_EVENT)
 472                         cdt->cdt_thread.t_flags &= ~SVC_EVENT;
 473
 474                 /* if coordinator is suspended continue to wait */
 475                 if (cdt->cdt_state == CDT_DISABLE) {
 476                         CDEBUG(D_HSM, "disable state, coordinator sleeps\n");
 477                         continue;
 478                 }
 479
 480                 CDEBUG(D_HSM, "coordinator starts reading llog\n");
 481
 482                 if (hsd.max_requests != cdt->cdt_max_requests) {
 483                         /* cdt_max_requests has changed,
 484                          * we need to allocate a new buffer
 485                          */
 486                         OBD_FREE(hsd.request, request_sz);
 487                         hsd.max_requests = cdt->cdt_max_requests;
 488                         request_sz = hsd.max_requests * sizeof(*hsd.request);
 489                         OBD_ALLOC(hsd.request, request_sz);
 490                         if (!hsd.request) {
 491                                 rc = -ENOMEM;
 492                                 break;
 493                         }
 494                 }
 495
 496                 hsd.request_cnt = 0;
 497
 498                 rc = cdt_llog_process(mti->mti_env, mdt,
 499                                       mdt_coordinator_cb, &hsd);
 500                 if (rc < 0)
 501                         goto clean_cb_alloc;
 502
 503                 CDEBUG(D_HSM, "found %d requests to send\n", hsd.request_cnt);
 504
 505                 if (list_empty(&cdt->cdt_agents)) {
 506                         CDEBUG(D_HSM, "no agent available, "
 507                                       "coordinator sleeps\n");
 508                         goto clean_cb_alloc;
 509                 }
 510
 511                 /* here hsd contains a list of requests to be started */
 512                 for (i = 0; i < hsd.request_cnt; i++) {
 513                         struct hsm_scan_request *request = &hsd.request[i];
 514                         struct hsm_action_list  *hal = request->hal;
 515                         struct hsm_action_item  *hai;
 516                         __u64                   *cookies;
 517                         int                      sz, j;
 518                         enum agent_req_status    status;
 519
 520                         /* still room for work ? */
 521                         if (atomic_read(&cdt->cdt_request_count) >=
 522                             cdt->cdt_max_requests)
 523                                 break;
 524
 525                         rc = mdt_hsm_agent_send(mti, hal, 0);
 526                         /* if failure, we suppose it is temporary
 527                          * if the copy tool failed to do the request
 528                          * it has to use hsm_progress
 529                          */
 530                         status = (rc ? ARS_WAITING : ARS_STARTED);
 531
 532                         /* set up cookie vector to set records status
 533                          * after copy tools start or failed
 534                          */
 535                         sz = hal->hal_count * sizeof(__u64);
 536                         OBD_ALLOC(cookies, sz);
 537                         if (cookies == NULL) {
 538                                 CERROR("%s: Cannot allocate memory (%d o) "
 539                                        "for cookies vector "LPX64"\n",
 540                                        mdt_obd_name(mdt), sz,
 541                                        hal->hal_compound_id);
 542                                 continue;
 543                         }
 544                         hai = hai_first(hal);
 545                         for (j = 0; j < hal->hal_count; j++) {
 546                                 cookies[j] = hai->hai_cookie;
 547                                 hai = hai_next(hai);
 548                         }
 549
 550                         rc = mdt_agent_record_update(mti->mti_env, mdt, cookies,
 551                                                      hal->hal_count, status);
 552                         if (rc)
 553                                 CERROR("%s: mdt_agent_record_update() failed, "
 554                                        "rc=%d, cannot update status to %s "
 555                                        "for %d cookies\n",
 556                                        mdt_obd_name(mdt), rc,
 557                                        agent_req_status2name(status),
 558                                        hal->hal_count);
 559
 560                         OBD_FREE(cookies, sz);
 561                 }
 562 clean_cb_alloc:
 563                 /* free hal allocated by callback */
 564                 for (i = 0; i < hsd.request_cnt; i++) {
 565                         struct hsm_scan_request *request = &hsd.request[i];
 566
 567                         OBD_FREE(request->hal, request->hal_sz);
 568                 }
 569         }
 570         EXIT;
 571 out:
 572         if (hsd.request)
 573                 OBD_FREE(hsd.request, request_sz);
 574
 575         if (cdt->cdt_state == CDT_STOPPING) {
 576                 /* request comes from /proc path, so we need to clean cdt
 577                  * struct */
 578                  mdt_hsm_cdt_stop(mdt);
 579                  mdt->mdt_opts.mo_coordinator = 0;
 580         } else {
 581                 /* request comes from a thread event, generated
 582                  * by mdt_stop_coordinator(), we have to ack
 583                  * and cdt cleaning will be done by event sender
 584                  */
 585                 cdt->cdt_thread.t_flags = SVC_STOPPED;
 586                 wake_up(&cdt->cdt_thread.t_ctl_waitq);
 587         }
 588
 589         if (rc != 0)
 590                 CERROR("%s: coordinator thread exiting, process=%d, rc=%d\n",
 591                        mdt_obd_name(mdt), current_pid(), rc);
 592         else
 593                 CDEBUG(D_HSM, "%s: coordinator thread exiting, process=%d,"
 594                               " no error\n",
 595                        mdt_obd_name(mdt), current_pid());
 596
 597         return rc;
 598 }
 599
 600 /**
 601  * lookup a restore handle by FID
 602  * caller needs to hold cdt_restore_lock
 603  * \param cdt [IN] coordinator
 604  * \param fid [IN] FID
 605  * \retval cdt_restore_handle found
 606  * \retval NULL not found
 607  */
 608 struct cdt_restore_handle *mdt_hsm_restore_hdl_find(struct coordinator *cdt,
 609                                                        const struct lu_fid *fid)
 610 {
 611         struct cdt_restore_handle       *crh;
 612         ENTRY;
 613
 614         list_for_each_entry(crh, &cdt->cdt_restore_hdl, crh_list) {
 615                 if (lu_fid_eq(&crh->crh_fid, fid))
 616                         RETURN(crh);
 617         }
 618         RETURN(NULL);
 619 }
 620
 621 /**
 622  * data passed to llog_cat_process() callback
 623  * to scan requests and take actions
 624  */
 625 struct hsm_restore_data {
 626         struct mdt_thread_info  *hrd_mti;
 627 };
 628
 629 /**
 630  *  llog_cat_process() callback, used to:
 631  *  - find restore request and allocate the restore handle
 632  * \param env [IN] environment
 633  * \param llh [IN] llog handle
 634  * \param hdr [IN] llog record
 635  * \param data [IN/OUT] cb data = struct hsm_restore_data
 636  * \retval 0 success
 637  * \retval -ve failure
 638  */
 639 static int hsm_restore_cb(const struct lu_env *env,
 640                           struct llog_handle *llh,
 641                           struct llog_rec_hdr *hdr, void *data)
 642 {
 643         struct llog_agent_req_rec       *larr;
 644         struct hsm_restore_data         *hrd;
 645         struct cdt_restore_handle       *crh;
 646         struct hsm_action_item          *hai;
 647         struct mdt_thread_info          *mti;
 648         struct coordinator              *cdt;
 649         struct mdt_object               *child;
 650         int rc;
 651         ENTRY;
 652
 653         hrd = data;
 654         mti = hrd->hrd_mti;
 655         cdt = &mti->mti_mdt->mdt_coordinator;
 656
 657         larr = (struct llog_agent_req_rec *)hdr;
 658         hai = &larr->arr_hai;
 659         if (hai->hai_cookie > cdt->cdt_last_cookie)
 660                 /* update the cookie to avoid collision */
 661                 cdt->cdt_last_cookie = hai->hai_cookie + 1;
 662
 663         if (hai->hai_action != HSMA_RESTORE ||
 664             agent_req_in_final_state(larr->arr_status))
 665                 RETURN(0);
 666
 667         /* restore request not in a final state */
 668
 669         /* force replay of restore requests left in started state from previous
 670          * CDT context, to be canceled later if finally found to be incompatible
 671          * when being re-started */
 672         if (larr->arr_status == ARS_STARTED) {
 673                 larr->arr_status = ARS_WAITING;
 674                 larr->arr_req_change = cfs_time_current_sec();
 675                 rc = llog_write(env, llh, hdr, hdr->lrh_index);
 676                 if (rc != 0)
 677                         GOTO(out, rc);
 678         }
 679
 680         OBD_SLAB_ALLOC_PTR(crh, mdt_hsm_cdt_kmem);
 681         if (crh == NULL)
 682                 RETURN(-ENOMEM);
 683
 684         crh->crh_fid = hai->hai_fid;
 685         /* in V1 all file is restored
 686         crh->extent.start = hai->hai_extent.offset;
 687         crh->extent.end = hai->hai_extent.offset + hai->hai_extent.length;
 688         */
 689         crh->crh_extent.start = 0;
 690         crh->crh_extent.end = hai->hai_extent.length;
 691         /* get the layout lock */
 692         mdt_lock_reg_init(&crh->crh_lh, LCK_EX);
 693         child = mdt_object_find_lock(mti, &crh->crh_fid, &crh->crh_lh,
 694                                      MDS_INODELOCK_LAYOUT);
 695         if (IS_ERR(child))
 696                 GOTO(out, rc = PTR_ERR(child));
 697
 698         rc = 0;
 699         /* we choose to not keep a reference
 700          * on the object during the restore time which can be very long */
 701         mdt_object_put(mti->mti_env, child);
 702
 703         mutex_lock(&cdt->cdt_restore_lock);
 704         list_add_tail(&crh->crh_list, &cdt->cdt_restore_hdl);
 705         mutex_unlock(&cdt->cdt_restore_lock);
 706
 707 out:
 708         RETURN(rc);
 709 }
 710
 711 /**
 712  * restore coordinator state at startup
 713  * the goal is to take a layout lock for each registered restore request
 714  * \param mti [IN] context
 715  */
 716 static int mdt_hsm_pending_restore(struct mdt_thread_info *mti)
 717 {
 718         struct hsm_restore_data  hrd;
 719         int                      rc;
 720         ENTRY;
 721
 722         hrd.hrd_mti = mti;
 723
 724         rc = cdt_llog_process(mti->mti_env, mti->mti_mdt,
 725                               hsm_restore_cb, &hrd);
 726
 727         RETURN(rc);
 728 }
 729
 730 static int hsm_init_ucred(struct lu_ucred *uc)
 731 {
 732         ENTRY;
 733
 734         uc->uc_valid = UCRED_OLD;
 735         uc->uc_o_uid = 0;
 736         uc->uc_o_gid = 0;
 737         uc->uc_o_fsuid = 0;
 738         uc->uc_o_fsgid = 0;
 739         uc->uc_uid = 0;
 740         uc->uc_gid = 0;
 741         uc->uc_fsuid = 0;
 742         uc->uc_fsgid = 0;
 743         uc->uc_suppgids[0] = -1;
 744         uc->uc_suppgids[1] = -1;
 745         uc->uc_cap = CFS_CAP_FS_MASK;
 746         uc->uc_umask = 0777;
 747         uc->uc_ginfo = NULL;
 748         uc->uc_identity = NULL;
 749
 750         RETURN(0);
 751 }
 752
 753 /**
 754  * wake up coordinator thread
 755  * \param mdt [IN] device
 756  * \retval 0 success
 757  * \retval -ve failure
 758  */
 759 int mdt_hsm_cdt_wakeup(struct mdt_device *mdt)
 760 {
 761         struct coordinator      *cdt = &mdt->mdt_coordinator;
 762         ENTRY;
 763
 764         if (cdt->cdt_state == CDT_STOPPED)
 765                 RETURN(-ESRCH);
 766
 767         /* wake up coordinator */
 768         cdt->cdt_thread.t_flags = SVC_EVENT;
 769         wake_up(&cdt->cdt_thread.t_ctl_waitq);
 770
 771         RETURN(0);
 772 }
 773
 774 /**
 775  * initialize coordinator struct
 776  * \param mdt [IN] device
 777  * \retval 0 success
 778  * \retval -ve failure
 779  */
 780 int mdt_hsm_cdt_init(struct mdt_device *mdt)
 781 {
 782         struct coordinator      *cdt = &mdt->mdt_coordinator;
 783         struct mdt_thread_info  *cdt_mti = NULL;
 784         int                      rc;
 785         ENTRY;
 786
 787         cdt->cdt_state = CDT_STOPPED;
 788
 789         init_waitqueue_head(&cdt->cdt_thread.t_ctl_waitq);
 790         mutex_init(&cdt->cdt_llog_lock);
 791         init_rwsem(&cdt->cdt_agent_lock);
 792         init_rwsem(&cdt->cdt_request_lock);
 793         mutex_init(&cdt->cdt_restore_lock);
 794
 795         INIT_LIST_HEAD(&cdt->cdt_requests);
 796         INIT_LIST_HEAD(&cdt->cdt_agents);
 797         INIT_LIST_HEAD(&cdt->cdt_restore_hdl);
 798
 799         rc = lu_env_init(&cdt->cdt_env, LCT_MD_THREAD);
 800         if (rc < 0)
 801                 RETURN(rc);
 802
 803         /* for mdt_ucred(), lu_ucred stored in lu_ucred_key */
 804         rc = lu_context_init(&cdt->cdt_session, LCT_SERVER_SESSION);
 805         if (rc == 0) {
 806                 lu_context_enter(&cdt->cdt_session);
 807                 cdt->cdt_env.le_ses = &cdt->cdt_session;
 808         } else {
 809                 lu_env_fini(&cdt->cdt_env);
 810                 RETURN(rc);
 811         }
 812
 813         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
 814         LASSERT(cdt_mti != NULL);
 815
 816         cdt_mti->mti_env = &cdt->cdt_env;
 817         cdt_mti->mti_mdt = mdt;
 818
 819         hsm_init_ucred(mdt_ucred(cdt_mti));
 820
 821         /* default values for /proc tunnables
 822          * can be override by MGS conf */
 823         cdt->cdt_default_archive_id = 1;
 824         cdt->cdt_grace_delay = 60;
 825         cdt->cdt_loop_period = 10;
 826         cdt->cdt_max_requests = 3;
 827         cdt->cdt_policy = CDT_DEFAULT_POLICY;
 828         cdt->cdt_active_req_timeout = 3600;
 829
 830         RETURN(0);
 831 }
 832
 833 /**
 834  * free a coordinator thread
 835  * \param mdt [IN] device
 836  */
 837 int  mdt_hsm_cdt_fini(struct mdt_device *mdt)
 838 {
 839         struct coordinator *cdt = &mdt->mdt_coordinator;
 840         ENTRY;
 841
 842         lu_context_exit(cdt->cdt_env.le_ses);
 843         lu_context_fini(cdt->cdt_env.le_ses);
 844
 845         lu_env_fini(&cdt->cdt_env);
 846
 847         RETURN(0);
 848 }
 849
 850 /**
 851  * start a coordinator thread
 852  * \param mdt [IN] device
 853  * \retval 0 success
 854  * \retval -ve failure
 855  */
 856 int mdt_hsm_cdt_start(struct mdt_device *mdt)
 857 {
 858         struct coordinator      *cdt = &mdt->mdt_coordinator;
 859         int                      rc;
 860         void                    *ptr;
 861         struct mdt_thread_info  *cdt_mti;
 862         struct task_struct      *task;
 863         ENTRY;
 864
 865         /* functions defined but not yet used
 866          * this avoid compilation warning
 867          */
 868         ptr = dump_requests;
 869
 870         if (cdt->cdt_state != CDT_STOPPED) {
 871                 CERROR("%s: Coordinator already started\n",
 872                        mdt_obd_name(mdt));
 873                 RETURN(-EALREADY);
 874         }
 875
 876         CLASSERT(1 << (CDT_POLICY_SHIFT_COUNT - 1) == CDT_POLICY_LAST);
 877         cdt->cdt_policy = CDT_DEFAULT_POLICY;
 878
 879         cdt->cdt_state = CDT_INIT;
 880
 881         atomic_set(&cdt->cdt_compound_id, cfs_time_current_sec());
 882         /* just need to be larger than previous one */
 883         /* cdt_last_cookie is protected by cdt_llog_lock */
 884         cdt->cdt_last_cookie = cfs_time_current_sec();
 885         atomic_set(&cdt->cdt_request_count, 0);
 886         cdt->cdt_user_request_mask = (1UL << HSMA_RESTORE);
 887         cdt->cdt_group_request_mask = (1UL << HSMA_RESTORE);
 888         cdt->cdt_other_request_mask = (1UL << HSMA_RESTORE);
 889
 890         /* to avoid deadlock when start is made through /proc
 891          * /proc entries are created by the coordinator thread */
 892
 893         /* set up list of started restore requests */
 894         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
 895         rc = mdt_hsm_pending_restore(cdt_mti);
 896         if (rc)
 897                 CERROR("%s: cannot take the layout locks needed"
 898                        " for registered restore: %d\n",
 899                        mdt_obd_name(mdt), rc);
 900
 901         task = kthread_run(mdt_coordinator, cdt_mti, "hsm_cdtr");
 902         if (IS_ERR(task)) {
 903                 rc = PTR_ERR(task);
 904                 cdt->cdt_state = CDT_STOPPED;
 905                 CERROR("%s: error starting coordinator thread: %d\n",
 906                        mdt_obd_name(mdt), rc);
 907                 RETURN(rc);
 908         } else {
 909                 CDEBUG(D_HSM, "%s: coordinator thread started\n",
 910                        mdt_obd_name(mdt));
 911                 rc = 0;
 912         }
 913
 914         wait_event(cdt->cdt_thread.t_ctl_waitq,
 915                        (cdt->cdt_thread.t_flags & SVC_RUNNING));
 916
 917         cdt->cdt_state = CDT_RUNNING;
 918         mdt->mdt_opts.mo_coordinator = 1;
 919         RETURN(0);
 920 }
 921
 922 /**
 923  * stop a coordinator thread
 924  * \param mdt [IN] device
 925  */
 926 int mdt_hsm_cdt_stop(struct mdt_device *mdt)
 927 {
 928         struct coordinator              *cdt = &mdt->mdt_coordinator;
 929         struct cdt_agent_req            *car, *tmp1;
 930         struct hsm_agent                *ha, *tmp2;
 931         struct cdt_restore_handle       *crh, *tmp3;
 932         struct mdt_thread_info          *cdt_mti;
 933         ENTRY;
 934
 935         if (cdt->cdt_state == CDT_STOPPED) {
 936                 CERROR("%s: Coordinator already stopped\n",
 937                        mdt_obd_name(mdt));
 938                 RETURN(-EALREADY);
 939         }
 940
 941         if (cdt->cdt_state != CDT_STOPPING) {
 942                 /* stop coordinator thread before cleaning */
 943                 cdt->cdt_thread.t_flags = SVC_STOPPING;
 944                 wake_up(&cdt->cdt_thread.t_ctl_waitq);
 945                 wait_event(cdt->cdt_thread.t_ctl_waitq,
 946                            cdt->cdt_thread.t_flags & SVC_STOPPED);
 947         }
 948         cdt->cdt_state = CDT_STOPPED;
 949
 950         /* start cleaning */
 951         down_write(&cdt->cdt_request_lock);
 952         list_for_each_entry_safe(car, tmp1, &cdt->cdt_requests,
 953                                  car_request_list) {
 954                 list_del(&car->car_request_list);
 955                 mdt_cdt_free_request(car);
 956         }
 957         up_write(&cdt->cdt_request_lock);
 958
 959         down_write(&cdt->cdt_agent_lock);
 960         list_for_each_entry_safe(ha, tmp2, &cdt->cdt_agents, ha_list) {
 961                 list_del(&ha->ha_list);
 962                 OBD_FREE_PTR(ha);
 963         }
 964         up_write(&cdt->cdt_agent_lock);
 965
 966         cdt_mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
 967         mutex_lock(&cdt->cdt_restore_lock);
 968         list_for_each_entry_safe(crh, tmp3, &cdt->cdt_restore_hdl, crh_list) {
 969                 struct mdt_object       *child;
 970
 971                 /* give back layout lock */
 972                 child = mdt_object_find(&cdt->cdt_env, mdt, &crh->crh_fid);
 973                 if (!IS_ERR(child))
 974                         mdt_object_unlock_put(cdt_mti, child, &crh->crh_lh, 1);
 975
 976                 list_del(&crh->crh_list);
 977
 978                 OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
 979         }
 980         mutex_unlock(&cdt->cdt_restore_lock);
 981
 982         mdt->mdt_opts.mo_coordinator = 0;
 983
 984         RETURN(0);
 985 }
 986
 987 /**
 988  * register all requests from an hal in the memory list
 989  * \param mti [IN] context
 990  * \param hal [IN] request
 991  * \param uuid [OUT] in case of CANCEL, the uuid of the agent
 992  *  which is running the CT
 993  * \retval 0 success
 994  * \retval -ve failure
 995  */
 996 int mdt_hsm_add_hal(struct mdt_thread_info *mti,
 997                     struct hsm_action_list *hal, struct obd_uuid *uuid)
 998 {
 999         struct mdt_device       *mdt = mti->mti_mdt;
1000         struct coordinator      *cdt = &mdt->mdt_coordinator;
1001         struct hsm_action_item  *hai;
1002         int                      rc = 0, i;
1003         ENTRY;
1004
1005         /* register request in memory list */
1006         hai = hai_first(hal);
1007         for (i = 0; i < hal->hal_count; i++, hai = hai_next(hai)) {
1008                 struct cdt_agent_req *car;
1009
1010                 /* in case of a cancel request, we first mark the ondisk
1011                  * record of the request we want to stop as canceled
1012                  * this does not change the cancel record
1013                  * it will be done when updating the request status
1014                  */
1015                 if (hai->hai_action == HSMA_CANCEL) {
1016                         rc = mdt_agent_record_update(mti->mti_env, mti->mti_mdt,
1017                                                      &hai->hai_cookie,
1018                                                      1, ARS_CANCELED);
1019                         if (rc) {
1020                                 CERROR("%s: mdt_agent_record_update() failed, "
1021                                        "rc=%d, cannot update status to %s "
1022                                        "for cookie "LPX64"\n",
1023                                        mdt_obd_name(mdt), rc,
1024                                        agent_req_status2name(ARS_CANCELED),
1025                                        hai->hai_cookie);
1026                                 GOTO(out, rc);
1027                         }
1028
1029                         /* find the running request to set it canceled */
1030                         car = mdt_cdt_find_request(cdt, hai->hai_cookie, NULL);
1031                         if (car != NULL) {
1032                                 car->car_canceled = 1;
1033                                 /* uuid has to be changed to the one running the
1034                                 * request to cancel */
1035                                 *uuid = car->car_uuid;
1036                                 mdt_cdt_put_request(car);
1037                         }
1038                         /* no need to memorize cancel request
1039                          * this also avoid a deadlock when we receive
1040                          * a purge all requests command
1041                          */
1042                         continue;
1043                 }
1044
1045                 if (hai->hai_action == HSMA_ARCHIVE) {
1046                         struct mdt_object *obj;
1047                         struct md_hsm hsm;
1048
1049                         obj = mdt_hsm_get_md_hsm(mti, &hai->hai_fid, &hsm);
1050                         if (IS_ERR(obj) && (PTR_ERR(obj) == -ENOENT))
1051                                 continue;
1052                         if (IS_ERR(obj))
1053                                 GOTO(out, rc = PTR_ERR(obj));
1054
1055                         hsm.mh_flags |= HS_EXISTS;
1056                         hsm.mh_arch_id = hal->hal_archive_id;
1057                         rc = mdt_hsm_attr_set(mti, obj, &hsm);
1058                         mdt_object_put(mti->mti_env, obj);
1059                         if (rc)
1060                                 GOTO(out, rc);
1061                 }
1062
1063                 car = mdt_cdt_alloc_request(hal->hal_compound_id,
1064                                             hal->hal_archive_id, hal->hal_flags,
1065                                             uuid, hai);
1066                 if (IS_ERR(car))
1067                         GOTO(out, rc = PTR_ERR(car));
1068
1069                 rc = mdt_cdt_add_request(cdt, car);
1070                 if (rc != 0)
1071                         mdt_cdt_free_request(car);
1072         }
1073 out:
1074         RETURN(rc);
1075 }
1076
1077 /**
1078  * swap layouts between 2 fids
1079  * \param mti [IN] context
1080  * \param fid1 [IN]
1081  * \param fid2 [IN]
1082  * \param mh_common [IN] MD HSM
1083  */
1084 static int hsm_swap_layouts(struct mdt_thread_info *mti,
1085                             const lustre_fid *fid, const lustre_fid *dfid,
1086                             struct md_hsm *mh_common)
1087 {
1088         struct mdt_device       *mdt = mti->mti_mdt;
1089         struct mdt_object       *child1, *child2;
1090         struct mdt_lock_handle  *lh2;
1091         int                      rc;
1092         ENTRY;
1093
1094         child1 = mdt_object_find(mti->mti_env, mdt, fid);
1095         if (IS_ERR(child1))
1096                 GOTO(out, rc = PTR_ERR(child1));
1097
1098         /* we already have layout lock on FID so take only
1099          * on dfid */
1100         lh2 = &mti->mti_lh[MDT_LH_OLD];
1101         mdt_lock_reg_init(lh2, LCK_EX);
1102         child2 = mdt_object_find_lock(mti, dfid, lh2, MDS_INODELOCK_LAYOUT);
1103         if (IS_ERR(child2))
1104                 GOTO(out_child1, rc = PTR_ERR(child2));
1105
1106         /* if copy tool closes the volatile before sending the final
1107          * progress through llapi_hsm_copy_end(), all the objects
1108          * are removed and mdd_swap_layout LBUG */
1109         if (!mdt_object_exists(child2)) {
1110                 CERROR("%s: Copytool has closed volatile file "DFID"\n",
1111                        mdt_obd_name(mti->mti_mdt), PFID(dfid));
1112                 GOTO(out_child2, rc = -ENOENT);
1113         }
1114         /* Since we only handle restores here, unconditionally use
1115          * SWAP_LAYOUTS_MDS_HSM flag to ensure original layout will
1116          * be preserved in case of failure during swap_layout and not
1117          * leave a file in an intermediate but incoherent state.
1118          * But need to setup HSM xattr of data FID before, reuse
1119          * mti and mh presets for FID in hsm_cdt_request_completed(),
1120          * only need to clear RELEASED and DIRTY.
1121          */
1122         mh_common->mh_flags &= ~(HS_RELEASED | HS_DIRTY);
1123         rc = mdt_hsm_attr_set(mti, child2, mh_common);
1124         if (rc == 0)
1125                 rc = mo_swap_layouts(mti->mti_env,
1126                                      mdt_object_child(child1),
1127                                      mdt_object_child(child2),
1128                                      SWAP_LAYOUTS_MDS_HSM);
1129
1130 out_child2:
1131         mdt_object_unlock_put(mti, child2, lh2, 1);
1132 out_child1:
1133         mdt_object_put(mti->mti_env, child1);
1134 out:
1135         RETURN(rc);
1136 }
1137
1138 /**
1139  * update status of a completed request
1140  * \param mti [IN] context
1141  * \param pgs [IN] progress of the copy tool
1142  * \param update_record [IN] update llog record
1143  * \retval 0 success
1144  * \retval -ve failure
1145  */
1146 static int hsm_cdt_request_completed(struct mdt_thread_info *mti,
1147                                      struct hsm_progress_kernel *pgs,
1148                                      const struct cdt_agent_req *car,
1149                                      enum agent_req_status *status)
1150 {
1151         const struct lu_env     *env = mti->mti_env;
1152         struct mdt_device       *mdt = mti->mti_mdt;
1153         struct coordinator      *cdt = &mdt->mdt_coordinator;
1154         struct mdt_object       *obj = NULL;
1155         int                      cl_flags = 0, rc = 0;
1156         struct md_hsm            mh;
1157         bool                     is_mh_changed;
1158         ENTRY;
1159
1160         /* default is to retry */
1161         *status = ARS_WAITING;
1162
1163         /* find object by FID */
1164         obj = mdt_hsm_get_md_hsm(mti, &car->car_hai->hai_fid, &mh);
1165         /* we will update MD HSM only if needed */
1166         is_mh_changed = false;
1167         if (IS_ERR(obj)) {
1168                 /* object removed */
1169                 *status = ARS_SUCCEED;
1170                 goto unlock;
1171         }
1172
1173         /* no need to change mh->mh_arch_id
1174          * mdt_hsm_get_md_hsm() got it from disk and it is still valid
1175          */
1176         if (pgs->hpk_errval != 0) {
1177                 switch (pgs->hpk_errval) {
1178                 case ENOSYS:
1179                         /* the copy tool does not support cancel
1180                          * so the cancel request is failed
1181                          * As we cannot distinguish a cancel progress
1182                          * from another action progress (they have the
1183                          * same cookie), we suppose here the CT returns
1184                          * ENOSYS only if does not support cancel
1185                          */
1186                         /* this can also happen when cdt calls it to
1187                          * for a timed out request */
1188                         *status = ARS_FAILED;
1189                         /* to have a cancel event in changelog */
1190                         pgs->hpk_errval = ECANCELED;
1191                         break;
1192                 case ECANCELED:
1193                         /* the request record has already been set to
1194                          * ARS_CANCELED, this set the cancel request
1195                          * to ARS_SUCCEED */
1196                         *status = ARS_SUCCEED;
1197                         break;
1198                 default:
1199                         *status = (cdt->cdt_policy & CDT_NORETRY_ACTION ||
1200                                    !(pgs->hpk_flags & HP_FLAG_RETRY) ?
1201                                    ARS_FAILED : ARS_WAITING);
1202                         break;
1203                 }
1204
1205                 if (pgs->hpk_errval > CLF_HSM_MAXERROR) {
1206                         CERROR("%s: Request "LPX64" on "DFID
1207                                " failed, error code %d too large\n",
1208                                mdt_obd_name(mdt),
1209                                pgs->hpk_cookie, PFID(&pgs->hpk_fid),
1210                                pgs->hpk_errval);
1211                         hsm_set_cl_error(&cl_flags,
1212                                          CLF_HSM_ERROVERFLOW);
1213                         rc = -EINVAL;
1214                 } else {
1215                         hsm_set_cl_error(&cl_flags, pgs->hpk_errval);
1216                 }
1217
1218                 switch (car->car_hai->hai_action) {
1219                 case HSMA_ARCHIVE:
1220                         hsm_set_cl_event(&cl_flags, HE_ARCHIVE);
1221                         break;
1222                 case HSMA_RESTORE:
1223                         hsm_set_cl_event(&cl_flags, HE_RESTORE);
1224                         break;
1225                 case HSMA_REMOVE:
1226                         hsm_set_cl_event(&cl_flags, HE_REMOVE);
1227                         break;
1228                 case HSMA_CANCEL:
1229                         hsm_set_cl_event(&cl_flags, HE_CANCEL);
1230                         CERROR("%s: Failed request "LPX64" on "DFID
1231                                " cannot be a CANCEL\n",
1232                                mdt_obd_name(mdt),
1233                                pgs->hpk_cookie,
1234                                PFID(&pgs->hpk_fid));
1235                         break;
1236                 default:
1237                         CERROR("%s: Failed request "LPX64" on "DFID
1238                                " %d is an unknown action\n",
1239                                mdt_obd_name(mdt),
1240                                pgs->hpk_cookie, PFID(&pgs->hpk_fid),
1241                                car->car_hai->hai_action);
1242                         rc = -EINVAL;
1243                         break;
1244                 }
1245         } else {
1246                 *status = ARS_SUCCEED;
1247                 switch (car->car_hai->hai_action) {
1248                 case HSMA_ARCHIVE:
1249                         hsm_set_cl_event(&cl_flags, HE_ARCHIVE);
1250                         /* set ARCHIVE keep EXIST and clear LOST and
1251                          * DIRTY */
1252                         mh.mh_arch_ver = pgs->hpk_data_version;
1253                         mh.mh_flags |= HS_ARCHIVED;
1254                         mh.mh_flags &= ~(HS_LOST|HS_DIRTY);
1255                         is_mh_changed = true;
1256                         break;
1257                 case HSMA_RESTORE:
1258                         hsm_set_cl_event(&cl_flags, HE_RESTORE);
1259
1260                         /* do not clear RELEASED and DIRTY here
1261                          * this will occur in hsm_swap_layouts()
1262                          */
1263
1264                         /* Restoring has changed the file version on
1265                          * disk. */
1266                         mh.mh_arch_ver = pgs->hpk_data_version;
1267                         is_mh_changed = true;
1268                         break;
1269                 case HSMA_REMOVE:
1270                         hsm_set_cl_event(&cl_flags, HE_REMOVE);
1271                         /* clear ARCHIVED EXISTS and LOST */
1272                         mh.mh_flags &= ~(HS_ARCHIVED | HS_EXISTS | HS_LOST);
1273                         is_mh_changed = true;
1274                         break;
1275                 case HSMA_CANCEL:
1276                         hsm_set_cl_event(&cl_flags, HE_CANCEL);
1277                         CERROR("%s: Successful request "LPX64
1278                                " on "DFID
1279                                " cannot be a CANCEL\n",
1280                                mdt_obd_name(mdt),
1281                                pgs->hpk_cookie,
1282                                PFID(&pgs->hpk_fid));
1283                         break;
1284                 default:
1285                         CERROR("%s: Successful request "LPX64
1286                                " on "DFID
1287                                " %d is an unknown action\n",
1288                                mdt_obd_name(mdt),
1289                                pgs->hpk_cookie, PFID(&pgs->hpk_fid),
1290                                car->car_hai->hai_action);
1291                         rc = -EINVAL;
1292                         break;
1293                 }
1294         }
1295
1296         /* rc != 0 means error when analysing action, it may come from
1297          * a crasy CT no need to manage DIRTY
1298          */
1299         if (rc == 0)
1300                 hsm_set_cl_flags(&cl_flags,
1301                                  mh.mh_flags & HS_DIRTY ? CLF_HSM_DIRTY : 0);
1302
1303         /* unlock is done later, after layout lock management */
1304         if (is_mh_changed)
1305                 rc = mdt_hsm_attr_set(mti, obj, &mh);
1306
1307 unlock:
1308         /* we give back layout lock only if restore was successful or
1309          * if restore was canceled or if policy is to not retry
1310          * in other cases we just unlock the object */
1311         if (car->car_hai->hai_action == HSMA_RESTORE &&
1312             (pgs->hpk_errval == 0 || pgs->hpk_errval == ECANCELED ||
1313              cdt->cdt_policy & CDT_NORETRY_ACTION)) {
1314                 struct cdt_restore_handle       *crh;
1315
1316                 /* restore in data FID done, we swap the layouts
1317                  * only if restore is successful */
1318                 if (pgs->hpk_errval == 0) {
1319                         rc = hsm_swap_layouts(mti, &car->car_hai->hai_fid,
1320                                               &car->car_hai->hai_dfid, &mh);
1321                         if (rc) {
1322                                 if (cdt->cdt_policy & CDT_NORETRY_ACTION)
1323                                         *status = ARS_FAILED;
1324                                 pgs->hpk_errval = -rc;
1325                         }
1326                 }
1327                 /* we have to retry, so keep layout lock */
1328                 if (*status == ARS_WAITING)
1329                         GOTO(out, rc);
1330
1331                 /* give back layout lock */
1332                 mutex_lock(&cdt->cdt_restore_lock);
1333                 crh = mdt_hsm_restore_hdl_find(cdt, &car->car_hai->hai_fid);
1334                 if (crh != NULL)
1335                         list_del(&crh->crh_list);
1336                 mutex_unlock(&cdt->cdt_restore_lock);
1337                 /* just give back layout lock, we keep
1338                  * the reference which is given back
1339                  * later with the lock for HSM flags */
1340                 if (!IS_ERR(obj) && crh != NULL)
1341                         mdt_object_unlock(mti, obj, &crh->crh_lh, 1);
1342
1343                 if (crh != NULL)
1344                         OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
1345         }
1346
1347         GOTO(out, rc);
1348
1349 out:
1350         if (obj != NULL && !IS_ERR(obj)) {
1351                 mo_changelog(env, CL_HSM, cl_flags,
1352                              mdt_object_child(obj));
1353                 mdt_object_put(mti->mti_env, obj);
1354         }
1355
1356         RETURN(rc);
1357 }
1358
1359 /**
1360  * update status of a request
1361  * \param mti [IN] context
1362  * \param pgs [IN] progress of the copy tool
1363  * \param update_record [IN] update llog record
1364  * \retval 0 success
1365  * \retval -ve failure
1366  */
1367 int mdt_hsm_update_request_state(struct mdt_thread_info *mti,
1368                                  struct hsm_progress_kernel *pgs,
1369                                  const int update_record)
1370 {
1371         struct mdt_device       *mdt = mti->mti_mdt;
1372         struct coordinator      *cdt = &mdt->mdt_coordinator;
1373         struct cdt_agent_req    *car;
1374         int                      rc = 0;
1375         ENTRY;
1376
1377         /* no coordinator started, so we cannot serve requests */
1378         if (cdt->cdt_state == CDT_STOPPED)
1379                 RETURN(-EAGAIN);
1380
1381         /* first do sanity checks */
1382         car = mdt_cdt_update_request(cdt, pgs);
1383         if (IS_ERR(car)) {
1384                 CERROR("%s: Cannot find running request for cookie "LPX64
1385                        " on fid="DFID"\n",
1386                        mdt_obd_name(mdt),
1387                        pgs->hpk_cookie, PFID(&pgs->hpk_fid));
1388
1389                 RETURN(PTR_ERR(car));
1390         }
1391
1392         CDEBUG(D_HSM, "Progress received for fid="DFID" cookie="LPX64
1393                       " action=%s flags=%d err=%d fid="DFID" dfid="DFID"\n",
1394                       PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1395                       hsm_copytool_action2name(car->car_hai->hai_action),
1396                       pgs->hpk_flags, pgs->hpk_errval,
1397                       PFID(&car->car_hai->hai_fid),
1398                       PFID(&car->car_hai->hai_dfid));
1399
1400         /* progress is done on FID or data FID depending of the action and
1401          * of the copy progress */
1402         /* for restore progress is used to send back the data FID to cdt */
1403         if (car->car_hai->hai_action == HSMA_RESTORE &&
1404             lu_fid_eq(&car->car_hai->hai_fid, &car->car_hai->hai_dfid))
1405                 car->car_hai->hai_dfid = pgs->hpk_fid;
1406
1407         if ((car->car_hai->hai_action == HSMA_RESTORE ||
1408              car->car_hai->hai_action == HSMA_ARCHIVE) &&
1409             (!lu_fid_eq(&pgs->hpk_fid, &car->car_hai->hai_dfid) &&
1410              !lu_fid_eq(&pgs->hpk_fid, &car->car_hai->hai_fid))) {
1411                 CERROR("%s: Progress on "DFID" for cookie "LPX64
1412                        " does not match request FID "DFID" nor data FID "
1413                        DFID"\n",
1414                        mdt_obd_name(mdt),
1415                        PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1416                        PFID(&car->car_hai->hai_fid),
1417                        PFID(&car->car_hai->hai_dfid));
1418                 GOTO(out, rc = -EINVAL);
1419         }
1420
1421         if (pgs->hpk_errval != 0 && !(pgs->hpk_flags & HP_FLAG_COMPLETED)) {
1422                 CERROR("%s: Progress on "DFID" for cookie "LPX64" action=%s"
1423                        " is not coherent (err=%d and not completed"
1424                        " (flags=%d))\n",
1425                        mdt_obd_name(mdt),
1426                        PFID(&pgs->hpk_fid), pgs->hpk_cookie,
1427                        hsm_copytool_action2name(car->car_hai->hai_action),
1428                        pgs->hpk_errval, pgs->hpk_flags);
1429                 GOTO(out, rc = -EINVAL);
1430         }
1431
1432         /* now progress is valid */
1433
1434         /* we use a root like ucred */
1435         hsm_init_ucred(mdt_ucred(mti));
1436
1437         if (pgs->hpk_flags & HP_FLAG_COMPLETED) {
1438                 enum agent_req_status    status;
1439
1440                 rc = hsm_cdt_request_completed(mti, pgs, car, &status);
1441
1442                 /* remove request from memory list */
1443                 mdt_cdt_remove_request(cdt, pgs->hpk_cookie);
1444
1445                 CDEBUG(D_HSM, "Updating record: fid="DFID" cookie="LPX64
1446                               " action=%s status=%s\n", PFID(&pgs->hpk_fid),
1447                        pgs->hpk_cookie,
1448                        hsm_copytool_action2name(car->car_hai->hai_action),
1449                        agent_req_status2name(status));
1450
1451                 if (update_record) {
1452                         int rc1;
1453
1454                         rc1 = mdt_agent_record_update(mti->mti_env, mdt,
1455                                                      &pgs->hpk_cookie, 1,
1456                                                      status);
1457                         if (rc1)
1458                                 CERROR("%s: mdt_agent_record_update() failed,"
1459                                        " rc=%d, cannot update status to %s"
1460                                        " for cookie "LPX64"\n",
1461                                        mdt_obd_name(mdt), rc1,
1462                                        agent_req_status2name(status),
1463                                        pgs->hpk_cookie);
1464                         rc = (rc != 0 ? rc : rc1);
1465                 }
1466                 /* ct has completed a request, so a slot is available, wakeup
1467                  * cdt to find new work */
1468                 mdt_hsm_cdt_wakeup(mdt);
1469         } else {
1470                 /* if copytool send a progress on a canceled request
1471                  * we inform copytool it should stop
1472                  */
1473                 if (car->car_canceled == 1)
1474                         rc = -ECANCELED;
1475         }
1476         GOTO(out, rc);
1477
1478 out:
1479         /* remove ref got from mdt_cdt_update_request() */
1480         mdt_cdt_put_request(car);
1481
1482         return rc;
1483 }
1484
1485
1486 /**
1487  * data passed to llog_cat_process() callback
1488  * to cancel requests
1489  */
1490 struct hsm_cancel_all_data {
1491         struct mdt_device       *mdt;
1492 };
1493
1494 /**
1495  *  llog_cat_process() callback, used to:
1496  *  - purge all requests
1497  * \param env [IN] environment
1498  * \param llh [IN] llog handle
1499  * \param hdr [IN] llog record
1500  * \param data [IN] cb data = struct hsm_cancel_all_data
1501  * \retval 0 success
1502  * \retval -ve failure
1503  */
1504 static int mdt_cancel_all_cb(const struct lu_env *env,
1505                              struct llog_handle *llh,
1506                              struct llog_rec_hdr *hdr, void *data)
1507 {
1508         struct llog_agent_req_rec       *larr;
1509         struct hsm_cancel_all_data      *hcad;
1510         int                              rc = 0;
1511         ENTRY;
1512
1513         larr = (struct llog_agent_req_rec *)hdr;
1514         hcad = data;
1515         if (larr->arr_status == ARS_WAITING ||
1516             larr->arr_status == ARS_STARTED) {
1517                 larr->arr_status = ARS_CANCELED;
1518                 larr->arr_req_change = cfs_time_current_sec();
1519                 rc = llog_write(env, llh, hdr, hdr->lrh_index);
1520         }
1521
1522         RETURN(rc);
1523 }
1524
1525 /**
1526  * cancel all actions
1527  * \param obd [IN] MDT device
1528  */
1529 static int hsm_cancel_all_actions(struct mdt_device *mdt)
1530 {
1531         struct mdt_thread_info          *mti;
1532         struct coordinator              *cdt = &mdt->mdt_coordinator;
1533         struct cdt_agent_req            *car;
1534         struct hsm_action_list          *hal = NULL;
1535         struct hsm_action_item          *hai;
1536         struct hsm_cancel_all_data       hcad;
1537         int                              hal_sz = 0, hal_len, rc;
1538         enum cdt_states                  save_state;
1539         ENTRY;
1540
1541         /* retrieve coordinator context */
1542         mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
1543
1544         /* disable coordinator */
1545         save_state = cdt->cdt_state;
1546         cdt->cdt_state = CDT_DISABLE;
1547
1548         /* send cancel to all running requests */
1549         down_read(&cdt->cdt_request_lock);
1550         list_for_each_entry(car, &cdt->cdt_requests, car_request_list) {
1551                 mdt_cdt_get_request(car);
1552                 /* request is not yet removed from list, it will be done
1553                  * when copytool will return progress
1554                  */
1555
1556                 if (car->car_hai->hai_action == HSMA_CANCEL) {
1557                         mdt_cdt_put_request(car);
1558                         continue;
1559                 }
1560
1561                 /* needed size */
1562                 hal_len = sizeof(*hal) + cfs_size_round(MTI_NAME_MAXLEN + 1) +
1563                           cfs_size_round(car->car_hai->hai_len);
1564
1565                 if (hal_len > hal_sz && hal_sz > 0) {
1566                         /* not enough room, free old buffer */
1567                         OBD_FREE(hal, hal_sz);
1568                         hal = NULL;
1569                 }
1570
1571                 /* empty buffer, allocate one */
1572                 if (hal == NULL) {
1573                         hal_sz = hal_len;
1574                         OBD_ALLOC(hal, hal_sz);
1575                         if (hal == NULL) {
1576                                 mdt_cdt_put_request(car);
1577                                 up_read(&cdt->cdt_request_lock);
1578                                 GOTO(out, rc = -ENOMEM);
1579                         }
1580                 }
1581
1582                 hal->hal_version = HAL_VERSION;
1583                 obd_uuid2fsname(hal->hal_fsname, mdt_obd_name(mdt),
1584                                 MTI_NAME_MAXLEN);
1585                 hal->hal_fsname[MTI_NAME_MAXLEN] = '\0';
1586                 hal->hal_compound_id = car->car_compound_id;
1587                 hal->hal_archive_id = car->car_archive_id;
1588                 hal->hal_flags = car->car_flags;
1589                 hal->hal_count = 0;
1590
1591                 hai = hai_first(hal);
1592                 memcpy(hai, car->car_hai, car->car_hai->hai_len);
1593                 hai->hai_action = HSMA_CANCEL;
1594                 hal->hal_count = 1;
1595
1596                 /* it is possible to safely call mdt_hsm_agent_send()
1597                  * (ie without a deadlock on cdt_request_lock), because the
1598                  * write lock is taken only if we are not in purge mode
1599                  * (mdt_hsm_agent_send() does not call mdt_cdt_add_request()
1600                  *   nor mdt_cdt_remove_request())
1601                  */
1602                 /* no conflict with cdt thread because cdt is disable and we
1603                  * have the request lock */
1604                 mdt_hsm_agent_send(mti, hal, 1);
1605
1606                 mdt_cdt_put_request(car);
1607         }
1608         up_read(&cdt->cdt_request_lock);
1609
1610         if (hal != NULL)
1611                 OBD_FREE(hal, hal_sz);
1612
1613         /* cancel all on-disk records */
1614         hcad.mdt = mdt;
1615
1616         rc = cdt_llog_process(mti->mti_env, mti->mti_mdt,
1617                               mdt_cancel_all_cb, &hcad);
1618 out:
1619         /* enable coordinator */
1620         cdt->cdt_state = save_state;
1621
1622         RETURN(rc);
1623 }
1624
1625 /**
1626  * check if a request is compatible with file status
1627  * \param hai [IN] request description
1628  * \param hal_an [IN] request archive number (not used)
1629  * \param rq_flags [IN] request flags
1630  * \param hsm [IN] file HSM metadata
1631  * \retval boolean
1632  */
1633 bool mdt_hsm_is_action_compat(const struct hsm_action_item *hai,
1634                               const int hal_an, const __u64 rq_flags,
1635                               const struct md_hsm *hsm)
1636 {
1637         int      is_compat = false;
1638         int      hsm_flags;
1639         ENTRY;
1640
1641         hsm_flags = hsm->mh_flags;
1642         switch (hai->hai_action) {
1643         case HSMA_ARCHIVE:
1644                 if (!(hsm_flags & HS_NOARCHIVE) &&
1645                     (hsm_flags & HS_DIRTY || !(hsm_flags & HS_ARCHIVED)))
1646                         is_compat = true;
1647                 break;
1648         case HSMA_RESTORE:
1649                 if (!(hsm_flags & HS_DIRTY) && (hsm_flags & HS_RELEASED) &&
1650                     hsm_flags & HS_ARCHIVED && !(hsm_flags & HS_LOST))
1651                         is_compat = true;
1652                 break;
1653         case HSMA_REMOVE:
1654                 if (!(hsm_flags & HS_RELEASED) &&
1655                     (hsm_flags & (HS_ARCHIVED | HS_EXISTS)))
1656                         is_compat = true;
1657                 break;
1658         case HSMA_CANCEL:
1659                 is_compat = true;
1660                 break;
1661         }
1662         CDEBUG(D_HSM, "fid="DFID" action=%s flags="LPX64
1663                       " extent="LPX64"-"LPX64" hsm_flags=%.8X %s\n",
1664                       PFID(&hai->hai_fid),
1665                       hsm_copytool_action2name(hai->hai_action), rq_flags,
1666                       hai->hai_extent.offset, hai->hai_extent.length,
1667                       hsm->mh_flags,
1668                       (is_compat ? "compatible" : "uncompatible"));
1669
1670         RETURN(is_compat);
1671 }
1672
1673 /*
1674  * /proc interface used to get/set HSM behaviour (cdt->cdt_policy)
1675  */
1676 static const struct {
1677         __u64            bit;
1678         char            *name;
1679         char            *nickname;
1680 } hsm_policy_names[] = {
1681         { CDT_NONBLOCKING_RESTORE,      "NonBlockingRestore",   "NBR"},
1682         { CDT_NORETRY_ACTION,           "NoRetryAction",        "NRA"},
1683         { 0 },
1684 };
1685
1686 /**
1687  * convert a policy name to a bit
1688  * \param name [IN] policy name
1689  * \retval 0 unknown
1690  * \retval   policy bit
1691  */
1692 static __u64 hsm_policy_str2bit(const char *name)
1693 {
1694         int      i;
1695
1696         for (i = 0; hsm_policy_names[i].bit != 0; i++)
1697                 if (strcmp(hsm_policy_names[i].nickname, name) == 0 ||
1698                     strcmp(hsm_policy_names[i].name, name) == 0)
1699                         return hsm_policy_names[i].bit;
1700         return 0;
1701 }
1702
1703 /**
1704  * convert a policy bit field to a string
1705  * \param mask [IN] policy bit field
1706  * \param hexa [IN] print mask before bit names
1707  * \param buffer [OUT] string
1708  * \param count [IN] size of buffer
1709  */
1710 static void hsm_policy_bit2str(struct seq_file *m, const __u64 mask,
1711                                 const bool hexa)
1712 {
1713         int      i, j;
1714         __u64    bit;
1715         ENTRY;
1716
1717         if (hexa)
1718                 seq_printf(m, "("LPX64") ", mask);
1719
1720         for (i = 0; i < CDT_POLICY_SHIFT_COUNT; i++) {
1721                 bit = (1ULL << i);
1722
1723                 for (j = 0; hsm_policy_names[j].bit != 0; j++) {
1724                         if (hsm_policy_names[j].bit == bit)
1725                                 break;
1726                 }
1727                 if (bit & mask)
1728                         seq_printf(m, "[%s] ", hsm_policy_names[j].name);
1729                 else
1730                         seq_printf(m, "%s ", hsm_policy_names[j].name);
1731         }
1732         /* remove last ' ' */
1733         m->count--;
1734         seq_putc(m, '\0');
1735 }
1736
1737 /* methods to read/write HSM policy flags */
1738 static int mdt_hsm_policy_seq_show(struct seq_file *m, void *data)
1739 {
1740         struct mdt_device       *mdt = m->private;
1741         struct coordinator      *cdt = &mdt->mdt_coordinator;
1742         ENTRY;
1743
1744         hsm_policy_bit2str(m, cdt->cdt_policy, false);
1745         RETURN(0);
1746 }
1747
1748 static ssize_t
1749 mdt_hsm_policy_seq_write(struct file *file, const char __user *buffer,
1750                          size_t count, loff_t *off)
1751 {
1752         struct seq_file         *m = file->private_data;
1753         struct mdt_device       *mdt = m->private;
1754         struct coordinator      *cdt = &mdt->mdt_coordinator;
1755         char                    *start, *token, sign;
1756         char                    *buf;
1757         __u64                    policy;
1758         __u64                    add_mask, remove_mask, set_mask;
1759         int                      rc;
1760         ENTRY;
1761
1762         if (count + 1 > PAGE_SIZE)
1763                 RETURN(-EINVAL);
1764
1765         OBD_ALLOC(buf, count + 1);
1766         if (buf == NULL)
1767                 RETURN(-ENOMEM);
1768
1769         if (copy_from_user(buf, buffer, count))
1770                 GOTO(out, rc = -EFAULT);
1771
1772         buf[count] = '\0';
1773
1774         start = buf;
1775         CDEBUG(D_HSM, "%s: receive new policy: '%s'\n", mdt_obd_name(mdt),
1776                start);
1777
1778         add_mask = remove_mask = set_mask = 0;
1779         do {
1780                 token = strsep(&start, "\n ");
1781                 sign = *token;
1782
1783                 if (sign == '\0')
1784                         continue;
1785
1786                 if (sign == '-' || sign == '+')
1787                         token++;
1788
1789                 policy = hsm_policy_str2bit(token);
1790                 if (policy == 0) {
1791                         CWARN("%s: '%s' is unknown, "
1792                               "supported policies are:\n", mdt_obd_name(mdt),
1793                               token);
1794                         hsm_policy_bit2str(m, 0, false);
1795                         GOTO(out, rc = -EINVAL);
1796                 }
1797                 switch (sign) {
1798                 case '-':
1799                         remove_mask |= policy;
1800                         break;
1801                 case '+':
1802                         add_mask |= policy;
1803                         break;
1804                 default:
1805                         set_mask |= policy;
1806                         break;
1807                 }
1808
1809         } while (start != NULL);
1810
1811         CDEBUG(D_HSM, "%s: new policy: rm="LPX64" add="LPX64" set="LPX64"\n",
1812                mdt_obd_name(mdt), remove_mask, add_mask, set_mask);
1813
1814         /* if no sign in all string, it is a clear and set
1815          * if some sign found, all unsigned are converted
1816          * to add
1817          * P1 P2 = set to P1 and P2
1818          * P1 -P2 = add P1 clear P2 same as +P1 -P2
1819          */
1820         if (remove_mask == 0 && add_mask == 0) {
1821                 cdt->cdt_policy = set_mask;
1822         } else {
1823                 cdt->cdt_policy |= set_mask | add_mask;
1824                 cdt->cdt_policy &= ~remove_mask;
1825         }
1826
1827         GOTO(out, rc = count);
1828
1829 out:
1830         OBD_FREE(buf, count + 1);
1831         RETURN(rc);
1832 }
1833 LPROC_SEQ_FOPS(mdt_hsm_policy);
1834
1835 #define GENERATE_PROC_METHOD(VAR)                                       \
1836 static int mdt_hsm_##VAR##_seq_show(struct seq_file *m, void *data)     \
1837 {                                                                       \
1838         struct mdt_device       *mdt = m->private;                      \
1839         struct coordinator      *cdt = &mdt->mdt_coordinator;           \
1840         ENTRY;                                                          \
1841                                                                         \
1842         seq_printf(m, LPU64"\n", (__u64)cdt->VAR);                      \
1843         RETURN(0);                                                      \
1844 }                                                                       \
1845 static ssize_t                                                          \
1846 mdt_hsm_##VAR##_seq_write(struct file *file, const char __user *buffer, \
1847                           size_t count, loff_t *off)                    \
1848                                                                         \
1849 {                                                                       \
1850         struct seq_file         *m = file->private_data;                \
1851         struct mdt_device       *mdt = m->private;                      \
1852         struct coordinator      *cdt = &mdt->mdt_coordinator;           \
1853         __s64                    val;                                   \
1854         int                      rc;                                    \
1855         ENTRY;                                                          \
1856                                                                         \
1857         rc = lprocfs_str_to_s64(buffer, count, &val);                   \
1858         if (rc)                                                         \
1859                 RETURN(rc);                                             \
1860         if (val > 0 && val < INT_MAX) {                                 \
1861                 cdt->VAR = val;                                         \
1862                 RETURN(count);                                          \
1863         }                                                               \
1864         RETURN(-EINVAL);                                                \
1865 }                                                                       \
1866
1867 GENERATE_PROC_METHOD(cdt_loop_period)
1868 GENERATE_PROC_METHOD(cdt_grace_delay)
1869 GENERATE_PROC_METHOD(cdt_active_req_timeout)
1870 GENERATE_PROC_METHOD(cdt_max_requests)
1871 GENERATE_PROC_METHOD(cdt_default_archive_id)
1872
1873 /*
1874  * procfs write method for MDT/hsm_control
1875  * proc entry is in mdt directory so data is mdt obd_device pointer
1876  */
1877 #define CDT_ENABLE_CMD   "enabled"
1878 #define CDT_STOP_CMD     "shutdown"
1879 #define CDT_DISABLE_CMD  "disabled"
1880 #define CDT_PURGE_CMD    "purge"
1881 #define CDT_HELP_CMD     "help"
1882 #define CDT_MAX_CMD_LEN  10
1883
1884 ssize_t
1885 mdt_hsm_cdt_control_seq_write(struct file *file, const char __user *buffer,
1886                               size_t count, loff_t *off)
1887 {
1888         struct seq_file         *m = file->private_data;
1889         struct obd_device       *obd = m->private;
1890         struct mdt_device       *mdt = mdt_dev(obd->obd_lu_dev);
1891         struct coordinator      *cdt = &(mdt->mdt_coordinator);
1892         int                      rc, usage = 0;
1893         char                     kernbuf[CDT_MAX_CMD_LEN];
1894         ENTRY;
1895
1896         if (count == 0 || count >= sizeof(kernbuf))
1897                 RETURN(-EINVAL);
1898
1899         if (copy_from_user(kernbuf, buffer, count))
1900                 RETURN(-EFAULT);
1901         kernbuf[count] = 0;
1902
1903         if (kernbuf[count - 1] == '\n')
1904                 kernbuf[count - 1] = 0;
1905
1906         rc = 0;
1907         if (strcmp(kernbuf, CDT_ENABLE_CMD) == 0) {
1908                 if (cdt->cdt_state == CDT_DISABLE) {
1909                         cdt->cdt_state = CDT_RUNNING;
1910                         mdt_hsm_cdt_wakeup(mdt);
1911                 } else {
1912                         rc = mdt_hsm_cdt_start(mdt);
1913                 }
1914         } else if (strcmp(kernbuf, CDT_STOP_CMD) == 0) {
1915                 if ((cdt->cdt_state == CDT_STOPPING) ||
1916                     (cdt->cdt_state == CDT_STOPPED)) {
1917                         CERROR("%s: Coordinator already stopped\n",
1918                                mdt_obd_name(mdt));
1919                         rc = -EALREADY;
1920                 } else {
1921                         cdt->cdt_state = CDT_STOPPING;
1922                 }
1923         } else if (strcmp(kernbuf, CDT_DISABLE_CMD) == 0) {
1924                 if ((cdt->cdt_state == CDT_STOPPING) ||
1925                     (cdt->cdt_state == CDT_STOPPED)) {
1926                         CERROR("%s: Coordinator is stopped\n",
1927                                mdt_obd_name(mdt));
1928                         rc = -EINVAL;
1929                 } else {
1930                         cdt->cdt_state = CDT_DISABLE;
1931                 }
1932         } else if (strcmp(kernbuf, CDT_PURGE_CMD) == 0) {
1933                 rc = hsm_cancel_all_actions(mdt);
1934         } else if (strcmp(kernbuf, CDT_HELP_CMD) == 0) {
1935                 usage = 1;
1936         } else {
1937                 usage = 1;
1938                 rc = -EINVAL;
1939         }
1940
1941         if (usage == 1)
1942                 CERROR("%s: Valid coordinator control commands are: "
1943                        "%s %s %s %s %s\n", mdt_obd_name(mdt),
1944                        CDT_ENABLE_CMD, CDT_STOP_CMD, CDT_DISABLE_CMD,
1945                        CDT_PURGE_CMD, CDT_HELP_CMD);
1946
1947         if (rc)
1948                 RETURN(rc);
1949
1950         RETURN(count);
1951 }
1952
1953 int mdt_hsm_cdt_control_seq_show(struct seq_file *m, void *data)
1954 {
1955         struct obd_device       *obd = m->private;
1956         struct coordinator      *cdt;
1957         ENTRY;
1958
1959         cdt = &(mdt_dev(obd->obd_lu_dev)->mdt_coordinator);
1960
1961         if (cdt->cdt_state == CDT_INIT)
1962                 seq_printf(m, "init\n");
1963         else if (cdt->cdt_state == CDT_RUNNING)
1964                 seq_printf(m, "enabled\n");
1965         else if (cdt->cdt_state == CDT_STOPPING)
1966                 seq_printf(m, "stopping\n");
1967         else if (cdt->cdt_state == CDT_STOPPED)
1968                 seq_printf(m, "stopped\n");
1969         else if (cdt->cdt_state == CDT_DISABLE)
1970                 seq_printf(m, "disabled\n");
1971         else
1972                 seq_printf(m, "unknown\n");
1973
1974         RETURN(0);
1975 }
1976
1977 static int
1978 mdt_hsm_request_mask_show(struct seq_file *m, __u64 mask)
1979 {
1980         bool first = true;
1981         int i;
1982         ENTRY;
1983
1984         for (i = 0; i < 8 * sizeof(mask); i++) {
1985                 if (mask & (1UL << i)) {
1986                         seq_printf(m, "%s%s", first ? "" : " ",
1987                                    hsm_copytool_action2name(i));
1988                         first = false;
1989                 }
1990         }
1991         seq_putc(m, '\n');
1992
1993         RETURN(0);
1994 }
1995
1996 static int
1997 mdt_hsm_user_request_mask_seq_show(struct seq_file *m, void *data)
1998 {
1999         struct mdt_device *mdt = m->private;
2000         struct coordinator *cdt = &mdt->mdt_coordinator;
2001
2002         return mdt_hsm_request_mask_show(m, cdt->cdt_user_request_mask);
2003 }
2004
2005 static int
2006 mdt_hsm_group_request_mask_seq_show(struct seq_file *m, void *data)
2007 {
2008         struct mdt_device *mdt = m->private;
2009         struct coordinator *cdt = &mdt->mdt_coordinator;
2010
2011         return mdt_hsm_request_mask_show(m, cdt->cdt_group_request_mask);
2012 }
2013
2014 static int
2015 mdt_hsm_other_request_mask_seq_show(struct seq_file *m, void *data)
2016 {
2017         struct mdt_device *mdt = m->private;
2018         struct coordinator *cdt = &mdt->mdt_coordinator;
2019
2020         return mdt_hsm_request_mask_show(m, cdt->cdt_other_request_mask);
2021 }
2022
2023 static inline enum hsm_copytool_action
2024 hsm_copytool_name2action(const char *name)
2025 {
2026         if (strcasecmp(name, "NOOP") == 0)
2027                 return HSMA_NONE;
2028         else if (strcasecmp(name, "ARCHIVE") == 0)
2029                 return HSMA_ARCHIVE;
2030         else if (strcasecmp(name, "RESTORE") == 0)
2031                 return HSMA_RESTORE;
2032         else if (strcasecmp(name, "REMOVE") == 0)
2033                 return HSMA_REMOVE;
2034         else if (strcasecmp(name, "CANCEL") == 0)
2035                 return HSMA_CANCEL;
2036         else
2037                 return -1;
2038 }
2039
2040 static ssize_t
2041 mdt_write_hsm_request_mask(struct file *file, const char __user *user_buf,
2042                             size_t user_count, __u64 *mask)
2043 {
2044         char *buf, *pos, *name;
2045         size_t buf_size;
2046         __u64 new_mask = 0;
2047         int rc;
2048         ENTRY;
2049
2050         if (!(user_count < 4096))
2051                 RETURN(-ENOMEM);
2052
2053         buf_size = user_count + 1;
2054
2055         OBD_ALLOC(buf, buf_size);
2056         if (buf == NULL)
2057                 RETURN(-ENOMEM);
2058
2059         if (copy_from_user(buf, user_buf, buf_size - 1))
2060                 GOTO(out, rc = -EFAULT);
2061
2062         buf[buf_size - 1] = '\0';
2063
2064         pos = buf;
2065         while ((name = strsep(&pos, " \t\v\n")) != NULL) {
2066                 int action;
2067
2068                 if (*name == '\0')
2069                         continue;
2070
2071                 action = hsm_copytool_name2action(name);
2072                 if (action < 0)
2073                         GOTO(out, rc = -EINVAL);
2074
2075                 new_mask |= (1UL << action);
2076         }
2077
2078         *mask = new_mask;
2079         rc = user_count;
2080 out:
2081         OBD_FREE(buf, buf_size);
2082
2083         RETURN(rc);
2084 }
2085
2086 static ssize_t
2087 mdt_hsm_user_request_mask_seq_write(struct file *file, const char __user *buf,
2088                                         size_t count, loff_t *off)
2089 {
2090         struct seq_file         *m = file->private_data;
2091         struct mdt_device       *mdt = m->private;
2092         struct coordinator *cdt = &mdt->mdt_coordinator;
2093
2094         return mdt_write_hsm_request_mask(file, buf, count,
2095                                            &cdt->cdt_user_request_mask);
2096 }
2097
2098 static ssize_t
2099 mdt_hsm_group_request_mask_seq_write(struct file *file, const char __user *buf,
2100                                         size_t count, loff_t *off)
2101 {
2102         struct seq_file         *m = file->private_data;
2103         struct mdt_device       *mdt = m->private;
2104         struct coordinator      *cdt = &mdt->mdt_coordinator;
2105
2106         return mdt_write_hsm_request_mask(file, buf, count,
2107                                            &cdt->cdt_group_request_mask);
2108 }
2109
2110 static ssize_t
2111 mdt_hsm_other_request_mask_seq_write(struct file *file, const char __user *buf,
2112                                         size_t count, loff_t *off)
2113 {
2114         struct seq_file         *m = file->private_data;
2115         struct mdt_device       *mdt = m->private;
2116         struct coordinator      *cdt = &mdt->mdt_coordinator;
2117
2118         return mdt_write_hsm_request_mask(file, buf, count,
2119                                            &cdt->cdt_other_request_mask);
2120 }
2121
2122 LPROC_SEQ_FOPS(mdt_hsm_cdt_loop_period);
2123 LPROC_SEQ_FOPS(mdt_hsm_cdt_grace_delay);
2124 LPROC_SEQ_FOPS(mdt_hsm_cdt_active_req_timeout);
2125 LPROC_SEQ_FOPS(mdt_hsm_cdt_max_requests);
2126 LPROC_SEQ_FOPS(mdt_hsm_cdt_default_archive_id);
2127 LPROC_SEQ_FOPS(mdt_hsm_user_request_mask);
2128 LPROC_SEQ_FOPS(mdt_hsm_group_request_mask);
2129 LPROC_SEQ_FOPS(mdt_hsm_other_request_mask);
2130
2131 static struct lprocfs_vars lprocfs_mdt_hsm_vars[] = {
2132         { .name =       "agents",
2133           .fops =       &mdt_hsm_agent_fops                     },
2134         { .name =       "actions",
2135           .fops =       &mdt_hsm_actions_fops,
2136           .proc_mode =  0444                                    },
2137         { .name =       "default_archive_id",
2138           .fops =       &mdt_hsm_cdt_default_archive_id_fops    },
2139         { .name =       "grace_delay",
2140           .fops =       &mdt_hsm_cdt_grace_delay_fops           },
2141         { .name =       "loop_period",
2142           .fops =       &mdt_hsm_cdt_loop_period_fops           },
2143         { .name =       "max_requests",
2144           .fops =       &mdt_hsm_cdt_max_requests_fops          },
2145         { .name =       "policy",
2146           .fops =       &mdt_hsm_policy_fops                    },
2147         { .name =       "active_request_timeout",
2148           .fops =       &mdt_hsm_cdt_active_req_timeout_fops    },
2149         { .name =       "active_requests",
2150           .fops =       &mdt_hsm_active_requests_fops           },
2151         { .name =       "user_request_mask",
2152           .fops =       &mdt_hsm_user_request_mask_fops,        },
2153         { .name =       "group_request_mask",
2154           .fops =       &mdt_hsm_group_request_mask_fops,       },
2155         { .name =       "other_request_mask",
2156           .fops =       &mdt_hsm_other_request_mask_fops,       },
2157         { 0 }
2158 };