net/ceph/cls_lock_client.c

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