X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fio.cc;h=885e23919edfea4865cfaab698f00b482afb6487;hb=bcf4f7d33fd0b0cb5bb599b300aa043effc62a33;hp=2b0866dc7ac9d86c6fd1626008272f3458fb5fe4;hpb=51351e672a6a1626e09d5a9d41e3229df44fc3c8;p=m6w6%2Flibmemcached diff --git a/libmemcached/io.cc b/libmemcached/io.cc index 2b0866dc..885e2391 100644 --- a/libmemcached/io.cc +++ b/libmemcached/io.cc @@ -67,24 +67,27 @@ 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()) { case EINTR: continue; +#if EWOULDBLOCK != EAGAIN case EWOULDBLOCK: -#ifdef USE_EAGAIN - case EAGAIN: #endif + case EAGAIN: #ifdef TARGET_OS_LINUX case ERESTART: #endif @@ -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; } @@ -131,7 +129,7 @@ static bool process_input_buffer(memcached_server_write_instance_st ptr) ** We might be able to process some of the response messages if we ** have a callback set up */ - if (ptr->root->callbacks != NULL && ptr->root->flags.use_udp == false) + if (ptr->root->callbacks != NULL) { /* * We might have responses... try to read them out and fire @@ -142,10 +140,8 @@ static bool process_input_buffer(memcached_server_write_instance_st ptr) memcached_set_processing_input((memcached_st *)ptr->root, true); char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE]; - memcached_return_t error; memcached_st *root= (memcached_st *)ptr->root; - error= memcached_response(ptr, buffer, sizeof(buffer), - &root->result); + memcached_return_t error= memcached_response(ptr, buffer, sizeof(buffer), &root->result); memcached_set_processing_input(root, false); @@ -155,7 +151,9 @@ static bool process_input_buffer(memcached_server_write_instance_st ptr) { error= (*cb.callback[x])(ptr->root, &root->result, cb.context); if (error != MEMCACHED_SUCCESS) + { break; + } } /* @todo what should I do with the error message??? */ @@ -170,20 +168,6 @@ static bool process_input_buffer(memcached_server_write_instance_st ptr) static memcached_return_t io_wait(memcached_server_write_instance_st ptr, const memc_read_or_write read_or_write) { - struct pollfd fds; - fds.fd= ptr->fd; - fds.events= POLLIN; - - 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 ** on write if we haven't read anything from our input buffer.. @@ -194,80 +178,91 @@ static memcached_return_t io_wait(memcached_server_write_instance_st ptr, */ if (read_or_write == MEM_WRITE) { - memcached_return_t rc= memcached_purge(ptr); - if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED) + if (memcached_purge(ptr) == false) { return MEMCACHED_FAILURE; } } + struct pollfd fds; + fds.fd= ptr->fd; + fds.events= POLLIN; + fds.revents= 0; + + if (read_or_write == MEM_WRITE) /* write */ + { + fds.events= POLLOUT; + ptr->io_wait_count.write++; + } + else + { + ptr->io_wait_count.read++; + } + 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); - int error= poll(&fds, 1, ptr->root->poll_timeout); - switch (error) + if (active_fd >= 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); - + assert_msg(active_fd == 1 , "poll() returned an unexpected value"); return MEMCACHED_SUCCESS; - - case 0: // Timeout occured, we let the while() loop do its thing. + } + else if (active_fd == 0) + { + ptr->io_wait_count.timeouts++; return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT); + } - default: - WATCHPOINT_ERRNO(get_socket_errno()); - switch (get_socket_errno()) - { + // 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) + { #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: - 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) - { - continue; - } - errno= err; + continue; } + local_errno= err; } - else - { - memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT); - } - int local_errno= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno - 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, @@ -280,11 +275,9 @@ static bool io_flush(memcached_server_write_instance_st ptr, ** in the purge function to avoid duplicating the logic.. */ { - memcached_return_t rc; WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); - rc= memcached_purge(ptr); - if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED) + if (memcached_purge(ptr) == false) { return false; } @@ -296,18 +289,6 @@ static bool io_flush(memcached_server_write_instance_st ptr, WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); - // UDP Sanity check, make sure that we are not sending somthing too big - if (memcached_is_udp(ptr->root) and write_length > MAX_UDP_DATAGRAM_LENGTH) - { - error= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT); - return false; - } - - if (ptr->write_buffer_offset == 0 or (memcached_is_udp(ptr->root) and ptr->write_buffer_offset == UDP_DATAGRAM_HEADER_LENGTH)) - { - return true; - } - /* Looking for memory overflows */ #if defined(DEBUG) if (write_length == MEMCACHED_MAX_BUFFER) @@ -319,25 +300,12 @@ static bool io_flush(memcached_server_write_instance_st ptr, { WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); WATCHPOINT_ASSERT(write_length > 0); - if (memcached_is_udp(ptr->root)) - { - increment_udp_message_id(ptr); - } - 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) { - memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT); #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()); @@ -346,10 +314,11 @@ static bool io_flush(memcached_server_write_instance_st ptr, { case ENOBUFS: continue; + +#if EWOULDBLOCK != EAGAIN case EWOULDBLOCK: -#ifdef USE_EAGAIN - case EAGAIN: #endif + case EAGAIN: { /* * We may be blocked on write because the input buffer @@ -369,7 +338,6 @@ static bool io_flush(memcached_server_write_instance_st ptr, } else if (rc == MEMCACHED_TIMEOUT) { - error= memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT); return false; } @@ -387,13 +355,6 @@ static bool io_flush(memcached_server_write_instance_st ptr, } } - if (memcached_is_udp(ptr->root) and size_t(sent_length) != write_length) - { - memcached_quit_server(ptr, true); - error= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT); - return false; - } - ptr->io_bytes_sent+= uint32_t(sent_length); local_write_ptr+= sent_length; @@ -401,14 +362,7 @@ static bool io_flush(memcached_server_write_instance_st ptr, } WATCHPOINT_ASSERT(write_length == 0); - if (memcached_is_udp(ptr->root)) - { - ptr->write_buffer_offset= UDP_DATAGRAM_HEADER_LENGTH; - } - else - { - ptr->write_buffer_offset= 0; - } + ptr->write_buffer_offset= 0; return true; } @@ -419,8 +373,9 @@ memcached_return_t memcached_io_wait_for_write(memcached_server_write_instance_s } memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, - void *buffer, size_t length, ssize_t *nread) + void *buffer, size_t length, ssize_t& nread) { + assert(memcached_is_udp(ptr->root) == false); assert_msg(ptr, "Programmer error, memcached_io_read() recieved an invalid memcached_server_write_instance_st"); // Programmer error char *buffer_ptr= static_cast(buffer); @@ -434,12 +389,12 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, while (length) { - if (not ptr->read_buffer_length) + if (ptr->read_buffer_length == 0) { ssize_t data_read; do { - data_read= recv(ptr->fd, ptr->read_buffer, MEMCACHED_MAX_BUFFER, MSG_DONTWAIT); + data_read= ::recv(ptr->fd, ptr->read_buffer, MEMCACHED_MAX_BUFFER, MSG_DONTWAIT); if (data_read == SOCKET_ERROR) { switch (get_socket_errno()) @@ -448,18 +403,22 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, continue; case ETIMEDOUT: // OSX +#if EWOULDBLOCK != EAGAIN case EWOULDBLOCK: -#ifdef USE_EAGAIN - case EAGAIN: #endif + case EAGAIN: #ifdef TARGET_OS_LINUX case ERESTART: #endif - if (memcached_success(io_wait(ptr, MEM_READ))) { - continue; + memcached_return_t io_wait_ret; + if (memcached_success(io_wait_ret= io_wait(ptr, MEM_READ))) + { + continue; + } + + return io_wait_ret; } - return MEMCACHED_IN_PROGRESS; /* fall through */ @@ -475,7 +434,7 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, default: { memcached_quit_server(ptr, true); - *nread= -1; + nread= -1; return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT); } } @@ -493,8 +452,9 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, */ WATCHPOINT_STRING("We had a zero length recv()"); memcached_quit_server(ptr, true); - *nread= -1; - return memcached_set_error(*ptr, MEMCACHED_UNKNOWN_READ_FAILURE, MEMCACHED_AT); + nread= -1; + return memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT, + memcached_literal_param("::rec() returned zero, server has disconnected")); } } while (data_read <= 0); @@ -526,7 +486,7 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, } } - *nread = (ssize_t)(buffer_ptr - (char*)buffer); + nread= ssize_t(buffer_ptr - (char*)buffer); return MEMCACHED_SUCCESS; } @@ -534,6 +494,7 @@ memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr, memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr) { assert_msg(ptr, "Programmer error, invalid memcached_server_write_instance_st"); + assert(memcached_is_udp(ptr->root) == false); if (ptr->fd == INVALID_SOCKET) { @@ -554,10 +515,10 @@ memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr) continue; case ETIMEDOUT: // OSX +#if EWOULDBLOCK != EAGAIN case EWOULDBLOCK: -#ifdef USE_EAGAIN - case EAGAIN: #endif + case EAGAIN: #ifdef TARGET_OS_LINUX case ERESTART: #endif @@ -570,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: @@ -587,36 +548,23 @@ 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(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; - size_t should_write; - size_t buffer_end; - - if (memcached_is_udp(ptr->root)) - { - //UDP does not support partial writes - buffer_end= MAX_UDP_DATAGRAM_LENGTH; - should_write= length; - if (ptr->write_buffer_offset + should_write > buffer_end) - { - return -1; - } - } - else - { - buffer_end= MEMCACHED_MAX_BUFFER; - should_write= buffer_end - ptr->write_buffer_offset; - should_write= (should_write < length) ? should_write : length; - } + size_t buffer_end= MEMCACHED_MAX_BUFFER; + size_t should_write= buffer_end -ptr->write_buffer_offset; + should_write= (should_write < length) ? should_write : length; write_ptr= ptr->write_buffer + ptr->write_buffer_offset; memcpy(write_ptr, buffer_ptr, should_write); @@ -624,14 +572,15 @@ static ssize_t _io_write(memcached_server_write_instance_st ptr, buffer_ptr+= should_write; length-= should_write; - if (ptr->write_buffer_offset == buffer_end and memcached_is_udp(ptr->root) == false) + if (ptr->write_buffer_offset == buffer_end) { WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET); memcached_return_t rc; if (io_flush(ptr, with_flush, rc) == false) { - return -1; + written= original_length -length; + return false; } } } @@ -642,53 +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; -size_t io_vector_total_size(libmemcached_io_vector_st* vector, const size_t number_of) -{ - ssize_t total= 0; - - for (size_t x= 0; x < number_of; x++) + if (_io_write(ptr, buffer, length, with_flush, written) == false) { - total+= vector->length; + return -1; } - return total; + return ssize_t(written); } -ssize_t memcached_io_writev(memcached_server_write_instance_st ptr, - const struct libmemcached_io_vector_st *vector, - size_t number_of, bool with_flush) +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; } } @@ -696,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); } @@ -731,9 +680,9 @@ memcached_server_write_instance_st memcached_io_get_readable_server(memcached_st { #define MAX_SERVERS_TO_POLL 100 struct pollfd fds[MAX_SERVERS_TO_POLL]; - unsigned int host_index= 0; + nfds_t host_index= 0; - for (uint32_t x= 0; x < memcached_server_count(memc) && host_index < MAX_SERVERS_TO_POLL; ++x) + for (uint32_t x= 0; x < memcached_server_count(memc) and host_index < MAX_SERVERS_TO_POLL; ++x) { memcached_server_write_instance_st instance= memcached_server_instance_fetch(memc, x); @@ -778,17 +727,18 @@ memcached_server_write_instance_st memcached_io_get_readable_server(memcached_st break; default: - for (size_t x= 0; x < host_index; ++x) + for (nfds_t x= 0; x < host_index; ++x) { if (fds[x].revents & POLLIN) { for (uint32_t y= 0; y < memcached_server_count(memc); ++y) { - memcached_server_write_instance_st instance= - memcached_server_instance_fetch(memc, y); + memcached_server_write_instance_st instance= memcached_server_instance_fetch(memc, y); if (instance->fd == fds[x].fd) + { return instance; + } } } } @@ -821,14 +771,14 @@ memcached_return_t memcached_safe_read(memcached_server_write_instance_st ptr, ssize_t nread; memcached_return_t rc; - while (memcached_continue(rc= memcached_io_read(ptr, data + offset, size - offset, &nread))) { }; + while (memcached_continue(rc= memcached_io_read(ptr, data + offset, size - offset, nread))) { }; if (memcached_failed(rc)) { return rc; } - offset+= (size_t) nread; + offset+= size_t(nread); } return MEMCACHED_SUCCESS; @@ -852,7 +802,7 @@ memcached_return_t memcached_io_readline(memcached_server_write_instance_st ptr, * the logic. */ ssize_t nread; - memcached_return_t rc= memcached_io_read(ptr, buffer_ptr, 1, &nread); + memcached_return_t rc= memcached_io_read(ptr, buffer_ptr, 1, nread); if (memcached_failed(rc) and rc == MEMCACHED_IN_PROGRESS) { memcached_quit_server(ptr, true); @@ -873,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')