1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2009 Brian Aker
9 * Redistribution and use in source and binary forms, with or without
10 * modification, are permitted provided that the following conditions are
13 * * Redistributions of source code must retain the above copyright
14 * notice, this list of conditions and the following disclaimer.
16 * * Redistributions in binary form must reproduce the above
17 * copyright notice, this list of conditions and the following disclaimer
18 * in the documentation and/or other materials provided with the
21 * * The names of its contributors may not be used to endorse or
22 * promote products derived from this software without specific prior
25 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
26 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
27 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
28 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
29 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
30 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
31 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
32 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
33 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
34 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
35 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
40 #include <libmemcached/common.h>
42 enum memc_read_or_write
{
48 * Try to fill the input buffer for a server with as much
51 * @param ptr the server to pack
53 static bool repack_input_buffer(memcached_server_write_instance_st ptr
)
55 if (ptr
->read_ptr
!= ptr
->read_buffer
)
57 /* Move all of the data to the beginning of the buffer so
58 ** that we can fit more data into the buffer...
60 memmove(ptr
->read_buffer
, ptr
->read_ptr
, ptr
->read_buffer_length
);
61 ptr
->read_ptr
= ptr
->read_buffer
;
62 ptr
->read_data_length
= ptr
->read_buffer_length
;
65 /* There is room in the buffer, try to fill it! */
66 if (ptr
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
69 /* Just try a single read to grab what's available */
71 if ((nr
= recv(ptr
->fd
,
72 ptr
->read_ptr
+ ptr
->read_data_length
,
73 MEMCACHED_MAX_BUFFER
- ptr
->read_data_length
,
78 memcached_set_error(*ptr
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
);
82 switch (get_socket_errno())
87 #if EWOULDBLOCK != EAGAIN
91 #ifdef TARGET_OS_LINUX
94 break; // No IO is fine, we can just move on
97 memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
103 else // We read data, append to our read buffer
105 ptr
->read_data_length
+= size_t(nr
);
106 ptr
->read_buffer_length
+= size_t(nr
);
117 * If the we have callbacks connected to this server structure
118 * we may start process the input queue and fire the callbacks
119 * for the incomming messages. This function is _only_ called
120 * when the input buffer is full, so that we _know_ that we have
121 * at least _one_ message to process.
123 * @param ptr the server to star processing iput messages for
124 * @return true if we processed anything, false otherwise
126 static bool process_input_buffer(memcached_server_write_instance_st ptr
)
129 ** We might be able to process some of the response messages if we
130 ** have a callback set up
132 if (ptr
->root
->callbacks
!= NULL
)
135 * We might have responses... try to read them out and fire
138 memcached_callback_st cb
= *ptr
->root
->callbacks
;
140 memcached_set_processing_input((memcached_st
*)ptr
->root
, true);
142 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
143 memcached_st
*root
= (memcached_st
*)ptr
->root
;
144 memcached_return_t error
= memcached_response(ptr
, buffer
, sizeof(buffer
), &root
->result
);
146 memcached_set_processing_input(root
, false);
148 if (error
== MEMCACHED_SUCCESS
)
150 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
152 error
= (*cb
.callback
[x
])(ptr
->root
, &root
->result
, cb
.context
);
153 if (error
!= MEMCACHED_SUCCESS
)
159 /* @todo what should I do with the error message??? */
161 /* @todo what should I do with other error messages?? */
168 static memcached_return_t
io_wait(memcached_server_write_instance_st ptr
,
169 const memc_read_or_write read_or_write
)
172 ** We are going to block on write, but at least on Solaris we might block
173 ** on write if we haven't read anything from our input buffer..
174 ** Try to purge the input buffer if we don't do any flow control in the
175 ** application layer (just sending a lot of data etc)
176 ** The test is moved down in the purge function to avoid duplication of
179 if (read_or_write
== MEM_WRITE
)
181 if (memcached_purge(ptr
) == false)
183 return MEMCACHED_FAILURE
;
192 if (read_or_write
== MEM_WRITE
) /* write */
195 ptr
->io_wait_count
.write
++;
199 ptr
->io_wait_count
.read
++;
202 if (ptr
->root
->poll_timeout
== 0) // Mimic 0 causes timeout behavior (not all platforms do this)
204 ptr
->io_wait_count
.timeouts
++;
205 return memcached_set_error(*ptr
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
209 while (--loop_max
) // While loop is for ERESTART or EINTR
211 int active_fd
= poll(&fds
, 1, ptr
->root
->poll_timeout
);
215 assert_msg(active_fd
== 1 , "poll() returned an unexpected number of active file descriptors");
216 if (fds
.revents
& POLLIN
or fds
.revents
& POLLOUT
)
218 return MEMCACHED_SUCCESS
;
221 if (fds
.revents
& POLLHUP
)
223 return memcached_set_error(*ptr
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
,
224 memcached_literal_param("poll() detected hang up"));
227 if (fds
.revents
& POLLERR
)
229 int local_errno
= EINVAL
;
231 socklen_t len
= sizeof (err
);
232 if (getsockopt(ptr
->fd
, SOL_SOCKET
, SO_ERROR
, &err
, &len
) == 0)
234 if (err
== 0) // treat this as EINTR
240 memcached_quit_server(ptr
, true);
241 return memcached_set_errno(*ptr
, local_errno
, MEMCACHED_AT
,
242 memcached_literal_param("poll() returned POLLHUP"));
245 return memcached_set_error(*ptr
, MEMCACHED_FAILURE
, MEMCACHED_AT
, memcached_literal_param("poll() returned a value that was not dealt with"));
250 ptr
->io_wait_count
.timeouts
++;
251 return memcached_set_error(*ptr
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
254 // Only an error should result in this code being called.
255 int local_errno
= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno
256 assert_msg(active_fd
== -1 , "poll() returned an unexpected value");
259 #ifdef TARGET_OS_LINUX
267 memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
270 memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid"));
273 memcached_set_errno(*ptr
, local_errno
, MEMCACHED_AT
, memcached_literal_param("poll"));
279 memcached_quit_server(ptr
, true);
281 if (memcached_has_error(ptr
))
283 return memcached_server_error_return(ptr
);
286 return memcached_set_error(*ptr
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
,
287 memcached_literal_param("number of attempts to call io_wait() failed"));
290 static bool io_flush(memcached_server_write_instance_st ptr
,
291 const bool with_flush
,
292 memcached_return_t
& error
)
295 ** We might want to purge the input buffer if we haven't consumed
296 ** any output yet... The test for the limits is the purge is inline
297 ** in the purge function to avoid duplicating the logic..
300 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
302 if (memcached_purge(ptr
) == false)
307 char *local_write_ptr
= ptr
->write_buffer
;
308 size_t write_length
= ptr
->write_buffer_offset
;
310 error
= MEMCACHED_SUCCESS
;
312 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
314 /* Looking for memory overflows */
316 if (write_length
== MEMCACHED_MAX_BUFFER
)
317 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
318 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
323 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
324 WATCHPOINT_ASSERT(write_length
> 0);
326 int flags
= with_flush
? MSG_NOSIGNAL
|MSG_DONTWAIT
: MSG_NOSIGNAL
|MSG_DONTWAIT
|MSG_MORE
;
327 ssize_t sent_length
= ::send(ptr
->fd
, local_write_ptr
, write_length
, flags
);
329 if (sent_length
== SOCKET_ERROR
)
331 #if 0 // @todo I should look at why we hit this bit of code hard frequently
332 WATCHPOINT_ERRNO(get_socket_errno());
333 WATCHPOINT_NUMBER(get_socket_errno());
335 switch (get_socket_errno())
340 #if EWOULDBLOCK != EAGAIN
346 * We may be blocked on write because the input buffer
347 * is full. Let's check if we have room in our input
348 * buffer for more data and retry the write before
351 if (repack_input_buffer(ptr
) or process_input_buffer(ptr
))
356 memcached_return_t rc
= io_wait(ptr
, MEM_WRITE
);
357 if (memcached_success(rc
))
361 else if (rc
== MEMCACHED_TIMEOUT
)
366 memcached_quit_server(ptr
, true);
367 error
= memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
373 memcached_quit_server(ptr
, true);
374 error
= memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
375 WATCHPOINT_ASSERT(ptr
->fd
== INVALID_SOCKET
);
380 ptr
->io_bytes_sent
+= uint32_t(sent_length
);
382 local_write_ptr
+= sent_length
;
383 write_length
-= uint32_t(sent_length
);
386 WATCHPOINT_ASSERT(write_length
== 0);
387 ptr
->write_buffer_offset
= 0;
392 memcached_return_t
memcached_io_wait_for_write(memcached_server_write_instance_st ptr
)
394 return io_wait(ptr
, MEM_WRITE
);
397 static memcached_return_t
_io_fill(memcached_server_write_instance_st ptr
)
402 data_read
= ::recv(ptr
->fd
, ptr
->read_buffer
, MEMCACHED_MAX_BUFFER
, MSG_DONTWAIT
);
403 if (data_read
== SOCKET_ERROR
)
405 switch (get_socket_errno())
407 case EINTR
: // We just retry
410 case ETIMEDOUT
: // OSX
411 #if EWOULDBLOCK != EAGAIN
415 #ifdef TARGET_OS_LINUX
419 memcached_return_t io_wait_ret
;
420 if (memcached_success(io_wait_ret
= io_wait(ptr
, MEM_READ
)))
430 case ENOTCONN
: // Programmer Error
431 WATCHPOINT_ASSERT(0);
433 WATCHPOINT_ASSERT(0);
435 assert_msg(ptr
->fd
!= INVALID_SOCKET
, "Programmer error, invalid socket");
440 memcached_quit_server(ptr
, true);
441 memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
445 return memcached_server_error_return(ptr
);
447 else if (data_read
== 0)
450 EOF. Any data received so far is incomplete
451 so discard it. This always reads by byte in case of TCP
452 and protocol enforcement happens at memcached_response()
453 looking for '\n'. We do not care for UDB which requests 8 bytes
454 at once. Generally, this means that connection went away. Since
455 for blocking I/O we do not return 0 and for non-blocking case
456 it will return EGAIN if data is not immediatly available.
458 memcached_quit_server(ptr
, true);
459 return memcached_set_error(*ptr
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
,
460 memcached_literal_param("::rec() returned zero, server has disconnected"));
462 ptr
->io_wait_count
._bytes_read
+= data_read
;
463 } while (data_read
<= 0);
465 ptr
->io_bytes_sent
= 0;
466 ptr
->read_data_length
= (size_t) data_read
;
467 ptr
->read_buffer_length
= (size_t) data_read
;
468 ptr
->read_ptr
= ptr
->read_buffer
;
470 return MEMCACHED_SUCCESS
;
473 memcached_return_t
memcached_io_read(memcached_server_write_instance_st ptr
,
474 void *buffer
, size_t length
, ssize_t
& nread
)
476 assert(memcached_is_udp(ptr
->root
) == false);
477 assert_msg(ptr
, "Programmer error, memcached_io_read() recieved an invalid memcached_server_write_instance_st"); // Programmer error
478 char *buffer_ptr
= static_cast<char *>(buffer
);
480 if (ptr
->fd
== INVALID_SOCKET
)
483 assert_msg(int(ptr
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
), "Programmer error, invalid socket state");
485 return MEMCACHED_CONNECTION_FAILURE
;
490 if (ptr
->read_buffer_length
== 0)
492 memcached_return_t io_fill_ret
;
493 if (memcached_fatal(io_fill_ret
= _io_fill(ptr
)))
502 size_t difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
504 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
505 length
-= difference
;
506 ptr
->read_ptr
+= difference
;
507 ptr
->read_buffer_length
-= difference
;
508 buffer_ptr
+= difference
;
512 *buffer_ptr
= *ptr
->read_ptr
;
514 ptr
->read_buffer_length
--;
520 nread
= ssize_t(buffer_ptr
- (char*)buffer
);
522 return MEMCACHED_SUCCESS
;
525 memcached_return_t
memcached_io_slurp(memcached_server_write_instance_st ptr
)
527 assert_msg(ptr
, "Programmer error, invalid memcached_server_write_instance_st");
528 assert(memcached_is_udp(ptr
->root
) == false);
530 if (ptr
->fd
== INVALID_SOCKET
)
532 assert_msg(int(ptr
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
), "Invalid socket state");
533 return MEMCACHED_CONNECTION_FAILURE
;
537 char buffer
[MEMCACHED_MAX_BUFFER
];
540 data_read
= recv(ptr
->fd
, ptr
->read_buffer
, sizeof(buffer
), MSG_DONTWAIT
);
541 if (data_read
== SOCKET_ERROR
)
543 switch (get_socket_errno())
545 case EINTR
: // We just retry
548 case ETIMEDOUT
: // OSX
549 #if EWOULDBLOCK != EAGAIN
553 #ifdef TARGET_OS_LINUX
556 if (memcached_success(io_wait(ptr
, MEM_READ
)))
560 return MEMCACHED_IN_PROGRESS
;
564 case ENOTCONN
: // Programmer Error
569 assert_msg(ptr
->fd
!= INVALID_SOCKET
, "Invalid socket state");
574 return MEMCACHED_CONNECTION_FAILURE
; // We want this!
577 } while (data_read
> 0);
579 return MEMCACHED_CONNECTION_FAILURE
;
582 static bool _io_write(memcached_server_write_instance_st ptr
,
583 const void *buffer
, size_t length
, bool with_flush
,
586 assert(ptr
->fd
!= INVALID_SOCKET
);
587 assert(memcached_is_udp(ptr
->root
) == false);
589 const char *buffer_ptr
= static_cast<const char *>(buffer
);
591 const size_t original_length
= length
;
596 size_t buffer_end
= MEMCACHED_MAX_BUFFER
;
597 size_t should_write
= buffer_end
-ptr
->write_buffer_offset
;
598 should_write
= (should_write
< length
) ? should_write
: length
;
600 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
601 memcpy(write_ptr
, buffer_ptr
, should_write
);
602 ptr
->write_buffer_offset
+= should_write
;
603 buffer_ptr
+= should_write
;
604 length
-= should_write
;
606 if (ptr
->write_buffer_offset
== buffer_end
)
608 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
610 memcached_return_t rc
;
611 if (io_flush(ptr
, with_flush
, rc
) == false)
613 written
= original_length
-length
;
621 memcached_return_t rc
;
622 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
623 if (io_flush(ptr
, with_flush
, rc
) == false)
625 written
= original_length
-length
;
630 written
= original_length
-length
;
635 bool memcached_io_write(memcached_server_write_instance_st ptr
)
638 return _io_write(ptr
, NULL
, 0, true, written
);
641 ssize_t
memcached_io_write(memcached_server_write_instance_st ptr
,
642 const void *buffer
, const size_t length
, const bool with_flush
)
646 if (_io_write(ptr
, buffer
, length
, with_flush
, written
) == false)
651 return ssize_t(written
);
654 bool memcached_io_writev(memcached_server_write_instance_st ptr
,
655 libmemcached_io_vector_st vector
[],
656 const size_t number_of
, const bool with_flush
)
658 ssize_t complete_total
= 0;
661 for (size_t x
= 0; x
< number_of
; x
++, vector
++)
663 complete_total
+= vector
->length
;
667 if ((_io_write(ptr
, vector
->buffer
, vector
->length
, false, written
)) == false)
677 if (memcached_io_write(ptr
) == false)
683 return (complete_total
== total
);
687 void memcached_io_close(memcached_server_write_instance_st ptr
)
689 if (ptr
->fd
== INVALID_SOCKET
)
694 /* in case of death shutdown to avoid blocking at close() */
695 if (shutdown(ptr
->fd
, SHUT_RDWR
) == SOCKET_ERROR
&& get_socket_errno() != ENOTCONN
)
697 WATCHPOINT_NUMBER(ptr
->fd
);
698 WATCHPOINT_ERRNO(get_socket_errno());
699 WATCHPOINT_ASSERT(get_socket_errno());
702 if (closesocket(ptr
->fd
) == SOCKET_ERROR
)
704 WATCHPOINT_ERRNO(get_socket_errno());
706 ptr
->state
= MEMCACHED_SERVER_STATE_NEW
;
707 ptr
->fd
= INVALID_SOCKET
;
710 memcached_server_write_instance_st
memcached_io_get_readable_server(memcached_st
*memc
)
712 #define MAX_SERVERS_TO_POLL 100
713 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
714 nfds_t host_index
= 0;
716 for (uint32_t x
= 0; x
< memcached_server_count(memc
) and host_index
< MAX_SERVERS_TO_POLL
; ++x
)
718 memcached_server_write_instance_st instance
= memcached_server_instance_fetch(memc
, x
);
720 if (instance
->read_buffer_length
> 0) /* I have data in the buffer */
725 if (memcached_server_response_count(instance
) > 0)
727 fds
[host_index
].events
= POLLIN
;
728 fds
[host_index
].revents
= 0;
729 fds
[host_index
].fd
= instance
->fd
;
736 /* We have 0 or 1 server with pending events.. */
737 for (uint32_t x
= 0; x
< memcached_server_count(memc
); ++x
)
739 memcached_server_write_instance_st instance
=
740 memcached_server_instance_fetch(memc
, x
);
742 if (memcached_server_response_count(instance
) > 0)
751 int error
= poll(fds
, host_index
, memc
->poll_timeout
);
755 memcached_set_errno(*memc
, get_socket_errno(), MEMCACHED_AT
);
761 for (nfds_t x
= 0; x
< host_index
; ++x
)
763 if (fds
[x
].revents
& POLLIN
)
765 for (uint32_t y
= 0; y
< memcached_server_count(memc
); ++y
)
767 memcached_server_write_instance_st instance
= memcached_server_instance_fetch(memc
, y
);
769 if (instance
->fd
== fds
[x
].fd
)
782 Eventually we will just kill off the server with the problem.
784 void memcached_io_reset(memcached_server_write_instance_st ptr
)
786 memcached_quit_server(ptr
, true);
790 * Read a given number of bytes from the server and place it into a specific
791 * buffer. Reset the IO channel on this server if an error occurs.
793 memcached_return_t
memcached_safe_read(memcached_server_write_instance_st ptr
,
798 char *data
= static_cast<char *>(dta
);
800 while (offset
< size
)
803 memcached_return_t rc
;
805 while (memcached_continue(rc
= memcached_io_read(ptr
, data
+ offset
, size
- offset
, nread
))) { };
807 if (memcached_failed(rc
))
812 offset
+= size_t(nread
);
815 return MEMCACHED_SUCCESS
;
818 memcached_return_t
memcached_io_readline(memcached_server_write_instance_st ptr
,
824 bool line_complete
= false;
826 while (line_complete
== false)
828 if (ptr
->read_buffer_length
== 0)
831 * We don't have any data in the buffer, so let's fill the read
832 * buffer. Call the standard read function to avoid duplicating
836 memcached_return_t rc
= memcached_io_read(ptr
, buffer_ptr
, 1, nread
);
837 if (memcached_failed(rc
) and rc
== MEMCACHED_IN_PROGRESS
)
839 memcached_quit_server(ptr
, true);
840 return memcached_set_error(*ptr
, rc
, MEMCACHED_AT
);
842 else if (memcached_failed(rc
))
847 if (*buffer_ptr
== '\n')
856 /* Now let's look in the buffer and copy as we go! */
857 while (ptr
->read_buffer_length
and total_nr
< size
and line_complete
== false)
859 *buffer_ptr
= *ptr
->read_ptr
;
860 if (*buffer_ptr
== '\n')
862 line_complete
= true;
864 --ptr
->read_buffer_length
;
870 if (total_nr
== size
)
872 return MEMCACHED_PROTOCOL_ERROR
;
876 return MEMCACHED_SUCCESS
;