2 Basic socket buffered IO
6 #include "memcached_io.h"
7 #include <sys/select.h>
15 static ssize_t
io_flush(memcached_server_st
*ptr
, memcached_return
*error
);
17 static memcached_return
io_wait(memcached_server_st
*ptr
,
18 memc_read_or_write read_or_write
)
24 if (read_or_write
== MEM_WRITE
) /* write */
25 flags
= POLLOUT
| POLLERR
;
27 flags
= POLLIN
| POLLERR
;
29 memset(&fds
, 0, sizeof(struct pollfd
));
33 error
= poll(fds
, 1, ptr
->root
->poll_timeout
);
36 return MEMCACHED_SUCCESS
;
39 return MEMCACHED_TIMEOUT
;
42 /* Imposssible for anything other then -1 */
43 WATCHPOINT_ASSERT(error
== -1);
44 memcached_quit_server(ptr
, 1);
46 return MEMCACHED_FAILURE
;
51 void memcached_io_preread(memcached_st
*ptr
)
57 for (x
= 0; x
< ptr
->number_of_hosts
; x
++)
59 if (memcached_server_response_count(ptr
, x
) &&
60 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
64 data_read
= read(ptr
->hosts
[x
].fd
,
65 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
66 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
);
70 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
71 ptr
->hosts
[x
].read_data_length
+= data_read
;
77 ssize_t
memcached_io_read(memcached_server_st
*ptr
,
78 void *buffer
, size_t length
)
87 if (!ptr
->read_buffer_length
)
93 data_read
= read(ptr
->fd
,
95 MEMCACHED_MAX_BUFFER
);
98 else if (data_read
== -1)
100 ptr
->cached_errno
= errno
;
107 rc
= io_wait(ptr
, MEM_READ
);
109 if (rc
== MEMCACHED_SUCCESS
)
115 memcached_quit_server(ptr
, 1);
127 ptr
->read_data_length
= data_read
;
128 ptr
->read_buffer_length
= data_read
;
129 ptr
->read_ptr
= ptr
->read_buffer
;
136 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
138 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
139 length
-= difference
;
140 ptr
->read_ptr
+= difference
;
141 ptr
->read_buffer_length
-= difference
;
142 buffer_ptr
+= difference
;
146 *buffer_ptr
= *ptr
->read_ptr
;
148 ptr
->read_buffer_length
--;
157 return (size_t)(buffer_ptr
- (char*)buffer
);
160 ssize_t
memcached_io_write(memcached_server_st
*ptr
,
161 const void *buffer
, size_t length
, char with_flush
)
163 size_t original_length
;
164 const char* buffer_ptr
;
166 original_length
= length
;
174 should_write
= MEMCACHED_MAX_BUFFER
- ptr
->write_buffer_offset
;
175 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
177 should_write
= (should_write
< length
) ? should_write
: length
;
179 memcpy(write_ptr
, buffer_ptr
, should_write
);
180 ptr
->write_buffer_offset
+= should_write
;
181 buffer_ptr
+= should_write
;
182 length
-= should_write
;
184 if (ptr
->write_buffer_offset
== MEMCACHED_MAX_BUFFER
)
189 sent_length
= io_flush(ptr
, &rc
);
190 if (sent_length
== -1)
193 WATCHPOINT_ASSERT(sent_length
== MEMCACHED_MAX_BUFFER
);
200 if (io_flush(ptr
, &rc
) == -1)
204 return original_length
;
207 memcached_return
memcached_io_close(memcached_server_st
*ptr
)
210 /* in case of death shutdown to avoid blocking at close() */
212 r
= shutdown(ptr
->fd
, SHUT_RDWR
);
215 if (r
&& errno
!= ENOTCONN
)
217 WATCHPOINT_ERRNO(errno
);
218 WATCHPOINT_ASSERT(errno
);
223 WATCHPOINT_ASSERT(r
== 0);
225 return MEMCACHED_SUCCESS
;
228 static ssize_t
io_flush(memcached_server_st
*ptr
,
229 memcached_return
*error
)
232 size_t return_length
;
233 char *local_write_ptr
= ptr
->write_buffer
;
234 size_t write_length
= ptr
->write_buffer_offset
;
236 *error
= MEMCACHED_SUCCESS
;
238 if (ptr
->write_buffer_offset
== 0)
241 /* Looking for memory overflows */
242 #if defined(HAVE_DEBUG)
243 if (write_length
== MEMCACHED_MAX_BUFFER
)
244 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
245 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
251 WATCHPOINT_ASSERT(write_length
> 0);
253 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
257 ai
= ptr
->address_info
;
259 /* Crappy test code */
260 char buffer
[HUGE_STRING_LEN
+ 8];
261 memset(buffer
, 0, HUGE_STRING_LEN
+ 8);
262 memcpy (buffer
+8, local_write_ptr
, write_length
);
271 sent_length
= sendto(ptr
->fd
, buffer
, write_length
+ 8, 0,
272 (struct sockaddr
*)ai
->ai_addr
,
274 if (sent_length
== -1)
276 WATCHPOINT_ERRNO(errno
);
277 WATCHPOINT_ASSERT(0);
279 sent_length
-= 8; /* We remove the header */
283 if ((sent_length
= write(ptr
->fd
, local_write_ptr
,
284 write_length
)) == -1)
293 rc
= io_wait(ptr
, MEM_WRITE
);
295 if (rc
== MEMCACHED_SUCCESS
)
298 memcached_quit_server(ptr
, 1);
302 memcached_quit_server(ptr
, 1);
303 ptr
->cached_errno
= errno
;
304 *error
= MEMCACHED_ERRNO
;
310 local_write_ptr
+= sent_length
;
311 write_length
-= sent_length
;
312 return_length
+= sent_length
;
315 WATCHPOINT_ASSERT(write_length
== 0);
316 WATCHPOINT_ASSERT(return_length
== ptr
->write_buffer_offset
);
317 ptr
->write_buffer_offset
= 0;
319 return return_length
;
323 Eventually we will just kill off the server with the problem.
325 void memcached_io_reset(memcached_server_st
*ptr
)
327 memcached_quit_server(ptr
, 0);