2 * Copyright (C) 2006-2009 Brian Aker
5 * Use and distribution licensed under the BSD license. See
6 * the COPYING file in the parent directory for full text.
8 * Summary: Server IO, Not public!
20 static ssize_t
io_flush(memcached_server_write_instance_st ptr
,
21 memcached_return_t
*error
);
22 static void increment_udp_message_id(memcached_server_write_instance_st ptr
);
24 static memcached_return_t
io_wait(memcached_server_write_instance_st ptr
,
25 memc_read_or_write read_or_write
)
33 if (read_or_write
== MEM_WRITE
) /* write */
36 WATCHPOINT_SET(ptr
->io_wait_count
.write
++);
40 WATCHPOINT_SET(ptr
->io_wait_count
.read
++);
44 ** We are going to block on write, but at least on Solaris we might block
45 ** on write if we haven't read anything from our input buffer..
46 ** Try to purge the input buffer if we don't do any flow control in the
47 ** application layer (just sending a lot of data etc)
48 ** The test is moved down in the purge function to avoid duplication of
51 if (read_or_write
== MEM_WRITE
)
53 memcached_return_t rc
= memcached_purge(ptr
);
54 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
55 return MEMCACHED_FAILURE
;
59 while (--loop_max
) // While loop is for ERESTART or EINTR
61 error
= poll(&fds
, 1, ptr
->root
->poll_timeout
);
66 WATCHPOINT_IF_LABELED_NUMBER(read_or_write
&& loop_max
< 4, "read() times we had to loop, decremented down from 5", loop_max
);
67 WATCHPOINT_IF_LABELED_NUMBER(!read_or_write
&& loop_max
< 4, "write() times we had to loop, decremented down from 5", loop_max
);
69 return MEMCACHED_SUCCESS
;
70 case 0: // Timeout occured, we let the while() loop do its thing.
71 return MEMCACHED_TIMEOUT
;
73 WATCHPOINT_ERRNO(get_socket_errno());
74 switch (get_socket_errno())
76 #ifdef TARGET_OS_LINUX
82 if (fds
.revents
& POLLERR
)
85 socklen_t len
= sizeof (err
);
86 (void)getsockopt(ptr
->fd
, SOL_SOCKET
, SO_ERROR
, &err
, &len
);
87 ptr
->cached_errno
= (err
== 0) ? get_socket_errno() : err
;
91 ptr
->cached_errno
= get_socket_errno();
93 memcached_quit_server(ptr
, true);
95 return MEMCACHED_FAILURE
;
100 /* Imposssible for anything other then -1 */
101 WATCHPOINT_ASSERT(error
== -1);
102 ptr
->cached_errno
= get_socket_errno();
103 memcached_quit_server(ptr
, true);
105 return MEMCACHED_FAILURE
;
109 * Try to fill the input buffer for a server with as much
112 * @param ptr the server to pack
114 static bool repack_input_buffer(memcached_server_write_instance_st ptr
)
116 if (ptr
->read_ptr
!= ptr
->read_buffer
)
118 /* Move all of the data to the beginning of the buffer so
119 ** that we can fit more data into the buffer...
121 memmove(ptr
->read_buffer
, ptr
->read_ptr
, ptr
->read_buffer_length
);
122 ptr
->read_ptr
= ptr
->read_buffer
;
123 ptr
->read_data_length
= ptr
->read_buffer_length
;
126 /* There is room in the buffer, try to fill it! */
127 if (ptr
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
129 /* Just try a single read to grab what's available */
130 ssize_t nr
= recv(ptr
->fd
,
131 ptr
->read_ptr
+ ptr
->read_data_length
,
132 MEMCACHED_MAX_BUFFER
- ptr
->read_data_length
,
137 ptr
->read_data_length
+= (size_t)nr
;
138 ptr
->read_buffer_length
+= (size_t)nr
;
146 * If the we have callbacks connected to this server structure
147 * we may start process the input queue and fire the callbacks
148 * for the incomming messages. This function is _only_ called
149 * when the input buffer is full, so that we _know_ that we have
150 * at least _one_ message to process.
152 * @param ptr the server to star processing iput messages for
153 * @return true if we processed anything, false otherwise
155 static bool process_input_buffer(memcached_server_write_instance_st ptr
)
158 ** We might be able to process some of the response messages if we
159 ** have a callback set up
161 if (ptr
->root
->callbacks
!= NULL
&& ptr
->root
->flags
.use_udp
== false)
164 * We might have responses... try to read them out and fire
167 memcached_callback_st cb
= *ptr
->root
->callbacks
;
169 memcached_set_processing_input((memcached_st
*)ptr
->root
, true);
171 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
172 memcached_return_t error
;
173 memcached_st
*root
= (memcached_st
*)ptr
->root
;
174 error
= memcached_response(ptr
, buffer
, sizeof(buffer
),
177 memcached_set_processing_input(root
, false);
179 if (error
== MEMCACHED_SUCCESS
)
181 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
183 error
= (*cb
.callback
[x
])(ptr
->root
, &root
->result
, cb
.context
);
184 if (error
!= MEMCACHED_SUCCESS
)
188 /* @todo what should I do with the error message??? */
190 /* @todo what should I do with other error messages?? */
197 static inline void memcached_io_cork_push(memcached_server_st
*ptr
)
201 if (ptr
->root
->flags
.cork
== false || ptr
->state
.is_corked
)
205 int err
= setsockopt(ptr
->fd
, IPPROTO_TCP
, CORK
,
206 &enable
, (socklen_t
)sizeof(int));
208 ptr
->state
.is_corked
= true;
210 WATCHPOINT_ASSERT(ptr
->state
.is_corked
== true);
214 static inline void memcached_io_cork_pop(memcached_server_st
*ptr
)
218 if (ptr
->root
->flags
.cork
== false || ptr
->state
.is_corked
== false)
222 int err
= setsockopt(ptr
->fd
, IPPROTO_TCP
, CORK
,
223 &enable
, (socklen_t
)sizeof(int));
225 ptr
->state
.is_corked
= false;
227 WATCHPOINT_ASSERT(ptr
->state
.is_corked
== false);
231 #if 0 // Dead code, this should be removed.
232 void memcached_io_preread(memcached_st
*ptr
)
238 for (x
= 0; x
< memcached_server_count(ptr
); x
++)
240 if (memcached_server_response_count(ptr
, x
) &&
241 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
245 data_read
= recv(ptr
->hosts
[x
].fd
,
246 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
247 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
, 0);
248 if (data_read
== SOCKET_ERROR
)
251 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
252 ptr
->hosts
[x
].read_data_length
+= data_read
;
258 memcached_return_t
memcached_io_read(memcached_server_write_instance_st ptr
,
259 void *buffer
, size_t length
, ssize_t
*nread
)
267 if (!ptr
->read_buffer_length
)
273 data_read
= recv(ptr
->fd
, ptr
->read_buffer
, MEMCACHED_MAX_BUFFER
, 0);
278 else if (data_read
== SOCKET_ERROR
)
280 ptr
->cached_errno
= get_socket_errno();
281 memcached_return_t rc
= MEMCACHED_ERRNO
;
282 switch (get_socket_errno())
289 #ifdef TARGET_OS_LINUX
292 if ((rc
= io_wait(ptr
, MEM_READ
)) == MEMCACHED_SUCCESS
)
298 memcached_quit_server(ptr
, true);
307 EOF. Any data received so far is incomplete
308 so discard it. This always reads by byte in case of TCP
309 and protocol enforcement happens at memcached_response()
310 looking for '\n'. We do not care for UDB which requests 8 bytes
311 at once. Generally, this means that connection went away. Since
312 for blocking I/O we do not return 0 and for non-blocking case
313 it will return EGAIN if data is not immediatly available.
315 WATCHPOINT_STRING("We had a zero length recv()");
316 memcached_quit_server(ptr
, true);
318 return MEMCACHED_UNKNOWN_READ_FAILURE
;
322 ptr
->io_bytes_sent
= 0;
323 ptr
->read_data_length
= (size_t) data_read
;
324 ptr
->read_buffer_length
= (size_t) data_read
;
325 ptr
->read_ptr
= ptr
->read_buffer
;
332 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
334 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
335 length
-= difference
;
336 ptr
->read_ptr
+= difference
;
337 ptr
->read_buffer_length
-= difference
;
338 buffer_ptr
+= difference
;
342 *buffer_ptr
= *ptr
->read_ptr
;
344 ptr
->read_buffer_length
--;
350 ptr
->server_failure_counter
= 0;
351 *nread
= (ssize_t
)(buffer_ptr
- (char*)buffer
);
352 return MEMCACHED_SUCCESS
;
355 static ssize_t
_io_write(memcached_server_write_instance_st ptr
,
356 const void *buffer
, size_t length
, bool with_flush
)
358 size_t original_length
;
359 const char* buffer_ptr
;
361 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
363 original_length
= length
;
366 /* more writable data is coming if a flush isn't required, so delay send */
369 memcached_io_cork_push(ptr
);
378 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
380 //UDP does not support partial writes
381 buffer_end
= MAX_UDP_DATAGRAM_LENGTH
;
382 should_write
= length
;
383 if (ptr
->write_buffer_offset
+ should_write
> buffer_end
)
390 buffer_end
= MEMCACHED_MAX_BUFFER
;
391 should_write
= buffer_end
- ptr
->write_buffer_offset
;
392 should_write
= (should_write
< length
) ? should_write
: length
;
395 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
396 memcpy(write_ptr
, buffer_ptr
, should_write
);
397 ptr
->write_buffer_offset
+= should_write
;
398 buffer_ptr
+= should_write
;
399 length
-= should_write
;
401 if (ptr
->write_buffer_offset
== buffer_end
&& ptr
->type
!= MEMCACHED_CONNECTION_UDP
)
403 memcached_return_t rc
;
406 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
407 sent_length
= io_flush(ptr
, &rc
);
408 if (sent_length
== -1)
413 /* If io_flush calls memcached_purge, sent_length may be 0 */
414 unlikely (sent_length
!= 0)
416 WATCHPOINT_ASSERT(sent_length
== (ssize_t
)buffer_end
);
423 memcached_return_t rc
;
424 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
425 if (io_flush(ptr
, &rc
) == -1)
430 memcached_io_cork_pop(ptr
);
433 return (ssize_t
) original_length
;
436 ssize_t
memcached_io_write(memcached_server_write_instance_st ptr
,
437 const void *buffer
, size_t length
, bool with_flush
)
439 return _io_write(ptr
, buffer
, length
, with_flush
);
442 ssize_t
memcached_io_writev(memcached_server_write_instance_st ptr
,
443 const struct libmemcached_io_vector_st
*vector
,
444 size_t number_of
, bool with_flush
)
448 for (size_t x
= 0; x
< number_of
; x
++, vector
++)
452 if ((returnable
= _io_write(ptr
, vector
->buffer
, vector
->length
, false)) == -1)
461 if (memcached_io_write(ptr
, NULL
, 0, true) == -1)
471 memcached_return_t
memcached_io_close(memcached_server_write_instance_st ptr
)
473 if (ptr
->fd
== INVALID_SOCKET
)
475 return MEMCACHED_SUCCESS
;
478 /* in case of death shutdown to avoid blocking at close() */
479 if (shutdown(ptr
->fd
, SHUT_RDWR
) == SOCKET_ERROR
&& get_socket_errno() != ENOTCONN
)
481 WATCHPOINT_NUMBER(ptr
->fd
);
482 WATCHPOINT_ERRNO(get_socket_errno());
483 WATCHPOINT_ASSERT(get_socket_errno());
486 if (closesocket(ptr
->fd
) == SOCKET_ERROR
)
488 WATCHPOINT_ERRNO(get_socket_errno());
491 return MEMCACHED_SUCCESS
;
494 memcached_server_write_instance_st
memcached_io_get_readable_server(memcached_st
*memc
)
496 #define MAX_SERVERS_TO_POLL 100
497 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
498 unsigned int host_index
= 0;
501 x
< memcached_server_count(memc
) && host_index
< MAX_SERVERS_TO_POLL
;
504 memcached_server_write_instance_st instance
=
505 memcached_server_instance_fetch(memc
, x
);
507 if (instance
->read_buffer_length
> 0) /* I have data in the buffer */
510 if (memcached_server_response_count(instance
) > 0)
512 fds
[host_index
].events
= POLLIN
;
513 fds
[host_index
].revents
= 0;
514 fds
[host_index
].fd
= instance
->fd
;
521 /* We have 0 or 1 server with pending events.. */
522 for (uint32_t x
= 0; x
< memcached_server_count(memc
); ++x
)
524 memcached_server_write_instance_st instance
=
525 memcached_server_instance_fetch(memc
, x
);
527 if (memcached_server_response_count(instance
) > 0)
536 int err
= poll(fds
, host_index
, memc
->poll_timeout
);
539 memc
->cached_errno
= get_socket_errno();
544 for (size_t x
= 0; x
< host_index
; ++x
)
546 if (fds
[x
].revents
& POLLIN
)
548 for (uint32_t y
= 0; y
< memcached_server_count(memc
); ++y
)
550 memcached_server_write_instance_st instance
=
551 memcached_server_instance_fetch(memc
, y
);
553 if (instance
->fd
== fds
[x
].fd
)
563 static ssize_t
io_flush(memcached_server_write_instance_st ptr
,
564 memcached_return_t
*error
)
567 ** We might want to purge the input buffer if we haven't consumed
568 ** any output yet... The test for the limits is the purge is inline
569 ** in the purge function to avoid duplicating the logic..
572 memcached_return_t rc
;
573 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
574 rc
= memcached_purge(ptr
);
576 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
582 size_t return_length
;
583 char *local_write_ptr
= ptr
->write_buffer
;
584 size_t write_length
= ptr
->write_buffer_offset
;
586 *error
= MEMCACHED_SUCCESS
;
588 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
590 // UDP Sanity check, make sure that we are not sending somthing too big
591 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& write_length
> MAX_UDP_DATAGRAM_LENGTH
)
596 if (ptr
->write_buffer_offset
== 0 || (ptr
->type
== MEMCACHED_CONNECTION_UDP
597 && ptr
->write_buffer_offset
== UDP_DATAGRAM_HEADER_LENGTH
))
600 /* Looking for memory overflows */
602 if (write_length
== MEMCACHED_MAX_BUFFER
)
603 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
604 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
610 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
611 WATCHPOINT_ASSERT(write_length
> 0);
613 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
614 increment_udp_message_id(ptr
);
616 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
617 sent_length
= send(ptr
->fd
, local_write_ptr
, write_length
, 0);
618 if (sent_length
== SOCKET_ERROR
)
620 ptr
->cached_errno
= get_socket_errno();
621 #if 0 // @todo I should look at why we hit this bit of code hard frequently
622 WATCHPOINT_ERRNO(get_socket_errno());
623 WATCHPOINT_NUMBER(get_socket_errno());
625 switch (get_socket_errno())
635 * We may be blocked on write because the input buffer
636 * is full. Let's check if we have room in our input
637 * buffer for more data and retry the write before
640 if (repack_input_buffer(ptr
) ||
641 process_input_buffer(ptr
))
644 memcached_return_t rc
;
645 rc
= io_wait(ptr
, MEM_WRITE
);
647 if (rc
== MEMCACHED_SUCCESS
|| rc
== MEMCACHED_TIMEOUT
)
650 memcached_quit_server(ptr
, true);
656 memcached_quit_server(ptr
, true);
657 *error
= MEMCACHED_ERRNO
;
658 WATCHPOINT_ASSERT(ptr
->fd
== -1);
663 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&&
664 (size_t)sent_length
!= write_length
)
666 memcached_quit_server(ptr
, true);
670 ptr
->io_bytes_sent
+= (uint32_t) sent_length
;
672 local_write_ptr
+= sent_length
;
673 write_length
-= (uint32_t) sent_length
;
674 return_length
+= (uint32_t) sent_length
;
677 WATCHPOINT_ASSERT(write_length
== 0);
678 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
679 // ptr->write_buffer_offset);
681 // if we are a udp server, the begining of the buffer is reserverd for
682 // the upd frame header
683 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
684 ptr
->write_buffer_offset
= UDP_DATAGRAM_HEADER_LENGTH
;
686 ptr
->write_buffer_offset
= 0;
688 return (ssize_t
) return_length
;
692 Eventually we will just kill off the server with the problem.
694 void memcached_io_reset(memcached_server_write_instance_st ptr
)
696 memcached_quit_server(ptr
, true);
700 * Read a given number of bytes from the server and place it into a specific
701 * buffer. Reset the IO channel on this server if an error occurs.
703 memcached_return_t
memcached_safe_read(memcached_server_write_instance_st ptr
,
710 while (offset
< size
)
713 memcached_return_t rc
= memcached_io_read(ptr
, data
+ offset
, size
- offset
,
715 if (rc
!= MEMCACHED_SUCCESS
)
718 offset
+= (size_t) nread
;
721 return MEMCACHED_SUCCESS
;
724 memcached_return_t
memcached_io_readline(memcached_server_write_instance_st ptr
,
728 bool line_complete
= false;
731 while (!line_complete
)
733 if (ptr
->read_buffer_length
== 0)
736 * We don't have any data in the buffer, so let's fill the read
737 * buffer. Call the standard read function to avoid duplicating
741 memcached_return_t rc
= memcached_io_read(ptr
, buffer_ptr
, 1, &nread
);
742 if (rc
!= MEMCACHED_SUCCESS
)
745 if (*buffer_ptr
== '\n')
752 /* Now let's look in the buffer and copy as we go! */
753 while (ptr
->read_buffer_length
&& total_nr
< size
&& !line_complete
)
755 *buffer_ptr
= *ptr
->read_ptr
;
756 if (*buffer_ptr
== '\n')
757 line_complete
= true;
758 --ptr
->read_buffer_length
;
764 if (total_nr
== size
)
765 return MEMCACHED_PROTOCOL_ERROR
;
768 return MEMCACHED_SUCCESS
;
772 * The udp request id consists of two seperate sections
774 * 2) The message number
775 * The thread id should only be set when the memcached_st struct is created
776 * and should not be changed.
778 * The message num is incremented for each new message we send, this function
779 * extracts the message number from message_id, increments it and then
780 * writes the new value back into the header
782 static void increment_udp_message_id(memcached_server_write_instance_st ptr
)
784 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
785 uint16_t cur_req
= get_udp_datagram_request_id(header
);
786 int msg_num
= get_msg_num_from_request_id(cur_req
);
787 int thread_id
= get_thread_id_from_request_id(cur_req
);
789 if (((++msg_num
) & UDP_REQUEST_ID_THREAD_MASK
) != 0)
792 header
->request_id
= htons((uint16_t) (thread_id
| msg_num
));
795 memcached_return_t
memcached_io_init_udp_header(memcached_server_write_instance_st ptr
, uint16_t thread_id
)
797 if (thread_id
> UDP_REQUEST_ID_MAX_THREAD_ID
)
798 return MEMCACHED_FAILURE
;
800 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
801 header
->request_id
= htons((uint16_t) (generate_udp_request_thread_id(thread_id
)));
802 header
->num_datagrams
= htons(1);
803 header
->sequence_number
= htons(0);
805 return MEMCACHED_SUCCESS
;