Merge lp:~clint-fewbar/gearmand/round-robin into lp:gearmand/1.0

[Clint Byrum]

[Clint Byrum (clint-fewbar)]

Adds tests for round-robin behavior

=== modified file '.bzrignore'
--- .bzrignore	2010-01-26 05:53:44 +0000
+++ .bzrignore	2010-02-16 17:23:15 +0000
@@ -72,3 +72,5 @@
 tests/worker_test.res
 tests/tokyocabinet_test
 tests/tokyocabinet_test.res
+tests/round_robin_test
+tests/round_robin_test.res

=== modified file 'tests/include.am'
--- tests/include.am	2010-02-10 01:48:52 +0000
+++ tests/include.am	2010-02-16 17:23:16 +0000
@@ -55,11 +55,20 @@
 tests_tokyocabinet_test_LDADD= ${TEST_LDADD}
 
 noinst_PROGRAMS+= \
+<<<<<<< TREE
 		  tests/client_test \
 		  tests/cpp_test \
 		  tests/internals_test \
 		  tests/regression_test \
 		  tests/worker_test
+=======
+		tests/client_test \
+		tests/cpp_test \
+		tests/internals_test \
+		tests/regression_test \
+    tests/round_robin_test \
+		tests/worker_test
+>>>>>>> MERGE-SOURCE
 
 noinst_HEADERS+= \
 		 tests/test.h \
@@ -75,6 +84,9 @@
 tests_regression_test_SOURCES= tests/regression.c
 tests_regression_test_LDADD= ${TEST_LDADD} libgearman/libgearmancore.la
 
+tests_round_robin_test_SOURCES= tests/round_robin.c
+tests_round_robin_test_LDADD= ${TEST_LDADD} libgearman/libgearmancore.la
+
 tests_worker_test_SOURCES= tests/worker_test.c
 tests_worker_test_LDADD= ${TEST_LDADD}
 
@@ -85,6 +97,9 @@
 test-client:
 	@tests/client_test $(ARG1) $(ARG2)
 
+test-round-robin:
+	tests/round_robin_test $(ARG1) $(ARG2)
+
 test-worker:
 	@tests/worker_test $(ARG1) $(ARG2)
 
@@ -104,11 +119,14 @@
 	$(TOKYOCABINET_TEST) $(ARG1) $(ARG2)
 	$(TOKYOCABINET_RM)
 
-check-local: test-client test-worker test-internals test-libmemcached test-sqlite test-tokyocabinet
+check-local: test-client test-round-robin test-worker test-internals test-libmemcached test-sqlite test-tokyocabinet
 
 gdb-client: ${noinst_PROGRAMS}
 	$(LIBTOOL) --mode=execute gdb tests/client_test
 
+gdb-round-robin: ${noinst_PROGRAMS}
+	$(LIBTOOL) --mode=execute gdb tests/round_robin_test
+
 gdb-worker: ${noinst_PROGRAMS}
 	$(LIBTOOL) --mode=execute gdb tests/worker_test
 
@@ -118,6 +136,9 @@
 valgrind-client:
 	$(VALGRIND_COMMAND) tests/client_test $(ARG1) $(ARG2)
 
+valgrind-round-robin:
+	$(VALGRIND_COMMAND) tests/round_robin_test $(ARG1) $(ARG2)
+
 valgrind-worker:
 	$(VALGRIND_COMMAND) tests/worker_test $(ARG1) $(ARG2)
 
@@ -137,6 +158,7 @@
 	$(VALGRIND_COMMAND) $(TOKYOCABINET_TEST) $(ARG1) $(ARG2)
 	$(TOKYOCABINET_RM)
 
+<<<<<<< TREE
 GEARMAN_CLIENT_TEST= bin/gearman
 GEARMAN_PIDFILE = ${abs_top_builddir}/tests/Xugear.pid
 
@@ -148,3 +170,6 @@
 
 valgrind: ${noinst_PROGRAMS} valgrind-client valgrind-worker valgrind-internals valgrind-sqlite valgrind-tokyocabinetn
 
