X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fio.c;h=3b5d44f570e2003313963a6c70a1600287f382ba;hb=d46b596b82e290141fd97e3c5381efdb792756a4;hp=070af6b19462d9056f7a3181ba4944e26ae7b9cf;hpb=96c84d98b1b2db8419072eea960e8800c189586b;p=awesomized%2Flibmemcached diff --git a/libmemcached/io.c b/libmemcached/io.c index 070af6b1..3b5d44f5 100644 --- a/libmemcached/io.c +++ b/libmemcached/io.c @@ -11,8 +11,8 @@ #include "common.h" -#include -#include + +#include typedef enum { MEM_READ, @@ -32,8 +32,15 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr, }; int error; - unlikely (read_or_write == MEM_WRITE) /* write */ + if (read_or_write == MEM_WRITE) /* write */ + { fds.events= POLLOUT; + WATCHPOINT_SET(ptr->io_wait_count.write++); + } + else + { + WATCHPOINT_SET(ptr->io_wait_count.read++); + } /* ** We are going to block on write, but at least on Solaris we might block @@ -50,42 +57,51 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr, return MEMCACHED_FAILURE; } - int timeout= ptr->root->poll_timeout; - if (ptr->root->flags.no_block == false) - timeout= -1; - size_t loop_max= 5; - while (--loop_max) + while (--loop_max) // While loop is for ERESTART or EINTR { - error= poll(&fds, 1, timeout); + error= poll(&fds, 1, ptr->root->poll_timeout); switch (error) { - case 1: + case 1: // Success! + WATCHPOINT_IF_LABELED_NUMBER(read_or_write && loop_max < 4, "read() times we had to loop, decremented down from 5", loop_max); + WATCHPOINT_IF_LABELED_NUMBER(!read_or_write && loop_max < 4, "write() times we had to loop, decremented down from 5", loop_max); + return MEMCACHED_SUCCESS; - case 0: - continue; + case 0: // Timeout occured, we let the while() loop do its thing. + return MEMCACHED_TIMEOUT; default: - WATCHPOINT_ERRNO(errno); + WATCHPOINT_ERRNO(get_socket_errno()); + switch (get_socket_errno()) { - switch (errno) +#ifdef TARGET_OS_LINUX + case ERESTART: +#endif + case EINTR: + break; + default: + if (fds.revents & POLLERR) { - default: - ptr->cached_errno= error; - memcached_quit_server(ptr, true); - - return MEMCACHED_FAILURE; + int err; + socklen_t len= sizeof (err); + (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len); + ptr->cached_errno= (err == 0) ? get_socket_errno() : err; + } + else + { + ptr->cached_errno= get_socket_errno(); } + memcached_quit_server(ptr, true); + + return MEMCACHED_FAILURE; } } } - if (loop_max == 0 && error == 0) - return MEMCACHED_TIMEOUT; - /* Imposssible for anything other then -1 */ WATCHPOINT_ASSERT(error == -1); - ptr->cached_errno= error; + ptr->cached_errno= get_socket_errno(); memcached_quit_server(ptr, true); return MEMCACHED_FAILURE; @@ -113,9 +129,10 @@ static bool repack_input_buffer(memcached_server_write_instance_st ptr) if (ptr->read_buffer_length != MEMCACHED_MAX_BUFFER) { /* Just try a single read to grab what's available */ - ssize_t nr= read(ptr->fd, + ssize_t nr= recv(ptr->fd, ptr->read_ptr + ptr->read_data_length, - MEMCACHED_MAX_BUFFER - ptr->read_data_length); + MEMCACHED_MAX_BUFFER - ptr->read_data_length, + 0); if (nr > 0) { @@ -227,10 +244,10 @@ void memcached_io_preread(memcached_st *ptr) { size_t data_read; - data_read= read(ptr->hosts[x].fd, + data_read= recv(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) + MEMCACHED_MAX_BUFFER - ptr->hosts[x].read_data_length, 0); + if (data_read == SOCKET_ERROR) continue; ptr->hosts[x].read_buffer_length+= data_read; @@ -255,16 +272,21 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, while (1) { - data_read= read(ptr->fd, ptr->read_buffer, MEMCACHED_MAX_BUFFER); + data_read= recv(ptr->fd, ptr->read_buffer, MEMCACHED_MAX_BUFFER, 0); if (data_read > 0) + { break; - else if (data_read == -1) + } + else if (data_read == SOCKET_ERROR) { - ptr->cached_errno= errno; - memcached_return_t rc= MEMCACHED_UNKNOWN_READ_FAILURE; - switch (errno) + ptr->cached_errno= get_socket_errno(); + memcached_return_t rc= MEMCACHED_ERRNO; + switch (get_socket_errno()) { + case EWOULDBLOCK: +#ifdef USE_EAGAIN case EAGAIN: +#endif case EINTR: #ifdef TARGET_OS_LINUX case ERESTART: @@ -292,6 +314,7 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, for blocking I/O we do not return 0 and for non-blocking case it will return EGAIN if data is not immediatly available. */ + WATCHPOINT_STRING("We had a zero length recv()"); memcached_quit_server(ptr, true); *nread= -1; return MEMCACHED_UNKNOWN_READ_FAILURE; @@ -337,7 +360,7 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, size_t original_length; const char* buffer_ptr; - WATCHPOINT_ASSERT(ptr->fd != -1); + WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); original_length= length; buffer_ptr= buffer; @@ -360,7 +383,10 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, buffer_end= MAX_UDP_DATAGRAM_LENGTH; should_write= length; if (ptr->write_buffer_offset + should_write > buffer_end) + { + fprintf(stderr, "%s:%d (%s)\n", __FILE__, __LINE__,__func__);fflush(stdout); return -1; + } } else { @@ -380,10 +406,12 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, memcached_return_t rc; ssize_t sent_length; - WATCHPOINT_ASSERT(ptr->fd != -1); + WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); sent_length= io_flush(ptr, &rc); if (sent_length == -1) + { return -1; + } /* If io_flush calls memcached_purge, sent_length may be 0 */ unlikely (sent_length != 0) @@ -396,9 +424,10 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, if (with_flush) { memcached_return_t rc; - WATCHPOINT_ASSERT(ptr->fd != -1); + WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); if (io_flush(ptr, &rc) == -1) { + fprintf(stderr, "%s:%d (%s)\n", __FILE__, __LINE__,__func__);fflush(stdout); return -1; } @@ -415,7 +444,7 @@ ssize_t memcached_io_write(memcached_server_write_instance_st ptr, } ssize_t memcached_io_writev(memcached_server_write_instance_st ptr, - const struct __write_vector_st *vector, + const struct libmemcached_io_vector_st *vector, size_t number_of, bool with_flush) { ssize_t total= 0; @@ -445,22 +474,22 @@ ssize_t memcached_io_writev(memcached_server_write_instance_st ptr, memcached_return_t memcached_io_close(memcached_server_write_instance_st ptr) { - if (ptr->fd == -1) + if (ptr->fd == INVALID_SOCKET) { return MEMCACHED_SUCCESS; } /* in case of death shutdown to avoid blocking at close() */ - if (shutdown(ptr->fd, SHUT_RDWR) == -1 && errno != ENOTCONN) + if (shutdown(ptr->fd, SHUT_RDWR) == SOCKET_ERROR && get_socket_errno() != ENOTCONN) { WATCHPOINT_NUMBER(ptr->fd); - WATCHPOINT_ERRNO(errno); - WATCHPOINT_ASSERT(errno); + WATCHPOINT_ERRNO(get_socket_errno()); + WATCHPOINT_ASSERT(get_socket_errno()); } - if (close(ptr->fd) == -1) + if (closesocket(ptr->fd) == SOCKET_ERROR) { - WATCHPOINT_ERRNO(errno); + WATCHPOINT_ERRNO(get_socket_errno()); } return MEMCACHED_SUCCESS; @@ -511,7 +540,7 @@ memcached_server_write_instance_st memcached_io_get_readable_server(memcached_st int err= poll(fds, host_index, memc->poll_timeout); switch (err) { case -1: - memc->cached_errno = errno; + memc->cached_errno = get_socket_errno(); /* FALLTHROUGH */ case 0: break; @@ -545,11 +574,14 @@ static ssize_t io_flush(memcached_server_write_instance_st ptr, */ { memcached_return_t rc; - WATCHPOINT_ASSERT(ptr->fd != -1); + WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); rc= memcached_purge(ptr); if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED) + { + fprintf(stderr, "%s:%d (%s)\n", __FILE__, __LINE__,__func__);fflush(stdout); return -1; + } } ssize_t sent_length; size_t return_length; @@ -558,11 +590,14 @@ static ssize_t io_flush(memcached_server_write_instance_st ptr, *error= MEMCACHED_SUCCESS; - WATCHPOINT_ASSERT(ptr->fd != -1); + WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); // UDP Sanity check, make sure that we are not sending somthing too big if (ptr->type == MEMCACHED_CONNECTION_UDP && write_length > MAX_UDP_DATAGRAM_LENGTH) + { + fprintf(stderr, "%s:%d (%s)\n", __FILE__, __LINE__,__func__);fflush(stdout); return -1; + } if (ptr->write_buffer_offset == 0 || (ptr->type == MEMCACHED_CONNECTION_UDP && ptr->write_buffer_offset == UDP_DATAGRAM_HEADER_LENGTH)) @@ -578,22 +613,29 @@ static ssize_t io_flush(memcached_server_write_instance_st ptr, return_length= 0; while (write_length) { - WATCHPOINT_ASSERT(ptr->fd != -1); + WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); WATCHPOINT_ASSERT(write_length > 0); sent_length= 0; if (ptr->type == MEMCACHED_CONNECTION_UDP) increment_udp_message_id(ptr); - sent_length= write(ptr->fd, local_write_ptr, write_length); - if (sent_length == -1) + assert(ptr->fd != -1); + sent_length= send(ptr->fd, local_write_ptr, write_length, 0); + if (sent_length == SOCKET_ERROR) { - ptr->cached_errno= errno; - WATCHPOINT_ERRNO(errno); - switch (errno) + ptr->cached_errno= get_socket_errno(); +#if 0 // @todo I should look at why we hit this bit of code hard frequently + WATCHPOINT_ERRNO(get_socket_errno()); + WATCHPOINT_NUMBER(get_socket_errno()); +#endif + switch (get_socket_errno()) { case ENOBUFS: continue; + case EWOULDBLOCK: +#ifdef USE_EAGAIN case EAGAIN: +#endif { /* * We may be blocked on write because the input buffer @@ -612,11 +654,17 @@ static ssize_t io_flush(memcached_server_write_instance_st ptr, continue; memcached_quit_server(ptr, true); + fprintf(stderr, "%s:%d (%s)\n", __FILE__, __LINE__,__func__);fflush(stdout); return -1; } + case ENOTCONN: + case EPIPE: default: + fprintf(stderr, "%s %u %u %u\n", ptr->hostname, ptr->port, ptr->io_bytes_sent, ptr->server_failure_counter); + fprintf(stderr, "%s:%d (%s)(%s)\n", __FILE__, __LINE__,__func__, strerror(errno));fflush(stdout); memcached_quit_server(ptr, true); *error= MEMCACHED_ERRNO; + assert(ptr->fd == -1); return -1; } } @@ -625,6 +673,7 @@ static ssize_t io_flush(memcached_server_write_instance_st ptr, (size_t)sent_length != write_length) { memcached_quit_server(ptr, true); + fprintf(stderr, "%s:%d (%s)\n", __FILE__, __LINE__,__func__);fflush(stdout); return -1; }