Basic socket buffered IO
*/
-#include <memcached.h>
+#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,
+typedef enum {
+ MEM_READ,
+ MEM_WRITE,
+} memc_read_or_write;
+
+static ssize_t io_flush(memcached_server_st *ptr, memcached_return *error);
+
+static memcached_return io_wait(memcached_server_st *ptr,
+ memc_read_or_write read_or_write)
+{
+ struct pollfd fds[1];
+ short flags= 0;
+ int error;
+
+ if (read_or_write == MEM_WRITE) /* write */
+ flags= POLLOUT | POLLERR;
+ else
+ flags= POLLIN | POLLERR;
+
+ memset(&fds, 0, sizeof(struct pollfd));
+ fds[0].fd= ptr->fd;
+ fds[0].events= flags;
+
+ error= poll(fds, 1, ptr->root->poll_timeout);
+
+ if (error == 1)
+ return MEMCACHED_SUCCESS;
+ else if (error == 0)
+ {
+ return MEMCACHED_TIMEOUT;
+ }
+
+ /* Imposssible for anything other then -1 */
+ WATCHPOINT_ASSERT(error == -1);
+ memcached_quit_server(ptr, 1);
+
+ return MEMCACHED_FAILURE;
+
+}
+
+#ifdef UNUSED
+void memcached_io_preread(memcached_st *ptr)
+{
+ unsigned int x;
+
+ return;
+
+ for (x= 0; x < ptr->number_of_hosts; x++)
+ {
+ if (memcached_server_response_count(ptr, x) &&
+ ptr->hosts[x].read_data_length < MEMCACHED_MAX_BUFFER )
+ {
+ size_t data_read;
+
+ data_read= read(ptr->hosts[x].fd,
+ ptr->hosts[x].read_ptr + ptr->hosts[x].read_data_length,
+ MEMCACHED_MAX_BUFFER - ptr->hosts[x].read_data_length);
+ if (data_read == -1)
+ continue;
+
+ ptr->hosts[x].read_buffer_length+= data_read;
+ ptr->hosts[x].read_data_length+= data_read;
+ }
+ }
+}
+#endif
+
+ssize_t memcached_io_read(memcached_server_st *ptr,
char *buffer, size_t length)
{
- size_t x;
char *buffer_ptr;
buffer_ptr= buffer;
- for (x= 0, buffer_ptr= buffer;
- x < length; x++)
+ while (length)
{
+ uint8_t found_eof= 0;
if (!ptr->read_buffer_length)
{
- ptr->read_buffer_length= recv(ptr->hosts[server_key].fd,
- ptr->read_buffer,
- MEMCACHED_MAX_BUFFER, 0);
+ ssize_t data_read;
+
+ while (1)
+ {
+ data_read= read(ptr->fd,
+ ptr->read_buffer,
+ MEMCACHED_MAX_BUFFER);
+ if (data_read > 0)
+ break;
+ else if (data_read == -1)
+ {
+ ptr->cached_errno= errno;
+ switch (errno)
+ {
+ case EAGAIN:
+ {
+ memcached_return rc;
+
+ rc= io_wait(ptr, MEM_READ);
+
+ if (rc == MEMCACHED_SUCCESS)
+ continue;
+ }
+ /* fall trough */
+ default:
+ {
+ memcached_quit_server(ptr, 1);
+ return -1;
+ }
+ }
+ }
+ else
+ {
+ WATCHPOINT_ASSERT(0);
+ found_eof= 1;
+ break;
+ }
+ }
+
+ ptr->read_data_length= data_read;
+ ptr->read_buffer_length= data_read;
ptr->read_ptr= ptr->read_buffer;
+ }
- if (ptr->read_buffer_length == -1)
- return -1;
- if (ptr->read_buffer_length == 0)
- return x;
+ if (length > 1)
+ {
+ size_t difference;
+
+ difference= (length > ptr->read_buffer_length) ? ptr->read_buffer_length : length;
+
+ memcpy(buffer_ptr, ptr->read_ptr, difference);
+ length -= difference;
+ ptr->read_ptr+= difference;
+ ptr->read_buffer_length-= difference;
+ buffer_ptr+= difference;
+ }
+ else
+ {
+ *buffer_ptr= *ptr->read_ptr;
+ ptr->read_ptr++;
+ ptr->read_buffer_length--;
+ buffer_ptr++;
+ break;
}
- *buffer_ptr= *ptr->read_ptr;
- buffer_ptr++;
- ptr->read_ptr++;
- ptr->read_buffer_length--;
+
+ if (found_eof)
+ break;
}
- return length;
+ return (size_t)(buffer_ptr - buffer);
}
-ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key,
- char *buffer, size_t length)
+ssize_t memcached_io_write(memcached_server_st *ptr,
+ char *buffer, size_t length, char with_flush)
{
- unsigned long long x;
+ size_t original_length;
+ char* buffer_ptr;
+
+ original_length= length;
+ buffer_ptr= buffer;
- for (x= 0; x < length; x++)
+ while (length)
{
- ptr->write_buffer[ptr->write_buffer_offset]= buffer[x];
- ptr->write_buffer_offset++;
+ char *write_ptr;
+ size_t should_write;
+
+ should_write= MEMCACHED_MAX_BUFFER - ptr->write_buffer_offset;
+ write_ptr= ptr->write_buffer + ptr->write_buffer_offset;
+
+ should_write= (should_write < length) ? should_write : length;
+
+ memcpy(write_ptr, buffer_ptr, should_write);
+ ptr->write_buffer_offset+= should_write;
+ buffer_ptr+= should_write;
+ length-= should_write;
+
if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
{
- size_t sent_length;
+ memcached_return rc;
+ ssize_t sent_length;
- if ((sent_length= send(ptr->hosts[server_key].fd, ptr->write_buffer,
- MEMCACHED_MAX_BUFFER, 0)) == -1)
+ sent_length= io_flush(ptr, &rc);
+ if (sent_length == -1)
return -1;
- assert(sent_length == MEMCACHED_MAX_BUFFER);
- ptr->write_between_flush+= MEMCACHED_MAX_BUFFER;
-
- ptr->write_buffer_offset= 0;
+ WATCHPOINT_ASSERT(sent_length == MEMCACHED_MAX_BUFFER);
}
}
- return length;
+ if (with_flush)
+ {
+ memcached_return rc;
+ if (io_flush(ptr, &rc) == -1)
+ return -1;
+ }
+
+ return original_length;
}
-ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key)
+memcached_return memcached_io_close(memcached_server_st *ptr)
+{
+ close(ptr->fd);
+
+ return MEMCACHED_SUCCESS;
+}
+
+static ssize_t io_flush(memcached_server_st *ptr,
+ memcached_return *error)
{
size_t sent_length;
+ size_t return_length;
+ char *local_write_ptr= ptr->write_buffer;
+ size_t write_length= ptr->write_buffer_offset;
+
+ *error= MEMCACHED_SUCCESS;
if (ptr->write_buffer_offset == 0)
return 0;
- if ((sent_length= send(ptr->hosts[server_key].fd, ptr->write_buffer,
- ptr->write_buffer_offset, 0)) == -1)
- return -1;
+ /* Looking for memory overflows */
+ if (write_length == MEMCACHED_MAX_BUFFER)
+ WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
+ WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
- assert(sent_length == ptr->write_buffer_offset);
+ return_length= 0;
+ while (write_length)
+ {
+ WATCHPOINT_ASSERT(write_length > 0);
+ sent_length= 0;
+ if (ptr->type == MEMCACHED_CONNECTION_UDP)
+ {
+ struct addrinfo *ai;
- sent_length+= ptr->write_between_flush;
+ ai= ptr->address_info;
+ /* Crappy test code */
+ char buffer[HUGE_STRING_LEN + 8];
+ memset(buffer, 0, HUGE_STRING_LEN + 8);
+ memcpy (buffer+8, local_write_ptr, write_length);
+ buffer[0]= 0;
+ buffer[1]= 0;
+ buffer[2]= 0;
+ buffer[3]= 0;
+ buffer[4]= 0;
+ buffer[5]= 1;
+ buffer[6]= 0;
+ buffer[7]= 0;
+ sent_length= sendto(ptr->fd, buffer, write_length + 8, 0,
+ (struct sockaddr *)ai->ai_addr,
+ ai->ai_addrlen);
+ if (sent_length == -1)
+ {
+ WATCHPOINT_ERRNO(errno);
+ WATCHPOINT_ASSERT(0);
+ }
+ sent_length-= 8; /* We remove the header */
+ }
+ else
+ {
+ if ((ssize_t)(sent_length= write(ptr->fd, local_write_ptr,
+ write_length)) == -1)
+ {
+ switch (errno)
+ {
+ case ENOBUFS:
+ continue;
+ case EAGAIN:
+ {
+ memcached_return rc;
+ rc= io_wait(ptr, MEM_WRITE);
+
+ if (rc == MEMCACHED_SUCCESS)
+ continue;
+
+ memcached_quit_server(ptr, 1);
+ return -1;
+ }
+ default:
+ memcached_quit_server(ptr, 1);
+ ptr->cached_errno= errno;
+ *error= MEMCACHED_ERRNO;
+ return -1;
+ }
+ }
+ }
+
+ local_write_ptr+= sent_length;
+ write_length-= sent_length;
+ return_length+= sent_length;
+ }
+
+ WATCHPOINT_ASSERT(write_length == 0);
+ WATCHPOINT_ASSERT(return_length == ptr->write_buffer_offset);
ptr->write_buffer_offset= 0;
- ptr->write_between_flush= 0;
- return sent_length;
+ return return_length;
}
/*
Eventually we will just kill off the server with the problem.
*/
-void memcached_io_reset(memcached_st *ptr, unsigned int server_key)
+void memcached_io_reset(memcached_server_st *ptr)
{
- ptr->write_buffer_offset= 0;
- memcached_quit(ptr);
+ memcached_quit_server(ptr, 0);
}
projects / m6w6 / libmemcached / blobdiff