1 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
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 ;-)
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
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
28 #include <libmemcachedprotocol-0.0/handler.h>
29 #include <libmemcached/socket.hpp>
30 #include <example/byteorder.h>
31 #include "example/storage.h"
32 #include "example/memcached_light.h"
44 #include <sys/types.h>
47 extern memcached_binary_protocol_callback_st interface_v0_impl
;
48 extern memcached_binary_protocol_callback_st interface_v1_impl
;
50 static memcached_socket_t server_sockets
[1024];
51 static int num_server_sockets
= 0;
59 /* The default maximum number of connections... (change with -c) */
60 static int maxconns
= 1024;
62 static struct connection
*socket_userdata_map
;
63 static struct event_base
*event_base
;
76 static options_st global_options
;
79 * Callback for driving a client connection
80 * @param fd the socket for the client socket
81 * @param which identifying the event that occurred (not used)
82 * @param arg the connection structure for the client
84 static void drive_client(memcached_socket_t fd
, short, void *arg
)
86 struct connection
*client
= (struct connection
*)arg
;
87 struct memcached_protocol_client_st
* c
= (struct memcached_protocol_client_st
*)client
->userdata
;
90 memcached_protocol_event_t events
= memcached_protocol_client_work(c
);
91 if (events
& MEMCACHED_PROTOCOL_ERROR_EVENT
)
93 if (global_options
.is_verbose
)
95 struct sockaddr_in sin
;
96 socklen_t addrlen
= sizeof(sin
);
98 if (getsockname(fd
, (struct sockaddr
*)&sin
, &addrlen
) != -1)
100 std::cout
<< __FILE__
<< ":" << __LINE__
101 << " close(MEMCACHED_PROTOCOL_ERROR_EVENT)"
102 << " " << inet_ntoa(sin
.sin_addr
) << ":" << sin
.sin_port
108 std::cout
<< __FILE__
<< ":" << __LINE__
<< "close() MEMCACHED_PROTOCOL_ERROR_EVENT" << std::endl
;
112 memcached_protocol_client_destroy(c
);
118 if (events
& MEMCACHED_PROTOCOL_WRITE_EVENT
)
123 if (events
& MEMCACHED_PROTOCOL_READ_EVENT
)
128 event_set(&client
->event
, (intptr_t)fd
, flags
, drive_client
, client
);
129 event_base_set(event_base
, &client
->event
);
131 if (event_add(&client
->event
, 0) == -1)
133 memcached_protocol_client_destroy(c
);
140 * Callback for accepting new connections
141 * @param fd the socket for the server socket
142 * @param which identifying the event that occurred (not used)
143 * @param arg the connection structure for the server
145 static void accept_handler(memcached_socket_t fd
, short, void *arg
)
147 struct connection
*server
= (struct connection
*)arg
;
148 /* accept new client */
149 struct sockaddr_storage addr
;
150 socklen_t addrlen
= sizeof(addr
);
151 memcached_socket_t sock
= accept(fd
, (struct sockaddr
*)&addr
, &addrlen
);
153 if (sock
== INVALID_SOCKET
)
155 perror("Failed to accept client");
159 if (sock
>= maxconns
)
166 struct memcached_protocol_client_st
* c
= memcached_protocol_create_client((memcached_protocol_st
*)server
->userdata
, sock
);
173 memcached_protocol_client_set_verbose(c
, global_options
.is_verbose
);
174 struct connection
*client
= &socket_userdata_map
[sock
];
177 event_set(&client
->event
, (intptr_t)sock
, EV_READ
, drive_client
, client
);
178 event_base_set(event_base
, &client
->event
);
179 if (event_add(&client
->event
, 0) == -1)
181 std::cerr
<< "Failed to add event for " << sock
<< std::endl
;
182 memcached_protocol_client_destroy(c
);
189 * Create a socket and bind it to a specific port number
190 * @param port the port number to bind to
192 static int server_socket(const char *port
)
195 struct addrinfo hints
;
196 memset(&hints
, 0, sizeof(struct addrinfo
));
198 hints
.ai_flags
= AI_PASSIVE
;
199 hints
.ai_family
= AF_UNSPEC
;
200 hints
.ai_socktype
= SOCK_STREAM
;
202 int error
= getaddrinfo("127.0.0.1", port
, &hints
, &ai
);
205 if (error
!= EAI_SYSTEM
)
207 std::cerr
<< "getaddrinfo(): " << gai_strerror(error
) << std::endl
;
211 std::cerr
<< "getaddrinfo(): " << strerror(errno
) << std::endl
;
217 struct linger ling
= {0, 0};
219 for (struct addrinfo
*next
= ai
; next
; next
= next
->ai_next
)
221 memcached_socket_t sock
= socket(ai
->ai_family
, ai
->ai_socktype
, ai
->ai_protocol
);
222 if (sock
== INVALID_SOCKET
)
224 std::cerr
<< "Failed to create socket: " << strerror(errno
) << std::endl
;
231 if (ioctlsocket(sock
, FIONBIO
, &arg
) == SOCKET_ERROR
)
233 std::cerr
<< "Failed to set nonblocking io: " << strerror(errno
) << std::endl
;
238 flags
= fcntl(sock
, F_GETFL
, 0);
241 std::cerr
<< "Failed to get socket flags: " << strerror(errno
) << std::endl
;
246 if ((flags
& O_NONBLOCK
) != O_NONBLOCK
)
248 if (fcntl(sock
, F_SETFL
, flags
| O_NONBLOCK
) == -1)
250 std::cerr
<< "Failed to set socket to nonblocking mode: " << strerror(errno
) << std::endl
;
258 if (setsockopt(sock
, SOL_SOCKET
, SO_REUSEADDR
, (void *)&flags
, sizeof(flags
)) != 0)
260 std::cerr
<< "Failed to set SO_REUSEADDR: " << strerror(errno
) << std::endl
;
263 if (setsockopt(sock
, SOL_SOCKET
, SO_KEEPALIVE
, (void *)&flags
, sizeof(flags
)) != 0)
265 std::cerr
<< "Failed to set SO_KEEPALIVE: " << strerror(errno
) << std::endl
;
268 if (setsockopt(sock
, SOL_SOCKET
, SO_LINGER
, (void *)&ling
, sizeof(ling
)) != 0)
270 std::cerr
<< "Failed to set SO_LINGER: " << strerror(errno
) << std::endl
;
273 if (setsockopt(sock
, IPPROTO_TCP
, TCP_NODELAY
, (void *)&flags
, sizeof(flags
)) != 0)
275 std::cerr
<< "Failed to set TCP_NODELAY: " << strerror(errno
) << std::endl
;
278 if (bind(sock
, next
->ai_addr
, next
->ai_addrlen
) == SOCKET_ERROR
)
280 if (get_socket_errno() != EADDRINUSE
)
282 std::cerr
<< "bind(): " << strerror(errno
) << std::endl
;
289 if (listen(sock
, 1024) == SOCKET_ERROR
)
291 std::cerr
<< "listen(): " << strerror(errno
) << std::endl
;
296 if (global_options
.is_verbose
)
298 std::cout
<< "Listening to " << port
<< std::endl
;
301 server_sockets
[num_server_sockets
++]= sock
;
306 return (num_server_sockets
> 0) ? 0 : 1;
310 * Convert a command code to a textual string
311 * @param cmd the comcode to convert
312 * @return a textual string with the command or NULL for unknown commands
314 static const char* comcode2str(uint8_t cmd
)
316 static const char * const text
[] = {
317 "GET", "SET", "ADD", "REPLACE", "DELETE",
318 "INCREMENT", "DECREMENT", "QUIT", "FLUSH",
319 "GETQ", "NOOP", "VERSION", "GETK", "GETKQ",
320 "APPEND", "PREPEND", "STAT", "SETQ", "ADDQ",
321 "REPLACEQ", "DELETEQ", "INCREMENTQ", "DECREMENTQ",
322 "QUITQ", "FLUSHQ", "APPENDQ", "PREPENDQ"
325 if (cmd
<= PROTOCOL_BINARY_CMD_PREPENDQ
)
334 * Print out the command we are about to execute
336 static void pre_execute(const void *cookie
,
337 protocol_binary_request_header
*header
)
339 if (global_options
.is_verbose
)
343 const char *cmd
= comcode2str(header
->request
.opcode
);
346 std::cout
<< "pre_execute from " << cookie
<< ": " << cmd
<< std::endl
;
350 std::cout
<< "pre_execute from " << cookie
<< ": " << header
->request
.opcode
<< std::endl
;
355 std::cout
<< "pre_execute from " << cookie
<< std::endl
;
361 * Print out the command we just executed
363 static void post_execute(const void *cookie
,
364 protocol_binary_request_header
*header
)
366 if (global_options
.is_verbose
)
370 const char *cmd
= comcode2str(header
->request
.opcode
);
373 std::cout
<< "post_execute from " << cookie
<< ": " << cmd
<< std::endl
;
377 std::cout
<< "post_execute from " << cookie
<< ": " << header
->request
.opcode
<< std::endl
;
382 std::cout
<< "post_execute from " << cookie
<< std::endl
;
388 * Callback handler for all unknown commands.
389 * Send an unknown command back to the client
391 static protocol_binary_response_status
unknown(const void *cookie
,
392 protocol_binary_request_header
*header
,
393 memcached_binary_protocol_raw_response_handler response_handler
)
395 protocol_binary_response_no_extras response
;
396 memset(&response
, 0, sizeof(protocol_binary_response_no_extras
));
398 response
.message
.header
.response
.magic
= PROTOCOL_BINARY_RES
;
399 response
.message
.header
.response
.opcode
= header
->request
.opcode
;
400 response
.message
.header
.response
.status
= htons(PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND
);
401 response
.message
.header
.response
.opaque
= header
->request
.opaque
;
403 return response_handler(cookie
, header
, (protocol_binary_response_header
*)&response
);
407 * Program entry point. Bind to the specified port(s) and serve clients
409 * @param argc number of items in the argument vector
410 * @param argv argument vector
411 * @return EXIT_SUCCESS on success, 1 otherwise
413 int main(int argc
, char **argv
)
415 memcached_binary_protocol_callback_st
*interface
= &interface_v0_impl
;
417 event_base
= event_init();
418 if (event_base
== NULL
)
420 std::cerr
<< "Failed to create an instance of libevent" << std::endl
;
425 * We need to initialize the handlers manually due to a bug in the
426 * warnings generated by struct initialization in gcc (all the way up to 4.4)
428 initialize_interface_v0_handler();
429 initialize_interface_v1_handler();
435 OPT_PROTOCOL_VERSION
,
442 static struct option long_options
[]=
444 {"help", no_argument
, NULL
, OPT_HELP
},
445 {"port", required_argument
, NULL
, OPT_PORT
},
446 {"verbose", no_argument
, NULL
, OPT_VERBOSE
},
447 {"protocol", required_argument
, NULL
, OPT_PROTOCOL_VERSION
},
448 {"version", no_argument
, NULL
, OPT_VERSION
},
449 {"max-connections", required_argument
, NULL
, OPT_MAX_CONNECTIONS
},
450 {"pid-file", required_argument
, NULL
, OPT_PIDFILE
},
455 bool has_port
= false;
457 while (done
== false)
459 switch (getopt_long(argc
, argv
, "", long_options
, &option_index
))
465 case OPT_PROTOCOL_VERSION
:
466 interface
= &interface_v1_impl
;
470 global_options
.pid_file
= strdup(optarg
);
474 global_options
.is_verbose
= true;
479 (void)server_socket(optarg
);
482 case OPT_MAX_CONNECTIONS
:
483 maxconns
= atoi(optarg
);
486 case OPT_HELP
: /* FALLTHROUGH */
487 std::cout
<< "Usage: " << argv
[0] << std::endl
;
488 for (struct option
*ptr_option
= long_options
; ptr_option
->name
; ptr_option
++)
490 std::cout
<< "\t" << ptr_option
->name
<< std::endl
;
496 std::cerr
<< "Unknown option: " << optarg
<< std::endl
;
502 if (has_port
== false)
504 (void)server_socket("9999");
508 if (! initialize_storage())
510 /* Error message already printed */
514 if (global_options
.pid_file
)
519 pid_file
= fopen(global_options
.pid_file
, "w+");
521 if (pid_file
== NULL
)
523 perror(strerror(get_socket_errno()));
527 pid
= (uint32_t)getpid();
528 if (global_options
.is_verbose
)
530 std::cout
<< "pid:" << pid
<< std::endl
;
535 if (num_server_sockets
== 0)
537 std::cerr
<< "No server sockets are available" << std::endl
;
542 * Create and initialize the handles to the protocol handlers. I want
543 * to be able to trace the traffic throught the pre/post handlers, and
544 * set up a common handler for unknown messages
546 interface
->pre_execute
= pre_execute
;
547 interface
->post_execute
= post_execute
;
548 interface
->unknown
= unknown
;
550 struct memcached_protocol_st
*protocol_handle
;
551 if ((protocol_handle
= memcached_protocol_create_instance()) == NULL
)
553 std::cerr
<< "Failed to allocate protocol handle" << std::endl
;
557 socket_userdata_map
= (struct connection
*)calloc((size_t)(maxconns
), sizeof(struct connection
));
558 if (socket_userdata_map
== NULL
)
560 std::cerr
<< "Failed to allocate room for connections" << std::endl
;
564 memcached_binary_protocol_set_callbacks(protocol_handle
, interface
);
565 memcached_binary_protocol_set_pedantic(protocol_handle
, true);
567 for (int xx
= 0; xx
< num_server_sockets
; ++xx
)
569 struct connection
*conn
= &socket_userdata_map
[server_sockets
[xx
]];
570 conn
->userdata
= protocol_handle
;
572 event_set(&conn
->event
, (intptr_t)server_sockets
[xx
], EV_READ
| EV_PERSIST
, accept_handler
, conn
);
574 event_base_set(event_base
, &conn
->event
);
575 if (event_add(&conn
->event
, 0) == -1)
577 std::cerr
<< "Failed to add event for " << server_sockets
[xx
] << std::endl
;
578 closesocket(server_sockets
[xx
]);
582 /* Serve all of the clients */
583 event_base_loop(event_base
, 0);