X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=tests%2Ffunction.c;fp=tests%2Ffunction.c;h=c505b1118088247d28f67aca9449ba6ab0079bcf;hb=f63d634e21459280599f663154089fa724903bd6;hp=fcf5b51104008b27b905923d8bec0103ff78565a;hpb=020e1c454c37818161ae905749a55a1568b93f15;p=m6w6%2Flibmemcached

diff --git a/tests/function.c b/tests/function.c
index fcf5b511..c505b111 100644
--- a/tests/function.c
+++ b/tests/function.c
@@ -1552,6 +1552,9 @@ static test_return  behavior_test(memcached_st *memc)
   value= memcached_behavior_get(memc, MEMCACHED_BEHAVIOR_SOCKET_RECV_SIZE);
   assert(value > 0);
 
+  value= memcached_behavior_get(memc, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS);
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS, value + 1);
+  assert((value + 1) == memcached_behavior_get(memc, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS));
   return 0;
 }
 
@@ -3049,6 +3052,24 @@ static memcached_return  pre_binary(memcached_st *memc)
   return rc;
 }
 
+static memcached_return pre_replication(memcached_st *memc)
+{
+  memcached_return rc= MEMCACHED_FAILURE;
+  if (pre_binary(memc) == MEMCACHED_SUCCESS) 
+  {
+    /*
+     * Make sure that we store the item on all servers 
+     * (master + replicas == number of servers) 
+     */
+    rc= memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS, 
+                               memc->number_of_hosts - 1);
+    assert(rc == MEMCACHED_SUCCESS);
+    assert(memcached_behavior_get(memc, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS) == memc->number_of_hosts - 1);
+  }
+
+  return rc;
+}
+
 static void my_free(memcached_st *ptr __attribute__((unused)), void *mem)
 {
   free(mem);
@@ -3550,6 +3571,179 @@ static test_return connection_pool_test(memcached_st *memc)
 }
 #endif
 
