New version update
[m6w6/libmemcached] / libmemcached / memcached_io.c
1 /*
2 Basic socket buffered IO
3 */
4
5 #include "common.h"
6 #include "memcached_io.h"
7 #include <sys/select.h>
8 #include <poll.h>
9
10 typedef enum {
11 MEM_READ,
12 MEM_WRITE,
13 } memc_read_or_write;
14
15 static ssize_t io_flush(memcached_server_st *ptr, memcached_return *error);
16
17 static memcached_return io_wait(memcached_server_st *ptr,
18 memc_read_or_write read_or_write)
19 {
20 struct pollfd fds[1];
21 short flags= 0;
22 int error;
23
24 if (read_or_write == MEM_WRITE) /* write */
25 flags= POLLOUT | POLLERR;
26 else
27 flags= POLLIN | POLLERR;
28
29 memset(&fds, 0, sizeof(struct pollfd));
30 fds[0].fd= ptr->fd;
31 fds[0].events= flags;
32
33 error= poll(fds, 1, ptr->root->poll_timeout);
34
35 if (error == 1)
36 return MEMCACHED_SUCCESS;
37 else if (error == 0)
38 {
39 return MEMCACHED_TIMEOUT;
40 }
41
42 /* Imposssible for anything other then -1 */
43 WATCHPOINT_ASSERT(error == -1);
44 memcached_quit_server(ptr, 1);
45
46 return MEMCACHED_FAILURE;
47
48 }
49
50 #ifdef UNUSED
51 void memcached_io_preread(memcached_st *ptr)
52 {
53 unsigned int x;
54
55 return;
56
57 for (x= 0; x < ptr->number_of_hosts; x++)
58 {
59 if (memcached_server_response_count(ptr, x) &&
60 ptr->hosts[x].read_data_length < MEMCACHED_MAX_BUFFER )
61 {
62 size_t data_read;
63
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);
67 if (data_read == -1)
68 continue;
69
70 ptr->hosts[x].read_buffer_length+= data_read;
71 ptr->hosts[x].read_data_length+= data_read;
72 }
73 }
74 }
75 #endif
76
77 ssize_t memcached_io_read(memcached_server_st *ptr,
78 char *buffer, size_t length)
79 {
80 char *buffer_ptr;
81
82 buffer_ptr= buffer;
83
84 while (length)
85 {
86 uint8_t found_eof= 0;
87 if (!ptr->read_buffer_length)
88 {
89 ssize_t data_read;
90
91 while (1)
92 {
93 data_read= read(ptr->fd,
94 ptr->read_buffer,
95 MEMCACHED_MAX_BUFFER);
96 if (data_read > 0)
97 break;
98 else if (data_read == -1)
99 {
100 ptr->cached_errno= errno;
101 switch (errno)
102 {
103 case EAGAIN:
104 {
105 memcached_return rc;
106
107 rc= io_wait(ptr, MEM_READ);
108
109 if (rc == MEMCACHED_SUCCESS)
110 continue;
111 }
112 /* fall trough */
113 default:
114 {
115 memcached_quit_server(ptr, 1);
116 return -1;
117 }
118 }
119 }
120 else
121 {
122 found_eof= 1;
123 break;
124 }
125 }
126
127 ptr->read_data_length= data_read;
128 ptr->read_buffer_length= data_read;
129 ptr->read_ptr= ptr->read_buffer;
130 }
131
132 if (length > 1)
133 {
134 size_t difference;
135
136 difference= (length > ptr->read_buffer_length) ? ptr->read_buffer_length : length;
137
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;
143 }
144 else
145 {
146 *buffer_ptr= *ptr->read_ptr;
147 ptr->read_ptr++;
148 ptr->read_buffer_length--;
149 buffer_ptr++;
150 break;
151 }
152
153 if (found_eof)
154 break;
155 }
156
157 return (size_t)(buffer_ptr - buffer);
158 }
159
160 ssize_t memcached_io_write(memcached_server_st *ptr,
161 char *buffer, size_t length, char with_flush)
162 {
163 size_t original_length;
164 char* buffer_ptr;
165
166 original_length= length;
167 buffer_ptr= buffer;
168
169 while (length)
170 {
171 char *write_ptr;
172 size_t should_write;
173
174 should_write= MEMCACHED_MAX_BUFFER - ptr->write_buffer_offset;
175 write_ptr= ptr->write_buffer + ptr->write_buffer_offset;
176
177 should_write= (should_write < length) ? should_write : length;
178
179 memcpy(write_ptr, buffer_ptr, should_write);
180 ptr->write_buffer_offset+= should_write;
181 buffer_ptr+= should_write;
182 length-= should_write;
183
184 if (ptr->write_buffer_offset == MEMCACHED_MAX_BUFFER)
185 {
186 memcached_return rc;
187 ssize_t sent_length;
188
189 sent_length= io_flush(ptr, &rc);
190 if (sent_length == -1)
191 return -1;
192
193 WATCHPOINT_ASSERT(sent_length == MEMCACHED_MAX_BUFFER);
194 }
195 }
196
197 if (with_flush)
198 {
199 memcached_return rc;
200 if (io_flush(ptr, &rc) == -1)
201 return -1;
202 }
203
204 return original_length;
205 }
206
207 memcached_return memcached_io_close(memcached_server_st *ptr)
208 {
209 close(ptr->fd);
210
211 return MEMCACHED_SUCCESS;
212 }
213
214 static ssize_t io_flush(memcached_server_st *ptr,
215 memcached_return *error)
216 {
217 size_t sent_length;
218 size_t return_length;
219 char *local_write_ptr= ptr->write_buffer;
220 size_t write_length= ptr->write_buffer_offset;
221
222 *error= MEMCACHED_SUCCESS;
223
224 if (ptr->write_buffer_offset == 0)
225 return 0;
226
227 /* Looking for memory overflows */
228 if (write_length == MEMCACHED_MAX_BUFFER)
229 WATCHPOINT_ASSERT(ptr->write_buffer == local_write_ptr);
230 WATCHPOINT_ASSERT((ptr->write_buffer + MEMCACHED_MAX_BUFFER) >= (local_write_ptr + write_length));
231
232 return_length= 0;
233 while (write_length)
234 {
235 WATCHPOINT_ASSERT(write_length > 0);
236 sent_length= 0;
237 if (ptr->type == MEMCACHED_CONNECTION_UDP)
238 {
239 struct addrinfo *ai;
240
241 ai= ptr->address_info;
242
243 /* Crappy test code */
244 char buffer[HUGE_STRING_LEN + 8];
245 memset(buffer, 0, HUGE_STRING_LEN + 8);
246 memcpy (buffer+8, local_write_ptr, write_length);
247 buffer[0]= 0;
248 buffer[1]= 0;
249 buffer[2]= 0;
250 buffer[3]= 0;
251 buffer[4]= 0;
252 buffer[5]= 1;
253 buffer[6]= 0;
254 buffer[7]= 0;
255 sent_length= sendto(ptr->fd, buffer, write_length + 8, 0,
256 (struct sockaddr *)ai->ai_addr,
257 ai->ai_addrlen);
258 if (sent_length == -1)
259 {
260 WATCHPOINT_ERRNO(errno);
261 WATCHPOINT_ASSERT(0);
262 }
263 sent_length-= 8; /* We remove the header */
264 }
265 else
266 {
267 if ((ssize_t)(sent_length= write(ptr->fd, local_write_ptr,
268 write_length)) == -1)
269 {
270 switch (errno)
271 {
272 case ENOBUFS:
273 continue;
274 case EAGAIN:
275 {
276 memcached_return rc;
277 rc= io_wait(ptr, MEM_WRITE);
278
279 if (rc == MEMCACHED_SUCCESS)
280 continue;
281
282 memcached_quit_server(ptr, 1);
283 return -1;
284 }
285 default:
286 memcached_quit_server(ptr, 1);
287 ptr->cached_errno= errno;
288 *error= MEMCACHED_ERRNO;
289 return -1;
290 }
291 }
292 }
293
294 local_write_ptr+= sent_length;
295 write_length-= sent_length;
296 return_length+= sent_length;
297 }
298
299 WATCHPOINT_ASSERT(write_length == 0);
300 WATCHPOINT_ASSERT(return_length == ptr->write_buffer_offset);
301 ptr->write_buffer_offset= 0;
302
303 return return_length;
304 }
305
306 /*
307 Eventually we will just kill off the server with the problem.
308 */
309 void memcached_io_reset(memcached_server_st *ptr)
310 {
311 memcached_quit_server(ptr, 0);
312 }