X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=lib%2Fmemcached_io.c;h=38580638d7f197c81d373373954cb299b8aabc98;hb=acca3c7ad60716bf073ae8de0652517542cdd224;hp=c0215ac08f917d83249a0c74dcfebd277d2a894d;hpb=3bcf5c0efffbf64b6b6eccf140f057d2373fa974;p=m6w6%2Flibmemcached diff --git a/lib/memcached_io.c b/lib/memcached_io.c index c0215ac0..38580638 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); + + 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; - return MEMCACHED_SUCCESS; +} + +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; + } + } } 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,26 +213,13 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key) return_length= 0; while (write_length) { - if (ptr->flags & MEM_NO_BLOCK) - { - memcached_return rc; - - rc= io_wait(ptr, server_key, 1); - if (rc != MEMCACHED_SUCCESS) - return -1; - } - sent_length= 0; if (ptr->hosts[server_key].type == MEMCACHED_CONNECTION_UDP) { - - sent_length= sendto(ptr->hosts[server_key].fd, write_ptr, write_length, - 0, 0, 0); - /* - rc = sendto(sd, argv[i], strlen(argv[i])+1, 0, - (struct sockaddr *) &remoteServAddr, - sizeof(remoteServAddr)); - */ + 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 { @@ -159,17 +229,22 @@ ssize_t memcached_io_flush(memcached_st *ptr, unsigned int server_key) switch (errno) { case ENOBUFS: - case EAGAIN: - WATCHPOINT; continue; - if (loop < 100) + case EAGAIN: { - loop++; - break; + 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; } - /* Yes, we want to fall through */ default: + memcached_quit_server(ptr, server_key, 1); ptr->cached_errno= errno; + *error= MEMCACHED_ERRNO; return -1; } } @@ -182,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; @@ -192,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); }