Merge working tree with build tree.
[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 if (memcached_fatal(memcached_purge(ptr)))
196 {
197 return MEMCACHED_FAILURE;
198 }
199 }
200
201 if (ptr->root->poll_timeout == 0) // Mimic 0 causes timeout behavior (not all platforms do this)
202 {
203 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
204 }
205
206 size_t loop_max= 5;
207 while (--loop_max) // While loop is for ERESTART or EINTR
208 {
209
210 int error= poll(&fds, 1, ptr->root->poll_timeout);
211 switch (error)
212 {
213 case 1: // Success!
214 WATCHPOINT_IF_LABELED_NUMBER(read_or_write && loop_max < 4, "read() times we had to loop, decremented down from 5", loop_max);
215 WATCHPOINT_IF_LABELED_NUMBER(!read_or_write && loop_max < 4, "write() times we had to loop, decremented down from 5", loop_max);
216
217 return MEMCACHED_SUCCESS;
218
219 case 0: // Timeout occured, we let the while() loop do its thing.
220 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
221
222 default:
223 WATCHPOINT_ERRNO(get_socket_errno());
224 switch (get_socket_errno())
225 {
226 #ifdef TARGET_OS_LINUX
227 case ERESTART:
228 #endif
229 case EINTR:
230 break;
231
232 case EFAULT:
233 case ENOMEM:
234 return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
235
236 case EINVAL:
237 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"));
238
239 default:
240 if (fds.revents & POLLERR)
241 {
242 int err;
243 socklen_t len= sizeof (err);
244 if (getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0)
245 {
246 if (err == 0)
247 {
248 continue;
249 }
250 errno= err;
251 }
252 }
253 else
254 {
255 memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
256 }
257 int local_errno= get_socket_errno(); // We cache in case memcached_quit_server() modifies errno
258 memcached_quit_server(ptr, true);
259
260 return memcached_set_errno(*ptr, local_errno, MEMCACHED_AT);
261 }
262 }
263 }
264
265 memcached_quit_server(ptr, true);
266
267 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
268 }
269
270 static bool io_flush(memcached_server_write_instance_st ptr,
271 const bool with_flush,
272 memcached_return_t& error)
273 {
274 /*
275 ** We might want to purge the input buffer if we haven't consumed
276 ** any output yet... The test for the limits is the purge is inline
277 ** in the purge function to avoid duplicating the logic..
278 */
279 {
280 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
281 memcached_return_t rc= memcached_purge(ptr);
282
283 if (rc != MEMCACHED_SUCCESS and rc != MEMCACHED_STORED)
284 {
285 return false;
286 }
287 }
288 char *local_write_ptr= ptr->write_buffer;
289 size_t write_length= ptr->write_buffer_offset;
290
291 error= MEMCACHED_SUCCESS;
292
293 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
294
295 /* Looking for memory overflows */
296 #if defined(DEBUG)
297 if (write_length == MEMCACHED_MAX_BUFFER)
298 WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
299 WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
300 #endif
301
302 while (write_length)
303 {
304 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
305 WATCHPOINT_ASSERT(write_length > 0);
306
307 ssize_t sent_length= 0;
308 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
309 if (with_flush)
310 {
311 sent_length= ::send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT);
312 }
313 else
314 {
315 sent_length= ::send(ptr->fd, local_write_ptr, write_length, MSG_NOSIGNAL|MSG_DONTWAIT|MSG_MORE);
316 }
317
318 if (sent_length == SOCKET_ERROR)
319 {
320 #if 0 // @todo I should look at why we hit this bit of code hard frequently
321 WATCHPOINT_ERRNO(get_socket_errno());
322 WATCHPOINT_NUMBER(get_socket_errno());
323 #endif
324 switch (get_socket_errno())
325 {
326 case ENOBUFS:
327 continue;
328
329 #if EWOULDBLOCK != EAGAIN
330 case EWOULDBLOCK:
331 #endif
332 case EAGAIN:
333 {
334 /*
335 * We may be blocked on write because the input buffer
336 * is full. Let's check if we have room in our input
337 * buffer for more data and retry the write before
338 * waiting..
339 */
340 if (repack_input_buffer(ptr) or process_input_buffer(ptr))
341 {
342 continue;
343 }
344
345 memcached_return_t rc= io_wait(ptr, MEM_WRITE);
346 if (memcached_success(rc))
347 {
348 continue;
349 }
350 else if (rc == MEMCACHED_TIMEOUT)
351 {
352 error= memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
353 return false;
354 }
355
356 memcached_quit_server(ptr, true);
357 error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
358 return false;
359 }
360 case ENOTCONN:
361 case EPIPE:
362 default:
363 memcached_quit_server(ptr, true);
364 error= memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
365 WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
366 return false;
367 }
368 }
369
370 ptr->io_bytes_sent+= uint32_t(sent_length);
371
372 local_write_ptr+= sent_length;
373 write_length-= uint32_t(sent_length);
374 }
375
376 WATCHPOINT_ASSERT(write_length == 0);
377 ptr->write_buffer_offset= 0;
378
379 return true;
380 }
381
382 memcached_return_t memcached_io_wait_for_write(memcached_server_write_instance_st ptr)
383 {
384 return io_wait(ptr, MEM_WRITE);
385 }
386
387 memcached_return_t memcached_io_read(memcached_server_write_instance_st ptr,
388 void *buffer, size_t length, ssize_t& nread)
389 {
390 assert(memcached_is_udp(ptr->root) == false);
391 assert_msg(ptr, "Programmer error, memcached_io_read() recieved an invalid memcached_server_write_instance_st"); // Programmer error
392 char *buffer_ptr= static_cast<char *>(buffer);
393
394 if (ptr->fd == INVALID_SOCKET)
395 {
396 #if 0
397 assert_msg(int(ptr->state) <= int(MEMCACHED_SERVER_STATE_ADDRINFO), "Programmer error, invalid socket state");
398 #endif
399 return MEMCACHED_CONNECTION_FAILURE;
400 }
401
402 while (length)
403 {
404 if (not ptr->read_buffer_length)
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 if (memcached_success(io_wait(ptr, MEM_READ)))
426 {
427 continue;
428 }
429 return MEMCACHED_IN_PROGRESS;
430
431 /* fall through */
432
433 case ENOTCONN: // Programmer Error
434 WATCHPOINT_ASSERT(0);
435 case ENOTSOCK:
436 WATCHPOINT_ASSERT(0);
437 case EBADF:
438 assert_msg(ptr->fd != INVALID_SOCKET, "Programmer error, invalid socket");
439 case EINVAL:
440 case EFAULT:
441 case ECONNREFUSED:
442 default:
443 {
444 memcached_quit_server(ptr, true);
445 nread= -1;
446 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
447 }
448 }
449 }
450 else if (data_read == 0)
451 {
452 /*
453 EOF. Any data received so far is incomplete
454 so discard it. This always reads by byte in case of TCP
455 and protocol enforcement happens at memcached_response()
456 looking for '\n'. We do not care for UDB which requests 8 bytes
457 at once. Generally, this means that connection went away. Since
458 for blocking I/O we do not return 0 and for non-blocking case
459 it will return EGAIN if data is not immediatly available.
460 */
461 WATCHPOINT_STRING("We had a zero length recv()");
462 memcached_quit_server(ptr, true);
463 nread= -1;
464 return memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT,
465 memcached_literal_param("::rec() returned zero, server has disconnected"));
466 }
467 } while (data_read <= 0);
468
469 ptr->io_bytes_sent = 0;
470 ptr->read_data_length= (size_t) data_read;
471 ptr->read_buffer_length= (size_t) data_read;
472 ptr->read_ptr= ptr->read_buffer;
473 }
474
475 if (length > 1)
476 {
477 size_t difference;
478
479 difference= (length > ptr->read_buffer_length) ? ptr->read_buffer_length : length;
480
481 memcpy(buffer_ptr, ptr->read_ptr, difference);
482 length -= difference;
483 ptr->read_ptr+= difference;
484 ptr->read_buffer_length-= difference;
485 buffer_ptr+= difference;
486 }
487 else
488 {
489 *buffer_ptr= *ptr->read_ptr;
490 ptr->read_ptr++;
491 ptr->read_buffer_length--;
492 buffer_ptr++;
493 break;
494 }
495 }
496
497 nread= ssize_t(buffer_ptr - (char*)buffer);
498
499 return MEMCACHED_SUCCESS;
500 }
501
502 memcached_return_t memcached_io_slurp(memcached_server_write_instance_st ptr)
503 {
504 assert_msg(ptr, "Programmer error, invalid memcached_server_write_instance_st");
505 assert(memcached_is_udp(ptr->root) == false);
506
507 if (ptr->fd == INVALID_SOCKET)
508 {
509 assert_msg(int(ptr->state) <= int(MEMCACHED_SERVER_STATE_ADDRINFO), "Invalid socket state");
510 return MEMCACHED_CONNECTION_FAILURE;
511 }
512
513 ssize_t data_read;
514 char buffer[MEMCACHED_MAX_BUFFER];
515 do
516 {
517 data_read= recv(ptr->fd, ptr->read_buffer, sizeof(buffer), MSG_DONTWAIT);
518 if (data_read == SOCKET_ERROR)
519 {
520 switch (get_socket_errno())
521 {
522 case EINTR: // We just retry
523 continue;
524
525 case ETIMEDOUT: // OSX
526 #if EWOULDBLOCK != EAGAIN
527 case EWOULDBLOCK:
528 #endif
529 case EAGAIN:
530 #ifdef TARGET_OS_LINUX
531 case ERESTART:
532 #endif
533 if (memcached_success(io_wait(ptr, MEM_READ)))
534 {
535 continue;
536 }
537 return MEMCACHED_IN_PROGRESS;
538
539 /* fall through */
540
541 case ENOTCONN: // Programmer Error
542 WATCHPOINT_ASSERT(0);
543 case ENOTSOCK:
544 WATCHPOINT_ASSERT(0);
545 case EBADF:
546 assert_msg(ptr->fd != INVALID_SOCKET, "Invalid socket state");
547 case EINVAL:
548 case EFAULT:
549 case ECONNREFUSED:
550 default:
551 return MEMCACHED_CONNECTION_FAILURE; // We want this!
552 }
553 }
554 } while (data_read > 0);
555
556 return MEMCACHED_CONNECTION_FAILURE;
557 }
558
559 static ssize_t _io_write(memcached_server_write_instance_st ptr,
560 const void *buffer, size_t length, bool with_flush)
561 {
562 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
563 assert(memcached_is_udp(ptr->root) == false);
564
565 size_t original_length= length;
566 const char *buffer_ptr= static_cast<const char *>(buffer);
567
568 while (length)
569 {
570 char *write_ptr;
571 size_t buffer_end= MEMCACHED_MAX_BUFFER;
572 size_t should_write= buffer_end -ptr->write_buffer_offset;
573 should_write= (should_write < length) ? should_write : length;
574
575 write_ptr= ptr->write_buffer + ptr->write_buffer_offset;
576 memcpy(write_ptr, buffer_ptr, should_write);
577 ptr->write_buffer_offset+= should_write;
578 buffer_ptr+= should_write;
579 length-= should_write;
580
581 if (ptr->write_buffer_offset == buffer_end)
582 {
583 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
584
585 memcached_return_t rc;
586 if (io_flush(ptr, with_flush, rc) == false)
587 {
588 return -1;
589 }
590 }
591 }
592
593 if (with_flush)
594 {
595 memcached_return_t rc;
596 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
597 if (io_flush(ptr, with_flush, rc) == false)
598 {
599 return -1;
600 }
601 }
602
603 return (ssize_t) original_length;
604 }
605
606 bool memcached_io_write(memcached_server_write_instance_st ptr)
607 {
608 return (_io_write(ptr, NULL, 0, true) >= 0);
609 }
610
611 ssize_t memcached_io_write(memcached_server_write_instance_st ptr,
612 const void *buffer, const size_t length, const bool with_flush)
613 {
614 return _io_write(ptr, buffer, length, with_flush);
615 }
616
617 ssize_t memcached_io_writev(memcached_server_write_instance_st ptr,
618 libmemcached_io_vector_st vector[],
619 const size_t number_of, const bool with_flush)
620 {
621 ssize_t total= 0;
622
623 for (size_t x= 0; x < number_of; x++, vector++)
624 {
625 ssize_t returnable;
626
627 if (vector->length)
628 {
629 if ((returnable= _io_write(ptr, vector->buffer, vector->length, false)) == -1)
630 {
631 return -1;
632 }
633 total+= returnable;
634 }
635 }
636
637 if (with_flush)
638 {
639 if (memcached_io_write(ptr) == false)
640 {
641 return -1;
642 }
643 }
644
645 return total;
646 }
647
648
649 void memcached_io_close(memcached_server_write_instance_st ptr)
650 {
651 if (ptr->fd == INVALID_SOCKET)
652 {
653 return;
654 }
655
656 /* in case of death shutdown to avoid blocking at close() */
657 if (shutdown(ptr->fd, SHUT_RDWR) == SOCKET_ERROR && get_socket_errno() != ENOTCONN)
658 {
659 WATCHPOINT_NUMBER(ptr->fd);
660 WATCHPOINT_ERRNO(get_socket_errno());
661 WATCHPOINT_ASSERT(get_socket_errno());
662 }
663
664 if (closesocket(ptr->fd) == SOCKET_ERROR)
665 {
666 WATCHPOINT_ERRNO(get_socket_errno());
667 }
668 ptr->state= MEMCACHED_SERVER_STATE_NEW;
669 ptr->fd= INVALID_SOCKET;
670 }
671
672 memcached_server_write_instance_st memcached_io_get_readable_server(memcached_st *memc)
673 {
674 #define MAX_SERVERS_TO_POLL 100
675 struct pollfd fds[MAX_SERVERS_TO_POLL];
676 nfds_t host_index= 0;
677
678 for (uint32_t x= 0; x < memcached_server_count(memc) and host_index < MAX_SERVERS_TO_POLL; ++x)
679 {
680 memcached_server_write_instance_st instance= memcached_server_instance_fetch(memc, x);
681
682 if (instance->read_buffer_length > 0) /* I have data in the buffer */
683 {
684 return instance;
685 }
686
687 if (memcached_server_response_count(instance) > 0)
688 {
689 fds[host_index].events = POLLIN;
690 fds[host_index].revents = 0;
691 fds[host_index].fd = instance->fd;
692 ++host_index;
693 }
694 }
695
696 if (host_index < 2)
697 {
698 /* We have 0 or 1 server with pending events.. */
699 for (uint32_t x= 0; x< memcached_server_count(memc); ++x)
700 {
701 memcached_server_write_instance_st instance=
702 memcached_server_instance_fetch(memc, x);
703
704 if (memcached_server_response_count(instance) > 0)
705 {
706 return instance;
707 }
708 }
709
710 return NULL;
711 }
712
713 int error= poll(fds, host_index, memc->poll_timeout);
714 switch (error)
715 {
716 case -1:
717 memcached_set_errno(*memc, get_socket_errno(), MEMCACHED_AT);
718 /* FALLTHROUGH */
719 case 0:
720 break;
721
722 default:
723 for (nfds_t x= 0; x < host_index; ++x)
724 {
725 if (fds[x].revents & POLLIN)
726 {
727 for (uint32_t y= 0; y < memcached_server_count(memc); ++y)
728 {
729 memcached_server_write_instance_st instance= memcached_server_instance_fetch(memc, y);
730
731 if (instance->fd == fds[x].fd)
732 {
733 return instance;
734 }
735 }
736 }
737 }
738 }
739
740 return NULL;
741 }
742
743 /*
744 Eventually we will just kill off the server with the problem.
745 */
746 void memcached_io_reset(memcached_server_write_instance_st ptr)
747 {
748 memcached_quit_server(ptr, true);
749 }
750
751 /**
752 * Read a given number of bytes from the server and place it into a specific
753 * buffer. Reset the IO channel on this server if an error occurs.
754 */
755 memcached_return_t memcached_safe_read(memcached_server_write_instance_st ptr,
756 void *dta,
757 const size_t size)
758 {
759 size_t offset= 0;
760 char *data= static_cast<char *>(dta);
761
762 while (offset < size)
763 {
764 ssize_t nread;
765 memcached_return_t rc;
766
767 while (memcached_continue(rc= memcached_io_read(ptr, data + offset, size - offset, nread))) { };
768
769 if (memcached_failed(rc))
770 {
771 return rc;
772 }
773
774 offset+= size_t(nread);
775 }
776
777 return MEMCACHED_SUCCESS;
778 }
779
780 memcached_return_t memcached_io_readline(memcached_server_write_instance_st ptr,
781 char *buffer_ptr,
782 size_t size,
783 size_t& total_nr)
784 {
785 total_nr= 0;
786 bool line_complete= false;
787
788 while (line_complete == false)
789 {
790 if (ptr->read_buffer_length == 0)
791 {
792 /*
793 * We don't have any data in the buffer, so let's fill the read
794 * buffer. Call the standard read function to avoid duplicating
795 * the logic.
796 */
797 ssize_t nread;
798 memcached_return_t rc= memcached_io_read(ptr, buffer_ptr, 1, nread);
799 if (memcached_failed(rc) and rc == MEMCACHED_IN_PROGRESS)
800 {
801 memcached_quit_server(ptr, true);
802 return memcached_set_error(*ptr, rc, MEMCACHED_AT);
803 }
804 else if (memcached_failed(rc))
805 {
806 return rc;
807 }
808
809 if (*buffer_ptr == '\n')
810 {
811 line_complete= true;
812 }
813
814 ++buffer_ptr;
815 ++total_nr;
816 }
817
818 /* Now let's look in the buffer and copy as we go! */
819 while (ptr->read_buffer_length && total_nr < size && !line_complete)
820 {
821 *buffer_ptr = *ptr->read_ptr;
822 if (*buffer_ptr == '\n')
823 {
824 line_complete = true;
825 }
826 --ptr->read_buffer_length;
827 ++ptr->read_ptr;
828 ++total_nr;
829 ++buffer_ptr;
830 }
831
832 if (total_nr == size)
833 {
834 return MEMCACHED_PROTOCOL_ERROR;
835 }
836 }
837
838 return MEMCACHED_SUCCESS;
839 }