+static test_return replication_set_test(memcached_st *memc)
+{
+  memcached_return rc;
+  memcached_st *clone= memcached_clone(NULL, memc);
+  memcached_behavior_set(clone, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS, 0);
+
+  rc= memcached_set(memc, "bubba", 5, "0", 1, 0, 0);
+  assert(rc == MEMCACHED_SUCCESS);
+
+  /*
+  ** "bubba" should now be stored on all of our servers. We don't have an
+  ** easy to use API to address each individual server, so I'll just iterate
+  ** through a bunch of "master keys" and I should most likely hit all of the
+  ** servers...
+  */
+  for (int x= 'a'; x <= 'z'; ++x)
+  {
+    char key[2]= { [0]= (char)x };
+    size_t len;
+    uint32_t flags;
+    char *val= memcached_get_by_key(clone, key, 1, "bubba", 5, 
+                                    &len, &flags, &rc);
+    assert(rc == MEMCACHED_SUCCESS);
+    assert(val != NULL);
+    free(val);
+  }
+
+  memcached_free(clone);
+
+  return TEST_SUCCESS;
+}
+
+static test_return replication_get_test(memcached_st *memc)
+{
+  memcached_return rc;
+
+  /*
+   * Don't do the following in your code. I am abusing the internal details
+   * within the library, and this is not a supported interface.
+   * This is to verify correct behavior in the library
+   */
+  for (uint32_t host= 0; host < memc->number_of_hosts; ++host) 
+  {
+    memcached_st *clone= memcached_clone(NULL, memc);
+    clone->hosts[host].port= 0;
+
+    for (int x= 'a'; x <= 'z'; ++x)
+    {
+      char key[2]= { [0]= (char)x };
+      size_t len;
+      uint32_t flags;
+      char *val= memcached_get_by_key(clone, key, 1, "bubba", 5, 
+                                      &len, &flags, &rc);
+      assert(rc == MEMCACHED_SUCCESS);
+      assert(val != NULL);
+      free(val);
+    }
+
+    memcached_free(clone);
+  }
+
+  return TEST_SUCCESS;
+}
+
+static test_return replication_mget_test(memcached_st *memc)
+{
+  memcached_return rc;
+  memcached_st *clone= memcached_clone(NULL, memc);
+  memcached_behavior_set(clone, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS, 0);
+
+  char *keys[]= { "bubba", "key1", "key2", "key3" };
+  size_t len[]= { 5, 4, 4, 4 };
+
+  for (int x=0; x< 4; ++x)
+  {
+    rc= memcached_set(memc, keys[x], len[x], "0", 1, 0, 0);
+    assert(rc == MEMCACHED_SUCCESS);
+  }
+
+  /*
+   * Don't do the following in your code. I am abusing the internal details
+   * within the library, and this is not a supported interface.
+   * This is to verify correct behavior in the library
+   */
+  memcached_result_st result_obj;
+  for (uint32_t host= 0; host < clone->number_of_hosts; host++) 
+  {
+    memcached_st *new_clone= memcached_clone(NULL, memc);
+    new_clone->hosts[host].port= 0;
+
+    for (int x= 'a'; x <= 'z'; ++x)
+    {
+      char key[2]= { [0]= (char)x };
+
+      rc= memcached_mget_by_key(new_clone, key, 1, keys, len, 4);
+      assert(rc == MEMCACHED_SUCCESS);
+
+      memcached_result_st *results= memcached_result_create(new_clone, &result_obj);
+      assert(results);
+
+      int hits= 0;
+      while ((results= memcached_fetch_result(new_clone, &result_obj, &rc)) != NULL)
+      {
+        hits++;
+      }
+      assert(hits == 4);
+      memcached_result_free(&result_obj);
+    }
+
+    memcached_free(new_clone);
+  }
+
+  return TEST_SUCCESS;
+}
+
+static test_return replication_delete_test(memcached_st *memc)
+{
+  memcached_return rc;
+  memcached_st *clone= memcached_clone(NULL, memc);
+  /* Delete the items from all of the servers except 1 */
+  uint64_t repl= memcached_behavior_get(memc,
+                                        MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS);
+  memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NUMBER_OF_REPLICAS, --repl);
+
+  char *keys[]= { "bubba", "key1", "key2", "key3" };
+  size_t len[]= { 5, 4, 4, 4 };
+
+  for (int x=0; x< 4; ++x)
+  {
+    rc= memcached_delete_by_key(memc, keys[0], len[0], keys[x], len[x], 0);
+    assert(rc == MEMCACHED_SUCCESS);
+  }
+
+  /*
+   * Don't do the following in your code. I am abusing the internal details
+   * within the library, and this is not a supported interface.
+   * This is to verify correct behavior in the library
+   */
+  uint32_t hash= memcached_generate_hash(memc, keys[0], len[0]);
+  for (uint32_t x= 0; x < (repl + 1); ++x) 
+  {
+    clone->hosts[hash].port= 0;
+    if (++hash == clone->number_of_hosts)
+      hash= 0;
+  }
+
+  memcached_result_st result_obj;
+  for (uint32_t host= 0; host < clone->number_of_hosts; ++host) 
+  {
+    for (int x= 'a'; x <= 'z'; ++x)
+    {
+      char key[2]= { [0]= (char)x };
+
+      rc= memcached_mget_by_key(clone, key, 1, keys, len, 4);
+      assert(rc == MEMCACHED_SUCCESS);
+
+      memcached_result_st *results= memcached_result_create(clone, &result_obj);
+      assert(results);
+
+      int hits= 0;
+      while ((results= memcached_fetch_result(clone, &result_obj, &rc)) != NULL)
+      {
+        ++hits;
+      }
+      assert(hits == 4);
+      memcached_result_free(&result_obj);
+    }
+  }
+  memcached_free(clone);
+
+  return TEST_SUCCESS;
+}
+
 static void increment_request_id(uint16_t *id)
 {
   (*id)++;
@@ -3562,6 +3756,7 @@ static uint16_t *get_udp_request_ids(memcached_st *memc)
   uint16_t *ids= malloc(sizeof(uint16_t) * memc->number_of_hosts);
   assert(ids != NULL);
   unsigned int x;
+
   for (x= 0; x < memc->number_of_hosts; x++)
     ids[x]= get_udp_datagram_request_id((struct udp_datagram_header_st *) memc->hosts[x].write_buffer);
 
@@ -3573,6 +3768,7 @@ static test_return post_udp_op_check(memcached_st *memc, uint16_t *expected_req_
   unsigned int x;
   memcached_server_st *cur_server = memc->hosts;
   uint16_t *cur_req_ids = get_udp_request_ids(memc);
+
   for (x= 0; x < memc->number_of_hosts; x++)
   {
     assert(cur_server[x].cursor_active == 0);
@@ -3580,6 +3776,7 @@ static test_return post_udp_op_check(memcached_st *memc, uint16_t *expected_req_
   }
   free(expected_req_ids);
   free(cur_req_ids);
+
   return TEST_SUCCESS;
 }
 
@@ -3601,6 +3798,7 @@ static memcached_return init_udp(memcached_st *memc)
   memcpy(servers, memc->hosts, sizeof(memcached_server_st) * num_hosts);
   for (x= 0; x < num_hosts; x++)
     memcached_server_free(&memc->hosts[x]);
+
   memc->number_of_hosts= 0;
   memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_USE_UDP, 1);
   for (x= 0; x < num_hosts; x++)
@@ -3608,6 +3806,7 @@ static memcached_return init_udp(memcached_st *memc)
     assert(memcached_server_add_udp(memc, servers[x].hostname, servers[x].port) == MEMCACHED_SUCCESS);
     assert(memc->hosts[x].write_buffer_offset == UDP_DATAGRAM_HEADER_LENGTH);
   }
+
   return MEMCACHED_SUCCESS;
 }
 
@@ -4255,6 +4454,14 @@ test_st user_tests[] ={
   {0, 0, 0}
 };
 
+test_st replication_tests[]= {
+  {"set", 1, replication_set_test },
+  {"get", 0, replication_get_test },
+  {"mget", 0, replication_mget_test },
+  {"delete", 0, replication_delete_test },
+  {0, 0, 0}
+};
+
 test_st generate_tests[] ={
   {"generate_pairs", 1, generate_pairs },
   {"generate_data", 1, generate_data },
@@ -4359,6 +4566,7 @@ collection_st collection[] ={
   {"consistent_ketama", pre_behavior_ketama, 0, consistent_tests},
   {"consistent_ketama_weighted", pre_behavior_ketama_weighted, 0, consistent_weighted_tests},
   {"test_hashes", 0, 0, hash_tests},
+  {"replication", pre_replication, 0, replication_tests},
   {0, 0, 0, 0}
 };