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  * @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  */
  23 int 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 }
  81 EXPORT_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  */
  89 int 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 }
 129 EXPORT_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  */
 138 int 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 }
 182 EXPORT_SYMBOL(ceph_cls_break_lock);
 183
 184 int 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 }
 233 EXPORT_SYMBOL(ceph_cls_set_cookie);
 234
 235 void 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 }
 243 EXPORT_SYMBOL(ceph_free_lockers);
 244
 245 static 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
 282 static 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
 317 err_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  */
 328 int 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 }
 381 EXPORT_SYMBOL(ceph_cls_lock_info);
 382
 383 int 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 }
 425 EXPORT_SYMBOL(ceph_cls_assert_locked);