#include "common.h"
#include "memcached_io.h"
#include <sys/select.h>
+#include <poll.h>
-ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key,
- char *buffer, size_t length)
-{
- size_t x;
- char *buffer_ptr;
+typedef enum {
+ MEM_READ,
+ MEM_WRITE,
+} memc_read_or_write;
- buffer_ptr= buffer;
+static ssize_t io_flush(memcached_st *ptr, unsigned int server_key,
+ memcached_return *error);
- for (x= 0, buffer_ptr= buffer;
- x < length; x++)
- {
- if (!ptr->read_buffer_length)
- {
- if (length > 1)
- {
+static memcached_return io_wait(memcached_st *ptr, unsigned int server_key,
+ memc_read_or_write read_or_write)
+{
+ struct pollfd fds[1];
+ short flags= 0;
+ int error;
- size_t data_read;
- data_read= recv(ptr->hosts[server_key].fd,
- buffer_ptr,
- length - x, 0);
- if (data_read == -1)
- {
- return -1;
- }
- if (data_read == 0)
- return x;
+ if (read_or_write == MEM_WRITE) /* write */
+ flags= POLLOUT | POLLERR;
+ else
+ flags= POLLIN | POLLERR;
- data_read+= x;
+ memset(&fds, 0, sizeof(struct pollfd));
+ fds[0].fd= ptr->hosts[server_key].fd;
+ fds[0].events= flags;
- return data_read;
- }
- else
- {
- size_t data_read;
-try_again:
+ error= poll(fds, 1, ptr->poll_timeout);
- if (ptr->flags & MEM_NO_BLOCK)
- {
- struct timeval local_tv;
- fd_set set;
+ if (error == 1)
+ return MEMCACHED_SUCCESS;
+ else if (error == 0)
+ {
+ WATCHPOINT_NUMBER(read_or_write);
+ return MEMCACHED_TIMEOUT;
+ }
- memset(&local_tv, 0, sizeof(struct timeval));
+ WATCHPOINT;
+ /* Imposssible for anything other then -1 */
+ WATCHPOINT_ASSERT(error == -1);
+ memcached_quit_server(ptr, server_key, 1);
+ return MEMCACHED_FAILURE;
- local_tv.tv_sec= 0;
- local_tv.tv_usec= 300;
+}
- FD_ZERO(&set);
- FD_SET(ptr->hosts[server_key].fd, &set);
+ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key,
+ char *buffer, size_t length)
+{
+ char *buffer_ptr;
- select(1, &set, NULL, NULL, &local_tv);
- }
+ buffer_ptr= buffer;
+
+ while (length)
+ {
+ if (!ptr->hosts[server_key].read_buffer_length)
+ {
+ size_t data_read;
- data_read= recv(ptr->hosts[server_key].fd,
- ptr->read_buffer,
- MEMCACHED_MAX_BUFFER, 0);
+ while (1)
+ {
+ data_read= read(ptr->hosts[server_key].fd,
+ ptr->hosts[server_key].read_buffer,
+ MEMCACHED_MAX_BUFFER);
if (data_read == -1)
{
- if (errno == EAGAIN)
- goto try_again;
- return -1;
+ switch (errno)
+ {
+ case EAGAIN:
+ {
+ memcached_return rc;
+ rc= io_wait(ptr, server_key, MEM_READ);
+
+ if (rc == MEMCACHED_SUCCESS)
+ continue;
+
+ memcached_quit_server(ptr, server_key, 1);
+ return -1;
+ }
+ default:
+ {
+ memcached_quit_server(ptr, server_key, 1);
+ ptr->cached_errno= errno;
+ return -1;
+ }
+ }
}
- ptr->read_buffer_length= data_read;
- ptr->read_ptr= ptr->read_buffer;
+ else if (data_read)
+ break;
+ /* If zero, just keep looping */
}
- if (ptr->read_buffer_length == -1)
- return -1;
- if (ptr->read_buffer_length == 0)
- return x;
+ ptr->hosts[server_key].read_buffer_length= data_read;
+ ptr->hosts[server_key].read_ptr= ptr->hosts[server_key].read_buffer;
}
- *buffer_ptr= *ptr->read_ptr;
+
+ *buffer_ptr= *ptr->hosts[server_key].read_ptr;
+ length--;
+ ptr->hosts[server_key].read_ptr++;
+ ptr->hosts[server_key].read_buffer_length--;
buffer_ptr++;
- ptr->read_ptr++;
- ptr->read_buffer_length--;
}
- return length;
+ return (size_t)(buffer_ptr - buffer);
}
ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key,
for (x= 0; x < length; x++)
{
- ptr->write_buffer[ptr->write_buffer_offset]= buffer[x];
- ptr->write_buffer_offset++;
-
- if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
+ if (ptr->hosts[server_key].write_ptr == 0)
+ ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer;
+ WATCHPOINT_ASSERT(ptr->hosts[server_key].write_ptr);
+ *ptr->hosts[server_key].write_ptr= buffer[x];
+ ptr->hosts[server_key].write_ptr++;
+ ptr->hosts[server_key].write_buffer_offset++;
+
+ if (ptr->hosts[server_key].write_buffer_offset == MEMCACHED_MAX_BUFFER)
{
+ memcached_return rc;
size_t sent_length;
- sent_length= memcached_io_flush(ptr, server_key);
+ sent_length= io_flush(ptr, server_key, &rc);
+ if (sent_length == -1)
+ return -1;
- assert(sent_length == MEMCACHED_MAX_BUFFER);
- ptr->write_buffer_offset= 0;
+ WATCHPOINT_ASSERT(sent_length == MEMCACHED_MAX_BUFFER);
+ ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer;
+ ptr->hosts[server_key].write_buffer_offset= 0;
}
}
if (with_flush)
{
- if (memcached_io_flush(ptr, server_key) == -1)
+ memcached_return rc;
+ if (io_flush(ptr, server_key, &rc) == -1)
return -1;
}
return length;
}
-ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
+memcached_return memcached_io_close(memcached_st *ptr, unsigned int server_key)
{
- size_t sent_length;
- char *write_ptr= ptr->write_buffer;
- size_t write_length= ptr->write_buffer_offset;
- unsigned int loop= 1;
+ memcached_return rc;
- if (ptr->write_buffer_offset == 0)
- return 0;
-
- while (write_length)
+ rc= MEMCACHED_SUCCESS;
+ if (ptr->flags & MEM_NO_BLOCK)
{
- if (ptr->flags & MEM_NO_BLOCK)
+ int error;
+ struct pollfd fds[1];
+ short flags= 0;
+
+ flags= POLLHUP | POLLERR;
+
+ memset(&fds, 0, sizeof(struct pollfd));
+ fds[0].fd= ptr->hosts[server_key].fd;
+ fds[0].events= flags;
+ fds[0].revents= 0;
+
+ error= poll(fds, 1, ptr->poll_timeout == -1 ? 100 : ptr->poll_timeout);
+
+ if (error == -1)
{
+ memcached_quit_server(ptr, server_key, 1);
+ return MEMCACHED_FAILURE;
+ }
+ else if (error == 0)
+ return MEMCACHED_FAILURE; /* Timeout occurred */
+ }
- while (1)
- {
- struct timeval local_tv;
- fd_set set;
- int select_return;
+ close(ptr->hosts[server_key].fd);
- local_tv.tv_sec= 0;
- local_tv.tv_usec= 300 * loop;
+ return rc;
+}
- FD_ZERO(&set);
- FD_SET(ptr->hosts[server_key].fd, &set);
+static ssize_t io_flush(memcached_st *ptr, unsigned int server_key,
+ memcached_return *error)
+{
+ size_t sent_length;
+ size_t return_length;
+ char *write_ptr= ptr->hosts[server_key].write_buffer;
+ size_t write_length= ptr->hosts[server_key].write_buffer_offset;
- select_return= select(1, NULL, &set, NULL, &local_tv);
+ *error= MEMCACHED_SUCCESS;
- if (select_return == -1)
- {
- ptr->my_errno= errno;
- return -1;
- }
- else if (!select_return)
- break;
- }
- }
+ if (ptr->hosts[server_key].write_buffer_offset == 0)
+ return 0;
+ return_length= 0;
+ while (write_length)
+ {
sent_length= 0;
- if ((sent_length= send(ptr->hosts[server_key].fd, write_ptr,
- write_length, 0)) == -1)
+ if (ptr->hosts[server_key].type == MEMCACHED_CONNECTION_UDP)
{
- switch (errno)
+ sent_length= sendto(ptr->hosts[server_key].fd,
+ write_ptr, write_length, 0,
+ (struct sockaddr *)&ptr->hosts[server_key].address_info->ai_addr,
+ sizeof(struct sockaddr));
+ }
+ else
+ {
+ if ((ssize_t)(sent_length= write(ptr->hosts[server_key].fd, write_ptr,
+ write_length)) == -1)
{
- case ENOBUFS:
- case EAGAIN:
- if (loop < 10)
+ switch (errno)
{
- loop++;
- break;
+ case ENOBUFS:
+ continue;
+ case EAGAIN:
+ {
+ memcached_return rc;
+ rc= io_wait(ptr, server_key, MEM_WRITE);
+
+ if (rc == MEMCACHED_SUCCESS)
+ continue;
+
+ memcached_quit_server(ptr, server_key, 1);
+ return -1;
+ }
+ default:
+ memcached_quit_server(ptr, server_key, 1);
+ ptr->cached_errno= errno;
+ *error= MEMCACHED_ERRNO;
+ return -1;
}
- /* Yes, we want to fall through */
- default:
- ptr->my_errno= errno;
- return -1;
}
}
- else
- {
- write_ptr+= sent_length;
- write_length-= sent_length;
- }
+
+ write_ptr+= sent_length;
+ write_length-= sent_length;
+ return_length+= sent_length;
}
- ptr->write_buffer_offset= 0;
+ WATCHPOINT_ASSERT(write_length == 0);
+ WATCHPOINT_ASSERT(return_length == ptr->hosts[server_key].write_buffer_offset);
+ ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer;
+ ptr->hosts[server_key].write_buffer_offset= 0;
- return sent_length;
+ return return_length;
}
/*
*/
void memcached_io_reset(memcached_st *ptr, unsigned int server_key)
{
- ptr->write_buffer_offset= 0;
+ ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer;
+ ptr->hosts[server_key].write_buffer_offset= 0;
memcached_quit(ptr);
}
projects / awesomized / libmemcached / blobdiff