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