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
));
34 ** We are going to block on write, but at least on Solaris we might block
35 ** on write if we haven't read anything from our input buffer..
36 ** Try to purge the input buffer if we don't do any flow control in the
37 ** application layer (just sending a lot of data etc)
38 ** The test is moved down in the purge function to avoid duplication of
41 if (read_or_write
== MEM_WRITE
)
44 error
= poll(fds
, 1, ptr
->root
->poll_timeout
);
47 return MEMCACHED_SUCCESS
;
50 return MEMCACHED_TIMEOUT
;
53 /* Imposssible for anything other then -1 */
54 WATCHPOINT_ASSERT(error
== -1);
55 memcached_quit_server(ptr
, 1);
57 return MEMCACHED_FAILURE
;
62 void memcached_io_preread(memcached_st
*ptr
)
68 for (x
= 0; x
< ptr
->number_of_hosts
; x
++)
70 if (memcached_server_response_count(ptr
, x
) &&
71 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
75 data_read
= read(ptr
->hosts
[x
].fd
,
76 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
77 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
);
81 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
82 ptr
->hosts
[x
].read_data_length
+= data_read
;
88 ssize_t
memcached_io_read(memcached_server_st
*ptr
,
89 void *buffer
, size_t length
)
98 if (!ptr
->read_buffer_length
)
104 data_read
= read(ptr
->fd
,
106 MEMCACHED_MAX_BUFFER
);
109 else if (data_read
== -1)
111 ptr
->cached_errno
= errno
;
118 rc
= io_wait(ptr
, MEM_READ
);
120 if (rc
== MEMCACHED_SUCCESS
)
126 memcached_quit_server(ptr
, 1);
138 ptr
->io_bytes_sent
= 0;
139 ptr
->read_data_length
= data_read
;
140 ptr
->read_buffer_length
= data_read
;
141 ptr
->read_ptr
= ptr
->read_buffer
;
148 difference
= (length
> ptr
->read_buffer_length
) ? ptr
->read_buffer_length
: length
;
150 memcpy(buffer_ptr
, ptr
->read_ptr
, difference
);
151 length
-= difference
;
152 ptr
->read_ptr
+= difference
;
153 ptr
->read_buffer_length
-= difference
;
154 buffer_ptr
+= difference
;
158 *buffer_ptr
= *ptr
->read_ptr
;
160 ptr
->read_buffer_length
--;
169 return (size_t)(buffer_ptr
- (char*)buffer
);
172 ssize_t
memcached_io_write(memcached_server_st
*ptr
,
173 const void *buffer
, size_t length
, char with_flush
)
175 size_t original_length
;
176 const char* buffer_ptr
;
178 original_length
= length
;
186 should_write
= MEMCACHED_MAX_BUFFER
- ptr
->write_buffer_offset
;
187 write_ptr
= ptr
->write_buffer
+ ptr
->write_buffer_offset
;
189 should_write
= (should_write
< length
) ? should_write
: length
;
191 memcpy(write_ptr
, buffer_ptr
, should_write
);
192 ptr
->write_buffer_offset
+= should_write
;
193 buffer_ptr
+= should_write
;
194 length
-= should_write
;
196 if (ptr
->write_buffer_offset
== MEMCACHED_MAX_BUFFER
)
201 sent_length
= io_flush(ptr
, &rc
);
202 if (sent_length
== -1)
205 WATCHPOINT_ASSERT(sent_length
== MEMCACHED_MAX_BUFFER
);
212 if (io_flush(ptr
, &rc
) == -1)
216 return original_length
;
219 memcached_return
memcached_io_close(memcached_server_st
*ptr
)
222 /* in case of death shutdown to avoid blocking at close() */
224 r
= shutdown(ptr
->fd
, SHUT_RDWR
);
227 if (r
&& errno
!= ENOTCONN
)
229 WATCHPOINT_ERRNO(errno
);
230 WATCHPOINT_ASSERT(errno
);
235 WATCHPOINT_ASSERT(r
== 0);
237 return MEMCACHED_SUCCESS
;
240 static ssize_t
io_flush(memcached_server_st
*ptr
,
241 memcached_return
*error
)
244 size_t return_length
;
245 char *local_write_ptr
= ptr
->write_buffer
;
246 size_t write_length
= ptr
->write_buffer_offset
;
248 *error
= MEMCACHED_SUCCESS
;
250 if (ptr
->write_buffer_offset
== 0)
253 /* Looking for memory overflows */
254 #if defined(HAVE_DEBUG)
255 if (write_length
== MEMCACHED_MAX_BUFFER
)
256 WATCHPOINT_ASSERT(ptr
->write_buffer
== local_write_ptr
);
257 WATCHPOINT_ASSERT((ptr
->write_buffer
+ MEMCACHED_MAX_BUFFER
) >= (local_write_ptr
+ write_length
));
263 WATCHPOINT_ASSERT(write_length
> 0);
265 if (ptr
->type
== MEMCACHED_CONNECTION_UDP
)
269 ai
= ptr
->address_info
;
271 /* Crappy test code */
272 char buffer
[HUGE_STRING_LEN
+ 8];
273 memset(buffer
, 0, HUGE_STRING_LEN
+ 8);
274 memcpy (buffer
+8, local_write_ptr
, write_length
);
283 sent_length
= sendto(ptr
->fd
, buffer
, write_length
+ 8, 0,
284 (struct sockaddr
*)ai
->ai_addr
,
286 if (sent_length
== -1)
288 WATCHPOINT_ERRNO(errno
);
289 WATCHPOINT_ASSERT(0);
291 sent_length
-= 8; /* We remove the header */
296 ** We might want to purge the input buffer if we haven't consumed
297 ** any output yet... The test for the limits is the purge is inline
298 ** in the purge function to avoid duplicating the logic..
300 memcached_purge(ptr
);
302 if ((sent_length
= write(ptr
->fd
, local_write_ptr
,
303 write_length
)) == -1)
312 rc
= io_wait(ptr
, MEM_WRITE
);
314 if (rc
== MEMCACHED_SUCCESS
)
317 memcached_quit_server(ptr
, 1);
321 memcached_quit_server(ptr
, 1);
322 ptr
->cached_errno
= errno
;
323 *error
= MEMCACHED_ERRNO
;
329 ptr
->io_bytes_sent
+= sent_length
;
331 local_write_ptr
+= sent_length
;
332 write_length
-= sent_length
;
333 return_length
+= sent_length
;
336 WATCHPOINT_ASSERT(write_length
== 0);
337 WATCHPOINT_ASSERT(return_length
== ptr
->write_buffer_offset
);
338 ptr
->write_buffer_offset
= 0;
340 return return_length
;
344 Eventually we will just kill off the server with the problem.
346 void memcached_io_reset(memcached_server_st
*ptr
)
348 memcached_quit_server(ptr
, 0);