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 */
ceph_cls_lock(struct ceph_osd_client * osdc,struct ceph_object_id * oid,struct ceph_object_locator * oloc,char * lock_name,u8 type,char * cookie,char * tag,char * desc,u8 flags)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 */
ceph_cls_unlock(struct ceph_osd_client * osdc,struct ceph_object_id * oid,struct ceph_object_locator * oloc,char * lock_name,char * cookie)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 */
ceph_cls_break_lock(struct ceph_osd_client * osdc,struct ceph_object_id * oid,struct ceph_object_locator * oloc,char * lock_name,char * cookie,struct ceph_entity_name * locker)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
ceph_cls_set_cookie(struct ceph_osd_client * osdc,struct ceph_object_id * oid,struct ceph_object_locator * oloc,char * lock_name,u8 type,char * old_cookie,char * tag,char * new_cookie)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
ceph_free_lockers(struct ceph_locker * lockers,u32 num_lockers)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
decode_locker(void ** p,void * end,struct ceph_locker * locker)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
decode_lockers(void ** p,void * end,u8 * type,char ** tag,struct ceph_locker ** lockers,u32 * num_lockers)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 */
ceph_cls_lock_info(struct ceph_osd_client * osdc,struct ceph_object_id * oid,struct ceph_object_locator * oloc,char * lock_name,u8 * type,char ** tag,struct ceph_locker ** lockers,u32 * num_lockers)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
ceph_cls_assert_locked(struct ceph_osd_request * req,int which,char * lock_name,u8 type,char * cookie,char * tag)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);
426