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!
14 #include <sys/select.h>
22 static ssize_t
io_flush(memcached_server_instance_st
*ptr
, memcached_return_t
*error
);
23 static void increment_udp_message_id(memcached_server_instance_st
*ptr
);
25 static memcached_return_t
io_wait(memcached_server_instance_st
*ptr
,
26 memc_read_or_write read_or_write
)
34 unlikely (read_or_write
== MEM_WRITE
) /* write */
38 ** We are going to block on write, but at least on Solaris we might block
39 ** on write if we haven't read anything from our input buffer..
40 ** Try to purge the input buffer if we don't do any flow control in the
41 ** application layer (just sending a lot of data etc)
42 ** The test is moved down in the purge function to avoid duplication of
45 if (read_or_write
== MEM_WRITE
)
47 memcached_return_t rc
= memcached_purge(ptr
);
48 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
49 return MEMCACHED_FAILURE
;
52 int timeout
= ptr
->root
->poll_timeout
;
53 if (ptr
->root
->flags
.no_block
== false)
59 error
= poll(&fds
, 1, timeout
);
64 return MEMCACHED_SUCCESS
;
66 return MEMCACHED_TIMEOUT
;
73 ptr
->cached_errno
= error
;
74 memcached_quit_server(ptr
, 1);
76 return MEMCACHED_FAILURE
;
80 /* Imposssible for anything other then -1 */
81 WATCHPOINT_ASSERT(error
== -1);
82 ptr
->cached_errno
= error
;
83 memcached_quit_server(ptr
, 1);
85 return MEMCACHED_FAILURE
;
89 * Try to fill the input buffer for a server with as much
92 * @param ptr the server to pack
94 static bool repack_input_buffer(memcached_server_instance_st
*ptr
)
96 if (ptr
->read_ptr
!= ptr
->read_buffer
)
98 /* Move all of the data to the beginning of the buffer so
99 ** that we can fit more data into the buffer...
101 memmove(ptr
->read_buffer
, ptr
->read_ptr
, ptr
->read_buffer_length
);
102 ptr
->read_ptr
= ptr
->read_buffer
;
103 ptr
->read_data_length
= ptr
->read_buffer_length
;
106 /* There is room in the buffer, try to fill it! */
107 if (ptr
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
109 /* Just try a single read to grab what's available */
110 ssize_t nr
= read(ptr
->fd
,
111 ptr
->read_ptr
+ ptr
->read_data_length
,
112 MEMCACHED_MAX_BUFFER
- ptr
->read_data_length
);
116 ptr
->read_data_length
+= (size_t)nr
;
117 ptr
->read_buffer_length
+= (size_t)nr
;
125 * If the we have callbacks connected to this server structure
126 * we may start process the input queue and fire the callbacks
127 * for the incomming messages. This function is _only_ called
128 * when the input buffer is full, so that we _know_ that we have
129 * at least _one_ message to process.
131 * @param ptr the server to star processing iput messages for
132 * @return true if we processed anything, false otherwise
134 static bool process_input_buffer(memcached_server_instance_st
*ptr
)
137 ** We might be able to process some of the response messages if we
138 ** have a callback set up
140 if (ptr
->root
->callbacks
!= NULL
&& ptr
->root
->flags
.use_udp
== false)
143 * We might have responses... try to read them out and fire
146 memcached_callback_st cb
= *ptr
->root
->callbacks
;
148 memcached_set_processing_input((memcached_st
*)ptr
->root
, true);
150 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
151 memcached_return_t error
;
152 memcached_st
*root
= (memcached_st
*)ptr
->root
;
153 error
= memcached_response(ptr
, buffer
, sizeof(buffer
),
156 memcached_set_processing_input(root
, false);
158 if (error
== MEMCACHED_SUCCESS
)
160 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
162 error
= (*cb
.callback
[x
])(ptr
->root
, &root
->result
, cb
.context
);
163 if (error
!= MEMCACHED_SUCCESS
)
167 /* @todo what should I do with the error message??? */
169 /* @todo what should I do with other error messages?? */
176 static inline void memcached_io_cork_push(memcached_server_st
*ptr
)
180 if (ptr
->root
->flags
.cork
== false || ptr
->state
.is_corked
)
184 int err
= setsockopt(ptr
->fd
, IPPROTO_TCP
, CORK
,
185 &enable
, (socklen_t
)sizeof(int));
187 ptr
->state
.is_corked
= true;
189 WATCHPOINT_ASSERT(ptr
->state
.is_corked
== true);
193 static inline void memcached_io_cork_pop(memcached_server_st
*ptr
)
197 if (ptr
->root
->flags
.cork
== false || ptr
->state
.is_corked
== false)
201 int err
= setsockopt(ptr
->fd
, IPPROTO_TCP
, CORK
,
202 &enable
, (socklen_t
)sizeof(int));
204 ptr
->state
.is_corked
= false;
206 WATCHPOINT_ASSERT(ptr
->state
.is_corked
== false);
210 #if 0 // Dead code, this should be removed.
211 void memcached_io_preread(memcached_st
*ptr
)
217 for (x
= 0; x
< memcached_server_count(ptr
); x
++)
219 if (memcached_server_response_count(ptr
, x
) &&
220 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
224 data_read
= read(ptr
->hosts
[x
].fd
,
225 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
226 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
);
230 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
231 ptr
->hosts
[x
].read_data_length
+= data_read
;
237 memcached_return_t
memcached_io_read(memcached_server_instance_st
*ptr
,
238 void *buffer
, size_t length
, ssize_t
*nread
)
246 if (!ptr
->read_buffer_length
)
252 data_read
= read(ptr
->fd
, ptr
->read_buffer
, MEMCACHED_MAX_BUFFER
);
255 else if (data_read
== -1)
257 ptr
->cached_errno
= errno
;
258 memcached_return_t rc
= MEMCACHED_UNKNOWN_READ_FAILURE
;
266 if ((rc
= io_wait(ptr
, MEM_READ
)) == MEMCACHED_SUCCESS
)
272 memcached_quit_server(ptr
, 1);
281 EOF. Any data received so far is incomplete
282 so discard it. This always reads by byte in case of TCP
283 and protocol enforcement happens at memcached_response()
284 looking for '\n'. We do not care for UDB which requests 8 bytes
285 at once. Generally, this means that connection went away. Since
286 for blocking I/O we do not return 0 and for non-blocking case
287 it will return EGAIN if data is not immediatly available.
289 memcached_quit_server(ptr
, 1);
291 return MEMCACHED_UNKNOWN_READ_FAILURE
;
295 ptr
->io_bytes_sent
= 0;
296 ptr
->read_data_length
= (size_t) data_read
;
297 ptr
->read_buffer_length
= (size_t) data_read
;
298 ptr
->read_ptr
= ptr
->read_buffer
;
305 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
307 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
308 length
-= difference
;
309 ptr
->read_ptr
+= difference
;
310 ptr
->read_buffer_length
-= difference
;
311 buffer_ptr
+= difference
;
315 *buffer_ptr
= *ptr
->read_ptr
;
317 ptr
->read_buffer_length
--;
323 ptr
->server_failure_counter
= 0;
324 *nread
= (ssize_t
)(buffer_ptr
- (char*)buffer
);
325 return MEMCACHED_SUCCESS
;
328 ssize_t
memcached_io_write(memcached_server_instance_st
*ptr
,
329 const void *buffer
, size_t length
, bool with_flush
)
331 size_t original_length
;
332 const char* buffer_ptr
;
334 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
336 original_length
= length
;
339 /* more writable data is coming if a flush isn't required, so delay send */
342 memcached_io_cork_push(ptr
);
351 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
353 //UDP does not support partial writes
354 buffer_end
= MAX_UDP_DATAGRAM_LENGTH
;
355 should_write
= length
;
356 if (ptr
->write_buffer_offset
+ should_write
> buffer_end
)
361 buffer_end
= MEMCACHED_MAX_BUFFER
;
362 should_write
= buffer_end
- ptr
->write_buffer_offset
;
363 should_write
= (should_write
< length
) ? should_write
: length
;
366 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
367 memcpy(write_ptr
, buffer_ptr
, should_write
);
368 ptr
->write_buffer_offset
+= should_write
;
369 buffer_ptr
+= should_write
;
370 length
-= should_write
;
372 if (ptr
->write_buffer_offset
== buffer_end
&& ptr
->type
!= MEMCACHED_CONNECTION_UDP
)
374 memcached_return_t rc
;
377 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
378 sent_length
= io_flush(ptr
, &rc
);
379 if (sent_length
== -1)
382 /* If io_flush calls memcached_purge, sent_length may be 0 */
383 unlikely (sent_length
!= 0)
385 WATCHPOINT_ASSERT(sent_length
== (ssize_t
)buffer_end
);
392 memcached_return_t rc
;
393 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
394 if (io_flush(ptr
, &rc
) == -1)
399 memcached_io_cork_pop(ptr
);
402 return (ssize_t
) original_length
;
405 memcached_return_t
memcached_io_close(memcached_server_instance_st
*ptr
)
409 return MEMCACHED_SUCCESS
;
412 /* in case of death shutdown to avoid blocking at close() */
413 if (shutdown(ptr
->fd
, SHUT_RDWR
) == -1 && errno
!= ENOTCONN
)
415 WATCHPOINT_NUMBER(ptr
->fd
);
416 WATCHPOINT_ERRNO(errno
);
417 WATCHPOINT_ASSERT(errno
);
420 if (close(ptr
->fd
) == -1)
422 WATCHPOINT_ERRNO(errno
);
425 return MEMCACHED_SUCCESS
;
428 memcached_server_instance_st
*memcached_io_get_readable_server(memcached_st
*memc
)
430 #define MAX_SERVERS_TO_POLL 100
431 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
432 unsigned int host_index
= 0;
435 x
< memcached_server_count(memc
) && host_index
< MAX_SERVERS_TO_POLL
;
438 memcached_server_instance_st
*instance
=
439 memcached_server_instance_fetch(memc
, x
);
441 if (instance
->read_buffer_length
> 0) /* I have data in the buffer */
444 if (memcached_server_response_count(instance
) > 0)
446 fds
[host_index
].events
= POLLIN
;
447 fds
[host_index
].revents
= 0;
448 fds
[host_index
].fd
= instance
->fd
;
455 /* We have 0 or 1 server with pending events.. */
456 for (uint32_t x
= 0; x
< memcached_server_count(memc
); ++x
)
458 memcached_server_instance_st
*instance
=
459 memcached_server_instance_fetch(memc
, x
);
461 if (memcached_server_response_count(instance
) > 0)
470 int err
= poll(fds
, host_index
, memc
->poll_timeout
);
473 memc
->cached_errno
= errno
;
478 for (size_t x
= 0; x
< host_index
; ++x
)
480 if (fds
[x
].revents
& POLLIN
)
482 for (uint32_t y
= 0; y
< memcached_server_count(memc
); ++y
)
484 memcached_server_instance_st
*instance
=
485 memcached_server_instance_fetch(memc
, y
);
487 if (instance
->fd
== fds
[x
].fd
)
497 static ssize_t
io_flush(memcached_server_instance_st
*ptr
,
498 memcached_return_t
*error
)
501 ** We might want to purge the input buffer if we haven't consumed
502 ** any output yet... The test for the limits is the purge is inline
503 ** in the purge function to avoid duplicating the logic..
506 memcached_return_t rc
;
507 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
508 rc
= memcached_purge(ptr
);
510 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
514 size_t return_length
;
515 char *local_write_ptr
= ptr
->write_buffer
;
516 size_t write_length
= ptr
->write_buffer_offset
;
518 *error
= MEMCACHED_SUCCESS
;
520 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
522 // UDP Sanity check, make sure that we are not sending somthing too big
523 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& write_length
> MAX_UDP_DATAGRAM_LENGTH
)
526 if (ptr
->write_buffer_offset
== 0 || (ptr
->type
== MEMCACHED_CONNECTION_UDP
527 && ptr
->write_buffer_offset
== UDP_DATAGRAM_HEADER_LENGTH
))
530 /* Looking for memory overflows */
532 if (write_length
== MEMCACHED_MAX_BUFFER
)
533 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
534 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
540 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
541 WATCHPOINT_ASSERT(write_length
> 0);
543 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
544 increment_udp_message_id(ptr
);
545 sent_length
= write(ptr
->fd
, local_write_ptr
, write_length
);
547 if (sent_length
== -1)
549 ptr
->cached_errno
= errno
;
557 * We may be blocked on write because the input buffer
558 * is full. Let's check if we have room in our input
559 * buffer for more data and retry the write before
562 if (repack_input_buffer(ptr
) ||
563 process_input_buffer(ptr
))
566 memcached_return_t rc
;
567 rc
= io_wait(ptr
, MEM_WRITE
);
569 if (rc
== MEMCACHED_SUCCESS
|| rc
== MEMCACHED_TIMEOUT
)
572 memcached_quit_server(ptr
, 1);
576 memcached_quit_server(ptr
, 1);
577 *error
= MEMCACHED_ERRNO
;
582 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&&
583 (size_t)sent_length
!= write_length
)
585 memcached_quit_server(ptr
, 1);
589 ptr
->io_bytes_sent
+= (uint32_t) sent_length
;
591 local_write_ptr
+= sent_length
;
592 write_length
-= (uint32_t) sent_length
;
593 return_length
+= (uint32_t) sent_length
;
596 WATCHPOINT_ASSERT(write_length
== 0);
597 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
598 // ptr->write_buffer_offset);
600 // if we are a udp server, the begining of the buffer is reserverd for
601 // the upd frame header
602 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
603 ptr
->write_buffer_offset
= UDP_DATAGRAM_HEADER_LENGTH
;
605 ptr
->write_buffer_offset
= 0;
607 return (ssize_t
) return_length
;
611 Eventually we will just kill off the server with the problem.
613 void memcached_io_reset(memcached_server_instance_st
*ptr
)
615 memcached_quit_server(ptr
, 1);
619 * Read a given number of bytes from the server and place it into a specific
620 * buffer. Reset the IO channel on this server if an error occurs.
622 memcached_return_t
memcached_safe_read(memcached_server_instance_st
*ptr
,
629 while (offset
< size
)
632 memcached_return_t rc
= memcached_io_read(ptr
, data
+ offset
, size
- offset
,
634 if (rc
!= MEMCACHED_SUCCESS
)
637 offset
+= (size_t) nread
;
640 return MEMCACHED_SUCCESS
;
643 memcached_return_t
memcached_io_readline(memcached_server_instance_st
*ptr
,
647 bool line_complete
= false;
650 while (!line_complete
)
652 if (ptr
->read_buffer_length
== 0)
655 * We don't have any data in the buffer, so let's fill the read
656 * buffer. Call the standard read function to avoid duplicating
660 memcached_return_t rc
= memcached_io_read(ptr
, buffer_ptr
, 1, &nread
);
661 if (rc
!= MEMCACHED_SUCCESS
)
664 if (*buffer_ptr
== '\n')
671 /* Now let's look in the buffer and copy as we go! */
672 while (ptr
->read_buffer_length
&& total_nr
< size
&& !line_complete
)
674 *buffer_ptr
= *ptr
->read_ptr
;
675 if (*buffer_ptr
== '\n')
676 line_complete
= true;
677 --ptr
->read_buffer_length
;
683 if (total_nr
== size
)
684 return MEMCACHED_PROTOCOL_ERROR
;
687 return MEMCACHED_SUCCESS
;
691 * The udp request id consists of two seperate sections
693 * 2) The message number
694 * The thread id should only be set when the memcached_st struct is created
695 * and should not be changed.
697 * The message num is incremented for each new message we send, this function
698 * extracts the message number from message_id, increments it and then
699 * writes the new value back into the header
701 static void increment_udp_message_id(memcached_server_instance_st
*ptr
)
703 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
704 uint16_t cur_req
= get_udp_datagram_request_id(header
);
705 int msg_num
= get_msg_num_from_request_id(cur_req
);
706 int thread_id
= get_thread_id_from_request_id(cur_req
);
708 if (((++msg_num
) & UDP_REQUEST_ID_THREAD_MASK
) != 0)
711 header
->request_id
= htons((uint16_t) (thread_id
| msg_num
));
714 memcached_return_t
memcached_io_init_udp_header(memcached_server_instance_st
*ptr
, uint16_t thread_id
)
716 if (thread_id
> UDP_REQUEST_ID_MAX_THREAD_ID
)
717 return MEMCACHED_FAILURE
;
719 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
720 header
->request_id
= htons((uint16_t) (generate_udp_request_thread_id(thread_id
)));
721 header
->num_datagrams
= htons(1);
722 header
->sequence_number
= htons(0);
724 return MEMCACHED_SUCCESS
;