~drizzle-trunk/drizzle/development

/*****************************************************************************

Copyright (C) 2007, 2010, Innobase Oy. All Rights Reserved.

This program is free software; you can redistribute it and/or modify it under

the terms of the GNU General Public License as published by the Free Software

Foundation; version 2 of the License.

This program is distributed in the hope that it will be useful, but WITHOUT

ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with

this program; if not, write to the Free Software Foundation, Inc., 51 Franklin

St, Fifth Floor, Boston, MA 02110-1301 USA

*****************************************************************************/

/**************************************************//**

@file trx/trx0i_s.c

INFORMATION SCHEMA innodb_trx, innodb_locks and

innodb_lock_waits tables fetch code.

The code below fetches information needed to fill those

3 dynamic tables and uploads it into a "transactions

table cache" for later retrieval.

Created July 17, 2007 Vasil Dimov

*******************************************************/

#include "config.h"

/* Found during the build of 5.5.3 on Linux 2.4 and early 2.6 kernels:

   The includes "univ.i" -> "my_global.h" cause a different path

   to be taken further down with pthread functions and types,

   so they must come first.

   From the symptoms, this is related to bug#46587 in the MySQL bug DB.

*/

#include "univ.i"

#if !defined(BUILD_DRIZZLE)

# include <mysql/plugin.h>

#endif

#include "mysql_addons.h"

#include "buf0buf.h"

#include "dict0dict.h"

#include "ha0storage.h"

#include "ha_prototypes.h"

#include "hash0hash.h"

#include "lock0iter.h"

#include "lock0lock.h"

#include "mem0mem.h"

#include "page0page.h"

#include "rem0rec.h"

#include "row0row.h"

#include "srv0srv.h"

#include "sync0rw.h"

#include "sync0sync.h"

#include "sync0types.h"

#include "trx0i_s.h"

#include "trx0sys.h"

#include "trx0trx.h"

#include "ut0mem.h"

#include "ut0ut.h"

/** Initial number of rows in the table cache */

#define TABLE_CACHE_INITIAL_ROWSNUM	1024

/** @brief The maximum number of chunks to allocate for a table cache.

The rows of a table cache are stored in a set of chunks. When a new

row is added a new chunk is allocated if necessary. Assuming that the

first one is 1024 rows (TABLE_CACHE_INITIAL_ROWSNUM) and each

subsequent is N/2 where N is the number of rows we have allocated till

now, then 39th chunk would accommodate 1677416425 rows and all chunks

would accommodate 3354832851 rows. */

#define MEM_CHUNKS_IN_TABLE_CACHE	39

/** The following are some testing auxiliary macros. Do not enable them

in a production environment. */

/* @{ */

#if 0

/** If this is enabled then lock folds will always be different

resulting in equal rows being put in a different cells of the hash

table. Checking for duplicates will be flawed because different

fold will be calculated when a row is searched in the hash table. */

#define TEST_LOCK_FOLD_ALWAYS_DIFFERENT

#endif

#if 0

/** This effectively kills the search-for-duplicate-before-adding-a-row

function, but searching in the hash is still performed. It will always

be assumed that lock is not present and insertion will be performed in

the hash table. */

#define TEST_NO_LOCKS_ROW_IS_EVER_EQUAL_TO_LOCK_T

#endif

#if 0

/** This aggressively repeats adding each row many times. Depending on

the above settings this may be noop or may result in lots of rows being

added. */

#define TEST_ADD_EACH_LOCKS_ROW_MANY_TIMES

#endif

#if 0

/** Very similar to TEST_NO_LOCKS_ROW_IS_EVER_EQUAL_TO_LOCK_T but hash

table search is not performed at all. */

#define TEST_DO_NOT_CHECK_FOR_DUPLICATE_ROWS

#endif

#if 0

/** Do not insert each row into the hash table, duplicates may appear

if this is enabled, also if this is enabled searching into the hash is

noop because it will be empty. */

#define TEST_DO_NOT_INSERT_INTO_THE_HASH_TABLE

#endif

/* @} */

/** Memory limit passed to ha_storage_put_memlim().

@param cache	hash storage

@return		maximum allowed allocation size */

#define MAX_ALLOWED_FOR_STORAGE(cache)		\

	(TRX_I_S_MEM_LIMIT			\

	 - (cache)->mem_allocd)

