aboutsummaryrefslogtreecommitdiffstats
path: root/fs
diff options
context:
space:
mode:
authorSage Weil <sage@newdream.net>2009-10-06 14:31:10 -0400
committerSage Weil <sage@newdream.net>2009-10-06 14:31:10 -0400
commitf24e9980eb860d8600cbe5ef3d2fd9295320d229 (patch)
tree10f43450ad2cd4d799dd02d33c02d4ed8bef39d6 /fs
parent2f2dc053404febedc9c273452d9d518fb31fde72 (diff)
ceph: OSD client
The OSD client is responsible for reading and writing data from/to the object storage pool. This includes determining where objects are stored in the cluster, and ensuring that requests are retried or redirected in the event of a node failure or data migration. If an OSD does not respond before a timeout expires, keepalive messages are sent across the lossless, ordered communications channel to ensure that any break in the TCP is discovered. If the session does reset, a reconnection is attempted and affected requests are resent (by the message transport layer). Signed-off-by: Sage Weil <sage@newdream.net>
Diffstat (limited to 'fs')
-rw-r--r--fs/ceph/osd_client.c1294
-rw-r--r--fs/ceph/osd_client.h144
-rw-r--r--fs/ceph/osdmap.c875
-rw-r--r--fs/ceph/osdmap.h123
4 files changed, 2436 insertions, 0 deletions
diff --git a/fs/ceph/osd_client.c b/fs/ceph/osd_client.c
new file mode 100644
index 00000000000..978593a4f46
--- /dev/null
+++ b/fs/ceph/osd_client.c
@@ -0,0 +1,1294 @@
1#include "ceph_debug.h"
2
3#include <linux/err.h>
4#include <linux/highmem.h>
5#include <linux/mm.h>
6#include <linux/pagemap.h>
7#include <linux/slab.h>
8#include <linux/uaccess.h>
9
10#include "super.h"
11#include "osd_client.h"
12#include "messenger.h"
13#include "decode.h"
14
15const static struct ceph_connection_operations osd_con_ops;
16
17static void kick_requests(struct ceph_osd_client *osdc, struct ceph_osd *osd);
18
19/*
20 * Implement client access to distributed object storage cluster.
21 *
22 * All data objects are stored within a cluster/cloud of OSDs, or
23 * "object storage devices." (Note that Ceph OSDs have _nothing_ to
24 * do with the T10 OSD extensions to SCSI.) Ceph OSDs are simply
25 * remote daemons serving up and coordinating consistent and safe
26 * access to storage.
27 *
28 * Cluster membership and the mapping of data objects onto storage devices
29 * are described by the osd map.
30 *
31 * We keep track of pending OSD requests (read, write), resubmit
32 * requests to different OSDs when the cluster topology/data layout
33 * change, or retry the affected requests when the communications
34 * channel with an OSD is reset.
35 */
36
37/*
38 * calculate the mapping of a file extent onto an object, and fill out the
39 * request accordingly. shorten extent as necessary if it crosses an
40 * object boundary.
41 *
42 * fill osd op in request message.
43 */
44static void calc_layout(struct ceph_osd_client *osdc,
45 struct ceph_vino vino, struct ceph_file_layout *layout,
46 u64 off, u64 *plen,
47 struct ceph_osd_request *req)
48{
49 struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
50 struct ceph_osd_op *op = (void *)(reqhead + 1);
51 u64 orig_len = *plen;
52 u64 objoff, objlen; /* extent in object */
53 u64 bno;
54
55 reqhead->snapid = cpu_to_le64(vino.snap);
56
57 /* object extent? */
58 ceph_calc_file_object_mapping(layout, off, plen, &bno,
59 &objoff, &objlen);
60 if (*plen < orig_len)
61 dout(" skipping last %llu, final file extent %llu~%llu\n",
62 orig_len - *plen, off, *plen);
63
64 sprintf(req->r_oid, "%llx.%08llx", vino.ino, bno);
65 req->r_oid_len = strlen(req->r_oid);
66
67 op->extent.offset = cpu_to_le64(objoff);
68 op->extent.length = cpu_to_le64(objlen);
69 req->r_num_pages = calc_pages_for(off, *plen);
70
71 dout("calc_layout %s (%d) %llu~%llu (%d pages)\n",
72 req->r_oid, req->r_oid_len, objoff, objlen, req->r_num_pages);
73}
74
75
76/*
77 * requests
78 */
79void ceph_osdc_put_request(struct ceph_osd_request *req)
80{
81 dout("osdc put_request %p %d -> %d\n", req, atomic_read(&req->r_ref),
82 atomic_read(&req->r_ref)-1);
83 BUG_ON(atomic_read(&req->r_ref) <= 0);
84 if (atomic_dec_and_test(&req->r_ref)) {
85 if (req->r_request)
86 ceph_msg_put(req->r_request);
87 if (req->r_reply)
88 ceph_msg_put(req->r_reply);
89 if (req->r_own_pages)
90 ceph_release_page_vector(req->r_pages,
91 req->r_num_pages);
92 ceph_put_snap_context(req->r_snapc);
93 if (req->r_mempool)
94 mempool_free(req, req->r_osdc->req_mempool);
95 else
96 kfree(req);
97 }
98}
99
100/*
101 * build new request AND message, calculate layout, and adjust file
102 * extent as needed.
103 *
104 * if the file was recently truncated, we include information about its
105 * old and new size so that the object can be updated appropriately. (we
106 * avoid synchronously deleting truncated objects because it's slow.)
107 *
108 * if @do_sync, include a 'startsync' command so that the osd will flush
109 * data quickly.
110 */
111struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *osdc,
112 struct ceph_file_layout *layout,
113 struct ceph_vino vino,
114 u64 off, u64 *plen,
115 int opcode, int flags,
116 struct ceph_snap_context *snapc,
117 int do_sync,
118 u32 truncate_seq,
119 u64 truncate_size,
120 struct timespec *mtime,
121 bool use_mempool, int num_reply)
122{
123 struct ceph_osd_request *req;
124 struct ceph_msg *msg;
125 struct ceph_osd_request_head *head;
126 struct ceph_osd_op *op;
127 void *p;
128 int do_trunc = truncate_seq && (off + *plen > truncate_size);
129 int num_op = 1 + do_sync + do_trunc;
130 size_t msg_size = sizeof(*head) + num_op*sizeof(*op);
131 int err, i;
132 u64 prevofs;
133
134 if (use_mempool) {
135 req = mempool_alloc(osdc->req_mempool, GFP_NOFS);
136 memset(req, 0, sizeof(*req));
137 } else {
138 req = kzalloc(sizeof(*req), GFP_NOFS);
139 }
140 if (req == NULL)
141 return ERR_PTR(-ENOMEM);
142
143 err = ceph_msgpool_resv(&osdc->msgpool_op_reply, num_reply);
144 if (err) {
145 ceph_osdc_put_request(req);
146 return ERR_PTR(-ENOMEM);
147 }
148
149 req->r_osdc = osdc;
150 req->r_mempool = use_mempool;
151 atomic_set(&req->r_ref, 1);
152 init_completion(&req->r_completion);
153 init_completion(&req->r_safe_completion);
154 INIT_LIST_HEAD(&req->r_unsafe_item);
155 req->r_flags = flags;
156
157 WARN_ON((flags & (CEPH_OSD_FLAG_READ|CEPH_OSD_FLAG_WRITE)) == 0);
158
159 /* create message; allow space for oid */
160 msg_size += 40;
161 if (snapc)
162 msg_size += sizeof(u64) * snapc->num_snaps;
163 if (use_mempool)
164 msg = ceph_msgpool_get(&osdc->msgpool_op);
165 else
166 msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, 0, 0, NULL);
167 if (IS_ERR(msg)) {
168 ceph_msgpool_resv(&osdc->msgpool_op_reply, num_reply);
169 ceph_osdc_put_request(req);
170 return ERR_PTR(PTR_ERR(msg));
171 }
172 msg->hdr.type = cpu_to_le16(CEPH_MSG_OSD_OP);
173 memset(msg->front.iov_base, 0, msg->front.iov_len);
174 head = msg->front.iov_base;
175 op = (void *)(head + 1);
176 p = (void *)(op + num_op);
177
178 req->r_request = msg;
179 req->r_snapc = ceph_get_snap_context(snapc);
180
181 head->client_inc = cpu_to_le32(1); /* always, for now. */
182 head->flags = cpu_to_le32(flags);
183 if (flags & CEPH_OSD_FLAG_WRITE)
184 ceph_encode_timespec(&head->mtime, mtime);
185 head->num_ops = cpu_to_le16(num_op);
186 op->op = cpu_to_le16(opcode);
187
188 /* calculate max write size */
189 calc_layout(osdc, vino, layout, off, plen, req);
190 req->r_file_layout = *layout; /* keep a copy */
191
192 if (flags & CEPH_OSD_FLAG_WRITE) {
193 req->r_request->hdr.data_off = cpu_to_le16(off);
194 req->r_request->hdr.data_len = cpu_to_le32(*plen);
195 op->payload_len = cpu_to_le32(*plen);
196 }
197
198 /* fill in oid */
199 head->object_len = cpu_to_le32(req->r_oid_len);
200 memcpy(p, req->r_oid, req->r_oid_len);
201 p += req->r_oid_len;
202
203 /* additional ops */
204 if (do_trunc) {
205 op++;
206 op->op = cpu_to_le16(opcode == CEPH_OSD_OP_READ ?
207 CEPH_OSD_OP_MASKTRUNC : CEPH_OSD_OP_SETTRUNC);
208 op->trunc.truncate_seq = cpu_to_le32(truncate_seq);
209 prevofs = le64_to_cpu((op-1)->extent.offset);
210 op->trunc.truncate_size = cpu_to_le64(truncate_size -
211 (off-prevofs));
212 }
213 if (do_sync) {
214 op++;
215 op->op = cpu_to_le16(CEPH_OSD_OP_STARTSYNC);
216 }
217 if (snapc) {
218 head->snap_seq = cpu_to_le64(snapc->seq);
219 head->num_snaps = cpu_to_le32(snapc->num_snaps);
220 for (i = 0; i < snapc->num_snaps; i++) {
221 put_unaligned_le64(snapc->snaps[i], p);
222 p += sizeof(u64);
223 }
224 }
225
226 BUG_ON(p > msg->front.iov_base + msg->front.iov_len);
227 return req;
228}
229
230/*
231 * We keep osd requests in an rbtree, sorted by ->r_tid.
232 */
233static void __insert_request(struct ceph_osd_client *osdc,
234 struct ceph_osd_request *new)
235{
236 struct rb_node **p = &osdc->requests.rb_node;
237 struct rb_node *parent = NULL;
238 struct ceph_osd_request *req = NULL;
239
240 while (*p) {
241 parent = *p;
242 req = rb_entry(parent, struct ceph_osd_request, r_node);
243 if (new->r_tid < req->r_tid)
244 p = &(*p)->rb_left;
245 else if (new->r_tid > req->r_tid)
246 p = &(*p)->rb_right;
247 else
248 BUG();
249 }
250
251 rb_link_node(&new->r_node, parent, p);
252 rb_insert_color(&new->r_node, &osdc->requests);
253}
254
255static struct ceph_osd_request *__lookup_request(struct ceph_osd_client *osdc,
256 u64 tid)
257{
258 struct ceph_osd_request *req;
259 struct rb_node *n = osdc->requests.rb_node;
260
261 while (n) {
262 req = rb_entry(n, struct ceph_osd_request, r_node);
263 if (tid < req->r_tid)
264 n = n->rb_left;
265 else if (tid > req->r_tid)
266 n = n->rb_right;
267 else
268 return req;
269 }
270 return NULL;
271}
272
273static struct ceph_osd_request *
274__lookup_request_ge(struct ceph_osd_client *osdc,
275 u64 tid)
276{
277 struct ceph_osd_request *req;
278 struct rb_node *n = osdc->requests.rb_node;
279
280 while (n) {
281 req = rb_entry(n, struct ceph_osd_request, r_node);
282 if (tid < req->r_tid) {
283 if (!n->rb_left)
284 return req;
285 n = n->rb_left;
286 } else if (tid > req->r_tid) {
287 n = n->rb_right;
288 } else {
289 return req;
290 }
291 }
292 return NULL;
293}
294
295
296/*
297 * The messaging layer will reconnect to the osd as needed. If the
298 * session has dropped, the OSD will have dropped the session state,
299 * and we'll get notified by the messaging layer. If that happens, we
300 * need to resubmit all requests for that osd.
301 */
302static void osd_reset(struct ceph_connection *con)
303{
304 struct ceph_osd *osd = con->private;
305 struct ceph_osd_client *osdc;
306
307 if (!osd)
308 return;
309 dout("osd_reset osd%d\n", osd->o_osd);
310 osdc = osd->o_osdc;
311 osd->o_incarnation++;
312 down_read(&osdc->map_sem);
313 kick_requests(osdc, osd);
314 up_read(&osdc->map_sem);
315}
316
317/*
318 * Track open sessions with osds.
319 */
320static struct ceph_osd *create_osd(struct ceph_osd_client *osdc)
321{
322 struct ceph_osd *osd;
323
324 osd = kzalloc(sizeof(*osd), GFP_NOFS);
325 if (!osd)
326 return NULL;
327
328 atomic_set(&osd->o_ref, 1);
329 osd->o_osdc = osdc;
330 INIT_LIST_HEAD(&osd->o_requests);
331 osd->o_incarnation = 1;
332
333 ceph_con_init(osdc->client->msgr, &osd->o_con);
334 osd->o_con.private = osd;
335 osd->o_con.ops = &osd_con_ops;
336 osd->o_con.peer_name.type = CEPH_ENTITY_TYPE_OSD;
337 return osd;
338}
339
340static struct ceph_osd *get_osd(struct ceph_osd *osd)
341{
342 if (atomic_inc_not_zero(&osd->o_ref)) {
343 dout("get_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref)-1,
344 atomic_read(&osd->o_ref));
345 return osd;
346 } else {
347 dout("get_osd %p FAIL\n", osd);
348 return NULL;
349 }
350}
351
352static void put_osd(struct ceph_osd *osd)
353{
354 dout("put_osd %p %d -> %d\n", osd, atomic_read(&osd->o_ref),
355 atomic_read(&osd->o_ref) - 1);
356 if (atomic_dec_and_test(&osd->o_ref)) {
357 ceph_con_shutdown(&osd->o_con);
358 kfree(osd);
359 }
360}
361
362/*
363 * remove an osd from our map
364 */
365static void remove_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
366{
367 dout("remove_osd %p\n", osd);
368 BUG_ON(!list_empty(&osd->o_requests));
369 rb_erase(&osd->o_node, &osdc->osds);
370 ceph_con_close(&osd->o_con);
371 put_osd(osd);
372}
373
374/*
375 * reset osd connect
376 */
377static int reset_osd(struct ceph_osd_client *osdc, struct ceph_osd *osd)
378{
379 int ret = 0;
380
381 dout("reset_osd %p osd%d\n", osd, osd->o_osd);
382 if (list_empty(&osd->o_requests)) {
383 remove_osd(osdc, osd);
384 } else {
385 ceph_con_close(&osd->o_con);
386 ceph_con_open(&osd->o_con, &osdc->osdmap->osd_addr[osd->o_osd]);
387 osd->o_incarnation++;
388 }
389 return ret;
390}
391
392static void __insert_osd(struct ceph_osd_client *osdc, struct ceph_osd *new)
393{
394 struct rb_node **p = &osdc->osds.rb_node;
395 struct rb_node *parent = NULL;
396 struct ceph_osd *osd = NULL;
397
398 while (*p) {
399 parent = *p;
400 osd = rb_entry(parent, struct ceph_osd, o_node);
401 if (new->o_osd < osd->o_osd)
402 p = &(*p)->rb_left;
403 else if (new->o_osd > osd->o_osd)
404 p = &(*p)->rb_right;
405 else
406 BUG();
407 }
408
409 rb_link_node(&new->o_node, parent, p);
410 rb_insert_color(&new->o_node, &osdc->osds);
411}
412
413static struct ceph_osd *__lookup_osd(struct ceph_osd_client *osdc, int o)
414{
415 struct ceph_osd *osd;
416 struct rb_node *n = osdc->osds.rb_node;
417
418 while (n) {
419 osd = rb_entry(n, struct ceph_osd, o_node);
420 if (o < osd->o_osd)
421 n = n->rb_left;
422 else if (o > osd->o_osd)
423 n = n->rb_right;
424 else
425 return osd;
426 }
427 return NULL;
428}
429
430
431/*
432 * Register request, assign tid. If this is the first request, set up
433 * the timeout event.
434 */
435static void register_request(struct ceph_osd_client *osdc,
436 struct ceph_osd_request *req)
437{
438 struct ceph_osd_request_head *head = req->r_request->front.iov_base;
439
440 mutex_lock(&osdc->request_mutex);
441 req->r_tid = ++osdc->last_tid;
442 head->tid = cpu_to_le64(req->r_tid);
443
444 dout("register_request %p tid %lld\n", req, req->r_tid);
445 __insert_request(osdc, req);
446 ceph_osdc_get_request(req);
447 osdc->num_requests++;
448
449 req->r_timeout_stamp =
450 jiffies + osdc->client->mount_args.osd_timeout*HZ;
451
452 if (osdc->num_requests == 1) {
453 osdc->timeout_tid = req->r_tid;
454 dout(" timeout on tid %llu at %lu\n", req->r_tid,
455 req->r_timeout_stamp);
456 schedule_delayed_work(&osdc->timeout_work,
457 round_jiffies_relative(req->r_timeout_stamp - jiffies));
458 }
459 mutex_unlock(&osdc->request_mutex);
460}
461
462/*
463 * called under osdc->request_mutex
464 */
465static void __unregister_request(struct ceph_osd_client *osdc,
466 struct ceph_osd_request *req)
467{
468 dout("__unregister_request %p tid %lld\n", req, req->r_tid);
469 rb_erase(&req->r_node, &osdc->requests);
470 osdc->num_requests--;
471
472 list_del_init(&req->r_osd_item);
473 if (list_empty(&req->r_osd->o_requests))
474 remove_osd(osdc, req->r_osd);
475 req->r_osd = NULL;
476
477 ceph_osdc_put_request(req);
478
479 if (req->r_tid == osdc->timeout_tid) {
480 if (osdc->num_requests == 0) {
481 dout("no requests, canceling timeout\n");
482 osdc->timeout_tid = 0;
483 cancel_delayed_work(&osdc->timeout_work);
484 } else {
485 req = rb_entry(rb_first(&osdc->requests),
486 struct ceph_osd_request, r_node);
487 osdc->timeout_tid = req->r_tid;
488 dout("rescheduled timeout on tid %llu at %lu\n",
489 req->r_tid, req->r_timeout_stamp);
490 schedule_delayed_work(&osdc->timeout_work,
491 round_jiffies_relative(req->r_timeout_stamp -
492 jiffies));
493 }
494 }
495}
496
497/*
498 * Cancel a previously queued request message
499 */
500static void __cancel_request(struct ceph_osd_request *req)
501{
502 if (req->r_sent) {
503 ceph_con_revoke(&req->r_osd->o_con, req->r_request);
504 req->r_sent = 0;
505 }
506}
507
508/*
509 * Pick an osd (the first 'up' osd in the pg), allocate the osd struct
510 * (as needed), and set the request r_osd appropriately. If there is
511 * no up osd, set r_osd to NULL.
512 *
513 * Return 0 if unchanged, 1 if changed, or negative on error.
514 *
515 * Caller should hold map_sem for read and request_mutex.
516 */
517static int __map_osds(struct ceph_osd_client *osdc,
518 struct ceph_osd_request *req)
519{
520 struct ceph_osd_request_head *reqhead = req->r_request->front.iov_base;
521 union ceph_pg pgid;
522 int o = -1;
523 int err;
524 struct ceph_osd *newosd = NULL;
525
526 dout("map_osds %p tid %lld\n", req, req->r_tid);
527 err = ceph_calc_object_layout(&reqhead->layout, req->r_oid,
528 &req->r_file_layout, osdc->osdmap);
529 if (err)
530 return err;
531 pgid.pg64 = le64_to_cpu(reqhead->layout.ol_pgid);
532 o = ceph_calc_pg_primary(osdc->osdmap, pgid);
533
534 if ((req->r_osd && req->r_osd->o_osd == o &&
535 req->r_sent >= req->r_osd->o_incarnation) ||
536 (req->r_osd == NULL && o == -1))
537 return 0; /* no change */
538
539 dout("map_osds tid %llu pgid %llx pool %d osd%d (was osd%d)\n",
540 req->r_tid, pgid.pg64, pgid.pg.pool, o,
541 req->r_osd ? req->r_osd->o_osd : -1);
542
543 if (req->r_osd) {
544 __cancel_request(req);
545 list_del_init(&req->r_osd_item);
546 if (list_empty(&req->r_osd->o_requests)) {
547 /* try to re-use r_osd if possible */
548 newosd = get_osd(req->r_osd);
549 remove_osd(osdc, newosd);
550 }
551 req->r_osd = NULL;
552 }
553
554 req->r_osd = __lookup_osd(osdc, o);
555 if (!req->r_osd && o >= 0) {
556 if (newosd) {
557 req->r_osd = newosd;
558 newosd = NULL;
559 } else {
560 err = -ENOMEM;
561 req->r_osd = create_osd(osdc);
562 if (!req->r_osd)
563 goto out;
564 }
565
566 dout("map_osds osd %p is osd%d\n", req->r_osd, o);
567 req->r_osd->o_osd = o;
568 req->r_osd->o_con.peer_name.num = cpu_to_le64(o);
569 __insert_osd(osdc, req->r_osd);
570
571 ceph_con_open(&req->r_osd->o_con, &osdc->osdmap->osd_addr[o]);
572 }
573
574 if (req->r_osd)
575 list_add(&req->r_osd_item, &req->r_osd->o_requests);
576 err = 1; /* osd changed */
577
578out:
579 if (newosd)
580 put_osd(newosd);
581 return err;
582}
583
584/*
585 * caller should hold map_sem (for read) and request_mutex
586 */
587static int __send_request(struct ceph_osd_client *osdc,
588 struct ceph_osd_request *req)
589{
590 struct ceph_osd_request_head *reqhead;
591 int err;
592
593 err = __map_osds(osdc, req);
594 if (err < 0)
595 return err;
596 if (req->r_osd == NULL) {
597 dout("send_request %p no up osds in pg\n", req);
598 ceph_monc_request_next_osdmap(&osdc->client->monc);
599 return 0;
600 }
601
602 dout("send_request %p tid %llu to osd%d flags %d\n",
603 req, req->r_tid, req->r_osd->o_osd, req->r_flags);
604
605 reqhead = req->r_request->front.iov_base;
606 reqhead->osdmap_epoch = cpu_to_le32(osdc->osdmap->epoch);
607 reqhead->flags |= cpu_to_le32(req->r_flags); /* e.g., RETRY */
608 reqhead->reassert_version = req->r_reassert_version;
609
610 req->r_timeout_stamp = jiffies+osdc->client->mount_args.osd_timeout*HZ;
611
612 ceph_msg_get(req->r_request); /* send consumes a ref */
613 ceph_con_send(&req->r_osd->o_con, req->r_request);
614 req->r_sent = req->r_osd->o_incarnation;
615 return 0;
616}
617
618/*
619 * Timeout callback, called every N seconds when 1 or more osd
620 * requests has been active for more than N seconds. When this
621 * happens, we ping all OSDs with requests who have timed out to
622 * ensure any communications channel reset is detected. Reset the
623 * request timeouts another N seconds in the future as we go.
624 * Reschedule the timeout event another N seconds in future (unless
625 * there are no open requests).
626 */
627static void handle_timeout(struct work_struct *work)
628{
629 struct ceph_osd_client *osdc =
630 container_of(work, struct ceph_osd_client, timeout_work.work);
631 struct ceph_osd_request *req;
632 struct ceph_osd *osd;
633 unsigned long timeout = osdc->client->mount_args.osd_timeout * HZ;
634 unsigned long next_timeout = timeout + jiffies;
635 struct rb_node *p;
636
637 dout("timeout\n");
638 down_read(&osdc->map_sem);
639
640 ceph_monc_request_next_osdmap(&osdc->client->monc);
641
642 mutex_lock(&osdc->request_mutex);
643 for (p = rb_first(&osdc->requests); p; p = rb_next(p)) {
644 req = rb_entry(p, struct ceph_osd_request, r_node);
645
646 if (req->r_resend) {
647 int err;
648
649 dout("osdc resending prev failed %lld\n", req->r_tid);
650 err = __send_request(osdc, req);
651 if (err)
652 dout("osdc failed again on %lld\n", req->r_tid);
653 else
654 req->r_resend = false;
655 continue;
656 }
657 }
658 for (p = rb_first(&osdc->osds); p; p = rb_next(p)) {
659 osd = rb_entry(p, struct ceph_osd, o_node);
660 if (list_empty(&osd->o_requests))
661 continue;
662 req = list_first_entry(&osd->o_requests,
663 struct ceph_osd_request, r_osd_item);
664 if (time_before(jiffies, req->r_timeout_stamp))
665 continue;
666
667 dout(" tid %llu (at least) timed out on osd%d\n",
668 req->r_tid, osd->o_osd);
669 req->r_timeout_stamp = next_timeout;
670 ceph_con_keepalive(&osd->o_con);
671 }
672
673 if (osdc->timeout_tid)
674 schedule_delayed_work(&osdc->timeout_work,
675 round_jiffies_relative(timeout));
676
677 mutex_unlock(&osdc->request_mutex);
678
679 up_read(&osdc->map_sem);
680}
681
682/*
683 * handle osd op reply. either call the callback if it is specified,
684 * or do the completion to wake up the waiting thread.
685 */
686static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg)
687{
688 struct ceph_osd_reply_head *rhead = msg->front.iov_base;
689 struct ceph_osd_request *req;
690 u64 tid;
691 int numops, object_len, flags;
692
693 if (msg->front.iov_len < sizeof(*rhead))
694 goto bad;
695 tid = le64_to_cpu(rhead->tid);
696 numops = le32_to_cpu(rhead->num_ops);
697 object_len = le32_to_cpu(rhead->object_len);
698 if (msg->front.iov_len != sizeof(*rhead) + object_len +
699 numops * sizeof(struct ceph_osd_op))
700 goto bad;
701 dout("handle_reply %p tid %llu\n", msg, tid);
702
703 /* lookup */
704 mutex_lock(&osdc->request_mutex);
705 req = __lookup_request(osdc, tid);
706 if (req == NULL) {
707 dout("handle_reply tid %llu dne\n", tid);
708 mutex_unlock(&osdc->request_mutex);
709 return;
710 }
711 ceph_osdc_get_request(req);
712 flags = le32_to_cpu(rhead->flags);
713
714 if (req->r_reply) {
715 /*
716 * once we see the message has been received, we don't
717 * need a ref (which is only needed for revoking
718 * pages)
719 */
720 ceph_msg_put(req->r_reply);
721 req->r_reply = NULL;
722 }
723
724 if (!req->r_got_reply) {
725 unsigned bytes;
726
727 req->r_result = le32_to_cpu(rhead->result);
728 bytes = le32_to_cpu(msg->hdr.data_len);
729 dout("handle_reply result %d bytes %d\n", req->r_result,
730 bytes);
731 if (req->r_result == 0)
732 req->r_result = bytes;
733
734 /* in case this is a write and we need to replay, */
735 req->r_reassert_version = rhead->reassert_version;
736
737 req->r_got_reply = 1;
738 } else if ((flags & CEPH_OSD_FLAG_ONDISK) == 0) {
739 dout("handle_reply tid %llu dup ack\n", tid);
740 goto done;
741 }
742
743 dout("handle_reply tid %llu flags %d\n", tid, flags);
744
745 /* either this is a read, or we got the safe response */
746 if ((flags & CEPH_OSD_FLAG_ONDISK) ||
747 ((flags & CEPH_OSD_FLAG_WRITE) == 0))
748 __unregister_request(osdc, req);
749
750 mutex_unlock(&osdc->request_mutex);
751
752 if (req->r_callback)
753 req->r_callback(req, msg);
754 else
755 complete(&req->r_completion);
756
757 if (flags & CEPH_OSD_FLAG_ONDISK) {
758 if (req->r_safe_callback)
759 req->r_safe_callback(req, msg);
760 complete(&req->r_safe_completion); /* fsync waiter */
761 }
762
763done:
764 ceph_osdc_put_request(req);
765 return;
766
767bad:
768 pr_err("corrupt osd_op_reply got %d %d expected %d\n",
769 (int)msg->front.iov_len, le32_to_cpu(msg->hdr.front_len),
770 (int)sizeof(*rhead));
771}
772
773
774/*
775 * Resubmit osd requests whose osd or osd address has changed. Request
776 * a new osd map if osds are down, or we are otherwise unable to determine
777 * how to direct a request.
778 *
779 * Close connections to down osds.
780 *
781 * If @who is specified, resubmit requests for that specific osd.
782 *
783 * Caller should hold map_sem for read and request_mutex.
784 */
785static void kick_requests(struct ceph_osd_client *osdc,
786 struct ceph_osd *kickosd)
787{
788 struct ceph_osd_request *req;
789 struct rb_node *p, *n;
790 int needmap = 0;
791 int err;
792
793 dout("kick_requests osd%d\n", kickosd ? kickosd->o_osd : -1);
794 mutex_lock(&osdc->request_mutex);
795 if (!kickosd) {
796 for (p = rb_first(&osdc->osds); p; p = n) {
797 struct ceph_osd *osd =
798 rb_entry(p, struct ceph_osd, o_node);
799
800 n = rb_next(p);
801 if (!ceph_osd_is_up(osdc->osdmap, osd->o_osd) ||
802 !ceph_entity_addr_equal(&osd->o_con.peer_addr,
803 ceph_osd_addr(osdc->osdmap,
804 osd->o_osd)))
805 reset_osd(osdc, osd);
806 }
807 }
808
809 for (p = rb_first(&osdc->requests); p; p = rb_next(p)) {
810 req = rb_entry(p, struct ceph_osd_request, r_node);
811
812 if (req->r_resend) {
813 dout(" r_resend set on tid %llu\n", req->r_tid);
814 goto kick;
815 }
816 if (req->r_osd && kickosd == req->r_osd)
817 goto kick;
818
819 err = __map_osds(osdc, req);
820 if (err == 0)
821 continue; /* no change */
822 if (err < 0) {
823 /*
824 * FIXME: really, we should set the request
825 * error and fail if this isn't a 'nofail'
826 * request, but that's a fair bit more
827 * complicated to do. So retry!
828 */
829 dout(" setting r_resend on %llu\n", req->r_tid);
830 req->r_resend = true;
831 continue;
832 }
833 if (req->r_osd == NULL) {
834 dout("tid %llu maps to no valid osd\n", req->r_tid);
835 needmap++; /* request a newer map */
836 continue;
837 }
838
839kick:
840 dout("kicking tid %llu osd%d\n", req->r_tid, req->r_osd->o_osd);
841 req->r_flags |= CEPH_OSD_FLAG_RETRY;
842 err = __send_request(osdc, req);
843 if (err) {
844 dout(" setting r_resend on %llu\n", req->r_tid);
845 req->r_resend = true;
846 }
847 }
848 mutex_unlock(&osdc->request_mutex);
849
850 if (needmap) {
851 dout("%d requests for down osds, need new map\n", needmap);
852 ceph_monc_request_next_osdmap(&osdc->client->monc);
853 }
854}
855
856/*
857 * Process updated osd map.
858 *
859 * The message contains any number of incremental and full maps, normally
860 * indicating some sort of topology change in the cluster. Kick requests
861 * off to different OSDs as needed.
862 */
863void ceph_osdc_handle_map(struct ceph_osd_client *osdc, struct ceph_msg *msg)
864{
865 void *p, *end, *next;
866 u32 nr_maps, maplen;
867 u32 epoch;
868 struct ceph_osdmap *newmap = NULL, *oldmap;
869 int err;
870 struct ceph_fsid fsid;
871
872 dout("handle_map have %u\n", osdc->osdmap ? osdc->osdmap->epoch : 0);
873 p = msg->front.iov_base;
874 end = p + msg->front.iov_len;
875
876 /* verify fsid */
877 ceph_decode_need(&p, end, sizeof(fsid), bad);
878 ceph_decode_copy(&p, &fsid, sizeof(fsid));
879 if (ceph_fsid_compare(&fsid, &osdc->client->monc.monmap->fsid)) {
880 pr_err("got osdmap with wrong fsid, ignoring\n");
881 return;
882 }
883
884 down_write(&osdc->map_sem);
885
886 /* incremental maps */
887 ceph_decode_32_safe(&p, end, nr_maps, bad);
888 dout(" %d inc maps\n", nr_maps);
889 while (nr_maps > 0) {
890 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
891 ceph_decode_32(&p, epoch);
892 ceph_decode_32(&p, maplen);
893 ceph_decode_need(&p, end, maplen, bad);
894 next = p + maplen;
895 if (osdc->osdmap && osdc->osdmap->epoch+1 == epoch) {
896 dout("applying incremental map %u len %d\n",
897 epoch, maplen);
898 newmap = osdmap_apply_incremental(&p, next,
899 osdc->osdmap,
900 osdc->client->msgr);
901 if (IS_ERR(newmap)) {
902 err = PTR_ERR(newmap);
903 goto bad;
904 }
905 if (newmap != osdc->osdmap) {
906 ceph_osdmap_destroy(osdc->osdmap);
907 osdc->osdmap = newmap;
908 }
909 } else {
910 dout("ignoring incremental map %u len %d\n",
911 epoch, maplen);
912 }
913 p = next;
914 nr_maps--;
915 }
916 if (newmap)
917 goto done;
918
919 /* full maps */
920 ceph_decode_32_safe(&p, end, nr_maps, bad);
921 dout(" %d full maps\n", nr_maps);
922 while (nr_maps) {
923 ceph_decode_need(&p, end, 2*sizeof(u32), bad);
924 ceph_decode_32(&p, epoch);
925 ceph_decode_32(&p, maplen);
926 ceph_decode_need(&p, end, maplen, bad);
927 if (nr_maps > 1) {
928 dout("skipping non-latest full map %u len %d\n",
929 epoch, maplen);
930 } else if (osdc->osdmap && osdc->osdmap->epoch >= epoch) {
931 dout("skipping full map %u len %d, "
932 "older than our %u\n", epoch, maplen,
933 osdc->osdmap->epoch);
934 } else {
935 dout("taking full map %u len %d\n", epoch, maplen);
936 newmap = osdmap_decode(&p, p+maplen);
937 if (IS_ERR(newmap)) {
938 err = PTR_ERR(newmap);
939 goto bad;
940 }
941 oldmap = osdc->osdmap;
942 osdc->osdmap = newmap;
943 if (oldmap)
944 ceph_osdmap_destroy(oldmap);
945 }
946 p += maplen;
947 nr_maps--;
948 }
949
950done:
951 downgrade_write(&osdc->map_sem);
952 ceph_monc_got_osdmap(&osdc->client->monc, osdc->osdmap->epoch);
953 if (newmap)
954 kick_requests(osdc, NULL);
955 up_read(&osdc->map_sem);
956 return;
957
958bad:
959 pr_err("osdc handle_map corrupt msg\n");
960 up_write(&osdc->map_sem);
961 return;
962}
963
964
965/*
966 * A read request prepares specific pages that data is to be read into.
967 * When a message is being read off the wire, we call prepare_pages to
968 * find those pages.
969 * 0 = success, -1 failure.
970 */
971static int prepare_pages(struct ceph_connection *con, struct ceph_msg *m,
972 int want)
973{
974 struct ceph_osd *osd = con->private;
975 struct ceph_osd_client *osdc;
976 struct ceph_osd_reply_head *rhead = m->front.iov_base;
977 struct ceph_osd_request *req;
978 u64 tid;
979 int ret = -1;
980 int type = le16_to_cpu(m->hdr.type);
981
982 if (!osd)
983 return -1;
984 osdc = osd->o_osdc;
985
986 dout("prepare_pages on msg %p want %d\n", m, want);
987 if (unlikely(type != CEPH_MSG_OSD_OPREPLY))
988 return -1; /* hmm! */
989
990 tid = le64_to_cpu(rhead->tid);
991 mutex_lock(&osdc->request_mutex);
992 req = __lookup_request(osdc, tid);
993 if (!req) {
994 dout("prepare_pages unknown tid %llu\n", tid);
995 goto out;
996 }
997 dout("prepare_pages tid %llu has %d pages, want %d\n",
998 tid, req->r_num_pages, want);
999 if (likely(req->r_num_pages >= want && !req->r_prepared_pages)) {
1000 m->pages = req->r_pages;
1001 m->nr_pages = req->r_num_pages;
1002 req->r_reply = m; /* only for duration of read over socket */
1003 ceph_msg_get(m);
1004 req->r_prepared_pages = 1;
1005 ret = 0; /* success */
1006 }
1007out:
1008 mutex_unlock(&osdc->request_mutex);
1009 return ret;
1010}
1011
1012/*
1013 * Register request, send initial attempt.
1014 */
1015int ceph_osdc_start_request(struct ceph_osd_client *osdc,
1016 struct ceph_osd_request *req,
1017 bool nofail)
1018{
1019 int rc;
1020
1021 req->r_request->pages = req->r_pages;
1022 req->r_request->nr_pages = req->r_num_pages;
1023
1024 register_request(osdc, req);
1025
1026 down_read(&osdc->map_sem);
1027 mutex_lock(&osdc->request_mutex);
1028 rc = __send_request(osdc, req);
1029 if (rc) {
1030 if (nofail) {
1031 dout("osdc_start_request failed send, marking %lld\n",
1032 req->r_tid);
1033 req->r_resend = true;
1034 rc = 0;
1035 } else {
1036 __unregister_request(osdc, req);
1037 }
1038 }
1039 mutex_unlock(&osdc->request_mutex);
1040 up_read(&osdc->map_sem);
1041 return rc;
1042}
1043
1044/*
1045 * wait for a request to complete
1046 */
1047int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
1048 struct ceph_osd_request *req)
1049{
1050 int rc;
1051
1052 rc = wait_for_completion_interruptible(&req->r_completion);
1053 if (rc < 0) {
1054 mutex_lock(&osdc->request_mutex);
1055 __cancel_request(req);
1056 mutex_unlock(&osdc->request_mutex);
1057 dout("wait_request tid %llu timed out\n", req->r_tid);
1058 return rc;
1059 }
1060
1061 dout("wait_request tid %llu result %d\n", req->r_tid, req->r_result);
1062 return req->r_result;
1063}
1064
1065/*
1066 * sync - wait for all in-flight requests to flush. avoid starvation.
1067 */
1068void ceph_osdc_sync(struct ceph_osd_client *osdc)
1069{
1070 struct ceph_osd_request *req;
1071 u64 last_tid, next_tid = 0;
1072
1073 mutex_lock(&osdc->request_mutex);
1074 last_tid = osdc->last_tid;
1075 while (1) {
1076 req = __lookup_request_ge(osdc, next_tid);
1077 if (!req)
1078 break;
1079 if (req->r_tid > last_tid)
1080 break;
1081
1082 next_tid = req->r_tid + 1;
1083 if ((req->r_flags & CEPH_OSD_FLAG_WRITE) == 0)
1084 continue;
1085
1086 ceph_osdc_get_request(req);
1087 mutex_unlock(&osdc->request_mutex);
1088 dout("sync waiting on tid %llu (last is %llu)\n",
1089 req->r_tid, last_tid);
1090 wait_for_completion(&req->r_safe_completion);
1091 mutex_lock(&osdc->request_mutex);
1092 ceph_osdc_put_request(req);
1093 }
1094 mutex_unlock(&osdc->request_mutex);
1095 dout("sync done (thru tid %llu)\n", last_tid);
1096}
1097
1098/*
1099 * init, shutdown
1100 */
1101int ceph_osdc_init(struct ceph_osd_client *osdc, struct ceph_client *client)
1102{
1103 int err;
1104
1105 dout("init\n");
1106 osdc->client = client;
1107 osdc->osdmap = NULL;
1108 init_rwsem(&osdc->map_sem);
1109 init_completion(&osdc->map_waiters);
1110 osdc->last_requested_map = 0;
1111 mutex_init(&osdc->request_mutex);
1112 osdc->timeout_tid = 0;
1113 osdc->last_tid = 0;
1114 osdc->osds = RB_ROOT;
1115 osdc->requests = RB_ROOT;
1116 osdc->num_requests = 0;
1117 INIT_DELAYED_WORK(&osdc->timeout_work, handle_timeout);
1118
1119 osdc->req_mempool = mempool_create_kmalloc_pool(10,
1120 sizeof(struct ceph_osd_request));
1121 if (!osdc->req_mempool)
1122 return -ENOMEM;
1123
1124 err = ceph_msgpool_init(&osdc->msgpool_op, 4096, 10, true);
1125 if (err < 0)
1126 return -ENOMEM;
1127 err = ceph_msgpool_init(&osdc->msgpool_op_reply, 512, 0, false);
1128 if (err < 0)
1129 return -ENOMEM;
1130
1131 return 0;
1132}
1133
1134void ceph_osdc_stop(struct ceph_osd_client *osdc)
1135{
1136 cancel_delayed_work_sync(&osdc->timeout_work);
1137 if (osdc->osdmap) {
1138 ceph_osdmap_destroy(osdc->osdmap);
1139 osdc->osdmap = NULL;
1140 }
1141 mempool_destroy(osdc->req_mempool);
1142 ceph_msgpool_destroy(&osdc->msgpool_op);
1143 ceph_msgpool_destroy(&osdc->msgpool_op_reply);
1144}
1145
1146/*
1147 * Read some contiguous pages. If we cross a stripe boundary, shorten
1148 * *plen. Return number of bytes read, or error.
1149 */
1150int ceph_osdc_readpages(struct ceph_osd_client *osdc,
1151 struct ceph_vino vino, struct ceph_file_layout *layout,
1152 u64 off, u64 *plen,
1153 u32 truncate_seq, u64 truncate_size,
1154 struct page **pages, int num_pages)
1155{
1156 struct ceph_osd_request *req;
1157 int rc = 0;
1158
1159 dout("readpages on ino %llx.%llx on %llu~%llu\n", vino.ino,
1160 vino.snap, off, *plen);
1161 req = ceph_osdc_new_request(osdc, layout, vino, off, plen,
1162 CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
1163 NULL, 0, truncate_seq, truncate_size, NULL,
1164 false, 1);
1165 if (IS_ERR(req))
1166 return PTR_ERR(req);
1167
1168 /* it may be a short read due to an object boundary */
1169 req->r_pages = pages;
1170 num_pages = calc_pages_for(off, *plen);
1171 req->r_num_pages = num_pages;
1172
1173 dout("readpages final extent is %llu~%llu (%d pages)\n",
1174 off, *plen, req->r_num_pages);
1175
1176 rc = ceph_osdc_start_request(osdc, req, false);
1177 if (!rc)
1178 rc = ceph_osdc_wait_request(osdc, req);
1179
1180 ceph_osdc_put_request(req);
1181 dout("readpages result %d\n", rc);
1182 return rc;
1183}
1184
1185/*
1186 * do a synchronous write on N pages
1187 */
1188int ceph_osdc_writepages(struct ceph_osd_client *osdc, struct ceph_vino vino,
1189 struct ceph_file_layout *layout,
1190 struct ceph_snap_context *snapc,
1191 u64 off, u64 len,
1192 u32 truncate_seq, u64 truncate_size,
1193 struct timespec *mtime,
1194 struct page **pages, int num_pages,
1195 int flags, int do_sync, bool nofail)
1196{
1197 struct ceph_osd_request *req;
1198 int rc = 0;
1199
1200 BUG_ON(vino.snap != CEPH_NOSNAP);
1201 req = ceph_osdc_new_request(osdc, layout, vino, off, &len,
1202 CEPH_OSD_OP_WRITE,
1203 flags | CEPH_OSD_FLAG_ONDISK |
1204 CEPH_OSD_FLAG_WRITE,
1205 snapc, do_sync,
1206 truncate_seq, truncate_size, mtime,
1207 nofail, 1);
1208 if (IS_ERR(req))
1209 return PTR_ERR(req);
1210
1211 /* it may be a short write due to an object boundary */
1212 req->r_pages = pages;
1213 req->r_num_pages = calc_pages_for(off, len);
1214 dout("writepages %llu~%llu (%d pages)\n", off, len,
1215 req->r_num_pages);
1216
1217 rc = ceph_osdc_start_request(osdc, req, nofail);
1218 if (!rc)
1219 rc = ceph_osdc_wait_request(osdc, req);
1220
1221 ceph_osdc_put_request(req);
1222 if (rc == 0)
1223 rc = len;
1224 dout("writepages result %d\n", rc);
1225 return rc;
1226}
1227
1228/*
1229 * handle incoming message
1230 */
1231static void dispatch(struct ceph_connection *con, struct ceph_msg *msg)
1232{
1233 struct ceph_osd *osd = con->private;
1234 struct ceph_osd_client *osdc = osd->o_osdc;
1235 int type = le16_to_cpu(msg->hdr.type);
1236
1237 if (!osd)
1238 return;
1239
1240 switch (type) {
1241 case CEPH_MSG_OSD_MAP:
1242 ceph_osdc_handle_map(osdc, msg);
1243 break;
1244 case CEPH_MSG_OSD_OPREPLY:
1245 handle_reply(osdc, msg);
1246 break;
1247
1248 default:
1249 pr_err("received unknown message type %d %s\n", type,
1250 ceph_msg_type_name(type));
1251 }
1252 ceph_msg_put(msg);
1253}
1254
1255static struct ceph_msg *alloc_msg(struct ceph_connection *con,
1256 struct ceph_msg_header *hdr)
1257{
1258 struct ceph_osd *osd = con->private;
1259 struct ceph_osd_client *osdc = osd->o_osdc;
1260 int type = le16_to_cpu(hdr->type);
1261
1262 switch (type) {
1263 case CEPH_MSG_OSD_OPREPLY:
1264 return ceph_msgpool_get(&osdc->msgpool_op_reply);
1265 }
1266 return ceph_alloc_msg(con, hdr);
1267}
1268
1269/*
1270 * Wrappers to refcount containing ceph_osd struct
1271 */
1272static struct ceph_connection *get_osd_con(struct ceph_connection *con)
1273{
1274 struct ceph_osd *osd = con->private;
1275 if (get_osd(osd))
1276 return con;
1277 return NULL;
1278}
1279
1280static void put_osd_con(struct ceph_connection *con)
1281{
1282 struct ceph_osd *osd = con->private;
1283 put_osd(osd);
1284}
1285
1286const static struct ceph_connection_operations osd_con_ops = {
1287 .get = get_osd_con,
1288 .put = put_osd_con,
1289 .dispatch = dispatch,
1290 .alloc_msg = alloc_msg,
1291 .peer_reset = osd_reset,
1292 .alloc_middle = ceph_alloc_middle,
1293 .prepare_pages = prepare_pages,
1294};
diff --git a/fs/ceph/osd_client.h b/fs/ceph/osd_client.h
new file mode 100644
index 00000000000..9a4addf7d65
--- /dev/null
+++ b/fs/ceph/osd_client.h
@@ -0,0 +1,144 @@
1#ifndef _FS_CEPH_OSD_CLIENT_H
2#define _FS_CEPH_OSD_CLIENT_H
3
4#include <linux/completion.h>
5#include <linux/mempool.h>
6#include <linux/rbtree.h>
7
8#include "types.h"
9#include "osdmap.h"
10#include "messenger.h"
11
12struct ceph_msg;
13struct ceph_snap_context;
14struct ceph_osd_request;
15struct ceph_osd_client;
16
17/*
18 * completion callback for async writepages
19 */
20typedef void (*ceph_osdc_callback_t)(struct ceph_osd_request *,
21 struct ceph_msg *);
22
23/* a given osd we're communicating with */
24struct ceph_osd {
25 atomic_t o_ref;
26 struct ceph_osd_client *o_osdc;
27 int o_osd;
28 int o_incarnation;
29 struct rb_node o_node;
30 struct ceph_connection o_con;
31 struct list_head o_requests;
32};
33
34/* an in-flight request */
35struct ceph_osd_request {
36 u64 r_tid; /* unique for this client */
37 struct rb_node r_node;
38 struct list_head r_osd_item;
39 struct ceph_osd *r_osd;
40
41 struct ceph_msg *r_request, *r_reply;
42 int r_result;
43 int r_flags; /* any additional flags for the osd */
44 u32 r_sent; /* >0 if r_request is sending/sent */
45 int r_prepared_pages, r_got_reply;
46
47 struct ceph_osd_client *r_osdc;
48 atomic_t r_ref;
49 bool r_mempool;
50 struct completion r_completion, r_safe_completion;
51 ceph_osdc_callback_t r_callback, r_safe_callback;
52 struct ceph_eversion r_reassert_version;
53 struct list_head r_unsafe_item;
54
55 struct inode *r_inode; /* for use by callbacks */
56 struct writeback_control *r_wbc; /* ditto */
57
58 char r_oid[40]; /* object name */
59 int r_oid_len;
60 unsigned long r_timeout_stamp;
61 bool r_resend; /* msg send failed, needs retry */
62
63 struct ceph_file_layout r_file_layout;
64 struct ceph_snap_context *r_snapc; /* snap context for writes */
65 unsigned r_num_pages; /* size of page array (follows) */
66 struct page **r_pages; /* pages for data payload */
67 int r_pages_from_pool;
68 int r_own_pages; /* if true, i own page list */
69};
70
71struct ceph_osd_client {
72 struct ceph_client *client;
73
74 struct ceph_osdmap *osdmap; /* current map */
75 struct rw_semaphore map_sem;
76 struct completion map_waiters;
77 u64 last_requested_map;
78
79 struct mutex request_mutex;
80 struct rb_root osds; /* osds */
81 u64 timeout_tid; /* tid of timeout triggering rq */
82 u64 last_tid; /* tid of last request */
83 struct rb_root requests; /* pending requests */
84 int num_requests;
85 struct delayed_work timeout_work;
86 struct dentry *debugfs_file;
87
88 mempool_t *req_mempool;
89
90 struct ceph_msgpool msgpool_op;
91 struct ceph_msgpool msgpool_op_reply;
92};
93
94extern int ceph_osdc_init(struct ceph_osd_client *osdc,
95 struct ceph_client *client);
96extern void ceph_osdc_stop(struct ceph_osd_client *osdc);
97
98extern void ceph_osdc_handle_reply(struct ceph_osd_client *osdc,
99 struct ceph_msg *msg);
100extern void ceph_osdc_handle_map(struct ceph_osd_client *osdc,
101 struct ceph_msg *msg);
102
103extern struct ceph_osd_request *ceph_osdc_new_request(struct ceph_osd_client *,
104 struct ceph_file_layout *layout,
105 struct ceph_vino vino,
106 u64 offset, u64 *len, int op, int flags,
107 struct ceph_snap_context *snapc,
108 int do_sync, u32 truncate_seq,
109 u64 truncate_size,
110 struct timespec *mtime,
111 bool use_mempool, int num_reply);
112
113static inline void ceph_osdc_get_request(struct ceph_osd_request *req)
114{
115 atomic_inc(&req->r_ref);
116}
117extern void ceph_osdc_put_request(struct ceph_osd_request *req);
118
119extern int ceph_osdc_start_request(struct ceph_osd_client *osdc,
120 struct ceph_osd_request *req,
121 bool nofail);
122extern int ceph_osdc_wait_request(struct ceph_osd_client *osdc,
123 struct ceph_osd_request *req);
124extern void ceph_osdc_sync(struct ceph_osd_client *osdc);
125
126extern int ceph_osdc_readpages(struct ceph_osd_client *osdc,
127 struct ceph_vino vino,
128 struct ceph_file_layout *layout,
129 u64 off, u64 *plen,
130 u32 truncate_seq, u64 truncate_size,
131 struct page **pages, int nr_pages);
132
133extern int ceph_osdc_writepages(struct ceph_osd_client *osdc,
134 struct ceph_vino vino,
135 struct ceph_file_layout *layout,
136 struct ceph_snap_context *sc,
137 u64 off, u64 len,
138 u32 truncate_seq, u64 truncate_size,
139 struct timespec *mtime,
140 struct page **pages, int nr_pages,
141 int flags, int do_sync, bool nofail);
142
143#endif
144
diff --git a/fs/ceph/osdmap.c b/fs/ceph/osdmap.c
new file mode 100644
index 00000000000..e38fe6309b1
--- /dev/null
+++ b/fs/ceph/osdmap.c
@@ -0,0 +1,875 @@
1
2#include <asm/div64.h>
3
4#include "super.h"
5#include "osdmap.h"
6#include "crush/hash.h"
7#include "crush/mapper.h"
8#include "decode.h"
9#include "ceph_debug.h"
10
11char *ceph_osdmap_state_str(char *str, int len, int state)
12{
13 int flag = 0;
14
15 if (!len)
16 goto done;
17
18 *str = '\0';
19 if (state) {
20 if (state & CEPH_OSD_EXISTS) {
21 snprintf(str, len, "exists");
22 flag = 1;
23 }
24 if (state & CEPH_OSD_UP) {
25 snprintf(str, len, "%s%s%s", str, (flag ? ", " : ""),
26 "up");
27 flag = 1;
28 }
29 } else {
30 snprintf(str, len, "doesn't exist");
31 }
32done:
33 return str;
34}
35
36/* maps */
37
38static int calc_bits_of(unsigned t)
39{
40 int b = 0;
41 while (t) {
42 t = t >> 1;
43 b++;
44 }
45 return b;
46}
47
48/*
49 * the foo_mask is the smallest value 2^n-1 that is >= foo.
50 */
51static void calc_pg_masks(struct ceph_pg_pool_info *pi)
52{
53 pi->pg_num_mask = (1 << calc_bits_of(le32_to_cpu(pi->v.pg_num)-1)) - 1;
54 pi->pgp_num_mask =
55 (1 << calc_bits_of(le32_to_cpu(pi->v.pgp_num)-1)) - 1;
56 pi->lpg_num_mask =
57 (1 << calc_bits_of(le32_to_cpu(pi->v.lpg_num)-1)) - 1;
58 pi->lpgp_num_mask =
59 (1 << calc_bits_of(le32_to_cpu(pi->v.lpgp_num)-1)) - 1;
60}
61
62/*
63 * decode crush map
64 */
65static int crush_decode_uniform_bucket(void **p, void *end,
66 struct crush_bucket_uniform *b)
67{
68 dout("crush_decode_uniform_bucket %p to %p\n", *p, end);
69 ceph_decode_need(p, end, (1+b->h.size) * sizeof(u32), bad);
70 ceph_decode_32(p, b->item_weight);
71 return 0;
72bad:
73 return -EINVAL;
74}
75
76static int crush_decode_list_bucket(void **p, void *end,
77 struct crush_bucket_list *b)
78{
79 int j;
80 dout("crush_decode_list_bucket %p to %p\n", *p, end);
81 b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
82 if (b->item_weights == NULL)
83 return -ENOMEM;
84 b->sum_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
85 if (b->sum_weights == NULL)
86 return -ENOMEM;
87 ceph_decode_need(p, end, 2 * b->h.size * sizeof(u32), bad);
88 for (j = 0; j < b->h.size; j++) {
89 ceph_decode_32(p, b->item_weights[j]);
90 ceph_decode_32(p, b->sum_weights[j]);
91 }
92 return 0;
93bad:
94 return -EINVAL;
95}
96
97static int crush_decode_tree_bucket(void **p, void *end,
98 struct crush_bucket_tree *b)
99{
100 int j;
101 dout("crush_decode_tree_bucket %p to %p\n", *p, end);
102 ceph_decode_32_safe(p, end, b->num_nodes, bad);
103 b->node_weights = kcalloc(b->num_nodes, sizeof(u32), GFP_NOFS);
104 if (b->node_weights == NULL)
105 return -ENOMEM;
106 ceph_decode_need(p, end, b->num_nodes * sizeof(u32), bad);
107 for (j = 0; j < b->num_nodes; j++)
108 ceph_decode_32(p, b->node_weights[j]);
109 return 0;
110bad:
111 return -EINVAL;
112}
113
114static int crush_decode_straw_bucket(void **p, void *end,
115 struct crush_bucket_straw *b)
116{
117 int j;
118 dout("crush_decode_straw_bucket %p to %p\n", *p, end);
119 b->item_weights = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
120 if (b->item_weights == NULL)
121 return -ENOMEM;
122 b->straws = kcalloc(b->h.size, sizeof(u32), GFP_NOFS);
123 if (b->straws == NULL)
124 return -ENOMEM;
125 ceph_decode_need(p, end, 2 * b->h.size * sizeof(u32), bad);
126 for (j = 0; j < b->h.size; j++) {
127 ceph_decode_32(p, b->item_weights[j]);
128 ceph_decode_32(p, b->straws[j]);
129 }
130 return 0;
131bad:
132 return -EINVAL;
133}
134
135static struct crush_map *crush_decode(void *pbyval, void *end)
136{
137 struct crush_map *c;
138 int err = -EINVAL;
139 int i, j;
140 void **p = &pbyval;
141 void *start = pbyval;
142 u32 magic;
143
144 dout("crush_decode %p to %p len %d\n", *p, end, (int)(end - *p));
145
146 c = kzalloc(sizeof(*c), GFP_NOFS);
147 if (c == NULL)
148 return ERR_PTR(-ENOMEM);
149
150 ceph_decode_need(p, end, 4*sizeof(u32), bad);
151 ceph_decode_32(p, magic);
152 if (magic != CRUSH_MAGIC) {
153 pr_err("crush_decode magic %x != current %x\n",
154 (unsigned)magic, (unsigned)CRUSH_MAGIC);
155 goto bad;
156 }
157 ceph_decode_32(p, c->max_buckets);
158 ceph_decode_32(p, c->max_rules);
159 ceph_decode_32(p, c->max_devices);
160
161 c->device_parents = kcalloc(c->max_devices, sizeof(u32), GFP_NOFS);
162 if (c->device_parents == NULL)
163 goto badmem;
164 c->bucket_parents = kcalloc(c->max_buckets, sizeof(u32), GFP_NOFS);
165 if (c->bucket_parents == NULL)
166 goto badmem;
167
168 c->buckets = kcalloc(c->max_buckets, sizeof(*c->buckets), GFP_NOFS);
169 if (c->buckets == NULL)
170 goto badmem;
171 c->rules = kcalloc(c->max_rules, sizeof(*c->rules), GFP_NOFS);
172 if (c->rules == NULL)
173 goto badmem;
174
175 /* buckets */
176 for (i = 0; i < c->max_buckets; i++) {
177 int size = 0;
178 u32 alg;
179 struct crush_bucket *b;
180
181 ceph_decode_32_safe(p, end, alg, bad);
182 if (alg == 0) {
183 c->buckets[i] = NULL;
184 continue;
185 }
186 dout("crush_decode bucket %d off %x %p to %p\n",
187 i, (int)(*p-start), *p, end);
188
189 switch (alg) {
190 case CRUSH_BUCKET_UNIFORM:
191 size = sizeof(struct crush_bucket_uniform);
192 break;
193 case CRUSH_BUCKET_LIST:
194 size = sizeof(struct crush_bucket_list);
195 break;
196 case CRUSH_BUCKET_TREE:
197 size = sizeof(struct crush_bucket_tree);
198 break;
199 case CRUSH_BUCKET_STRAW:
200 size = sizeof(struct crush_bucket_straw);
201 break;
202 default:
203 goto bad;
204 }
205 BUG_ON(size == 0);
206 b = c->buckets[i] = kzalloc(size, GFP_NOFS);
207 if (b == NULL)
208 goto badmem;
209
210 ceph_decode_need(p, end, 4*sizeof(u32), bad);
211 ceph_decode_32(p, b->id);
212 ceph_decode_16(p, b->type);
213 ceph_decode_16(p, b->alg);
214 ceph_decode_32(p, b->weight);
215 ceph_decode_32(p, b->size);
216
217 dout("crush_decode bucket size %d off %x %p to %p\n",
218 b->size, (int)(*p-start), *p, end);
219
220 b->items = kcalloc(b->size, sizeof(__s32), GFP_NOFS);
221 if (b->items == NULL)
222 goto badmem;
223 b->perm = kcalloc(b->size, sizeof(u32), GFP_NOFS);
224 if (b->perm == NULL)
225 goto badmem;
226 b->perm_n = 0;
227
228 ceph_decode_need(p, end, b->size*sizeof(u32), bad);
229 for (j = 0; j < b->size; j++)
230 ceph_decode_32(p, b->items[j]);
231
232 switch (b->alg) {
233 case CRUSH_BUCKET_UNIFORM:
234 err = crush_decode_uniform_bucket(p, end,
235 (struct crush_bucket_uniform *)b);
236 if (err < 0)
237 goto bad;
238 break;
239 case CRUSH_BUCKET_LIST:
240 err = crush_decode_list_bucket(p, end,
241 (struct crush_bucket_list *)b);
242 if (err < 0)
243 goto bad;
244 break;
245 case CRUSH_BUCKET_TREE:
246 err = crush_decode_tree_bucket(p, end,
247 (struct crush_bucket_tree *)b);
248 if (err < 0)
249 goto bad;
250 break;
251 case CRUSH_BUCKET_STRAW:
252 err = crush_decode_straw_bucket(p, end,
253 (struct crush_bucket_straw *)b);
254 if (err < 0)
255 goto bad;
256 break;
257 }
258 }
259
260 /* rules */
261 dout("rule vec is %p\n", c->rules);
262 for (i = 0; i < c->max_rules; i++) {
263 u32 yes;
264 struct crush_rule *r;
265
266 ceph_decode_32_safe(p, end, yes, bad);
267 if (!yes) {
268 dout("crush_decode NO rule %d off %x %p to %p\n",
269 i, (int)(*p-start), *p, end);
270 c->rules[i] = NULL;
271 continue;
272 }
273
274 dout("crush_decode rule %d off %x %p to %p\n",
275 i, (int)(*p-start), *p, end);
276
277 /* len */
278 ceph_decode_32_safe(p, end, yes, bad);
279#if BITS_PER_LONG == 32
280 if (yes > ULONG_MAX / sizeof(struct crush_rule_step))
281 goto bad;
282#endif
283 r = c->rules[i] = kmalloc(sizeof(*r) +
284 yes*sizeof(struct crush_rule_step),
285 GFP_NOFS);
286 if (r == NULL)
287 goto badmem;
288 dout(" rule %d is at %p\n", i, r);
289 r->len = yes;
290 ceph_decode_copy_safe(p, end, &r->mask, 4, bad); /* 4 u8's */
291 ceph_decode_need(p, end, r->len*3*sizeof(u32), bad);
292 for (j = 0; j < r->len; j++) {
293 ceph_decode_32(p, r->steps[j].op);
294 ceph_decode_32(p, r->steps[j].arg1);
295 ceph_decode_32(p, r->steps[j].arg2);
296 }
297 }
298
299 /* ignore trailing name maps. */
300
301 dout("crush_decode success\n");
302 return c;
303
304badmem:
305 err = -ENOMEM;
306bad:
307 dout("crush_decode fail %d\n", err);
308 crush_destroy(c);
309 return ERR_PTR(err);
310}
311
312
313/*
314 * osd map
315 */
316void ceph_osdmap_destroy(struct ceph_osdmap *map)
317{
318 dout("osdmap_destroy %p\n", map);
319 if (map->crush)
320 crush_destroy(map->crush);
321 while (!RB_EMPTY_ROOT(&map->pg_temp))
322 rb_erase(rb_first(&map->pg_temp), &map->pg_temp);
323 kfree(map->osd_state);
324 kfree(map->osd_weight);
325 kfree(map->pg_pool);
326 kfree(map->osd_addr);
327 kfree(map);
328}
329
330/*
331 * adjust max osd value. reallocate arrays.
332 */
333static int osdmap_set_max_osd(struct ceph_osdmap *map, int max)
334{
335 u8 *state;
336 struct ceph_entity_addr *addr;
337 u32 *weight;
338
339 state = kcalloc(max, sizeof(*state), GFP_NOFS);
340 addr = kcalloc(max, sizeof(*addr), GFP_NOFS);
341 weight = kcalloc(max, sizeof(*weight), GFP_NOFS);
342 if (state == NULL || addr == NULL || weight == NULL) {
343 kfree(state);
344 kfree(addr);
345 kfree(weight);
346 return -ENOMEM;
347 }
348
349 /* copy old? */
350 if (map->osd_state) {
351 memcpy(state, map->osd_state, map->max_osd*sizeof(*state));
352 memcpy(addr, map->osd_addr, map->max_osd*sizeof(*addr));
353 memcpy(weight, map->osd_weight, map->max_osd*sizeof(*weight));
354 kfree(map->osd_state);
355 kfree(map->osd_addr);
356 kfree(map->osd_weight);
357 }
358
359 map->osd_state = state;
360 map->osd_weight = weight;
361 map->osd_addr = addr;
362 map->max_osd = max;
363 return 0;
364}
365
366/*
367 * Insert a new pg_temp mapping
368 */
369static void __insert_pg_mapping(struct ceph_pg_mapping *new,
370 struct rb_root *root)
371{
372 struct rb_node **p = &root->rb_node;
373 struct rb_node *parent = NULL;
374 struct ceph_pg_mapping *pg = NULL;
375
376 while (*p) {
377 parent = *p;
378 pg = rb_entry(parent, struct ceph_pg_mapping, node);
379 if (new->pgid < pg->pgid)
380 p = &(*p)->rb_left;
381 else if (new->pgid > pg->pgid)
382 p = &(*p)->rb_right;
383 else
384 BUG();
385 }
386
387 rb_link_node(&new->node, parent, p);
388 rb_insert_color(&new->node, root);
389}
390
391/*
392 * decode a full map.
393 */
394struct ceph_osdmap *osdmap_decode(void **p, void *end)
395{
396 struct ceph_osdmap *map;
397 u16 version;
398 u32 len, max, i;
399 int err = -EINVAL;
400 void *start = *p;
401
402 dout("osdmap_decode %p to %p len %d\n", *p, end, (int)(end - *p));
403
404 map = kzalloc(sizeof(*map), GFP_NOFS);
405 if (map == NULL)
406 return ERR_PTR(-ENOMEM);
407 map->pg_temp = RB_ROOT;
408
409 ceph_decode_16_safe(p, end, version, bad);
410
411 ceph_decode_need(p, end, 2*sizeof(u64)+6*sizeof(u32), bad);
412 ceph_decode_copy(p, &map->fsid, sizeof(map->fsid));
413 ceph_decode_32(p, map->epoch);
414 ceph_decode_copy(p, &map->created, sizeof(map->created));
415 ceph_decode_copy(p, &map->modified, sizeof(map->modified));
416
417 ceph_decode_32(p, map->num_pools);
418 map->pg_pool = kcalloc(map->num_pools, sizeof(*map->pg_pool),
419 GFP_NOFS);
420 if (!map->pg_pool) {
421 err = -ENOMEM;
422 goto bad;
423 }
424 ceph_decode_32_safe(p, end, max, bad);
425 while (max--) {
426 ceph_decode_need(p, end, 4+sizeof(map->pg_pool->v), bad);
427 ceph_decode_32(p, i);
428 if (i >= map->num_pools)
429 goto bad;
430 ceph_decode_copy(p, &map->pg_pool[i].v,
431 sizeof(map->pg_pool->v));
432 calc_pg_masks(&map->pg_pool[i]);
433 p += le32_to_cpu(map->pg_pool[i].v.num_snaps) * sizeof(u64);
434 p += le32_to_cpu(map->pg_pool[i].v.num_removed_snap_intervals)
435 * sizeof(u64) * 2;
436 }
437
438 ceph_decode_32_safe(p, end, map->flags, bad);
439
440 ceph_decode_32(p, max);
441
442 /* (re)alloc osd arrays */
443 err = osdmap_set_max_osd(map, max);
444 if (err < 0)
445 goto bad;
446 dout("osdmap_decode max_osd = %d\n", map->max_osd);
447
448 /* osds */
449 err = -EINVAL;
450 ceph_decode_need(p, end, 3*sizeof(u32) +
451 map->max_osd*(1 + sizeof(*map->osd_weight) +
452 sizeof(*map->osd_addr)), bad);
453 *p += 4; /* skip length field (should match max) */
454 ceph_decode_copy(p, map->osd_state, map->max_osd);
455
456 *p += 4; /* skip length field (should match max) */
457 for (i = 0; i < map->max_osd; i++)
458 ceph_decode_32(p, map->osd_weight[i]);
459
460 *p += 4; /* skip length field (should match max) */
461 ceph_decode_copy(p, map->osd_addr, map->max_osd*sizeof(*map->osd_addr));
462
463 /* pg_temp */
464 ceph_decode_32_safe(p, end, len, bad);
465 for (i = 0; i < len; i++) {
466 int n, j;
467 u64 pgid;
468 struct ceph_pg_mapping *pg;
469
470 ceph_decode_need(p, end, sizeof(u32) + sizeof(u64), bad);
471 ceph_decode_64(p, pgid);
472 ceph_decode_32(p, n);
473 ceph_decode_need(p, end, n * sizeof(u32), bad);
474 pg = kmalloc(sizeof(*pg) + n*sizeof(u32), GFP_NOFS);
475 if (!pg) {
476 err = -ENOMEM;
477 goto bad;
478 }
479 pg->pgid = pgid;
480 pg->len = n;
481 for (j = 0; j < n; j++)
482 ceph_decode_32(p, pg->osds[j]);
483
484 __insert_pg_mapping(pg, &map->pg_temp);
485 dout(" added pg_temp %llx len %d\n", pgid, len);
486 }
487
488 /* crush */
489 ceph_decode_32_safe(p, end, len, bad);
490 dout("osdmap_decode crush len %d from off 0x%x\n", len,
491 (int)(*p - start));
492 ceph_decode_need(p, end, len, bad);
493 map->crush = crush_decode(*p, end);
494 *p += len;
495 if (IS_ERR(map->crush)) {
496 err = PTR_ERR(map->crush);
497 map->crush = NULL;
498 goto bad;
499 }
500
501 /* ignore the rest of the map */
502 *p = end;
503
504 dout("osdmap_decode done %p %p\n", *p, end);
505 return map;
506
507bad:
508 dout("osdmap_decode fail\n");
509 ceph_osdmap_destroy(map);
510 return ERR_PTR(err);
511}
512
513/*
514 * decode and apply an incremental map update.
515 */
516struct ceph_osdmap *osdmap_apply_incremental(void **p, void *end,
517 struct ceph_osdmap *map,
518 struct ceph_messenger *msgr)
519{
520 struct ceph_osdmap *newmap = map;
521 struct crush_map *newcrush = NULL;
522 struct ceph_fsid fsid;
523 u32 epoch = 0;
524 struct ceph_timespec modified;
525 u32 len, pool;
526 __s32 new_flags, max;
527 void *start = *p;
528 int err = -EINVAL;
529 u16 version;
530 struct rb_node *rbp;
531
532 ceph_decode_16_safe(p, end, version, bad);
533
534 ceph_decode_need(p, end, sizeof(fsid)+sizeof(modified)+2*sizeof(u32),
535 bad);
536 ceph_decode_copy(p, &fsid, sizeof(fsid));
537 ceph_decode_32(p, epoch);
538 BUG_ON(epoch != map->epoch+1);
539 ceph_decode_copy(p, &modified, sizeof(modified));
540 ceph_decode_32(p, new_flags);
541
542 /* full map? */
543 ceph_decode_32_safe(p, end, len, bad);
544 if (len > 0) {
545 dout("apply_incremental full map len %d, %p to %p\n",
546 len, *p, end);
547 newmap = osdmap_decode(p, min(*p+len, end));
548 return newmap; /* error or not */
549 }
550
551 /* new crush? */
552 ceph_decode_32_safe(p, end, len, bad);
553 if (len > 0) {
554 dout("apply_incremental new crush map len %d, %p to %p\n",
555 len, *p, end);
556 newcrush = crush_decode(*p, min(*p+len, end));
557 if (IS_ERR(newcrush))
558 return ERR_PTR(PTR_ERR(newcrush));
559 }
560
561 /* new flags? */
562 if (new_flags >= 0)
563 map->flags = new_flags;
564
565 ceph_decode_need(p, end, 5*sizeof(u32), bad);
566
567 /* new max? */
568 ceph_decode_32(p, max);
569 if (max >= 0) {
570 err = osdmap_set_max_osd(map, max);
571 if (err < 0)
572 goto bad;
573 }
574
575 map->epoch++;
576 map->modified = map->modified;
577 if (newcrush) {
578 if (map->crush)
579 crush_destroy(map->crush);
580 map->crush = newcrush;
581 newcrush = NULL;
582 }
583
584 /* new_pool */
585 ceph_decode_32_safe(p, end, len, bad);
586 while (len--) {
587 ceph_decode_32_safe(p, end, pool, bad);
588 if (pool >= map->num_pools) {
589 void *pg_pool = kcalloc(pool + 1,
590 sizeof(*map->pg_pool),
591 GFP_NOFS);
592 if (!pg_pool) {
593 err = -ENOMEM;
594 goto bad;
595 }
596 memcpy(pg_pool, map->pg_pool,
597 map->num_pools * sizeof(*map->pg_pool));
598 kfree(map->pg_pool);
599 map->pg_pool = pg_pool;
600 map->num_pools = pool+1;
601 }
602 ceph_decode_copy(p, &map->pg_pool[pool].v,
603 sizeof(map->pg_pool->v));
604 calc_pg_masks(&map->pg_pool[pool]);
605 }
606
607 /* old_pool (ignore) */
608 ceph_decode_32_safe(p, end, len, bad);
609 *p += len * sizeof(u32);
610
611 /* new_up */
612 err = -EINVAL;
613 ceph_decode_32_safe(p, end, len, bad);
614 while (len--) {
615 u32 osd;
616 struct ceph_entity_addr addr;
617 ceph_decode_32_safe(p, end, osd, bad);
618 ceph_decode_copy_safe(p, end, &addr, sizeof(addr), bad);
619 pr_info("osd%d up\n", osd);
620 BUG_ON(osd >= map->max_osd);
621 map->osd_state[osd] |= CEPH_OSD_UP;
622 map->osd_addr[osd] = addr;
623 }
624
625 /* new_down */
626 ceph_decode_32_safe(p, end, len, bad);
627 while (len--) {
628 u32 osd;
629 ceph_decode_32_safe(p, end, osd, bad);
630 (*p)++; /* clean flag */
631 pr_info("ceph osd%d down\n", osd);
632 if (osd < map->max_osd)
633 map->osd_state[osd] &= ~CEPH_OSD_UP;
634 }
635
636 /* new_weight */
637 ceph_decode_32_safe(p, end, len, bad);
638 while (len--) {
639 u32 osd, off;
640 ceph_decode_need(p, end, sizeof(u32)*2, bad);
641 ceph_decode_32(p, osd);
642 ceph_decode_32(p, off);
643 pr_info("osd%d weight 0x%x %s\n", osd, off,
644 off == CEPH_OSD_IN ? "(in)" :
645 (off == CEPH_OSD_OUT ? "(out)" : ""));
646 if (osd < map->max_osd)
647 map->osd_weight[osd] = off;
648 }
649
650 /* new_pg_temp */
651 rbp = rb_first(&map->pg_temp);
652 ceph_decode_32_safe(p, end, len, bad);
653 while (len--) {
654 struct ceph_pg_mapping *pg;
655 int j;
656 u64 pgid;
657 u32 pglen;
658 ceph_decode_need(p, end, sizeof(u64) + sizeof(u32), bad);
659 ceph_decode_64(p, pgid);
660 ceph_decode_32(p, pglen);
661
662 /* remove any? */
663 while (rbp && rb_entry(rbp, struct ceph_pg_mapping,
664 node)->pgid <= pgid) {
665 struct rb_node *cur = rbp;
666 rbp = rb_next(rbp);
667 dout(" removed pg_temp %llx\n",
668 rb_entry(cur, struct ceph_pg_mapping, node)->pgid);
669 rb_erase(cur, &map->pg_temp);
670 }
671
672 if (pglen) {
673 /* insert */
674 ceph_decode_need(p, end, pglen*sizeof(u32), bad);
675 pg = kmalloc(sizeof(*pg) + sizeof(u32)*pglen, GFP_NOFS);
676 if (!pg) {
677 err = -ENOMEM;
678 goto bad;
679 }
680 pg->pgid = pgid;
681 pg->len = pglen;
682 for (j = 0; j < len; j++)
683 ceph_decode_32(p, pg->osds[j]);
684 __insert_pg_mapping(pg, &map->pg_temp);
685 dout(" added pg_temp %llx len %d\n", pgid, pglen);
686 }
687 }
688 while (rbp) {
689 struct rb_node *cur = rbp;
690 rbp = rb_next(rbp);
691 dout(" removed pg_temp %llx\n",
692 rb_entry(cur, struct ceph_pg_mapping, node)->pgid);
693 rb_erase(cur, &map->pg_temp);
694 }
695
696 /* ignore the rest */
697 *p = end;
698 return map;
699
700bad:
701 pr_err("corrupt inc osdmap epoch %d off %d (%p of %p-%p)\n",
702 epoch, (int)(*p - start), *p, start, end);
703 if (newcrush)
704 crush_destroy(newcrush);
705 return ERR_PTR(err);
706}
707
708
709
710
711/*
712 * calculate file layout from given offset, length.
713 * fill in correct oid, logical length, and object extent
714 * offset, length.
715 *
716 * for now, we write only a single su, until we can
717 * pass a stride back to the caller.
718 */
719void ceph_calc_file_object_mapping(struct ceph_file_layout *layout,
720 u64 off, u64 *plen,
721 u64 *bno,
722 u64 *oxoff, u64 *oxlen)
723{
724 u32 osize = le32_to_cpu(layout->fl_object_size);
725 u32 su = le32_to_cpu(layout->fl_stripe_unit);
726 u32 sc = le32_to_cpu(layout->fl_stripe_count);
727 u32 bl, stripeno, stripepos, objsetno;
728 u32 su_per_object;
729 u64 t;
730
731 dout("mapping %llu~%llu osize %u fl_su %u\n", off, *plen,
732 osize, su);
733 su_per_object = osize / le32_to_cpu(layout->fl_stripe_unit);
734 dout("osize %u / su %u = su_per_object %u\n", osize, su,
735 su_per_object);
736
737 BUG_ON((su & ~PAGE_MASK) != 0);
738 /* bl = *off / su; */
739 t = off;
740 do_div(t, su);
741 bl = t;
742 dout("off %llu / su %u = bl %u\n", off, su, bl);
743
744 stripeno = bl / sc;
745 stripepos = bl % sc;
746 objsetno = stripeno / su_per_object;
747
748 *bno = objsetno * sc + stripepos;
749 dout("objset %u * sc %u = bno %u\n", objsetno, sc, (unsigned)*bno);
750 /* *oxoff = *off / layout->fl_stripe_unit; */
751 t = off;
752 *oxoff = do_div(t, su);
753 *oxlen = min_t(u64, *plen, su - *oxoff);
754 *plen = *oxlen;
755
756 dout(" obj extent %llu~%llu\n", *oxoff, *oxlen);
757}
758
759/*
760 * calculate an object layout (i.e. pgid) from an oid,
761 * file_layout, and osdmap
762 */
763int ceph_calc_object_layout(struct ceph_object_layout *ol,
764 const char *oid,
765 struct ceph_file_layout *fl,
766 struct ceph_osdmap *osdmap)
767{
768 unsigned num, num_mask;
769 union ceph_pg pgid;
770 s32 preferred = (s32)le32_to_cpu(fl->fl_pg_preferred);
771 int poolid = le32_to_cpu(fl->fl_pg_pool);
772 struct ceph_pg_pool_info *pool;
773
774 if (poolid >= osdmap->num_pools)
775 return -EIO;
776 pool = &osdmap->pg_pool[poolid];
777
778 if (preferred >= 0) {
779 num = le32_to_cpu(pool->v.lpg_num);
780 num_mask = pool->lpg_num_mask;
781 } else {
782 num = le32_to_cpu(pool->v.pg_num);
783 num_mask = pool->pg_num_mask;
784 }
785
786 pgid.pg64 = 0; /* start with it zeroed out */
787 pgid.pg.ps = ceph_full_name_hash(oid, strlen(oid));
788 pgid.pg.preferred = preferred;
789 pgid.pg.pool = le32_to_cpu(fl->fl_pg_pool);
790 if (preferred >= 0)
791 dout("calc_object_layout '%s' pgid %d.%xp%d (%llx)\n", oid,
792 pgid.pg.pool, pgid.pg.ps, (int)preferred, pgid.pg64);
793 else
794 dout("calc_object_layout '%s' pgid %d.%x (%llx)\n", oid,
795 pgid.pg.pool, pgid.pg.ps, pgid.pg64);
796
797 ol->ol_pgid = cpu_to_le64(pgid.pg64);
798 ol->ol_stripe_unit = fl->fl_object_stripe_unit;
799
800 return 0;
801}
802
803/*
804 * Calculate raw osd vector for the given pgid. Return pointer to osd
805 * array, or NULL on failure.
806 */
807static int *calc_pg_raw(struct ceph_osdmap *osdmap, union ceph_pg pgid,
808 int *osds, int *num)
809{
810 struct rb_node *n = osdmap->pg_temp.rb_node;
811 struct ceph_pg_mapping *pg;
812 struct ceph_pg_pool_info *pool;
813 int ruleno;
814 unsigned pps; /* placement ps */
815
816 /* pg_temp? */
817 while (n) {
818 pg = rb_entry(n, struct ceph_pg_mapping, node);
819 if (pgid.pg64 < pg->pgid)
820 n = n->rb_left;
821 else if (pgid.pg64 > pg->pgid)
822 n = n->rb_right;
823 else {
824 *num = pg->len;
825 return pg->osds;
826 }
827 }
828
829 /* crush */
830 if (pgid.pg.pool >= osdmap->num_pools)
831 return NULL;
832 pool = &osdmap->pg_pool[pgid.pg.pool];
833 ruleno = crush_find_rule(osdmap->crush, pool->v.crush_ruleset,
834 pool->v.type, pool->v.size);
835 if (ruleno < 0) {
836 pr_err("no crush rule pool %d type %d size %d\n",
837 pgid.pg.pool, pool->v.type, pool->v.size);
838 return NULL;
839 }
840
841 if (pgid.pg.preferred >= 0)
842 pps = ceph_stable_mod(pgid.pg.ps,
843 le32_to_cpu(pool->v.lpgp_num),
844 pool->lpgp_num_mask);
845 else
846 pps = ceph_stable_mod(pgid.pg.ps,
847 le32_to_cpu(pool->v.pgp_num),
848 pool->pgp_num_mask);
849 pps += pgid.pg.pool;
850 *num = crush_do_rule(osdmap->crush, ruleno, pps, osds,
851 min_t(int, pool->v.size, *num),
852 pgid.pg.preferred, osdmap->osd_weight);
853 return osds;
854}
855
856/*
857 * Return primary osd for given pgid, or -1 if none.
858 */
859int ceph_calc_pg_primary(struct ceph_osdmap *osdmap, union ceph_pg pgid)
860{
861 int rawosds[10], *osds;
862 int i, num = ARRAY_SIZE(rawosds);
863
864 osds = calc_pg_raw(osdmap, pgid, rawosds, &num);
865 if (!osds)
866 return -1;
867
868 /* primary is first up osd */
869 for (i = 0; i < num; i++)
870 if (ceph_osd_is_up(osdmap, osds[i])) {
871 return osds[i];
872 break;
873 }
874 return -1;
875}
diff --git a/fs/ceph/osdmap.h b/fs/ceph/osdmap.h
new file mode 100644
index 00000000000..07127c6fb13
--- /dev/null
+++ b/fs/ceph/osdmap.h
@@ -0,0 +1,123 @@
1#ifndef _FS_CEPH_OSDMAP_H
2#define _FS_CEPH_OSDMAP_H
3
4#include <linux/rbtree.h>
5#include "types.h"
6#include "ceph_fs.h"
7#include "crush/crush.h"
8
9/*
10 * The osd map describes the current membership of the osd cluster and
11 * specifies the mapping of objects to placement groups and placement
12 * groups to (sets of) osds. That is, it completely specifies the
13 * (desired) distribution of all data objects in the system at some
14 * point in time.
15 *
16 * Each map version is identified by an epoch, which increases monotonically.
17 *
18 * The map can be updated either via an incremental map (diff) describing
19 * the change between two successive epochs, or as a fully encoded map.
20 */
21struct ceph_pg_pool_info {
22 struct ceph_pg_pool v;
23 int pg_num_mask, pgp_num_mask, lpg_num_mask, lpgp_num_mask;
24};
25
26struct ceph_pg_mapping {
27 struct rb_node node;
28 u64 pgid;
29 int len;
30 int osds[];
31};
32
33struct ceph_osdmap {
34 struct ceph_fsid fsid;
35 u32 epoch;
36 u32 mkfs_epoch;
37 struct ceph_timespec created, modified;
38
39 u32 flags; /* CEPH_OSDMAP_* */
40
41 u32 max_osd; /* size of osd_state, _offload, _addr arrays */
42 u8 *osd_state; /* CEPH_OSD_* */
43 u32 *osd_weight; /* 0 = failed, 0x10000 = 100% normal */
44 struct ceph_entity_addr *osd_addr;
45
46 struct rb_root pg_temp;
47
48 u32 num_pools;
49 struct ceph_pg_pool_info *pg_pool;
50
51 /* the CRUSH map specifies the mapping of placement groups to
52 * the list of osds that store+replicate them. */
53 struct crush_map *crush;
54};
55
56/*
57 * file layout helpers
58 */
59#define ceph_file_layout_su(l) ((__s32)le32_to_cpu((l).fl_stripe_unit))
60#define ceph_file_layout_stripe_count(l) \
61 ((__s32)le32_to_cpu((l).fl_stripe_count))
62#define ceph_file_layout_object_size(l) ((__s32)le32_to_cpu((l).fl_object_size))
63#define ceph_file_layout_cas_hash(l) ((__s32)le32_to_cpu((l).fl_cas_hash))
64#define ceph_file_layout_object_su(l) \
65 ((__s32)le32_to_cpu((l).fl_object_stripe_unit))
66#define ceph_file_layout_pg_preferred(l) \
67 ((__s32)le32_to_cpu((l).fl_pg_preferred))
68#define ceph_file_layout_pg_pool(l) \
69 ((__s32)le32_to_cpu((l).fl_pg_pool))
70
71static inline unsigned ceph_file_layout_stripe_width(struct ceph_file_layout *l)
72{
73 return le32_to_cpu(l->fl_stripe_unit) *
74 le32_to_cpu(l->fl_stripe_count);
75}
76
77/* "period" == bytes before i start on a new set of objects */
78static inline unsigned ceph_file_layout_period(struct ceph_file_layout *l)
79{
80 return le32_to_cpu(l->fl_object_size) *
81 le32_to_cpu(l->fl_stripe_count);
82}
83
84
85static inline int ceph_osd_is_up(struct ceph_osdmap *map, int osd)
86{
87 return (osd < map->max_osd) && (map->osd_state[osd] & CEPH_OSD_UP);
88}
89
90static inline bool ceph_osdmap_flag(struct ceph_osdmap *map, int flag)
91{
92 return map && (map->flags & flag);
93}
94
95extern char *ceph_osdmap_state_str(char *str, int len, int state);
96
97static inline struct ceph_entity_addr *ceph_osd_addr(struct ceph_osdmap *map,
98 int osd)
99{
100 if (osd >= map->max_osd)
101 return NULL;
102 return &map->osd_addr[osd];
103}
104
105extern struct ceph_osdmap *osdmap_decode(void **p, void *end);
106extern struct ceph_osdmap *osdmap_apply_incremental(void **p, void *end,
107 struct ceph_osdmap *map,
108 struct ceph_messenger *msgr);
109extern void ceph_osdmap_destroy(struct ceph_osdmap *map);
110
111/* calculate mapping of a file extent to an object */
112extern void ceph_calc_file_object_mapping(struct ceph_file_layout *layout,
113 u64 off, u64 *plen,
114 u64 *bno, u64 *oxoff, u64 *oxlen);
115
116/* calculate mapping of object to a placement group */
117extern int ceph_calc_object_layout(struct ceph_object_layout *ol,
118 const char *oid,
119 struct ceph_file_layout *fl,
120 struct ceph_osdmap *osdmap);
121extern int ceph_calc_pg_primary(struct ceph_osdmap *osdmap, union ceph_pg pgid);
122
123#endif
generated by cgit v1.2.2 (git 2.25.0) at 2024-12-28 12:30:48 -0500