git.m6w6.name Git - m6w6/libmemcached/blob - lib/memcached_io.c

   1 /*
   2   Basic socket buffered IO
   3 */
   4
   5 #include "common.h"
   6 #include "memcached_io.h"
   7 #include <sys/select.h>
   8 #include <poll.h>
   9
  10 int io_wait(memcached_st *ptr, unsigned int server_key, unsigned read_or_write)
  11 {
  12   struct pollfd fds[1];
  13   short flags= 0;
  14
  15   if (read_or_write)
  16     flags= POLLOUT |  POLLERR;
  17   else
  18     flags= POLLIN | POLLERR;
  19
  20   memset(&fds, 0, sizeof(struct pollfd));
  21   fds[0].fd= ptr->hosts[server_key].fd;
  22   fds[0].events= flags;
  23
  24   if (poll(fds, 1, -1) < 0)
  25     return MEMCACHED_FAILURE;
  26
  27   return MEMCACHED_SUCCESS;
  28 #ifdef OLD
  29   while (1)
  30   {
  31     int select_return;
  32     struct timeval local_tv;
  33     fd_set set;
  34
  35     memset(&local_tv, 0, sizeof(struct timeval));
  36
  37     local_tv.tv_sec= 0;
  38     local_tv.tv_usec= 300;
  39
  40     FD_ZERO(&set);
  41     FD_SET(ptr->hosts[server_key].fd, &set);
  42
  43     if (read_or_write)
  44       select_return= select(1, &set, NULL, NULL, &local_tv);
  45     else
  46       select_return= select(1, NULL, &set, NULL, &local_tv);
  47
  48     if (select_return == -1)
  49     {
  50       ptr->my_errno= errno;
  51       return MEMCACHED_FAILURE;
  52     }
  53     else if (!select_return)
  54       break;
  55   }
  56
  57   return MEMCACHED_SUCCESS;
  58 #endif
  59 }
  60
  61 ssize_t memcached_io_read(memcached_st *ptr, unsigned  int server_key,
  62                           char *buffer, size_t length)
  63 {
  64   char *buffer_ptr;
  65
  66   buffer_ptr= buffer;
  67
  68   while (length)
  69   {
  70     if (!ptr->read_buffer_length)
  71     {
  72       size_t data_read;
  73
  74       while (1)
  75       {
  76         if (ptr->flags & MEM_NO_BLOCK)
  77         {
  78           memcached_return rc;
  79
  80           rc= io_wait(ptr, server_key, 0);
  81           if (rc != MEMCACHED_SUCCESS)
  82             return -1;
  83         }
  84
  85         data_read= recv(ptr->hosts[server_key].fd,
  86                         ptr->read_buffer,
  87                         MEMCACHED_MAX_BUFFER, 0);
  88         if (data_read == -1)
  89         {
  90           switch (errno)
  91           {
  92           case EAGAIN:
  93             break;
  94           default:
  95             {
  96               ptr->my_errno= errno;
  97               return -1;
  98             }
  99           }
 100         }
 101         else if (data_read)
 102           break;
 103         /* If zero, just keep looping */
 104       }
 105
 106       ptr->read_buffer_length= data_read;
 107       ptr->read_ptr= ptr->read_buffer;
 108     }
 109
 110     *buffer_ptr= *ptr->read_ptr;
 111     length--;
 112     ptr->read_ptr++;
 113     ptr->read_buffer_length--;
 114     buffer_ptr++;
 115   }
 116
 117   return (size_t)(buffer_ptr - buffer);
 118 }
 119
 120 ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key,
 121                         char *buffer, size_t length, char with_flush)
 122 {
 123   unsigned long long x;
 124
 125   for (x= 0; x < length; x++)
 126   {
 127     ptr->write_buffer[ptr->write_buffer_offset]= buffer[x];
 128     ptr->write_buffer_offset++;
 129
 130     if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
 131     {
 132       size_t sent_length;
 133
 134       sent_length= memcached_io_flush(ptr, server_key);
 135
 136       assert(sent_length == MEMCACHED_MAX_BUFFER);
 137       ptr->write_buffer_offset= 0;
 138     }
 139   }
 140
 141   if (with_flush)
 142   {
 143     if (memcached_io_flush(ptr, server_key) == -1)
 144       return -1;
 145   }
 146
 147   return length;
 148 }
 149
 150 ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
 151 {
 152   size_t sent_length;
 153   char *write_ptr= ptr->write_buffer;
 154   size_t write_length= ptr->write_buffer_offset;
 155   unsigned int loop= 1;
 156
 157   if (ptr->write_buffer_offset == 0)
 158     return 0;
 159
 160   while (write_length)
 161   {
 162     if (ptr->flags & MEM_NO_BLOCK)
 163     {
 164       memcached_return rc;
 165
 166       rc= io_wait(ptr, server_key, 1);
 167       if (rc != MEMCACHED_SUCCESS)
 168         return -1;
 169     }
 170
 171     sent_length= 0;
 172 #ifdef orig
 173     if ((sent_length= send(ptr->hosts[server_key].fd, write_ptr,
 174                            write_length, 0)) == -1)
 175 #endif
 176       if ((sent_length= write(ptr->hosts[server_key].fd, write_ptr,
 177                               write_length)) == -1)
 178     {
 179       switch (errno)
 180       {
 181       case ENOBUFS:
 182       case EAGAIN:
 183         if (loop < 10)
 184         {
 185           loop++;
 186           break;
 187         }
 188         /* Yes, we want to fall through */
 189       default:
 190         ptr->my_errno= errno;
 191         return -1;
 192       }
 193     }
 194     else
 195     {
 196       write_ptr+= sent_length;
 197       write_length-= sent_length;
 198     }
 199   }
 200
 201   ptr->write_buffer_offset= 0;
 202
 203   return sent_length;
 204 }
 205
 206 /*
 207   Eventually we will just kill off the server with the problem.
 208 */
 209 void memcached_io_reset(memcached_st *ptr, unsigned int server_key)
 210 {
 211   ptr->write_buffer_offset= 0;
 212   memcached_quit(ptr);
 213 }