/** Memory limit in table_cache_create_empty_row().

@param cache	hash storage

@return		maximum allowed allocation size */

#define MAX_ALLOWED_FOR_ALLOC(cache)		\

	(TRX_I_S_MEM_LIMIT			\

	 - (cache)->mem_allocd			\

	 - ha_storage_get_size((cache)->storage))

/** Memory for each table in the intermediate buffer is allocated in

separate chunks. These chunks are considered to be concatenated to

represent one flat array of rows. */

typedef struct i_s_mem_chunk_struct {

	ulint	offset;		/*!< offset, in number of rows */

	ulint	rows_allocd;	/*!< the size of this chunk, in number

				of rows */

	void*	base;		/*!< start of the chunk */

} i_s_mem_chunk_t;

/** This represents one table's cache. */

typedef struct i_s_table_cache_struct {

	ulint		rows_used;	/*!< number of used rows */

	ulint		rows_allocd;	/*!< number of allocated rows */

	ulint		row_size;	/*!< size of a single row */

	i_s_mem_chunk_t	chunks[MEM_CHUNKS_IN_TABLE_CACHE]; /*!< array of

					memory chunks that stores the

					rows */

} i_s_table_cache_t;

/** This structure describes the intermediate buffer */

struct trx_i_s_cache_struct {

	rw_lock_t	rw_lock;	/*!< read-write lock protecting

					the rest of this structure */

	ullint		last_read;	/*!< last time the cache was read;

					measured in microseconds since

					epoch */

	mutex_t		last_read_mutex;/*!< mutex protecting the

					last_read member - it is updated

					inside a shared lock of the

					rw_lock member */

	i_s_table_cache_t innodb_trx;	/*!< innodb_trx table */

	i_s_table_cache_t innodb_locks;	/*!< innodb_locks table */

	i_s_table_cache_t innodb_lock_waits;/*!< innodb_lock_waits table */

/** the hash table size is LOCKS_HASH_CELLS_NUM * sizeof(void*) bytes */

#define LOCKS_HASH_CELLS_NUM		10000

	hash_table_t*	locks_hash;	/*!< hash table used to eliminate

					duplicate entries in the

					innodb_locks table */

/** Initial size of the cache storage */

#define CACHE_STORAGE_INITIAL_SIZE	1024

/** Number of hash cells in the cache storage */

#define CACHE_STORAGE_HASH_CELLS	2048

	ha_storage_t*	storage;	/*!< storage for external volatile

					data that can possibly not be

					available later, when we release

					the kernel mutex */

	ulint		mem_allocd;	/*!< the amount of memory

					allocated with mem_alloc*() */

	ibool		is_truncated;	/*!< this is TRUE if the memory

					limit was hit and thus the data

					in the cache is truncated */

};

/** This is the intermediate buffer where data needed to fill the

INFORMATION SCHEMA tables is fetched and later retrieved by the C++

code in handler/i_s.cc. */

static trx_i_s_cache_t	trx_i_s_cache_static;

/** This is the intermediate buffer where data needed to fill the

INFORMATION SCHEMA tables is fetched and later retrieved by the C++

code in handler/i_s.cc. */

UNIV_INTERN trx_i_s_cache_t*	trx_i_s_cache = &trx_i_s_cache_static;

/* Key to register the lock/mutex with performance schema */

#ifdef UNIV_PFS_RWLOCK

UNIV_INTERN mysql_pfs_key_t	trx_i_s_cache_lock_key;

#endif /* UNIV_PFS_RWLOCK */

#ifdef UNIV_PFS_MUTEX

UNIV_INTERN mysql_pfs_key_t	cache_last_read_mutex_key;

#endif /* UNIV_PFS_MUTEX */

/*******************************************************************//**

For a record lock that is in waiting state retrieves the only bit that

is set, for a table lock returns ULINT_UNDEFINED.

@return	record number within the heap */

static

ulint

wait_lock_get_heap_no(

/*==================*/

	const lock_t*	lock)	/*!< in: lock */

{

	ulint	ret;

	switch (lock_get_type(lock)) {

	case LOCK_REC:

		ret = lock_rec_find_set_bit(lock);

		ut_a(ret != ULINT_UNDEFINED);

		break;

	case LOCK_TABLE:

		ret = ULINT_UNDEFINED;

		break;

	default:

		ut_error;

	}

	return(ret);

}

/*******************************************************************//**

Initializes the members of a table cache. */

static

void

