Basic socket buffered IO
*/
-#include <memcached.h>
+#include "common.h"
+#include "memcached_io.h"
+#include <sys/select.h>
+#include <poll.h>
+
+int io_wait(memcached_st *ptr, unsigned int server_key, unsigned read_or_write)
+{
+ struct pollfd fds[1];
+ short flags= 0;
+
+ if (read_or_write)
+ flags= POLLOUT | POLLERR;
+ else
+ flags= POLLIN | POLLERR;
+
+ memset(&fds, 0, sizeof(struct pollfd));
+ fds[0].fd= ptr->hosts[server_key].fd;
+ fds[0].events= flags;
+
+ if (poll(fds, 1, -1) < 0)
+ return MEMCACHED_FAILURE;
+
+ return MEMCACHED_SUCCESS;
+}
ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key,
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)
{
if (!ptr->read_buffer_length)
{
- ptr->read_buffer_length= recv(ptr->hosts[server_key].fd,
- ptr->read_buffer,
- MEMCACHED_MAX_BUFFER, 0);
+ size_t data_read;
+
+ while (1)
+ {
+ if (ptr->flags & MEM_NO_BLOCK)
+ {
+ memcached_return rc;
+
+ rc= io_wait(ptr, server_key, 0);
+ if (rc != MEMCACHED_SUCCESS)
+ return -1;
+ }
+
+ data_read= recv(ptr->hosts[server_key].fd,
+ ptr->read_buffer,
+ MEMCACHED_MAX_BUFFER, 0);
+ if (data_read == -1)
+ {
+ switch (errno)
+ {
+ case EAGAIN:
+ break;
+ default:
+ {
+ ptr->my_errno= errno;
+ return -1;
+ }
+ }
+ }
+ else if (data_read)
+ break;
+ /* If zero, just keep looping */
+ }
+
+ 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;
}
+
*buffer_ptr= *ptr->read_ptr;
- buffer_ptr++;
+ length--;
ptr->read_ptr++;
ptr->read_buffer_length--;
+ buffer_ptr++;
}
- 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)
+ char *buffer, size_t length, char with_flush)
{
unsigned long long x;
{
ptr->write_buffer[ptr->write_buffer_offset]= buffer[x];
ptr->write_buffer_offset++;
+
if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
{
size_t sent_length;
- if ((sent_length= send(ptr->hosts[server_key].fd, ptr->write_buffer,
- MEMCACHED_MAX_BUFFER, 0)) == -1)
- return -1;
+ sent_length= memcached_io_flush(ptr, server_key);
assert(sent_length == MEMCACHED_MAX_BUFFER);
- ptr->write_between_flush+= MEMCACHED_MAX_BUFFER;
-
ptr->write_buffer_offset= 0;
}
}
+ if (with_flush)
+ {
+ if (memcached_io_flush(ptr, server_key) == -1)
+ return -1;
+ }
+
return length;
}
ssize_t memcached_io_flush(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;
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;
+ while (write_length)
+ {
+ if (ptr->flags & MEM_NO_BLOCK)
+ {
+ memcached_return rc;
- assert(sent_length == ptr->write_buffer_offset);
+ rc= io_wait(ptr, server_key, 1);
+ if (rc != MEMCACHED_SUCCESS)
+ return -1;
+ }
- sent_length+= ptr->write_between_flush;
+ sent_length= 0;
+ if ((sent_length= write(ptr->hosts[server_key].fd, write_ptr,
+ write_length)) == -1)
+ {
+ switch (errno)
+ {
+ case ENOBUFS:
+ case EAGAIN:
+ if (loop < 10)
+ {
+ loop++;
+ break;
+ }
+ /* Yes, we want to fall through */
+ default:
+ ptr->my_errno= errno;
+ return -1;
+ }
+ }
+ else
+ {
+ write_ptr+= sent_length;
+ write_length-= sent_length;
+ }
+ }
ptr->write_buffer_offset= 0;
- ptr->write_between_flush= 0;
return sent_length;
}