Fix
[m6w6/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;
334 if (with_flush)
335 {
336 flags= MSG_NOSIGNAL|MSG_DONTWAIT;
337 }
338 else
339 {
340 flags= MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE;
341 }
342
343 ssize_t sent_length= ::send(ptr->fd, local_write_ptr, write_length, flags);
344
345 if (sent_length == SOCKET_ERROR)
346 {
347 #if 0 // @todo I should look at why we hit this bit of code hard frequently
348 WATCHPOINT_ERRNO(get_socket_errno());
349 WATCHPOINT_NUMBER(get_socket_errno());
350 #endif
351 switch (get_socket_errno())
352 {
353 case ENOBUFS:
354 continue;
355
356 #if EWOULDBLOCK != EAGAIN
357 case EWOULDBLOCK:
358 #endif
359 case EAGAIN:
360 {
361 /*
362 * We may be blocked on write because the input buffer
363 * is full. Let's check if we have room in our input
364 * buffer for more data and retry the write before
365 * waiting..
366 */
367 if (repack_input_buffer(ptr) or process_input_buffer(ptr))
368 {
369 continue;
370 }
371
372 memcached_return_t rc= io_wait(ptr, MEM_WRITE);
373 if (memcached_success(rc))
374 {
375 continue;
376 }
377 else if (rc == MEMCACHED_TIMEOUT)
378 {
379 return false;
380 }
381
382 memcached_quit_server(ptr, true);
383 error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
384 return false;
385 }
386 case ENOTCONN:
387 case EPIPE:
388 default:
389 memcached_quit_server(ptr, true);
390 error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
391 WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
392 return false;
393 }
394 }
395
396 ptr->io_bytes_sent+= uint32_t(sent_length);
397
398 local_write_ptr+= sent_length;
399 write_length-= uint32_t(sent_length);
400 }
401
402 WATCHPOINT_ASSERT(write_length == 0);
403 ptr->write_buffer_offset= 0;
404
405 return true;
406 }
407
408 memcached_return_t memcached_io_wait_for_write(org::libmemcached::Instance* ptr)
409 {
410 return io_wait(ptr, MEM_WRITE);
411 }
412
413 static memcached_return_t _io_fill(org::libmemcached::Instance* ptr)
414 {
415 ssize_t data_read;
416 do
417 {
418 data_read= ::recv(ptr->fd, ptr->read_buffer, MEMCACHED_MAX_BUFFER, MSG_DONTWAIT);
419 if (data_read == SOCKET_ERROR)
420 {
421 switch (get_socket_errno())
422 {
423 case EINTR: // We just retry
424 continue;
425
426 case ETIMEDOUT: // OSX
427 #if EWOULDBLOCK != EAGAIN
428 case EWOULDBLOCK:
429 #endif
430 case EAGAIN:
431 #ifdef TARGET_OS_LINUX
432 case ERESTART:
433 #endif
434 {
435 memcached_return_t io_wait_ret;
436 if (memcached_success(io_wait_ret= io_wait(ptr, MEM_READ)))
437 {
438 continue;
439 }
440
441 return io_wait_ret;
442 }
443
444 /* fall through */
445
446 case ENOTCONN: // Programmer Error
447 WATCHPOINT_ASSERT(0);
448 case ENOTSOCK:
449 WATCHPOINT_ASSERT(0);
450 case EBADF:
451 assert_msg(ptr->fd != INVALID_SOCKET, "Programmer error, invalid socket");
452 case EINVAL:
453 case EFAULT:
454 case ECONNREFUSED:
455 default:
456 memcached_quit_server(ptr, true);
457 memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
458 break;
459 }
460
461 return memcached_instance_error_return(ptr);
462 }
463 else if (data_read == 0)
464 {
465 /*
466 EOF. Any data received so far is incomplete
467 so discard it. This always reads by byte in case of TCP
468 and protocol enforcement happens at memcached_response()
469 looking for '\n'. We do not care for UDB which requests 8 bytes
470 at once. Generally, this means that connection went away. Since
471 for blocking I/O we do not return 0 and for non-blocking case
472 it will return EGAIN if data is not immediatly available.
473 */
474 memcached_quit_server(ptr, true);
475 return memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT,
476 memcached_literal_param("::rec() returned zero, server has disconnected"));
477 }
478 ptr->io_wait_count._bytes_read+= data_read;
479 } while (data_read <= 0);
480
481 ptr->io_bytes_sent= 0;
482 ptr->read_data_length= (size_t) data_read;
483 ptr->read_buffer_length= (size_t) data_read;
484 ptr->read_ptr= ptr->read_buffer;
485
486 return MEMCACHED_SUCCESS;
487 }
488
489 memcached_return_t memcached_io_read(org::libmemcached::Instance* ptr,
490 void *buffer, size_t length, ssize_t& nread)
491 {
492 assert(memcached_is_udp(ptr->root) == false);
493 assert_msg(ptr, "Programmer error, memcached_io_read() recieved an invalid Instance"); // Programmer error
494 char *buffer_ptr= static_cast<char *>(buffer);
495
496 if (ptr->fd == INVALID_SOCKET)
497 {
498 #if 0
499 assert_msg(int(ptr->state) <= int(MEMCACHED_SERVER_STATE_ADDRINFO), "Programmer error, invalid socket state");
500 #endif
501 return MEMCACHED_CONNECTION_FAILURE;
502 }
503
504 while (length)
505 {
506 if (ptr->read_buffer_length == 0)
507 {
508 memcached_return_t io_fill_ret;
509 if (memcached_fatal(io_fill_ret= _io_fill(ptr)))
510 {
511 nread= -1;
512 return io_fill_ret;
513 }
514 }
515
516 if (length > 1)
517 {
518 size_t difference= (length > ptr->read_buffer_length) ? ptr->read_buffer_length : length;
519
520 memcpy(buffer_ptr, ptr->read_ptr, difference);
521 length -= difference;
522 ptr->read_ptr+= difference;
523 ptr->read_buffer_length-= difference;
524 buffer_ptr+= difference;
525 }
526 else
527 {
528 *buffer_ptr= *ptr->read_ptr;
529 ptr->read_ptr++;
530 ptr->read_buffer_length--;
531 buffer_ptr++;
532 break;
533 }
534 }
535
536 nread= ssize_t(buffer_ptr - (char*)buffer);
537
538 return MEMCACHED_SUCCESS;
539 }
540
541 memcached_return_t memcached_io_slurp(org::libmemcached::Instance* ptr)
542 {
543 assert_msg(ptr, "Programmer error, invalid Instance");
544 assert(memcached_is_udp(ptr->root) == false);
545
546 if (ptr->fd == INVALID_SOCKET)
547 {
548 assert_msg(int(ptr->state) <= int(MEMCACHED_SERVER_STATE_ADDRINFO), "Invalid socket state");
549 return MEMCACHED_CONNECTION_FAILURE;
550 }
551
552 ssize_t data_read;
553 char buffer[MEMCACHED_MAX_BUFFER];
554 do
555 {
556 data_read= recv(ptr->fd, ptr->read_buffer, sizeof(buffer), MSG_DONTWAIT);
557 if (data_read == SOCKET_ERROR)
558 {
559 switch (get_socket_errno())
560 {
561 case EINTR: // We just retry
562 continue;
563
564 case ETIMEDOUT: // OSX
565 #if EWOULDBLOCK != EAGAIN
566 case EWOULDBLOCK:
567 #endif
568 case EAGAIN:
569 #ifdef TARGET_OS_LINUX
570 case ERESTART:
571 #endif
572 if (memcached_success(io_wait(ptr, MEM_READ)))
573 {
574 continue;
575 }
576 return MEMCACHED_IN_PROGRESS;
577
578 /* fall through */
579
580 case ENOTCONN: // Programmer Error
581 assert(0);
582 case ENOTSOCK:
583 assert(0);
584 case EBADF:
585 assert_msg(ptr->fd != INVALID_SOCKET, "Invalid socket state");
586 case EINVAL:
587 case EFAULT:
588 case ECONNREFUSED:
589 default:
590 return MEMCACHED_CONNECTION_FAILURE; // We want this!
591 }
592 }
593 } while (data_read > 0);
594
595 return MEMCACHED_CONNECTION_FAILURE;
596 }
597
598 static bool _io_write(org::libmemcached::Instance* ptr,
599 const void *buffer, size_t length, bool with_flush,
600 size_t& written)
601 {
602 assert(ptr->fd != INVALID_SOCKET);
603 assert(memcached_is_udp(ptr->root) == false);
604
605 const char *buffer_ptr= static_cast<const char *>(buffer);
606
607 const size_t original_length= length;
608
609 while (length)
610 {
611 char *write_ptr;
612 size_t buffer_end= MEMCACHED_MAX_BUFFER;
613 size_t should_write= buffer_end -ptr->write_buffer_offset;
614 should_write= (should_write < length) ? should_write : length;
615
616 write_ptr= ptr->write_buffer + ptr->write_buffer_offset;
617 memcpy(write_ptr, buffer_ptr, should_write);
618 ptr->write_buffer_offset+= should_write;
619 buffer_ptr+= should_write;
620 length-= should_write;
621
622 if (ptr->write_buffer_offset == buffer_end)
623 {
624 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
625
626 memcached_return_t rc;
627 if (io_flush(ptr, with_flush, rc) == false)
628 {
629 written= original_length -length;
630 return false;
631 }
632 }
633 }
634
635 if (with_flush)
636 {
637 memcached_return_t rc;
638 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
639 if (io_flush(ptr, with_flush, rc) == false)
640 {
641 written= original_length -length;
642 return false;
643 }
644 }
645
646 written= original_length -length;
647
648 return true;
649 }
650
651 bool memcached_io_write(org::libmemcached::Instance* ptr)
652 {
653 size_t written;
654 return _io_write(ptr, NULL, 0, true, written);
655 }
656
657 ssize_t memcached_io_write(org::libmemcached::Instance* ptr,
658 const void *buffer, const size_t length, const bool with_flush)
659 {
660 size_t written;
661
662 if (_io_write(ptr, buffer, length, with_flush, written) == false)
663 {
664 return -1;
665 }
666
667 return ssize_t(written);
668 }
669
670 bool memcached_io_writev(org::libmemcached::Instance* ptr,
671 libmemcached_io_vector_st vector[],
672 const size_t number_of, const bool with_flush)
673 {
674 ssize_t complete_total= 0;
675 ssize_t total= 0;
676
677 for (size_t x= 0; x < number_of; x++, vector++)
678 {
679 complete_total+= vector->length;
680 if (vector->length)
681 {
682 size_t written;
683 if ((_io_write(ptr, vector->buffer, vector->length, false, written)) == false)
684 {
685 return false;
686 }
687 total+= written;
688 }
689 }
690
691 if (with_flush)
692 {
693 if (memcached_io_write(ptr) == false)
694 {
695 return false;
696 }
697 }
698
699 return (complete_total == total);
700 }
701
702
703 void memcached_io_close(org::libmemcached::Instance* ptr)
704 {
705 if (ptr->fd == INVALID_SOCKET)
706 {
707 return;
708 }
709
710 /* in case of death shutdown to avoid blocking at close() */
711 if (shutdown(ptr->fd, SHUT_RDWR) == SOCKET_ERROR && get_socket_errno() != ENOTCONN)
712 {
713 WATCHPOINT_NUMBER(ptr->fd);
714 WATCHPOINT_ERRNO(get_socket_errno());
715 WATCHPOINT_ASSERT(get_socket_errno());
716 }
717
718 if (closesocket(ptr->fd) == SOCKET_ERROR)
719 {
720 WATCHPOINT_ERRNO(get_socket_errno());
721 }
722 ptr->state= MEMCACHED_SERVER_STATE_NEW;
723 ptr->fd= INVALID_SOCKET;
724 }
725
726 org::libmemcached::Instance* memcached_io_get_readable_server(memcached_st *memc)
727 {
728 #define MAX_SERVERS_TO_POLL 100
729 struct pollfd fds[MAX_SERVERS_TO_POLL];
730 nfds_t host_index= 0;
731
732 for (uint32_t x= 0; x < memcached_server_count(memc) and host_index < MAX_SERVERS_TO_POLL; ++x)
733 {
734 org::libmemcached::Instance* instance= memcached_instance_fetch(memc, x);
735
736 if (instance->read_buffer_length > 0) /* I have data in the buffer */
737 {
738 return instance;
739 }
740
741 if (memcached_instance_response_count(instance) > 0)
742 {
743 fds[host_index].events= POLLIN;
744 fds[host_index].revents= 0;
745 fds[host_index].fd= instance->fd;
746 ++host_index;
747 }
748 }
749
750 if (host_index < 2)
751 {
752 /* We have 0 or 1 server with pending events.. */
753 for (uint32_t x= 0; x< memcached_server_count(memc); ++x)
754 {
755 org::libmemcached::Instance* instance= memcached_instance_fetch(memc, x);
756
757 if (memcached_instance_response_count(instance) > 0)
758 {
759 return instance;
760 }
761 }
762
763 return NULL;
764 }
765
766 int error= poll(fds, host_index, memc->poll_timeout);
767 switch (error)
768 {
769 case -1:
770 memcached_set_errno(*memc, get_socket_errno(), MEMCACHED_AT);
771 /* FALLTHROUGH */
772 case 0:
773 break;
774
775 default:
776 for (nfds_t x= 0; x < host_index; ++x)
777 {
778 if (fds[x].revents & POLLIN)
779 {
780 for (uint32_t y= 0; y < memcached_server_count(memc); ++y)
781 {
782 org::libmemcached::Instance* instance= memcached_instance_fetch(memc, y);
783
784 if (instance->fd == fds[x].fd)
785 {
786 return instance;
787 }
788 }
789 }
790 }
791 }
792
793 return NULL;
794 }
795
796 /*
797 Eventually we will just kill off the server with the problem.
798 */
799 void memcached_io_reset(org::libmemcached::Instance* ptr)
800 {
801 memcached_quit_server(ptr, true);
802 }
803
804 /**
805 * Read a given number of bytes from the server and place it into a specific
806 * buffer. Reset the IO channel on this server if an error occurs.
807 */
808 memcached_return_t memcached_safe_read(org::libmemcached::Instance* ptr,
809 void *dta,
810 const size_t size)
811 {
812 size_t offset= 0;
813 char *data= static_cast<char *>(dta);
814
815 while (offset < size)
816 {
817 ssize_t nread;
818 memcached_return_t rc;
819
820 while (memcached_continue(rc= memcached_io_read(ptr, data + offset, size - offset, nread))) { };
821
822 if (memcached_failed(rc))
823 {
824 return rc;
825 }
826
827 offset+= size_t(nread);
828 }
829
830 return MEMCACHED_SUCCESS;
831 }
832
833 memcached_return_t memcached_io_readline(org::libmemcached::Instance* ptr,
834 char *buffer_ptr,
835 size_t size,
836 size_t& total_nr)
837 {
838 total_nr= 0;
839 bool line_complete= false;
840
841 while (line_complete == false)
842 {
843 if (ptr->read_buffer_length == 0)
844 {
845 /*
846 * We don't have any data in the buffer, so let's fill the read
847 * buffer. Call the standard read function to avoid duplicating
848 * the logic.
849 */
850 ssize_t nread;
851 memcached_return_t rc= memcached_io_read(ptr, buffer_ptr, 1, nread);
852 if (memcached_failed(rc) and rc == MEMCACHED_IN_PROGRESS)
853 {
854 memcached_quit_server(ptr, true);
855 return memcached_set_error(*ptr, rc, MEMCACHED_AT);
856 }
857 else if (memcached_failed(rc))
858 {
859 return rc;
860 }
861
862 if (*buffer_ptr == '\n')
863 {
864 line_complete= true;
865 }
866
867 ++buffer_ptr;
868 ++total_nr;
869 }
870
871 /* Now let's look in the buffer and copy as we go! */
872 while (ptr->read_buffer_length and total_nr < size and line_complete == false)
873 {
874 *buffer_ptr = *ptr->read_ptr;
875 if (*buffer_ptr == '\n')
876 {
877 line_complete = true;
878 }
879 --ptr->read_buffer_length;
880 ++ptr->read_ptr;
881 ++total_nr;
882 ++buffer_ptr;
883 }
884
885 if (total_nr == size)
886 {
887 return MEMCACHED_PROTOCOL_ERROR;
888 }
889 }
890
891 return MEMCACHED_SUCCESS;
892 }