table_cache_init(

/*=============*/

	i_s_table_cache_t*	table_cache,	/*!< out: table cache */

	size_t			row_size)	/*!< in: the size of a

						row */

{

	ulint	i;

	table_cache->rows_used = 0;

	table_cache->rows_allocd = 0;

	table_cache->row_size = row_size;

	for (i = 0; i < MEM_CHUNKS_IN_TABLE_CACHE; i++) {

		/* the memory is actually allocated in

		table_cache_create_empty_row() */

		table_cache->chunks[i].base = NULL;

	}

}

/*******************************************************************//**

Frees a table cache. */

static

void

table_cache_free(

/*=============*/

	i_s_table_cache_t*	table_cache)	/*!< in/out: table cache */

{

	ulint	i;

	for (i = 0; i < MEM_CHUNKS_IN_TABLE_CACHE; i++) {

		/* the memory is actually allocated in

		table_cache_create_empty_row() */

		if (table_cache->chunks[i].base) {

			mem_free(table_cache->chunks[i].base);

			table_cache->chunks[i].base = NULL;

		}

	}

}

/*******************************************************************//**

Returns an empty row from a table cache. The row is allocated if no more

empty rows are available. The number of used rows is incremented.

If the memory limit is hit then NULL is returned and nothing is

allocated.

@return	empty row, or NULL if out of memory */

static

void*

table_cache_create_empty_row(

/*=========================*/

	i_s_table_cache_t*	table_cache,	/*!< in/out: table cache */

	trx_i_s_cache_t*	cache)		/*!< in/out: cache to record

						how many bytes are

						allocated */

{

	ulint	i;

	void*	row;

	ut_a(table_cache->rows_used <= table_cache->rows_allocd);

	if (table_cache->rows_used == table_cache->rows_allocd) {

		/* rows_used == rows_allocd means that new chunk needs

		to be allocated: either no more empty rows in the

		last allocated chunk or nothing has been allocated yet

		(rows_num == rows_allocd == 0); */

		i_s_mem_chunk_t*	chunk;

		ulint			req_bytes;

		ulint			got_bytes;

		ulint			req_rows;

		ulint			got_rows;

		/* find the first not allocated chunk */

		for (i = 0; i < MEM_CHUNKS_IN_TABLE_CACHE; i++) {

			if (table_cache->chunks[i].base == NULL) {

				break;

			}

		}

		/* i == MEM_CHUNKS_IN_TABLE_CACHE means that all chunks

		have been allocated :-X */

		ut_a(i < MEM_CHUNKS_IN_TABLE_CACHE);

		/* allocate the chunk we just found */

		if (i == 0) {

			/* first chunk, nothing is allocated yet */

			req_rows = TABLE_CACHE_INITIAL_ROWSNUM;

		} else {

			/* Memory is increased by the formula

			new = old + old / 2; We are trying not to be

			aggressive here (= using the common new = old * 2)

			because the allocated memory will not be freed

			until InnoDB exit (it is reused). So it is better

			to once allocate the memory in more steps, but

			have less unused/wasted memory than to use less

			steps in allocation (which is done once in a

			lifetime) but end up with lots of unused/wasted

			memory. */

			req_rows = table_cache->rows_allocd / 2;

		}

		req_bytes = req_rows * table_cache->row_size;

		if (req_bytes > MAX_ALLOWED_FOR_ALLOC(cache)) {

			return(NULL);

		}

		chunk = &table_cache->chunks[i];

		chunk->base = mem_alloc2(req_bytes, &got_bytes);

		got_rows = got_bytes / table_cache->row_size;

		cache->mem_allocd += got_bytes;

#if 0

		printf("allocating chunk %d req bytes=%lu, got bytes=%lu, "

		       "row size=%lu, "

		       "req rows=%lu, got rows=%lu\n",

		       i, req_bytes, got_bytes,

		       table_cache->row_size,

		       req_rows, got_rows);

#endif

		chunk->rows_allocd = got_rows;

		table_cache->rows_allocd += got_rows;

		/* adjust the offset of the next chunk */

		if (i < MEM_CHUNKS_IN_TABLE_CACHE - 1) {

			table_cache->chunks[i + 1].offset

				= chunk->offset + chunk->rows_allocd;

		}

		/* return the first empty row in the newly allocated

		chunk */

		row = chunk->base;

	} else {

		char*	chunk_start;

		ulint	offset;

		/* there is an empty row, no need to allocate new

		chunks */

		/* find the first chunk that contains allocated but

		empty/unused rows */

		for (i = 0; i < MEM_CHUNKS_IN_TABLE_CACHE; i++) {

			if (table_cache->chunks[i].offset

			    + table_cache->chunks[i].rows_allocd

			    > table_cache->rows_used) {

				break;

			}

		}

		/* i == MEM_CHUNKS_IN_TABLE_CACHE means that all chunks

		are full, but

		table_cache->rows_used != table_cache->rows_allocd means

		exactly the opposite - there are allocated but

		empty/unused rows :-X */

		ut_a(i < MEM_CHUNKS_IN_TABLE_CACHE);

		chunk_start = (char*) table_cache->chunks[i].base;

		offset = table_cache->rows_used

			- table_cache->chunks[i].offset;

		row = chunk_start + offset * table_cache->row_size;

	}

	table_cache->rows_used++;

	return(row);

}

