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