[PATCH] chunkd: add cp command, for local intra-table copies

[PATCH] chunkd: add cp command, for local intra-table copies

[Jeff Garzik]

The following patch, against current hail.git, adds the "CP" command to
chunkd, permitting copying from object->object inside a single table.

It also adds a worker thread pool that may be used by other background
tasks as well.


diff --git a/chunkd/chunkd.h b/chunkd/chunkd.h
index e019f0d..f1f7d04 100644
--- a/chunkd/chunkd.h
+++ b/chunkd/chunkd.h
@@ -104,6 +104,8 @@ struct client {
 	unsigned int		req_used;	/* amount of req_buf in use */
 	void			*req_ptr;	/* start of unexamined data */
 	uint16_t		key_len;
+	unsigned int		var_len;
+	bool			second_var;
 
 	char			*hdr_start;	/* current hdr start */
 	char			*hdr_end;	/* current hdr end (so far) */
@@ -124,6 +126,7 @@ struct client {
 	char			netbuf_out[CLI_DATA_BUF_SZ];
 	char			key[CHD_KEY_SZ];
 	char			table[CHD_KEY_SZ];
+	char			key2[CHD_KEY_SZ];
 };
 
 struct backend_obj {
@@ -162,6 +165,16 @@ struct volume_entry {
 	char			*owner;		/* obj owner username */
 };
 
+enum chunk_worker_op {
+	CW_CP,					/* local obj copy */
+};
+
+struct worker_info {
+	enum chunk_worker_op	op;
+	struct client		*cli;
+	enum chunk_errcode	err;
+};
+
 struct server_stats {
 	unsigned long		poll;		/* number polls */
 	unsigned long		event;		/* events dispatched */
@@ -209,6 +222,10 @@ struct server {
 
 	GHashTable		*fd_info;
 
+	GThreadPool		*workers;
+	int			max_workers;
+	int			worker_pipe[2];
+
 	struct list_head	wr_trash;
 	unsigned int		trash_sz;
 
@@ -278,6 +295,9 @@ extern int fs_obj_do_sum(const char *fn, unsigned int klen, char **csump);
 extern bool object_del(struct client *cli);
 extern bool object_put(struct client *cli);
 extern bool object_get(struct client *cli, bool want_body);
+extern bool object_cp(struct client *cli);
+extern void worker_cp_thr(struct worker_info *wi);
+extern void worker_cp_pipe(struct worker_info *wi);
 extern bool cli_evt_data_in(struct client *cli, unsigned int events);
 extern void cli_out_end(struct client *cli);
 extern void cli_in_end(struct client *cli);
@@ -314,12 +334,14 @@ extern bool cli_err(struct client *cli, enum chunk_errcode code, bool recycle_ok
 extern int cli_writeq(struct client *cli, const void *buf, unsigned int buflen,
 		     cli_write_func cb, void *cb_data);
 extern bool cli_wr_sendfile(struct client *, cli_write_func);
+extern bool cli_rd_set_poll(struct client *cli, bool readable);
 extern void cli_wr_set_poll(struct client *cli, bool writable);
 extern bool cli_cb_free(struct client *cli, struct client_write *wr,
 			bool done);
 extern bool cli_write_start(struct client *cli);
 extern int cli_req_avail(struct client *cli);
 extern int cli_poll_mod(struct client *cli);
+extern bool tcp_cli_event(int fd, short events, void *userdata);
 extern void resp_init_req(struct chunksrv_resp *resp,
 		   const struct chunksrv_req *req);
 
diff --git a/chunkd/object.c b/chunkd/object.c
index 116792f..105eb84 100644
--- a/chunkd/object.c
+++ b/chunkd/object.c
@@ -25,6 +25,7 @@
 #include <unistd.h>
 #include <string.h>
 #include <errno.h>
+#include <poll.h>
 #include <stdio.h>
 #include <syslog.h>
 #include <glib.h>
@@ -356,3 +357,102 @@ start_write:
 	return cli_write_start(cli);
 }
 
+bool object_cp(struct client *cli)
+{
+	enum chunk_errcode err = che_InternalError;
+	struct worker_info *wi;
+
+	wi = calloc(1, sizeof(*wi));
+	if (!wi)
+		return cli_err(cli, err, false);
+
+	wi->op = CW_CP;
+	wi->cli = cli;
+
+	g_thread_pool_push(chunkd_srv.workers, wi, NULL);
+
+	cli_rd_set_poll(cli, false);
+
+	return false;
+}
+
+void worker_cp_thr(struct worker_info *wi)
+{
+	unsigned char md[SHA_DIGEST_LENGTH];
+	char hashstr[50];
+	struct client *cli = wi->cli;
+	struct backend_obj *obj = NULL, *out_obj = NULL;
+	enum chunk_errcode err = che_InternalError;
+	void *buf;
+	static const unsigned bufsz = 1 * 1024 * 1024;
+
+	buf = malloc(bufsz);
+	if (!buf)
+		goto out;
+
+	cli->in_obj = obj = fs_obj_open(cli->table_id, cli->user, cli->key2,
+					cli->var_len, &err);
+	if (!obj)
+		goto out;
+
+	cli->in_len = obj->size;
+
+	cli->out_bo = out_obj = fs_obj_new(cli->table_id,
+					   cli->key, cli->key_len, &err);
+	if (!cli->out_bo)
+		goto out;
+
+	SHA1_Init(&cli->out_hash);
+
+	while (cli->in_len > 0) {
+		ssize_t rrc, wrc;
+
+		rrc = fs_obj_read(obj, buf, MIN(cli->in_len, bufsz));
+		if (rrc < 0)
+			goto err_out;
+		if (rrc == 0)
+			break;
+
+		SHA1_Update(&cli->out_hash, buf, rrc);
+
+		wrc = fs_obj_write(out_obj, buf, rrc);
+		if (wrc < 0)
+			goto err_out;
+
+		cli->in_len -= wrc;
+	}
+
+	SHA1_Final(md, &cli->out_hash);
+	hexstr(md, SHA_DIGEST_LENGTH, hashstr);
+
+	if (!fs_obj_write_commit(out_obj, cli->user, hashstr, false))
+		goto err_out;
+
+	err = che_Success;
+
+out:
+	if (buf)
+		free(buf);
+	if (obj)
+		fs_obj_free(obj);
+	if (out_obj)
+		fs_obj_free(out_obj);
+	wi->err = err;
+	write(chunkd_srv.worker_pipe[1], &wi, sizeof(wi));
+
+err_out:
+	/* FIXME: remove half-written destination object */
+	goto out;
+}
+
+void worker_cp_pipe(struct worker_info *wi)
+{
+	struct client *cli = wi->cli;
+	bool rcb;
+
+	cli_rd_set_poll(cli, true);
+
+	rcb = cli_err(cli, wi->err, (wi->err == che_Success) ? true : false);
+	if (rcb)
+		tcp_cli_event(cli->fd, POLLIN | POLLOUT, cli);
+}
diff --git a/chunkd/server.c b/chunkd/server.c
index abd29ad..0fec33a 100644
--- a/chunkd/server.c
+++ b/chunkd/server.c
@@ -418,6 +418,16 @@ static bool cli_evt_recycle(struct client *cli, unsigned int events)
 	return true;
 }
 
+bool cli_rd_set_poll(struct client *cli, bool readable)
+{
+	if (readable)
+		srv_poll_mask(cli->fd, POLLIN, 0);
+	else
+		srv_poll_mask(cli->fd, 0, POLLIN);
+	
+	return true;
+}
+
 void cli_wr_set_poll(struct client *cli, bool writable)
 {
 	if (writable)
@@ -1049,6 +1059,7 @@ static const char *op2str(enum chunksrv_ops op)
 	case CHO_CHECK_START:	return "CHO_CHECK_START";
 	case CHO_CHECK_STATUS:	return "CHO_CHECK_STATUS";
 	case CHO_START_TLS:	return "CHO_START_TLS";
+	case CHO_CP:		return "CHO_CP";
 
 	default:
 		return "BUG/UNKNOWN!";
@@ -1143,6 +1154,9 @@ static bool cli_evt_exec_req(struct client *cli, unsigned int events)
 	case CHO_DEL:
 		rcb = object_del(cli);
 		break;
+	case CHO_CP:
+		rcb = object_cp(cli);
+		break;
 	case CHO_LIST:
 		rcb = volume_list(cli);
 		break;
@@ -1226,8 +1240,10 @@ static bool cli_evt_read_fixed(struct client *cli, unsigned int events)
 
 	/* otherwise, go to read-variable-len-record state */
 	cli->req_ptr = &cli->key;
+	cli->var_len = cli->key_len;
 	cli->req_used = 0;
 	cli->state = evt_read_var;
+	cli->second_var = false;
 
 	return true;
 }
@@ -1235,7 +1251,7 @@ static bool cli_evt_read_fixed(struct client *cli, unsigned int events)
 static bool cli_evt_read_var(struct client *cli, unsigned int events)
 {
 	int rc = cli_read_data(cli, cli->req_ptr,
-			       cli->key_len - cli->req_used);
+			       cli->var_len - cli->req_used);
 	if (rc < 0) {
 		cli->state = evt_dispose;
 		return true;
@@ -1245,10 +1261,17 @@ static bool cli_evt_read_var(struct client *cli, unsigned int events)
 	cli->req_used += rc;
 
 	/* poll for more, if variable-length record not yet received */
-	if (cli->req_used < cli->key_len)
+	if (cli->req_used < cli->var_len)
 		return false;
 
-	cli->state = evt_exec_req;
+	if (cli->creq.op == CHO_CP && !cli->second_var) {
+		cli->req_ptr = &cli->key2;
+		cli->var_len = le64_to_cpu(cli->creq.data_len);
+		cli->req_used = 0;
+		cli->state = evt_read_var;
+		cli->second_var = true;
+	} else
+		cli->state = evt_exec_req;
 
 	return true;
 }
@@ -1301,7 +1324,7 @@ static void tcp_cli_wr_event(int fd, short events, void *userdata)
 		cli_writable(cli);
 }
 
-static bool tcp_cli_event(int fd, short events, void *userdata)
+bool tcp_cli_event(int fd, short events, void *userdata)
 {
 	struct client *cli = userdata;
 	bool loop = false, disposing = false;
@@ -1419,6 +1442,25 @@ static int net_write_port(const char *port_file, const char *port_str)
 	return 0;
 }
 
+static bool pipe_watch(int pipe_fd_0, 
+		       bool (*cb)(int fd, short events, void *userdata),
+		       void *userdata)
+{
+	struct server_poll *sp;
+
+	sp = calloc(1, sizeof(*sp));
+	if (!sp)
+		return false;
+
+	sp->events = POLLIN;
+	sp->cb = cb;
+	sp->userdata = userdata;
+
+	g_hash_table_insert(chunkd_srv.fd_info, GINT_TO_POINTER(pipe_fd_0), sp);
+
+	return true;
+}
+
 static int net_open_socket(const struct listen_cfg *cfg,
 			   int addr_fam, int sock_type, int sock_prot,
 			   int addr_len, void *addr_ptr)
@@ -1634,6 +1676,33 @@ static int net_open(struct listen_cfg *cfg)
 		return net_open_known(cfg);
 }
 
+static void worker_thread(gpointer data, gpointer userdata)
+{
+	struct worker_info *wi = data;
+
+	switch (wi->op) {
+	case CW_CP:
+		worker_cp_thr(wi);
+		break;
+	}
+}
+
+static bool worker_pipe_evt(int fd, short events, void *userdata)
+{
+	struct worker_info *wi = NULL;
+
+	if (read(fd, &wi, sizeof(wi)) != sizeof(wi))
+		return false;
+
+	switch (wi->op) {
+	case CW_CP:
+		worker_cp_pipe(wi);
+		break;
+	}
+
+	return true;
+}
+
 static void fill_poll_arr(gpointer key, gpointer val, gpointer userdata)
 {
 	int fd = GPOINTER_TO_INT(key);
@@ -1830,21 +1899,38 @@ int main (int argc, char *argv[])
 		goto err_out_session;
 	}
 
+	chunkd_srv.max_workers = 10;
+	chunkd_srv.workers = g_thread_pool_new(worker_thread, NULL,
+					       chunkd_srv.max_workers,
+					       FALSE, NULL);
+	if (!chunkd_srv.workers) {
+		rc = 1;
+		goto err_out_fd_info;
+	}
+
 	if (objcache_init(&chunkd_srv.actives) != 0) {
 		rc = 1;
-		goto err_out_objcache;
+		goto err_out_workers;
 	}
 
 	chunkd_srv.trash_sz = 0;
 
 	if (pipe(chunkd_srv.chk_pipe) < 0) {
 		rc = 1;
-		goto err_out_pipe;
+		goto err_out_objcache;
+	}
+	if (pipe(chunkd_srv.worker_pipe) < 0) {
+		rc = 1;
+		goto err_out_chk_pipe;
+	}
+	if (!pipe_watch(chunkd_srv.worker_pipe[0], worker_pipe_evt, NULL)) {
+		rc = 1;
+		goto err_out_chk_pipe;
 	}
 
 	if (fs_open()) {
 		rc = 1;
-		goto err_out_fs;
+		goto err_out_worker_pipe;
 	}
 
 	/* set up server networking */
@@ -1874,13 +1960,17 @@ err_out_cld:
 	/* net_close(); */
 err_out_listen:
 	fs_close();
-err_out_fs:
+err_out_worker_pipe:
+err_out_chk_pipe:
 	cmd = CHK_CMD_EXIT;
 	write(chunkd_srv.chk_pipe[1], &cmd, 1);
 	close(chunkd_srv.chk_pipe[1]);
-err_out_pipe:
-	objcache_fini(&chunkd_srv.actives);
 err_out_objcache:
+	objcache_fini(&chunkd_srv.actives);
+err_out_workers:
+	if (strict_free)
+		g_thread_pool_free(chunkd_srv.workers, TRUE, FALSE);
+err_out_fd_info:
 	if (strict_free)
 		g_hash_table_destroy(chunkd_srv.fd_info);
 err_out_session:
diff --git a/include/chunk_msg.h b/include/chunk_msg.h
index 4a3c15d..3ee1c6a 100644
--- a/include/chunk_msg.h
+++ b/include/chunk_msg.h
@@ -48,6 +48,8 @@ enum chunksrv_ops {
 	 * functions' success/failure is sufficient indication.
 	 */
 	CHO_START_TLS		= 10,	/* Encrypt all subsequent msgs */
+
+	CHO_CP			= 11,	/* local object copy (intra-table) */
 };
 
 enum chunk_errcode {
diff --git a/include/chunkc.h b/include/chunkc.h
index 683992e..005ba58 100644
--- a/include/chunkc.h
+++ b/include/chunkc.h
@@ -85,6 +85,9 @@ extern bool stc_put_sync(struct st_client *stc);
 extern bool stc_put_inline(struct st_client *stc, const void *key,
 			   size_t key_len, void *data, uint64_t len,
 			   uint32_t flags);
+extern bool stc_cp(struct st_client *stc,
+		   const void *dest_key, size_t dest_key_len,
+		   const void *src_key, size_t src_key_len);
 
 extern bool stc_del(struct st_client *stc, const void *key, size_t key_len);
 extern bool stc_ping(struct st_client *stc);
@@ -133,4 +136,11 @@ static inline bool stc_table_openz(struct st_client *stc, const char *key,
 	return stc_table_open(stc, key, strlen(key) + 1, flags);
 }
 
+static inline bool stc_cpz(struct st_client *stc,
+			   const char *dest_key, const char *src_key)
+{
+	return stc_cp(stc, dest_key, strlen(dest_key),
+		      src_key, strlen(src_key));
+}
+
 #endif /* __STC_H__ */
diff --git a/lib/chunkdc.c b/lib/chunkdc.c
index 7441662..260f4d9 100644
--- a/lib/chunkdc.c
+++ b/lib/chunkdc.c
@@ -1091,6 +1091,62 @@ bool stc_check_status(struct st_client *stc, struct chunk_check_status *out)
 	return true;
 }
 
+bool stc_cp(struct st_client *stc,
+	    const void *dest_key, size_t dest_key_len,
+	    const void *src_key, size_t src_key_len)
+{
+	struct chunksrv_resp resp;
+	struct chunksrv_req *req;
+	void *p;
+	bool rcb = false;
+	size_t alloc_len;
+
+	if (stc->verbose)
+		fprintf(stderr, "libstc: CP\n");
+
+	alloc_len = sizeof(*req) + src_key_len + dest_key_len;
+	req = malloc(alloc_len);
+	if (!req)
+		return false;
+
+	/* initialize request */
+	req_init(stc, req);
+	req->op = CHO_CP;
+	req->data_len = cpu_to_le64(src_key_len);
+
+	/* store destination (new) key in key (1st) buffer area */
+	req_set_key(req, dest_key, dest_key_len);
+
+	/* store source (old) key in data (2nd) buffer area */
+	p = (req + 1);
+	p += dest_key_len;
+	memcpy(p, src_key, src_key_len);
+
+	/* sign request */
+	chreq_sign(req, stc->key, req->sig);
+
+	/* write request */
+	if (!net_write(stc, req, alloc_len))
+		goto out;
+
+	/* read response header */
+	if (!net_read(stc, &resp, sizeof(resp)))
+		goto out;
+
+	/* check response code */
+	if (resp.resp_code != che_Success) {
+		if (stc->verbose)
+			fprintf(stderr, "CP resp code: %d\n", resp.resp_code);
+		goto out;
+	}
+
+	rcb = true;
+
+out:
+	free(req);
+	return rcb;
+}
+
 /*
  * For extra safety, call stc_init after g_thread_init, if present.
  * Currently we just call srand(), but since we use GLib, we may need
diff --git a/test/chunkd/Makefile.am b/test/chunkd/Makefile.am
index 1cb9875..390462b 100644
--- a/test/chunkd/Makefile.am
+++ b/test/chunkd/Makefile.am
@@ -24,13 +24,14 @@ TESTS =				\
 	nop			\
 	basic-object		\
 	auth			\
+	cp			\
 	large-object		\
 	lotsa-objects		\
 	selfcheck-unit		\
 	stop-daemon		\
 	clean-db
 
-check_PROGRAMS		= auth basic-object it-works large-object \
+check_PROGRAMS		= auth basic-object cp it-works large-object \
 			  lotsa-objects nop objcache-unit selfcheck-unit
 
 TESTLDADD		= ../../lib/libhail.la	\
@@ -38,6 +39,7 @@ TESTLDADD		= ../../lib/libhail.la	\
 			  @GLIB_LIBS@ @CRYPTO_LIBS@ \
 			  @XML_LIBS@ @SSL_LIBS@
 basic_object_LDADD	= $(TESTLDADD)
+cp_LDADD		= $(TESTLDADD)
 auth_LDADD		= $(TESTLDADD)
 it_works_LDADD		= $(TESTLDADD)
 large_object_LDADD	= $(TESTLDADD)
diff --git a/test/chunkd/cp.c b/test/chunkd/cp.c
new file mode 100644
index 0000000..85f61c4
--- /dev/null
+++ b/test/chunkd/cp.c
@@ -0,0 +1,116 @@
+
+/*
+ * Copyright 2009-2010 Red Hat, Inc.
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; see the file COPYING.  If not, write to
+ * the Free Software Foundation, 675 Mass Ave, Cambridge, MA 02139, USA.
+ *
+ */
+
+#define _GNU_SOURCE
+#include "hail-config.h"
+
+#include <sys/types.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <time.h>
+#include <string.h>
+#include <locale.h>
+#include <chunkc.h>
+#include "test.h"
+
+static void test(bool do_encrypt)
+{
+	struct st_object *obj;
+	struct st_keylist *klist;
+	struct st_client *stc;
+	int port;
+	bool rcb;
+	char val[] = "my first value";
+	char key[64] = "deadbeef";
+	char key2[64] = "deadcab0";
+	size_t len = 0;
+	void *mem;
+
+	port = stc_readport(TEST_PORTFILE);
+	OK(port > 0);
+
+	stc = stc_new(TEST_HOST, port, TEST_USER, TEST_USER_KEY, do_encrypt);
+	OK(stc);
+
+	rcb = stc_table_openz(stc, TEST_TABLE, 0);
+	OK(rcb);
+
+	/* store object */
+	rcb = stc_put_inlinez(stc, key, val, strlen(val), 0);
+	OK(rcb);
+
+	/* make sure object appears in list of volume keys */
+	klist = stc_keys(stc);
+	OK(klist);
+	OK(klist->contents);
+	OK(klist->contents->next == NULL);
+
+	obj = klist->contents->data;
+	OK(obj);
+	OK(obj->name);
+	OK(!strcmp(obj->name, key));
+	OK(obj->time_mod);
+	OK(obj->etag);
+	OK(obj->size == strlen(val));
+	OK(obj->owner);
+
+	stc_free_keylist(klist);
+
+	/* get object */
+	mem = stc_get_inlinez(stc, key, &len);
+	OK(mem);
+	OK(len == strlen(val));
+	OK(!memcmp(val, mem, strlen(val)));
+
+	free(mem);
+
+	/* copy object */
+	rcb = stc_cpz(stc, key2, key);
+	OK(rcb);
+
+	/* get object copy */
+	mem = stc_get_inlinez(stc, key2, &len);
+	OK(mem);
+	OK(len == strlen(val));
+	OK(!memcmp(val, mem, strlen(val)));
+
+	free(mem);
+
+	/* delete objects */
+	rcb = stc_delz(stc, key);
+	OK(rcb);
+	rcb = stc_delz(stc, key2);
+	OK(rcb);
+
+	stc_free(stc);
+}
+
+int main(int argc, char *argv[])
+{
+	setlocale(LC_ALL, "C");
+
+	stc_init();
+	SSL_library_init();
+	SSL_load_error_strings();
+
+	test(false);
+	test(true);
+
+	return 0;
+}

Re: [PATCH] chunkd: add cp command, for local intra-table copies

[Pete Zaitcev]

On Tue, 6 Jul 2010 03:24:29 -0400
Jeff Garzik <jeff@garzik.org> wrote:

> The following patch, against current hail.git, adds the "CP" command to
> chunkd, permitting copying from object->object inside a single table.

What is it for?

-- Pete

Re: [PATCH] chunkd: add cp command, for local intra-table copies

[Jeff Garzik]

On 07/06/2010 11:17 AM, Pete Zaitcev wrote:
> On Tue, 6 Jul 2010 03:24:29 -0400
> Jeff Garzik<jeff@garzik.org>  wrote:
>
>> The following patch, against current hail.git, adds the "CP" command to
>> chunkd, permitting copying from object->object inside a single table.
>
> What is it for?

Fun!  :)

