xref: /openbmc/linux/net/ceph/cls_lock_client.c (revision b830f94f)
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);
426