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