X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fio.cc;h=ed33f05a2e6ba4d94ba7d74774456c9c18a93e2a;hb=0b1f3751f2f2131482edd2fd23bd02b5242b5aac;hp=94686d0590851a26c16b9cce32bc3858080b0661;hpb=015b6142ad6c3b496437d397416aacb776a42aed;p=m6w6%2Flibmemcached diff --git a/libmemcached/io.cc b/libmemcached/io.cc index 94686d05..ed33f05a 100644 --- a/libmemcached/io.cc +++ b/libmemcached/io.cc @@ -67,14 +67,17 @@ static bool repack_input_buffer(memcached_server_write_instance_st ptr) { do { /* Just try a single read to grab what's available */ - ssize_t nr= recv(ptr->fd, - ptr->read_ptr + ptr->read_data_length, - MEMCACHED_MAX_BUFFER - ptr->read_data_length, - MSG_DONTWAIT); - - switch (nr) + ssize_t nr; + if ((nr= recv(ptr->fd, + ptr->read_ptr + ptr->read_data_length, + MEMCACHED_MAX_BUFFER - ptr->read_data_length, + MSG_DONTWAIT)) <= 0) { - case SOCKET_ERROR: + if (nr == 0) + { + memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); + } + else { switch (get_socket_errno()) { @@ -94,24 +97,19 @@ static bool repack_input_buffer(memcached_server_write_instance_st ptr) memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT); } } - break; - case 0: // Shutdown on the socket has occurred - { - memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); - } break; + } + else // We read data, append to our read buffer + { + ptr->read_data_length+= size_t(nr); + ptr->read_buffer_length+= size_t(nr); - default: - { - ptr->read_data_length+= size_t(nr); - ptr->read_buffer_length+= size_t(nr); - return true; - } - break; + return true; } - } while (0); + } while (false); } + return false; } @@ -180,16 +178,16 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr, */ if (read_or_write == MEM_WRITE) { - if (memcached_fatal(memcached_purge(ptr))) + if (memcached_purge(ptr) == false) { return MEMCACHED_FAILURE; } } struct pollfd fds; - memset(&fds, 0, sizeof(pollfd)); fds.fd= ptr->fd; fds.events= POLLIN; + fds.revents= 0; if (read_or_write == MEM_WRITE) /* write */ { @@ -203,17 +201,19 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr, if (ptr->root->poll_timeout == 0) // Mimic 0 causes timeout behavior (not all platforms do this) { + ptr->io_wait_count.timeouts++; return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT); } + int local_errno; size_t loop_max= 5; while (--loop_max) // While loop is for ERESTART or EINTR { int active_fd= poll(&fds, 1, ptr->root->poll_timeout); - assert_msg(active_fd <= 1 , "poll() returned an unexpected value"); - if (active_fd == 1) + if (active_fd >= 1) { + assert_msg(active_fd == 1 , "poll() returned an unexpected value"); return MEMCACHED_SUCCESS; } else if (active_fd == 0) @@ -221,48 +221,48 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr, ptr->io_wait_count.timeouts++; return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT); } - else // -1 + + // Only an error should result in this code being called. + local_errno= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno + assert_msg(active_fd == -1 , "poll() returned an unexpected value"); + switch (local_errno) { - switch (get_socket_errno()) - { #ifdef TARGET_OS_LINUX - case ERESTART: + case ERESTART: #endif - case EINTR: - break; + case EINTR: + continue; - case EFAULT: - case ENOMEM: - return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT); + case EFAULT: + case ENOMEM: + return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT); - case EINVAL: - return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid")); + case EINVAL: + return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid")); - default: - int local_errno= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno - if (fds.revents & POLLERR) + default: + if (fds.revents & POLLERR) + { + int err; + socklen_t len= sizeof (err); + if (getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0) { - int err; - socklen_t len= sizeof (err); - if (getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0) + if (err == 0) // treat this as EINTR { - if (err == 0) // treat this as EINTR - { - continue; - } - local_errno= err; + continue; } + local_errno= err; } - memcached_quit_server(ptr, true); - - return memcached_set_errno(*ptr, local_errno, MEMCACHED_AT); } + break; } + + break; // should only occur from poll error } memcached_quit_server(ptr, true); - return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT); + return memcached_set_errno(*ptr, local_errno, MEMCACHED_AT); } static bool io_flush(memcached_server_write_instance_st ptr, @@ -276,9 +276,8 @@ static bool io_flush(memcached_server_write_instance_st ptr, */ { WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); - memcached_return_t rc= memcached_purge(ptr); - if (rc != MEMCACHED_SUCCESS and rc != MEMCACHED_STORED) + if (memcached_purge(ptr) == false) { return false; } @@ -302,16 +301,8 @@ static bool io_flush(memcached_server_write_instance_st ptr, WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); WATCHPOINT_ASSERT(write_length > 0); - ssize_t sent_length= 0; - WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); - if (with_flush) - { - sent_length= ::send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT); - } - else - { - sent_length= ::send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE); - } + int flags= with_flush ? MSG_NOSIGNAL|MSG_DONTWAIT : MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE; + ssize_t sent_length= ::send(ptr->fd, local_write_ptr, write_length, flags); if (sent_length == SOCKET_ERROR) { @@ -347,8 +338,6 @@ static bool io_flush(memcached_server_write_instance_st ptr, } else if (rc == MEMCACHED_TIMEOUT) { - ptr->io_wait_count.timeouts++; - error= memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT); return false; } @@ -542,9 +531,9 @@ memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr) /* fall through */ case ENOTCONN: // Programmer Error - WATCHPOINT_ASSERT(0); + assert(0); case ENOTSOCK: - WATCHPOINT_ASSERT(0); + assert(0); case EBADF: assert_msg(ptr->fd != INVALID_SOCKET, "Invalid socket state"); case EINVAL: @@ -559,15 +548,17 @@ memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr) return MEMCACHED_CONNECTION_FAILURE; } -static ssize_t _io_write(memcached_server_write_instance_st ptr, - const void *buffer, size_t length, bool with_flush) +static bool _io_write(memcached_server_write_instance_st ptr, + const void *buffer, size_t length, bool with_flush, + size_t& written) { - WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); + assert(ptr->fd != INVALID_SOCKET); assert(memcached_is_udp(ptr->root) == false); - size_t original_length= length; const char *buffer_ptr= static_cast(buffer); + const size_t original_length= length; + while (length) { char *write_ptr; @@ -588,7 +579,8 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, memcached_return_t rc; if (io_flush(ptr, with_flush, rc) == false) { - return -1; + written= original_length -length; + return false; } } } @@ -599,41 +591,53 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); if (io_flush(ptr, with_flush, rc) == false) { - return -1; + written= original_length -length; + return false; } } - return ssize_t(original_length); + written= original_length -length; + + return true; } bool memcached_io_write(memcached_server_write_instance_st ptr) { - return (_io_write(ptr, NULL, 0, true) >= 0); + size_t written; + return _io_write(ptr, NULL, 0, true, written); } ssize_t memcached_io_write(memcached_server_write_instance_st ptr, const void *buffer, const size_t length, const bool with_flush) { - return _io_write(ptr, buffer, length, with_flush); + size_t written; + + if (_io_write(ptr, buffer, length, with_flush, written) == false) + { + return -1; + } + + return ssize_t(written); } -ssize_t memcached_io_writev(memcached_server_write_instance_st ptr, +bool memcached_io_writev(memcached_server_write_instance_st ptr, libmemcached_io_vector_st vector[], const size_t number_of, const bool with_flush) { + ssize_t complete_total= 0; ssize_t total= 0; for (size_t x= 0; x < number_of; x++, vector++) { - ssize_t returnable; - + complete_total+= vector->length; if (vector->length) { - if ((returnable= _io_write(ptr, vector->buffer, vector->length, false)) == -1) + size_t written; + if ((_io_write(ptr, vector->buffer, vector->length, false, written)) == false) { - return -1; + return false; } - total+= returnable; + total+= written; } } @@ -641,11 +645,11 @@ ssize_t memcached_io_writev(memcached_server_write_instance_st ptr, { if (memcached_io_write(ptr) == false) { - return -1; + return false; } } - return total; + return (complete_total == total); } @@ -819,7 +823,7 @@ memcached_return_t memcached_io_readline(memcached_server_write_instance_st ptr, } /* Now let's look in the buffer and copy as we go! */ - while (ptr->read_buffer_length && total_nr < size && !line_complete) + while (ptr->read_buffer_length and total_nr < size and line_complete == false) { *buffer_ptr = *ptr->read_ptr; if (*buffer_ptr == '\n')