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