net/ceph/cls_lock_client.c

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