Updating for 1.0.2 release
[awesomized/libmemcached] / libmemcached / io.cc
1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
2 *
3 * LibMemcached
4 *
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2009 Brian Aker
7 * All rights reserved.
8 *
9 * Redistribution and use in source and binary forms, with or without
10 * modification, are permitted provided that the following conditions are
11 * met:
12 *
13 * * Redistributions of source code must retain the above copyright
14 * notice, this list of conditions and the following disclaimer.
15 *
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
19 * distribution.
20 *
21 * * The names of its contributors may not be used to endorse or
22 * promote products derived from this software without specific prior
23 * written permission.
24 *
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.
36 *
37 */
38
39
40 #include <libmemcached/common.h>
41
42 enum memc_read_or_write {
43 MEM_READ,
44 MEM_WRITE
45 };
46
47 /*
48 * The udp request id consists of two seperate sections
49 * 1) The thread id
50 * 2) The message number
51 * The thread id should only be set when the memcached_st struct is created
52 * and should not be changed.
53 *
54 * The message num is incremented for each new message we send, this function
55 * extracts the message number from message_id, increments it and then
56 * writes the new value back into the header
57 */
58 static void increment_udp_message_id(memcached_server_write_instance_st ptr)
59 {
60 struct udp_datagram_header_st *header= (struct udp_datagram_header_st *)ptr->write_buffer;
61 uint16_t cur_req= get_udp_datagram_request_id(header);
62 int msg_num= get_msg_num_from_request_id(cur_req);
63 int thread_id= get_thread_id_from_request_id(cur_req);
64
65 if (((++msg_num) & UDP_REQUEST_ID_THREAD_MASK) != 0)
66 msg_num= 0;
67
68 header->request_id= htons((uint16_t) (thread_id | msg_num));
69 }
70
71 /**
72 * Try to fill the input buffer for a server with as much
73 * data as possible.
74 *
75 * @param ptr the server to pack
76 */
77 static bool repack_input_buffer(memcached_server_write_instance_st ptr)
78 {
79 if (ptr->read_ptr != ptr->read_buffer)
80 {
81 /* Move all of the data to the beginning of the buffer so
82 ** that we can fit more data into the buffer...
83 */
84 memmove(ptr->read_buffer, ptr->read_ptr, ptr->read_buffer_length);
85 ptr->read_ptr= ptr->read_buffer;
86 ptr->read_data_length= ptr->read_buffer_length;
87 }
88
89 /* There is room in the buffer, try to fill it! */
90 if (ptr->read_buffer_length != MEMCACHED_MAX_BUFFER)
91 {
92 do {
93 /* Just try a single read to grab what's available */
94 ssize_t nr= recv(ptr->fd,
95 ptr->read_ptr + ptr->read_data_length,
96 MEMCACHED_MAX_BUFFER - ptr->read_data_length,
97 MSG_DONTWAIT);
98
99 switch (nr)
100 {
101 case SOCKET_ERROR:
102 {
103 switch (get_socket_errno())
104 {
105 case EINTR:
106 continue;
107
108 case EWOULDBLOCK:
109 #ifdef USE_EAGAIN
110 case EAGAIN:
111 #endif
112 #ifdef TARGET_OS_LINUX
113 case ERESTART:
114 #endif
115 break; // No IO is fine, we can just move on
116
117 default:
118 memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
119 }
120 }
121 break;
122
123 case 0: // Shutdown on the socket has occurred
124 {
125 memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT);
126 }
127 break;
128
129 default:
130 {
131 ptr->read_data_length+= size_t(nr);
132 ptr->read_buffer_length+= size_t(nr);
133 return true;
134 }
135 break;
136 }
137 } while (0);
138 }
139 return false;
140 }
141
142 /**
143 * If the we have callbacks connected to this server structure
144 * we may start process the input queue and fire the callbacks
145 * for the incomming messages. This function is _only_ called
146 * when the input buffer is full, so that we _know_ that we have
147 * at least _one_ message to process.
148 *
149 * @param ptr the server to star processing iput messages for
150 * @return true if we processed anything, false otherwise
151 */
152 static bool process_input_buffer(memcached_server_write_instance_st ptr)
153 {
154 /*
155 ** We might be able to process some of the response messages if we
156 ** have a callback set up
157 */
158 if (ptr->root->callbacks != NULL && ptr->root->flags.use_udp == false)
159 {
160 /*
161 * We might have responses... try to read them out and fire
162 * callbacks
163 */
164 memcached_callback_st cb= *ptr->root->callbacks;
165
166 memcached_set_processing_input((memcached_st *)ptr->root, true);
167
168 char buffer[MEMCACHED_DEFAULT_COMMAND_SIZE];
169 memcached_return_t error;
170 memcached_st *root= (memcached_st *)ptr->root;
171 error= memcached_response(ptr, buffer, sizeof(buffer),
172 &root->result);
173
174 memcached_set_processing_input(root, false);
175
176 if (error == MEMCACHED_SUCCESS)
177 {
178 for (unsigned int x= 0; x < cb.number_of_callback; x++)
179 {
180 error= (*cb.callback[x])(ptr->root, &root->result, cb.context);
181 if (error != MEMCACHED_SUCCESS)
182 break;
183 }
184
185 /* @todo what should I do with the error message??? */
186 }
187 /* @todo what should I do with other error messages?? */
188 return true;
189 }
190
191 return false;
192 }
193
194 static memcached_return_t io_wait(memcached_server_write_instance_st ptr,
195 memc_read_or_write read_or_write)
196 {
197 struct pollfd fds;
198 fds.fd= ptr->fd;
199 fds.events= POLLIN;
200
201 if (read_or_write == MEM_WRITE) /* write */
202 {
203 fds.events= POLLOUT;
204 WATCHPOINT_SET(ptr->io_wait_count.write++);
205 }
206 else
207 {
208 WATCHPOINT_SET(ptr->io_wait_count.read++);
209 }
210
211 /*
212 ** We are going to block on write, but at least on Solaris we might block
213 ** on write if we haven't read anything from our input buffer..
214 ** Try to purge the input buffer if we don't do any flow control in the
215 ** application layer (just sending a lot of data etc)
216 ** The test is moved down in the purge function to avoid duplication of
217 ** the test.
218 */
219 if (read_or_write == MEM_WRITE)
220 {
221 memcached_return_t rc= memcached_purge(ptr);
222 if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED)
223 {
224 return MEMCACHED_FAILURE;
225 }
226 }
227
228 if (ptr->root->poll_timeout == 0) // Mimic 0 causes timeout behavior (not all platforms do this)
229 {
230 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
231 }
232
233 size_t loop_max= 5;
234 while (--loop_max) // While loop is for ERESTART or EINTR
235 {
236
237 int error= poll(&fds, 1, ptr->root->poll_timeout);
238 switch (error)
239 {
240 case 1: // Success!
241 WATCHPOINT_IF_LABELED_NUMBER(read_or_write && loop_max < 4, "read() times we had to loop, decremented down from 5", loop_max);
242 WATCHPOINT_IF_LABELED_NUMBER(!read_or_write && loop_max < 4, "write() times we had to loop, decremented down from 5", loop_max);
243
244 return MEMCACHED_SUCCESS;
245
246 case 0: // Timeout occured, we let the while() loop do its thing.
247 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
248
249 default:
250 WATCHPOINT_ERRNO(get_socket_errno());
251 switch (get_socket_errno())
252 {
253 #ifdef TARGET_OS_LINUX
254 case ERESTART:
255 #endif
256 case EINTR:
257 break;
258
259 case EFAULT:
260 case ENOMEM:
261 return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
262
263 case EINVAL:
264 return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid"));
265
266 default:
267 if (fds.revents & POLLERR)
268 {
269 int err;
270 socklen_t len= sizeof (err);
271 (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len);
272 memcached_set_errno(*ptr, (err == 0) ? get_socket_errno() : err, MEMCACHED_AT);
273 }
274 else
275 {
276 memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
277 }
278 memcached_quit_server(ptr, true);
279
280 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
281 }
282 }
283 }
284
285 memcached_quit_server(ptr, true);
286
287 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
288 }
289
290 static ssize_t io_flush(memcached_server_write_instance_st ptr,
291 const bool with_flush,
292 memcached_return_t *error)
293 {
294 /*
295 ** We might want to purge the input buffer if we haven't consumed
296 ** any output yet... The test for the limits is the purge is inline
297 ** in the purge function to avoid duplicating the logic..
298 */
299 {
300 memcached_return_t rc;
301 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
302 rc= memcached_purge(ptr);
303
304 if (rc != MEMCACHED_SUCCESS && rc != MEMCACHED_STORED)
305 {
306 return -1;
307 }
308 }
309 size_t return_length;
310 char *local_write_ptr= ptr->write_buffer;
311 size_t write_length= ptr->write_buffer_offset;
312
313 *error= MEMCACHED_SUCCESS;
314
315 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
316
317 // UDP Sanity check, make sure that we are not sending somthing too big
318 if (memcached_is_udp(ptr->root) and write_length > MAX_UDP_DATAGRAM_LENGTH)
319 {
320 *error= MEMCACHED_WRITE_FAILURE;
321 return -1;
322 }
323
324 if (ptr->write_buffer_offset == 0 or (memcached_is_udp(ptr->root) and ptr->write_buffer_offset == UDP_DATAGRAM_HEADER_LENGTH))
325 {
326 return 0;
327 }
328
329 /* Looking for memory overflows */
330 #if defined(DEBUG)
331 if (write_length == MEMCACHED_MAX_BUFFER)
332 WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
333 WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
334 #endif
335
336 return_length= 0;
337 while (write_length)
338 {
339 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
340 WATCHPOINT_ASSERT(write_length > 0);
341 if (memcached_is_udp(ptr->root))
342 {
343 increment_udp_message_id(ptr);
344 }
345
346 ssize_t sent_length= 0;
347 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
348 if (with_flush)
349 {
350 sent_length= send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT);
351 }
352 else
353 {
354 sent_length= send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE);
355 }
356
357 if (sent_length == SOCKET_ERROR)
358 {
359 memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
360 #if 0 // @todo I should look at why we hit this bit of code hard frequently
361 WATCHPOINT_ERRNO(get_socket_errno());
362 WATCHPOINT_NUMBER(get_socket_errno());
363 #endif
364 switch (get_socket_errno())
365 {
366 case ENOBUFS:
367 continue;
368 case EWOULDBLOCK:
369 #ifdef USE_EAGAIN
370 case EAGAIN:
371 #endif
372 {
373 /*
374 * We may be blocked on write because the input buffer
375 * is full. Let's check if we have room in our input
376 * buffer for more data and retry the write before
377 * waiting..
378 */
379 if (repack_input_buffer(ptr) or process_input_buffer(ptr))
380 {
381 continue;
382 }
383
384 memcached_return_t rc= io_wait(ptr, MEM_WRITE);
385 if (memcached_success(rc))
386 {
387 continue;
388 }
389 else if (rc == MEMCACHED_TIMEOUT)
390 {
391 *error= memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
392 return -1;
393 }
394
395 memcached_quit_server(ptr, true);
396 *error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
397 return -1;
398 }
399 case ENOTCONN:
400 case EPIPE:
401 default:
402 memcached_quit_server(ptr, true);
403 *error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
404 WATCHPOINT_ASSERT(ptr->fd == -1);
405 return -1;
406 }
407 }
408
409 if (memcached_is_udp(ptr->root) and size_t(sent_length) != write_length)
410 {
411 memcached_quit_server(ptr, true);
412 *error= memcached_set_error(*ptr, MEMCACHED_WRITE_FAILURE, MEMCACHED_AT);
413 return -1;
414 }
415
416 ptr->io_bytes_sent+= uint32_t(sent_length);
417
418 local_write_ptr+= sent_length;
419 write_length-= uint32_t(sent_length);
420 return_length+= uint32_t(sent_length);
421 }
422
423 WATCHPOINT_ASSERT(write_length == 0);
424 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
425 // ptr->write_buffer_offset);
426
427 // if we are a udp server, the begining of the buffer is reserverd for
428 // the upd frame header
429 if (memcached_is_udp(ptr->root))
430 {
431 ptr->write_buffer_offset= UDP_DATAGRAM_HEADER_LENGTH;
432 }
433 else
434 {
435 ptr->write_buffer_offset= 0;
436 }
437
438 return (ssize_t) return_length;
439 }
440
441 memcached_return_t memcached_io_wait_for_write(memcached_server_write_instance_st ptr)
442 {
443 return io_wait(ptr, MEM_WRITE);
444 }
445
446 memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr,
447 void *buffer, size_t length, ssize_t *nread)
448 {
449 assert_msg(ptr, "Programmer error, memcached_io_read() recieved an invalid memcached_server_write_instance_st"); // Programmer error
450 char *buffer_ptr= static_cast<char *>(buffer);
451
452 if (ptr->fd == INVALID_SOCKET)
453 {
454 assert_msg(int(ptr->state) <= int(MEMCACHED_SERVER_STATE_ADDRINFO), "Programmer error, invalid socket state");
455 return MEMCACHED_CONNECTION_FAILURE;
456 }
457
458 while (length)
459 {
460 if (not ptr->read_buffer_length)
461 {
462 ssize_t data_read;
463 do
464 {
465 data_read= recv(ptr->fd, ptr->read_buffer, MEMCACHED_MAX_BUFFER, MSG_DONTWAIT);
466 if (data_read == SOCKET_ERROR)
467 {
468 switch (get_socket_errno())
469 {
470 case EINTR: // We just retry
471 continue;
472
473 case ETIMEDOUT: // OSX
474 case EWOULDBLOCK:
475 #ifdef USE_EAGAIN
476 case EAGAIN:
477 #endif
478 #ifdef TARGET_OS_LINUX
479 case ERESTART:
480 #endif
481 if (memcached_success(io_wait(ptr, MEM_READ)))
482 {
483 continue;
484 }
485 return MEMCACHED_IN_PROGRESS;
486
487 /* fall through */
488
489 case ENOTCONN: // Programmer Error
490 WATCHPOINT_ASSERT(0);
491 case ENOTSOCK:
492 WATCHPOINT_ASSERT(0);
493 case EBADF:
494 assert_msg(ptr->fd != INVALID_SOCKET, "Programmer error, invalid socket");
495 case EINVAL:
496 case EFAULT:
497 case ECONNREFUSED:
498 default:
499 {
500 memcached_quit_server(ptr, true);
501 *nread= -1;
502 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
503 }
504 }
505 }
506 else if (data_read == 0)
507 {
508 /*
509 EOF. Any data received so far is incomplete
510 so discard it. This always reads by byte in case of TCP
511 and protocol enforcement happens at memcached_response()
512 looking for '\n'. We do not care for UDB which requests 8 bytes
513 at once. Generally, this means that connection went away. Since
514 for blocking I/O we do not return 0 and for non-blocking case
515 it will return EGAIN if data is not immediatly available.
516 */
517 WATCHPOINT_STRING("We had a zero length recv()");
518 memcached_quit_server(ptr, true);
519 *nread= -1;
520 return memcached_set_error(*ptr, MEMCACHED_UNKNOWN_READ_FAILURE, MEMCACHED_AT);
521 }
522 } while (data_read <= 0);
523
524 ptr->io_bytes_sent = 0;
525 ptr->read_data_length= (size_t) data_read;
526 ptr->read_buffer_length= (size_t) data_read;
527 ptr->read_ptr= ptr->read_buffer;
528 }
529
530 if (length > 1)
531 {
532 size_t difference;
533
534 difference= (length > ptr->read_buffer_length) ? ptr->read_buffer_length : length;
535
536 memcpy(buffer_ptr, ptr->read_ptr, difference);
537 length -= difference;
538 ptr->read_ptr+= difference;
539 ptr->read_buffer_length-= difference;
540 buffer_ptr+= difference;
541 }
542 else
543 {
544 *buffer_ptr= *ptr->read_ptr;
545 ptr->read_ptr++;
546 ptr->read_buffer_length--;
547 buffer_ptr++;
548 break;
549 }
550 }
551
552 *nread = (ssize_t)(buffer_ptr - (char*)buffer);
553
554 return MEMCACHED_SUCCESS;
555 }
556
557 memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr)
558 {
559 assert_msg(ptr, "Programmer error, invalid memcached_server_write_instance_st");
560
561 if (ptr->fd == INVALID_SOCKET)
562 {
563 assert_msg(int(ptr->state) <= int(MEMCACHED_SERVER_STATE_ADDRINFO), "Invalid socket state");
564 return MEMCACHED_CONNECTION_FAILURE;
565 }
566
567 ssize_t data_read;
568 char buffer[MEMCACHED_MAX_BUFFER];
569 do
570 {
571 data_read= recv(ptr->fd, ptr->read_buffer, sizeof(buffer), MSG_DONTWAIT);
572 if (data_read == SOCKET_ERROR)
573 {
574 switch (get_socket_errno())
575 {
576 case EINTR: // We just retry
577 continue;
578
579 case ETIMEDOUT: // OSX
580 case EWOULDBLOCK:
581 #ifdef USE_EAGAIN
582 case EAGAIN:
583 #endif
584 #ifdef TARGET_OS_LINUX
585 case ERESTART:
586 #endif
587 if (memcached_success(io_wait(ptr, MEM_READ)))
588 {
589 continue;
590 }
591 return MEMCACHED_IN_PROGRESS;
592
593 /* fall through */
594
595 case ENOTCONN: // Programmer Error
596 WATCHPOINT_ASSERT(0);
597 case ENOTSOCK:
598 WATCHPOINT_ASSERT(0);
599 case EBADF:
600 assert_msg(ptr->fd != INVALID_SOCKET, "Invalid socket state");
601 case EINVAL:
602 case EFAULT:
603 case ECONNREFUSED:
604 default:
605 return MEMCACHED_CONNECTION_FAILURE; // We want this!
606 }
607 }
608 } while (data_read > 0);
609
610 return MEMCACHED_CONNECTION_FAILURE;
611 }
612
613 static ssize_t _io_write(memcached_server_write_instance_st ptr,
614 const void *buffer, size_t length, bool with_flush)
615 {
616 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
617
618 size_t original_length= length;
619 const char *buffer_ptr= static_cast<const char *>(buffer);
620
621 while (length)
622 {
623 char *write_ptr;
624 size_t should_write;
625 size_t buffer_end;
626
627 if (memcached_is_udp(ptr->root))
628 {
629 //UDP does not support partial writes
630 buffer_end= MAX_UDP_DATAGRAM_LENGTH;
631 should_write= length;
632 if (ptr->write_buffer_offset + should_write > buffer_end)
633 {
634 return -1;
635 }
636 }
637 else
638 {
639 buffer_end= MEMCACHED_MAX_BUFFER;
640 should_write= buffer_end - ptr->write_buffer_offset;
641 should_write= (should_write < length) ? should_write : length;
642 }
643
644 write_ptr= ptr->write_buffer + ptr->write_buffer_offset;
645 memcpy(write_ptr, buffer_ptr, should_write);
646 ptr->write_buffer_offset+= should_write;
647 buffer_ptr+= should_write;
648 length-= should_write;
649
650 if (ptr->write_buffer_offset == buffer_end and memcached_is_udp(ptr->root) == false)
651 {
652 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
653
654 memcached_return_t rc;
655 ssize_t sent_length= io_flush(ptr, with_flush, &rc);
656 if (sent_length == -1)
657 {
658 return -1;
659 }
660
661 /* If io_flush calls memcached_purge, sent_length may be 0 */
662 unlikely (sent_length != 0)
663 {
664 WATCHPOINT_ASSERT(sent_length == (ssize_t)buffer_end);
665 }
666 }
667 }
668
669 if (with_flush)
670 {
671 memcached_return_t rc;
672 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
673 if (io_flush(ptr, with_flush, &rc) == -1)
674 {
675 return -1;
676 }
677 }
678
679 return (ssize_t) original_length;
680 }
681
682 ssize_t memcached_io_write(memcached_server_write_instance_st ptr,
683 const void *buffer, size_t length, bool with_flush)
684 {
685 return _io_write(ptr, buffer, length, with_flush);
686 }
687
688 ssize_t memcached_io_writev(memcached_server_write_instance_st ptr,
689 const struct libmemcached_io_vector_st *vector,
690 size_t number_of, bool with_flush)
691 {
692 ssize_t total= 0;
693
694 for (size_t x= 0; x < number_of; x++, vector++)
695 {
696 ssize_t returnable;
697
698 if ((returnable= _io_write(ptr, vector->buffer, vector->length, false)) == -1)
699 {
700 return -1;
701 }
702 total+= returnable;
703 }
704
705 if (with_flush)
706 {
707 if (memcached_io_write(ptr, NULL, 0, true) == -1)
708 {
709 return -1;
710 }
711 }
712
713 return total;
714 }
715
716
717 void memcached_io_close(memcached_server_write_instance_st ptr)
718 {
719 if (ptr->fd == INVALID_SOCKET)
720 {
721 return;
722 }
723
724 /* in case of death shutdown to avoid blocking at close() */
725 if (shutdown(ptr->fd, SHUT_RDWR) == SOCKET_ERROR && get_socket_errno() != ENOTCONN)
726 {
727 WATCHPOINT_NUMBER(ptr->fd);
728 WATCHPOINT_ERRNO(get_socket_errno());
729 WATCHPOINT_ASSERT(get_socket_errno());
730 }
731
732 if (closesocket(ptr->fd) == SOCKET_ERROR)
733 {
734 WATCHPOINT_ERRNO(get_socket_errno());
735 }
736 ptr->state= MEMCACHED_SERVER_STATE_NEW;
737 ptr->fd= INVALID_SOCKET;
738 }
739
740 memcached_server_write_instance_st memcached_io_get_readable_server(memcached_st *memc)
741 {
742 #define MAX_SERVERS_TO_POLL 100
743 struct pollfd fds[MAX_SERVERS_TO_POLL];
744 unsigned int host_index= 0;
745
746 for (uint32_t x= 0; x < memcached_server_count(memc) && host_index < MAX_SERVERS_TO_POLL; ++x)
747 {
748 memcached_server_write_instance_st instance=
749 memcached_server_instance_fetch(memc, x);
750
751 if (instance->read_buffer_length > 0) /* I have data in the buffer */
752 return instance;
753
754 if (memcached_server_response_count(instance) > 0)
755 {
756 fds[host_index].events = POLLIN;
757 fds[host_index].revents = 0;
758 fds[host_index].fd = instance->fd;
759 ++host_index;
760 }
761 }
762
763 if (host_index < 2)
764 {
765 /* We have 0 or 1 server with pending events.. */
766 for (uint32_t x= 0; x< memcached_server_count(memc); ++x)
767 {
768 memcached_server_write_instance_st instance=
769 memcached_server_instance_fetch(memc, x);
770
771 if (memcached_server_response_count(instance) > 0)
772 {
773 return instance;
774 }
775 }
776
777 return NULL;
778 }
779
780 int error= poll(fds, host_index, memc->poll_timeout);
781 switch (error)
782 {
783 case -1:
784 memcached_set_errno(*memc, get_socket_errno(), MEMCACHED_AT);
785 /* FALLTHROUGH */
786 case 0:
787 break;
788
789 default:
790 for (size_t x= 0; x < host_index; ++x)
791 {
792 if (fds[x].revents & POLLIN)
793 {
794 for (uint32_t y= 0; y < memcached_server_count(memc); ++y)
795 {
796 memcached_server_write_instance_st instance=
797 memcached_server_instance_fetch(memc, y);
798
799 if (instance->fd == fds[x].fd)
800 return instance;
801 }
802 }
803 }
804 }
805
806 return NULL;
807 }
808
809 /*
810 Eventually we will just kill off the server with the problem.
811 */
812 void memcached_io_reset(memcached_server_write_instance_st ptr)
813 {
814 memcached_quit_server(ptr, true);
815 }
816
817 /**
818 * Read a given number of bytes from the server and place it into a specific
819 * buffer. Reset the IO channel on this server if an error occurs.
820 */
821 memcached_return_t memcached_safe_read(memcached_server_write_instance_st ptr,
822 void *dta,
823 size_t size)
824 {
825 size_t offset= 0;
826 char *data= static_cast<char *>(dta);
827
828 while (offset < size)
829 {
830 ssize_t nread;
831 memcached_return_t rc;
832
833 while (memcached_continue(rc= memcached_io_read(ptr, data + offset, size - offset, &nread))) { };
834
835 if (memcached_failed(rc))
836 {
837 return rc;
838 }
839
840 offset+= (size_t) nread;
841 }
842
843 return MEMCACHED_SUCCESS;
844 }
845
846 memcached_return_t memcached_io_readline(memcached_server_write_instance_st ptr,
847 char *buffer_ptr,
848 size_t size,
849 size_t& total_nr)
850 {
851 total_nr= 0;
852 bool line_complete= false;
853
854 while (not line_complete)
855 {
856 if (ptr->read_buffer_length == 0)
857 {
858 /*
859 * We don't have any data in the buffer, so let's fill the read
860 * buffer. Call the standard read function to avoid duplicating
861 * the logic.
862 */
863 ssize_t nread;
864 memcached_return_t rc= memcached_io_read(ptr, buffer_ptr, 1, &nread);
865 if (memcached_failed(rc) and rc == MEMCACHED_IN_PROGRESS)
866 {
867 memcached_quit_server(ptr, true);
868 return memcached_set_error(*ptr, rc, MEMCACHED_AT);
869 }
870 else if (memcached_failed(rc))
871 {
872 return rc;
873 }
874
875 if (*buffer_ptr == '\n')
876 line_complete= true;
877
878 ++buffer_ptr;
879 ++total_nr;
880 }
881
882 /* Now let's look in the buffer and copy as we go! */
883 while (ptr->read_buffer_length && total_nr < size && !line_complete)
884 {
885 *buffer_ptr = *ptr->read_ptr;
886 if (*buffer_ptr == '\n')
887 line_complete = true;
888 --ptr->read_buffer_length;
889 ++ptr->read_ptr;
890 ++total_nr;
891 ++buffer_ptr;
892 }
893
894 if (total_nr == size)
895 return MEMCACHED_PROTOCOL_ERROR;
896 }
897
898 return MEMCACHED_SUCCESS;
899 }
900
901 memcached_return_t memcached_io_init_udp_header(memcached_server_write_instance_st ptr, uint16_t thread_id)
902 {
903 if (thread_id > UDP_REQUEST_ID_MAX_THREAD_ID)
904 return MEMCACHED_FAILURE;
905
906 struct udp_datagram_header_st *header= (struct udp_datagram_header_st *)ptr->write_buffer;
907 header->request_id= htons((uint16_t) (generate_udp_request_thread_id(thread_id)));
908 header->num_datagrams= htons(1);
909 header->sequence_number= htons(0);
910
911 return MEMCACHED_SUCCESS;
912 }