2 Basic socket buffered IO
6 #include "memcached_io.h"
7 #include <sys/select.h>
15 static ssize_t
io_flush(memcached_server_st
*ptr
, memcached_return
*error
);
16 static void increment_udp_message_id(memcached_server_st
*ptr
);
18 static memcached_return
io_wait(memcached_server_st
*ptr
,
19 memc_read_or_write read_or_write
)
27 unlikely (read_or_write
== MEM_WRITE
) /* write */
31 ** We are going to block on write, but at least on Solaris we might block
32 ** on write if we haven't read anything from our input buffer..
33 ** Try to purge the input buffer if we don't do any flow control in the
34 ** application layer (just sending a lot of data etc)
35 ** The test is moved down in the purge function to avoid duplication of
38 if (read_or_write
== MEM_WRITE
)
40 memcached_return rc
= memcached_purge(ptr
);
41 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
42 return MEMCACHED_FAILURE
;
45 int timeout
= ptr
->root
->poll_timeout
;
46 if ((ptr
->root
->flags
& MEM_NO_BLOCK
) == 0)
49 error
= poll(&fds
, 1, timeout
);
52 return MEMCACHED_SUCCESS
;
54 return MEMCACHED_TIMEOUT
;
56 /* Imposssible for anything other then -1 */
57 WATCHPOINT_ASSERT(error
== -1);
58 memcached_quit_server(ptr
, 1);
60 return MEMCACHED_FAILURE
;
64 * Try to fill the input buffer for a server with as much
67 * @param ptr the server to pack
69 static bool repack_input_buffer(memcached_server_st
*ptr
)
71 if (ptr
->read_ptr
!= ptr
->read_buffer
)
73 /* Move all of the data to the beginning of the buffer so
74 ** that we can fit more data into the buffer...
76 memmove(ptr
->read_buffer
, ptr
->read_ptr
, ptr
->read_buffer_length
);
77 ptr
->read_ptr
= ptr
->read_buffer
;
78 ptr
->read_data_length
= ptr
->read_buffer_length
;
81 /* There is room in the buffer, try to fill it! */
82 if (ptr
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
84 /* Just try a single read to grab what's available */
85 ssize_t nr
= read(ptr
->fd
,
86 ptr
->read_ptr
+ ptr
->read_data_length
,
87 MEMCACHED_MAX_BUFFER
- ptr
->read_data_length
);
91 ptr
->read_data_length
+= (size_t)nr
;
92 ptr
->read_buffer_length
+= (size_t)nr
;
100 * If the we have callbacks connected to this server structure
101 * we may start process the input queue and fire the callbacks
102 * for the incomming messages. This function is _only_ called
103 * when the input buffer is full, so that we _know_ that we have
104 * at least _one_ message to process.
106 * @param ptr the server to star processing iput messages for
107 * @return true if we processed anything, false otherwise
109 static bool process_input_buffer(memcached_server_st
*ptr
)
112 ** We might be able to process some of the response messages if we
113 ** have a callback set up
115 if (ptr
->root
->callbacks
!= NULL
&& (ptr
->root
->flags
& MEM_USE_UDP
) == 0)
118 * We might have responses... try to read them out and fire
121 memcached_callback_st cb
= *ptr
->root
->callbacks
;
123 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
124 memcached_return error
;
125 error
= memcached_response(ptr
, buffer
, sizeof(buffer
),
127 if (error
== MEMCACHED_SUCCESS
)
129 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
131 error
= (*cb
.callback
[x
])(ptr
->root
, &ptr
->root
->result
, cb
.context
);
132 if (error
!= MEMCACHED_SUCCESS
)
136 /* @todo what should I do with the error message??? */
138 /* @todo what should I do with other error messages?? */
146 #define CORK TCP_CORK
147 #elif defined TCP_NOPUSH
148 #define CORK TCP_NOPUSH
151 static void memcached_io_cork(memcached_server_st
*ptr
, int enable
)
154 if (ptr
->type
!= MEMCACHED_CONNECTION_TCP
)
157 if ((enable
&& ptr
->is_corked
) || (!enable
&& !ptr
->is_corked
))
160 int err
= setsockopt(ptr
->fd
, IPPROTO_TCP
, CORK
,
161 &enable
, (socklen_t
)sizeof(int));
163 ptr
->is_corked
= enable
;
168 void memcached_io_preread(memcached_st
*ptr
)
174 for (x
= 0; x
< ptr
->number_of_hosts
; x
++)
176 if (memcached_server_response_count(ptr
, x
) &&
177 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
181 data_read
= read(ptr
->hosts
[x
].fd
,
182 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
183 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
);
187 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
188 ptr
->hosts
[x
].read_data_length
+= data_read
;
194 memcached_return
memcached_io_read(memcached_server_st
*ptr
,
195 void *buffer
, size_t length
, ssize_t
*nread
)
203 if (!ptr
->read_buffer_length
)
209 data_read
= read(ptr
->fd
, ptr
->read_buffer
, MEMCACHED_MAX_BUFFER
);
212 else if (data_read
== -1)
214 ptr
->cached_errno
= errno
;
215 memcached_return rc
= MEMCACHED_UNKNOWN_READ_FAILURE
;
220 if ((rc
= io_wait(ptr
, MEM_READ
)) == MEMCACHED_SUCCESS
)
226 memcached_quit_server(ptr
, 1);
235 EOF. Any data received so far is incomplete
236 so discard it. This always reads by byte in case of TCP
237 and protocol enforcement happens at memcached_response()
238 looking for '\n'. We do not care for UDB which requests 8 bytes
239 at once. Generally, this means that connection went away. Since
240 for blocking I/O we do not return 0 and for non-blocking case
241 it will return EGAIN if data is not immediatly available.
243 memcached_quit_server(ptr
, 1);
245 return MEMCACHED_UNKNOWN_READ_FAILURE
;
249 ptr
->io_bytes_sent
= 0;
250 ptr
->read_data_length
= (size_t) data_read
;
251 ptr
->read_buffer_length
= (size_t) data_read
;
252 ptr
->read_ptr
= ptr
->read_buffer
;
259 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
261 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
262 length
-= difference
;
263 ptr
->read_ptr
+= difference
;
264 ptr
->read_buffer_length
-= difference
;
265 buffer_ptr
+= difference
;
269 *buffer_ptr
= *ptr
->read_ptr
;
271 ptr
->read_buffer_length
--;
277 ptr
->server_failure_counter
= 0;
278 *nread
= (ssize_t
)(buffer_ptr
- (char*)buffer
);
279 return MEMCACHED_SUCCESS
;
282 ssize_t
memcached_io_write(memcached_server_st
*ptr
,
283 const void *buffer
, size_t length
, char with_flush
)
285 size_t original_length
;
286 const char* buffer_ptr
;
288 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
290 original_length
= length
;
293 /* more writable data is coming if a flush isn't required, so delay send */
295 memcached_io_cork(ptr
, 1);
303 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
305 //UDP does not support partial writes
306 buffer_end
= MAX_UDP_DATAGRAM_LENGTH
;
307 should_write
= length
;
308 if (ptr
->write_buffer_offset
+ should_write
> buffer_end
)
313 buffer_end
= MEMCACHED_MAX_BUFFER
;
314 should_write
= buffer_end
- ptr
->write_buffer_offset
;
315 should_write
= (should_write
< length
) ? should_write
: length
;
318 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
319 memcpy(write_ptr
, buffer_ptr
, should_write
);
320 ptr
->write_buffer_offset
+= should_write
;
321 buffer_ptr
+= should_write
;
322 length
-= should_write
;
324 if (ptr
->write_buffer_offset
== buffer_end
&& ptr
->type
!= MEMCACHED_CONNECTION_UDP
)
329 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
330 sent_length
= io_flush(ptr
, &rc
);
331 if (sent_length
== -1)
334 /* If io_flush calls memcached_purge, sent_length may be 0 */
335 unlikely (sent_length
!= 0)
337 WATCHPOINT_ASSERT(sent_length
== (ssize_t
)buffer_end
);
345 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
346 if (io_flush(ptr
, &rc
) == -1)
348 memcached_io_cork(ptr
, 0);
351 return (ssize_t
) original_length
;
354 memcached_return
memcached_io_close(memcached_server_st
*ptr
)
359 return MEMCACHED_SUCCESS
;
361 /* in case of death shutdown to avoid blocking at close() */
364 r
= shutdown(ptr
->fd
, SHUT_RDWR
);
367 if (r
&& errno
!= ENOTCONN
)
369 WATCHPOINT_NUMBER(ptr
->fd
);
370 WATCHPOINT_ERRNO(errno
);
371 WATCHPOINT_ASSERT(errno
);
379 WATCHPOINT_ERRNO(errno
);
382 return MEMCACHED_SUCCESS
;
385 memcached_server_st
*memcached_io_get_readable_server(memcached_st
*memc
)
387 #define MAX_SERVERS_TO_POLL 100
388 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
389 unsigned int host_index
= 0;
391 for (unsigned int x
= 0;
392 x
< memc
->number_of_hosts
&& host_index
< MAX_SERVERS_TO_POLL
;
395 if (memc
->hosts
[x
].read_buffer_length
> 0) /* I have data in the buffer */
396 return &memc
->hosts
[x
];
398 if (memcached_server_response_count(&memc
->hosts
[x
]) > 0)
400 fds
[host_index
].events
= POLLIN
;
401 fds
[host_index
].revents
= 0;
402 fds
[host_index
].fd
= memc
->hosts
[x
].fd
;
409 /* We have 0 or 1 server with pending events.. */
410 for (unsigned int x
= 0; x
< memc
->number_of_hosts
; ++x
)
411 if (memcached_server_response_count(&memc
->hosts
[x
]) > 0)
412 return &memc
->hosts
[x
];
417 int err
= poll(fds
, host_index
, memc
->poll_timeout
);
420 memc
->cached_errno
= errno
;
425 for (unsigned int x
= 0; x
< host_index
; ++x
)
426 if (fds
[x
].revents
& POLLIN
)
427 for (unsigned int y
= 0; y
< memc
->number_of_hosts
; ++y
)
428 if (memc
->hosts
[y
].fd
== fds
[x
].fd
)
429 return &memc
->hosts
[y
];
435 static ssize_t
io_flush(memcached_server_st
*ptr
,
436 memcached_return
*error
)
439 ** We might want to purge the input buffer if we haven't consumed
440 ** any output yet... The test for the limits is the purge is inline
441 ** in the purge function to avoid duplicating the logic..
445 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
446 rc
= memcached_purge(ptr
);
448 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
452 size_t return_length
;
453 char *local_write_ptr
= ptr
->write_buffer
;
454 size_t write_length
= ptr
->write_buffer_offset
;
456 *error
= MEMCACHED_SUCCESS
;
458 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
460 // UDP Sanity check, make sure that we are not sending somthing too big
461 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& write_length
> MAX_UDP_DATAGRAM_LENGTH
)
464 if (ptr
->write_buffer_offset
== 0 || (ptr
->type
== MEMCACHED_CONNECTION_UDP
465 && ptr
->write_buffer_offset
== UDP_DATAGRAM_HEADER_LENGTH
))
468 /* Looking for memory overflows */
470 if (write_length
== MEMCACHED_MAX_BUFFER
)
471 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
472 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
478 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
479 WATCHPOINT_ASSERT(write_length
> 0);
481 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
482 increment_udp_message_id(ptr
);
483 sent_length
= write(ptr
->fd
, local_write_ptr
, write_length
);
485 if (sent_length
== -1)
487 ptr
->cached_errno
= errno
;
495 * We may be blocked on write because the input buffer
496 * is full. Let's check if we have room in our input
497 * buffer for more data and retry the write before
500 if (repack_input_buffer(ptr
) ||
501 process_input_buffer(ptr
))
505 rc
= io_wait(ptr
, MEM_WRITE
);
507 if (rc
== MEMCACHED_SUCCESS
|| rc
== MEMCACHED_TIMEOUT
)
510 memcached_quit_server(ptr
, 1);
514 memcached_quit_server(ptr
, 1);
515 *error
= MEMCACHED_ERRNO
;
520 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&&
521 (size_t)sent_length
!= write_length
)
523 memcached_quit_server(ptr
, 1);
527 ptr
->io_bytes_sent
+= (uint32_t) sent_length
;
529 local_write_ptr
+= sent_length
;
530 write_length
-= (uint32_t) sent_length
;
531 return_length
+= (uint32_t) sent_length
;
534 WATCHPOINT_ASSERT(write_length
== 0);
535 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
536 // ptr->write_buffer_offset);
538 // if we are a udp server, the begining of the buffer is reserverd for
539 // the upd frame header
540 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
541 ptr
->write_buffer_offset
= UDP_DATAGRAM_HEADER_LENGTH
;
543 ptr
->write_buffer_offset
= 0;
545 return (ssize_t
) return_length
;
549 Eventually we will just kill off the server with the problem.
551 void memcached_io_reset(memcached_server_st
*ptr
)
553 memcached_quit_server(ptr
, 1);
557 * Read a given number of bytes from the server and place it into a specific
558 * buffer. Reset the IO channel on this server if an error occurs.
560 memcached_return
memcached_safe_read(memcached_server_st
*ptr
,
567 while (offset
< size
)
570 memcached_return rc
= memcached_io_read(ptr
, data
+ offset
, size
- offset
,
572 if (rc
!= MEMCACHED_SUCCESS
)
575 offset
+= (size_t) nread
;
578 return MEMCACHED_SUCCESS
;
581 memcached_return
memcached_io_readline(memcached_server_st
*ptr
,
585 bool line_complete
= false;
588 while (!line_complete
)
590 if (ptr
->read_buffer_length
== 0)
593 * We don't have any data in the buffer, so let's fill the read
594 * buffer. Call the standard read function to avoid duplicating
598 memcached_return rc
= memcached_io_read(ptr
, buffer_ptr
, 1, &nread
);
599 if (rc
!= MEMCACHED_SUCCESS
)
602 if (*buffer_ptr
== '\n')
609 /* Now let's look in the buffer and copy as we go! */
610 while (ptr
->read_buffer_length
&& total_nr
< size
&& !line_complete
)
612 *buffer_ptr
= *ptr
->read_ptr
;
613 if (*buffer_ptr
== '\n')
614 line_complete
= true;
615 --ptr
->read_buffer_length
;
621 if (total_nr
== size
)
622 return MEMCACHED_PROTOCOL_ERROR
;
625 return MEMCACHED_SUCCESS
;
629 * The udp request id consists of two seperate sections
631 * 2) The message number
632 * The thread id should only be set when the memcached_st struct is created
633 * and should not be changed.
635 * The message num is incremented for each new message we send, this function
636 * extracts the message number from message_id, increments it and then
637 * writes the new value back into the header
639 static void increment_udp_message_id(memcached_server_st
*ptr
)
641 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
642 uint16_t cur_req
= get_udp_datagram_request_id(header
);
643 int msg_num
= get_msg_num_from_request_id(cur_req
);
644 int thread_id
= get_thread_id_from_request_id(cur_req
);
646 if (((++msg_num
) & UDP_REQUEST_ID_THREAD_MASK
) != 0)
649 header
->request_id
= htons((uint16_t) (thread_id
| msg_num
));
652 memcached_return
memcached_io_init_udp_header(memcached_server_st
*ptr
, uint16_t thread_id
)
654 if (thread_id
> UDP_REQUEST_ID_MAX_THREAD_ID
)
655 return MEMCACHED_FAILURE
;
657 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
658 header
->request_id
= htons((uint16_t) (generate_udp_request_thread_id(thread_id
)));
659 header
->num_datagrams
= htons(1);
660 header
->sequence_number
= htons(0);
662 return MEMCACHED_SUCCESS
;