git.m6w6.name Git - awesomized/libmemcached/blob - libmemcachedprotocol/handler.c

   1 /*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
   2  *
   3  *  Libmemcached library
   4  *
   5  *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
   6  *
   7  *  Redistribution and use in source and binary forms, with or without
   8  *  modification, are permitted provided that the following conditions are
   9  *  met:
  10  *
  11  *      * Redistributions of source code must retain the above copyright
  12  *  notice, this list of conditions and the following disclaimer.
  13  *
  14  *      * Redistributions in binary form must reproduce the above
  15  *  copyright notice, this list of conditions and the following disclaimer
  16  *  in the documentation and/or other materials provided with the
  17  *  distribution.
  18  *
  19  *      * The names of its contributors may not be used to endorse or
  20  *  promote products derived from this software without specific prior
  21  *  written permission.
  22  *
  23  *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  24  *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  25  *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  26  *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  27  *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  28  *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  29  *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  30  *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  31  *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  32  *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  33  *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  34  *
  35  */
  36
  37 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
  38 #include <libmemcachedprotocol/common.h>
  39
  40 #include <stdlib.h>
  41 #include <sys/types.h>
  42 #include <errno.h>
  43 #include <stdbool.h>
  44 #include <string.h>
  45 #include <strings.h>
  46 #include <ctype.h>
  47 #include <stdio.h>
  48
  49 /*
  50 ** **********************************************************************
  51 ** INTERNAL INTERFACE
  52 ** **********************************************************************
  53 */
  54
  55 /**
  56  * The default function to receive data from the client. This function
  57  * just wraps the recv function to receive from a socket.
  58  * See man -s3socket recv for more information.
  59  *
  60  * @param cookie cookie indentifying a client, not used
  61  * @param sock socket to read from
  62  * @param buf the destination buffer
  63  * @param nbytes the number of bytes to read
  64  * @return the number of bytes transferred of -1 upon error
  65  */
  66 static ssize_t default_recv(const void *cookie,
  67                             memcached_socket_t sock,
  68                             void *buf,
  69                             size_t nbytes)
  70 {
  71   (void)cookie;
  72   return recv(sock, buf, nbytes, 0);
  73 }
  74
  75 /**
  76  * The default function to send data to the server. This function
  77  * just wraps the send function to send through a socket.
  78  * See man -s3socket send for more information.
  79  *
  80  * @param cookie cookie indentifying a client, not used
  81  * @param sock socket to send to
  82  * @param buf the source buffer
  83  * @param nbytes the number of bytes to send
  84  * @return the number of bytes transferred of -1 upon error
  85  */
  86 static ssize_t default_send(const void *cookie,
  87                             memcached_socket_t fd,
  88                             const void *buf,
  89                             size_t nbytes)
  90 {
  91   (void)cookie;
  92   return send(fd, buf, nbytes, 0);
  93 }
  94
  95 /**
  96  * Try to drain the output buffers without blocking
  97  *
  98  * @param client the client to drain
  99  * @return false if an error occured (connection should be shut down)
 100  *         true otherwise (please note that there may be more data to
 101  *              left in the buffer to send)
 102  */
 103 static bool drain_output(struct memcached_protocol_client_st *client)
 104 {
 105   ssize_t len;
 106
 107   /* Do we have pending data to send? */
 108   while (client->output != NULL)
 109   {
 110     len= client->root->send(client,
 111                             client->sock,
 112                             client->output->data + client->output->offset,
 113                             client->output->nbytes - client->output->offset);
 114
 115     if (len == -1)
 116     {
 117       if (get_socket_errno() == EWOULDBLOCK)
 118       {
 119         return true;
 120       }
 121       else if (get_socket_errno() != EINTR)
 122       {
 123         client->error= get_socket_errno();
 124         return false;
 125       }
 126     }
 127     else
 128     {
 129       client->output->offset += (size_t)len;
 130       if (client->output->offset == client->output->nbytes)
 131       {
 132         /* This was the complete buffer */
 133         struct chunk_st *old= client->output;
 134         client->output= client->output->next;
 135         if (client->output == NULL)
 136         {
 137           client->output_tail= NULL;
 138         }
 139         cache_free(client->root->buffer_cache, old);
 140       }
 141     }
 142   }
 143
 144   return true;
 145 }
 146
 147 /**
 148  * Allocate an output buffer and chain it into the output list
 149  *
 150  * @param client the client that needs the buffer
 151  * @return pointer to the new chunk if the allocation succeeds, NULL otherwise
 152  */
 153 static struct chunk_st *allocate_output_chunk(struct memcached_protocol_client_st *client)
 154 {
 155   struct chunk_st *ret= cache_alloc(client->root->buffer_cache);
 156
 157   if (ret == NULL)
 158   {
 159     return NULL;
 160   }
 161
 162   ret->offset= ret->nbytes= 0;
 163   ret->next= NULL;
 164   ret->size= CHUNK_BUFFERSIZE;
 165   ret->data= (void*)(ret + 1);
 166   if (client->output == NULL)
 167   {
 168     client->output= client->output_tail= ret;
 169   }
 170   else
 171   {
 172     client->output_tail->next= ret;
 173     client->output_tail= ret;
 174   }
 175
 176   return ret;
 177 }
 178
 179 /**
 180  * Spool data into the send-buffer for a client.
 181  *
 182  * @param client the client to spool the data for
 183  * @param data the data to spool
 184  * @param length the number of bytes of data to spool
 185  * @return PROTOCOL_BINARY_RESPONSE_SUCCESS if success,
 186  *         PROTOCOL_BINARY_RESPONSE_ENOMEM if we failed to allocate memory
 187  */
 188 static protocol_binary_response_status spool_output(struct memcached_protocol_client_st *client,
 189                                                     const void *data,
 190                                                     size_t length)
 191 {
 192   if (client->mute)
 193   {
 194     return PROTOCOL_BINARY_RESPONSE_SUCCESS;
 195   }
 196
 197   size_t offset= 0;
 198
 199   struct chunk_st *chunk= client->output;
 200   while (offset < length)
 201   {
 202     if (chunk == NULL || (chunk->size - chunk->nbytes) == 0)
 203     {
 204       if ((chunk= allocate_output_chunk(client)) == NULL)
 205       {
 206         return PROTOCOL_BINARY_RESPONSE_ENOMEM;
 207       }
 208     }
 209
 210     size_t bulk= length - offset;
 211     if (bulk > chunk->size - chunk->nbytes)
 212     {
 213       bulk= chunk->size - chunk->nbytes;
 214     }
 215
 216     memcpy(chunk->data + chunk->nbytes, data, bulk);
 217     chunk->nbytes += bulk;
 218     offset += bulk;
 219   }
 220
 221   return PROTOCOL_BINARY_RESPONSE_SUCCESS;
 222 }
 223
 224 /**
 225  * Try to determine the protocol used on this connection.
 226  * If the first byte contains the magic byte PROTOCOL_BINARY_REQ we should
 227  * be using the binary protocol on the connection. I implemented the support
 228  * for the ASCII protocol by wrapping into the simple interface (aka v1),
 229  * so the implementors needs to provide an implementation of that interface
 230  *
 231  */
 232 static memcached_protocol_event_t determine_protocol(struct memcached_protocol_client_st *client, ssize_t *length, void **endptr)
 233 {
 234   if (*client->root->input_buffer == (uint8_t)PROTOCOL_BINARY_REQ)
 235   {
 236     client->work= memcached_binary_protocol_process_data;
 237   }
 238   else if (client->root->callback->interface_version == 1)
 239   {
 240     /*
 241      * The ASCII protocol can only be used if the implementors provide
 242      * an implementation for the version 1 of the interface..
 243      *
 244      * @todo I should allow the implementors to provide an implementation
 245      *       for version 0 and 1 at the same time and set the preferred
 246      *       interface to use...
 247      */
 248     client->work= memcached_ascii_protocol_process_data;
 249   }
 250   else
 251   {
 252     /* Let's just output a warning the way it is supposed to look like
 253      * in the ASCII protocol...
 254      */
 255     const char *err= "CLIENT_ERROR: Unsupported protocol\r\n";
 256     client->root->spool(client, err, strlen(err));
 257     client->root->drain(client);
 258     return MEMCACHED_PROTOCOL_ERROR_EVENT; /* Unsupported protocol */
 259   }
 260
 261   return client->work(client, length, endptr);
 262 }
 263
 264 /*
 265 ** **********************************************************************
 266 ** * PUBLIC INTERFACE
 267 ** * See protocol_handler.h for function description
 268 ** **********************************************************************
 269 */
 270 struct memcached_protocol_st *memcached_protocol_create_instance(void)
 271 {
 272   struct memcached_protocol_st *ret= calloc(1, sizeof(*ret));
 273   if (ret != NULL)
 274   {
 275     ret->recv= default_recv;
 276     ret->send= default_send;
 277     ret->drain= drain_output;
 278     ret->spool= spool_output;
 279     ret->input_buffer_size= 1 * 1024 * 1024;
 280     ret->input_buffer= malloc(ret->input_buffer_size);
 281     if (ret->input_buffer == NULL)
 282     {
 283       free(ret);
 284       ret= NULL;
 285       return NULL;
 286     }
 287
 288     ret->buffer_cache= cache_create("protocol_handler",
 289                                      CHUNK_BUFFERSIZE + sizeof(struct chunk_st),
 290                                      0, NULL, NULL);
 291     if (ret->buffer_cache == NULL)
 292     {
 293       free(ret->input_buffer);
 294       free(ret);
 295     }
 296   }
 297
 298   return ret;
 299 }
 300
 301 void memcached_protocol_destroy_instance(struct memcached_protocol_st *instance)
 302 {
 303   cache_destroy(instance->buffer_cache);
 304   free(instance->input_buffer);
 305   free(instance);
 306 }
 307
 308 struct memcached_protocol_client_st *memcached_protocol_create_client(struct memcached_protocol_st *instance, memcached_socket_t sock)
 309 {
 310   struct memcached_protocol_client_st *ret= calloc(1, sizeof(*ret));
 311   if (ret != NULL)
 312   {
 313     ret->root= instance;
 314     ret->sock= sock;
 315     ret->work= determine_protocol;
 316   }
 317
 318   return ret;
 319 }
 320
 321 void memcached_protocol_client_destroy(struct memcached_protocol_client_st *client)
 322 {
 323   free(client);
 324 }
 325
 326 memcached_protocol_event_t memcached_protocol_client_work(struct memcached_protocol_client_st *client)
 327 {
 328   /* Try to send data and read from the socket */
 329   bool more_data= true;
 330   do
 331   {
 332     ssize_t len= client->root->recv(client,
 333                                     client->sock,
 334                                     client->root->input_buffer + client->input_buffer_offset,
 335                                     client->root->input_buffer_size - client->input_buffer_offset);
 336
 337     if (len > 0)
 338     {
 339       /* Do we have the complete packet? */
 340       if (client->input_buffer_offset > 0)
 341       {
 342         memcpy(client->root->input_buffer, client->input_buffer,
 343                client->input_buffer_offset);
 344         len += (ssize_t)client->input_buffer_offset;
 345
 346         /* @todo use buffer-cache! */
 347         free(client->input_buffer);
 348         client->input_buffer_offset= 0;
 349       }
 350
 351       void *endptr;
 352       memcached_protocol_event_t events= client->work(client, &len, &endptr);
 353       if (events == MEMCACHED_PROTOCOL_ERROR_EVENT)
 354       {
 355         return MEMCACHED_PROTOCOL_ERROR_EVENT;
 356       }
 357
 358       if (len > 0)
 359       {
 360         /* save the data for later on */
 361         /* @todo use buffer-cache */
 362         client->input_buffer= malloc((size_t)len);
 363         if (client->input_buffer == NULL)
 364         {
 365           client->error= ENOMEM;
 366           return MEMCACHED_PROTOCOL_ERROR_EVENT;
 367         }
 368         memcpy(client->input_buffer, endptr, (size_t)len);
 369         client->input_buffer_offset= (size_t)len;
 370         more_data= false;
 371       }
 372     }
 373     else if (len == 0)
 374     {
 375       /* Connection closed */
 376       drain_output(client);
 377       return MEMCACHED_PROTOCOL_ERROR_EVENT;
 378     }
 379     else
 380     {
 381       if (get_socket_errno() != EWOULDBLOCK)
 382       {
 383         client->error= get_socket_errno();
 384         /* mark this client as terminated! */
 385         return MEMCACHED_PROTOCOL_ERROR_EVENT;
 386       }
 387       more_data= false;
 388     }
 389   } while (more_data);
 390
 391   if (!drain_output(client))
 392   {
 393     return MEMCACHED_PROTOCOL_ERROR_EVENT;
 394   }
 395
 396   memcached_protocol_event_t ret= MEMCACHED_PROTOCOL_READ_EVENT;
 397   if (client->output)
 398     ret|= MEMCACHED_PROTOCOL_READ_EVENT;
 399
 400   return ret;
 401 }