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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/ptlrpc/llog_server.c
  37  *
  38  * remote api for llog - server side
  39  *
  40  * Author: Andreas Dilger <adilger@clusterfs.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_LOG
  44
  45 #ifndef __KERNEL__
  46 #include <liblustre.h>
  47 #endif
  48
  49 #include <obd_class.h>
  50 #include <lu_target.h>
  51 #include <lustre_log.h>
  52 #include <lustre_net.h>
  53
  54 #if defined(__KERNEL__) && defined(LUSTRE_LOG_SERVER)
  55 static int llog_origin_close(const struct lu_env *env, struct llog_handle *lgh)
  56 {
  57         if (lgh->lgh_hdr != NULL && lgh->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
  58                 return llog_cat_close(env, lgh);
  59         else
  60                 return llog_close(env, lgh);
  61 }
  62
  63 /* Only open is supported, no new llog can be created remotely */
  64 int llog_origin_handle_open(struct ptlrpc_request *req)
  65 {
  66         struct obd_export       *exp = req->rq_export;
  67         struct obd_device       *obd = exp->exp_obd;
  68         struct llog_handle      *loghandle;
  69         struct llogd_body       *body;
  70         struct llog_logid       *logid = NULL;
  71         struct llog_ctxt        *ctxt;
  72         char                    *name = NULL;
  73         int                      rc;
  74
  75         ENTRY;
  76
  77         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
  78         if (body == NULL)
  79                 RETURN(err_serious(-EFAULT));
  80
  81         rc = req_capsule_server_pack(&req->rq_pill);
  82         if (rc)
  83                 RETURN(err_serious(-ENOMEM));
  84
  85         if (ostid_id(&body->lgd_logid.lgl_oi) > 0)
  86                 logid = &body->lgd_logid;
  87
  88         if (req_capsule_field_present(&req->rq_pill, &RMF_NAME, RCL_CLIENT)) {
  89                 name = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
  90                 if (name == NULL)
  91                         RETURN(-EFAULT);
  92                 CDEBUG(D_INFO, "%s: opening log %s\n", obd->obd_name, name);
  93         }
  94
  95         ctxt = llog_get_context(obd, body->lgd_ctxt_idx);
  96         if (ctxt == NULL) {
  97                 CDEBUG(D_WARNING, "%s: no ctxt. group=%p idx=%d name=%s\n",
  98                        obd->obd_name, &obd->obd_olg, body->lgd_ctxt_idx, name);
  99                 RETURN(-ENODEV);
 100         }
 101
 102         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle, logid,
 103                        name, LLOG_OPEN_EXISTS);
 104         if (rc)
 105                 GOTO(out_ctxt, rc);
 106
 107         body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 108         body->lgd_logid = loghandle->lgh_id;
 109
 110         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 111         EXIT;
 112 out_ctxt:
 113         llog_ctxt_put(ctxt);
 114         return rc;
 115 }
 116 EXPORT_SYMBOL(llog_origin_handle_open);
 117
 118 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 119 {
 120         struct llogd_body       *body;
 121         struct llog_logid       *logid = NULL;
 122         struct llog_ctxt        *ctxt;
 123         int                      rc;
 124
 125         ENTRY;
 126
 127         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 128         if (body == NULL)
 129                 RETURN(err_serious(-EFAULT));
 130
 131         rc = req_capsule_server_pack(&req->rq_pill);
 132         if (rc < 0)
 133                 RETURN(err_serious(-ENOMEM));
 134
 135         if (ostid_id(&body->lgd_logid.lgl_oi) > 0)
 136                 logid = &body->lgd_logid;
 137
 138         if (!(body->lgd_llh_flags & LLOG_F_IS_PLAIN))
 139                 CERROR("%s: wrong llog flags %x\n",
 140                        req->rq_export->exp_obd->obd_name, body->lgd_llh_flags);
 141
 142         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 143         if (ctxt == NULL)
 144                 RETURN(-ENODEV);
 145
 146         rc = llog_erase(req->rq_svc_thread->t_env, ctxt, logid, NULL);
 147         llog_ctxt_put(ctxt);
 148         RETURN(rc);
 149 }
 150 EXPORT_SYMBOL(llog_origin_handle_destroy);
 151
 152 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 153 {
 154         struct llog_handle      *loghandle;
 155         struct llogd_body       *body;
 156         struct llogd_body       *repbody;
 157         struct llog_ctxt        *ctxt;
 158         __u32                    flags;
 159         void                    *ptr;
 160         int                      rc;
 161
 162         ENTRY;
 163
 164         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 165         if (body == NULL)
 166                 RETURN(err_serious(-EFAULT));
 167
 168         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 169                              LLOG_CHUNK_SIZE);
 170         rc = req_capsule_server_pack(&req->rq_pill);
 171         if (rc)
 172                 RETURN(err_serious(-ENOMEM));
 173
 174         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 175         if (ctxt == NULL)
 176                 RETURN(-ENODEV);
 177
 178         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle,
 179                        &body->lgd_logid, NULL, LLOG_OPEN_EXISTS);
 180         if (rc)
 181                 GOTO(out_ctxt, rc);
 182
 183         flags = body->lgd_llh_flags;
 184         rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
 185                               NULL);
 186         if (rc)
 187                 GOTO(out_close, rc);
 188
 189         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 190         *repbody = *body;
 191
 192         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 193         rc = llog_next_block(req->rq_svc_thread->t_env, loghandle,
 194                              &repbody->lgd_saved_index, repbody->lgd_index,
 195                              &repbody->lgd_cur_offset, ptr, LLOG_CHUNK_SIZE);
 196         if (rc)
 197                 GOTO(out_close, rc);
 198         EXIT;
 199 out_close:
 200         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 201 out_ctxt:
 202         llog_ctxt_put(ctxt);
 203         return rc;
 204 }
 205 EXPORT_SYMBOL(llog_origin_handle_next_block);
 206
 207 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 208 {
 209         struct llog_handle      *loghandle;
 210         struct llogd_body       *body;
 211         struct llogd_body       *repbody;
 212         struct llog_ctxt        *ctxt;
 213         __u32                    flags;
 214         void                    *ptr;
 215         int                      rc;
 216
 217         ENTRY;
 218
 219         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 220         if (body == NULL)
 221                 RETURN(err_serious(-EFAULT));
 222
 223         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
 224                              LLOG_CHUNK_SIZE);
 225         rc = req_capsule_server_pack(&req->rq_pill);
 226         if (rc)
 227                 RETURN(err_serious(-ENOMEM));
 228
 229         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 230         if (ctxt == NULL)
 231                 RETURN(-ENODEV);
 232
 233         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle,
 234                          &body->lgd_logid, NULL, LLOG_OPEN_EXISTS);
 235         if (rc)
 236                 GOTO(out_ctxt, rc);
 237
 238         flags = body->lgd_llh_flags;
 239         rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
 240                               NULL);
 241         if (rc)
 242                 GOTO(out_close, rc);
 243
 244         repbody = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);
 245         *repbody = *body;
 246
 247         ptr = req_capsule_server_get(&req->rq_pill, &RMF_EADATA);
 248         rc = llog_prev_block(req->rq_svc_thread->t_env, loghandle,
 249                              body->lgd_index, ptr, LLOG_CHUNK_SIZE);
 250         if (rc)
 251                 GOTO(out_close, rc);
 252
 253         EXIT;
 254 out_close:
 255         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 256 out_ctxt:
 257         llog_ctxt_put(ctxt);
 258         return rc;
 259 }
 260 EXPORT_SYMBOL(llog_origin_handle_prev_block);
 261
 262 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 263 {
 264         struct llog_handle      *loghandle;
 265         struct llogd_body       *body;
 266         struct llog_log_hdr     *hdr;
 267         struct llog_ctxt        *ctxt;
 268         __u32                    flags;
 269         int                      rc;
 270
 271         ENTRY;
 272
 273         body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);
 274         if (body == NULL)
 275                 RETURN(err_serious(-EFAULT));
 276
 277         rc = req_capsule_server_pack(&req->rq_pill);
 278         if (rc)
 279                 RETURN(err_serious(-ENOMEM));
 280
 281         ctxt = llog_get_context(req->rq_export->exp_obd, body->lgd_ctxt_idx);
 282         if (ctxt == NULL)
 283                 RETURN(-ENODEV);
 284
 285         rc = llog_open(req->rq_svc_thread->t_env, ctxt, &loghandle,
 286                        &body->lgd_logid, NULL, LLOG_OPEN_EXISTS);
 287         if (rc)
 288                 GOTO(out_ctxt, rc);
 289
 290         /*
 291          * llog_init_handle() reads the llog header
 292          */
 293         flags = body->lgd_llh_flags;
 294         rc = llog_init_handle(req->rq_svc_thread->t_env, loghandle, flags,
 295                               NULL);
 296         if (rc)
 297                 GOTO(out_close, rc);
 298         flags = loghandle->lgh_hdr->llh_flags;
 299
 300         hdr = req_capsule_server_get(&req->rq_pill, &RMF_LLOG_LOG_HDR);
 301         *hdr = *loghandle->lgh_hdr;
 302         EXIT;
 303 out_close:
 304         llog_origin_close(req->rq_svc_thread->t_env, loghandle);
 305 out_ctxt:
 306         llog_ctxt_put(ctxt);
 307         return rc;
 308 }
 309 EXPORT_SYMBOL(llog_origin_handle_read_header);
 310
 311 int llog_origin_handle_close(struct ptlrpc_request *req)
 312 {
 313         int      rc;
 314
 315         ENTRY;
 316
 317         rc = req_capsule_server_pack(&req->rq_pill);
 318         if (rc)
 319                 RETURN(err_serious(-ENOMEM));
 320         RETURN(0);
 321 }
 322 EXPORT_SYMBOL(llog_origin_handle_close);
 323
 324 #else /* !__KERNEL__ */
 325 int llog_origin_handle_open(struct ptlrpc_request *req)
 326 {
 327         LBUG();
 328         return 0;
 329 }
 330
 331 int llog_origin_handle_destroy(struct ptlrpc_request *req)
 332 {
 333         LBUG();
 334         return 0;
 335 }
 336
 337 int llog_origin_handle_next_block(struct ptlrpc_request *req)
 338 {
 339         LBUG();
 340         return 0;
 341 }
 342 int llog_origin_handle_prev_block(struct ptlrpc_request *req)
 343 {
 344         LBUG();
 345         return 0;
 346 }
 347 int llog_origin_handle_read_header(struct ptlrpc_request *req)
 348 {
 349         LBUG();
 350         return 0;
 351 }
 352 int llog_origin_handle_close(struct ptlrpc_request *req)
 353 {
 354         LBUG();
 355         return 0;
 356 }
 357 #endif