net/ceph/cls_lock_client.c

0001 // SPDX-License-Identifier: GPL-2.0
0002 #include <linux/ceph/ceph_debug.h>
0003
0004 #include <linux/types.h>
0005 #include <linux/slab.h>
0006
0007 #include <linux/ceph/cls_lock_client.h>
0008 #include <linux/ceph/decode.h>
0009 #include <linux/ceph/libceph.h>
0010
0011 /**
0012  * ceph_cls_lock - grab rados lock for object
0013  * @osdc: OSD client instance
0014  * @oid: object to lock
0015  * @oloc: object to lock
0016  * @lock_name: the name of the lock
0017  * @type: lock type (CEPH_CLS_LOCK_EXCLUSIVE or CEPH_CLS_LOCK_SHARED)
0018  * @cookie: user-defined identifier for this instance of the lock
0019  * @tag: user-defined tag
0020  * @desc: user-defined lock description
0021  * @flags: lock flags
0022  *
0023  * All operations on the same lock should use the same tag.
0024  */
0025 int ceph_cls_lock(struct ceph_osd_client *osdc,
0026           struct ceph_object_id *oid,
0027           struct ceph_object_locator *oloc,
0028           char *lock_name, u8 type, char *cookie,
0029           char *tag, char *desc, u8 flags)
0030 {
0031     int lock_op_buf_size;
0032     int name_len = strlen(lock_name);
0033     int cookie_len = strlen(cookie);
0034     int tag_len = strlen(tag);
0035     int desc_len = strlen(desc);
0036     void *p, *end;
0037     struct page *lock_op_page;
0038     struct timespec64 mtime;
0039     int ret;
0040
0041     lock_op_buf_size = name_len + sizeof(__le32) +
0042                cookie_len + sizeof(__le32) +
0043                tag_len + sizeof(__le32) +
0044                desc_len + sizeof(__le32) +
0045                sizeof(struct ceph_timespec) +
0046                /* flag and type */
0047                sizeof(u8) + sizeof(u8) +
0048                CEPH_ENCODING_START_BLK_LEN;
0049     if (lock_op_buf_size > PAGE_SIZE)
0050         return -E2BIG;
0051
0052     lock_op_page = alloc_page(GFP_NOIO);
0053     if (!lock_op_page)
0054         return -ENOMEM;
0055
0056     p = page_address(lock_op_page);
0057     end = p + lock_op_buf_size;
0058
0059     /* encode cls_lock_lock_op struct */
0060     ceph_start_encoding(&p, 1, 1,
0061                 lock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
0062     ceph_encode_string(&p, end, lock_name, name_len);
0063     ceph_encode_8(&p, type);
0064     ceph_encode_string(&p, end, cookie, cookie_len);
0065     ceph_encode_string(&p, end, tag, tag_len);
0066     ceph_encode_string(&p, end, desc, desc_len);
0067     /* only support infinite duration */
0068     memset(&mtime, 0, sizeof(mtime));
0069     ceph_encode_timespec64(p, &mtime);
0070     p += sizeof(struct ceph_timespec);
0071     ceph_encode_8(&p, flags);
0072
0073     dout("%s lock_name %s type %d cookie %s tag %s desc %s flags 0x%x\n",
0074          __func__, lock_name, type, cookie, tag, desc, flags);
0075     ret = ceph_osdc_call(osdc, oid, oloc, "lock", "lock",
0076                  CEPH_OSD_FLAG_WRITE, lock_op_page,
0077                  lock_op_buf_size, NULL, NULL);
0078
0079     dout("%s: status %d\n", __func__, ret);
0080     __free_page(lock_op_page);
0081     return ret;
0082 }
0083 EXPORT_SYMBOL(ceph_cls_lock);
0084
0085 /**
0086  * ceph_cls_unlock - release rados lock for object
0087  * @osdc: OSD client instance
0088  * @oid: object to lock
0089  * @oloc: object to lock
0090  * @lock_name: the name of the lock
0091  * @cookie: user-defined identifier for this instance of the lock
0092  */
0093 int ceph_cls_unlock(struct ceph_osd_client *osdc,
0094             struct ceph_object_id *oid,
0095             struct ceph_object_locator *oloc,
0096             char *lock_name, char *cookie)
0097 {
0098     int unlock_op_buf_size;
0099     int name_len = strlen(lock_name);
0100     int cookie_len = strlen(cookie);
0101     void *p, *end;
0102     struct page *unlock_op_page;
0103     int ret;
0104
0105     unlock_op_buf_size = name_len + sizeof(__le32) +
0106                  cookie_len + sizeof(__le32) +
0107                  CEPH_ENCODING_START_BLK_LEN;
0108     if (unlock_op_buf_size > PAGE_SIZE)
0109         return -E2BIG;
0110
0111     unlock_op_page = alloc_page(GFP_NOIO);
0112     if (!unlock_op_page)
0113         return -ENOMEM;
0114
0115     p = page_address(unlock_op_page);
0116     end = p + unlock_op_buf_size;
0117
0118     /* encode cls_lock_unlock_op struct */
0119     ceph_start_encoding(&p, 1, 1,
0120                 unlock_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
0121     ceph_encode_string(&p, end, lock_name, name_len);
0122     ceph_encode_string(&p, end, cookie, cookie_len);
0123
0124     dout("%s lock_name %s cookie %s\n", __func__, lock_name, cookie);
0125     ret = ceph_osdc_call(osdc, oid, oloc, "lock", "unlock",
0126                  CEPH_OSD_FLAG_WRITE, unlock_op_page,
0127                  unlock_op_buf_size, NULL, NULL);
0128
0129     dout("%s: status %d\n", __func__, ret);
0130     __free_page(unlock_op_page);
0131     return ret;
0132 }
0133 EXPORT_SYMBOL(ceph_cls_unlock);
0134
0135 /**
0136  * ceph_cls_break_lock - release rados lock for object for specified client
0137  * @osdc: OSD client instance
0138  * @oid: object to lock
0139  * @oloc: object to lock
0140  * @lock_name: the name of the lock
0141  * @cookie: user-defined identifier for this instance of the lock
0142  * @locker: current lock owner
0143  */
0144 int ceph_cls_break_lock(struct ceph_osd_client *osdc,
0145             struct ceph_object_id *oid,
0146             struct ceph_object_locator *oloc,
0147             char *lock_name, char *cookie,
0148             struct ceph_entity_name *locker)
0149 {
0150     int break_op_buf_size;
0151     int name_len = strlen(lock_name);
0152     int cookie_len = strlen(cookie);
0153     struct page *break_op_page;
0154     void *p, *end;
0155     int ret;
0156
0157     break_op_buf_size = name_len + sizeof(__le32) +
0158                 cookie_len + sizeof(__le32) +
0159                 sizeof(u8) + sizeof(__le64) +
0160                 CEPH_ENCODING_START_BLK_LEN;
0161     if (break_op_buf_size > PAGE_SIZE)
0162         return -E2BIG;
0163
0164     break_op_page = alloc_page(GFP_NOIO);
0165     if (!break_op_page)
0166         return -ENOMEM;
0167
0168     p = page_address(break_op_page);
0169     end = p + break_op_buf_size;
0170
0171     /* encode cls_lock_break_op struct */
0172     ceph_start_encoding(&p, 1, 1,
0173                 break_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
0174     ceph_encode_string(&p, end, lock_name, name_len);
0175     ceph_encode_copy(&p, locker, sizeof(*locker));
0176     ceph_encode_string(&p, end, cookie, cookie_len);
0177
0178     dout("%s lock_name %s cookie %s locker %s%llu\n", __func__, lock_name,
0179          cookie, ENTITY_NAME(*locker));
0180     ret = ceph_osdc_call(osdc, oid, oloc, "lock", "break_lock",
0181                  CEPH_OSD_FLAG_WRITE, break_op_page,
0182                  break_op_buf_size, NULL, NULL);
0183
0184     dout("%s: status %d\n", __func__, ret);
0185     __free_page(break_op_page);
0186     return ret;
0187 }
0188 EXPORT_SYMBOL(ceph_cls_break_lock);
0189
0190 int ceph_cls_set_cookie(struct ceph_osd_client *osdc,
0191             struct ceph_object_id *oid,
0192             struct ceph_object_locator *oloc,
0193             char *lock_name, u8 type, char *old_cookie,
0194             char *tag, char *new_cookie)
0195 {
0196     int cookie_op_buf_size;
0197     int name_len = strlen(lock_name);
0198     int old_cookie_len = strlen(old_cookie);
0199     int tag_len = strlen(tag);
0200     int new_cookie_len = strlen(new_cookie);
0201     void *p, *end;
0202     struct page *cookie_op_page;
0203     int ret;
0204
0205     cookie_op_buf_size = name_len + sizeof(__le32) +
0206                  old_cookie_len + sizeof(__le32) +
0207                  tag_len + sizeof(__le32) +
0208                  new_cookie_len + sizeof(__le32) +
0209                  sizeof(u8) + CEPH_ENCODING_START_BLK_LEN;
0210     if (cookie_op_buf_size > PAGE_SIZE)
0211         return -E2BIG;
0212
0213     cookie_op_page = alloc_page(GFP_NOIO);
0214     if (!cookie_op_page)
0215         return -ENOMEM;
0216
0217     p = page_address(cookie_op_page);
0218     end = p + cookie_op_buf_size;
0219
0220     /* encode cls_lock_set_cookie_op struct */
0221     ceph_start_encoding(&p, 1, 1,
0222                 cookie_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
0223     ceph_encode_string(&p, end, lock_name, name_len);
0224     ceph_encode_8(&p, type);
0225     ceph_encode_string(&p, end, old_cookie, old_cookie_len);
0226     ceph_encode_string(&p, end, tag, tag_len);
0227     ceph_encode_string(&p, end, new_cookie, new_cookie_len);
0228
0229     dout("%s lock_name %s type %d old_cookie %s tag %s new_cookie %s\n",
0230          __func__, lock_name, type, old_cookie, tag, new_cookie);
0231     ret = ceph_osdc_call(osdc, oid, oloc, "lock", "set_cookie",
0232                  CEPH_OSD_FLAG_WRITE, cookie_op_page,
0233                  cookie_op_buf_size, NULL, NULL);
0234
0235     dout("%s: status %d\n", __func__, ret);
0236     __free_page(cookie_op_page);
0237     return ret;
0238 }
0239 EXPORT_SYMBOL(ceph_cls_set_cookie);
0240
0241 void ceph_free_lockers(struct ceph_locker *lockers, u32 num_lockers)
0242 {
0243     int i;
0244
0245     for (i = 0; i < num_lockers; i++)
0246         kfree(lockers[i].id.cookie);
0247     kfree(lockers);
0248 }
0249 EXPORT_SYMBOL(ceph_free_lockers);
0250
0251 static int decode_locker(void **p, void *end, struct ceph_locker *locker)
0252 {
0253     u8 struct_v;
0254     u32 len;
0255     char *s;
0256     int ret;
0257
0258     ret = ceph_start_decoding(p, end, 1, "locker_id_t", &struct_v, &len);
0259     if (ret)
0260         return ret;
0261
0262     ceph_decode_copy(p, &locker->id.name, sizeof(locker->id.name));
0263     s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
0264     if (IS_ERR(s))
0265         return PTR_ERR(s);
0266
0267     locker->id.cookie = s;
0268
0269     ret = ceph_start_decoding(p, end, 1, "locker_info_t", &struct_v, &len);
0270     if (ret)
0271         return ret;
0272
0273     *p += sizeof(struct ceph_timespec); /* skip expiration */
0274
0275     ret = ceph_decode_entity_addr(p, end, &locker->info.addr);
0276     if (ret)
0277         return ret;
0278
0279     len = ceph_decode_32(p);
0280     *p += len; /* skip description */
0281
0282     dout("%s %s%llu cookie %s addr %s\n", __func__,
0283          ENTITY_NAME(locker->id.name), locker->id.cookie,
0284          ceph_pr_addr(&locker->info.addr));
0285     return 0;
0286 }
0287
0288 static int decode_lockers(void **p, void *end, u8 *type, char **tag,
0289               struct ceph_locker **lockers, u32 *num_lockers)
0290 {
0291     u8 struct_v;
0292     u32 struct_len;
0293     char *s;
0294     int i;
0295     int ret;
0296
0297     ret = ceph_start_decoding(p, end, 1, "cls_lock_get_info_reply",
0298                   &struct_v, &struct_len);
0299     if (ret)
0300         return ret;
0301
0302     *num_lockers = ceph_decode_32(p);
0303     *lockers = kcalloc(*num_lockers, sizeof(**lockers), GFP_NOIO);
0304     if (!*lockers)
0305         return -ENOMEM;
0306
0307     for (i = 0; i < *num_lockers; i++) {
0308         ret = decode_locker(p, end, *lockers + i);
0309         if (ret)
0310             goto err_free_lockers;
0311     }
0312
0313     *type = ceph_decode_8(p);
0314     s = ceph_extract_encoded_string(p, end, NULL, GFP_NOIO);
0315     if (IS_ERR(s)) {
0316         ret = PTR_ERR(s);
0317         goto err_free_lockers;
0318     }
0319
0320     *tag = s;
0321     return 0;
0322
0323 err_free_lockers:
0324     ceph_free_lockers(*lockers, *num_lockers);
0325     return ret;
0326 }
0327
0328 /*
0329  * On success, the caller is responsible for:
0330  *
0331  *     kfree(tag);
0332  *     ceph_free_lockers(lockers, num_lockers);
0333  */
0334 int ceph_cls_lock_info(struct ceph_osd_client *osdc,
0335                struct ceph_object_id *oid,
0336                struct ceph_object_locator *oloc,
0337                char *lock_name, u8 *type, char **tag,
0338                struct ceph_locker **lockers, u32 *num_lockers)
0339 {
0340     int get_info_op_buf_size;
0341     int name_len = strlen(lock_name);
0342     struct page *get_info_op_page, *reply_page;
0343     size_t reply_len = PAGE_SIZE;
0344     void *p, *end;
0345     int ret;
0346
0347     get_info_op_buf_size = name_len + sizeof(__le32) +
0348                    CEPH_ENCODING_START_BLK_LEN;
0349     if (get_info_op_buf_size > PAGE_SIZE)
0350         return -E2BIG;
0351
0352     get_info_op_page = alloc_page(GFP_NOIO);
0353     if (!get_info_op_page)
0354         return -ENOMEM;
0355
0356     reply_page = alloc_page(GFP_NOIO);
0357     if (!reply_page) {
0358         __free_page(get_info_op_page);
0359         return -ENOMEM;
0360     }
0361
0362     p = page_address(get_info_op_page);
0363     end = p + get_info_op_buf_size;
0364
0365     /* encode cls_lock_get_info_op struct */
0366     ceph_start_encoding(&p, 1, 1,
0367                 get_info_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
0368     ceph_encode_string(&p, end, lock_name, name_len);
0369
0370     dout("%s lock_name %s\n", __func__, lock_name);
0371     ret = ceph_osdc_call(osdc, oid, oloc, "lock", "get_info",
0372                  CEPH_OSD_FLAG_READ, get_info_op_page,
0373                  get_info_op_buf_size, &reply_page, &reply_len);
0374
0375     dout("%s: status %d\n", __func__, ret);
0376     if (ret >= 0) {
0377         p = page_address(reply_page);
0378         end = p + reply_len;
0379
0380         ret = decode_lockers(&p, end, type, tag, lockers, num_lockers);
0381     }
0382
0383     __free_page(get_info_op_page);
0384     __free_page(reply_page);
0385     return ret;
0386 }
0387 EXPORT_SYMBOL(ceph_cls_lock_info);
0388
0389 int ceph_cls_assert_locked(struct ceph_osd_request *req, int which,
0390                char *lock_name, u8 type, char *cookie, char *tag)
0391 {
0392     int assert_op_buf_size;
0393     int name_len = strlen(lock_name);
0394     int cookie_len = strlen(cookie);
0395     int tag_len = strlen(tag);
0396     struct page **pages;
0397     void *p, *end;
0398     int ret;
0399
0400     assert_op_buf_size = name_len + sizeof(__le32) +
0401                  cookie_len + sizeof(__le32) +
0402                  tag_len + sizeof(__le32) +
0403                  sizeof(u8) + CEPH_ENCODING_START_BLK_LEN;
0404     if (assert_op_buf_size > PAGE_SIZE)
0405         return -E2BIG;
0406
0407     ret = osd_req_op_cls_init(req, which, "lock", "assert_locked");
0408     if (ret)
0409         return ret;
0410
0411     pages = ceph_alloc_page_vector(1, GFP_NOIO);
0412     if (IS_ERR(pages))
0413         return PTR_ERR(pages);
0414
0415     p = page_address(pages[0]);
0416     end = p + assert_op_buf_size;
0417
0418     /* encode cls_lock_assert_op struct */
0419     ceph_start_encoding(&p, 1, 1,
0420                 assert_op_buf_size - CEPH_ENCODING_START_BLK_LEN);
0421     ceph_encode_string(&p, end, lock_name, name_len);
0422     ceph_encode_8(&p, type);
0423     ceph_encode_string(&p, end, cookie, cookie_len);
0424     ceph_encode_string(&p, end, tag, tag_len);
0425     WARN_ON(p != end);
0426
0427     osd_req_op_cls_request_data_pages(req, which, pages, assert_op_buf_size,
0428                       0, false, true);
0429     return 0;
0430 }
0431 EXPORT_SYMBOL(ceph_cls_assert_locked);