2 Basic socket buffered IO
6 #include "memcached_io.h"
7 #include <sys/select.h>
10 static int io_wait(memcached_st
*ptr
, unsigned int server_key
, unsigned read_or_write
)
18 flags
= POLLOUT
| POLLERR
;
20 flags
= POLLIN
| POLLERR
;
22 memset(&fds
, 0, sizeof(struct pollfd
));
23 fds
[0].fd
= ptr
->hosts
[server_key
].fd
;
28 error
= poll(fds
, 1, ptr
->poll_timeout
);
31 return MEMCACHED_SUCCESS
;
34 memcached_quit_server(ptr
, server_key
, 1);
35 return MEMCACHED_FAILURE
;
39 /* This is impossible */
41 return MEMCACHED_FAILURE
;
47 memcached_quit_server(ptr
, server_key
, 1);
49 return MEMCACHED_FAILURE
; /* Timeout occurred */
52 ssize_t
memcached_io_read(memcached_st
*ptr
, unsigned int server_key
,
53 char *buffer
, size_t length
)
61 if (!ptr
->hosts
[server_key
].read_buffer_length
)
67 if (ptr
->flags
& MEM_NO_BLOCK
)
71 rc
= io_wait(ptr
, server_key
, 0);
72 if (rc
!= MEMCACHED_SUCCESS
)
76 data_read
= read(ptr
->hosts
[server_key
].fd
,
77 ptr
->hosts
[server_key
].read_buffer
,
78 MEMCACHED_MAX_BUFFER
);
87 memcached_quit_server(ptr
, server_key
, 1);
88 ptr
->cached_errno
= errno
;
95 /* If zero, just keep looping */
98 ptr
->hosts
[server_key
].read_buffer_length
= data_read
;
99 ptr
->hosts
[server_key
].read_ptr
= ptr
->hosts
[server_key
].read_buffer
;
102 *buffer_ptr
= *ptr
->hosts
[server_key
].read_ptr
;
104 ptr
->hosts
[server_key
].read_ptr
++;
105 ptr
->hosts
[server_key
].read_buffer_length
--;
109 return (size_t)(buffer_ptr
- buffer
);
112 ssize_t
memcached_io_write(memcached_st
*ptr
, unsigned int server_key
,
113 char *buffer
, size_t length
, char with_flush
)
115 unsigned long long x
;
117 for (x
= 0; x
< length
; x
++)
119 if (ptr
->hosts
[server_key
].write_ptr
== 0)
120 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
121 WATCHPOINT_ASSERT(ptr
->hosts
[server_key
].write_ptr
);
122 *ptr
->hosts
[server_key
].write_ptr
= buffer
[x
];
123 ptr
->hosts
[server_key
].write_ptr
++;
124 ptr
->hosts
[server_key
].write_buffer_offset
++;
126 if (ptr
->hosts
[server_key
].write_buffer_offset
== MEMCACHED_MAX_BUFFER
)
130 sent_length
= memcached_io_flush(ptr
, server_key
);
131 if (sent_length
== -1)
134 WATCHPOINT_ASSERT(sent_length
== MEMCACHED_MAX_BUFFER
);
135 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
136 ptr
->hosts
[server_key
].write_buffer_offset
= 0;
142 if (memcached_io_flush(ptr
, server_key
) == -1)
149 memcached_return
memcached_io_close(memcached_st
*ptr
, unsigned int server_key
)
153 rc
= MEMCACHED_SUCCESS
;
154 if (ptr
->flags
& MEM_NO_BLOCK
)
157 struct pollfd fds
[1];
160 flags
= POLLHUP
| POLLERR
;
162 memset(&fds
, 0, sizeof(struct pollfd
));
163 fds
[0].fd
= ptr
->hosts
[server_key
].fd
;
164 fds
[0].events
= flags
;
167 error
= poll(fds
, 1, ptr
->poll_timeout
== -1 ? 100 : ptr
->poll_timeout
);
171 memcached_quit_server(ptr
, server_key
, 1);
172 return MEMCACHED_FAILURE
;
175 return MEMCACHED_FAILURE
; /* Timeout occurred */
178 close(ptr
->hosts
[server_key
].fd
);
183 ssize_t
memcached_io_flush(memcached_st
*ptr
, unsigned int server_key
)
186 size_t return_length
;
187 char *write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
188 size_t write_length
= ptr
->hosts
[server_key
].write_buffer_offset
;
189 unsigned int loop
= 1;
191 if (ptr
->hosts
[server_key
].write_buffer_offset
== 0)
197 if (ptr
->flags
& MEM_NO_BLOCK
)
201 rc
= io_wait(ptr
, server_key
, 1);
202 if (rc
!= MEMCACHED_SUCCESS
)
207 if (ptr
->hosts
[server_key
].type
== MEMCACHED_CONNECTION_UDP
)
209 sent_length
= sendto(ptr
->hosts
[server_key
].fd
,
210 write_ptr
, write_length
, 0,
211 (struct sockaddr
*)&ptr
->hosts
[server_key
].address_info
->ai_addr
,
212 sizeof(struct sockaddr
));
216 if ((ssize_t
)(sent_length
= write(ptr
->hosts
[server_key
].fd
, write_ptr
,
217 write_length
)) == -1)
230 /* Yes, we want to fall through */
232 memcached_quit_server(ptr
, server_key
, 1);
233 ptr
->cached_errno
= errno
;
239 write_ptr
+= sent_length
;
240 write_length
-= sent_length
;
241 return_length
+= sent_length
;
244 WATCHPOINT_ASSERT(write_length
== 0);
245 WATCHPOINT_ASSERT(return_length
== ptr
->hosts
[server_key
].write_buffer_offset
);
246 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
247 ptr
->hosts
[server_key
].write_buffer_offset
= 0;
249 return return_length
;
253 Eventually we will just kill off the server with the problem.
255 void memcached_io_reset(memcached_st
*ptr
, unsigned int server_key
)
257 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
258 ptr
->hosts
[server_key
].write_buffer_offset
= 0;