0657dfd65da7f2e2a3e5bfb2df116596178627e3
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
)
14 struct timespec timer
;
19 flags
= POLLOUT
| POLLERR
;
21 flags
= POLLIN
| POLLERR
;
23 memset(&fds
, 0, sizeof(struct pollfd
));
24 fds
[0].fd
= ptr
->hosts
[server_key
].fd
;
27 if (poll(fds
, 1, ptr
->poll_timeout
) < 0)
28 return MEMCACHED_FAILURE
;
30 return MEMCACHED_SUCCESS
;
33 ssize_t
memcached_io_read(memcached_st
*ptr
, unsigned int server_key
,
34 char *buffer
, size_t length
)
42 if (!ptr
->hosts
[server_key
].read_buffer_length
)
48 if (ptr
->flags
& MEM_NO_BLOCK
)
52 rc
= io_wait(ptr
, server_key
, 0);
53 if (rc
!= MEMCACHED_SUCCESS
)
57 data_read
= read(ptr
->hosts
[server_key
].fd
,
58 ptr
->hosts
[server_key
].read_buffer
,
59 MEMCACHED_MAX_BUFFER
);
68 ptr
->cached_errno
= errno
;
75 /* If zero, just keep looping */
78 ptr
->hosts
[server_key
].read_buffer_length
= data_read
;
79 ptr
->hosts
[server_key
].read_ptr
= ptr
->hosts
[server_key
].read_buffer
;
82 *buffer_ptr
= *ptr
->hosts
[server_key
].read_ptr
;
84 ptr
->hosts
[server_key
].read_ptr
++;
85 ptr
->hosts
[server_key
].read_buffer_length
--;
89 return (size_t)(buffer_ptr
- buffer
);
92 ssize_t
memcached_io_write(memcached_st
*ptr
, unsigned int server_key
,
93 char *buffer
, size_t length
, char with_flush
)
97 for (x
= 0; x
< length
; x
++)
99 if (ptr
->hosts
[server_key
].write_ptr
== 0)
100 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
101 WATCHPOINT_ASSERT(ptr
->hosts
[server_key
].write_ptr
);
102 *ptr
->hosts
[server_key
].write_ptr
= buffer
[x
];
103 ptr
->hosts
[server_key
].write_ptr
++;
104 ptr
->hosts
[server_key
].write_buffer_offset
++;
106 if (ptr
->hosts
[server_key
].write_buffer_offset
== MEMCACHED_MAX_BUFFER
)
110 sent_length
= memcached_io_flush(ptr
, server_key
);
112 WATCHPOINT_ASSERT(sent_length
== MEMCACHED_MAX_BUFFER
);
113 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
114 ptr
->hosts
[server_key
].write_buffer_offset
= 0;
120 if (memcached_io_flush(ptr
, server_key
) == -1)
127 memcached_return
memcached_io_close(memcached_st
*ptr
, unsigned int server_key
)
129 struct pollfd fds
[1];
131 struct timespec timer
;
136 flags
= POLLHUP
| POLLERR
;
138 memset(&fds
, 0, sizeof(struct pollfd
));
139 fds
[0].fd
= ptr
->hosts
[server_key
].fd
;
140 fds
[0].events
= flags
;
141 fds
[0].revents
= flags
;
143 if (poll(fds
, 1, ptr
->poll_timeout
) < 0)
144 rc
= MEMCACHED_FAILURE
;
146 rc
= MEMCACHED_SUCCESS
;
148 close(ptr
->hosts
[server_key
].fd
);
153 ssize_t
memcached_io_flush(memcached_st
*ptr
, unsigned int server_key
)
156 size_t return_length
;
157 char *write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
158 size_t write_length
= ptr
->hosts
[server_key
].write_buffer_offset
;
159 unsigned int loop
= 1;
161 if (ptr
->hosts
[server_key
].write_buffer_offset
== 0)
167 if (ptr
->flags
& MEM_NO_BLOCK
)
171 rc
= io_wait(ptr
, server_key
, 1);
172 if (rc
!= MEMCACHED_SUCCESS
)
177 if (ptr
->hosts
[server_key
].type
== MEMCACHED_CONNECTION_UDP
)
179 sent_length
= sendto(ptr
->hosts
[server_key
].fd
,
180 write_ptr
, write_length
, 0,
181 (struct sockaddr
*)&ptr
->hosts
[server_key
].address_info
->ai_addr
,
182 sizeof(struct sockaddr
));
186 if ((ssize_t
)(sent_length
= write(ptr
->hosts
[server_key
].fd
, write_ptr
,
187 write_length
)) == -1)
200 /* Yes, we want to fall through */
202 ptr
->cached_errno
= errno
;
208 write_ptr
+= sent_length
;
209 write_length
-= sent_length
;
210 return_length
+= sent_length
;
213 WATCHPOINT_ASSERT(write_length
== 0);
214 WATCHPOINT_ASSERT(return_length
== ptr
->hosts
[server_key
].write_buffer_offset
);
215 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
216 ptr
->hosts
[server_key
].write_buffer_offset
= 0;
218 return return_length
;
222 Eventually we will just kill off the server with the problem.
224 void memcached_io_reset(memcached_st
*ptr
, unsigned int server_key
)
226 ptr
->hosts
[server_key
].write_ptr
= ptr
->hosts
[server_key
].write_buffer
;
227 ptr
->hosts
[server_key
].write_buffer_offset
= 0;