1 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
9 #include <embedded_innodb-1.0/innodb.h>
13 const char *tablename
= "memcached/items";
16 #define data_col_idx 1
17 #define flags_col_idx 2
24 * To avoid cluttering down all the code with error checking I use the
25 * following macro. It will execute the statement and verify that the
26 * result of the operation is DB_SUCCESS. If any other error code is
27 * returned it will print an "assert-like" output and jump to the
28 * label error_exit. There I release resources before returning out of
31 * @param a the expression to execute
34 #define checked(expression) \
36 ib_err_t checked_err= expression; \
37 if (checked_err != DB_SUCCESS) \
39 fprintf(stderr, "ERROR: %s at %u: Failed: <%s>\n\t%s\n", \
40 __FILE__, __LINE__, #expression, \
41 ib_strerror(checked_err)); \
47 * Create the database schema.
48 * @return true if the database schema was created without any problems
51 static bool create_schema(void) {
52 ib_tbl_sch_t schema
= NULL
;
53 ib_idx_sch_t dbindex
= NULL
;
55 if (ib_database_create("memcached") != IB_TRUE
)
57 fprintf(stderr
, "Failed to create database\n");
61 ib_trx_t transaction
= ib_trx_begin(IB_TRX_SERIALIZABLE
);
64 checked(ib_table_schema_create(tablename
, &schema
, IB_TBL_COMPACT
, 0));
65 checked(ib_table_schema_add_col(schema
, "key", IB_BLOB
,
66 IB_COL_NOT_NULL
, 0, 32767));
67 checked(ib_table_schema_add_col(schema
, "data", IB_BLOB
,
68 IB_COL_NONE
, 0, 1024*1024));
69 checked(ib_table_schema_add_col(schema
, "flags", IB_INT
,
70 IB_COL_UNSIGNED
, 0, 4));
71 checked(ib_table_schema_add_col(schema
, "cas", IB_INT
,
72 IB_COL_UNSIGNED
, 0, 8));
73 checked(ib_table_schema_add_col(schema
, "exp", IB_INT
,
74 IB_COL_UNSIGNED
, 0, 4));
75 checked(ib_table_schema_add_index(schema
, "PRIMARY_KEY", &dbindex
));
76 checked(ib_index_schema_add_col(dbindex
, "key", 0));
77 checked(ib_index_schema_set_clustered(dbindex
));
78 checked(ib_schema_lock_exclusive(transaction
));
79 checked(ib_table_create(transaction
, schema
, &table_id
));
80 checked(ib_trx_commit(transaction
));
81 ib_table_schema_delete(schema
);
86 /* @todo release resources! */
88 ib_err_t error
= ib_trx_rollback(transaction
);
89 if (error
!= DB_SUCCESS
)
90 fprintf(stderr
, "Failed to roll back the transaction:\n\t%s\n",
97 * Store an item into the database. Update the CAS id on the item before
98 * storing it in the database.
100 * @param trx the transaction to use
101 * @param item the item to store
102 * @return true if we can go ahead and commit the transaction, false otherwise
104 static bool do_put_item(ib_trx_t trx
, struct item
* item
) {
107 ib_crsr_t cursor
= NULL
;
108 ib_tpl_t tuple
= NULL
;
111 checked(ib_cursor_open_table(tablename
, trx
, &cursor
));
112 checked(ib_cursor_lock(cursor
, IB_LOCK_X
));
113 tuple
= ib_clust_read_tuple_create(cursor
);
115 checked(ib_col_set_value(tuple
, key_col_idx
, item
->key
, item
->nkey
));
116 checked(ib_col_set_value(tuple
, data_col_idx
, item
->data
, item
->size
));
117 checked(ib_tuple_write_u32(tuple
, flags_col_idx
, item
->flags
));
118 checked(ib_tuple_write_u64(tuple
, cas_col_idx
, item
->cas
));
119 checked(ib_tuple_write_u32(tuple
, exp_col_idx
, (ib_u32_t
)item
->exp
));
120 checked(ib_cursor_insert_row(cursor
, tuple
));
123 /* Release resources: */
128 ib_tuple_delete(tuple
);
131 ib_cursor_close(cursor
);
137 * Try to locate an item in the database. Return a cursor and the tuple to
138 * the item if I found it in the database.
140 * @param trx the transaction to use
141 * @param key the key of the item to look up
142 * @param nkey the size of the key
143 * @param cursor where to store the cursor (OUT)
144 * @param tuple where to store the tuple (OUT)
145 * @return true if I found the object, false otherwise
147 static bool do_locate_item(ib_trx_t trx
,
153 ib_tpl_t tuple
= NULL
;
157 checked(ib_cursor_open_table(tablename
, trx
, cursor
));
158 tuple
= ib_clust_search_tuple_create(*cursor
);
161 fprintf(stderr
, "Failed to allocate tuple object\n");
165 checked(ib_col_set_value(tuple
, key_col_idx
, key
, nkey
));
166 ib_err_t err
= ib_cursor_moveto(*cursor
, tuple
, IB_CUR_GE
, &res
);
168 if (err
== DB_SUCCESS
&& res
== 0)
170 ib_tuple_delete(tuple
);
173 else if (err
!= DB_SUCCESS
&&
174 err
!= DB_RECORD_NOT_FOUND
&&
175 err
!= DB_END_OF_INDEX
)
177 fprintf(stderr
, "ERROR: ib_cursor_moveto(): %s\n", ib_strerror(err
));
182 ib_tuple_delete(tuple
);
184 ib_cursor_close(*cursor
);
191 * Try to get an item from the database
193 * @param trx the transaction to use
194 * @param key the key to get
195 * @param nkey the lenght of the key
196 * @return a pointer to the item if I found it in the database
198 static struct item
* do_get_item(ib_trx_t trx
, const void* key
, size_t nkey
) {
199 ib_crsr_t cursor
= NULL
;
200 ib_tpl_t tuple
= NULL
;
201 struct item
* retval
= NULL
;
203 if (do_locate_item(trx
, key
, nkey
, &cursor
)) {
204 tuple
= ib_clust_read_tuple_create(cursor
);
207 fprintf(stderr
, "Failed to create read tuple\n");
210 checked(ib_cursor_read_row(cursor
, tuple
));
212 ib_ulint_t datalen
= ib_col_get_meta(tuple
, data_col_idx
, &meta
);
213 ib_ulint_t flaglen
= ib_col_get_meta(tuple
, flags_col_idx
, &meta
);
214 ib_ulint_t caslen
= ib_col_get_meta(tuple
, cas_col_idx
, &meta
);
215 ib_ulint_t explen
= ib_col_get_meta(tuple
, exp_col_idx
, &meta
);
216 const void *dataptr
= ib_col_get_value(tuple
, data_col_idx
);
218 retval
= create_item(key
, nkey
, dataptr
, datalen
, 0, 0);
219 if (retval
== NULL
) {
220 fprintf(stderr
, "Failed to allocate memory\n");
226 checked(ib_tuple_read_u32(tuple
, flags_col_idx
, &val
));
227 retval
->flags
= (uint32_t)val
;
231 checked(ib_tuple_read_u64(tuple
, cas_col_idx
, &val
));
232 retval
->cas
= (uint64_t)val
;
236 checked(ib_tuple_read_u32(tuple
, exp_col_idx
, &val
));
237 retval
->exp
= (time_t)val
;
241 /* Release resources */
246 ib_tuple_delete(tuple
);
249 ib_cursor_close(cursor
);
255 * Delete an item from the cache
256 * @param trx the transaction to use
257 * @param key the key of the item to delete
258 * @param nkey the length of the key
259 * @return true if we should go ahead and commit the transaction
260 * or false if we should roll back (if the key didn't exists)
262 static bool do_delete_item(ib_trx_t trx
, const void* key
, size_t nkey
) {
263 ib_crsr_t cursor
= NULL
;
266 if (do_locate_item(trx
, key
, nkey
, &cursor
))
268 checked(ib_cursor_lock(cursor
, IB_LOCK_X
));
269 checked(ib_cursor_delete_row(cursor
));
272 /* Release resources */
277 ib_cursor_close(cursor
);
283 /****************************************************************************
285 ***************************************************************************/
288 * Initialize the database storage
289 * @return true if the database was initialized successfully, false otherwise
291 bool initialize_storage(void) {
296 checked(ib_cfg_set_text("data_home_dir", "/tmp/memcached_light"));
297 checked(ib_cfg_set_text("log_group_home_dir", "/tmp/memcached_light"));
298 checked(ib_cfg_set_bool_on("file_per_table"));
299 checked(ib_startup("barracuda"));
301 /* check to see if the table exists or if we should create the schema */
302 error
= ib_table_get_id(tablename
, &tid
);
303 if (error
== DB_TABLE_NOT_FOUND
) {
304 if (!create_schema()) {
307 } else if (error
!= DB_SUCCESS
) {
308 fprintf(stderr
, "Failed to get table id: %s\n", ib_strerror(error
));
319 * Shut down this storage engine
321 void shutdown_storage(void) {
322 checked(ib_shutdown());
328 * Store an item in the databse
330 * @param item the item to store
332 void put_item(struct item
* item
) {
333 ib_trx_t transaction
= ib_trx_begin(IB_TRX_SERIALIZABLE
);
334 if (do_put_item(transaction
, item
)) {
335 ib_err_t error
= ib_trx_commit(transaction
);
336 if (error
!= DB_SUCCESS
) {
337 fprintf(stderr
, "Failed to store key:\n\t%s\n",
341 ib_err_t error
= ib_trx_rollback(transaction
);
342 if (error
!= DB_SUCCESS
)
343 fprintf(stderr
, "Failed to roll back the transaction:\n\t%s\n",
349 * Get an item from the engine
350 * @param key the key to grab
351 * @param nkey number of bytes in the key
352 * @return pointer to the item if found
354 struct item
* get_item(const void* key
, size_t nkey
) {
355 ib_trx_t transaction
= ib_trx_begin(IB_TRX_SERIALIZABLE
);
356 struct item
* ret
= do_get_item(transaction
, key
, nkey
);
357 ib_err_t error
= ib_trx_rollback(transaction
);
358 if (error
!= DB_SUCCESS
)
359 fprintf(stderr
, "Failed to roll back the transaction:\n\t%s\n",
366 * Create an item structure and initialize it with the content
368 * @param key the key for the item
369 * @param nkey the number of bytes in the key
370 * @param data pointer to the value for the item (may be NULL)
371 * @param size the size of the data
372 * @param flags the flags to store with the data
373 * @param exp the expiry time for the item
374 * @return pointer to an initialized item object or NULL if allocation failed
376 struct item
* create_item(const void* key
, size_t nkey
, const void* data
,
377 size_t size
, uint32_t flags
, time_t exp
)
379 struct item
* ret
= calloc(1, sizeof(*ret
));
382 ret
->key
= malloc(nkey
);
385 ret
->data
= malloc(size
);
388 if (ret
->key
== NULL
|| (size
> 0 && ret
->data
== NULL
))
396 memcpy(ret
->key
, key
, nkey
);
399 memcpy(ret
->data
, data
, size
);
412 * Delete an item from the cache
413 * @param key the key of the item to delete
414 * @param nkey the length of the key
415 * @return true if the item was deleted from the cache
417 bool delete_item(const void* key
, size_t nkey
) {
418 ib_trx_t transaction
= ib_trx_begin(IB_TRX_REPEATABLE_READ
);
420 bool ret
= do_delete_item(transaction
, key
, nkey
);
424 /* object found. commit transaction */
425 ib_err_t error
= ib_trx_commit(transaction
);
426 if (error
!= DB_SUCCESS
)
428 fprintf(stderr
, "Failed to delete key:\n\t%s\n",
435 ib_err_t error
= ib_trx_rollback(transaction
);
436 if (error
!= DB_SUCCESS
)
437 fprintf(stderr
, "Failed to roll back the transaction:\n\t%s\n",
445 * Flush the entire cache
446 * @param when when the cache should be flushed (0 == immediately)
448 void flush(uint32_t when
__attribute__((unused
))) {
449 /* @TODO implement support for when != 0 */
450 ib_trx_t transaction
= ib_trx_begin(IB_TRX_REPEATABLE_READ
);
451 ib_crsr_t cursor
= NULL
;
452 ib_err_t err
= DB_SUCCESS
;
454 checked(ib_cursor_open_table(tablename
, transaction
, &cursor
));
455 checked(ib_cursor_first(cursor
));
456 checked(ib_cursor_lock(cursor
, IB_LOCK_X
));
459 checked(ib_cursor_delete_row(cursor
));
460 } while ((err
= ib_cursor_next(cursor
)) == DB_SUCCESS
);
462 if (err
!= DB_END_OF_INDEX
)
464 fprintf(stderr
, "Failed to flush the cache: %s\n", ib_strerror(err
));
467 ib_cursor_close(cursor
);
469 checked(ib_trx_commit(transaction
));
474 ib_cursor_close(cursor
);
476 ib_err_t error
= ib_trx_rollback(transaction
);
477 if (error
!= DB_SUCCESS
)
478 fprintf(stderr
, "Failed to roll back the transaction:\n\t%s\n",
483 * Update the cas ID in the item structure
484 * @param item the item to update
486 void update_cas(struct item
* item
) {
491 * Release all the resources allocated by the item
492 * @param item the item to release
494 void release_item(struct item
* item
) {