+=======
+valgrind: ${noinst_PROGRAMS} valgrind-client valgrind-round-robin valgrind-worker valgrind-internals valgrind-sqlite valgrind-tokyocabinetn
+>>>>>>> MERGE-SOURCE

=== added file 'tests/round_robin.c'
--- tests/round_robin.c	1970-01-01 00:00:00 +0000
+++ tests/round_robin.c	2010-02-16 17:23:16 +0000
@@ -0,0 +1,199 @@
+/* Gearman server and library
+ * Copyright (C) 2008 Brian Aker, Eric Day
+ * All rights reserved.
+ *
+ * Use and distribution licensed under the BSD license.  See
+ * the COPYING file in the parent directory for full text.
+ */
+
+#include "config.h"
+
+#if defined(NDEBUG)
+# undef NDEBUG
+#endif
+
+#include <assert.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <libgearman/gearman.h>
+#include "test.h"
+#include "test_gearmand.h"
+
+#define WORKER_TEST_PORT 32123
+
+typedef struct
+{
+  pid_t gearmand_pid;
+  gearman_worker_st worker;
+  bool run_worker;
+} worker_test_st;
+
+/* Prototypes */
+test_return_t queue_add(void *object);
+test_return_t queue_worker(void *object);
+
+test_return_t pre(void *object);
+test_return_t post(void *object);
+
+void *world_create(test_return_t *error);
+test_return_t world_destroy(void *object);
+
+/* append test for worker */
+static void *append_function(gearman_job_st *job __attribute__((unused)),
+                             void *context, size_t *result_size,
+                             gearman_return_t *ret_ptr __attribute__((unused)))
+{
+  /* this will will set the last char in the context (buffer) to the */
+  /* first char of the work */
+  char * buf = (char *)context;
+  char * work = (char *)gearman_job_workload(job);
+  buf += strlen(buf);
+  *buf = *work;
+  *result_size= 0;
+  return NULL;
+}
+
+test_return_t queue_add(void *object)
+{
+  worker_test_st *test= (worker_test_st *)object;
+  gearman_client_st client;
+  char job_handle[GEARMAN_JOB_HANDLE_SIZE];
+
+  uint8_t * value= (uint8_t *)strdup("0");
+  size_t value_length= 1;
+  uint8_t i;
+
+  test->run_worker= false;
+
+  if (gearman_client_create(&client) == NULL)
+    return TEST_FAILURE;
+
+  if (gearman_client_add_server(&client, NULL,
+                                WORKER_TEST_PORT) != GEARMAN_SUCCESS)
+  {
+    return TEST_FAILURE;
+  }
+
+  /* send strings "0", "1" ... "9" to alternating between 2 queues */
+  /* queue1 = 1,3,5,7,9 */
+  /* queue2 = 0,2,4,6,8 */
+  for (i=0; i<10; i++) {
+    if (gearman_client_do_background(&client, i % 2 ? "queue1" : "queue2", NULL, value,
+                                     value_length, job_handle) != GEARMAN_SUCCESS)
+    {
+      return TEST_FAILURE;
+    }
+    *value = (uint8_t)(*value + 1);
+  }
+
+  gearman_client_free(&client);
+  free(value);
+
+  test->run_worker= true;
+  return TEST_SUCCESS;
+}
+
+test_return_t queue_worker(void *object)
+{
+  worker_test_st *test= (worker_test_st *)object;
+  gearman_worker_st *worker= &(test->worker);
+  char buffer[11];
+  int i;
+  memset(buffer, 0, 11);
+
+  if (!test->run_worker)
+    return TEST_FAILURE;
+
+  if (gearman_worker_add_function(worker, "queue1", 5, append_function,
+                                  buffer) != GEARMAN_SUCCESS)
+  {
+    return TEST_FAILURE;
+  }
+
+  if (gearman_worker_add_function(worker, "queue2", 5, append_function,
+                                  buffer) != GEARMAN_SUCCESS)
+  {
+    return TEST_FAILURE;
+  }
+
+  for (i=0; i<10; i++) {
+    if (gearman_worker_work(worker) != GEARMAN_SUCCESS)
+      return TEST_FAILURE;
+  }
+
+  // expect buffer to be reassembled in a predictable round robin order
+  if( strcmp(buffer, "1032547698") ) 
+  {
+    fprintf(stderr, "\n\nexpecting 0123456789, got %s\n\n", buffer);
+    return TEST_FAILURE;
+  }
+
+  return TEST_SUCCESS;
+}
+
+
+void *world_create(test_return_t *error)
+{
+  worker_test_st *test;
+  const char *argv[2]= { "test_gearmand", "--round-robin"};
+  pid_t gearmand_pid;
+
+  gearmand_pid= test_gearmand_start(WORKER_TEST_PORT, NULL, (char **)argv, 2);
+
+  test= malloc(sizeof(worker_test_st));
+  if (! test)
+  {
+    *error= TEST_MEMORY_ALLOCATION_FAILURE;
+    return NULL;
+  }
+
+  memset(test, 0, sizeof(worker_test_st));
+  if (gearman_worker_create(&(test->worker)) == NULL)
+  {
+    *error= TEST_FAILURE;
+    return NULL;
+  }
+
+  if (gearman_worker_add_server(&(test->worker), NULL, WORKER_TEST_PORT) != GEARMAN_SUCCESS)
+  {
+    *error= TEST_FAILURE;
+    return NULL;
+  }
+
+  test->gearmand_pid= gearmand_pid;
+
+  *error= TEST_SUCCESS;
+
+  return (void *)test;
+}
+
+test_return_t world_destroy(void *object)
+{
+  worker_test_st *test= (worker_test_st *)object;
+  gearman_worker_free(&(test->worker));
+  test_gearmand_stop(test->gearmand_pid);
+  free(test);
+
+  return TEST_SUCCESS;
+}
+
+test_st tests[] ={
+  {"add", 0, queue_add },
+  {"worker", 0, queue_worker },
+  {0, 0, 0}
+};
+
+collection_st collection[] ={
+  {"round_robin", 0, 0, tests},
+  {0, 0, 0, 0}
+};
+
+void get_world(world_st *world)
+{
+  world->collections= collection;
+  world->create= world_create;
+  world->destroy= world_destroy;
+}

