1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2009 Brian Aker
9 * Redistribution and use in source and binary forms, with or without
10 * modification, are permitted provided that the following conditions are
13 * * Redistributions of source code must retain the above copyright
14 * notice, this list of conditions and the following disclaimer.
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
21 * * The names of its contributors may not be used to endorse or
22 * promote products derived from this software without specific prior
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.
40 #include <libmemcached/common.h>
43 enum memc_read_or_write
{
48 static ssize_t
io_flush(memcached_server_write_instance_st ptr
,
49 const bool with_flush
,
50 memcached_return_t
*error
);
51 static void increment_udp_message_id(memcached_server_write_instance_st ptr
);
53 static memcached_return_t
io_wait(memcached_server_write_instance_st ptr
,
54 memc_read_or_write read_or_write
)
60 if (read_or_write
== MEM_WRITE
) /* write */
63 WATCHPOINT_SET(ptr
->io_wait_count
.write
++);
67 WATCHPOINT_SET(ptr
->io_wait_count
.read
++);
71 ** We are going to block on write, but at least on Solaris we might block
72 ** on write if we haven't read anything from our input buffer..
73 ** Try to purge the input buffer if we don't do any flow control in the
74 ** application layer (just sending a lot of data etc)
75 ** The test is moved down in the purge function to avoid duplication of
78 if (read_or_write
== MEM_WRITE
)
80 memcached_return_t rc
= memcached_purge(ptr
);
81 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
83 return MEMCACHED_FAILURE
;
87 if (ptr
->root
->poll_timeout
== 0) // Mimic 0 causes timeout behavior (not all platforms do this)
89 return memcached_set_error(*ptr
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
93 while (--loop_max
) // While loop is for ERESTART or EINTR
96 int error
= poll(&fds
, 1, ptr
->root
->poll_timeout
);
100 WATCHPOINT_IF_LABELED_NUMBER(read_or_write
&& loop_max
< 4, "read() times we had to loop, decremented down from 5", loop_max
);
101 WATCHPOINT_IF_LABELED_NUMBER(!read_or_write
&& loop_max
< 4, "write() times we had to loop, decremented down from 5", loop_max
);
103 return MEMCACHED_SUCCESS
;
105 case 0: // Timeout occured, we let the while() loop do its thing.
106 return memcached_set_error(*ptr
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
109 WATCHPOINT_ERRNO(get_socket_errno());
110 switch (get_socket_errno())
112 #ifdef TARGET_OS_LINUX
120 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
123 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"));
126 if (fds
.revents
& POLLERR
)
129 socklen_t len
= sizeof (err
);
130 (void)getsockopt(ptr
->fd
, SOL_SOCKET
, SO_ERROR
, &err
, &len
);
131 ptr
->cached_errno
= (err
== 0) ? get_socket_errno() : err
;
135 ptr
->cached_errno
= get_socket_errno();
137 memcached_quit_server(ptr
, true);
139 return memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
144 ptr
->cached_errno
= get_socket_errno();
145 memcached_quit_server(ptr
, true);
147 return memcached_set_error(*ptr
, MEMCACHED_FAILURE
, MEMCACHED_AT
);
150 memcached_return_t
memcached_io_wait_for_write(memcached_server_write_instance_st ptr
)
152 return io_wait(ptr
, MEM_WRITE
);
156 * Try to fill the input buffer for a server with as much
159 * @param ptr the server to pack
161 static bool repack_input_buffer(memcached_server_write_instance_st ptr
)
163 if (ptr
->read_ptr
!= ptr
->read_buffer
)
165 /* Move all of the data to the beginning of the buffer so
166 ** that we can fit more data into the buffer...
168 memmove(ptr
->read_buffer
, ptr
->read_ptr
, ptr
->read_buffer_length
);
169 ptr
->read_ptr
= ptr
->read_buffer
;
170 ptr
->read_data_length
= ptr
->read_buffer_length
;
173 /* There is room in the buffer, try to fill it! */
174 if (ptr
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
177 /* Just try a single read to grab what's available */
178 ssize_t nr
= recv(ptr
->fd
,
179 ptr
->read_ptr
+ ptr
->read_data_length
,
180 MEMCACHED_MAX_BUFFER
- ptr
->read_data_length
,
187 switch (get_socket_errno())
196 #ifdef TARGET_OS_LINUX
199 break; // No IO is fine, we can just move on
202 memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
207 case 0: // Shutdown on the socket has occurred
209 memcached_set_error(*ptr
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
);
215 ptr
->read_data_length
+= size_t(nr
);
216 ptr
->read_buffer_length
+= size_t(nr
);
227 * If the we have callbacks connected to this server structure
228 * we may start process the input queue and fire the callbacks
229 * for the incomming messages. This function is _only_ called
230 * when the input buffer is full, so that we _know_ that we have
231 * at least _one_ message to process.
233 * @param ptr the server to star processing iput messages for
234 * @return true if we processed anything, false otherwise
236 static bool process_input_buffer(memcached_server_write_instance_st ptr
)
239 ** We might be able to process some of the response messages if we
240 ** have a callback set up
242 if (ptr
->root
->callbacks
!= NULL
&& ptr
->root
->flags
.use_udp
== false)
245 * We might have responses... try to read them out and fire
248 memcached_callback_st cb
= *ptr
->root
->callbacks
;
250 memcached_set_processing_input((memcached_st
*)ptr
->root
, true);
252 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
253 memcached_return_t error
;
254 memcached_st
*root
= (memcached_st
*)ptr
->root
;
255 error
= memcached_response(ptr
, buffer
, sizeof(buffer
),
258 memcached_set_processing_input(root
, false);
260 if (error
== MEMCACHED_SUCCESS
)
262 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
264 error
= (*cb
.callback
[x
])(ptr
->root
, &root
->result
, cb
.context
);
265 if (error
!= MEMCACHED_SUCCESS
)
269 /* @todo what should I do with the error message??? */
271 /* @todo what should I do with other error messages?? */
278 memcached_return_t
memcached_io_read(memcached_server_write_instance_st ptr
,
279 void *buffer
, size_t length
, ssize_t
*nread
)
281 assert(ptr
); // Programmer error
282 char *buffer_ptr
= static_cast<char *>(buffer
);
284 if (ptr
->fd
== INVALID_SOCKET
)
286 assert(int(ptr
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
));
287 return MEMCACHED_CONNECTION_FAILURE
;
292 if (not ptr
->read_buffer_length
)
297 data_read
= recv(ptr
->fd
, ptr
->read_buffer
, MEMCACHED_MAX_BUFFER
, MSG_DONTWAIT
);
298 if (data_read
== SOCKET_ERROR
)
300 switch (get_socket_errno())
302 case EINTR
: // We just retry
305 case ETIMEDOUT
: // OSX
310 #ifdef TARGET_OS_LINUX
313 if (memcached_success(io_wait(ptr
, MEM_READ
)))
317 return MEMCACHED_IN_PROGRESS
;
321 case ENOTCONN
: // Programmer Error
322 WATCHPOINT_ASSERT(0);
324 WATCHPOINT_ASSERT(0);
326 assert(ptr
->fd
!= INVALID_SOCKET
);
332 memcached_quit_server(ptr
, true);
334 return memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
338 else if (data_read
== 0)
341 EOF. Any data received so far is incomplete
342 so discard it. This always reads by byte in case of TCP
343 and protocol enforcement happens at memcached_response()
344 looking for '\n'. We do not care for UDB which requests 8 bytes
345 at once. Generally, this means that connection went away. Since
346 for blocking I/O we do not return 0 and for non-blocking case
347 it will return EGAIN if data is not immediatly available.
349 WATCHPOINT_STRING("We had a zero length recv()");
351 memcached_quit_server(ptr
, true);
353 return memcached_set_error(*ptr
, MEMCACHED_UNKNOWN_READ_FAILURE
, MEMCACHED_AT
);
355 } while (data_read
<= 0);
357 ptr
->io_bytes_sent
= 0;
358 ptr
->read_data_length
= (size_t) data_read
;
359 ptr
->read_buffer_length
= (size_t) data_read
;
360 ptr
->read_ptr
= ptr
->read_buffer
;
367 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
369 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
370 length
-= difference
;
371 ptr
->read_ptr
+= difference
;
372 ptr
->read_buffer_length
-= difference
;
373 buffer_ptr
+= difference
;
377 *buffer_ptr
= *ptr
->read_ptr
;
379 ptr
->read_buffer_length
--;
385 ptr
->server_failure_counter
= 0;
386 *nread
= (ssize_t
)(buffer_ptr
- (char*)buffer
);
387 return MEMCACHED_SUCCESS
;
390 memcached_return_t
memcached_io_slurp(memcached_server_write_instance_st ptr
)
392 assert(ptr
); // Programmer error
394 if (ptr
->fd
== INVALID_SOCKET
)
396 assert(int(ptr
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
));
397 return MEMCACHED_CONNECTION_FAILURE
;
401 char buffer
[MEMCACHED_MAX_BUFFER
];
404 data_read
= recv(ptr
->fd
, ptr
->read_buffer
, sizeof(buffer
), MSG_DONTWAIT
);
405 if (data_read
== SOCKET_ERROR
)
407 switch (get_socket_errno())
409 case EINTR
: // We just retry
412 case ETIMEDOUT
: // OSX
417 #ifdef TARGET_OS_LINUX
420 if (memcached_success(io_wait(ptr
, MEM_READ
)))
424 return MEMCACHED_IN_PROGRESS
;
428 case ENOTCONN
: // Programmer Error
429 WATCHPOINT_ASSERT(0);
431 WATCHPOINT_ASSERT(0);
433 assert(ptr
->fd
!= INVALID_SOCKET
);
438 return MEMCACHED_CONNECTION_FAILURE
; // We want this!
441 } while (data_read
> 0);
443 return MEMCACHED_CONNECTION_FAILURE
;
446 static ssize_t
_io_write(memcached_server_write_instance_st ptr
,
447 const void *buffer
, size_t length
, bool with_flush
)
449 size_t original_length
;
450 const char* buffer_ptr
;
452 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
454 original_length
= length
;
455 buffer_ptr
= static_cast<const char *>(buffer
);
463 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
465 //UDP does not support partial writes
466 buffer_end
= MAX_UDP_DATAGRAM_LENGTH
;
467 should_write
= length
;
468 if (ptr
->write_buffer_offset
+ should_write
> buffer_end
)
475 buffer_end
= MEMCACHED_MAX_BUFFER
;
476 should_write
= buffer_end
- ptr
->write_buffer_offset
;
477 should_write
= (should_write
< length
) ? should_write
: length
;
480 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
481 memcpy(write_ptr
, buffer_ptr
, should_write
);
482 ptr
->write_buffer_offset
+= should_write
;
483 buffer_ptr
+= should_write
;
484 length
-= should_write
;
486 if (ptr
->write_buffer_offset
== buffer_end
&& ptr
->type
!= MEMCACHED_CONNECTION_UDP
)
488 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
490 memcached_return_t rc
;
491 ssize_t sent_length
= io_flush(ptr
, with_flush
, &rc
);
492 if (sent_length
== -1)
497 /* If io_flush calls memcached_purge, sent_length may be 0 */
498 unlikely (sent_length
!= 0)
500 WATCHPOINT_ASSERT(sent_length
== (ssize_t
)buffer_end
);
507 memcached_return_t rc
;
508 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
509 if (io_flush(ptr
, with_flush
, &rc
) == -1)
515 return (ssize_t
) original_length
;
518 ssize_t
memcached_io_write(memcached_server_write_instance_st ptr
,
519 const void *buffer
, size_t length
, bool with_flush
)
521 return _io_write(ptr
, buffer
, length
, with_flush
);
524 ssize_t
memcached_io_writev(memcached_server_write_instance_st ptr
,
525 const struct libmemcached_io_vector_st
*vector
,
526 size_t number_of
, bool with_flush
)
530 for (size_t x
= 0; x
< number_of
; x
++, vector
++)
534 if ((returnable
= _io_write(ptr
, vector
->buffer
, vector
->length
, false)) == -1)
543 if (memcached_io_write(ptr
, NULL
, 0, true) == -1)
553 void memcached_io_close(memcached_server_write_instance_st ptr
)
555 if (ptr
->fd
== INVALID_SOCKET
)
560 /* in case of death shutdown to avoid blocking at close() */
561 if (shutdown(ptr
->fd
, SHUT_RDWR
) == SOCKET_ERROR
&& get_socket_errno() != ENOTCONN
)
563 WATCHPOINT_NUMBER(ptr
->fd
);
564 WATCHPOINT_ERRNO(get_socket_errno());
565 WATCHPOINT_ASSERT(get_socket_errno());
568 if (closesocket(ptr
->fd
) == SOCKET_ERROR
)
570 WATCHPOINT_ERRNO(get_socket_errno());
572 ptr
->state
= MEMCACHED_SERVER_STATE_NEW
;
573 ptr
->fd
= INVALID_SOCKET
;
576 memcached_server_write_instance_st
memcached_io_get_readable_server(memcached_st
*memc
)
578 #define MAX_SERVERS_TO_POLL 100
579 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
580 unsigned int host_index
= 0;
582 for (uint32_t x
= 0; x
< memcached_server_count(memc
) && host_index
< MAX_SERVERS_TO_POLL
; ++x
)
584 memcached_server_write_instance_st instance
=
585 memcached_server_instance_fetch(memc
, x
);
587 if (instance
->read_buffer_length
> 0) /* I have data in the buffer */
590 if (memcached_server_response_count(instance
) > 0)
592 fds
[host_index
].events
= POLLIN
;
593 fds
[host_index
].revents
= 0;
594 fds
[host_index
].fd
= instance
->fd
;
601 /* We have 0 or 1 server with pending events.. */
602 for (uint32_t x
= 0; x
< memcached_server_count(memc
); ++x
)
604 memcached_server_write_instance_st instance
=
605 memcached_server_instance_fetch(memc
, x
);
607 if (memcached_server_response_count(instance
) > 0)
616 int error
= poll(fds
, host_index
, memc
->poll_timeout
);
620 memcached_set_errno(*memc
, get_socket_errno(), MEMCACHED_AT
);
626 for (size_t x
= 0; x
< host_index
; ++x
)
628 if (fds
[x
].revents
& POLLIN
)
630 for (uint32_t y
= 0; y
< memcached_server_count(memc
); ++y
)
632 memcached_server_write_instance_st instance
=
633 memcached_server_instance_fetch(memc
, y
);
635 if (instance
->fd
== fds
[x
].fd
)
645 static ssize_t
io_flush(memcached_server_write_instance_st ptr
,
646 const bool with_flush
,
647 memcached_return_t
*error
)
650 ** We might want to purge the input buffer if we haven't consumed
651 ** any output yet... The test for the limits is the purge is inline
652 ** in the purge function to avoid duplicating the logic..
655 memcached_return_t rc
;
656 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
657 rc
= memcached_purge(ptr
);
659 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
665 size_t return_length
;
666 char *local_write_ptr
= ptr
->write_buffer
;
667 size_t write_length
= ptr
->write_buffer_offset
;
669 *error
= MEMCACHED_SUCCESS
;
671 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
673 // UDP Sanity check, make sure that we are not sending somthing too big
674 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& write_length
> MAX_UDP_DATAGRAM_LENGTH
)
676 *error
= MEMCACHED_WRITE_FAILURE
;
680 if (ptr
->write_buffer_offset
== 0 || (ptr
->type
== MEMCACHED_CONNECTION_UDP
681 && ptr
->write_buffer_offset
== UDP_DATAGRAM_HEADER_LENGTH
))
686 /* Looking for memory overflows */
688 if (write_length
== MEMCACHED_MAX_BUFFER
)
689 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
690 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
696 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
697 WATCHPOINT_ASSERT(write_length
> 0);
699 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
700 increment_udp_message_id(ptr
);
702 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
705 sent_length
= send(ptr
->fd
, local_write_ptr
, write_length
, MSG_NOSIGNAL
|MSG_DONTWAIT
);
709 sent_length
= send(ptr
->fd
, local_write_ptr
, write_length
, MSG_NOSIGNAL
|MSG_DONTWAIT
|MSG_MORE
);
712 if (sent_length
== SOCKET_ERROR
)
714 ptr
->cached_errno
= get_socket_errno();
715 #if 0 // @todo I should look at why we hit this bit of code hard frequently
716 WATCHPOINT_ERRNO(get_socket_errno());
717 WATCHPOINT_NUMBER(get_socket_errno());
719 switch (get_socket_errno())
729 * We may be blocked on write because the input buffer
730 * is full. Let's check if we have room in our input
731 * buffer for more data and retry the write before
734 if (repack_input_buffer(ptr
) or
735 process_input_buffer(ptr
))
740 memcached_return_t rc
= io_wait(ptr
, MEM_WRITE
);
741 if (memcached_success(rc
))
745 else if (rc
== MEMCACHED_TIMEOUT
)
747 *error
= memcached_set_error(*ptr
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
751 memcached_quit_server(ptr
, true);
752 *error
= memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
758 memcached_quit_server(ptr
, true);
759 *error
= memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
760 WATCHPOINT_ASSERT(ptr
->fd
== -1);
765 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
and
766 (size_t)sent_length
!= write_length
)
768 memcached_quit_server(ptr
, true);
769 *error
= memcached_set_error(*ptr
, MEMCACHED_WRITE_FAILURE
, MEMCACHED_AT
);
773 ptr
->io_bytes_sent
+= (uint32_t) sent_length
;
775 local_write_ptr
+= sent_length
;
776 write_length
-= (uint32_t) sent_length
;
777 return_length
+= (uint32_t) sent_length
;
780 WATCHPOINT_ASSERT(write_length
== 0);
781 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
782 // ptr->write_buffer_offset);
784 // if we are a udp server, the begining of the buffer is reserverd for
785 // the upd frame header
786 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
787 ptr
->write_buffer_offset
= UDP_DATAGRAM_HEADER_LENGTH
;
789 ptr
->write_buffer_offset
= 0;
791 return (ssize_t
) return_length
;
795 Eventually we will just kill off the server with the problem.
797 void memcached_io_reset(memcached_server_write_instance_st ptr
)
799 memcached_quit_server(ptr
, true);
803 * Read a given number of bytes from the server and place it into a specific
804 * buffer. Reset the IO channel on this server if an error occurs.
806 memcached_return_t
memcached_safe_read(memcached_server_write_instance_st ptr
,
811 char *data
= static_cast<char *>(dta
);
813 while (offset
< size
)
816 memcached_return_t rc
;
818 while (memcached_continue(rc
= memcached_io_read(ptr
, data
+ offset
, size
- offset
, &nread
))) { };
820 if (memcached_failed(rc
))
825 offset
+= (size_t) nread
;
828 return MEMCACHED_SUCCESS
;
831 memcached_return_t
memcached_io_readline(memcached_server_write_instance_st ptr
,
835 bool line_complete
= false;
838 while (not line_complete
)
840 if (ptr
->read_buffer_length
== 0)
843 * We don't have any data in the buffer, so let's fill the read
844 * buffer. Call the standard read function to avoid duplicating
848 memcached_return_t rc
= memcached_io_read(ptr
, buffer_ptr
, 1, &nread
);
849 if (memcached_failed(rc
) and rc
== MEMCACHED_IN_PROGRESS
)
851 memcached_quit_server(ptr
, true);
852 return memcached_set_error(*ptr
, rc
, MEMCACHED_AT
);
854 else if (memcached_failed(rc
))
859 if (*buffer_ptr
== '\n')
866 /* Now let's look in the buffer and copy as we go! */
867 while (ptr
->read_buffer_length
&& total_nr
< size
&& !line_complete
)
869 *buffer_ptr
= *ptr
->read_ptr
;
870 if (*buffer_ptr
== '\n')
871 line_complete
= true;
872 --ptr
->read_buffer_length
;
878 if (total_nr
== size
)
879 return MEMCACHED_PROTOCOL_ERROR
;
882 return MEMCACHED_SUCCESS
;
886 * The udp request id consists of two seperate sections
888 * 2) The message number
889 * The thread id should only be set when the memcached_st struct is created
890 * and should not be changed.
892 * The message num is incremented for each new message we send, this function
893 * extracts the message number from message_id, increments it and then
894 * writes the new value back into the header
896 static void increment_udp_message_id(memcached_server_write_instance_st ptr
)
898 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
899 uint16_t cur_req
= get_udp_datagram_request_id(header
);
900 int msg_num
= get_msg_num_from_request_id(cur_req
);
901 int thread_id
= get_thread_id_from_request_id(cur_req
);
903 if (((++msg_num
) & UDP_REQUEST_ID_THREAD_MASK
) != 0)
906 header
->request_id
= htons((uint16_t) (thread_id
| msg_num
));
909 memcached_return_t
memcached_io_init_udp_header(memcached_server_write_instance_st ptr
, uint16_t thread_id
)
911 if (thread_id
> UDP_REQUEST_ID_MAX_THREAD_ID
)
912 return MEMCACHED_FAILURE
;
914 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
915 header
->request_id
= htons((uint16_t) (generate_udp_request_thread_id(thread_id
)));
916 header
->num_datagrams
= htons(1);
917 header
->sequence_number
= htons(0);
919 return MEMCACHED_SUCCESS
;