memcached_quit should clear internal server variables
[m6w6/libmemcached] / example / memcached_light.c
1 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
2 /**
3 * What is a library without an example to show you how to use the library?
4 * This example use both interfaces to implement a small memcached server.
5 * Please note that this is an exemple on how to use the library, not
6 * an implementation of a scalable memcached server. If you look closely
7 * at the example it isn't even multithreaded ;-)
8 *
9 * With that in mind, let me give you some pointers into the source:
10 * storage.c/h - Implements the item store for this server and not really
11 * interesting for this example.
12 * interface_v0.c - Shows an implementation of the memcached server by using
13 * the "raw" access to the packets as they arrive
14 * interface_v1.c - Shows an implementation of the memcached server by using
15 * the more "logical" interface.
16 * memcached_light.c - This file sets up all of the sockets and run the main
17 * message loop.
18 *
19 *
20 * config.h is included so that I can use the ntohll/htonll on platforms that
21 * doesn't have that (this is a private function inside libmemcached, so you
22 * cannot use it directly from libmemcached without special modifications to
23 * the library)
24 */
25
26 #include "config.h"
27 #include <assert.h>
28 #include <sys/types.h>
29 #include <sys/socket.h>
30 #include <netdb.h>
31 #include <netinet/tcp.h>
32 #include <stdio.h>
33 #include <unistd.h>
34 #include <fcntl.h>
35 #include <errno.h>
36 #include <stdlib.h>
37 #include <string.h>
38 #include <poll.h>
39
40 #include <libmemcached/protocol_handler.h>
41 #include <libmemcached/byteorder.h>
42 #include "storage.h"
43 #include "memcached_light.h"
44
45 extern memcached_binary_protocol_callback_st interface_v0_impl;
46 extern memcached_binary_protocol_callback_st interface_v1_impl;
47
48 static int server_sockets[1024];
49 static int num_server_sockets= 0;
50 static void* socket_userdata_map[1024];
51 static bool verbose= false;
52
53 /**
54 * Create a socket and bind it to a specific port number
55 * @param port the port number to bind to
56 */
57 static int server_socket(const char *port) {
58 struct addrinfo *ai;
59 struct addrinfo hints= { .ai_flags= AI_PASSIVE,
60 .ai_family= AF_UNSPEC,
61 .ai_socktype= SOCK_STREAM };
62
63 int error= getaddrinfo("127.0.0.1", port, &hints, &ai);
64 if (error != 0)
65 {
66 if (error != EAI_SYSTEM)
67 fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error));
68 else
69 perror("getaddrinfo()");
70
71 return 1;
72 }
73
74 struct linger ling= {0, 0};
75
76 for (struct addrinfo *next= ai; next; next= next->ai_next)
77 {
78 int sock= socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
79 if (sock == -1)
80 {
81 perror("Failed to create socket");
82 continue;
83 }
84
85 int flags= fcntl(sock, F_GETFL, 0);
86 if (flags == -1)
87 {
88 perror("Failed to get socket flags");
89 close(sock);
90 continue;
91 }
92
93 if ((flags & O_NONBLOCK) != O_NONBLOCK)
94 {
95 if (fcntl(sock, F_SETFL, flags | O_NONBLOCK) == -1)
96 {
97 perror("Failed to set socket to nonblocking mode");
98 close(sock);
99 continue;
100 }
101 }
102
103 flags= 1;
104 if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)) != 0)
105 perror("Failed to set SO_REUSEADDR");
106
107 if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)) != 0)
108 perror("Failed to set SO_KEEPALIVE");
109
110 if (setsockopt(sock, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)) != 0)
111 perror("Failed to set SO_LINGER");
112
113 if (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags)) != 0)
114 perror("Failed to set TCP_NODELAY");
115
116 if (bind(sock, next->ai_addr, next->ai_addrlen) == -1)
117 {
118 if (errno != EADDRINUSE)
119 {
120 perror("bind()");
121 freeaddrinfo(ai);
122 }
123 close(sock);
124 continue;
125 }
126
127 if (listen(sock, 1024) == -1)
128 {
129 perror("listen()");
130 close(sock);
131 continue;
132 }
133
134 server_sockets[num_server_sockets++]= sock;
135 }
136
137 freeaddrinfo(ai);
138
139 return (num_server_sockets > 0) ? 0 : 1;
140 }
141
142 /**
143 * Convert a command code to a textual string
144 * @param cmd the comcode to convert
145 * @return a textual string with the command or NULL for unknown commands
146 */
147 static const char* comcode2str(uint8_t cmd)
148 {
149 static const char * const text[] = {
150 "GET", "SET", "ADD", "REPLACE", "DELETE",
151 "INCREMENT", "DECREMENT", "QUIT", "FLUSH",
152 "GETQ", "NOOP", "VERSION", "GETK", "GETKQ",
153 "APPEND", "PREPEND", "STAT", "SETQ", "ADDQ",
154 "REPLACEQ", "DELETEQ", "INCREMENTQ", "DECREMENTQ",
155 "QUITQ", "FLUSHQ", "APPENDQ", "PREPENDQ"
156 };
157
158 if (cmd <= PROTOCOL_BINARY_CMD_PREPENDQ)
159 return text[cmd];
160
161 return NULL;
162 }
163
164 /**
165 * Print out the command we are about to execute
166 */
167 static void pre_execute(const void *cookie __attribute__((unused)),
168 protocol_binary_request_header *header __attribute__((unused)))
169 {
170 if (verbose)
171 {
172 const char *cmd= comcode2str(header->request.opcode);
173 if (cmd != NULL)
174 fprintf(stderr, "pre_execute from %p: %s\n", cookie, cmd);
175 else
176 fprintf(stderr, "pre_execute from %p: 0x%02x\n", cookie, header->request.opcode);
177 }
178 }
179
180 /**
181 * Print out the command we just executed
182 */
183 static void post_execute(const void *cookie __attribute__((unused)),
184 protocol_binary_request_header *header __attribute__((unused)))
185 {
186 if (verbose)
187 {
188 const char *cmd= comcode2str(header->request.opcode);
189 if (cmd != NULL)
190 fprintf(stderr, "post_execute from %p: %s\n", cookie, cmd);
191 else
192 fprintf(stderr, "post_execute from %p: 0x%02x\n", cookie, header->request.opcode);
193 }
194 }
195
196 /**
197 * Callback handler for all unknown commands.
198 * Send an unknown command back to the client
199 */
200 static protocol_binary_response_status unknown(const void *cookie,
201 protocol_binary_request_header *header,
202 memcached_binary_protocol_raw_response_handler response_handler)
203 {
204 protocol_binary_response_no_extras response= {
205 .message= {
206 .header.response= {
207 .magic= PROTOCOL_BINARY_RES,
208 .opcode= header->request.opcode,
209 .status= htons(PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND),
210 .opaque= header->request.opaque
211 }
212 }
213 };
214
215 return response_handler(cookie, header, (void*)&response);
216 }
217
218 static void work(void);
219
220 /**
221 * Program entry point. Bind to the specified port(s) and serve clients
222 *
223 * @param argc number of items in the argument vector
224 * @param argv argument vector
225 * @return 0 on success, 1 otherwise
226 */
227 int main(int argc, char **argv)
228 {
229 bool port_specified= false;
230 int cmd;
231 memcached_binary_protocol_callback_st *interface= &interface_v0_impl;
232
233 /*
234 * We need to initialize the handlers manually due to a bug in the
235 * warnings generated by struct initialization in gcc (all the way up to 4.4)
236 */
237 initialize_interface_v0_handler();
238
239 while ((cmd= getopt(argc, argv, "v1p:?")) != EOF)
240 {
241 switch (cmd) {
242 case '1':
243 interface= &interface_v1_impl;
244 break;
245 case 'p':
246 port_specified= true;
247 (void)server_socket(optarg);
248 break;
249 case 'v':
250 verbose= true;
251 break;
252 case '?': /* FALLTHROUGH */
253 default:
254 (void)fprintf(stderr, "Usage: %s [-p port] [-v] [-1]\n", argv[0]);
255 return 1;
256 }
257 }
258
259 if (!initialize_storage())
260 {
261 /* Error message already printed */
262 return 1;
263 }
264
265 if (!port_specified)
266 (void)server_socket("9999");
267
268 if (num_server_sockets == 0)
269 {
270 fprintf(stderr, "I don't have any server sockets\n");
271 return 1;
272 }
273
274 /*
275 * Create and initialize the handles to the protocol handlers. I want
276 * to be able to trace the traffic throught the pre/post handlers, and
277 * set up a common handler for unknown messages
278 */
279 interface->pre_execute= pre_execute;
280 interface->post_execute= post_execute;
281 interface->unknown= unknown;
282
283 struct memcached_protocol_st *protocol_handle;
284 if ((protocol_handle= memcached_protocol_create_instance()) == NULL)
285 {
286 fprintf(stderr, "Failed to allocate protocol handle\n");
287 return 1;
288 }
289
290 memcached_binary_protocol_set_callbacks(protocol_handle, interface);
291 memcached_binary_protocol_set_pedantic(protocol_handle, true);
292
293 for (int xx= 0; xx < num_server_sockets; ++xx)
294 socket_userdata_map[server_sockets[xx]]= protocol_handle;
295
296 /* Serve all of the clients */
297 work();
298
299 /* NOTREACHED */
300 return 0;
301 }
302
303 static void work(void)
304 {
305 #define MAX_SERVERS_TO_POLL 100
306 struct pollfd fds[MAX_SERVERS_TO_POLL];
307 int max_poll;
308
309 for (max_poll= 0; max_poll < num_server_sockets; ++max_poll)
310 {
311 fds[max_poll].events= POLLIN;
312 fds[max_poll].revents= 0;
313 fds[max_poll].fd= server_sockets[max_poll];
314 ++max_poll;
315 }
316
317 while (true)
318 {
319 int err= poll(fds, (nfds_t)max_poll, -1);
320
321 if (err == 0 || (err == -1 && errno != EINTR))
322 {
323 perror("poll() failed");
324 abort();
325 }
326
327 /* find the available filedescriptors */
328 for (int x= max_poll - 1; x > -1 && err > 0; --x)
329 {
330 if (fds[x].revents != 0)
331 {
332 --err;
333 if (x < num_server_sockets)
334 {
335 /* accept new client */
336 struct sockaddr_storage addr;
337 socklen_t addrlen= sizeof(addr);
338 int sock= accept(fds[x].fd, (struct sockaddr *)&addr,
339 &addrlen);
340
341 if (sock == -1)
342 {
343 perror("Failed to accept client");
344 continue;
345 }
346
347 struct memcached_protocol_st *protocol;
348 protocol= socket_userdata_map[fds[x].fd];
349
350 struct memcached_protocol_client_st* c;
351 c= memcached_protocol_create_client(protocol, sock);
352 if (c == NULL)
353 {
354 fprintf(stderr, "Failed to create client\n");
355 close(sock);
356 }
357 else
358 {
359 socket_userdata_map[sock]= c;
360 fds[max_poll].events= POLLIN;
361 fds[max_poll].revents= 0;
362 fds[max_poll].fd= sock;
363 ++max_poll;
364 }
365 }
366 else
367 {
368 /* drive the client */
369 struct memcached_protocol_client_st* c;
370 c= socket_userdata_map[fds[x].fd];
371 assert(c != NULL);
372 fds[max_poll].events= 0;
373
374 memcached_protocol_event_t events= memcached_protocol_client_work(c);
375 if (events & MEMCACHED_PROTOCOL_WRITE_EVENT)
376 fds[max_poll].events= POLLOUT;
377
378 if (events & MEMCACHED_PROTOCOL_READ_EVENT)
379 fds[max_poll].events= POLLIN;
380
381 if (!(events & MEMCACHED_PROTOCOL_PAUSE_EVENT ||
382 fds[max_poll].events != 0))
383 {
384 memcached_protocol_client_destroy(c);
385 close(fds[x].fd);
386 fds[x].events= 0;
387
388 if (x != max_poll - 1)
389 memmove(fds + x, fds + x + 1, (size_t)(max_poll - x));
390
391 --max_poll;
392 }
393 }
394 }
395 }
396 }
397 }