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