Updating for 1.0.2 release
[awesomized/libmemcached] / libmemcachedprotocol / handler.c
1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
2 *
3 * Libmemcached library
4 *
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 *
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions are
9 * met:
10 *
11 * * Redistributions of source code must retain the above copyright
12 * notice, this list of conditions and the following disclaimer.
13 *
14 * * Redistributions in binary form must reproduce the above
15 * copyright notice, this list of conditions and the following disclaimer
16 * in the documentation and/or other materials provided with the
17 * distribution.
18 *
19 * * The names of its contributors may not be used to endorse or
20 * promote products derived from this software without specific prior
21 * written permission.
22 *
23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
24 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
25 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
26 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
27 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
28 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
29 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
30 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
31 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
32 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
33 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
34 *
35 */
36
37 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
38 #include <libmemcachedprotocol/common.h>
39
40 #include <stdlib.h>
41 #include <sys/types.h>
42 #include <errno.h>
43 #include <stdbool.h>
44 #include <string.h>
45 #include <strings.h>
46 #include <ctype.h>
47 #include <stdio.h>
48
49 /*
50 ** **********************************************************************
51 ** INTERNAL INTERFACE
52 ** **********************************************************************
53 */
54
55 /**
56 * The default function to receive data from the client. This function
57 * just wraps the recv function to receive from a socket.
58 * See man -s3socket recv for more information.
59 *
60 * @param cookie cookie indentifying a client, not used
61 * @param sock socket to read from
62 * @param buf the destination buffer
63 * @param nbytes the number of bytes to read
64 * @return the number of bytes transferred of -1 upon error
65 */
66 static ssize_t default_recv(const void *cookie,
67 memcached_socket_t sock,
68 void *buf,
69 size_t nbytes)
70 {
71 (void)cookie;
72 return recv(sock, buf, nbytes, 0);
73 }
74
75 /**
76 * The default function to send data to the server. This function
77 * just wraps the send function to send through a socket.
78 * See man -s3socket send for more information.
79 *
80 * @param cookie cookie indentifying a client, not used
81 * @param sock socket to send to
82 * @param buf the source buffer
83 * @param nbytes the number of bytes to send
84 * @return the number of bytes transferred of -1 upon error
85 */
86 static ssize_t default_send(const void *cookie,
87 memcached_socket_t fd,
88 const void *buf,
89 size_t nbytes)
90 {
91 (void)cookie;
92 return send(fd, buf, nbytes, 0);
93 }
94
95 /**
96 * Try to drain the output buffers without blocking
97 *
98 * @param client the client to drain
99 * @return false if an error occured (connection should be shut down)
100 * true otherwise (please note that there may be more data to
101 * left in the buffer to send)
102 */
103 static bool drain_output(struct memcached_protocol_client_st *client)
104 {
105 ssize_t len;
106
107 /* Do we have pending data to send? */
108 while (client->output != NULL)
109 {
110 len= client->root->send(client,
111 client->sock,
112 client->output->data + client->output->offset,
113 client->output->nbytes - client->output->offset);
114
115 if (len == -1)
116 {
117 if (get_socket_errno() == EWOULDBLOCK)
118 {
119 return true;
120 }
121 else if (get_socket_errno() != EINTR)
122 {
123 client->error= get_socket_errno();
124 return false;
125 }
126 }
127 else
128 {
129 client->output->offset += (size_t)len;
130 if (client->output->offset == client->output->nbytes)
131 {
132 /* This was the complete buffer */
133 struct chunk_st *old= client->output;
134 client->output= client->output->next;
135 if (client->output == NULL)
136 {
137 client->output_tail= NULL;
138 }
139 cache_free(client->root->buffer_cache, old);
140 }
141 }
142 }
143
144 return true;
145 }
146
147 /**
148 * Allocate an output buffer and chain it into the output list
149 *
150 * @param client the client that needs the buffer
151 * @return pointer to the new chunk if the allocation succeeds, NULL otherwise
152 */
153 static struct chunk_st *allocate_output_chunk(struct memcached_protocol_client_st *client)
154 {
155 struct chunk_st *ret= cache_alloc(client->root->buffer_cache);
156
157 if (ret == NULL)
158 {
159 return NULL;
160 }
161
162 ret->offset= ret->nbytes= 0;
163 ret->next= NULL;
164 ret->size= CHUNK_BUFFERSIZE;
165 ret->data= (void*)(ret + 1);
166 if (client->output == NULL)
167 {
168 client->output= client->output_tail= ret;
169 }
170 else
171 {
172 client->output_tail->next= ret;
173 client->output_tail= ret;
174 }
175
176 return ret;
177 }
178
179 /**
180 * Spool data into the send-buffer for a client.
181 *
182 * @param client the client to spool the data for
183 * @param data the data to spool
184 * @param length the number of bytes of data to spool
185 * @return PROTOCOL_BINARY_RESPONSE_SUCCESS if success,
186 * PROTOCOL_BINARY_RESPONSE_ENOMEM if we failed to allocate memory
187 */
188 static protocol_binary_response_status spool_output(struct memcached_protocol_client_st *client,
189 const void *data,
190 size_t length)
191 {
192 if (client->mute)
193 {
194 return PROTOCOL_BINARY_RESPONSE_SUCCESS;
195 }
196
197 size_t offset= 0;
198
199 struct chunk_st *chunk= client->output;
200 while (offset < length)
201 {
202 if (chunk == NULL || (chunk->size - chunk->nbytes) == 0)
203 {
204 if ((chunk= allocate_output_chunk(client)) == NULL)
205 {
206 return PROTOCOL_BINARY_RESPONSE_ENOMEM;
207 }
208 }
209
210 size_t bulk= length - offset;
211 if (bulk > chunk->size - chunk->nbytes)
212 {
213 bulk= chunk->size - chunk->nbytes;
214 }
215
216 memcpy(chunk->data + chunk->nbytes, data, bulk);
217 chunk->nbytes += bulk;
218 offset += bulk;
219 }
220
221 return PROTOCOL_BINARY_RESPONSE_SUCCESS;
222 }
223
224 /**
225 * Try to determine the protocol used on this connection.
226 * If the first byte contains the magic byte PROTOCOL_BINARY_REQ we should
227 * be using the binary protocol on the connection. I implemented the support
228 * for the ASCII protocol by wrapping into the simple interface (aka v1),
229 * so the implementors needs to provide an implementation of that interface
230 *
231 */
232 static memcached_protocol_event_t determine_protocol(struct memcached_protocol_client_st *client, ssize_t *length, void **endptr)
233 {
234 if (*client->root->input_buffer == (uint8_t)PROTOCOL_BINARY_REQ)
235 {
236 client->work= memcached_binary_protocol_process_data;
237 }
238 else if (client->root->callback->interface_version == 1)
239 {
240 /*
241 * The ASCII protocol can only be used if the implementors provide
242 * an implementation for the version 1 of the interface..
243 *
244 * @todo I should allow the implementors to provide an implementation
245 * for version 0 and 1 at the same time and set the preferred
246 * interface to use...
247 */
248 client->work= memcached_ascii_protocol_process_data;
249 }
250 else
251 {
252 /* Let's just output a warning the way it is supposed to look like
253 * in the ASCII protocol...
254 */
255 const char *err= "CLIENT_ERROR: Unsupported protocol\r\n";
256 client->root->spool(client, err, strlen(err));
257 client->root->drain(client);
258 return MEMCACHED_PROTOCOL_ERROR_EVENT; /* Unsupported protocol */
259 }
260
261 return client->work(client, length, endptr);
262 }
263
264 /*
265 ** **********************************************************************
266 ** * PUBLIC INTERFACE
267 ** * See protocol_handler.h for function description
268 ** **********************************************************************
269 */
270 struct memcached_protocol_st *memcached_protocol_create_instance(void)
271 {
272 struct memcached_protocol_st *ret= calloc(1, sizeof(*ret));
273 if (ret != NULL)
274 {
275 ret->recv= default_recv;
276 ret->send= default_send;
277 ret->drain= drain_output;
278 ret->spool= spool_output;
279 ret->input_buffer_size= 1 * 1024 * 1024;
280 ret->input_buffer= malloc(ret->input_buffer_size);
281 if (ret->input_buffer == NULL)
282 {
283 free(ret);
284 ret= NULL;
285 return NULL;
286 }
287
288 ret->buffer_cache= cache_create("protocol_handler",
289 CHUNK_BUFFERSIZE + sizeof(struct chunk_st),
290 0, NULL, NULL);
291 if (ret->buffer_cache == NULL)
292 {
293 free(ret->input_buffer);
294 free(ret);
295 }
296 }
297
298 return ret;
299 }
300
301 void memcached_protocol_destroy_instance(struct memcached_protocol_st *instance)
302 {
303 cache_destroy(instance->buffer_cache);
304 free(instance->input_buffer);
305 free(instance);
306 }
307
308 struct memcached_protocol_client_st *memcached_protocol_create_client(struct memcached_protocol_st *instance, memcached_socket_t sock)
309 {
310 struct memcached_protocol_client_st *ret= calloc(1, sizeof(*ret));
311 if (ret != NULL)
312 {
313 ret->root= instance;
314 ret->sock= sock;
315 ret->work= determine_protocol;
316 }
317
318 return ret;
319 }
320
321 void memcached_protocol_client_destroy(struct memcached_protocol_client_st *client)
322 {
323 free(client);
324 }
325
326 memcached_protocol_event_t memcached_protocol_client_work(struct memcached_protocol_client_st *client)
327 {
328 /* Try to send data and read from the socket */
329 bool more_data= true;
330 do
331 {
332 ssize_t len= client->root->recv(client,
333 client->sock,
334 client->root->input_buffer + client->input_buffer_offset,
335 client->root->input_buffer_size - client->input_buffer_offset);
336
337 if (len > 0)
338 {
339 /* Do we have the complete packet? */
340 if (client->input_buffer_offset > 0)
341 {
342 memcpy(client->root->input_buffer, client->input_buffer,
343 client->input_buffer_offset);
344 len += (ssize_t)client->input_buffer_offset;
345
346 /* @todo use buffer-cache! */
347 free(client->input_buffer);
348 client->input_buffer_offset= 0;
349 }
350
351 void *endptr;
352 memcached_protocol_event_t events= client->work(client, &len, &endptr);
353 if (events == MEMCACHED_PROTOCOL_ERROR_EVENT)
354 {
355 return MEMCACHED_PROTOCOL_ERROR_EVENT;
356 }
357
358 if (len > 0)
359 {
360 /* save the data for later on */
361 /* @todo use buffer-cache */
362 client->input_buffer= malloc((size_t)len);
363 if (client->input_buffer == NULL)
364 {
365 client->error= ENOMEM;
366 return MEMCACHED_PROTOCOL_ERROR_EVENT;
367 }
368 memcpy(client->input_buffer, endptr, (size_t)len);
369 client->input_buffer_offset= (size_t)len;
370 more_data= false;
371 }
372 }
373 else if (len == 0)
374 {
375 /* Connection closed */
376 drain_output(client);
377 return MEMCACHED_PROTOCOL_ERROR_EVENT;
378 }
379 else
380 {
381 if (get_socket_errno() != EWOULDBLOCK)
382 {
383 client->error= get_socket_errno();
384 /* mark this client as terminated! */
385 return MEMCACHED_PROTOCOL_ERROR_EVENT;
386 }
387 more_data= false;
388 }
389 } while (more_data);
390
391 if (!drain_output(client))
392 {
393 return MEMCACHED_PROTOCOL_ERROR_EVENT;
394 }
395
396 memcached_protocol_event_t ret= MEMCACHED_PROTOCOL_READ_EVENT;
397 if (client->output)
398 ret|= MEMCACHED_PROTOCOL_READ_EVENT;
399
400 return ret;
401 }