=== modified file 'tests/test_gearmand.c'
--- tests/test_gearmand.c	2010-01-28 21:45:14 +0000
+++ tests/test_gearmand.c	2010-02-16 17:23:16 +0000
@@ -46,6 +46,10 @@
   pid_t gearmand_pid;
   gearmand_st *gearmand;
   gearman_conf_st conf;
+  gearman_conf_module_st module;
+  const char *name;
+  const char *value;
+  bool round_robin= false;
 
   gearmand_pid= fork();
   assert(gearmand_pid != -1);
@@ -74,12 +78,32 @@
     assert(ret == GEARMAN_SUCCESS);
 #endif
 
+    (void)gearman_conf_module_create(&conf, &module, NULL);
+
+#ifndef MCO
+#define MCO(__name, __short, __value, __help) \
+      gearman_conf_module_add_option(&module, __name, __short, __value, __help);
+#endif
+
+  MCO("round-robin", 'R', NULL, "Assign work in round-robin order per worker"
+      "connection. The default is to assign work in the order of functions "
+      "added by the worker.")
+
     ret= gearman_conf_parse_args(&conf, argc, argv);
     assert(ret == GEARMAN_SUCCESS);
 
+    /* Check for option values that were given. */
+    while (gearman_conf_module_value(&module, &name, &value))
+    {
+      if (!strcmp(name, "round-robin"))
+        round_robin++;
+    }
+
     gearmand= gearmand_create(NULL, port);
     assert(gearmand != NULL);
 
+    gearmand_set_round_robin(gearmand, round_robin);
+
     if (queue_type != NULL)
     {
       assert(argc);
@@ -121,6 +145,8 @@
       }
     }
 
+    
+
     ret= gearmand_run(gearmand);
     assert(ret != GEARMAN_SUCCESS);