git.m6w6.name Git - m6w6/libmemcached/blob - src/libmemcached/instance.cc

   1 /*
   2     +--------------------------------------------------------------------+
   3     | libmemcached - C/C++ Client Library for memcached                  |
   4     +--------------------------------------------------------------------+
   5     | Redistribution and use in source and binary forms, with or without |
   6     | modification, are permitted under the terms of the BSD license.    |
   7     | You should have received a copy of the license in a bundled file   |
   8     | named LICENSE; in case you did not receive a copy you can review   |
   9     | the terms online at: https://opensource.org/licenses/BSD-3-Clause  |
  10     +--------------------------------------------------------------------+
  11     | Copyright (c) 2006-2014 Brian Aker   https://datadifferential.com/ |
  12     | Copyright (c) 2020 Michael Wallner   <mike@php.net>                |
  13     +--------------------------------------------------------------------+
  14 */
  15
  16 #include "libmemcached/common.h"
  17 #include "p9y/socket.hpp"
  18
  19 static inline void _server_init(memcached_instance_st *self, Memcached *root,
  20                                 const memcached_string_t &hostname, in_port_t port, uint32_t weight,
  21                                 memcached_connection_t type) {
  22   self->options.is_shutting_down = false;
  23   self->options.is_dead = false;
  24   self->options.ready = false;
  25   self->_events = 0;
  26   self->_revents = 0;
  27   self->cursor_active_ = 0;
  28   self->port_ = port;
  29   self->fd = INVALID_SOCKET;
  30   self->io_bytes_sent = 0;
  31   self->request_id = 0;
  32   self->server_failure_counter = 0;
  33   self->server_failure_counter_query_id = 0;
  34   self->server_timeout_counter = 0;
  35   self->server_timeout_counter_query_id = 0;
  36   self->weight = weight ? weight : 1; // 1 is the default weight value
  37   self->io_wait_count.read = 0;
  38   self->io_wait_count.write = 0;
  39   self->io_wait_count.timeouts = 0;
  40   self->io_wait_count._bytes_read = 0;
  41   self->major_version = UINT8_MAX;
  42   self->micro_version = UINT8_MAX;
  43   self->minor_version = UINT8_MAX;
  44   self->type = type;
  45   self->error_messages = NULL;
  46   self->read_ptr = self->read_buffer;
  47   self->read_buffer_length = 0;
  48   self->write_buffer_offset = 0;
  49   self->address_info = NULL;
  50   self->address_info_next = NULL;
  51
  52   self->state = MEMCACHED_SERVER_STATE_NEW;
  53   self->next_retry = 0;
  54
  55   self->root = root;
  56   if (root) {
  57     self->version = ++root->server_info.version;
  58   } else {
  59     self->version = UINT_MAX;
  60   }
  61   self->limit_maxbytes = 0;
  62   self->hostname(hostname);
  63 }
  64
  65 static memcached_instance_st *_server_create(memcached_instance_st *self,
  66                                              const memcached_st *memc) {
  67   if (self == NULL) {
  68     self = libmemcached_xmalloc(memc, memcached_instance_st);
  69
  70     if (self == NULL) {
  71       return NULL; /*  MEMCACHED_MEMORY_ALLOCATION_FAILURE */
  72     }
  73
  74     self->options.is_allocated = true;
  75   } else {
  76     self->options.is_allocated = false;
  77   }
  78
  79   self->options.is_initialized = true;
  80
  81   return self;
  82 }
  83
  84 void memcached_instance_st::events(short arg) {
  85   if ((_events | arg) == _events) {
  86     return;
  87   }
  88
  89   _events |= arg;
  90 }
  91
  92 void memcached_instance_st::revents(short arg) {
  93   if (arg) {
  94     options.ready = true;
  95   }
  96
  97   _revents = arg;
  98   _events &= short(~arg);
  99 }
 100
 101 memcached_instance_st *__instance_create_with(memcached_st *memc, memcached_instance_st *self,
 102                                               const memcached_string_t &hostname,
 103                                               const in_port_t port, uint32_t weight,
 104                                               const memcached_connection_t type) {
 105   if (memcached_is_valid_servername(hostname) == false) {
 106     memcached_set_error(*memc, MEMCACHED_INVALID_ARGUMENTS, MEMCACHED_AT,
 107                         memcached_literal_param("Invalid hostname provided"));
 108     return NULL;
 109   }
 110
 111   self = _server_create(self, memc);
 112
 113   if (self == NULL) {
 114     return NULL;
 115   }
 116
 117   _server_init(self, const_cast<memcached_st *>(memc), hostname, port, weight, type);
 118
 119   if (memc and memcached_is_udp(memc)) {
 120     self->write_buffer_offset = UDP_DATAGRAM_HEADER_LENGTH;
 121     memcached_io_init_udp_header(self, 0);
 122   }
 123
 124   return self;
 125 }
 126
 127 void __instance_free(memcached_instance_st *self) {
 128   memcached_quit_server(self, false);
 129
 130   self->clear_addrinfo();
 131   assert(self->address_info_next == NULL);
 132
 133   memcached_error_free(*self);
 134
 135   if (memcached_is_allocated(self)) {
 136     libmemcached_free(self->root, self);
 137   } else {
 138     self->options.is_initialized = false;
 139   }
 140 }
 141
 142 void memcached_instance_free(memcached_instance_st *self) {
 143   if (self) {
 144     __instance_free(self);
 145   }
 146 }
 147
 148 memcached_return_t memcached_server_cursor(const memcached_st *shell,
 149                                            const memcached_server_fn *callback, void *context,
 150                                            uint32_t number_of_callbacks) {
 151   const Memcached *memc = memcached2Memcached(shell);
 152   memcached_return_t rc;
 153   if (memcached_failed(rc = initialize_const_query(memc))) {
 154     return rc;
 155   }
 156
 157   size_t errors = 0;
 158   for (uint32_t x = 0; x < memcached_instance_list_count(memc); x++) {
 159     memcached_instance_st *instance = memcached_instance_by_position(memc, x);
 160
 161     for (uint32_t y = 0; y < number_of_callbacks; y++) {
 162       memcached_return_t ret = (*callback[y])(memc, instance, context);
 163
 164       if (memcached_failed(ret)) {
 165         errors++;
 166         continue;
 167       }
 168     }
 169   }
 170
 171   return errors ? MEMCACHED_SOME_ERRORS : MEMCACHED_SUCCESS;
 172 }
 173
 174 memcached_return_t memcached_server_execute(memcached_st *memc,
 175                                             memcached_server_execute_fn callback, void *context) {
 176   if (callback == NULL) {
 177     return MEMCACHED_INVALID_ARGUMENTS;
 178   }
 179
 180   bool some_errors = false;
 181   ;
 182   for (uint32_t x = 0; x < memcached_instance_list_count(memc); x++) {
 183     memcached_instance_st *instance = memcached_instance_fetch(memc, x);
 184
 185     memcached_return_t rc = (*callback)(memc, instance, context);
 186     if (rc == MEMCACHED_INVALID_ARGUMENTS) {
 187       return rc;
 188     } else if (memcached_fatal(rc)) {
 189       some_errors = true;
 190     }
 191   }
 192
 193   (void) some_errors;
 194   return MEMCACHED_SUCCESS;
 195 }
 196
 197 const memcached_instance_st *memcached_server_by_key(memcached_st *shell, const char *key,
 198                                                      size_t key_length, memcached_return_t *error) {
 199   Memcached *memc = memcached2Memcached(shell);
 200   memcached_return_t unused;
 201   if (error == NULL) {
 202     error = &unused;
 203   }
 204
 205   memcached_return_t rc;
 206   if (memcached_failed(rc = initialize_const_query(memc))) {
 207     *error = rc;
 208     return NULL;
 209   }
 210
 211   if (memcached_failed((memcached_key_test(*memc, (const char **) &key, &key_length, 1)))) {
 212     *error = memcached_last_error(memc);
 213     return NULL;
 214   }
 215
 216   uint32_t server_key = memcached_generate_hash(memc, key, key_length);
 217   return memcached_instance_by_position(memc, server_key);
 218 }
 219
 220 /*
 221   If we do not have a valid object to clone from, we toss an error.
 222 */
 223 static memcached_instance_st *memcached_instance_clone(memcached_instance_st *source) {
 224   /* We just do a normal create if source is missing */
 225   if (source == NULL) {
 226     return NULL;
 227   }
 228
 229   memcached_string_t hostname_ = {memcached_string_make_from_cstr(source->hostname())};
 230   return __instance_create_with(source->root, NULL, hostname_, source->port(), source->weight,
 231                                 source->type);
 232 }
 233
 234 void set_last_disconnected_host(memcached_instance_st *self) {
 235   assert(self->root);
 236   if (self->root) {
 237     if (memcached_server_get_last_disconnect(self->root)
 238         and memcached_server_get_last_disconnect(self->root)->version == self->version)
 239     {
 240       return;
 241     }
 242
 243     // const_cast
 244     memcached_st *root = (memcached_st *) self->root;
 245
 246     memcached_instance_free((memcached_instance_st *) (root->last_disconnected_server));
 247
 248     // We set is_parsing so that no lookup happens
 249     root->state.is_parsing = true;
 250     root->last_disconnected_server = memcached_instance_clone(self);
 251     root->state.is_parsing = false;
 252
 253     ((memcached_instance_st *) memcached_server_get_last_disconnect(root))->version = self->version;
 254   }
 255 }
 256
 257 const memcached_instance_st *memcached_server_get_last_disconnect(const memcached_st *shell) {
 258   const Memcached *self = memcached2Memcached(shell);
 259   if (self) {
 260     return (const memcached_instance_st *) self->last_disconnected_server;
 261   }
 262
 263   return 0;
 264 }
 265
 266 void memcached_instance_next_retry(const memcached_instance_st *self, const time_t absolute_time) {
 267   WATCHPOINT_ASSERT(self);
 268   if (self) {
 269     ((memcached_instance_st *) self)->next_retry = absolute_time;
 270   }
 271 }
 272
 273 bool memcached_instance_st::valid() const {
 274   if (fd == INVALID_SOCKET) {
 275     return false;
 276   }
 277
 278   return true;
 279 }
 280
 281 bool memcached_instance_st::is_shutting_down() const {
 282   return options.is_shutting_down;
 283 }