X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fdo.cc;h=59f9d5a3ada657c68d74da26837e5d90212fde30;hb=9973d386ac8476cd09cc4d9cf7bd2234a42740c7;hp=a51472ba66562a50a18353b1c843bd5c3a3afd7c;hpb=fd8a1d785722f36d111915b0c96331e907df2746;p=awesomized%2Flibmemcached diff --git a/libmemcached/do.cc b/libmemcached/do.cc index a51472ba..59f9d5a3 100644 --- a/libmemcached/do.cc +++ b/libmemcached/do.cc @@ -11,20 +11,72 @@ #include -memcached_return_t memcached_vdo(memcached_server_write_instance_st instance, - libmemcached_io_vector_st *vector, +static memcached_return_t _vdo_udp(memcached_instance_st* instance, + libmemcached_io_vector_st vector[], + const size_t count) +{ +#ifndef __MINGW32__ + if (vector[0].buffer or vector[0].length) + { + return memcached_set_error(*instance->root, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT, + memcached_literal_param("UDP messages was attempted, but vector was not setup for it")); + } + + struct msghdr msg; + memset(&msg, 0, sizeof(msg)); + + increment_udp_message_id(instance); + vector[0].buffer= instance->write_buffer; + vector[0].length= UDP_DATAGRAM_HEADER_LENGTH; + + msg.msg_iov= (struct iovec*)vector; +#ifdef __APPLE__ + msg.msg_iovlen= int(count); +#else + msg.msg_iovlen= count; +#endif + + uint32_t retry= 5; + while (--retry) + { + ssize_t sendmsg_length= ::sendmsg(instance->fd, &msg, 0); + if (sendmsg_length > 0) + { + break; + } + else if (sendmsg_length < 0) + { + if (errno == EMSGSIZE) + { + return memcached_set_error(*instance, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT); + } + + return memcached_set_errno(*instance, errno, MEMCACHED_AT); + } + } + + return MEMCACHED_SUCCESS; +#else + (void)instance; + (void)vector; + (void)count; + return MEMCACHED_FAILURE; +#endif +} + +memcached_return_t memcached_vdo(memcached_instance_st* instance, + libmemcached_io_vector_st vector[], const size_t count, const bool with_flush) { memcached_return_t rc; - WATCHPOINT_ASSERT(count); - WATCHPOINT_ASSERT(vector); + assert_msg(vector, "Invalid vector passed"); if (memcached_failed(rc= memcached_connect(instance))) { WATCHPOINT_ERROR(rc); - assert_msg(instance->error_messages, "memcached_connect() returned an error but the memcached_server_write_instance_st showed none."); + assert_msg(instance->error_messages, "memcached_connect() returned an error but the Instance showed none."); return rc; } @@ -33,38 +85,23 @@ memcached_return_t memcached_vdo(memcached_server_write_instance_st instance, ** before they start writing, if there is any data in buffer, clear it out, ** otherwise we might get a partial write. **/ + bool sent_success; if (memcached_is_udp(instance->root)) { - if (vector->buffer or vector->length) - { - return memcached_set_error(*instance->root, MEMCACHED_NOT_SUPPORTED, MEMCACHED_AT, - memcached_literal_param("UDP messages was attempted, but vector was not setup for it")); - } - - size_t write_length= io_vector_total_size(vector, 11) +UDP_DATAGRAM_HEADER_LENGTH; - - if (write_length > MAX_UDP_DATAGRAM_LENGTH - UDP_DATAGRAM_HEADER_LENGTH) - { - return MEMCACHED_WRITE_FAILURE; - } - - return MEMCACHED_NOT_SUPPORTED; + sent_success= memcached_success(rc= _vdo_udp(instance, vector, count)); + } else { + sent_success= memcached_io_writev(instance, vector, count, with_flush); } - - ssize_t sent_length= memcached_io_writev(instance, vector, count, with_flush); - size_t command_length= 0; - for (uint32_t x= 0; x < count; ++x, vector++) - { - command_length+= vector->length; - } - - if (sent_length == -1 or size_t(sent_length) != command_length) + if (sent_success == false) { - rc= MEMCACHED_WRITE_FAILURE; - WATCHPOINT_ERROR(rc); - WATCHPOINT_ERRNO(errno); + rc= memcached_last_error(instance->root); + if (rc == MEMCACHED_SUCCESS) + { + memcached_set_error(*instance, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT); + } + memcached_io_reset(instance); } - else if (memcached_is_replying(instance->root)) + else if (memcached_is_replying(instance->root) && !memcached_is_udp(instance->root)) { memcached_server_response_increment(instance); }