lustre/ptlrpc/llog_server.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) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2015, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/ptlrpc/llog_server.c
  32  *
  33  * remote api for llog - server side
  34  *
  35  * Author: Andreas Dilger <adilger@clusterfs.com>
  36  */
  37
  38 #define DEBUG_SUBSYSTEM S_LOG
  39
  40 #include <obd_class.h>
  41 #include <lu_target.h>
  42 #include <lustre_log.h>
  43 #include <lustre_net.h>
  44
  45 static int llog_origin_close(const struct lu_env *env, struct llog_handle *lgh)
  46 {
  47         if (lgh->lgh_hdr != NULL && lgh->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
  48                 return llog_cat_close(env, lgh);
  49         else
  50                 return llog_close(env, lgh);
  51 }
  52
  53 /* Only open is supported, no new llog can be created remotely */
  54 int llog_origin_handle_open(struct ptlrpc_request *req)
  55 {
  56         struct obd_export       *exp = req->rq_export;
  57         struct obd_device       *obd = exp->exp_obd;
  58         struct llog_handle      *loghandle;
  59         struct llogd_body       *body;
  60         struct llog_logid       *logid = NULL;
  61         struct llog_ctxt        *ctxt;
  62         char                    *name = NULL;
  63         int                      rc;
  64
  65         ENTRY;
  66
  67         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
  68         if (body == NULL)
  69                 RETURN(err_serious(-EFAULT));
  70
  71         rc = req_capsule_server_pack(&req->rq_pill);
  72         if (rc)
  73                 RETURN(err_serious(-ENOMEM));
  74
  75         if (ostid_id(&body->lgd_logid.lgl_oi) > 0)
  76                 logid = &body->lgd_logid;
  77
  78         if (req_capsule_field_present(&req->rq_pill, &RMF_NAME, RCL_CLIENT)) {
  79                 name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
  80                 if (name == NULL)
  81                         RETURN(-EFAULT);
  82                 CDEBUG(D_INFO, "%s: opening log %s\n", obd->obd_name, name);
  83         }
  84
  85         if (body->lgd_ctxt_idx >= LLOG_MAX_CTXTS) {
  86                 CDEBUG(D_WARNING, "%s: bad ctxt ID: idx=%d name=%s\n",
  87                        obd->obd_name, body->lgd_ctxt_idx, name);
  88                 RETURN(-EPROTO);
  89         }
  90
  91         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
  92         if (ctxt == NULL) {
  93                 CDEBUG(D_WARNING, "%s: no ctxt. group=%p idx=%d name=%s\n",
  94                        obd->obd_name, &obd->obd_olg, body->lgd_ctxt_idx, name);
  95                 RETURN(-ENODEV);
  96         }
  97
  98         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle, logid,
  99                        name, LLOG_OPEN_EXISTS);
 100         if (rc)
 101                 GOTO(out_ctxt, rc);
 102
 103         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 104         body->lgd_logid = loghandle->lgh_id;
 105
 106         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 107         EXIT;
 108 out_ctxt:
 109         llog_ctxt_put(ctxt);
 110         return rc;
 111 }
 112
 113 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 114 {
 115         struct llog_handle      *loghandle;
 116         struct llogd_body       *body;
 117         struct llogd_body       *repbody;
 118         struct llog_ctxt        *ctxt;
 119         __u32                    flags;
 120         void                    *ptr;
 121         int                      rc;
 122
 123         ENTRY;
 124
 125         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 126         if (body == NULL)
 127                 RETURN(err_serious(-EFAULT));
 128
 129         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 130                              LLOG_MIN_CHUNK_SIZE);
 131         rc = req_capsule_server_pack(&req->rq_pill);
 132         if (rc)
 133                 RETURN(err_serious(-ENOMEM));
 134
 135         if (body->lgd_ctxt_idx >= LLOG_MAX_CTXTS) {
 136                 CDEBUG(D_WARNING, "%s: bad ctxt ID: idx=%d\n",
 137                        req->rq_export->exp_obd->obd_name, body->lgd_ctxt_idx);
 138                 RETURN(-EPROTO);
 139         }
 140
 141         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 142         if (ctxt == NULL)
 143                 RETURN(-ENODEV);
 144         if (OBD_FAIL_PRECHECK(OBD_FAIL_MDS_LLOG_UMOUNT_RACE))
 145                 cfs_fail_val = 1;
 146
 147         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle,
 148                        &body->lgd_logid, NULL, LLOG_OPEN_EXISTS);
 149         if (rc)
 150                 GOTO(out_ctxt, rc);
 151
 152         flags = body->lgd_llh_flags;
 153         rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
 154                               NULL);
 155         if (rc)
 156                 GOTO(out_close, rc);
 157
 158         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 159         *repbody = *body;
 160
 161         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 162         rc = llog_next_block(req->rq_svc_thread->t_env, loghandle,
 163                              &repbody->lgd_saved_index, repbody->lgd_index,
 164                              &repbody->lgd_cur_offset, ptr,
 165                              LLOG_MIN_CHUNK_SIZE);
 166         if (rc)
 167                 GOTO(out_close, rc);
 168         EXIT;
 169 out_close:
 170         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 171 out_ctxt:
 172         llog_ctxt_put(ctxt);
 173         return rc;
 174 }
 175
 176 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 177 {
 178         struct llog_handle      *loghandle;
 179         struct llogd_body       *body;
 180         struct llogd_body       *repbody;
 181         struct llog_ctxt        *ctxt;
 182         __u32                    flags;
 183         void                    *ptr;
 184         int                      rc;
 185
 186         ENTRY;
 187
 188         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 189         if (body == NULL)
 190                 RETURN(err_serious(-EFAULT));
 191
 192         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 193                              LLOG_MIN_CHUNK_SIZE);
 194         rc = req_capsule_server_pack(&req->rq_pill);
 195         if (rc)
 196                 RETURN(err_serious(-ENOMEM));
 197
 198         if (body->lgd_ctxt_idx >= LLOG_MAX_CTXTS) {
 199                 CDEBUG(D_WARNING, "%s: bad ctxt ID: idx=%d\n",
 200                        req->rq_export->exp_obd->obd_name, body->lgd_ctxt_idx);
 201                 RETURN(-EPROTO);
 202         }
 203
 204         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 205         if (ctxt == NULL)
 206                 RETURN(-ENODEV);
 207
 208         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle,
 209                          &body->lgd_logid, NULL, LLOG_OPEN_EXISTS);
 210         if (rc)
 211                 GOTO(out_ctxt, rc);
 212
 213         flags = body->lgd_llh_flags;
 214         rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
 215                               NULL);
 216         if (rc)
 217                 GOTO(out_close, rc);
 218
 219         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 220         *repbody = *body;
 221
 222         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 223         rc = llog_prev_block(req->rq_svc_thread->t_env, loghandle,
 224                              body->lgd_index, ptr, LLOG_MIN_CHUNK_SIZE);
 225         if (rc)
 226                 GOTO(out_close, rc);
 227
 228         EXIT;
 229 out_close:
 230         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 231 out_ctxt:
 232         llog_ctxt_put(ctxt);
 233         return rc;
 234 }
 235
 236 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 237 {
 238         struct llog_handle      *loghandle;
 239         struct llogd_body       *body;
 240         struct llog_log_hdr     *hdr;
 241         struct llog_ctxt        *ctxt;
 242         __u32                    flags;
 243         int                      rc;
 244
 245         ENTRY;
 246
 247         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 248         if (body == NULL)
 249                 RETURN(err_serious(-EFAULT));
 250
 251         rc = req_capsule_server_pack(&req->rq_pill);
 252         if (rc)
 253                 RETURN(err_serious(-ENOMEM));
 254
 255         if (body->lgd_ctxt_idx >= LLOG_MAX_CTXTS) {
 256                 CDEBUG(D_WARNING, "%s: bad ctxt ID: idx=%d\n",
 257                        req->rq_export->exp_obd->obd_name, body->lgd_ctxt_idx);
 258                 RETURN(-EPROTO);
 259         }
 260
 261         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 262         if (ctxt == NULL)
 263                 RETURN(-ENODEV);
 264
 265         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle,
 266                        &body->lgd_logid, NULL, LLOG_OPEN_EXISTS);
 267         if (rc)
 268                 GOTO(out_ctxt, rc);
 269
 270         /*
 271          * llog_init_handle() reads the llog header
 272          */
 273         flags = body->lgd_llh_flags;
 274         rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
 275                               NULL);
 276         if (rc)
 277                 GOTO(out_close, rc);
 278         flags = loghandle->lgh_hdr->llh_flags;
 279
 280         hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
 281         *hdr = *loghandle->lgh_hdr;
 282         EXIT;
 283 out_close:
 284         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 285 out_ctxt:
 286         llog_ctxt_put(ctxt);
 287         return rc;
 288 }