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 memcached_set_errno(*ptr
, (err
== 0) ? get_socket_errno() : err
, MEMCACHED_AT
);
135 memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
137 memcached_quit_server(ptr
, true);
139 return memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
144 memcached_quit_server(ptr
, true);
146 return memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
149 memcached_return_t
memcached_io_wait_for_write(memcached_server_write_instance_st ptr
)
151 return io_wait(ptr
, MEM_WRITE
);
155 * Try to fill the input buffer for a server with as much
158 * @param ptr the server to pack
160 static bool repack_input_buffer(memcached_server_write_instance_st ptr
)
162 if (ptr
->read_ptr
!= ptr
->read_buffer
)
164 /* Move all of the data to the beginning of the buffer so
165 ** that we can fit more data into the buffer...
167 memmove(ptr
->read_buffer
, ptr
->read_ptr
, ptr
->read_buffer_length
);
168 ptr
->read_ptr
= ptr
->read_buffer
;
169 ptr
->read_data_length
= ptr
->read_buffer_length
;
172 /* There is room in the buffer, try to fill it! */
173 if (ptr
->read_buffer_length
!= MEMCACHED_MAX_BUFFER
)
176 /* Just try a single read to grab what's available */
177 ssize_t nr
= recv(ptr
->fd
,
178 ptr
->read_ptr
+ ptr
->read_data_length
,
179 MEMCACHED_MAX_BUFFER
- ptr
->read_data_length
,
186 switch (get_socket_errno())
195 #ifdef TARGET_OS_LINUX
198 break; // No IO is fine, we can just move on
201 memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
206 case 0: // Shutdown on the socket has occurred
208 memcached_set_error(*ptr
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
);
214 ptr
->read_data_length
+= size_t(nr
);
215 ptr
->read_buffer_length
+= size_t(nr
);
226 * If the we have callbacks connected to this server structure
227 * we may start process the input queue and fire the callbacks
228 * for the incomming messages. This function is _only_ called
229 * when the input buffer is full, so that we _know_ that we have
230 * at least _one_ message to process.
232 * @param ptr the server to star processing iput messages for
233 * @return true if we processed anything, false otherwise
235 static bool process_input_buffer(memcached_server_write_instance_st ptr
)
238 ** We might be able to process some of the response messages if we
239 ** have a callback set up
241 if (ptr
->root
->callbacks
!= NULL
&& ptr
->root
->flags
.use_udp
== false)
244 * We might have responses... try to read them out and fire
247 memcached_callback_st cb
= *ptr
->root
->callbacks
;
249 memcached_set_processing_input((memcached_st
*)ptr
->root
, true);
251 char buffer
[MEMCACHED_DEFAULT_COMMAND_SIZE
];
252 memcached_return_t error
;
253 memcached_st
*root
= (memcached_st
*)ptr
->root
;
254 error
= memcached_response(ptr
, buffer
, sizeof(buffer
),
257 memcached_set_processing_input(root
, false);
259 if (error
== MEMCACHED_SUCCESS
)
261 for (unsigned int x
= 0; x
< cb
.number_of_callback
; x
++)
263 error
= (*cb
.callback
[x
])(ptr
->root
, &root
->result
, cb
.context
);
264 if (error
!= MEMCACHED_SUCCESS
)
268 /* @todo what should I do with the error message??? */
270 /* @todo what should I do with other error messages?? */
277 memcached_return_t
memcached_io_read(memcached_server_write_instance_st ptr
,
278 void *buffer
, size_t length
, ssize_t
*nread
)
280 assert(ptr
); // Programmer error
281 char *buffer_ptr
= static_cast<char *>(buffer
);
283 if (ptr
->fd
== INVALID_SOCKET
)
285 assert(int(ptr
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
));
286 return MEMCACHED_CONNECTION_FAILURE
;
291 if (not ptr
->read_buffer_length
)
296 data_read
= recv(ptr
->fd
, ptr
->read_buffer
, MEMCACHED_MAX_BUFFER
, MSG_DONTWAIT
);
297 if (data_read
== SOCKET_ERROR
)
299 switch (get_socket_errno())
301 case EINTR
: // We just retry
304 case ETIMEDOUT
: // OSX
309 #ifdef TARGET_OS_LINUX
312 if (memcached_success(io_wait(ptr
, MEM_READ
)))
316 return MEMCACHED_IN_PROGRESS
;
320 case ENOTCONN
: // Programmer Error
321 WATCHPOINT_ASSERT(0);
323 WATCHPOINT_ASSERT(0);
325 assert(ptr
->fd
!= INVALID_SOCKET
);
331 memcached_quit_server(ptr
, true);
333 return memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
337 else if (data_read
== 0)
340 EOF. Any data received so far is incomplete
341 so discard it. This always reads by byte in case of TCP
342 and protocol enforcement happens at memcached_response()
343 looking for '\n'. We do not care for UDB which requests 8 bytes
344 at once. Generally, this means that connection went away. Since
345 for blocking I/O we do not return 0 and for non-blocking case
346 it will return EGAIN if data is not immediatly available.
348 WATCHPOINT_STRING("We had a zero length recv()");
350 memcached_quit_server(ptr
, true);
352 return memcached_set_error(*ptr
, MEMCACHED_UNKNOWN_READ_FAILURE
, MEMCACHED_AT
);
354 } while (data_read
<= 0);
356 ptr
->io_bytes_sent
= 0;
357 ptr
->read_data_length
= (size_t) data_read
;
358 ptr
->read_buffer_length
= (size_t) data_read
;
359 ptr
->read_ptr
= ptr
->read_buffer
;
366 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
368 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
369 length
-= difference
;
370 ptr
->read_ptr
+= difference
;
371 ptr
->read_buffer_length
-= difference
;
372 buffer_ptr
+= difference
;
376 *buffer_ptr
= *ptr
->read_ptr
;
378 ptr
->read_buffer_length
--;
384 ptr
->server_failure_counter
= 0;
385 *nread
= (ssize_t
)(buffer_ptr
- (char*)buffer
);
386 return MEMCACHED_SUCCESS
;
389 memcached_return_t
memcached_io_slurp(memcached_server_write_instance_st ptr
)
391 assert(ptr
); // Programmer error
393 if (ptr
->fd
== INVALID_SOCKET
)
395 assert(int(ptr
->state
) <= int(MEMCACHED_SERVER_STATE_ADDRINFO
));
396 return MEMCACHED_CONNECTION_FAILURE
;
400 char buffer
[MEMCACHED_MAX_BUFFER
];
403 data_read
= recv(ptr
->fd
, ptr
->read_buffer
, sizeof(buffer
), MSG_DONTWAIT
);
404 if (data_read
== SOCKET_ERROR
)
406 switch (get_socket_errno())
408 case EINTR
: // We just retry
411 case ETIMEDOUT
: // OSX
416 #ifdef TARGET_OS_LINUX
419 if (memcached_success(io_wait(ptr
, MEM_READ
)))
423 return MEMCACHED_IN_PROGRESS
;
427 case ENOTCONN
: // Programmer Error
428 WATCHPOINT_ASSERT(0);
430 WATCHPOINT_ASSERT(0);
432 assert(ptr
->fd
!= INVALID_SOCKET
);
437 return MEMCACHED_CONNECTION_FAILURE
; // We want this!
440 } while (data_read
> 0);
442 return MEMCACHED_CONNECTION_FAILURE
;
445 static ssize_t
_io_write(memcached_server_write_instance_st ptr
,
446 const void *buffer
, size_t length
, bool with_flush
)
448 size_t original_length
;
449 const char* buffer_ptr
;
451 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
453 original_length
= length
;
454 buffer_ptr
= static_cast<const char *>(buffer
);
462 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
464 //UDP does not support partial writes
465 buffer_end
= MAX_UDP_DATAGRAM_LENGTH
;
466 should_write
= length
;
467 if (ptr
->write_buffer_offset
+ should_write
> buffer_end
)
474 buffer_end
= MEMCACHED_MAX_BUFFER
;
475 should_write
= buffer_end
- ptr
->write_buffer_offset
;
476 should_write
= (should_write
< length
) ? should_write
: length
;
479 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
480 memcpy(write_ptr
, buffer_ptr
, should_write
);
481 ptr
->write_buffer_offset
+= should_write
;
482 buffer_ptr
+= should_write
;
483 length
-= should_write
;
485 if (ptr
->write_buffer_offset
== buffer_end
&& ptr
->type
!= MEMCACHED_CONNECTION_UDP
)
487 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
489 memcached_return_t rc
;
490 ssize_t sent_length
= io_flush(ptr
, with_flush
, &rc
);
491 if (sent_length
== -1)
496 /* If io_flush calls memcached_purge, sent_length may be 0 */
497 unlikely (sent_length
!= 0)
499 WATCHPOINT_ASSERT(sent_length
== (ssize_t
)buffer_end
);
506 memcached_return_t rc
;
507 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
508 if (io_flush(ptr
, with_flush
, &rc
) == -1)
514 return (ssize_t
) original_length
;
517 ssize_t
memcached_io_write(memcached_server_write_instance_st ptr
,
518 const void *buffer
, size_t length
, bool with_flush
)
520 return _io_write(ptr
, buffer
, length
, with_flush
);
523 ssize_t
memcached_io_writev(memcached_server_write_instance_st ptr
,
524 const struct libmemcached_io_vector_st
*vector
,
525 size_t number_of
, bool with_flush
)
529 for (size_t x
= 0; x
< number_of
; x
++, vector
++)
533 if ((returnable
= _io_write(ptr
, vector
->buffer
, vector
->length
, false)) == -1)
542 if (memcached_io_write(ptr
, NULL
, 0, true) == -1)
552 void memcached_io_close(memcached_server_write_instance_st ptr
)
554 if (ptr
->fd
== INVALID_SOCKET
)
559 /* in case of death shutdown to avoid blocking at close() */
560 if (shutdown(ptr
->fd
, SHUT_RDWR
) == SOCKET_ERROR
&& get_socket_errno() != ENOTCONN
)
562 WATCHPOINT_NUMBER(ptr
->fd
);
563 WATCHPOINT_ERRNO(get_socket_errno());
564 WATCHPOINT_ASSERT(get_socket_errno());
567 if (closesocket(ptr
->fd
) == SOCKET_ERROR
)
569 WATCHPOINT_ERRNO(get_socket_errno());
571 ptr
->state
= MEMCACHED_SERVER_STATE_NEW
;
572 ptr
->fd
= INVALID_SOCKET
;
575 memcached_server_write_instance_st
memcached_io_get_readable_server(memcached_st
*memc
)
577 #define MAX_SERVERS_TO_POLL 100
578 struct pollfd fds
[MAX_SERVERS_TO_POLL
];
579 unsigned int host_index
= 0;
581 for (uint32_t x
= 0; x
< memcached_server_count(memc
) && host_index
< MAX_SERVERS_TO_POLL
; ++x
)
583 memcached_server_write_instance_st instance
=
584 memcached_server_instance_fetch(memc
, x
);
586 if (instance
->read_buffer_length
> 0) /* I have data in the buffer */
589 if (memcached_server_response_count(instance
) > 0)
591 fds
[host_index
].events
= POLLIN
;
592 fds
[host_index
].revents
= 0;
593 fds
[host_index
].fd
= instance
->fd
;
600 /* We have 0 or 1 server with pending events.. */
601 for (uint32_t x
= 0; x
< memcached_server_count(memc
); ++x
)
603 memcached_server_write_instance_st instance
=
604 memcached_server_instance_fetch(memc
, x
);
606 if (memcached_server_response_count(instance
) > 0)
615 int error
= poll(fds
, host_index
, memc
->poll_timeout
);
619 memcached_set_errno(*memc
, get_socket_errno(), MEMCACHED_AT
);
625 for (size_t x
= 0; x
< host_index
; ++x
)
627 if (fds
[x
].revents
& POLLIN
)
629 for (uint32_t y
= 0; y
< memcached_server_count(memc
); ++y
)
631 memcached_server_write_instance_st instance
=
632 memcached_server_instance_fetch(memc
, y
);
634 if (instance
->fd
== fds
[x
].fd
)
644 static ssize_t
io_flush(memcached_server_write_instance_st ptr
,
645 const bool with_flush
,
646 memcached_return_t
*error
)
649 ** We might want to purge the input buffer if we haven't consumed
650 ** any output yet... The test for the limits is the purge is inline
651 ** in the purge function to avoid duplicating the logic..
654 memcached_return_t rc
;
655 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
656 rc
= memcached_purge(ptr
);
658 if (rc
!= MEMCACHED_SUCCESS
&& rc
!= MEMCACHED_STORED
)
664 size_t return_length
;
665 char *local_write_ptr
= ptr
->write_buffer
;
666 size_t write_length
= ptr
->write_buffer_offset
;
668 *error
= MEMCACHED_SUCCESS
;
670 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
672 // UDP Sanity check, make sure that we are not sending somthing too big
673 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
&& write_length
> MAX_UDP_DATAGRAM_LENGTH
)
675 *error
= MEMCACHED_WRITE_FAILURE
;
679 if (ptr
->write_buffer_offset
== 0 || (ptr
->type
== MEMCACHED_CONNECTION_UDP
680 && ptr
->write_buffer_offset
== UDP_DATAGRAM_HEADER_LENGTH
))
685 /* Looking for memory overflows */
687 if (write_length
== MEMCACHED_MAX_BUFFER
)
688 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
689 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
695 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
696 WATCHPOINT_ASSERT(write_length
> 0);
698 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
699 increment_udp_message_id(ptr
);
701 WATCHPOINT_ASSERT(ptr
->fd
!= INVALID_SOCKET
);
704 sent_length
= send(ptr
->fd
, local_write_ptr
, write_length
, MSG_NOSIGNAL
|MSG_DONTWAIT
);
708 sent_length
= send(ptr
->fd
, local_write_ptr
, write_length
, MSG_NOSIGNAL
|MSG_DONTWAIT
|MSG_MORE
);
711 if (sent_length
== SOCKET_ERROR
)
713 memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
714 #if 0 // @todo I should look at why we hit this bit of code hard frequently
715 WATCHPOINT_ERRNO(get_socket_errno());
716 WATCHPOINT_NUMBER(get_socket_errno());
718 switch (get_socket_errno())
728 * We may be blocked on write because the input buffer
729 * is full. Let's check if we have room in our input
730 * buffer for more data and retry the write before
733 if (repack_input_buffer(ptr
) or
734 process_input_buffer(ptr
))
739 memcached_return_t rc
= io_wait(ptr
, MEM_WRITE
);
740 if (memcached_success(rc
))
744 else if (rc
== MEMCACHED_TIMEOUT
)
746 *error
= memcached_set_error(*ptr
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
750 memcached_quit_server(ptr
, true);
751 *error
= memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
757 memcached_quit_server(ptr
, true);
758 *error
= memcached_set_errno(*ptr
, get_socket_errno(), MEMCACHED_AT
);
759 WATCHPOINT_ASSERT(ptr
->fd
== -1);
764 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
and
765 (size_t)sent_length
!= write_length
)
767 memcached_quit_server(ptr
, true);
768 *error
= memcached_set_error(*ptr
, MEMCACHED_WRITE_FAILURE
, MEMCACHED_AT
);
772 ptr
->io_bytes_sent
+= (uint32_t) sent_length
;
774 local_write_ptr
+= sent_length
;
775 write_length
-= (uint32_t) sent_length
;
776 return_length
+= (uint32_t) sent_length
;
779 WATCHPOINT_ASSERT(write_length
== 0);
780 // Need to study this assert() WATCHPOINT_ASSERT(return_length ==
781 // ptr->write_buffer_offset);
783 // if we are a udp server, the begining of the buffer is reserverd for
784 // the upd frame header
785 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
786 ptr
->write_buffer_offset
= UDP_DATAGRAM_HEADER_LENGTH
;
788 ptr
->write_buffer_offset
= 0;
790 return (ssize_t
) return_length
;
794 Eventually we will just kill off the server with the problem.
796 void memcached_io_reset(memcached_server_write_instance_st ptr
)
798 memcached_quit_server(ptr
, true);
802 * Read a given number of bytes from the server and place it into a specific
803 * buffer. Reset the IO channel on this server if an error occurs.
805 memcached_return_t
memcached_safe_read(memcached_server_write_instance_st ptr
,
810 char *data
= static_cast<char *>(dta
);
812 while (offset
< size
)
815 memcached_return_t rc
;
817 while (memcached_continue(rc
= memcached_io_read(ptr
, data
+ offset
, size
- offset
, &nread
))) { };
819 if (memcached_failed(rc
))
824 offset
+= (size_t) nread
;
827 return MEMCACHED_SUCCESS
;
830 memcached_return_t
memcached_io_readline(memcached_server_write_instance_st ptr
,
834 bool line_complete
= false;
837 while (not line_complete
)
839 if (ptr
->read_buffer_length
== 0)
842 * We don't have any data in the buffer, so let's fill the read
843 * buffer. Call the standard read function to avoid duplicating
847 memcached_return_t rc
= memcached_io_read(ptr
, buffer_ptr
, 1, &nread
);
848 if (memcached_failed(rc
) and rc
== MEMCACHED_IN_PROGRESS
)
850 memcached_quit_server(ptr
, true);
851 return memcached_set_error(*ptr
, rc
, MEMCACHED_AT
);
853 else if (memcached_failed(rc
))
858 if (*buffer_ptr
== '\n')
865 /* Now let's look in the buffer and copy as we go! */
866 while (ptr
->read_buffer_length
&& total_nr
< size
&& !line_complete
)
868 *buffer_ptr
= *ptr
->read_ptr
;
869 if (*buffer_ptr
== '\n')
870 line_complete
= true;
871 --ptr
->read_buffer_length
;
877 if (total_nr
== size
)
878 return MEMCACHED_PROTOCOL_ERROR
;
881 return MEMCACHED_SUCCESS
;
885 * The udp request id consists of two seperate sections
887 * 2) The message number
888 * The thread id should only be set when the memcached_st struct is created
889 * and should not be changed.
891 * The message num is incremented for each new message we send, this function
892 * extracts the message number from message_id, increments it and then
893 * writes the new value back into the header
895 static void increment_udp_message_id(memcached_server_write_instance_st ptr
)
897 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
898 uint16_t cur_req
= get_udp_datagram_request_id(header
);
899 int msg_num
= get_msg_num_from_request_id(cur_req
);
900 int thread_id
= get_thread_id_from_request_id(cur_req
);
902 if (((++msg_num
) & UDP_REQUEST_ID_THREAD_MASK
) != 0)
905 header
->request_id
= htons((uint16_t) (thread_id
| msg_num
));
908 memcached_return_t
memcached_io_init_udp_header(memcached_server_write_instance_st ptr
, uint16_t thread_id
)
910 if (thread_id
> UDP_REQUEST_ID_MAX_THREAD_ID
)
911 return MEMCACHED_FAILURE
;
913 struct udp_datagram_header_st
*header
= (struct udp_datagram_header_st
*)ptr
->write_buffer
;
914 header
->request_id
= htons((uint16_t) (generate_udp_request_thread_id(thread_id
)));
915 header
->num_datagrams
= htons(1);
916 header
->sequence_number
= htons(0);
918 return MEMCACHED_SUCCESS
;