2 Basic socket buffered IO
6 #include "memcached_io.h"
7 #include <sys/select.h>
15 static ssize_t
io_flush(memcached_st
*ptr
, unsigned int server_key
,
16 memcached_return
*error
);
18 static memcached_return
io_wait(memcached_st
*ptr
, unsigned int server_key
,
19 memc_read_or_write read_or_write
)
25 if (read_or_write
== MEM_WRITE
) /* write */
26 flags
= POLLOUT
| POLLERR
;
28 flags
= POLLIN
| POLLERR
;
30 memset(&fds
, 0, sizeof(struct pollfd
));
31 fds
[0].fd
= ptr
->hosts
[server_key
].fd
;
34 error
= poll(fds
, 1, ptr
->poll_timeout
);
37 return MEMCACHED_SUCCESS
;
40 return MEMCACHED_TIMEOUT
;
44 /* Imposssible for anything other then -1 */
45 WATCHPOINT_ASSERT(error
== -1);
46 memcached_quit_server(ptr
, server_key
, 1);
48 return MEMCACHED_FAILURE
;
52 void memcached_io_preread(memcached_st
*ptr
)
58 for (x
= 0; x
< ptr
->number_of_hosts
; x
++)
60 if (memcached_server_response_count(ptr
, x
) &&
61 ptr
->hosts
[x
].read_data_length
< MEMCACHED_MAX_BUFFER
)
65 data_read
= read(ptr
->hosts
[x
].fd
,
66 ptr
->hosts
[x
].read_ptr
+ ptr
->hosts
[x
].read_data_length
,
67 MEMCACHED_MAX_BUFFER
- ptr
->hosts
[x
].read_data_length
);
71 ptr
->hosts
[x
].read_buffer_length
+= data_read
;
72 ptr
->hosts
[x
].read_data_length
+= data_read
;
77 ssize_t
memcached_io_read(memcached_st
*ptr
, unsigned int server_key
,
78 char *buffer
, size_t length
)
87 if (!ptr
->hosts
[server_key
].read_buffer_length
)
93 data_read
= read(ptr
->hosts
[server_key
].fd
,
94 ptr
->hosts
[server_key
].read_buffer
,
95 MEMCACHED_MAX_BUFFER
);
104 rc
= io_wait(ptr
, server_key
, MEM_READ
);
106 if (rc
== MEMCACHED_SUCCESS
)
109 memcached_quit_server(ptr
, server_key
, 1);
114 memcached_quit_server(ptr
, server_key
, 1);
115 ptr
->cached_errno
= errno
;
122 /* If zero, just keep looping unless testing, then assert() */
125 WATCHPOINT_ASSERT(0);
131 ptr
->hosts
[server_key
].read_data_length
= data_read
;
132 ptr
->hosts
[server_key
].read_buffer_length
= data_read
;
133 ptr
->hosts
[server_key
].read_ptr
= ptr
->hosts
[server_key
].read_buffer
;
140 difference
= (length
> ptr
->hosts
[server_key
].read_buffer_length
) ? ptr
->hosts
[server_key
].read_buffer_length
: length
;
142 memcpy(buffer_ptr
, ptr
->hosts
[server_key
].read_ptr
, difference
);
143 length
-= difference
;
144 ptr
->hosts
[server_key
].read_ptr
+= difference
;
145 ptr
->hosts
[server_key
].read_buffer_length
-= difference
;
146 buffer_ptr
+= difference
;
150 *buffer_ptr
= *ptr
->hosts
[server_key
].read_ptr
;
152 ptr
->hosts
[server_key
].read_ptr
++;
153 ptr
->hosts
[server_key
].read_buffer_length
--;
161 return (size_t)(buffer_ptr
- buffer
);
164 ssize_t
memcached_io_write(memcached_st
*ptr
, unsigned int server_key
,
165 char *buffer
, size_t length
, char with_flush
)
167 unsigned long long x
;
169 for (x
= 0; x
< length
; x
++)
171 if (ptr
->hosts
[server_key
].write_ptr
== 0)
172 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
173 WATCHPOINT_ASSERT(ptr
->hosts
[server_key
].write_ptr
);
174 *ptr
->hosts
[server_key
].write_ptr
= buffer
[x
];
175 ptr
->hosts
[server_key
].write_ptr
++;
176 ptr
->hosts
[server_key
].write_buffer_offset
++;
178 if (ptr
->hosts
[server_key
].write_buffer_offset
== MEMCACHED_MAX_BUFFER
)
183 sent_length
= io_flush(ptr
, server_key
, &rc
);
184 if (sent_length
== -1)
187 WATCHPOINT_ASSERT(sent_length
== MEMCACHED_MAX_BUFFER
);
188 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
189 ptr
->hosts
[server_key
].write_buffer_offset
= 0;
196 if (io_flush(ptr
, server_key
, &rc
) == -1)
203 memcached_return
memcached_io_close(memcached_st
*ptr
, unsigned int server_key
)
205 close(ptr
->hosts
[server_key
].fd
);
207 return MEMCACHED_SUCCESS
;
210 static ssize_t
io_flush(memcached_st
*ptr
, unsigned int server_key
,
211 memcached_return
*error
)
214 size_t return_length
;
215 char *write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
216 size_t write_length
= ptr
->hosts
[server_key
].write_buffer_offset
;
218 *error
= MEMCACHED_SUCCESS
;
220 if (ptr
->hosts
[server_key
].write_buffer_offset
== 0)
227 if (ptr
->hosts
[server_key
].type
== MEMCACHED_CONNECTION_UDP
)
229 sent_length
= sendto(ptr
->hosts
[server_key
].fd
,
230 write_ptr
, write_length
, 0,
231 (struct sockaddr
*)&ptr
->hosts
[server_key
].address_info
->ai_addr
,
232 sizeof(struct sockaddr
));
236 if ((ssize_t
)(sent_length
= write(ptr
->hosts
[server_key
].fd
, write_ptr
,
237 write_length
)) == -1)
246 rc
= io_wait(ptr
, server_key
, MEM_WRITE
);
248 if (rc
== MEMCACHED_SUCCESS
)
251 memcached_quit_server(ptr
, server_key
, 1);
255 memcached_quit_server(ptr
, server_key
, 1);
256 ptr
->cached_errno
= errno
;
257 *error
= MEMCACHED_ERRNO
;
263 write_ptr
+= sent_length
;
264 write_length
-= sent_length
;
265 return_length
+= sent_length
;
268 WATCHPOINT_ASSERT(write_length
== 0);
269 WATCHPOINT_ASSERT(return_length
== ptr
->hosts
[server_key
].write_buffer_offset
);
270 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
271 ptr
->hosts
[server_key
].write_buffer_offset
= 0;
273 return return_length
;
277 Eventually we will just kill off the server with the problem.
279 void memcached_io_reset(memcached_st
*ptr
, unsigned int server_key
)
281 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
282 ptr
->hosts
[server_key
].write_buffer_offset
= 0;