#ifdef UNIV_DEBUG

/*******************************************************************//**

Validates a row in the locks cache.

@return	TRUE if valid */

static

ibool

i_s_locks_row_validate(

/*===================*/

	const i_s_locks_row_t*	row)	/*!< in: row to validate */

{

	ut_ad(row->lock_trx_id != 0);

	ut_ad(row->lock_mode != NULL);

	ut_ad(row->lock_type != NULL);

	ut_ad(row->lock_table != NULL);

	ut_ad(row->lock_table_id != 0);

	if (row->lock_space == ULINT_UNDEFINED) {

		/* table lock */

		ut_ad(!strcmp("TABLE", row->lock_type));

		ut_ad(row->lock_index == NULL);

		ut_ad(row->lock_data == NULL);

		ut_ad(row->lock_page == ULINT_UNDEFINED);

		ut_ad(row->lock_rec == ULINT_UNDEFINED);

	} else {

		/* record lock */

		ut_ad(!strcmp("RECORD", row->lock_type));

		ut_ad(row->lock_index != NULL);

		ut_ad(row->lock_data != NULL);

		ut_ad(row->lock_page != ULINT_UNDEFINED);

		ut_ad(row->lock_rec != ULINT_UNDEFINED);

	}

	return(TRUE);

}

#endif /* UNIV_DEBUG */

/*******************************************************************//**

Fills i_s_trx_row_t object.

If memory can not be allocated then FALSE is returned.

@return	FALSE if allocation fails */

static

ibool

fill_trx_row(

/*=========*/

	i_s_trx_row_t*		row,		/*!< out: result object

						that's filled */

	const trx_t*		trx,		/*!< in: transaction to

						get data from */

	const i_s_locks_row_t*	requested_lock_row,/*!< in: pointer to the

						corresponding row in

						innodb_locks if trx is

						waiting or NULL if trx

						is not waiting */

	trx_i_s_cache_t*	cache)		/*!< in/out: cache into

						which to copy volatile

						strings */

