[RFC PATCH 21/35] libceph: Make notify code use ceph_databuf_enc_start/stop

[David Howells <dhowells@xxxxxxxxxx>]

Make the ceph_osdc_notify*() functions use ceph_databuf_enc_start() and
ceph_databuf_enc_stop() when filling out the request data.  Also use
ceph_encode_*() rather than ceph_databuf_encode_*() as the latter will do
an iterator copy to deal with page crossing and misalignment (the latter
being something that the CPU will handle on some arches).

Signed-off-by: David Howells <dhowells@xxxxxxxxxx>
cc: Viacheslav Dubeyko <slava@xxxxxxxxxxx>
cc: Alex Markuze <amarkuze@xxxxxxxxxx>
cc: Ilya Dryomov <idryomov@xxxxxxxxx>
cc: ceph-devel@xxxxxxxxxxxxxxx
cc: linux-fsdevel@xxxxxxxxxxxxxxx
---
 net/ceph/osd_client.c | 55 +++++++++++++++++++++----------------------
 1 file changed, 27 insertions(+), 28 deletions(-)

diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 0ac439e7e730..1a0cb2cdcc52 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -4759,7 +4759,10 @@ static int osd_req_op_notify_ack_init(struct ceph_osd_request *req, int which,
 {
 	struct ceph_osd_req_op *op;
 	struct ceph_databuf *dbuf;
-	int ret;
+	void *p;
+
+	if (!payload)
+		payload_len = 0;
 
 	op = osd_req_op_init(req, which, CEPH_OSD_OP_NOTIFY_ACK, 0);
 
@@ -4767,18 +4770,13 @@ static int osd_req_op_notify_ack_init(struct ceph_osd_request *req, int which,
 	if (!dbuf)
 		return -ENOMEM;
 
-	ret = ceph_databuf_encode_64(dbuf, notify_id);
-	ret |= ceph_databuf_encode_64(dbuf, cookie);
-	if (payload) {
-		ret |= ceph_databuf_encode_32(dbuf, payload_len);
-		ret |= ceph_databuf_append(dbuf, payload, payload_len);
-	} else {
-		ret |= ceph_databuf_encode_32(dbuf, 0);
-	}
-	if (ret) {
-		ceph_databuf_release(dbuf);
-		return -ENOMEM;
-	}
+	p = ceph_databuf_enc_start(dbuf);
+	ceph_encode_64(&p, notify_id);
+	ceph_encode_64(&p, cookie);
+	ceph_encode_32(&p, payload_len);
+	if (payload)
+		ceph_encode_copy(&p, payload, payload_len);
+	ceph_databuf_enc_stop(dbuf, p);
 
 	ceph_osd_databuf_init(&op->notify_ack.request_data, dbuf);
 	op->indata_len = ceph_databuf_len(dbuf);
@@ -4840,8 +4838,12 @@ int ceph_osdc_notify(struct ceph_osd_client *osdc,
 		     size_t *preply_len)
 {
 	struct ceph_osd_linger_request *lreq;
+	void *p;
 	int ret;
 
+	if (WARN_ON_ONCE(payload_len > PAGE_SIZE - 3 * 4))
+		return -EIO;
+
 	WARN_ON(!timeout);
 	if (preply_pages) {
 		*preply_pages = NULL;
@@ -4852,20 +4854,19 @@ int ceph_osdc_notify(struct ceph_osd_client *osdc,
 	if (!lreq)
 		return -ENOMEM;
 
-	lreq->request_pl = ceph_databuf_req_alloc(1, PAGE_SIZE, GFP_NOIO);
+	lreq->request_pl = ceph_databuf_req_alloc(0, 3 * 4 + payload_len,
+						  GFP_NOIO);
 	if (!lreq->request_pl) {
 		ret = -ENOMEM;
 		goto out_put_lreq;
 	}
 
-	ret = ceph_databuf_encode_32(lreq->request_pl, 1); /* prot_ver */
-	ret |= ceph_databuf_encode_32(lreq->request_pl, timeout);
-	ret |= ceph_databuf_encode_32(lreq->request_pl, payload_len);
-	ret |= ceph_databuf_append(lreq->request_pl, payload, payload_len);
-	if (ret) {
-		ret = -ENOMEM;
-		goto out_put_lreq;
-	}
+	p = ceph_databuf_enc_start(lreq->request_pl);
+	ceph_encode_32(&p, 1); /* prot_ver */
+	ceph_encode_32(&p, timeout);
+	ceph_encode_32(&p, payload_len);
+	ceph_encode_copy(&p, payload, payload_len);
+	ceph_databuf_enc_stop(lreq->request_pl, p);
 
 	/* for notify_id */
 	lreq->notify_id_buf = ceph_databuf_reply_alloc(1, PAGE_SIZE, GFP_NOIO);
@@ -5217,7 +5218,7 @@ int osd_req_op_copy_from_init(struct ceph_osd_request *req,
 {
 	struct ceph_osd_req_op *op;
 	struct ceph_databuf *dbuf;
-	void *p, *end;
+	void *p;
 
 	dbuf = ceph_databuf_req_alloc(1, PAGE_SIZE, GFP_KERNEL);
 	if (!dbuf)
@@ -5230,15 +5231,13 @@ int osd_req_op_copy_from_init(struct ceph_osd_request *req,
 	op->copy_from.flags = copy_from_flags;
 	op->copy_from.src_fadvise_flags = src_fadvise_flags;
 
-	p = kmap_ceph_databuf_page(dbuf, 0);
-	end = p + PAGE_SIZE;
+	p = ceph_databuf_enc_start(dbuf);
 	ceph_encode_string(&p, src_oid->name, src_oid->name_len);
 	encode_oloc(&p, src_oloc);
 	ceph_encode_32(&p, truncate_seq);
 	ceph_encode_64(&p, truncate_size);
-	op->indata_len = PAGE_SIZE - (end - p);
-	ceph_databuf_added_data(dbuf, op->indata_len);
-	kunmap_local(p);
+	ceph_databuf_enc_stop(dbuf, p);
+	op->indata_len = ceph_databuf_len(dbuf);
 
 	ceph_osd_databuf_init(&op->copy_from.osd_data, dbuf);
 	return 0;