The main process_work now gets a struct request passed to it to simply the
code in all instanes, and a new process_cluster_work is used for the cluster
ops where we can't simplify pass the request.

Recommendations for a better name for process_work_cluster are welcome.

Signed-off-by: Christoph Hellwig <hch@lst.de>
---
 sheep/group.c      |    4 -
 sheep/ops.c        |  136 ++++++++++++++++++++++++++---------------------------
 sheep/sdnet.c      |    2 
 sheep/sheep_priv.h |    7 +-
 sheep/store.c      |    2 
 5 files changed, 78 insertions(+), 73 deletions(-)

Index: sheepdog/sheep/ops.c
===================================================================
--- sheepdog.orig/sheep/ops.c	2012-05-07 10:54:32.285461465 +0200
+++ sheepdog/sheep/ops.c	2012-05-07 10:54:39.665461654 +0200
@@ -39,24 +39,24 @@ struct sd_op_template {
 	int force;
 
 	/*
-	 * process_work() will be called in the worker thread, and
-	 * process_main() will be called in the main thread.
+	 * process_work() and process_work_cluster will be called in a
+	 * worker thread, and process_main() will be called in the main thread.
 	 *
-	 * If type is SD_OP_TYPE_CLUSTER, it is guaranteed that only
-	 * one node processes a cluster operation at the same time.
-	 * We can use this for something like distributed locking.
-	 * process_work() will be called on the local node, and
-	 * process_main() will be called on every nodes.
+	 * If type is SD_OP_TYPE_CLUSTER, it is guaranteed that only one node
+	 * processes a cluster operation at the same time.  We can use this for
+	 * for example to implement distributed locking.  process_work_cluster()
+	 * will be called on the local node, and process_main() will be called
+	 * on every node.
 	 *
-	 * If type is SD_OP_TYPE_LOCAL, both process_work() and
-	 * process_main() will be called on the local node.
+	 * If type is SD_OP_TYPE_LOCAL, both process_work() and process_main()
+	 * will be called on the local node.
 	 *
-	 * If type is SD_OP_TYPE_IO, neither process_work() nor
-	 * process_main() is used because this type of operation is
-	 * heavily intertwined with Sheepdog core codes.  We will be
-	 * unlikely to add new operations of this type.
+	 * If type is SD_OP_TYPE_IO, only process_work() will be called, and it
+	 * will be called on the local node.
 	 */
-	int (*process_work)(const struct sd_req *req, struct sd_rsp *rsp, void *data);
+	int (*process_work)(struct request *req);
+	int (*process_work_cluster)(const struct sd_req *req,
+			struct sd_rsp *rsp, void *data);
 	int (*process_main)(const struct sd_req *req, struct sd_rsp *rsp, void *data);
 };
 
@@ -319,8 +319,7 @@ static int cluster_get_vdi_attr(const st
 	return ret;
 }
 