{

	const char*	stmt;

	size_t		stmt_len;

	const char*	s;

	ut_ad(mutex_own(&kernel_mutex));

	row->trx_id = trx->id;

	row->trx_started = (ib_time_t) trx->start_time;

	row->trx_state = trx_get_que_state_str(trx);

	row->requested_lock_row = requested_lock_row;

	ut_ad(requested_lock_row == NULL

	      || i_s_locks_row_validate(requested_lock_row));

	if (trx->wait_lock != NULL) {

		ut_a(requested_lock_row != NULL);

		row->trx_wait_started = (ib_time_t) trx->wait_started;

	} else {

		ut_a(requested_lock_row == NULL);

		row->trx_wait_started = 0;

	}

	row->trx_weight = (ullint) TRX_WEIGHT(trx);

	if (trx->mysql_thd == NULL) {

		/* For internal transactions e.g., purge and transactions

		being recovered at startup there is no associated MySQL

		thread data structure. */

		row->trx_mysql_thread_id = 0;

		row->trx_query = NULL;

		goto thd_done;

	}

	row->trx_mysql_thread_id = session_get_thread_id(trx->mysql_thd);

	stmt = innobase_get_stmt(trx->mysql_thd, &stmt_len);

	if (stmt != NULL) {

		char	query[TRX_I_S_TRX_QUERY_MAX_LEN + 1];

		if (stmt_len > TRX_I_S_TRX_QUERY_MAX_LEN) {

			stmt_len = TRX_I_S_TRX_QUERY_MAX_LEN;

		}

		memcpy(query, stmt, stmt_len);

		query[stmt_len] = '\0';

		row->trx_query = ha_storage_put_memlim(

			cache->storage, stmt, stmt_len + 1,

			MAX_ALLOWED_FOR_STORAGE(cache));

		if (row->trx_query == NULL) {

			return(FALSE);

		}

	} else {

		row->trx_query = NULL;

	}

thd_done:

	s = trx->op_info;

	if (s != NULL && s[0] != '\0') {

		TRX_I_S_STRING_COPY(s, row->trx_operation_state,

				    TRX_I_S_TRX_OP_STATE_MAX_LEN, cache);

		if (row->trx_operation_state == NULL) {

			return(FALSE);

		}

	} else {

		row->trx_operation_state = NULL;

	}

//	row->trx_tables_in_use = trx->n_mysql_tables_in_use;

	row->trx_tables_locked = trx->mysql_n_tables_locked;

	row->trx_lock_structs = UT_LIST_GET_LEN(trx->trx_locks);

	row->trx_lock_memory_bytes = mem_heap_get_size(trx->lock_heap);

	row->trx_rows_locked = lock_number_of_rows_locked(trx);

	row->trx_rows_modified = trx->undo_no;

	row->trx_concurrency_tickets = trx->n_tickets_to_enter_innodb;

	switch (trx->isolation_level) {

	case TRX_ISO_READ_UNCOMMITTED:

		row->trx_isolation_level = "READ UNCOMMITTED";

		break;

	case TRX_ISO_READ_COMMITTED:

		row->trx_isolation_level = "READ COMMITTED";

		break;

	case TRX_ISO_REPEATABLE_READ:

		row->trx_isolation_level = "REPEATABLE READ";

		break;

	case TRX_ISO_SERIALIZABLE:

		row->trx_isolation_level = "SERIALIZABLE";

		break;

	/* Should not happen as TRX_ISO_READ_COMMITTED is default */

	default:

		row->trx_isolation_level = "UNKNOWN";

	}

	row->trx_unique_checks = (ibool) trx->check_unique_secondary;

	row->trx_foreign_key_checks = (ibool) trx->check_foreigns;

	s = trx->detailed_error;

	if (s != NULL && s[0] != '\0') {

		TRX_I_S_STRING_COPY(s,

				    row->trx_foreign_key_error,

				    TRX_I_S_TRX_FK_ERROR_MAX_LEN, cache);

		if (row->trx_foreign_key_error == NULL) {

			return(FALSE);

		}

	} else {

		row->trx_foreign_key_error = NULL;

	}

	row->trx_has_search_latch = (ibool) trx->has_search_latch;

	row->trx_search_latch_timeout = trx->search_latch_timeout;

	return(TRUE);

}

/*******************************************************************//**

Format the nth field of "rec" and put it in "buf". The result is always

NUL-terminated. Returns the number of bytes that were written to "buf"

(including the terminating NUL).

@return	end of the result */

static

ulint

put_nth_field(

/*==========*/

	char*			buf,	/*!< out: buffer */

	ulint			buf_size,/*!< in: buffer size in bytes */

	ulint			n,	/*!< in: number of field */

	const dict_index_t*	index,	/*!< in: index */

	const rec_t*		rec,	/*!< in: record */

	const ulint*		offsets)/*!< in: record offsets, returned

					by rec_get_offsets() */

{

	const byte*	data;

	ulint		data_len;

	dict_field_t*	dict_field;

	ulint		ret;

	ut_ad(rec_offs_validate(rec, NULL, offsets));

	if (buf_size == 0) {

		return(0);

	}

	ret = 0;

	if (n > 0) {

		/* we must append ", " before the actual data */

		if (buf_size < 3) {

			buf[0] = '\0';

			return(1);

		}

		memcpy(buf, ", ", 3);

		buf += 2;

		buf_size -= 2;

		ret += 2;

	}

	/* now buf_size >= 1 */

	data = rec_get_nth_field(rec, offsets, n, &data_len);

	dict_field = dict_index_get_nth_field(index, n);

	ret += row_raw_format((const char*) data, data_len,

			      dict_field, buf, buf_size);

	return(ret);

}

