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
)
25 if (read_or_write
== MEM_WRITE
) /* write */
26 flags
= POLLOUT
| POLLERR
;
28 flags
= POLLIN
| POLLERR
;
30 memset(&fds
, 0, sizeof(struct pollfd
));
35 ** We are going to block on write, but at least on Solaris we might block
36 ** on write if we haven't read anything from our input buffer..
37 ** Try to purge the input buffer if we don't do any flow control in the
38 ** application layer (just sending a lot of data etc)
39 ** The test is moved down in the purge function to avoid duplication of
42 if (read_or_write
== MEM_WRITE
)
44 memcached_return rc
=memcached_purge(ptr
);
45 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
46 return MEMCACHED_FAILURE
;
49 error
= poll(fds
, 1, ptr
->root
->poll_timeout
);
52 return MEMCACHED_SUCCESS
;
55 return MEMCACHED_TIMEOUT
;
58 /* Imposssible for anything other then -1 */
59 WATCHPOINT_ASSERT(error
== -1);
60 memcached_quit_server(ptr
, 1);
62 return MEMCACHED_FAILURE
;
67 void memcached_io_preread(memcached_st
*ptr
)
73 for (x
= 0; x
< ptr
->number_of_hosts
; x
++)
75 if (memcached_server_response_count(ptr
, x
) &&
76 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
80 data_read
= read(ptr
->hosts
[x
].fd
,
81 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
82 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
);
86 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
87 ptr
->hosts
[x
].read_data_length
+= data_read
;
93 ssize_t
memcached_io_read(memcached_server_st
*ptr
,
94 void *buffer
, size_t length
)
102 if (!ptr
->read_buffer_length
)
108 data_read
= read(ptr
->fd
,
110 MEMCACHED_MAX_BUFFER
);
113 else if (data_read
== -1)
115 ptr
->cached_errno
= errno
;
123 rc
= io_wait(ptr
, MEM_READ
);
125 if (rc
== MEMCACHED_SUCCESS
)
131 memcached_quit_server(ptr
, 1);
139 EOF. Any data received so far is incomplete
140 so discard it. This always reads by byte in case of TCP
141 and protocol enforcement happens at memcached_response()
142 looking for '\n'. We do not care for UDB which requests 8 bytes
143 at once. Generally, this means that connection went away. Since
144 for blocking I/O we do not return 0 and for non-blocking case
145 it will return EGAIN if data is not immediatly available.
147 memcached_quit_server(ptr
, 1);
152 ptr
->io_bytes_sent
= 0;
153 ptr
->read_data_length
= data_read
;
154 ptr
->read_buffer_length
= data_read
;
155 ptr
->read_ptr
= ptr
->read_buffer
;
162 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
164 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
165 length
-= difference
;
166 ptr
->read_ptr
+= difference
;
167 ptr
->read_buffer_length
-= difference
;
168 buffer_ptr
+= difference
;
172 *buffer_ptr
= *ptr
->read_ptr
;
174 ptr
->read_buffer_length
--;
180 return (size_t)(buffer_ptr
- (char*)buffer
);
183 ssize_t
memcached_io_write(memcached_server_st
*ptr
,
184 const void *buffer
, size_t length
, char with_flush
)
186 size_t original_length
;
187 const char* buffer_ptr
;
189 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
191 original_length
= length
;
200 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
202 //UDP does not support partial writes
203 buffer_end
= MAX_UDP_DATAGRAM_LENGTH
;
204 should_write
= length
;
205 if (ptr
->write_buffer_offset
+ should_write
> buffer_end
)
210 buffer_end
= MEMCACHED_MAX_BUFFER
;
211 should_write
= buffer_end
- ptr
->write_buffer_offset
;
212 should_write
= (should_write
< length
) ? should_write
: length
;
215 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
216 memcpy(write_ptr
, buffer_ptr
, should_write
);
217 ptr
->write_buffer_offset
+= should_write
;
218 buffer_ptr
+= should_write
;
219 length
-= should_write
;
221 if (ptr
->write_buffer_offset
== buffer_end
&& ptr
->type
!= MEMCACHED_CONNECTION_UDP
)
226 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
227 sent_length
= io_flush(ptr
, &rc
);
228 if (sent_length
== -1)
231 /* If io_flush calls memcached_purge, sent_length may be 0 */
232 if (sent_length
!= 0)
233 WATCHPOINT_ASSERT(sent_length
== buffer_end
);
240 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
241 if (io_flush(ptr
, &rc
) == -1)
245 return original_length
;
248 memcached_return
memcached_io_close(memcached_server_st
*ptr
)
253 return MEMCACHED_SUCCESS
;
255 /* in case of death shutdown to avoid blocking at close() */
258 r
= shutdown(ptr
->fd
, SHUT_RDWR
);
261 if (r
&& errno
!= ENOTCONN
)
263 WATCHPOINT_NUMBER(ptr
->fd
);
264 WATCHPOINT_ERRNO(errno
);
265 WATCHPOINT_ASSERT(errno
);
273 WATCHPOINT_ERRNO(errno
);
276 return MEMCACHED_SUCCESS
;
279 static ssize_t
io_flush(memcached_server_st
*ptr
,
280 memcached_return
*error
)
283 ** We might want to purge the input buffer if we haven't consumed
284 ** any output yet... The test for the limits is the purge is inline
285 ** in the purge function to avoid duplicating the logic..
289 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
290 rc
= memcached_purge(ptr
);
292 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
296 size_t return_length
;
297 char *local_write_ptr
= ptr
->write_buffer
;
298 size_t write_length
= ptr
->write_buffer_offset
;
300 *error
= MEMCACHED_SUCCESS
;
302 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
304 // UDP Sanity check, make sure that we are not sending somthing too big
305 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& write_length
> MAX_UDP_DATAGRAM_LENGTH
)
308 if (ptr
->write_buffer_offset
== 0 || (ptr
->type
== MEMCACHED_CONNECTION_UDP
309 && ptr
->write_buffer_offset
== UDP_DATAGRAM_HEADER_LENGTH
))
312 /* Looking for memory overflows */
313 #if defined(HAVE_DEBUG)
314 if (write_length
== MEMCACHED_MAX_BUFFER
)
315 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
316 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
322 WATCHPOINT_ASSERT(ptr
->fd
!= -1);
323 WATCHPOINT_ASSERT(write_length
> 0);
325 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
326 increment_udp_message_id(ptr
);
327 sent_length
= write(ptr
->fd
, local_write_ptr
, write_length
);
329 if (sent_length
== -1)
331 ptr
->cached_errno
= errno
;
339 rc
= io_wait(ptr
, MEM_WRITE
);
341 if (rc
== MEMCACHED_SUCCESS
|| rc
== MEMCACHED_TIMEOUT
)
344 memcached_quit_server(ptr
, 1);
348 memcached_quit_server(ptr
, 1);
349 *error
= MEMCACHED_ERRNO
;
354 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& sent_length
!= write_length
)
356 memcached_quit_server(ptr
, 1);
360 ptr
->io_bytes_sent
+= sent_length
;
362 local_write_ptr
+= sent_length
;
363 write_length
-= sent_length
;
364 return_length
+= sent_length
;
367 WATCHPOINT_ASSERT(write_length
== 0);
368 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
369 // ptr->write_buffer_offset);
371 // if we are a udp server, the begining of the buffer is reserverd for
372 // the upd frame header
373 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
374 ptr
->write_buffer_offset
= UDP_DATAGRAM_HEADER_LENGTH
;
376 ptr
->write_buffer_offset
= 0;
378 return return_length
;
382 Eventually we will just kill off the server with the problem.
384 void memcached_io_reset(memcached_server_st
*ptr
)
386 memcached_quit_server(ptr
, 1);
390 * Read a given number of bytes from the server and place it into a specific
391 * buffer. Reset the IO channel on this server if an error occurs.
393 memcached_return
memcached_safe_read(memcached_server_st
*ptr
,
400 while (offset
< size
)
402 ssize_t nread
= memcached_io_read(ptr
, data
+ offset
, size
- offset
);
405 memcached_io_reset(ptr
);
406 return MEMCACHED_UNKNOWN_READ_FAILURE
;
411 return MEMCACHED_SUCCESS
;
414 memcached_return
memcached_io_readline(memcached_server_st
*ptr
,
418 bool line_complete
= false;
421 while (!line_complete
)
423 if (ptr
->read_buffer_length
== 0)
426 * We don't have any data in the buffer, so let's fill the read
427 * buffer. Call the standard read function to avoid duplicating
430 if (memcached_io_read(ptr
, buffer_ptr
, 1) != 1)
431 return MEMCACHED_UNKNOWN_READ_FAILURE
;
433 if (*buffer_ptr
== '\n')
440 /* Now let's look in the buffer and copy as we go! */
441 while (ptr
->read_buffer_length
&& total_nr
< size
&& !line_complete
)
443 *buffer_ptr
= *ptr
->read_ptr
;
444 if (*buffer_ptr
== '\n')
445 line_complete
= true;
446 --ptr
->read_buffer_length
;
452 if (total_nr
== size
)
453 return MEMCACHED_PROTOCOL_ERROR
;
456 return MEMCACHED_SUCCESS
;
460 * The udp request id consists of two seperate sections
462 * 2) The message number
463 * The thread id should only be set when the memcached_st struct is created
464 * and should not be changed.
466 * The message num is incremented for each new message we send, this function
467 * extracts the message number from message_id, increments it and then
468 * writes the new value back into the header
470 static void increment_udp_message_id(memcached_server_st
*ptr
)
472 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
473 uint16_t cur_req
= get_udp_datagram_request_id(header
);
474 uint16_t msg_num
= get_msg_num_from_request_id(cur_req
);
475 uint16_t thread_id
= get_thread_id_from_request_id(cur_req
);
477 if (((++msg_num
) & UDP_REQUEST_ID_THREAD_MASK
) != 0)
480 header
->request_id
= htons(thread_id
| msg_num
);
483 memcached_return
memcached_io_init_udp_header(memcached_server_st
*ptr
, uint16_t thread_id
)
485 if (thread_id
> UDP_REQUEST_ID_MAX_THREAD_ID
)
486 return MEMCACHED_FAILURE
;
488 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
489 header
->request_id
= htons(generate_udp_request_thread_id(thread_id
));
490 header
->num_datagrams
= htons(1);
491 header
->sequence_number
= htons(0);
493 return MEMCACHED_SUCCESS
;