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 #ifdef HAVE_SYS_SOCKET_H
43 # include <sys/socket.h>
46 void initialize_binary_request(memcached_instance_st
* server
, protocol_binary_request_header
& header
)
49 header
.request
.magic
= PROTOCOL_BINARY_REQ
;
50 header
.request
.opaque
= htons(server
->request_id
);
53 enum memc_read_or_write
{
59 * Try to fill the input buffer for a server with as much
62 * @param instance the server to pack
64 static bool repack_input_buffer(memcached_instance_st
* instance
)
66 if (instance
->read_ptr
!= instance
->read_buffer
)
68 /* Move all of the data to the beginning of the buffer so
69 ** that we can fit more data into the buffer...
71 memmove(instance
->read_buffer
, instance
->read_ptr
, instance
->read_buffer_length
);
72 instance
->read_ptr
= instance
->read_buffer
;
73 instance
->read_data_length
= instance
->read_buffer_length
;
76 /* There is room in the buffer, try to fill it! */
77 if (instance
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
80 /* Just try a single read to grab what's available */
82 if ((nr
= ::recv(instance
->fd
,
83 instance
->read_ptr
+ instance
->read_data_length
,
84 MEMCACHED_MAX_BUFFER
- instance
->read_data_length
,
89 memcached_set_error(*instance
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
);
93 switch (get_socket_errno())
98 #if EWOULDBLOCK != EAGAIN
105 break; // No IO is fine, we can just move on
108 memcached_set_errno(*instance
, get_socket_errno(), MEMCACHED_AT
);
114 else // We read data, append to our read buffer
116 instance
->read_data_length
+= size_t(nr
);
117 instance
->read_buffer_length
+= size_t(nr
);
128 * If the we have callbacks connected to this server structure
129 * we may start process the input queue and fire the callbacks
130 * for the incomming messages. This function is _only_ called
131 * when the input buffer is full, so that we _know_ that we have
132 * at least _one_ message to process.
134 * @param instance the server to star processing iput messages for
135 * @return true if we processed anything, false otherwise
137 static bool process_input_buffer(memcached_instance_st
* instance
)
140 ** We might be able to process some of the response messages if we
141 ** have a callback set up
143 if (instance
->root
->callbacks
!= NULL
)
146 * We might have responses... try to read them out and fire
149 memcached_callback_st cb
= *instance
->root
->callbacks
;
151 memcached_set_processing_input((Memcached
*)instance
->root
, true);
153 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
154 Memcached
*root
= (Memcached
*)instance
->root
;
155 memcached_return_t error
= memcached_response(instance
, buffer
, sizeof(buffer
), &root
->result
);
157 memcached_set_processing_input(root
, false);
159 if (error
== MEMCACHED_SUCCESS
)
161 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
163 error
= (*cb
.callback
[x
])(instance
->root
, &root
->result
, cb
.context
);
164 if (error
!= MEMCACHED_SUCCESS
)
170 /* @todo what should I do with the error message??? */
172 /* @todo what should I do with other error messages?? */
179 static memcached_return_t
io_wait(memcached_instance_st
* instance
,
183 ** We are going to block on write, but at least on Solaris we might block
184 ** on write if we haven't read anything from our input buffer..
185 ** Try to purge the input buffer if we don't do any flow control in the
186 ** application layer (just sending a lot of data etc)
187 ** The test is moved down in the purge function to avoid duplication of
190 if (events
& POLLOUT
)
192 if (memcached_purge(instance
) == false)
194 return MEMCACHED_FAILURE
;
199 fds
.fd
= instance
->fd
;
203 if (fds
.events
& POLLOUT
) /* write */
205 instance
->io_wait_count
.write
++;
209 instance
->io_wait_count
.read
++;
212 if (instance
->root
->poll_timeout
== 0) // Mimic 0 causes timeout behavior (not all platforms do this)
214 return memcached_set_error(*instance
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
, memcached_literal_param("poll_timeout() was set to zero"));
218 while (--loop_max
) // While loop is for ERESTART or EINTR
220 int active_fd
= poll(&fds
, 1, instance
->root
->poll_timeout
);
224 assert_msg(active_fd
== 1 , "poll() returned an unexpected number of active file descriptors");
225 if (fds
.revents
& POLLIN
or fds
.revents
& POLLOUT
)
227 return MEMCACHED_SUCCESS
;
230 if (fds
.revents
& POLLHUP
)
232 return memcached_set_error(*instance
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
,
233 memcached_literal_param("poll() detected hang up"));
236 if (fds
.revents
& POLLERR
)
238 int local_errno
= EINVAL
;
240 socklen_t len
= sizeof (err
);
241 if (getsockopt(instance
->fd
, SOL_SOCKET
, SO_ERROR
, (char*)&err
, &len
) == 0)
243 if (err
== 0) // treat this as EINTR
249 memcached_quit_server(instance
, true);
250 return memcached_set_errno(*instance
, local_errno
, MEMCACHED_AT
,
251 memcached_literal_param("poll() returned POLLHUP"));
254 return memcached_set_error(*instance
, MEMCACHED_FAILURE
, MEMCACHED_AT
, memcached_literal_param("poll() returned a value that was not dealt with"));
259 return memcached_set_error(*instance
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
, memcached_literal_param("No active_fd were found"));
262 // Only an error should result in this code being called.
263 int local_errno
= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno
264 assert_msg(active_fd
== -1 , "poll() returned an unexpected value");
275 memcached_set_error(*instance
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
278 memcached_set_error(*instance
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid"));
281 memcached_set_errno(*instance
, local_errno
, MEMCACHED_AT
, memcached_literal_param("poll"));
287 memcached_quit_server(instance
, true);
289 if (memcached_has_error(instance
))
291 return memcached_instance_error_return(instance
);
294 return memcached_set_error(*instance
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
,
295 memcached_literal_param("number of attempts to call io_wait() failed"));
298 static bool io_flush(memcached_instance_st
* instance
,
299 const bool with_flush
,
300 memcached_return_t
& error
)
303 ** We might want to purge the input buffer if we haven't consumed
304 ** any output yet... The test for the limits is the purge is inline
305 ** in the purge function to avoid duplicating the logic..
308 WATCHPOINT_ASSERT(instance
->fd
!= INVALID_SOCKET
);
310 if (memcached_purge(instance
) == false)
315 char *local_write_ptr
= instance
->write_buffer
;
316 size_t write_length
= instance
->write_buffer_offset
;
318 error
= MEMCACHED_SUCCESS
;
320 WATCHPOINT_ASSERT(instance
->fd
!= INVALID_SOCKET
);
322 /* Looking for memory overflows */
324 if (write_length
== MEMCACHED_MAX_BUFFER
)
325 WATCHPOINT_ASSERT(instance
->write_buffer
== local_write_ptr
);
326 WATCHPOINT_ASSERT((instance
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
331 WATCHPOINT_ASSERT(instance
->fd
!= INVALID_SOCKET
);
332 WATCHPOINT_ASSERT(write_length
> 0);
341 flags
= MSG_NOSIGNAL
|MSG_MORE
;
344 ssize_t sent_length
= ::send(instance
->fd
, local_write_ptr
, write_length
, flags
);
345 int local_errno
= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno
347 if (sent_length
== SOCKET_ERROR
)
349 #if 0 // @todo I should look at why we hit this bit of code hard frequently
350 WATCHPOINT_ERRNO(get_socket_errno());
351 WATCHPOINT_NUMBER(get_socket_errno());
353 switch (get_socket_errno())
358 #if EWOULDBLOCK != EAGAIN
364 * We may be blocked on write because the input buffer
365 * is full. Let's check if we have room in our input
366 * buffer for more data and retry the write before
369 if (repack_input_buffer(instance
) or process_input_buffer(instance
))
374 memcached_return_t rc
= io_wait(instance
, POLLOUT
);
375 if (memcached_success(rc
))
379 else if (rc
== MEMCACHED_TIMEOUT
)
384 memcached_quit_server(instance
, true);
385 error
= memcached_set_errno(*instance
, local_errno
, MEMCACHED_AT
);
391 memcached_quit_server(instance
, true);
392 error
= memcached_set_errno(*instance
, local_errno
, MEMCACHED_AT
);
393 WATCHPOINT_ASSERT(instance
->fd
== INVALID_SOCKET
);
398 instance
->io_bytes_sent
+= uint32_t(sent_length
);
400 local_write_ptr
+= sent_length
;
401 write_length
-= uint32_t(sent_length
);
404 WATCHPOINT_ASSERT(write_length
== 0);
405 instance
->write_buffer_offset
= 0;
410 memcached_return_t
memcached_io_wait_for_write(memcached_instance_st
* instance
)
412 return io_wait(instance
, POLLOUT
);
415 memcached_return_t
memcached_io_wait_for_read(memcached_instance_st
* instance
)
417 return io_wait(instance
, POLLIN
);
420 static memcached_return_t
_io_fill(memcached_instance_st
* instance
)
425 data_read
= ::recv(instance
->fd
, instance
->read_buffer
, MEMCACHED_MAX_BUFFER
, MSG_NOSIGNAL
);
426 int local_errno
= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno
428 if (data_read
== SOCKET_ERROR
)
430 switch (get_socket_errno())
432 case EINTR
: // We just retry
435 case ETIMEDOUT
: // OSX
436 #if EWOULDBLOCK != EAGAIN
444 memcached_return_t io_wait_ret
;
445 if (memcached_success(io_wait_ret
= io_wait(instance
, POLLIN
)))
455 case ENOTCONN
: // Programmer Error
456 WATCHPOINT_ASSERT(0);
458 WATCHPOINT_ASSERT(0);
460 assert_msg(instance
->fd
!= INVALID_SOCKET
, "Programmer error, invalid socket");
465 memcached_quit_server(instance
, true);
466 memcached_set_errno(*instance
, local_errno
, MEMCACHED_AT
);
470 return memcached_instance_error_return(instance
);
472 else if (data_read
== 0)
475 EOF. Any data received so far is incomplete
476 so discard it. This always reads by byte in case of TCP
477 and protocol enforcement happens at memcached_response()
478 looking for '\n'. We do not care for UDB which requests 8 bytes
479 at once. Generally, this means that connection went away. Since
480 for blocking I/O we do not return 0 and for non-blocking case
481 it will return EGAIN if data is not immediatly available.
483 memcached_quit_server(instance
, true);
484 return memcached_set_error(*instance
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
,
485 memcached_literal_param("::rec() returned zero, server has disconnected"));
487 instance
->io_wait_count
._bytes_read
+= data_read
;
488 } while (data_read
<= 0);
490 instance
->io_bytes_sent
= 0;
491 instance
->read_data_length
= (size_t) data_read
;
492 instance
->read_buffer_length
= (size_t) data_read
;
493 instance
->read_ptr
= instance
->read_buffer
;
495 return MEMCACHED_SUCCESS
;
498 memcached_return_t
memcached_io_read(memcached_instance_st
* instance
,
499 void *buffer
, size_t length
, ssize_t
& nread
)
501 assert(memcached_is_udp(instance
->root
) == false);
502 assert_msg(instance
, "Programmer error, memcached_io_read() recieved an invalid Instance"); // Programmer error
503 char *buffer_ptr
= static_cast<char *>(buffer
);
505 if (instance
->fd
== INVALID_SOCKET
)
508 assert_msg(int(instance
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
), "Programmer error, invalid socket state");
510 return MEMCACHED_CONNECTION_FAILURE
;
515 if (instance
->read_buffer_length
== 0)
517 memcached_return_t io_fill_ret
;
518 if (memcached_fatal(io_fill_ret
= _io_fill(instance
)))
527 size_t difference
= (length
> instance
->read_buffer_length
) ? instance
->read_buffer_length
: length
;
529 memcpy(buffer_ptr
, instance
->read_ptr
, difference
);
530 length
-= difference
;
531 instance
->read_ptr
+= difference
;
532 instance
->read_buffer_length
-= difference
;
533 buffer_ptr
+= difference
;
537 *buffer_ptr
= *instance
->read_ptr
;
538 instance
->read_ptr
++;
539 instance
->read_buffer_length
--;
545 nread
= ssize_t(buffer_ptr
- (char*)buffer
);
547 return MEMCACHED_SUCCESS
;
550 memcached_return_t
memcached_io_slurp(memcached_instance_st
* instance
)
552 assert_msg(instance
, "Programmer error, invalid Instance");
553 assert(memcached_is_udp(instance
->root
) == false);
555 if (instance
->fd
== INVALID_SOCKET
)
557 assert_msg(int(instance
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
), "Invalid socket state");
558 return MEMCACHED_CONNECTION_FAILURE
;
562 char buffer
[MEMCACHED_MAX_BUFFER
];
565 data_read
= ::recv(instance
->fd
, instance
->read_buffer
, sizeof(buffer
), MSG_NOSIGNAL
);
566 if (data_read
== SOCKET_ERROR
)
568 switch (get_socket_errno())
570 case EINTR
: // We just retry
573 case ETIMEDOUT
: // OSX
574 #if EWOULDBLOCK != EAGAIN
581 if (memcached_success(io_wait(instance
, POLLIN
)))
585 return MEMCACHED_IN_PROGRESS
;
589 case ENOTCONN
: // Programmer Error
594 assert_msg(instance
->fd
!= INVALID_SOCKET
, "Invalid socket state");
599 return MEMCACHED_CONNECTION_FAILURE
; // We want this!
602 } while (data_read
> 0);
604 return MEMCACHED_CONNECTION_FAILURE
;
607 static bool _io_write(memcached_instance_st
* instance
,
608 const void *buffer
, size_t length
, bool with_flush
,
611 assert(instance
->fd
!= INVALID_SOCKET
);
612 assert(memcached_is_udp(instance
->root
) == false);
614 const char *buffer_ptr
= static_cast<const char *>(buffer
);
616 const size_t original_length
= length
;
621 size_t buffer_end
= MEMCACHED_MAX_BUFFER
;
622 size_t should_write
= buffer_end
-instance
->write_buffer_offset
;
623 should_write
= (should_write
< length
) ? should_write
: length
;
625 write_ptr
= instance
->write_buffer
+ instance
->write_buffer_offset
;
626 memcpy(write_ptr
, buffer_ptr
, should_write
);
627 instance
->write_buffer_offset
+= should_write
;
628 buffer_ptr
+= should_write
;
629 length
-= should_write
;
631 if (instance
->write_buffer_offset
== buffer_end
)
633 WATCHPOINT_ASSERT(instance
->fd
!= INVALID_SOCKET
);
635 memcached_return_t rc
;
636 if (io_flush(instance
, with_flush
, rc
) == false)
638 written
= original_length
-length
;
646 memcached_return_t rc
;
647 WATCHPOINT_ASSERT(instance
->fd
!= INVALID_SOCKET
);
648 if (io_flush(instance
, with_flush
, rc
) == false)
650 written
= original_length
-length
;
655 written
= original_length
-length
;
660 bool memcached_io_write(memcached_instance_st
* instance
)
663 return _io_write(instance
, NULL
, 0, true, written
);
666 ssize_t
memcached_io_write(memcached_instance_st
* instance
,
667 const void *buffer
, const size_t length
, const bool with_flush
)
671 if (_io_write(instance
, buffer
, length
, with_flush
, written
) == false)
676 return ssize_t(written
);
679 bool memcached_io_writev(memcached_instance_st
* instance
,
680 libmemcached_io_vector_st vector
[],
681 const size_t number_of
, const bool with_flush
)
683 ssize_t complete_total
= 0;
686 for (size_t x
= 0; x
< number_of
; x
++, vector
++)
688 complete_total
+= vector
->length
;
692 if ((_io_write(instance
, vector
->buffer
, vector
->length
, false, written
)) == false)
702 if (memcached_io_write(instance
) == false)
708 return (complete_total
== total
);
711 void memcached_instance_st::start_close_socket()
713 if (fd
!= INVALID_SOCKET
)
715 shutdown(fd
, SHUT_WR
);
716 options
.is_shutting_down
= true;
720 void memcached_instance_st::reset_socket()
722 if (fd
!= INVALID_SOCKET
)
724 (void)closesocket(fd
);
729 void memcached_instance_st::close_socket()
731 if (fd
!= INVALID_SOCKET
)
733 int shutdown_options
= SHUT_RD
;
734 if (options
.is_shutting_down
== false)
736 shutdown_options
= SHUT_RDWR
;
739 /* in case of death shutdown to avoid blocking at close() */
740 if (shutdown(fd
, shutdown_options
) == SOCKET_ERROR
and get_socket_errno() != ENOTCONN
)
742 WATCHPOINT_NUMBER(fd
);
743 WATCHPOINT_ERRNO(get_socket_errno());
744 WATCHPOINT_ASSERT(get_socket_errno());
748 state
= MEMCACHED_SERVER_STATE_NEW
;
751 state
= MEMCACHED_SERVER_STATE_NEW
;
754 write_buffer_offset
= size_t(root
and memcached_is_udp(root
) ? UDP_DATAGRAM_HEADER_LENGTH
: 0);
755 read_buffer_length
= 0;
756 read_ptr
= read_buffer
;
757 options
.is_shutting_down
= false;
758 memcached_server_response_reset(this);
760 // We reset the version so that if we end up talking to a different server
761 // we don't have stale server version information.
762 major_version
= minor_version
= micro_version
= UINT8_MAX
;
765 memcached_instance_st
* memcached_io_get_readable_server(Memcached
*memc
, memcached_return_t
&)
767 #define MAX_SERVERS_TO_POLL 100
768 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
769 nfds_t host_index
= 0;
771 for (uint32_t x
= 0; x
< memcached_server_count(memc
) and host_index
< MAX_SERVERS_TO_POLL
; ++x
)
773 memcached_instance_st
* instance
= memcached_instance_fetch(memc
, x
);
775 if (instance
->read_buffer_length
> 0) /* I have data in the buffer */
780 if (instance
->response_count() > 0)
782 fds
[host_index
].events
= POLLIN
;
783 fds
[host_index
].revents
= 0;
784 fds
[host_index
].fd
= instance
->fd
;
791 /* We have 0 or 1 server with pending events.. */
792 for (uint32_t x
= 0; x
< memcached_server_count(memc
); ++x
)
794 memcached_instance_st
* instance
= memcached_instance_fetch(memc
, x
);
796 if (instance
->response_count() > 0)
805 int error
= poll(fds
, host_index
, memc
->poll_timeout
);
809 memcached_set_errno(*memc
, get_socket_errno(), MEMCACHED_AT
);
815 for (nfds_t x
= 0; x
< host_index
; ++x
)
817 if (fds
[x
].revents
& POLLIN
)
819 for (uint32_t y
= 0; y
< memcached_server_count(memc
); ++y
)
821 memcached_instance_st
* instance
= memcached_instance_fetch(memc
, y
);
823 if (instance
->fd
== fds
[x
].fd
)
836 Eventually we will just kill off the server with the problem.
838 void memcached_io_reset(memcached_instance_st
* instance
)
840 memcached_quit_server(instance
, true);
844 * Read a given number of bytes from the server and place it into a specific
845 * buffer. Reset the IO channel on this server if an error occurs.
847 memcached_return_t
memcached_safe_read(memcached_instance_st
* instance
,
852 char *data
= static_cast<char *>(dta
);
854 while (offset
< size
)
857 memcached_return_t rc
;
859 while (memcached_continue(rc
= memcached_io_read(instance
, data
+ offset
, size
- offset
, nread
))) { };
861 if (memcached_failed(rc
))
866 offset
+= size_t(nread
);
869 return MEMCACHED_SUCCESS
;
872 memcached_return_t
memcached_io_readline(memcached_instance_st
* instance
,
878 bool line_complete
= false;
880 while (line_complete
== false)
882 if (instance
->read_buffer_length
== 0)
885 * We don't have any data in the buffer, so let's fill the read
886 * buffer. Call the standard read function to avoid duplicating
890 memcached_return_t rc
= memcached_io_read(instance
, buffer_ptr
, 1, nread
);
891 if (memcached_failed(rc
) and rc
== MEMCACHED_IN_PROGRESS
)
893 memcached_quit_server(instance
, true);
894 return memcached_set_error(*instance
, rc
, MEMCACHED_AT
);
896 else if (memcached_failed(rc
))
901 if (*buffer_ptr
== '\n')
910 /* Now let's look in the buffer and copy as we go! */
911 while (instance
->read_buffer_length
and total_nr
< size
and line_complete
== false)
913 *buffer_ptr
= *instance
->read_ptr
;
914 if (*buffer_ptr
== '\n')
916 line_complete
= true;
918 --instance
->read_buffer_length
;
919 ++instance
->read_ptr
;
924 if (total_nr
== size
)
926 return MEMCACHED_PROTOCOL_ERROR
;
930 return MEMCACHED_SUCCESS
;