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