/*******************************************************************//**

Fills the "lock_data" member of i_s_locks_row_t object.

If memory can not be allocated then FALSE is returned.

@return	FALSE if allocation fails */

static

ibool

fill_lock_data(

/*===========*/

	const char**		lock_data,/*!< out: "lock_data" to fill */

	const lock_t*		lock,	/*!< in: lock used to find the data */

	ulint			heap_no,/*!< in: rec num used to find the data */

	trx_i_s_cache_t*	cache)	/*!< in/out: cache where to store

					volatile data */

{

	mtr_t			mtr;

	const buf_block_t*	block;

	const page_t*		page;

	const rec_t*		rec;

	ut_a(lock_get_type(lock) == LOCK_REC);

	mtr_start(&mtr);

	block = buf_page_try_get(lock_rec_get_space_id(lock),

				 lock_rec_get_page_no(lock),

				 &mtr);

	if (block == NULL) {

		*lock_data = NULL;

		mtr_commit(&mtr);

		return(TRUE);

	}

	page = (const page_t*) buf_block_get_frame(block);

	rec = page_find_rec_with_heap_no(page, heap_no);

	if (page_rec_is_infimum(rec)) {

		*lock_data = ha_storage_put_str_memlim(

			cache->storage, "infimum pseudo-record",

			MAX_ALLOWED_FOR_STORAGE(cache));

	} else if (page_rec_is_supremum(rec)) {

		*lock_data = ha_storage_put_str_memlim(

			cache->storage, "supremum pseudo-record",

			MAX_ALLOWED_FOR_STORAGE(cache));

	} else {

		const dict_index_t*	index;

		ulint			n_fields;

		mem_heap_t*		heap;

		ulint			offsets_onstack[REC_OFFS_NORMAL_SIZE];

		ulint*			offsets;

		char			buf[TRX_I_S_LOCK_DATA_MAX_LEN];

		ulint			buf_used;

		ulint			i;

		rec_offs_init(offsets_onstack);

		offsets = offsets_onstack;

		index = lock_rec_get_index(lock);

		n_fields = dict_index_get_n_unique(index);

		ut_a(n_fields > 0);

		heap = NULL;

		offsets = rec_get_offsets(rec, index, offsets, n_fields,

					  &heap);

		/* format and store the data */

		buf_used = 0;

		for (i = 0; i < n_fields; i++) {

			buf_used += put_nth_field(

				buf + buf_used, sizeof(buf) - buf_used,

				i, index, rec, offsets) - 1;

		}

		*lock_data = (const char*) ha_storage_put_memlim(

			cache->storage, buf, buf_used + 1,

			MAX_ALLOWED_FOR_STORAGE(cache));

		if (UNIV_UNLIKELY(heap != NULL)) {

			/* this means that rec_get_offsets() has created a new

			heap and has stored offsets in it; check that this is

			really the case and free the heap */

			ut_a(offsets != offsets_onstack);

			mem_heap_free(heap);

		}

	}

	mtr_commit(&mtr);

	if (*lock_data == NULL) {

		return(FALSE);

	}

	return(TRUE);

}

/*******************************************************************//**

Fills i_s_locks_row_t object. Returns its first argument.

If memory can not be allocated then FALSE is returned.

@return	FALSE if allocation fails */

static

ibool

fill_locks_row(

/*===========*/

	i_s_locks_row_t* row,	/*!< out: result object that's filled */

	const lock_t*	lock,	/*!< in: lock to get data from */

	ulint		heap_no,/*!< in: lock's record number

				or ULINT_UNDEFINED if the lock

				is a table lock */

	trx_i_s_cache_t* cache)	/*!< in/out: cache into which to copy

				volatile strings */

{

	row->lock_trx_id = lock_get_trx_id(lock);

	row->lock_mode = lock_get_mode_str(lock);

	row->lock_type = lock_get_type_str(lock);

	row->lock_table = ha_storage_put_str_memlim(

		cache->storage, lock_get_table_name(lock),

		MAX_ALLOWED_FOR_STORAGE(cache));

	/* memory could not be allocated */

	if (row->lock_table == NULL) {

		return(FALSE);

	}

	switch (lock_get_type(lock)) {

	case LOCK_REC:

		row->lock_index = ha_storage_put_str_memlim(

			cache->storage, lock_rec_get_index_name(lock),

			MAX_ALLOWED_FOR_STORAGE(cache));

		/* memory could not be allocated */

		if (row->lock_index == NULL) {