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