-static int local_get_store_list(const struct sd_req *req, struct sd_rsp *rsp,
-				void *data)
+static int local_get_store_list(struct request *req)
 {
 	struct strbuf buf = STRBUF_INIT;
 	struct store_driver *driver;
@@ -328,7 +327,7 @@ static int local_get_store_list(const st
 	list_for_each_entry(driver, &store_drivers, list) {
 		strbuf_addf(&buf, "%s ", driver->name);
 	}
-	memcpy(data, buf.buf, buf.len);
+	memcpy(req->data, buf.buf, buf.len);
 
 	strbuf_release(&buf);
 	return SD_RES_SUCCESS;
@@ -366,11 +365,10 @@ static int local_get_node_list(const str
 	return SD_RES_SUCCESS;
 }
 
-static int local_stat_sheep(const struct sd_req *req, struct sd_rsp *rsp,
-			    void *data)
+static int local_stat_sheep(struct request *req)
 {
-	struct sd_node_rsp *node_rsp = (struct sd_node_rsp *)rsp;
-	uint32_t epoch = req->epoch;
+	struct sd_node_rsp *node_rsp = (struct sd_node_rsp *)&req->rp;
+	uint32_t epoch = req->rq.epoch;
 
 	return stat_sheep(&node_rsp->store_size, &node_rsp->store_free, epoch);
 }
@@ -386,9 +384,9 @@ static int local_stat_recovery(const str
 	return SD_RES_UNKNOWN;
 }
 
-static int local_stat_cluster(const struct sd_req *req, struct sd_rsp *rsp,
-			      void *data)
+static int local_stat_cluster(struct request *req)
 {
+	struct sd_rsp *rsp = &req->rp;
 	struct epoch_log *log;
 	int i, max_logs;
 	uint32_t sys_stat = sys_stat_get(), epoch;
@@ -400,7 +398,7 @@ static int local_stat_cluster(const stru
 		if (epoch <= 0)
 			break;
 
-		log = (struct epoch_log *)data + i;
+		log = (struct epoch_log *)req->data + i;
 		log->epoch = epoch;
 		log->ctime = get_cluster_ctime();
 		log->nr_nodes = epoch_log_read(epoch, (char *)log->nodes,
@@ -436,28 +434,27 @@ static int local_stat_cluster(const stru
 	}
 }
 
-static int local_kill_node(const struct sd_req *req, struct sd_rsp *rsp,
-			   void *data)
+static int local_kill_node(struct request *req)
 {
 	exit(1);
 }
 
-static int local_get_obj_list(const struct sd_req *req, struct sd_rsp *rsp,
-			      void *data)
+static int local_get_obj_list(struct request *req)
 {
-	return get_obj_list((const struct sd_list_req *)req,
-			    (struct sd_list_rsp *)rsp, data);
+	return get_obj_list((const struct sd_list_req *)&req->rq,
+			    (struct sd_list_rsp *)&req->rp, req->data);
 }
 
-static int local_get_epoch(const struct sd_req *req, struct sd_rsp *rsp,
-			   void *data)
+static int local_get_epoch(struct request *req)
 {
-	const struct sd_obj_req *obj_req = (const struct sd_obj_req *)req;
-	struct sd_obj_rsp *obj_rsp = (struct sd_obj_rsp *)rsp;
+	const struct sd_obj_req *obj_req = (const struct sd_obj_req *)&req->rq;
+	struct sd_obj_rsp *obj_rsp = (struct sd_obj_rsp *)&req->rp;
 	uint32_t epoch = obj_req->tgt_epoch;
 	int len, ret;
+
 	dprintf("%d\n", epoch);
-	len = epoch_log_read(epoch, (char *)data, obj_req->data_length);
+
+	len = epoch_log_read(epoch, req->data, obj_req->data_length);
 	if (len == -1) {
 		ret = SD_RES_NO_TAG;
 		obj_rsp->data_length = 0;
@@ -564,15 +561,14 @@ static int cluster_restore(const struct
 	return ret;
 }
 
-static int local_get_snap_file(const struct sd_req *req, struct sd_rsp *rsp,
-			    void *data)
+static int local_get_snap_file(struct request *req)
 {
 	int ret;
-	struct siocb iocb = { .buf = data };
+	struct siocb iocb = { .buf = req->data };
 
 	if (sd_store->get_snap_file) {
 		ret = sd_store->get_snap_file(&iocb);
-		rsp->data_length = iocb.length;
+		req->rp.data_length = iocb.length;
 	} else
 		ret = SD_RES_NO_SUPPORT;
 
@@ -595,9 +591,9 @@ static void flush_vdi_done(struct work *
 	free(fw);
 }
 
-static int local_flush_vdi(const struct sd_req *req, struct sd_rsp *rsp, void *data)
+static int local_flush_vdi(struct request *req)
 {
-	struct sd_obj_req *hdr = (struct sd_obj_req *)req;
+	struct sd_obj_req *hdr = (struct sd_obj_req *)&req->rq;
 	uint64_t oid = hdr->oid;
 	uint32_t vid = oid_to_vid(oid);
 	struct object_cache *cache = find_object_cache(vid, 0);
@@ -709,9 +705,9 @@ out:
 	return ret;
 }
 
-static int store_remove_obj(const struct sd_req *req, struct sd_rsp *rsp, void *data)
+static int store_remove_obj(struct request *req)
 {
-	struct sd_obj_req *hdr = (struct sd_obj_req *)req;
+	struct sd_obj_req *hdr = (struct sd_obj_req *)&req->rq;
 	uint32_t epoch = hdr->epoch;
 	struct strbuf buf = STRBUF_INIT;
 	int ret = SD_RES_SUCCESS;
@@ -735,11 +731,10 @@ static int store_remove_obj(const struct
 	return ret;
 }
 
-static int store_read_obj(const struct sd_req *req, struct sd_rsp *rsp, void *data)
+static int store_read_obj(struct request *req)
 {
-	struct sd_obj_req *hdr = (struct sd_obj_req *)req;
-	struct sd_obj_rsp *rsps = (struct sd_obj_rsp *)rsp;
-	struct request *request = (struct request *)data;
+	struct sd_obj_req *hdr = (struct sd_obj_req *)&req->rq;
+	struct sd_obj_rsp *rsps = (struct sd_obj_rsp *)&req->rp;
 	int ret;
 	uint32_t epoch = hdr->epoch;
 	struct siocb iocb;
@@ -751,7 +746,7 @@ static int store_read_obj(const struct s
 	if (ret != SD_RES_SUCCESS)
 		return ret;
 
-	iocb.buf = request->data;
+	iocb.buf = req->data;
 	iocb.length = hdr->data_length;
 	iocb.offset = hdr->offset;
 	ret = sd_store->read(hdr->oid, &iocb);
@@ -795,11 +790,9 @@ static int do_write_obj(struct siocb *io
 	return ret;
 }
 
-static int store_write_obj(const struct sd_req *req, struct sd_rsp *rsp,
-			   void *data)
+static int store_write_obj(struct request *req)
 {
-	struct sd_obj_req *hdr = (struct sd_obj_req *)req;
-	struct request *request = (struct request *)data;
+	struct sd_obj_req *hdr = (struct sd_obj_req *)&req->rq;
 	int ret;
 	uint32_t epoch = hdr->epoch;
 	struct siocb iocb;
@@ -811,17 +804,15 @@ static int store_write_obj(const struct
 	if (ret != SD_RES_SUCCESS)
 		return ret;
 
-	ret = do_write_obj(&iocb, hdr, epoch, request->data);
+	ret = do_write_obj(&iocb, hdr, epoch, req->data);
 
 	sd_store->close(hdr->oid, &iocb);
 	return ret;
 }
 
-static int store_create_and_write_obj(const struct sd_req *req,
-				struct sd_rsp *rsp, void *data)
+static int store_create_and_write_obj(struct request *req)
 {
-	struct sd_obj_req *hdr = (struct sd_obj_req *)req;
-	struct request *request = (struct request *)data;
+	struct sd_obj_req *hdr = (struct sd_obj_req *)&req->rq;
 	struct sd_obj_req cow_hdr;
 	int ret;
 	uint32_t epoch = hdr->epoch;
@@ -852,21 +843,22 @@ static int store_create_and_write_obj(co
 			goto out;
 		}
 		if (hdr->data_length != SD_DATA_OBJ_SIZE) {
-			ret = read_copy_from_replica(request, hdr->epoch, hdr->cow_oid, buf);
+			ret = read_copy_from_replica(req, hdr->epoch,
+						     hdr->cow_oid, buf);
 			if (ret != SD_RES_SUCCESS) {
 				eprintf("failed to read cow object\n");
 				goto out;
 			}
 		}
 
-		memcpy(buf + hdr->offset, request->data, hdr->data_length);
+		memcpy(buf + hdr->offset, req->data, hdr->data_length);
 		memcpy(&cow_hdr, hdr, sizeof(cow_hdr));
 		cow_hdr.offset = 0;
 		cow_hdr.data_length = SD_DATA_OBJ_SIZE;
 
 		ret = do_write_obj(&iocb, &cow_hdr, epoch, buf);
 	} else
-		ret = do_write_obj(&iocb, hdr, epoch, request->data);
+		ret = do_write_obj(&iocb, hdr, epoch, req->data);
 
 	if (SD_RES_SUCCESS == ret)
 		check_and_insert_objlist_cache(hdr->oid);
@@ -882,23 +874,23 @@ static struct sd_op_template sd_ops[] =
 	/* cluster operations */
 	[SD_OP_NEW_VDI] = {
 		.type = SD_OP_TYPE_CLUSTER,
-		.process_work = cluster_new_vdi,
+		.process_work_cluster = cluster_new_vdi,
 		.process_main = post_cluster_new_vdi,
 	},
 
 	[SD_OP_DEL_VDI] = {
 		.type = SD_OP_TYPE_CLUSTER,
-		.process_work = cluster_del_vdi,
+		.process_work_cluster = cluster_del_vdi,
 	},
 
 	[SD_OP_GET_VDI_INFO] = {
 		.type = SD_OP_TYPE_CLUSTER,
-		.process_work = cluster_get_vdi_info,
+		.process_work_cluster = cluster_get_vdi_info,
 	},
 
 	[SD_OP_LOCK_VDI] = {
 		.type = SD_OP_TYPE_CLUSTER,
-		.process_work = cluster_get_vdi_info,
+		.process_work_cluster = cluster_get_vdi_info,
 	},
 
 	[SD_OP_MAKE_FS] = {
@@ -914,7 +906,7 @@ static struct sd_op_template sd_ops[] =
 
 	[SD_OP_GET_VDI_ATTR] = {
 		.type = SD_OP_TYPE_CLUSTER,
-		.process_work = cluster_get_vdi_attr,
+		.process_work_cluster = cluster_get_vdi_attr,
 	},
 
 	[SD_OP_RELEASE_VDI] = {
@@ -1074,15 +1066,25 @@ int has_process_work(struct sd_op_templa
 	return !!op->process_work;
 }
 
+int has_process_work_cluster(struct sd_op_template *op)
+{
+	return !!op->process_work_cluster;
+}
+
 int has_process_main(struct sd_op_template *op)
 {
 	return !!op->process_main;
 }
 
-int do_process_work(struct sd_op_template *op, const struct sd_req *req,
+int do_process_work(struct request *req)
+{
+	return req->op->process_work(req);
+}
+
+int do_process_work_cluster(struct sd_op_template *op, const struct sd_req *req,
 		    struct sd_rsp *rsp, void *data)
 {
-	return op->process_work(req, rsp, data);
+	return op->process_work_cluster(req, rsp, data);
 }
 
 int do_process_main(struct sd_op_template *op, const struct sd_req *req,
Index: sheepdog/sheep/sdnet.c
===================================================================
--- sheepdog.orig/sheep/sdnet.c	2012-05-07 10:54:33.681461501 +0200
+++ sheepdog/sheep/sdnet.c	2012-05-07 10:54:39.669461654 +0200
@@ -183,7 +183,7 @@ static void do_local_request(struct work
 	int ret = SD_RES_SUCCESS;
 
 	if (has_process_work(req->op))
-		ret = do_process_work(req->op, &req->rq, &req->rp, req->data);
+		ret = do_process_work(req);
 
 	rsp->result = ret;
 }
Index: sheepdog/sheep/sheep_priv.h
===================================================================
--- sheepdog.orig/sheep/sheep_priv.h	2012-05-07 10:54:33.681461501 +0200
+++ sheepdog/sheep/sheep_priv.h	2012-05-07 10:54:39.669461654 +0200
@@ -322,11 +322,14 @@ int is_io_op(struct sd_op_template *op);
 int is_force_op(struct sd_op_template *op);
 int has_process_work(struct sd_op_template *op);
 int has_process_main(struct sd_op_template *op);
-int do_process_work(struct sd_op_template *op, const struct sd_req *req,
-		    struct sd_rsp *rsp, void *data);
+int do_process_work(struct request *req);
 int do_process_main(struct sd_op_template *op, const struct sd_req *req,
 		    struct sd_rsp *rsp, void *data);
 
+int has_process_work_cluster(struct sd_op_template *op);
+int do_process_work_cluster(struct sd_op_template *op, const struct sd_req *req,
+			    struct sd_rsp *rsp, void *data);
+
 /* Journal */
 struct jrnl_descriptor *jrnl_begin(const void *buf, size_t count, off_t offset,
 				   const char *path, const char *jrnl_dir);
Index: sheepdog/sheep/store.c
===================================================================
--- sheepdog.orig/sheep/store.c	2012-05-07 10:54:32.285461465 +0200
+++ sheepdog/sheep/store.c	2012-05-07 10:54:39.669461654 +0200
@@ -53,7 +53,7 @@ static int do_local_io(struct request *r
 	hdr->epoch = epoch;
 	dprintf("%x, %" PRIx64" , %u\n", hdr->opcode, hdr->oid, epoch);
 
-	return do_process_work(req->op, &req->rq, &req->rp, req);
+	return do_process_work(req);
 }
 
 static int forward_read_obj_req(struct request *req)
Index: sheepdog/sheep/group.c
===================================================================
--- sheepdog.orig/sheep/group.c	2012-05-07 10:54:33.681461501 +0200
+++ sheepdog/sheep/group.c	2012-05-07 10:54:39.669461654 +0200
@@ -219,7 +219,7 @@ static void do_cluster_op(void *arg)
 		data = msg->data;
 	else
 		data = req->data;
-	ret = do_process_work(req->op, (const struct sd_req *)&msg->req,
+	ret = do_process_work_cluster(req->op, (const struct sd_req *)&msg->req,
 			      (struct sd_rsp *)&msg->rsp, data);
 
 	msg->rsp.result = ret;
@@ -252,7 +252,7 @@ void do_cluster_request(struct work *wor
 
 	list_add_tail(&req->pending_list, &sys->pending_list);
 
-	if (has_process_work(req->op))
+	if (has_process_work_cluster(req->op))
 		sys->cdrv->notify(msg, size, do_cluster_op);
 	else {
 		msg->rsp.result = SD_RES_SUCCESS;


