diff options
author | Linus Torvalds <torvalds@linux-foundation.org> | 2016-10-10 13:52:05 -0700 |
---|---|---|
committer | Linus Torvalds <torvalds@linux-foundation.org> | 2016-10-10 13:52:05 -0700 |
commit | 8dfb790b15e779232d5d4e3f0102af2bea21ca55 (patch) | |
tree | 7208241fc93d39f769dcec0c227c8582f117dfce | |
parent | fed41f7d039bad02f94cad9059e4b14cd81d13f2 (diff) | |
parent | 64f77566e1c84990d6c448bb3960f899521c0b7d (diff) |
Merge tag 'ceph-for-4.9-rc1' of git://github.com/ceph/ceph-client
Pull Ceph updates from Ilya Dryomov:
"The big ticket item here is support for rbd exclusive-lock feature,
with maintenance operations offloaded to userspace (Douglas Fuller,
Mike Christie and myself). Another block device bullet is a series
fixing up layering error paths (myself).
On the filesystem side, we've got patches that improve our handling of
buffered vs dio write races (Neil Brown) and a few assorted fixes from
Zheng. Also included a couple of random cleanups and a minor CRUSH
update"
* tag 'ceph-for-4.9-rc1' of git://github.com/ceph/ceph-client: (39 commits)
crush: remove redundant local variable
crush: don't normalize input of crush_ln iteratively
libceph: ceph_build_auth() doesn't need ceph_auth_build_hello()
libceph: use CEPH_AUTH_UNKNOWN in ceph_auth_build_hello()
ceph: fix description for rsize and rasize mount options
rbd: use kmalloc_array() in rbd_header_from_disk()
ceph: use list_move instead of list_del/list_add
ceph: handle CEPH_SESSION_REJECT message
ceph: avoid accessing / when mounting a subpath
ceph: fix mandatory flock check
ceph: remove warning when ceph_releasepage() is called on dirty page
ceph: ignore error from invalidate_inode_pages2_range() in direct write
ceph: fix error handling of start_read()
rbd: add rbd_obj_request_error() helper
rbd: img_data requests don't own their page array
rbd: don't call rbd_osd_req_format_read() for !img_data requests
rbd: rework rbd_img_obj_exists_submit() error paths
rbd: don't crash or leak on errors in rbd_img_obj_parent_read_full_callback()
rbd: move bumping img_request refcount into rbd_obj_request_submit()
rbd: mark the original request as done if stat request fails
...
-rw-r--r-- | Documentation/ABI/testing/sysfs-bus-rbd | 29 | ||||
-rw-r--r-- | Documentation/filesystems/ceph.txt | 4 | ||||
-rw-r--r-- | drivers/block/rbd.c | 1432 | ||||
-rw-r--r-- | drivers/block/rbd_types.h | 11 | ||||
-rw-r--r-- | fs/ceph/addr.c | 24 | ||||
-rw-r--r-- | fs/ceph/file.c | 4 | ||||
-rw-r--r-- | fs/ceph/locks.c | 4 | ||||
-rw-r--r-- | fs/ceph/mds_client.c | 30 | ||||
-rw-r--r-- | fs/ceph/mds_client.h | 1 | ||||
-rw-r--r-- | fs/ceph/strings.c | 2 | ||||
-rw-r--r-- | fs/ceph/super.c | 49 | ||||
-rw-r--r-- | include/linux/ceph/auth.h | 2 | ||||
-rw-r--r-- | include/linux/ceph/ceph_fs.h | 12 | ||||
-rw-r--r-- | include/linux/ceph/cls_lock_client.h | 49 | ||||
-rw-r--r-- | include/linux/ceph/libceph.h | 3 | ||||
-rw-r--r-- | include/linux/ceph/mon_client.h | 3 | ||||
-rw-r--r-- | include/linux/ceph/osd_client.h | 23 | ||||
-rw-r--r-- | net/ceph/Makefile | 1 | ||||
-rw-r--r-- | net/ceph/auth.c | 13 | ||||
-rw-r--r-- | net/ceph/auth_none.c | 2 | ||||
-rw-r--r-- | net/ceph/ceph_common.c | 13 | ||||
-rw-r--r-- | net/ceph/ceph_strings.c | 1 | ||||
-rw-r--r-- | net/ceph/cls_lock_client.c | 325 | ||||
-rw-r--r-- | net/ceph/crush/mapper.c | 17 | ||||
-rw-r--r-- | net/ceph/mon_client.c | 82 | ||||
-rw-r--r-- | net/ceph/osd_client.c | 169 |
26 files changed, 1966 insertions, 339 deletions
diff --git a/Documentation/ABI/testing/sysfs-bus-rbd b/Documentation/ABI/testing/sysfs-bus-rbd index 2ddd680929d8..f208ac58d613 100644 --- a/Documentation/ABI/testing/sysfs-bus-rbd +++ b/Documentation/ABI/testing/sysfs-bus-rbd @@ -6,7 +6,7 @@ Description: Being used for adding and removing rbd block devices. -Usage: <mon ip addr> <options> <pool name> <rbd image name> [snap name] +Usage: <mon ip addr> <options> <pool name> <rbd image name> [<snap name>] $ echo "192.168.0.1 name=admin rbd foo" > /sys/bus/rbd/add @@ -14,9 +14,13 @@ The snapshot name can be "-" or omitted to map the image read/write. A <dev-id> will be assigned for any registered block device. If snapshot is used, it will be mapped read-only. -Removal of a device: +Usage: <dev-id> [force] - $ echo <dev-id> > /sys/bus/rbd/remove + $ echo 2 > /sys/bus/rbd/remove + +Optional "force" argument which when passed will wait for running requests and +then unmap the image. Requests sent to the driver after initiating the removal +will be failed. (August 2016, since 4.9.) What: /sys/bus/rbd/add_single_major Date: December 2013 @@ -43,10 +47,25 @@ Description: Available only if rbd module is inserted with single_major Entries under /sys/bus/rbd/devices/<dev-id>/ -------------------------------------------- +client_addr + + The ceph unique client entity_addr_t (address + nonce). + The format is <address>:<port>/<nonce>: '1.2.3.4:1234/5678' or + '[1:2:3:4:5:6:7:8]:1234/5678'. (August 2016, since 4.9.) + client_id The ceph unique client id that was assigned for this specific session. +cluster_fsid + + The ceph cluster UUID. (August 2016, since 4.9.) + +config_info + + The string written into /sys/bus/rbd/add{,_single_major}. (August + 2016, since 4.9.) + features A hexadecimal encoding of the feature bits for this image. @@ -92,6 +111,10 @@ current_snap The current snapshot for which the device is mapped. +snap_id + + The current snapshot's id. (August 2016, since 4.9.) + parent Information identifying the chain of parent images in a layered rbd diff --git a/Documentation/filesystems/ceph.txt b/Documentation/filesystems/ceph.txt index d6030aa33376..f5306ee40ea9 100644 --- a/Documentation/filesystems/ceph.txt +++ b/Documentation/filesystems/ceph.txt @@ -98,6 +98,10 @@ Mount Options size. rsize=X + Specify the maximum read size in bytes. By default there is no + maximum. + + rasize=X Specify the maximum readahead. mount_timeout=X diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c index c1f84df7838b..abb71628ab61 100644 --- a/drivers/block/rbd.c +++ b/drivers/block/rbd.c @@ -31,6 +31,7 @@ #include <linux/ceph/libceph.h> #include <linux/ceph/osd_client.h> #include <linux/ceph/mon_client.h> +#include <linux/ceph/cls_lock_client.h> #include <linux/ceph/decode.h> #include <linux/parser.h> #include <linux/bsearch.h> @@ -114,12 +115,17 @@ static int atomic_dec_return_safe(atomic_t *v) #define RBD_OBJ_PREFIX_LEN_MAX 64 +#define RBD_NOTIFY_TIMEOUT 5 /* seconds */ +#define RBD_RETRY_DELAY msecs_to_jiffies(1000) + /* Feature bits */ #define RBD_FEATURE_LAYERING (1<<0) #define RBD_FEATURE_STRIPINGV2 (1<<1) -#define RBD_FEATURES_ALL \ - (RBD_FEATURE_LAYERING | RBD_FEATURE_STRIPINGV2) +#define RBD_FEATURE_EXCLUSIVE_LOCK (1<<2) +#define RBD_FEATURES_ALL (RBD_FEATURE_LAYERING | \ + RBD_FEATURE_STRIPINGV2 | \ + RBD_FEATURE_EXCLUSIVE_LOCK) /* Features supported by this (client software) implementation. */ @@ -128,11 +134,8 @@ static int atomic_dec_return_safe(atomic_t *v) /* * An RBD device name will be "rbd#", where the "rbd" comes from * RBD_DRV_NAME above, and # is a unique integer identifier. - * MAX_INT_FORMAT_WIDTH is used in ensuring DEV_NAME_LEN is big - * enough to hold all possible device names. */ #define DEV_NAME_LEN 32 -#define MAX_INT_FORMAT_WIDTH ((5 * sizeof (int)) / 2 + 1) /* * block device image metadata (in-memory version) @@ -322,6 +325,24 @@ struct rbd_img_request { #define for_each_obj_request_safe(ireq, oreq, n) \ list_for_each_entry_safe_reverse(oreq, n, &(ireq)->obj_requests, links) +enum rbd_watch_state { + RBD_WATCH_STATE_UNREGISTERED, + RBD_WATCH_STATE_REGISTERED, + RBD_WATCH_STATE_ERROR, +}; + +enum rbd_lock_state { + RBD_LOCK_STATE_UNLOCKED, + RBD_LOCK_STATE_LOCKED, + RBD_LOCK_STATE_RELEASING, +}; + +/* WatchNotify::ClientId */ +struct rbd_client_id { + u64 gid; + u64 handle; +}; + struct rbd_mapping { u64 size; u64 features; @@ -349,13 +370,29 @@ struct rbd_device { unsigned long flags; /* possibly lock protected */ struct rbd_spec *spec; struct rbd_options *opts; + char *config_info; /* add{,_single_major} string */ struct ceph_object_id header_oid; struct ceph_object_locator header_oloc; - struct ceph_file_layout layout; + struct ceph_file_layout layout; /* used for all rbd requests */ + struct mutex watch_mutex; + enum rbd_watch_state watch_state; struct ceph_osd_linger_request *watch_handle; + u64 watch_cookie; + struct delayed_work watch_dwork; + + struct rw_semaphore lock_rwsem; + enum rbd_lock_state lock_state; + struct rbd_client_id owner_cid; + struct work_struct acquired_lock_work; + struct work_struct released_lock_work; + struct delayed_work lock_dwork; + struct work_struct unlock_work; + wait_queue_head_t lock_waitq; + + struct workqueue_struct *task_wq; struct rbd_spec *parent_spec; u64 parent_overlap; @@ -439,6 +476,29 @@ static int minor_to_rbd_dev_id(int minor) return minor >> RBD_SINGLE_MAJOR_PART_SHIFT; } +static bool rbd_is_lock_supported(struct rbd_device *rbd_dev) +{ + return (rbd_dev->header.features & RBD_FEATURE_EXCLUSIVE_LOCK) && + rbd_dev->spec->snap_id == CEPH_NOSNAP && + !rbd_dev->mapping.read_only; +} + +static bool __rbd_is_lock_owner(struct rbd_device *rbd_dev) +{ + return rbd_dev->lock_state == RBD_LOCK_STATE_LOCKED || + rbd_dev->lock_state == RBD_LOCK_STATE_RELEASING; +} + +static bool rbd_is_lock_owner(struct rbd_device *rbd_dev) +{ + bool is_lock_owner; + + down_read(&rbd_dev->lock_rwsem); + is_lock_owner = __rbd_is_lock_owner(rbd_dev); + up_read(&rbd_dev->lock_rwsem); + return is_lock_owner; +} + static BUS_ATTR(add, S_IWUSR, NULL, rbd_add); static BUS_ATTR(remove, S_IWUSR, NULL, rbd_remove); static BUS_ATTR(add_single_major, S_IWUSR, NULL, rbd_add_single_major); @@ -735,6 +795,7 @@ enum { /* string args above */ Opt_read_only, Opt_read_write, + Opt_lock_on_read, Opt_err }; @@ -746,16 +807,19 @@ static match_table_t rbd_opts_tokens = { {Opt_read_only, "ro"}, /* Alternate spelling */ {Opt_read_write, "read_write"}, {Opt_read_write, "rw"}, /* Alternate spelling */ + {Opt_lock_on_read, "lock_on_read"}, {Opt_err, NULL} }; struct rbd_options { int queue_depth; bool read_only; + bool lock_on_read; }; #define RBD_QUEUE_DEPTH_DEFAULT BLKDEV_MAX_RQ #define RBD_READ_ONLY_DEFAULT false +#define RBD_LOCK_ON_READ_DEFAULT false static int parse_rbd_opts_token(char *c, void *private) { @@ -791,6 +855,9 @@ static int parse_rbd_opts_token(char *c, void *private) case Opt_read_write: rbd_opts->read_only = false; break; + case Opt_lock_on_read: + rbd_opts->lock_on_read = true; + break; default: /* libceph prints "bad option" msg */ return -EINVAL; @@ -919,7 +986,6 @@ static int rbd_header_from_disk(struct rbd_device *rbd_dev, char *snap_names = NULL; u64 *snap_sizes = NULL; u32 snap_count; - size_t size; int ret = -ENOMEM; u32 i; @@ -957,9 +1023,9 @@ static int rbd_header_from_disk(struct rbd_device *rbd_dev, goto out_err; /* ...as well as the array of their sizes. */ - - size = snap_count * sizeof (*header->snap_sizes); - snap_sizes = kmalloc(size, GFP_KERNEL); + snap_sizes = kmalloc_array(snap_count, + sizeof(*header->snap_sizes), + GFP_KERNEL); if (!snap_sizes) goto out_err; @@ -1551,11 +1617,18 @@ static bool obj_request_type_valid(enum obj_request_type type) } } -static int rbd_obj_request_submit(struct ceph_osd_client *osdc, - struct rbd_obj_request *obj_request) +static void rbd_img_obj_callback(struct rbd_obj_request *obj_request); + +static void rbd_obj_request_submit(struct rbd_obj_request *obj_request) { - dout("%s %p\n", __func__, obj_request); - return ceph_osdc_start_request(osdc, obj_request->osd_req, false); + struct ceph_osd_request *osd_req = obj_request->osd_req; + + dout("%s %p osd_req %p\n", __func__, obj_request, osd_req); + if (obj_request_img_data_test(obj_request)) { + WARN_ON(obj_request->callback != rbd_img_obj_callback); + rbd_img_request_get(obj_request->img_request); + } + ceph_osdc_start_request(osd_req->r_osdc, osd_req, false); } static void rbd_obj_request_end(struct rbd_obj_request *obj_request) @@ -1745,6 +1818,22 @@ static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) complete_all(&obj_request->completion); } +static void rbd_obj_request_error(struct rbd_obj_request *obj_request, int err) +{ + obj_request->result = err; + obj_request->xferred = 0; + /* + * kludge - mirror rbd_obj_request_submit() to match a put in + * rbd_img_obj_callback() + */ + if (obj_request_img_data_test(obj_request)) { + WARN_ON(obj_request->callback != rbd_img_obj_callback); + rbd_img_request_get(obj_request->img_request); + } + obj_request_done_set(obj_request); + rbd_obj_request_complete(obj_request); +} + static void rbd_osd_read_callback(struct rbd_obj_request *obj_request) { struct rbd_img_request *img_request = NULL; @@ -1877,11 +1966,10 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req) static void rbd_osd_req_format_read(struct rbd_obj_request *obj_request) { - struct rbd_img_request *img_request = obj_request->img_request; struct ceph_osd_request *osd_req = obj_request->osd_req; - if (img_request) - osd_req->r_snapid = img_request->snap_id; + rbd_assert(obj_request_img_data_test(obj_request)); + osd_req->r_snapid = obj_request->img_request->snap_id; } static void rbd_osd_req_format_write(struct rbd_obj_request *obj_request) @@ -2074,7 +2162,9 @@ static void rbd_obj_request_destroy(struct kref *kref) bio_chain_put(obj_request->bio_list); break; case OBJ_REQUEST_PAGES: - if (obj_request->pages) + /* img_data requests don't own their page array */ + if (obj_request->pages && + !obj_request_img_data_test(obj_request)) ceph_release_page_vector(obj_request->pages, obj_request->page_count); break; @@ -2295,13 +2385,6 @@ static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request) xferred = obj_request->length; } - /* Image object requests don't own their page array */ - - if (obj_request->type == OBJ_REQUEST_PAGES) { - obj_request->pages = NULL; - obj_request->page_count = 0; - } - if (img_request_child_test(img_request)) { rbd_assert(img_request->obj_request != NULL); more = obj_request->which < img_request->obj_request_count - 1; @@ -2520,8 +2603,6 @@ static int rbd_img_request_fill(struct rbd_img_request *img_request, rbd_img_obj_request_fill(obj_request, osd_req, op_type, 0); - rbd_img_request_get(img_request); - img_offset += length; resid -= length; } @@ -2579,7 +2660,6 @@ rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request) { struct rbd_obj_request *orig_request; struct ceph_osd_request *osd_req; - struct ceph_osd_client *osdc; struct rbd_device *rbd_dev; struct page **pages; enum obj_operation_type op_type; @@ -2603,7 +2683,7 @@ rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request) rbd_assert(obj_request_type_valid(orig_request->type)); img_result = img_request->result; parent_length = img_request->length; - rbd_assert(parent_length == img_request->xferred); + rbd_assert(img_result || parent_length == img_request->xferred); rbd_img_request_put(img_request); rbd_assert(orig_request->img_request); @@ -2616,13 +2696,9 @@ rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request) * and re-submit the original write request. */ if (!rbd_dev->parent_overlap) { - struct ceph_osd_client *osdc; - ceph_release_page_vector(pages, page_count); - osdc = &rbd_dev->rbd_client->client->osdc; - img_result = rbd_obj_request_submit(osdc, orig_request); - if (!img_result) - return; + rbd_obj_request_submit(orig_request); + return; } if (img_result) @@ -2656,17 +2732,12 @@ rbd_img_obj_parent_read_full_callback(struct rbd_img_request *img_request) /* All set, send it off. */ - osdc = &rbd_dev->rbd_client->client->osdc; - img_result = rbd_obj_request_submit(osdc, orig_request); - if (!img_result) - return; -out_err: - /* Record the error code and complete the request */ + rbd_obj_request_submit(orig_request); + return; - orig_request->result = img_result; - orig_request->xferred = 0; - obj_request_done_set(orig_request); - rbd_obj_request_complete(orig_request); +out_err: + ceph_release_page_vector(pages, page_count); + rbd_obj_request_error(orig_request, img_result); } /* @@ -2680,26 +2751,19 @@ out_err: * When the read completes, this page array will be transferred to * the original object request for the copyup operation. * - * If an error occurs, record it as the result of the original - * object request and mark it done so it gets completed. + * If an error occurs, it is recorded as the result of the original + * object request in rbd_img_obj_exists_callback(). */ static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request) { - struct rbd_img_request *img_request = NULL; + struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev; struct rbd_img_request *parent_request = NULL; - struct rbd_device *rbd_dev; u64 img_offset; u64 length; struct page **pages = NULL; u32 page_count; int result; - rbd_assert(obj_request_img_data_test(obj_request)); - rbd_assert(obj_request_type_valid(obj_request->type)); - - img_request = obj_request->img_request; - rbd_assert(img_request != NULL); - rbd_dev = img_request->rbd_dev; rbd_assert(rbd_dev->parent != NULL); /* @@ -2740,10 +2804,11 @@ static int rbd_img_obj_parent_read_full(struct rbd_obj_request *obj_request) result = rbd_img_request_fill(parent_request, OBJ_REQUEST_PAGES, pages); if (result) goto out_err; + parent_request->copyup_pages = pages; parent_request->copyup_page_count = page_count; - parent_request->callback = rbd_img_obj_parent_read_full_callback; + result = rbd_img_request_submit(parent_request); if (!result) return 0; @@ -2757,10 +2822,6 @@ out_err: ceph_release_page_vector(pages, page_count); if (parent_request) rbd_img_request_put(parent_request); - obj_request->result = result; - obj_request->xferred = 0; - obj_request_done_set(obj_request); - return result; } @@ -2793,17 +2854,13 @@ static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request) /* * If the overlap has become 0 (most likely because the - * image has been flattened) we need to free the pages - * and re-submit the original write request. + * image has been flattened) we need to re-submit the + * original request. */ rbd_dev = orig_request->img_request->rbd_dev; if (!rbd_dev->parent_overlap) { - struct ceph_osd_client *osdc; - - osdc = &rbd_dev->rbd_client->client->osdc; - result = rbd_obj_request_submit(osdc, orig_request); - if (!result) - return; + rbd_obj_request_submit(orig_request); + return; } /* @@ -2816,31 +2873,45 @@ static void rbd_img_obj_exists_callback(struct rbd_obj_request *obj_request) obj_request_existence_set(orig_request, true); } else if (result == -ENOENT) { obj_request_existence_set(orig_request, false); - } else if (result) { - orig_request->result = result; - goto out; + } else { + goto fail_orig_request; } /* * Resubmit the original request now that we have recorded * whether the target object exists. */ - orig_request->result = rbd_img_obj_request_submit(orig_request); -out: - if (orig_request->result) - rbd_obj_request_complete(orig_request); + result = rbd_img_obj_request_submit(orig_request); + if (result) + goto fail_orig_request; + + return; + +fail_orig_request: + rbd_obj_request_error(orig_request, result); } static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request) { + struct rbd_device *rbd_dev = obj_request->img_request->rbd_dev; struct rbd_obj_request *stat_request; - struct rbd_device *rbd_dev; - struct ceph_osd_client *osdc; - struct page **pages = NULL; + struct page **pages; u32 page_count; size_t size; int ret; + stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0, + OBJ_REQUEST_PAGES); + if (!stat_request) + return -ENOMEM; + + stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1, + stat_request); + if (!stat_request->osd_req) { + ret = -ENOMEM; + goto fail_stat_request; + } + /* * The response data for a STAT call consists of: * le64 length; @@ -2852,52 +2923,33 @@ static int rbd_img_obj_exists_submit(struct rbd_obj_request *obj_request) size = sizeof (__le64) + sizeof (__le32) + sizeof (__le32); page_count = (u32)calc_pages_for(0, size); pages = ceph_alloc_page_vector(page_count, GFP_KERNEL); - if (IS_ERR(pages)) - return PTR_ERR(pages); + if (IS_ERR(pages)) { + ret = PTR_ERR(pages); + goto fail_stat_request; + } - ret = -ENOMEM; - stat_request = rbd_obj_request_create(obj_request->object_name, 0, 0, - OBJ_REQUEST_PAGES); - if (!stat_request) - goto out; + osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT, 0); + osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0, + false, false); rbd_obj_request_get(obj_request); stat_request->obj_request = obj_request; stat_request->pages = pages; stat_request->page_count = page_count; - - rbd_assert(obj_request->img_request); - rbd_dev = obj_request->img_request->rbd_dev; - stat_request->osd_req = rbd_osd_req_create(rbd_dev, OBJ_OP_READ, 1, - stat_request); - if (!stat_request->osd_req) - goto out; stat_request->callback = rbd_img_obj_exists_callback; - osd_req_op_init(stat_request->osd_req, 0, CEPH_OSD_OP_STAT, 0); - osd_req_op_raw_data_in_pages(stat_request->osd_req, 0, pages, size, 0, - false, false); - rbd_osd_req_format_read(stat_request); - - osdc = &rbd_dev->rbd_client->client->osdc; - ret = rbd_obj_request_submit(osdc, stat_request); -out: - if (ret) - rbd_obj_request_put(obj_request); + rbd_obj_request_submit(stat_request); + return 0; +fail_stat_request: + rbd_obj_request_put(stat_request); return ret; } static bool img_obj_request_simple(struct rbd_obj_request *obj_request) { - struct rbd_img_request *img_request; - struct rbd_device *rbd_dev; - - rbd_assert(obj_request_img_data_test(obj_request)); - - img_request = obj_request->img_request; - rbd_assert(img_request); - rbd_dev = img_request->rbd_dev; + struct rbd_img_request *img_request = obj_request->img_request; + struct rbd_device *rbd_dev = img_request->rbd_dev; /* Reads */ if (!img_request_write_test(img_request) && @@ -2936,14 +2988,13 @@ static bool img_obj_request_simple(struct rbd_obj_request *obj_request) static int rbd_img_obj_request_submit(struct rbd_obj_request *obj_request) { - if (img_obj_request_simple(obj_request)) { - struct rbd_device *rbd_dev; - struct ceph_osd_client *osdc; - - rbd_dev = obj_request->img_request->rbd_dev; - osdc = &rbd_dev->rbd_client->client->osdc; + rbd_assert(obj_request_img_data_test(obj_request)); + rbd_assert(obj_request_type_valid(obj_request->type)); + rbd_assert(obj_request->img_request); - return rbd_obj_request_submit(osdc, obj_request); + if (img_obj_request_simple(obj_request)) { + rbd_obj_request_submit(obj_request); + return 0; } /* @@ -3006,12 +3057,8 @@ static void rbd_img_parent_read_callback(struct rbd_img_request *img_request) rbd_assert(obj_request->img_request); rbd_dev = obj_request->img_request->rbd_dev; if (!rbd_dev->parent_overlap) { - struct ceph_osd_client *osdc; - - osdc = &rbd_dev->rbd_client->client->osdc; - img_result = rbd_obj_request_submit(osdc, obj_request); - if (!img_result) - return; + rbd_obj_request_submit(obj_request); + return; } obj_request->result = img_result; @@ -3084,65 +3131,724 @@ out_err: obj_request_done_set(obj_request); } -static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev); -static void __rbd_dev_header_unwatch_sync(struct rbd_device *rbd_dev); +static const struct rbd_client_id rbd_empty_cid; -static void rbd_watch_cb(void *arg, u64 notify_id, u64 cookie, - u64 notifier_id, void *data, size_t data_len) +static bool rbd_cid_equal(const struct rbd_client_id *lhs, + const struct rbd_client_id *rhs) +{ + return lhs->gid == rhs->gid && lhs->handle == rhs->handle; +} + +static struct rbd_client_id rbd_get_cid(struct rbd_device *rbd_dev) +{ + struct rbd_client_id cid; + + mutex_lock(&rbd_dev->watch_mutex); + cid.gid = ceph_client_gid(rbd_dev->rbd_client->client); + cid.handle = rbd_dev->watch_cookie; + mutex_unlock(&rbd_dev->watch_mutex); + return cid; +} + +/* + * lock_rwsem must be held for write + */ +static void rbd_set_owner_cid(struct rbd_device *rbd_dev, + const struct rbd_client_id *cid) +{ + dout("%s rbd_dev %p %llu-%llu -> %llu-%llu\n", __func__, rbd_dev, + rbd_dev->owner_cid.gid, rbd_dev->owner_cid.handle, + cid->gid, cid->handle); + rbd_dev->owner_cid = *cid; /* struct */ +} + +static void format_lock_cookie(struct rbd_device *rbd_dev, char *buf) +{ + mutex_lock(&rbd_dev->watch_mutex); + sprintf(buf, "%s %llu", RBD_LOCK_COOKIE_PREFIX, rbd_dev->watch_cookie); + mutex_unlock(&rbd_dev->watch_mutex); +} + +/* + * lock_rwsem must be held for write + */ +static int rbd_lock(struct rbd_device *rbd_dev) { - struct rbd_device *rbd_dev = arg; struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; + struct rbd_client_id cid = rbd_get_cid(rbd_dev); + char cookie[32]; + int ret; + + WARN_ON(__rbd_is_lock_owner(rbd_dev)); + + format_lock_cookie(rbd_dev, cookie); + ret = ceph_cls_lock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc, + RBD_LOCK_NAME, CEPH_CLS_LOCK_EXCLUSIVE, cookie, + RBD_LOCK_TAG, "", 0); + if (ret) + return ret; + + rbd_dev->lock_state = RBD_LOCK_STATE_LOCKED; + rbd_set_owner_cid(rbd_dev, &cid); + queue_work(rbd_dev->task_wq, &rbd_dev->acquired_lock_work); + return 0; +} + +/* + * lock_rwsem must be held for write + */ +static int rbd_unlock(struct rbd_device *rbd_dev) +{ + struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; + char cookie[32]; + int ret; + + WARN_ON(!__rbd_is_lock_owner(rbd_dev)); + + rbd_dev->lock_state = RBD_LOCK_STATE_UNLOCKED; + + format_lock_cookie(rbd_dev, cookie); + ret = ceph_cls_unlock(osdc, &rbd_dev->header_oid, &rbd_dev->header_oloc, + RBD_LOCK_NAME, cookie); + if (ret && ret != -ENOENT) { + rbd_warn(rbd_dev, "cls_unlock failed: %d", ret); + return ret; + } + + rbd_set_owner_cid(rbd_dev, &rbd_empty_cid); + queue_work(rbd_dev->task_wq, &rbd_dev->released_lock_work); + return 0; +} + +static int __rbd_notify_op_lock(struct rbd_device *rbd_dev, + enum rbd_notify_op notify_op, + struct page ***preply_pages, + size_t *preply_len) +{ + struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc; + struct rbd_client_id cid = rbd_get_cid(rbd_dev); + int buf_size = 4 + 8 + 8 + CEPH_ENCODING_START_BLK_LEN; + char buf[buf_size]; + void *p = buf; + + dout("%s rbd_dev %p notify_op %d\n", __func__, rbd_dev, notify_op); + + /* encode *LockPayload NotifyMessage (op + ClientId) */ + ceph_start_encoding(&p, 2, 1, buf_size - CEPH_ENCODING_START_BLK_LEN); + ceph_encode_32(&p, notify_op); + ceph_encode_64(&p, cid.gid); + ceph_encode_64(&p, cid.handle); + + return ceph_osdc_notify(osdc, &rbd_dev->header_oid, + &rbd_dev->header_oloc, buf, buf_size, + RBD_NOTIFY_TIMEOUT, preply_pages, preply_len); +} + +static void rbd_notify_op_lock(struct rbd_device *rbd_dev, + enum rbd_notify_op notify_op) +{ + struct page **reply_pages; + size_t reply_len; + + __rbd_notify_op_lock(rbd_dev, notify_op, &reply_pages, &reply_len); + ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len)); +} + +static void rbd_notify_acquired_lock(struct work_struct *work) +{ + struct rbd_device *rbd_dev = container_of(work, struct rbd_device, + acquired_lock_work); + + rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_ACQUIRED_LOCK); +} + +static void rbd_notify_released_lock(struct work_struct *work) +{ + struct rbd_device *rbd_dev = container_of(work, struct rbd_device, + released_lock_work); + + rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_RELEASED_LOCK); +} + +static int rbd_request_lock(struct rbd_device *rbd_dev) +{ + struct page **reply_pages; + size_t reply_len; + bool lock_owner_responded = false; int ret; - dout("%s rbd_dev %p cookie %llu notify_id %llu\n", __func__, rbd_dev, - cookie, notify_id); + dout("%s rbd_dev %p\n", __func__, rbd_dev); + + ret = __rbd_notify_op_lock(rbd_dev, RBD_NOTIFY_OP_REQUEST_LOCK, + &reply_pages, &reply_len); + if (ret && ret != -ETIMEDOUT) { + rbd_warn(rbd_dev, "failed to request lock: %d", ret); + goto out; + } + + if (reply_len > 0 && reply_len <= PAGE_SIZE) { + void *p = page_address(reply_pages[0]); + void *const end = p + reply_len; + u32 n; + + ceph_decode_32_safe(&p, end, n, e_inval); /* num_acks */ + while (n--) { + u8 struct_v; + u32 len; + ceph_decode_need(&p, end, 8 + 8, e_inval); + p += 8 + 8; /* skip gid and cookie */ + + ceph_decode_32_safe(&p, end, len, e_inval); + if (!len) + continue; + + if (lock_owner_responded) { + rbd_warn(rbd_dev, + "duplicate lock owners detected"); + ret = -EIO; + goto out; + } + + lock_owner_responded = true; + ret = ceph_start_decoding(&p, end, 1, "ResponseMessage", + &struct_v, &len); + if (ret) { + rbd_warn(rbd_dev, + "failed to decode ResponseMessage: %d", + ret); + goto e_inval; + } + + ret = ceph_decode_32(&p); + } + } + + if (!lock_owner_responded) { + rbd_warn(rbd_dev, "no lock owners detected"); + ret = -ETIMEDOUT; + } + +out: + ceph_release_page_vector(reply_pages, calc_pages_for(0, reply_len)); + return ret; + +e_inval: + ret = -EINVAL; + goto out; +} + +static void wake_requests(struct rbd_device *rbd_dev, bool wake_all) +{ + dout("%s rbd_dev %p wake_all %d\n", __func__, rbd_dev, wake_all); + + cancel_delayed_work(&rbd_dev->lock_dwork); + if (wake_all) + wake_up_all(&rbd_dev->lock_waitq); + else + wake_up(&rbd_dev->lock_waitq); +} + +static int get_lock_owner_info(struct rbd_device *rbd_dev, + |