X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=lib%2Fmemcached_io.c;h=84928d2b40b5459e42acd1c7160d2e4d0a0acd31;hb=3bd95102380ff8d9f2599f0965c09776344d0691;hp=473b8881dc2bd61f1d6703d948ff35887da307e2;hpb=c1f907e3ade57bd12d08277f6b805c958f3cf581;p=m6w6%2Flibmemcached diff --git a/lib/memcached_io.c b/lib/memcached_io.c index 473b8881..84928d2b 100644 --- a/lib/memcached_io.c +++ b/lib/memcached_io.c @@ -7,12 +7,22 @@ #include #include -static int io_wait(memcached_st *ptr, unsigned int server_key, unsigned read_or_write) +typedef enum { + MEM_READ, + MEM_WRITE, +} memc_read_or_write; + +static ssize_t io_flush(memcached_st *ptr, unsigned int server_key, + memcached_return *error); + +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; - if (read_or_write) + if (read_or_write == MEM_WRITE) /* write */ flags= POLLOUT | POLLERR; else flags= POLLIN | POLLERR; @@ -21,10 +31,47 @@ static int io_wait(memcached_st *ptr, unsigned int server_key, unsigned read_or_ fds[0].fd= ptr->hosts[server_key].fd; fds[0].events= flags; - if (poll(fds, 1, -1) < 0) - return MEMCACHED_FAILURE; + error= poll(fds, 1, ptr->poll_timeout); - return MEMCACHED_SUCCESS; + if (error == 1) + return MEMCACHED_SUCCESS; + else if (error == 0) + { + return MEMCACHED_TIMEOUT; + } + + WATCHPOINT; + /* Imposssible for anything other then -1 */ + WATCHPOINT_ASSERT(error == -1); + memcached_quit_server(ptr, server_key, 1); + + return MEMCACHED_FAILURE; + +} + +void memcached_io_preread(memcached_st *ptr) +{ + unsigned int x; + + return; + + for (x= 0; x < ptr->number_of_hosts; x++) + { + if (ptr->hosts[x].cursor_active && + 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; + } + } } ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key, @@ -42,15 +89,6 @@ ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key, 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= read(ptr->hosts[server_key].fd, ptr->hosts[server_key].read_buffer, MEMCACHED_MAX_BUFFER); @@ -59,9 +97,20 @@ ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key, switch (errno) { case EAGAIN: - break; + { + 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; } @@ -72,15 +121,31 @@ ssize_t memcached_io_read(memcached_st *ptr, unsigned int server_key, /* If zero, just keep looping */ } + ptr->hosts[server_key].read_data_length= data_read; ptr->hosts[server_key].read_buffer_length= data_read; ptr->hosts[server_key].read_ptr= ptr->hosts[server_key].read_buffer; } - *buffer_ptr= *ptr->hosts[server_key].read_ptr; - length--; - ptr->hosts[server_key].read_ptr++; - ptr->hosts[server_key].read_buffer_length--; - buffer_ptr++; + if (length > 1) + { + size_t difference; + + difference= (length > ptr->hosts[server_key].read_buffer_length) ? ptr->hosts[server_key].read_buffer_length : length; + + memcpy(buffer_ptr, ptr->hosts[server_key].read_ptr, difference); + length -= difference; + ptr->hosts[server_key].read_ptr+= difference; + ptr->hosts[server_key].read_buffer_length-= difference; + buffer_ptr+= difference; + } + else + { + *buffer_ptr= *ptr->hosts[server_key].read_ptr; + length--; + ptr->hosts[server_key].read_ptr++; + ptr->hosts[server_key].read_buffer_length--; + buffer_ptr++; + } } return (size_t)(buffer_ptr - buffer); @@ -93,36 +158,54 @@ ssize_t memcached_io_write(memcached_st *ptr, unsigned int server_key, for (x= 0; x < length; x++) { - ptr->hosts[server_key].write_buffer[ptr->hosts[server_key].write_buffer_offset]= buffer[x]; + 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; 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) +{ + close(ptr->hosts[server_key].fd); + + return MEMCACHED_SUCCESS; +} + +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; - unsigned int loop= 1; + + *error= MEMCACHED_SUCCESS; if (ptr->hosts[server_key].write_buffer_offset == 0) return 0; @@ -130,34 +213,40 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key) return_length= 0; while (write_length) { - if (ptr->flags & MEM_NO_BLOCK) + sent_length= 0; + if (ptr->hosts[server_key].type == MEMCACHED_CONNECTION_UDP) { - memcached_return rc; - - rc= io_wait(ptr, server_key, 1); - if (rc != MEMCACHED_SUCCESS) - return -1; + 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)); } - - sent_length= 0; - if ((ssize_t)(sent_length= write(ptr->hosts[server_key].fd, write_ptr, - write_length)) == -1) + else { - switch (errno) + if ((ssize_t)(sent_length= write(ptr->hosts[server_key].fd, write_ptr, + write_length)) == -1) { - case ENOBUFS: - case EAGAIN: - WATCHPOINT; - continue; - if (loop < 100) + 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->cached_errno= errno; - return -1; } } @@ -168,6 +257,7 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key) 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 return_length; @@ -178,6 +268,7 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key) */ void memcached_io_reset(memcached_st *ptr, unsigned int server_key) { + ptr->hosts[server_key].write_ptr= ptr->hosts[server_key].write_buffer; ptr->hosts[server_key].write_buffer_offset= 0; memcached_quit(ptr); }