More seriously, it is mainly an infrastructure patch, adding things that 
the upcoming RCP command will use.  As CP is far less complex, this 
allows me to verify several bits of machinery before moving forward.  I 
imagine CP will be tangentially helpful, but not a crucial feature in 
and of itself.

	Jeff

Re: [PATCH] chunkd: add cp command, for local intra-table copies

[Jeff Garzik]

On 07/06/2010 11:17 AM, Pete Zaitcev wrote:
> On Tue, 6 Jul 2010 03:24:29 -0400
> Jeff Garzik<jeff@garzik.org>  wrote:
>
>> The following patch, against current hail.git, adds the "CP" command to
>> chunkd, permitting copying from object->object inside a single table.
>
> What is it for?

Here's a real-world example.

Quoting from the S3 documentation, this describes the "PUT (copy)" 
operation, something that tabled does not yet support, but should:

	This implementation of the PUT operation creates a copy of an
	object that is already stored in Amazon S3. A PUT copy
	operation is the same as performing a GET and then a PUT.
	Adding the request header, x-amz-copy-source, makes the PUT
	operation copy the source object into the destination bucket.

Assuming that a given tabled object is already fully replicated -- 
HOPEFULLY the common case for us -- the least expensive way to implement 
this is

	for each chunkd containing object OLD_KEY

		CHO_CP(object OLD_KEY -> object NEW_KEY)

Assuming each chunkd node has the necessary free space, this method 
totally avoids using network bandwidth, when creating a copy of an object

	Jeff