LXR linux/net/ceph/cls_lock

   1// SPDX-License-Identifier: GPL-2.0
   2#include <linux/ceph/ceph_debug.h>
   3
   4#include <linux/types.h>
   5#include <linux/slab.h>
   6
   7#include <linux/ceph/cls_lock_client.h>
   8#include <linux/ceph/decode.h>
   9#include <linux/ceph/libceph.h>
  10
  11/**
  12 * ceph_cls_lock - grab rados lock for object
  13 * @oid, @oloc: object to lock
  14 * @lock_name: the name of the lock
  15 * @type: lock type (CEPH_CLS_LOCK_EXCLUSIVE or CEPH_CLS_LOCK_SHARED)
  16 * @cookie: user-defined identifier for this instance of the lock
  17 * @tag: user-defined tag
  18 * @desc: user-defined lock description
  19 * @flags: lock flags
  20 *
  21 * All operations on the same lock should use the same tag.
  22 */
  23int ceph_cls_lock(struct ceph_osd_client *osdc,
  24                  struct ceph_object_id *oid,
  25                  struct ceph_object_locator *oloc,
  26                  char *lock_name, u8 type, char *cookie,
  27                  char *tag, char *desc, u8 flags)
  28{
  29        int lock_op_buf_size;
  30        int name_len = strlen(lock_name);
  31        int cookie_len = strlen(cookie);
  32        int tag_len = strlen(tag);
  33        int desc_len = strlen(desc);
  34        void *p, *end;
  35        struct page *lock_op_page;
  36        struct timespec64 mtime;
  37        int ret;
  38
  39        lock_op_buf_size = name_len + sizeof(__le32) +
  40                           cookie_len + sizeof(__le32) +
  41                           tag_len + sizeof(__le32) +
  42                           desc_len + sizeof(__le32) +
  43                           sizeof(struct ceph_timespec) +
  44                           /* flag and type */
  45                           sizeof(u8) + sizeof(u8) +
  46                           CEPH_ENCODING_START_BLK_LEN;
  47        if (lock_op_buf_size > PAGE_SIZE)
  48                return -E2BIG;
  49
  50        lock_op_page = alloc_page(GFP_NOIO);
  51        if (!lock_op_page)
  52                return -ENOMEM;
  53
  54        p = page_address(lock_op_page);
  55        end = p + lock_op_buf_size;
  56
  57        /* encode cls_lock_lock_op struct */
  58        ceph_start_encoding(&p, 1, 1,
  59                            lock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
  60        ceph_encode_string(&p, end, lock_name, name_len);
  61        ceph_encode_8(&p, type);
  62        ceph_encode_string(&p, end, cookie, cookie_len);
  63        ceph_encode_string(&p, end, tag, tag_len);
  64        ceph_encode_string(&p, end, desc, desc_len);
  65        /* only support infinite duration */
  66        memset(&mtime, 0, sizeof(mtime));
  67        ceph_encode_timespec64(p, &mtime);
  68        p += sizeof(struct ceph_timespec);
  69        ceph_encode_8(&p, flags);
  70
  71        dout("%s lock_name %s type %d cookie %s tag %s desc %s flags 0x%x\n",
  72             __func__, lock_name, type, cookie, tag, desc, flags);
  73        ret = ceph_osdc_call(osdc, oid, oloc, "lock", "lock",
  74                             CEPH_OSD_FLAG_WRITE, lock_op_page,
  75                             lock_op_buf_size, NULL, NULL);
  76
  77        dout("%s: status %d\n", __func__, ret);
  78        __free_page(lock_op_page);
  79        return ret;
  80}
  81EXPORT_SYMBOL(ceph_cls_lock);
  82
  83/**
  84 * ceph_cls_unlock - release rados lock for object
  85 * @oid, @oloc: object to lock
  86 * @lock_name: the name of the lock
  87 * @cookie: user-defined identifier for this instance of the lock
  88 */
  89int ceph_cls_unlock(struct ceph_osd_client *osdc,
  90                    struct ceph_object_id *oid,
  91                    struct ceph_object_locator *oloc,
  92                    char *lock_name, char *cookie)
  93{
  94        int unlock_op_buf_size;
  95        int name_len = strlen(lock_name);
  96        int cookie_len = strlen(cookie);
  97        void *p, *end;
  98        struct page *unlock_op_page;
  99        int ret;
 100
 101        unlock_op_buf_size = name_len + sizeof(__le32) +
 102                             cookie_len + sizeof(__le32) +
 103                             CEPH_ENCODING_START_BLK_LEN;
 104        if (unlock_op_buf_size > PAGE_SIZE)
 105                return -E2BIG;
 106
 107        unlock_op_page = alloc_page(GFP_NOIO);
 108        if (!unlock_op_page)
 109                return -ENOMEM;
 110
 111        p = page_address(unlock_op_page);
 112        end = p + unlock_op_buf_size;
 113
 114        /* encode cls_lock_unlock_op struct */
 115        ceph_start_encoding(&p, 1, 1,
 116                            unlock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
 117        ceph_encode_string(&p, end, lock_name, name_len);
 118        ceph_encode_string(&p, end, cookie, cookie_len);
 119
 120        dout("%s lock_name %s cookie %s\n", __func__, lock_name, cookie);
 121        ret = ceph_osdc_call(osdc, oid, oloc, "lock", "unlock",
 122                             CEPH_OSD_FLAG_WRITE, unlock_op_page,
 123                             unlock_op_buf_size, NULL, NULL);
 124
 125        dout("%s: status %d\n", __func__, ret);
 126        __free_page(unlock_op_page);
 127        return ret;
 128}
 129EXPORT_SYMBOL(ceph_cls_unlock);
 130
 131/**
 132 * ceph_cls_break_lock - release rados lock for object for specified client
 133 * @oid, @oloc: object to lock
 134 * @lock_name: the name of the lock
 135 * @cookie: user-defined identifier for this instance of the lock
 136 * @locker: current lock owner
 137 */
 138int ceph_cls_break_lock(struct ceph_osd_client *osdc,
 139                        struct ceph_object_id *oid,
 140                        struct ceph_object_locator *oloc,
 141                        char *lock_name, char *cookie,
 142                        struct ceph_entity_name *locker)
 143{
 144        int break_op_buf_size;
 145        int name_len = strlen(lock_name);
 146        int cookie_len = strlen(cookie);
 147        struct page *break_op_page;
 148        void *p, *end;
 149        int ret;
 150
 151        break_op_buf_size = name_len + sizeof(__le32) +
 152                            cookie_len + sizeof(__le32) +
 153                            sizeof(u8) + sizeof(__le64) +
 154                            CEPH_ENCODING_START_BLK_LEN;
 155        if (break_op_buf_size > PAGE_SIZE)
 156                return -E2BIG;
 157
 158        break_op_page = alloc_page(GFP_NOIO);
 159        if (!break_op_page)
 160                return -ENOMEM;
 161
 162        p = page_address(break_op_page);
 163        end = p + break_op_buf_size;
 164
 165        /* encode cls_lock_break_op struct */
 166        ceph_start_encoding(&p, 1, 1,
 167                            break_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
 168        ceph_encode_string(&p, end, lock_name, name_len);
 169        ceph_encode_copy(&p, locker, sizeof(*locker));
 170        ceph_encode_string(&p, end, cookie, cookie_len);
 171
 172        dout("%s lock_name %s cookie %s locker %s%llu\n", __func__, lock_name,
 173             cookie, ENTITY_NAME(*locker));
 174        ret = ceph_osdc_call(osdc, oid, oloc, "lock", "break_lock",
 175                             CEPH_OSD_FLAG_WRITE, break_op_page,
 176                             break_op_buf_size, NULL, NULL);
 177
 178        dout("%s: status %d\n", __func__, ret);
 179        __free_page(break_op_page);
 180        return ret;
 181}
 182EXPORT_SYMBOL(ceph_cls_break_lock);
 183
 184int ceph_cls_set_cookie(struct ceph_osd_client *osdc,
 185                        struct ceph_object_id *oid,
 186                        struct ceph_object_locator *oloc,
 187                        char *lock_name, u8 type, char *old_cookie,
 188                        char *tag, char *new_cookie)
 189{
 190        int cookie_op_buf_size;
 191        int name_len = strlen(lock_name);
 192        int old_cookie_len = strlen(old_cookie);
 193        int tag_len = strlen(tag);
 194        int new_cookie_len = strlen(new_cookie);
 195        void *p, *end;
 196        struct page *cookie_op_page;
 197        int ret;
 198
 199        cookie_op_buf_size = name_len + sizeof(__le32) +
 200                             old_cookie_len + sizeof(__le32) +
 201                             tag_len + sizeof(__le32) +
 202                             new_cookie_len + sizeof(__le32) +
 203                             sizeof(u8) + CEPH_ENCODING_START_BLK_LEN;
 204        if (cookie_op_buf_size > PAGE_SIZE)
 205                return -E2BIG;
 206
 207        cookie_op_page = alloc_page(GFP_NOIO);
 208        if (!cookie_op_page)
 209                return -ENOMEM;
 210
 211        p = page_address(cookie_op_page);
 212        end = p + cookie_op_buf_size;
 213
 214        /* encode cls_lock_set_cookie_op struct */
 215        ceph_start_encoding(&p, 1, 1,
 216                            cookie_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
 217        ceph_encode_string(&p, end, lock_name, name_len);
 218        ceph_encode_8(&p, type);
 219        ceph_encode_string(&p, end, old_cookie, old_cookie_len);
 220        ceph_encode_string(&p, end, tag, tag_len);
 221        ceph_encode_string(&p, end, new_cookie, new_cookie_len);
 222
 223        dout("%s lock_name %s type %d old_cookie %s tag %s new_cookie %s\n",
 224             __func__, lock_name, type, old_cookie, tag, new_cookie);
 225        ret = ceph_osdc_call(osdc, oid, oloc, "lock", "set_cookie",
 226                             CEPH_OSD_FLAG_WRITE, cookie_op_page,
 227                             cookie_op_buf_size, NULL, NULL);
 228
 229        dout("%s: status %d\n", __func__, ret);
 230        __free_page(cookie_op_page);
 231        return ret;
 232}
 233EXPORT_SYMBOL(ceph_cls_set_cookie);
 234
 235void ceph_free_lockers(struct ceph_locker *lockers, u32 num_lockers)
 236{
 237        int i;
 238
 239        for (i = 0; i < num_lockers; i++)
 240                kfree(lockers[i].id.cookie);
 241        kfree(lockers);
 242}
 243EXPORT_SYMBOL(ceph_free_lockers);
 244
 245static int decode_locker(void **p, void *end, struct ceph_locker *locker)
 246{
 247        u8 struct_v;
 248        u32 len;
 249        char *s;
 250        int ret;
 251
 252        ret = ceph_start_decoding(p, end, 1, "locker_id_t", &struct_v, &len);
 253        if (ret)
 254                return ret;
 255
 256        ceph_decode_copy(p, &locker->id.name, sizeof(locker->id.name));
 257        s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
 258        if (IS_ERR(s))
 259                return PTR_ERR(s);
 260
 261        locker->id.cookie = s;
 262
 263        ret = ceph_start_decoding(p, end, 1, "locker_info_t", &struct_v, &len);
 264        if (ret)
 265                return ret;
 266
 267        *p += sizeof(struct ceph_timespec); /* skip expiration */
 268
 269        ret = ceph_decode_entity_addr(p, end, &locker->info.addr);
 270        if (ret)
 271                return ret;
 272
 273        len = ceph_decode_32(p);
 274        *p += len; /* skip description */
 275
 276        dout("%s %s%llu cookie %s addr %s\n", __func__,
 277             ENTITY_NAME(locker->id.name), locker->id.cookie,
 278             ceph_pr_addr(&locker->info.addr));
 279        return 0;
 280}
 281
 282static int decode_lockers(void **p, void *end, u8 *type, char **tag,
 283                          struct ceph_locker **lockers, u32 *num_lockers)
 284{
 285        u8 struct_v;
 286        u32 struct_len;
 287        char *s;
 288        int i;
 289        int ret;
 290
 291        ret = ceph_start_decoding(p, end, 1, "cls_lock_get_info_reply",
 292                                  &struct_v, &struct_len);
 293        if (ret)
 294                return ret;
 295
 296        *num_lockers = ceph_decode_32(p);
 297        *lockers = kcalloc(*num_lockers, sizeof(**lockers), GFP_NOIO);
 298        if (!*lockers)
 299                return -ENOMEM;
 300
 301        for (i = 0; i < *num_lockers; i++) {
 302                ret = decode_locker(p, end, *lockers + i);
 303                if (ret)
 304                        goto err_free_lockers;
 305        }
 306
 307        *type = ceph_decode_8(p);
 308        s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
 309        if (IS_ERR(s)) {
 310                ret = PTR_ERR(s);
 311                goto err_free_lockers;
 312        }
 313
 314        *tag = s;
 315        return 0;
 316
 317err_free_lockers:
 318        ceph_free_lockers(*lockers, *num_lockers);
 319        return ret;
 320}
 321
 322/*
 323 * On success, the caller is responsible for:
 324 *
 325 *     kfree(tag);
 326 *     ceph_free_lockers(lockers, num_lockers);
 327 */
 328int ceph_cls_lock_info(struct ceph_osd_client *osdc,
 329                       struct ceph_object_id *oid,
 330                       struct ceph_object_locator *oloc,
 331                       char *lock_name, u8 *type, char **tag,
 332                       struct ceph_locker **lockers, u32 *num_lockers)
 333{
 334        int get_info_op_buf_size;
 335        int name_len = strlen(lock_name);
 336        struct page *get_info_op_page, *reply_page;
 337        size_t reply_len = PAGE_SIZE;
 338        void *p, *end;
 339        int ret;
 340
 341        get_info_op_buf_size = name_len + sizeof(__le32) +
 342                               CEPH_ENCODING_START_BLK_LEN;
 343        if (get_info_op_buf_size > PAGE_SIZE)
 344                return -E2BIG;
 345
 346        get_info_op_page = alloc_page(GFP_NOIO);
 347        if (!get_info_op_page)
 348                return -ENOMEM;
 349
 350        reply_page = alloc_page(GFP_NOIO);
 351        if (!reply_page) {
 352                __free_page(get_info_op_page);
 353                return -ENOMEM;
 354        }
 355
 356        p = page_address(get_info_op_page);
 357        end = p + get_info_op_buf_size;
 358
 359        /* encode cls_lock_get_info_op struct */
 360        ceph_start_encoding(&p, 1, 1,
 361                            get_info_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
 362        ceph_encode_string(&p, end, lock_name, name_len);
 363
 364        dout("%s lock_name %s\n", __func__, lock_name);
 365        ret = ceph_osdc_call(osdc, oid, oloc, "lock", "get_info",
 366                             CEPH_OSD_FLAG_READ, get_info_op_page,
 367                             get_info_op_buf_size, &reply_page, &reply_len);
 368
 369        dout("%s: status %d\n", __func__, ret);
 370        if (ret >= 0) {
 371                p = page_address(reply_page);
 372                end = p + reply_len;
 373
 374                ret = decode_lockers(&p, end, type, tag, lockers, num_lockers);
 375        }
 376
 377        __free_page(get_info_op_page);
 378        __free_page(reply_page);
 379        return ret;
 380}
 381EXPORT_SYMBOL(ceph_cls_lock_info);
 382
 383int ceph_cls_assert_locked(struct ceph_osd_request *req, int which,
 384                           char *lock_name, u8 type, char *cookie, char *tag)
 385{
 386        int assert_op_buf_size;
 387        int name_len = strlen(lock_name);
 388        int cookie_len = strlen(cookie);
 389        int tag_len = strlen(tag);
 390        struct page **pages;
 391        void *p, *end;
 392        int ret;
 393
 394        assert_op_buf_size = name_len + sizeof(__le32) +
 395                             cookie_len + sizeof(__le32) +
 396                             tag_len + sizeof(__le32) +
 397                             sizeof(u8) + CEPH_ENCODING_START_BLK_LEN;
 398        if (assert_op_buf_size > PAGE_SIZE)
 399                return -E2BIG;
 400
 401        ret = osd_req_op_cls_init(req, which, "lock", "assert_locked");
 402        if (ret)
 403                return ret;
 404
 405        pages = ceph_alloc_page_vector(1, GFP_NOIO);
 406        if (IS_ERR(pages))
 407                return PTR_ERR(pages);
 408
 409        p = page_address(pages[0]);
 410        end = p + assert_op_buf_size;
 411
 412        /* encode cls_lock_assert_op struct */
 413        ceph_start_encoding(&p, 1, 1,
 414                            assert_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
 415        ceph_encode_string(&p, end, lock_name, name_len);
 416        ceph_encode_8(&p, type);
 417        ceph_encode_string(&p, end, cookie, cookie_len);
 418        ceph_encode_string(&p, end, tag, tag_len);
 419        WARN_ON(p != end);
 420
 421        osd_req_op_cls_request_data_pages(req, which, pages, assert_op_buf_size,
 422                                          0, false, true);
 423        return 0;
 424}
 425EXPORT_SYMBOL(ceph_cls_assert_locked);
 426