mirror/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2025-01-19 21:42:35 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

ndb - bug#21948 & bug#17605

Browse source 
fix alloc/free extent in undo log
  allow extent to be reused once a lcp is finished (instead of when next lcp starts)
This commit is contained in:
jonas@perch.ndb.mysql.com 2006-12-07 15:25:02 +01:00
parent 4456144c58
commit 3ab75d971d
12 changed files with 577 additions and 163 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
storage/ndb/include/kernel/signaldata/Extent.hpp
View file

@ -74,6 +74,8 @@ struct FreeExtentReq {
Local_key key; Local_key key;
Uint32 table_id; Uint32 table_id;
Uint32 tablespace_id; Uint32 tablespace_id;
Uint32 lsn_hi;
Uint32 lsn_lo;
} request; } request;
struct struct
{ {

7
storage/ndb/src/kernel/blocks/dblqh/DblqhMain.cpp
View file

@ -11276,7 +11276,7 @@ void Dblqh::execLCP_PREPARE_REF(Signal* signal)
/** /**
* First fragment mean that last LCP is complete :-) * First fragment mean that last LCP is complete :-)
*/ */
EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length()); EXECUTE_DIRECT(TSMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
jamEntry(); jamEntry();
} }
@ -11327,7 +11327,7 @@ void Dblqh::execLCP_PREPARE_CONF(Signal* signal)
/** /**
* First fragment mean that last LCP is complete :-) * First fragment mean that last LCP is complete :-)
*/ */
EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, signal->length()); EXECUTE_DIRECT(TSMAN, GSN_LCP_FRAG_ORD, signal, signal->length());
jamEntry(); jamEntry();
} }
@ -11611,6 +11611,9 @@ void Dblqh::completeLcpRoundLab(Signal* signal, Uint32 lcpId)
sendSignal(LGMAN_REF, GSN_END_LCP_REQ, signal, sendSignal(LGMAN_REF, GSN_END_LCP_REQ, signal,
EndLcpReq::SignalLength, JBB); EndLcpReq::SignalLength, JBB);
EXECUTE_DIRECT(TSMAN, GSN_END_LCP_REQ, signal, EndLcpReq::SignalLength);
jamEntry();
lcpPtr.i = 0; lcpPtr.i = 0;
ptrAss(lcpPtr, lcpRecord); ptrAss(lcpPtr, lcpRecord);
lcpPtr.p->m_outstanding = 3; lcpPtr.p->m_outstanding = 3;

45
storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp
View file

@ -625,7 +625,8 @@ struct Fragrecord {
DLList<ScanOp>::Head m_scanList; DLList<ScanOp>::Head m_scanList;
bool m_undo_complete; enum { UC_LCP = 1, UC_CREATE = 2 };
Uint32 m_undo_complete;
Uint32 m_tablespace_id; Uint32 m_tablespace_id;
Uint32 m_logfile_group_id; Uint32 m_logfile_group_id;
Disk_alloc_info m_disk_alloc_info; Disk_alloc_info m_disk_alloc_info;
@ -989,6 +990,9 @@ ArrayPool<TupTriggerData> c_triggerPool;
,UNDO_UPDATE = File_formats::Undofile::UNDO_TUP_UPDATE ,UNDO_UPDATE = File_formats::Undofile::UNDO_TUP_UPDATE
,UNDO_FREE = File_formats::Undofile::UNDO_TUP_FREE ,UNDO_FREE = File_formats::Undofile::UNDO_TUP_FREE
,UNDO_CREATE = File_formats::Undofile::UNDO_TUP_CREATE ,UNDO_CREATE = File_formats::Undofile::UNDO_TUP_CREATE
,UNDO_DROP = File_formats::Undofile::UNDO_TUP_DROP
,UNDO_ALLOC_EXTENT = File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT
,UNDO_FREE_EXTENT = File_formats::Undofile::UNDO_TUP_FREE_EXTENT
}; };
struct Alloc struct Alloc
@ -1021,6 +1025,30 @@ ArrayPool<TupTriggerData> c_triggerPool;
Uint32 m_table; Uint32 m_table;
Uint32 m_type_length; // 16 bit type, 16 bit length Uint32 m_type_length; // 16 bit type, 16 bit length
}; };
struct Drop
{
Uint32 m_table;
Uint32 m_type_length; // 16 bit type, 16 bit length
};
struct AllocExtent
{
Uint32 m_table;
Uint32 m_fragment;
Uint32 m_page_no;
Uint32 m_file_no;
Uint32 m_type_length;
};
struct FreeExtent
{
Uint32 m_table;
Uint32 m_fragment;
Uint32 m_page_no;
Uint32 m_file_no;
Uint32 m_type_length;
};
}; };
Extent_info_pool c_extent_pool; Extent_info_pool c_extent_pool;
@ -1420,7 +1448,7 @@ public:
int nr_delete(Signal*, Uint32, Uint32 fragPtr, const Local_key*, Uint32 gci); int nr_delete(Signal*, Uint32, Uint32 fragPtr, const Local_key*, Uint32 gci);
void nr_delete_page_callback(Signal*, Uint32 op, Uint32 page); void nr_delete_page_callback(Signal*, Uint32 op, Uint32 page);
void nr_delete_logbuffer_callback(Signal*, Uint32 op, Uint32 page); void nr_delete_log_buffer_callback(Signal*, Uint32 op, Uint32 page);
private: private:
BLOCK_DEFINES(Dbtup); BLOCK_DEFINES(Dbtup);
@ -2345,9 +2373,10 @@ private:
Uint32 fragId); Uint32 fragId);
void releaseFragment(Signal* signal, Uint32 tableId); void releaseFragment(Signal* signal, Uint32 tableId, Uint32);
void drop_fragment_free_var_pages(Signal*); void drop_fragment_free_var_pages(Signal*);
void drop_fragment_free_exent(Signal*, TablerecPtr, FragrecordPtr, Uint32); void drop_fragment_free_extent(Signal*, TablerecPtr, FragrecordPtr, Uint32);
void drop_fragment_free_extent_log_buffer_callback(Signal*, Uint32, Uint32);
void drop_fragment_unmap_pages(Signal*, TablerecPtr, FragrecordPtr, Uint32); void drop_fragment_unmap_pages(Signal*, TablerecPtr, FragrecordPtr, Uint32);
void drop_fragment_unmap_page_callback(Signal* signal, Uint32, Uint32); void drop_fragment_unmap_page_callback(Signal* signal, Uint32, Uint32);
@ -2631,6 +2660,9 @@ private:
void disk_page_commit_callback(Signal*, Uint32 opPtrI, Uint32 page_id); void disk_page_commit_callback(Signal*, Uint32 opPtrI, Uint32 page_id);
void disk_page_log_buffer_callback(Signal*, Uint32 opPtrI, Uint32); void disk_page_log_buffer_callback(Signal*, Uint32 opPtrI, Uint32);
void disk_page_alloc_extent_log_buffer_callback(Signal*, Uint32, Uint32);
void disk_page_free_extent_log_buffer_callback(Signal*, Uint32, Uint32);
Uint64 disk_page_undo_alloc(Page*, const Local_key*, Uint64 disk_page_undo_alloc(Page*, const Local_key*,
Uint32 sz, Uint32 gci, Uint32 logfile_group_id); Uint32 sz, Uint32 gci, Uint32 logfile_group_id);
@ -2646,6 +2678,9 @@ private:
void undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused); void undo_createtable_callback(Signal* signal, Uint32 opPtrI, Uint32 unused);
void undo_createtable_logsync_callback(Signal* signal, Uint32, Uint32); void undo_createtable_logsync_callback(Signal* signal, Uint32, Uint32);
void drop_table_log_buffer_callback(Signal*, Uint32, Uint32);
void drop_table_logsync_callback(Signal*, Uint32, Uint32);
void disk_page_set_dirty(Ptr<Page>); void disk_page_set_dirty(Ptr<Page>);
void restart_setup_page(Disk_alloc_info&, Ptr<Page>); void restart_setup_page(Disk_alloc_info&, Ptr<Page>);
void update_extent_pos(Disk_alloc_info&, Ptr<Extent_info>); void update_extent_pos(Disk_alloc_info&, Ptr<Extent_info>);
@ -2679,7 +2714,7 @@ public:
private: private:
void disk_restart_undo_next(Signal*); void disk_restart_undo_next(Signal*);
void disk_restart_undo_lcp(Uint32, Uint32); void disk_restart_undo_lcp(Uint32, Uint32, Uint32 flag);
void disk_restart_undo_callback(Signal* signal, Uint32, Uint32); void disk_restart_undo_callback(Signal* signal, Uint32, Uint32);
void disk_restart_undo_alloc(Apply_undo*); void disk_restart_undo_alloc(Apply_undo*);
void disk_restart_undo_update(Apply_undo*); void disk_restart_undo_update(Apply_undo*);

156
storage/ndb/src/kernel/blocks/dbtup/DbtupDiskAlloc.cpp
View file

@ -68,6 +68,26 @@ operator<<(NdbOut& out, const Ptr<Dbtup::Extent_info> & ptr)
return out; return out;
} }
#if NOT_YET_FREE_EXTENT
static
inline
bool
check_free(const Dbtup::Extent_info* extP)
{
Uint32 res = 0;
for (Uint32 i = 1; i<MAX_FREE_LIST; i++)
res += extP->m_free_page_count[i];
return res;
}
#error "Code for deallocting extents when they get empty"
#error "This code is not yet complete"
#endif
#if NOT_YET_UNDO_ALLOC_EXTENT
#error "This is needed for deallocting extents when they get empty"
#error "This code is not complete yet"
#endif
void void
Dbtup::dump_disk_alloc(Dbtup::Disk_alloc_info & alloc) Dbtup::dump_disk_alloc(Dbtup::Disk_alloc_info & alloc)
{ {
@ -441,23 +461,71 @@ Dbtup::disk_page_prealloc(Signal* signal,
/** /**
* We need to alloc an extent * We need to alloc an extent
*/ */
#if NOT_YET_UNDO_ALLOC_EXTENT
Uint32 logfile_group_id = fragPtr.p->m_logfile_group_id;
err = c_lgman->alloc_log_space(logfile_group_id,
sizeof(Disk_undo::AllocExtent)>>2);
if(unlikely(err))
{
return -err;
}
#endif
if (!c_extent_pool.seize(ext)) if (!c_extent_pool.seize(ext))
{ {
jam();
//XXX //XXX
err= 2; err= 2;
#if NOT_YET_UNDO_ALLOC_EXTENT
c_lgman->free_log_space(logfile_group_id,
sizeof(Disk_undo::AllocExtent)>>2);
#endif
c_page_request_pool.release(req); c_page_request_pool.release(req);
ndbout_c("no free extent info"); ndbout_c("no free extent info");
return -err; return -err;
} }
if ((err= tsman.alloc_extent(&ext.p->m_key)) < 0) if ((err= tsman.alloc_extent(&ext.p->m_key)) < 0)
{ {
jam();
#if NOT_YET_UNDO_ALLOC_EXTENT
c_lgman->free_log_space(logfile_group_id,
sizeof(Disk_undo::AllocExtent)>>2);
#endif
c_extent_pool.release(ext); c_extent_pool.release(ext);
c_page_request_pool.release(req); c_page_request_pool.release(req);
return err; return err;
} }
int pages= err; int pages= err;
#if NOT_YET_UNDO_ALLOC_EXTENT
{
/**
* Do something here
*/
{
Callback cb;
cb.m_callbackData= ext.i;
cb.m_callbackFunction =
safe_cast(&Dbtup::disk_page_alloc_extent_log_buffer_callback);
Uint32 sz= sizeof(Disk_undo::AllocExtent)>>2;
Logfile_client lgman(this, c_lgman, logfile_group_id);
int res= lgman.get_log_buffer(signal, sz, &cb);
switch(res){
case 0:
break;
case -1:
ndbrequire("NOT YET IMPLEMENTED" == 0);
break;
default:
execute(signal, cb, res);
}
}
}
#endif
ndbout << "allocated " << pages << " pages: " << ext.p->m_key << endl; ndbout << "allocated " << pages << " pages: " << ext.p->m_key << endl;
ext.p->m_first_page_no = ext.p->m_key.m_page_no; ext.p->m_first_page_no = ext.p->m_key.m_page_no;
bzero(ext.p->m_free_page_count, sizeof(ext.p->m_free_page_count)); bzero(ext.p->m_free_page_count, sizeof(ext.p->m_free_page_count));
@ -1007,6 +1075,12 @@ Dbtup::disk_page_free(Signal *signal,
extentPtr.p->m_free_space += sz; extentPtr.p->m_free_space += sz;
update_extent_pos(alloc, extentPtr); update_extent_pos(alloc, extentPtr);
#if NOT_YET_FREE_EXTENT
if (check_free(extentPtr.p) == 0)
{
ndbout_c("free: extent is free");
}
#endif
} }
void void
@ -1104,14 +1178,56 @@ Dbtup::disk_page_abort_prealloc_callback_1(Signal* signal,
extentPtr.p->m_free_space += sz; extentPtr.p->m_free_space += sz;
update_extent_pos(alloc, extentPtr); update_extent_pos(alloc, extentPtr);
#if NOT_YET_FREE_EXTENT
if (check_free(extentPtr.p) == 0)
{
ndbout_c("abort: extent is free");
}
#endif
} }
#if NOT_YET_UNDO_ALLOC_EXTENT
void
Dbtup::disk_page_alloc_extent_log_buffer_callback(Signal* signal,
Uint32 extentPtrI,
Uint32 unused)
{
Ptr<Extent_info> extentPtr;
c_extent_pool.getPtr(extentPtr, extentPtrI);
Local_key key = extentPtr.p->m_key;
Tablespace_client2 tsman(signal, c_tsman, &key);
Ptr<Tablerec> tabPtr;
tabPtr.i= tsman.m_table_id;
ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
Ptr<Fragrecord> fragPtr;
getFragmentrec(fragPtr, tsman.m_fragment_id, tabPtr.p);
Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
Disk_undo::AllocExtent alloc;
alloc.m_table = tabPtr.i;
alloc.m_fragment = tsman.m_fragment_id;
alloc.m_page_no = key.m_page_no;
alloc.m_file_no = key.m_file_no;
alloc.m_type_length = (Disk_undo::UNDO_ALLOC_EXTENT<<16)|(sizeof(alloc)>> 2);
Logfile_client::Change c[1] = {{ &alloc, sizeof(alloc) >> 2 } };
Uint64 lsn= lgman.add_entry(c, 1);
tsman.update_lsn(&key, lsn);
}
#endif
Uint64 Uint64
Dbtup::disk_page_undo_alloc(Page* page, const Local_key* key, Dbtup::disk_page_undo_alloc(Page* page, const Local_key* key,
Uint32 sz, Uint32 gci, Uint32 logfile_group_id) Uint32 sz, Uint32 gci, Uint32 logfile_group_id)
{ {
Logfile_client lsman(this, c_lgman, logfile_group_id); Logfile_client lgman(this, c_lgman, logfile_group_id);
Disk_undo::Alloc alloc; Disk_undo::Alloc alloc;
alloc.m_type_length= (Disk_undo::UNDO_ALLOC << 16) | (sizeof(alloc) >> 2); alloc.m_type_length= (Disk_undo::UNDO_ALLOC << 16) | (sizeof(alloc) >> 2);
alloc.m_page_no = key->m_page_no; alloc.m_page_no = key->m_page_no;
@ -1119,7 +1235,7 @@ Dbtup::disk_page_undo_alloc(Page* page, const Local_key* key,
Logfile_client::Change c[1] = {{ &alloc, sizeof(alloc) >> 2 } }; Logfile_client::Change c[1] = {{ &alloc, sizeof(alloc) >> 2 } };
Uint64 lsn= lsman.add_entry(c, 1); Uint64 lsn= lgman.add_entry(c, 1);
m_pgman.update_lsn(* key, lsn); m_pgman.update_lsn(* key, lsn);
return lsn; return lsn;
@ -1130,7 +1246,7 @@ Dbtup::disk_page_undo_update(Page* page, const Local_key* key,
const Uint32* src, Uint32 sz, const Uint32* src, Uint32 sz,
Uint32 gci, Uint32 logfile_group_id) Uint32 gci, Uint32 logfile_group_id)
{ {
Logfile_client lsman(this, c_lgman, logfile_group_id); Logfile_client lgman(this, c_lgman, logfile_group_id);
Disk_undo::Update update; Disk_undo::Update update;
update.m_page_no = key->m_page_no; update.m_page_no = key->m_page_no;
@ -1148,7 +1264,7 @@ Dbtup::disk_page_undo_update(Page* page, const Local_key* key,
ndbassert(4*(3 + sz + 1) == (sizeof(update) + 4*sz - 4)); ndbassert(4*(3 + sz + 1) == (sizeof(update) + 4*sz - 4));
Uint64 lsn= lsman.add_entry(c, 3); Uint64 lsn= lgman.add_entry(c, 3);
m_pgman.update_lsn(* key, lsn); m_pgman.update_lsn(* key, lsn);
return lsn; return lsn;
@ -1159,7 +1275,7 @@ Dbtup::disk_page_undo_free(Page* page, const Local_key* key,
const Uint32* src, Uint32 sz, const Uint32* src, Uint32 sz,
Uint32 gci, Uint32 logfile_group_id) Uint32 gci, Uint32 logfile_group_id)
{ {
Logfile_client lsman(this, c_lgman, logfile_group_id); Logfile_client lgman(this, c_lgman, logfile_group_id);
Disk_undo::Free free; Disk_undo::Free free;
free.m_page_no = key->m_page_no; free.m_page_no = key->m_page_no;
@ -1177,7 +1293,7 @@ Dbtup::disk_page_undo_free(Page* page, const Local_key* key,
ndbassert(4*(3 + sz + 1) == (sizeof(free) + 4*sz - 4)); ndbassert(4*(3 + sz + 1) == (sizeof(free) + 4*sz - 4));
Uint64 lsn= lsman.add_entry(c, 3); Uint64 lsn= lgman.add_entry(c, 3);
m_pgman.update_lsn(* key, lsn); m_pgman.update_lsn(* key, lsn);
return lsn; return lsn;
@ -1207,7 +1323,7 @@ Dbtup::disk_restart_undo(Signal* signal, Uint64 lsn,
ndbrequire(len == 3); ndbrequire(len == 3);
Uint32 tableId = ptr[1] >> 16; Uint32 tableId = ptr[1] >> 16;
Uint32 fragId = ptr[1] & 0xFFFF; Uint32 fragId = ptr[1] & 0xFFFF;
disk_restart_undo_lcp(tableId, fragId); disk_restart_undo_lcp(tableId, fragId, Fragrecord::UC_LCP);
disk_restart_undo_next(signal); disk_restart_undo_next(signal);
return; return;
} }
@ -1246,10 +1362,20 @@ Dbtup::disk_restart_undo(Signal* signal, Uint64 lsn,
ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec); ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
for(Uint32 i = 0; i<MAX_FRAG_PER_NODE; i++) for(Uint32 i = 0; i<MAX_FRAG_PER_NODE; i++)
if (tabPtr.p->fragrec[i] != RNIL) if (tabPtr.p->fragrec[i] != RNIL)
disk_restart_undo_lcp(tabPtr.i, tabPtr.p->fragid[i]); disk_restart_undo_lcp(tabPtr.i, tabPtr.p->fragid[i],
Fragrecord::UC_CREATE);
disk_restart_undo_next(signal); disk_restart_undo_next(signal);
return; return;
} }
case File_formats::Undofile::UNDO_TUP_DROP:
jam();
case File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT:
jam();
case File_formats::Undofile::UNDO_TUP_FREE_EXTENT:
jam();
disk_restart_undo_next(signal);
return;
case File_formats::Undofile::UNDO_END: case File_formats::Undofile::UNDO_END:
f_undo_done = true; f_undo_done = true;
return; return;
@ -1283,7 +1409,7 @@ Dbtup::disk_restart_undo_next(Signal* signal)
} }
void void
Dbtup::disk_restart_undo_lcp(Uint32 tableId, Uint32 fragId) Dbtup::disk_restart_undo_lcp(Uint32 tableId, Uint32 fragId, Uint32 flag)
{ {
Ptr<Tablerec> tabPtr; Ptr<Tablerec> tabPtr;
tabPtr.i= tableId; tabPtr.i= tableId;
@ -1295,7 +1421,7 @@ Dbtup::disk_restart_undo_lcp(Uint32 tableId, Uint32 fragId)
getFragmentrec(fragPtr, fragId, tabPtr.p); getFragmentrec(fragPtr, fragId, tabPtr.p);
if (!fragPtr.isNull()) if (!fragPtr.isNull())
{ {
fragPtr.p->m_undo_complete = true; fragPtr.p->m_undo_complete |= flag;
} }
} }
} }
@ -1502,6 +1628,12 @@ Dbtup::disk_restart_alloc_extent(Uint32 tableId, Uint32 fragId,
if (tabPtr.p->tableStatus == DEFINED) if (tabPtr.p->tableStatus == DEFINED)
{ {
getFragmentrec(fragPtr, fragId, tabPtr.p); getFragmentrec(fragPtr, fragId, tabPtr.p);
if (fragPtr.p->m_undo_complete & Fragrecord::UC_CREATE)
{
jam();
return -1;
}
if (!fragPtr.isNull()) if (!fragPtr.isNull())
{ {
Disk_alloc_info& alloc= fragPtr.p->m_disk_alloc_info; Disk_alloc_info& alloc= fragPtr.p->m_disk_alloc_info;

6
storage/ndb/src/kernel/blocks/dbtup/DbtupExecQuery.cpp
View file

@ -3129,7 +3129,7 @@ Dbtup::nr_delete(Signal* signal, Uint32 senderData,
disk_page_set_dirty(disk_page); disk_page_set_dirty(disk_page);
preq.m_callback.m_callbackFunction = preq.m_callback.m_callbackFunction =
safe_cast(&Dbtup::nr_delete_logbuffer_callback); safe_cast(&Dbtup::nr_delete_log_buffer_callback);
Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id); Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
res= lgman.get_log_buffer(signal, sz, &preq.m_callback); res= lgman.get_log_buffer(signal, sz, &preq.m_callback);
switch(res){ switch(res){
@ -3182,7 +3182,7 @@ Dbtup::nr_delete_page_callback(Signal* signal,
Callback cb; Callback cb;
cb.m_callbackData = userpointer; cb.m_callbackData = userpointer;
cb.m_callbackFunction = cb.m_callbackFunction =
safe_cast(&Dbtup::nr_delete_logbuffer_callback); safe_cast(&Dbtup::nr_delete_log_buffer_callback);
Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id); Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
int res= lgman.get_log_buffer(signal, sz, &cb); int res= lgman.get_log_buffer(signal, sz, &cb);
switch(res){ switch(res){
@ -3202,7 +3202,7 @@ Dbtup::nr_delete_page_callback(Signal* signal,
} }
void void
Dbtup::nr_delete_logbuffer_callback(Signal* signal, Dbtup::nr_delete_log_buffer_callback(Signal* signal,
Uint32 userpointer, Uint32 userpointer,
Uint32 unused) Uint32 unused)
{ {

4
storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp
View file

@ -164,7 +164,7 @@ void Dbtup::execCONTINUEB(Signal* signal)
break; break;
case ZREL_FRAG: case ZREL_FRAG:
ljam(); ljam();
releaseFragment(signal, dataPtr); releaseFragment(signal, dataPtr, signal->theData[2]);
break; break;
case ZREPORT_MEMORY_USAGE:{ case ZREPORT_MEMORY_USAGE:{
ljam(); ljam();
@ -223,7 +223,7 @@ void Dbtup::execCONTINUEB(Signal* signal)
fragPtr.i= signal->theData[2]; fragPtr.i= signal->theData[2];
ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec); ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord); ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
drop_fragment_free_exent(signal, tabPtr, fragPtr, signal->theData[3]); drop_fragment_free_extent(signal, tabPtr, fragPtr, signal->theData[3]);
return; return;
} }
case ZUNMAP_PAGES: case ZUNMAP_PAGES:

231
storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp
View file

@ -309,6 +309,7 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
if(lastAttr) if(lastAttr)
{ {
jam();
/** /**
* Init Disk_alloc_info * Init Disk_alloc_info
*/ */
@ -320,6 +321,11 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
ndbrequire(tsman.get_tablespace_info(&rep) == 0); ndbrequire(tsman.get_tablespace_info(&rep) == 0);
regFragPtr.p->m_logfile_group_id= rep.tablespace.logfile_group_id; regFragPtr.p->m_logfile_group_id= rep.tablespace.logfile_group_id;
} }
else
{
jam();
regFragPtr.p->m_logfile_group_id = RNIL;
}
new (&regFragPtr.p->m_disk_alloc_info) new (&regFragPtr.p->m_disk_alloc_info)
Disk_alloc_info(regTabPtr.p, rep.tablespace.extent_size); Disk_alloc_info(regTabPtr.p, rep.tablespace.extent_size);
releaseFragoperrec(fragOperPtr); releaseFragoperrec(fragOperPtr);
@ -564,7 +570,12 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
regFragPtr.p->m_tablespace_id); regFragPtr.p->m_tablespace_id);
ndbrequire(tsman.get_tablespace_info(&rep) == 0); ndbrequire(tsman.get_tablespace_info(&rep) == 0);
regFragPtr.p->m_logfile_group_id= rep.tablespace.logfile_group_id; regFragPtr.p->m_logfile_group_id= rep.tablespace.logfile_group_id;
} }
else
{
jam();
regFragPtr.p->m_logfile_group_id = RNIL;
}
new (&regFragPtr.p->m_disk_alloc_info) new (&regFragPtr.p->m_disk_alloc_info)
Disk_alloc_info(regTabPtr.p, rep.tablespace.extent_size); Disk_alloc_info(regTabPtr.p, rep.tablespace.extent_size);
@ -597,7 +608,7 @@ void Dbtup::execTUP_ADD_ATTRREQ(Signal* signal)
ndbrequire("NOT YET IMPLEMENTED" == 0); ndbrequire("NOT YET IMPLEMENTED" == 0);
break; break;
} }
execute(signal, cb, 0); execute(signal, cb, regFragPtr.p->m_logfile_group_id);
return; return;
} }
} }
@ -874,7 +885,8 @@ Dbtup::execDROP_TAB_REQ(Signal* signal)
signal->theData[0]= ZREL_FRAG; signal->theData[0]= ZREL_FRAG;
signal->theData[1]= tabPtr.i; signal->theData[1]= tabPtr.i;
sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB); signal->theData[2]= RNIL;
sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);
} }
void Dbtup::releaseTabDescr(Tablerec* const regTabPtr) void Dbtup::releaseTabDescr(Tablerec* const regTabPtr)
@ -902,7 +914,8 @@ void Dbtup::releaseTabDescr(Tablerec* const regTabPtr)
} }
} }
void Dbtup::releaseFragment(Signal* signal, Uint32 tableId) void Dbtup::releaseFragment(Signal* signal, Uint32 tableId,
Uint32 logfile_group_id)
{ {
TablerecPtr tabPtr; TablerecPtr tabPtr;
tabPtr.i= tableId; tabPtr.i= tableId;
@ -929,16 +942,35 @@ void Dbtup::releaseFragment(Signal* signal, Uint32 tableId)
sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB); sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB);
return; return;
} }
#if NOT_YET_UNDO_DROP_TABLE
#error "This code is complete, but I prefer not to enable it until I need it"
if (logfile_group_id != RNIL)
{
Callback cb;
cb.m_callbackData= tabPtr.i;
cb.m_callbackFunction =
safe_cast(&Dbtup::drop_table_log_buffer_callback);
Uint32 sz= sizeof(Disk_undo::Drop) >> 2;
int r0 = c_lgman->alloc_log_space(logfile_group_id, sz);
Logfile_client lgman(this, c_lgman, logfile_group_id);
int res= lgman.get_log_buffer(signal, sz, &cb);
switch(res){
case 0:
ljam();
return;
case -1:
ndbrequire("NOT YET IMPLEMENTED" == 0);
break;
default:
execute(signal, cb, logfile_group_id);
return;
}
}
#endif
DropTabConf * const dropConf= (DropTabConf *)signal->getDataPtrSend(); drop_table_logsync_callback(signal, tabPtr.i, RNIL);
dropConf->senderRef= reference();
dropConf->senderData= tabPtr.p->m_dropTable.tabUserPtr;
dropConf->tableId= tabPtr.i;
sendSignal(tabPtr.p->m_dropTable.tabUserRef, GSN_DROP_TAB_CONF,
signal, DropTabConf::SignalLength, JBB);
releaseTabDescr(tabPtr.p);
initTab(tabPtr.p);
} }
void void
@ -965,7 +997,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
alloc_info.m_curr_extent_info_ptr_i= RNIL; alloc_info.m_curr_extent_info_ptr_i= RNIL;
} }
drop_fragment_free_exent(signal, tabPtr, fragPtr, 0); drop_fragment_free_extent(signal, tabPtr, fragPtr, 0);
return; return;
} }
@ -998,7 +1030,7 @@ Dbtup::drop_fragment_unmap_pages(Signal *signal,
} }
return; return;
} }
drop_fragment_free_exent(signal, tabPtr, fragPtr, 0); drop_fragment_free_extent(signal, tabPtr, fragPtr, 0);
} }
void void
@ -1031,10 +1063,10 @@ Dbtup::drop_fragment_unmap_page_callback(Signal* signal,
} }
void void
Dbtup::drop_fragment_free_exent(Signal *signal, Dbtup::drop_fragment_free_extent(Signal *signal,
TablerecPtr tabPtr, TablerecPtr tabPtr,
FragrecordPtr fragPtr, FragrecordPtr fragPtr,
Uint32 pos) Uint32 pos)
{ {
if (tabPtr.p->m_no_of_disk_attributes) if (tabPtr.p->m_no_of_disk_attributes)
{ {
@ -1044,25 +1076,32 @@ Dbtup::drop_fragment_free_exent(Signal *signal,
if(!alloc_info.m_free_extents[pos].isEmpty()) if(!alloc_info.m_free_extents[pos].isEmpty())
{ {
jam(); jam();
Local_extent_info_list Callback cb;
list(c_extent_pool, alloc_info.m_free_extents[pos]); cb.m_callbackData= fragPtr.i;
Ptr<Extent_info> ext_ptr; cb.m_callbackFunction =
list.first(ext_ptr); safe_cast(&Dbtup::drop_fragment_free_extent_log_buffer_callback);
#if NOT_YET_UNDO_FREE_EXTENT
Uint32 sz= sizeof(Disk_undo::FreeExtent) >> 2;
int r0 = c_lgman->alloc_log_space(fragPtr.p->m_logfile_group_id, sz);
Tablespace_client tsman(signal, c_tsman, tabPtr.i, Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
fragPtr.p->fragmentId,
fragPtr.p->m_tablespace_id);
tsman.free_extent(&ext_ptr.p->m_key); int res= lgman.get_log_buffer(signal, sz, &cb);
c_extent_hash.remove(ext_ptr); switch(res){
list.release(ext_ptr); case 0:
ljam();
signal->theData[0] = ZFREE_EXTENT; return;
signal->theData[1] = tabPtr.i; case -1:
signal->theData[2] = fragPtr.i; ndbrequire("NOT YET IMPLEMENTED" == 0);
signal->theData[3] = pos; break;
sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB); default:
execute(signal, cb, fragPtr.p->m_logfile_group_id);
return;
}
#else
execute(signal, cb, fragPtr.p->m_logfile_group_id);
return; return;
#endif
} }
} }
@ -1081,6 +1120,123 @@ Dbtup::drop_fragment_free_exent(Signal *signal,
sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB); sendSignal(reference(), GSN_CONTINUEB, signal, 3, JBB);
} }
void
Dbtup::drop_table_log_buffer_callback(Signal* signal, Uint32 tablePtrI,
Uint32 logfile_group_id)
{
TablerecPtr tabPtr;
tabPtr.i = tablePtrI;
ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
ndbrequire(tabPtr.p->m_no_of_disk_attributes);
Disk_undo::Drop drop;
drop.m_table = tabPtr.i;
drop.m_type_length =
(Disk_undo::UNDO_DROP << 16) | (sizeof(drop) >> 2);
Logfile_client lgman(this, c_lgman, logfile_group_id);
Logfile_client::Change c[1] = {{ &drop, sizeof(drop) >> 2 } };
Uint64 lsn = lgman.add_entry(c, 1);
Logfile_client::Request req;
req.m_callback.m_callbackData= tablePtrI;
req.m_callback.m_callbackFunction =
safe_cast(&Dbtup::drop_table_logsync_callback);
int ret = lgman.sync_lsn(signal, lsn, &req, 0);
switch(ret){
case 0:
return;
default:
ndbout_c("ret: %d", ret);
ndbrequire(false);
}
}
void
Dbtup::drop_table_logsync_callback(Signal* signal,
Uint32 tabPtrI,
Uint32 logfile_group_id)
{
TablerecPtr tabPtr;
tabPtr.i = tabPtrI;
ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
DropTabConf * const dropConf= (DropTabConf *)signal->getDataPtrSend();
dropConf->senderRef= reference();
dropConf->senderData= tabPtr.p->m_dropTable.tabUserPtr;
dropConf->tableId= tabPtr.i;
sendSignal(tabPtr.p->m_dropTable.tabUserRef, GSN_DROP_TAB_CONF,
signal, DropTabConf::SignalLength, JBB);
releaseTabDescr(tabPtr.p);
initTab(tabPtr.p);
}
void
Dbtup::drop_fragment_free_extent_log_buffer_callback(Signal* signal,
Uint32 fragPtrI,
Uint32 unused)
{
FragrecordPtr fragPtr;
fragPtr.i = fragPtrI;
ptrCheckGuard(fragPtr, cnoOfFragrec, fragrecord);
TablerecPtr tabPtr;
tabPtr.i = fragPtr.p->fragTableId;
ptrCheckGuard(tabPtr, cnoOfTablerec, tablerec);
ndbrequire(tabPtr.p->m_no_of_disk_attributes);
Disk_alloc_info& alloc_info= fragPtr.p->m_disk_alloc_info;
for(Uint32 pos = 0; pos<EXTENT_SEARCH_MATRIX_SIZE; pos++)
{
if(!alloc_info.m_free_extents[pos].isEmpty())
{
jam();
Local_extent_info_list
list(c_extent_pool, alloc_info.m_free_extents[pos]);
Ptr<Extent_info> ext_ptr;
list.first(ext_ptr);
#if NOT_YET_UNDO_FREE_EXTENT
#error "This code is complete"
#error "but not needed until we do dealloc of empty extents"
Disk_undo::FreeExtent free;
free.m_table = tabPtr.i;
free.m_fragment = fragPtr.p->fragmentId;
free.m_file_no = ext_ptr.p->m_key.m_file_no;
free.m_page_no = ext_ptr.p->m_key.m_page_no;
free.m_type_length =
(Disk_undo::UNDO_FREE_EXTENT << 16) | (sizeof(free) >> 2);
Logfile_client lgman(this, c_lgman, fragPtr.p->m_logfile_group_id);
Logfile_client::Change c[1] = {{ &free, sizeof(free) >> 2 } };
Uint64 lsn = lgman.add_entry(c, 1);
#else
Uint64 lsn = 0;
#endif
Tablespace_client tsman(signal, c_tsman, tabPtr.i,
fragPtr.p->fragmentId,
fragPtr.p->m_tablespace_id);
tsman.free_extent(&ext_ptr.p->m_key, lsn);
c_extent_hash.remove(ext_ptr);
list.release(ext_ptr);
signal->theData[0] = ZFREE_EXTENT;
signal->theData[1] = tabPtr.i;
signal->theData[2] = fragPtr.i;
signal->theData[3] = pos;
sendSignal(cownref, GSN_CONTINUEB, signal, 4, JBB);
return;
}
}
ndbrequire(false);
}
void void
Dbtup::drop_fragment_free_var_pages(Signal* signal) Dbtup::drop_fragment_free_var_pages(Signal* signal)
{ {
@ -1112,7 +1268,7 @@ Dbtup::drop_fragment_free_var_pages(Signal* signal)
sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB); sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);
return; return;
} }
Uint32 logfile_group_id = fragPtr.p->m_logfile_group_id ;
releaseFragPages(fragPtr.p); releaseFragPages(fragPtr.p);
Uint32 i; Uint32 i;
for(i= 0; i<MAX_FRAG_PER_NODE; i++) for(i= 0; i<MAX_FRAG_PER_NODE; i++)
@ -1126,7 +1282,8 @@ Dbtup::drop_fragment_free_var_pages(Signal* signal)
signal->theData[0]= ZREL_FRAG; signal->theData[0]= ZREL_FRAG;
signal->theData[1]= tabPtr.i; signal->theData[1]= tabPtr.i;
sendSignal(cownref, GSN_CONTINUEB, signal, 2, JBB); signal->theData[2]= logfile_group_id;
sendSignal(cownref, GSN_CONTINUEB, signal, 3, JBB);
return; return;
} }

5
storage/ndb/src/kernel/blocks/diskpage.hpp
View file

@ -154,7 +154,10 @@ struct File_formats
,UNDO_TUP_UPDATE = 4 ,UNDO_TUP_UPDATE = 4
,UNDO_TUP_FREE = 5 ,UNDO_TUP_FREE = 5
,UNDO_TUP_CREATE = 6 ,UNDO_TUP_CREATE = 6
,UNDO_TUP_DROP = 7
,UNDO_TUP_ALLOC_EXTENT = 8
,UNDO_TUP_FREE_EXTENT = 9
,UNDO_END = 0x7FFF ,UNDO_END = 0x7FFF
,UNDO_NEXT_LSN = 0x8000 ,UNDO_NEXT_LSN = 0x8000
}; };

11
storage/ndb/src/kernel/blocks/lgman.cpp
View file

@ -1160,14 +1160,15 @@ Lgman::process_log_sync_waiters(Signal* signal, Ptr<Logfile_group> ptr)
bool removed= false; bool removed= false;
Ptr<Log_waiter> waiter; Ptr<Log_waiter> waiter;
list.first(waiter); list.first(waiter);
Uint32 logfile_group_id = ptr.p->m_logfile_group_id;
if(waiter.p->m_sync_lsn <= ptr.p->m_last_synced_lsn) if(waiter.p->m_sync_lsn <= ptr.p->m_last_synced_lsn)
{ {
removed= true; removed= true;
Uint32 block = waiter.p->m_block; Uint32 block = waiter.p->m_block;
SimulatedBlock* b = globalData.getBlock(block); SimulatedBlock* b = globalData.getBlock(block);
b->execute(signal, waiter.p->m_callback, 0); b->execute(signal, waiter.p->m_callback, logfile_group_id);
list.releaseFirst(waiter); list.releaseFirst(waiter);
} }
@ -1522,12 +1523,13 @@ Lgman::process_log_buffer_waiters(Signal* signal, Ptr<Logfile_group> ptr)
bool removed= false; bool removed= false;
Ptr<Log_waiter> waiter; Ptr<Log_waiter> waiter;
list.first(waiter); list.first(waiter);
Uint32 logfile_group_id = ptr.p->m_logfile_group_id;
if(waiter.p->m_size + 2*File_formats::UNDO_PAGE_WORDS < free_buffer) if(waiter.p->m_size + 2*File_formats::UNDO_PAGE_WORDS < free_buffer)
{ {
removed= true; removed= true;
Uint32 block = waiter.p->m_block; Uint32 block = waiter.p->m_block;
SimulatedBlock* b = globalData.getBlock(block); SimulatedBlock* b = globalData.getBlock(block);
b->execute(signal, waiter.p->m_callback, 0); b->execute(signal, waiter.p->m_callback, logfile_group_id);
list.releaseFirst(waiter); list.releaseFirst(waiter);
} }
@ -2699,6 +2701,9 @@ Lgman::execute_undo_record(Signal* signal)
case File_formats::Undofile::UNDO_TUP_UPDATE: case File_formats::Undofile::UNDO_TUP_UPDATE:
case File_formats::Undofile::UNDO_TUP_FREE: case File_formats::Undofile::UNDO_TUP_FREE:
case File_formats::Undofile::UNDO_TUP_CREATE: case File_formats::Undofile::UNDO_TUP_CREATE:
case File_formats::Undofile::UNDO_TUP_DROP:
case File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT:
case File_formats::Undofile::UNDO_TUP_FREE_EXTENT:
tup->disk_restart_undo(signal, lsn, mask, ptr - len + 1, len); tup->disk_restart_undo(signal, lsn, mask, ptr - len + 1, len);
return; return;
default: default:

48
storage/ndb/src/kernel/blocks/print_file.cpp
View file

@ -304,6 +304,8 @@ print_undo_page(int count, void* ptr, Uint32 sz){
case File_formats::Undofile::UNDO_LCP: case File_formats::Undofile::UNDO_LCP:
printf("[ %lld LCP %d tab: %d frag: %d ]", lsn, printf("[ %lld LCP %d tab: %d frag: %d ]", lsn,
src[0], src[1] >> 16, src[1] & 0xFFFF); src[0], src[1] >> 16, src[1] & 0xFFFF);
if(g_verbosity <= 3)
printf("\n");
break; break;
case File_formats::Undofile::UNDO_TUP_ALLOC: case File_formats::Undofile::UNDO_TUP_ALLOC:
if(g_verbosity > 3) if(g_verbosity > 3)
@ -340,6 +342,48 @@ print_undo_page(int count, void* ptr, Uint32 sz){
req->m_gci); req->m_gci);
} }
break; break;
case File_formats::Undofile::UNDO_TUP_CREATE:
{
Dbtup::Disk_undo::Create *req = (Dbtup::Disk_undo::Create*)src;
printf("[ %lld Create %d ]", lsn, req->m_table);
if(g_verbosity <= 3)
printf("\n");
break;
}
case File_formats::Undofile::UNDO_TUP_DROP:
{
Dbtup::Disk_undo::Drop *req = (Dbtup::Disk_undo::Drop*)src;
printf("[ %lld Drop %d ]", lsn, req->m_table);
if(g_verbosity <= 3)
printf("\n");
break;
}
case File_formats::Undofile::UNDO_TUP_ALLOC_EXTENT:
{
Dbtup::Disk_undo::AllocExtent *req = (Dbtup::Disk_undo::AllocExtent*)src;
printf("[ %lld AllocExtent tab: %d frag: %d file: %d page: %d ]",
lsn,
req->m_table,
req->m_fragment,
req->m_file_no,
req->m_page_no);
if(g_verbosity <= 3)
printf("\n");
break;
}
case File_formats::Undofile::UNDO_TUP_FREE_EXTENT:
{
Dbtup::Disk_undo::FreeExtent *req = (Dbtup::Disk_undo::FreeExtent*)src;
printf("[ %lld FreeExtent tab: %d frag: %d file: %d page: %d ]",
lsn,
req->m_table,
req->m_fragment,
req->m_file_no,
req->m_page_no);
if(g_verbosity <= 3)
printf("\n");
break;
}
default: default:
ndbout_c("[ Unknown type %d len: %d, pos: %d ]", type, len, pos); ndbout_c("[ Unknown type %d len: %d, pos: %d ]", type, len, pos);
if(!(len && type)) if(!(len && type))
@ -361,11 +405,11 @@ print_undo_page(int count, void* ptr, Uint32 sz){
} }
} }
} }
if(count == g_uf_zero.m_undo_pages + 1) if(count == g_uf_zero.m_undo_pages + 1)
{ {
} }
return 0; return 0;
} }

171
storage/ndb/src/kernel/blocks/tsman.cpp
View file

@ -79,12 +79,14 @@ Tsman::Tsman(Block_context& ctx,
addRecSignal(GSN_START_RECREQ, &Tsman::execSTART_RECREQ); addRecSignal(GSN_START_RECREQ, &Tsman::execSTART_RECREQ);
addRecSignal(GSN_LCP_FRAG_ORD, &Tsman::execLCP_FRAG_ORD);
addRecSignal(GSN_END_LCP_REQ, &Tsman::execEND_LCP_REQ); addRecSignal(GSN_END_LCP_REQ, &Tsman::execEND_LCP_REQ);
addRecSignal(GSN_GET_TABINFOREQ, &Tsman::execGET_TABINFOREQ); addRecSignal(GSN_GET_TABINFOREQ, &Tsman::execGET_TABINFOREQ);
m_tablespace_hash.setSize(10); m_tablespace_hash.setSize(10);
m_file_hash.setSize(10); m_file_hash.setSize(10);
m_lcp_ongoing = false;
} }
Tsman::~Tsman() Tsman::~Tsman()
@ -1101,6 +1103,7 @@ Tsman::load_extent_page_callback(Signal* signal,
ptr.p->m_online.m_lcp_free_extent_tail = RNIL; ptr.p->m_online.m_lcp_free_extent_tail = RNIL;
ptr.p->m_online.m_data_pages = data_pages; ptr.p->m_online.m_data_pages = data_pages;
ptr.p->m_online.m_used_extent_cnt = 0; ptr.p->m_online.m_used_extent_cnt = 0;
ptr.p->m_online.m_extent_headers_per_extent_page = per_page;
Ptr<Tablespace> ts_ptr; Ptr<Tablespace> ts_ptr;
m_tablespace_pool.getPtr(ts_ptr, ptr.p->m_tablespace_ptr_i); m_tablespace_pool.getPtr(ts_ptr, ptr.p->m_tablespace_ptr_i);
@ -1182,9 +1185,8 @@ Tsman::scan_extent_headers(Signal* signal, Ptr<Datafile> ptr)
m_tablespace_pool.getPtr(lg_ptr, ptr.p->m_tablespace_ptr_i); m_tablespace_pool.getPtr(lg_ptr, ptr.p->m_tablespace_ptr_i);
Uint32 firstFree= RNIL; Uint32 firstFree= RNIL;
Uint32 size = lg_ptr.p->m_extent_size; Uint32 size = ptr.p->m_extent_size;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size); Uint32 per_page = ptr.p->m_online.m_extent_headers_per_extent_page;
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE; Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
Uint32 pages= ptr.p->m_online.m_offset_data_pages - 1; Uint32 pages= ptr.p->m_online.m_offset_data_pages - 1;
Uint32 datapages= ptr.p->m_online.m_data_pages; Uint32 datapages= ptr.p->m_online.m_data_pages;
@ -1410,23 +1412,21 @@ Tsman::execALLOC_EXTENT_REQ(Signal* signal)
AllocExtentReq::ErrorCode err; AllocExtentReq::ErrorCode err;
ndbrequire(m_tablespace_hash.find(ts_ptr, req.request.tablespace_id)); ndbrequire(m_tablespace_hash.find(ts_ptr, req.request.tablespace_id));
Uint32 size = ts_ptr.p->m_extent_size;
Local_datafile_list tmp(m_file_pool, ts_ptr.p->m_free_files); Local_datafile_list tmp(m_file_pool, ts_ptr.p->m_free_files);
if (tmp.first(file_ptr)) if (tmp.first(file_ptr))
{ {
Uint32 eh_words = File_formats::Datafile::extent_header_words(size); Uint32 size = file_ptr.p->m_extent_size;
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 extent = file_ptr.p->m_online.m_first_free_extent; Uint32 extent = file_ptr.p->m_online.m_first_free_extent;
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages; Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 page_no = extent / per_page; Uint32 page_no = extent / per_page;
Uint32 extent_no = extent % per_page; Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = page_no; preq.m_page.m_page_no = page_no;
preq.m_page.m_file_no = file_ptr.p->m_file_no; preq.m_page.m_file_no = file_ptr.p->m_file_no;
preq.m_page.m_page_idx = extent;
/** /**
* Handling of unmapped extent header pages is not implemented * Handling of unmapped extent header pages is not implemented
@ -1501,28 +1501,21 @@ void
Tsman::execFREE_EXTENT_REQ(Signal* signal) Tsman::execFREE_EXTENT_REQ(Signal* signal)
{ {
jamEntry(); jamEntry();
Ptr<Tablespace> ts_ptr;
Ptr<Datafile> file_ptr; Ptr<Datafile> file_ptr;
FreeExtentReq req = *(FreeExtentReq*)signal->getDataPtr(); FreeExtentReq req = *(FreeExtentReq*)signal->getDataPtr();
FreeExtentReq::ErrorCode err = (FreeExtentReq::ErrorCode)0; FreeExtentReq::ErrorCode err = (FreeExtentReq::ErrorCode)0;
ndbrequire(m_tablespace_hash.find(ts_ptr, req.request.tablespace_id));
Datafile file_key; Datafile file_key;
file_key.m_file_no = req.request.key.m_file_no; file_key.m_file_no = req.request.key.m_file_no;
ndbrequire(m_file_hash.find(file_ptr, file_key)); ndbrequire(m_file_hash.find(file_ptr, file_key));
Uint32 size = ts_ptr.p->m_extent_size;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size); struct req val = lookup_extent(req.request.key.m_page_no, file_ptr.p);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words; Uint32 extent =
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages; (req.request.key.m_page_no - val.m_extent_pages) / val.m_extent_size +
Uint32 extent = (req.request.key.m_page_no - data_off) / size + per_page; file_ptr.p->m_online.m_extent_headers_per_extent_page;
Uint32 page_no = extent / per_page;
Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = page_no; preq.m_page.m_page_no = val.m_extent_page_no;
preq.m_page.m_file_no = req.request.key.m_file_no; preq.m_page.m_file_no = req.request.key.m_file_no;
ndbout << "Free extent: " << req.request.key << endl; ndbout << "Free extent: " << req.request.key << endl;
@ -1539,16 +1532,38 @@ Tsman::execFREE_EXTENT_REQ(Signal* signal)
File_formats::Datafile::Extent_page* page = File_formats::Datafile::Extent_page* page =
(File_formats::Datafile::Extent_page*)ptr_p; (File_formats::Datafile::Extent_page*)ptr_p;
File_formats::Datafile::Extent_header* header = File_formats::Datafile::Extent_header* header =
page->get_header(extent_no, size); page->get_header(val.m_extent_no, val.m_extent_size);
ndbrequire(header->m_table == req.request.table_id); ndbrequire(header->m_table == req.request.table_id);
header->m_table = RNIL; header->m_table = RNIL;
header->m_next_free_extent= file_ptr.p->m_online.m_lcp_free_extent_head;
if(file_ptr.p->m_online.m_lcp_free_extent_head == RNIL)
file_ptr.p->m_online.m_lcp_free_extent_tail= extent;
file_ptr.p->m_online.m_lcp_free_extent_head= extent;
file_ptr.p->m_online.m_used_extent_cnt--; file_ptr.p->m_online.m_used_extent_cnt--;
if (m_lcp_ongoing)
{
jam();
header->m_next_free_extent= file_ptr.p->m_online.m_lcp_free_extent_head;
if(file_ptr.p->m_online.m_lcp_free_extent_head == RNIL)
file_ptr.p->m_online.m_lcp_free_extent_tail= extent;
file_ptr.p->m_online.m_lcp_free_extent_head= extent;
}
else
{
jam();
header->m_next_free_extent = file_ptr.p->m_online.m_first_free_extent;
if (file_ptr.p->m_online.m_first_free_extent == RNIL)
{
/**
* Move from full to free
*/
Ptr<Tablespace> ptr;
m_tablespace_pool.getPtr(ptr, file_ptr.p->m_tablespace_ptr_i);
Local_datafile_list free(m_file_pool, ptr.p->m_free_files);
Local_datafile_list full(m_file_pool, ptr.p->m_full_files);
full.remove(file_ptr);
free.add(file_ptr);
}
file_ptr.p->m_online.m_first_free_extent = extent;
}
} }
else else
{ {
@ -1583,18 +1598,10 @@ Tsman::update_page_free_bits(Signal* signal,
file_key.m_file_no = key->m_file_no; file_key.m_file_no = key->m_file_no;
ndbrequire(m_file_hash.find(file_ptr, file_key)); ndbrequire(m_file_hash.find(file_ptr, file_key));
Uint32 size = file_ptr.p->m_extent_size; struct req val = lookup_extent(key->m_page_no, file_ptr.p);
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
Uint32 extent = (key->m_page_no - data_off) / size + per_page;
Uint32 page_no = extent / per_page;
Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = page_no; preq.m_page.m_page_no = val.m_extent_page_no;
preq.m_page.m_file_no = key->m_file_no; preq.m_page.m_file_no = key->m_file_no;
/** /**
@ -1609,12 +1616,12 @@ Tsman::update_page_free_bits(Signal* signal,
File_formats::Datafile::Extent_page* page = File_formats::Datafile::Extent_page* page =
(File_formats::Datafile::Extent_page*)ptr_p; (File_formats::Datafile::Extent_page*)ptr_p;
File_formats::Datafile::Extent_header* header = File_formats::Datafile::Extent_header* header =
page->get_header(extent_no, size); page->get_header(val.m_extent_no, val.m_extent_size);
ndbrequire(header->m_table != RNIL); ndbrequire(header->m_table != RNIL);
Uint32 page_no_in_extent = (key->m_page_no - data_off) % size; Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
/** /**
* Toggle word * Toggle word
*/ */
@ -1637,26 +1644,15 @@ Tsman::get_page_free_bits(Signal* signal, Local_key *key,
{ {
jamEntry(); jamEntry();
/**
* XXX make into subroutine
*/
Ptr<Datafile> file_ptr; Ptr<Datafile> file_ptr;
Datafile file_key; Datafile file_key;
file_key.m_file_no = key->m_file_no; file_key.m_file_no = key->m_file_no;
ndbrequire(m_file_hash.find(file_ptr, file_key)); ndbrequire(m_file_hash.find(file_ptr, file_key));
Uint32 size = file_ptr.p->m_extent_size;
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
Uint32 extent = (key->m_page_no - data_off) / size + per_page; struct req val = lookup_extent(key->m_page_no, file_ptr.p);
Uint32 page_no = extent / per_page;
Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = page_no; preq.m_page.m_page_no = val.m_extent_page_no;
preq.m_page.m_file_no = key->m_file_no; preq.m_page.m_file_no = key->m_file_no;
/** /**
@ -1671,11 +1667,11 @@ Tsman::get_page_free_bits(Signal* signal, Local_key *key,
File_formats::Datafile::Extent_page* page = File_formats::Datafile::Extent_page* page =
(File_formats::Datafile::Extent_page*)ptr_p; (File_formats::Datafile::Extent_page*)ptr_p;
File_formats::Datafile::Extent_header* header = File_formats::Datafile::Extent_header* header =
page->get_header(extent_no, size); page->get_header(val.m_extent_no, val.m_extent_size);
ndbrequire(header->m_table != RNIL); ndbrequire(header->m_table != RNIL);
Uint32 page_no_in_extent = (key->m_page_no - data_off) % size; Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
Uint32 bits = header->get_free_bits(page_no_in_extent); Uint32 bits = header->get_free_bits(page_no_in_extent);
*uncommitted = (bits & UNCOMMITTED_MASK) >> UNCOMMITTED_SHIFT; *uncommitted = (bits & UNCOMMITTED_MASK) >> UNCOMMITTED_SHIFT;
*committed = (bits & COMMITTED_MASK); *committed = (bits & COMMITTED_MASK);
@ -1700,19 +1696,11 @@ Tsman::unmap_page(Signal* signal, Local_key *key, Uint32 uncommitted_bits)
Datafile file_key; Datafile file_key;
file_key.m_file_no = key->m_file_no; file_key.m_file_no = key->m_file_no;
ndbrequire(m_file_hash.find(file_ptr, file_key)); ndbrequire(m_file_hash.find(file_ptr, file_key));
Uint32 size = file_ptr.p->m_extent_size; struct req val = lookup_extent(key->m_page_no, file_ptr.p);
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
Uint32 extent = (key->m_page_no - data_off) / size + per_page;
Uint32 page_no = extent / per_page;
Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = page_no; preq.m_page.m_page_no = val.m_extent_page_no;
preq.m_page.m_file_no = key->m_file_no; preq.m_page.m_file_no = key->m_file_no;
/** /**
@ -1727,12 +1715,12 @@ Tsman::unmap_page(Signal* signal, Local_key *key, Uint32 uncommitted_bits)
File_formats::Datafile::Extent_page* page = File_formats::Datafile::Extent_page* page =
(File_formats::Datafile::Extent_page*)ptr_p; (File_formats::Datafile::Extent_page*)ptr_p;
File_formats::Datafile::Extent_header* header = File_formats::Datafile::Extent_header* header =
page->get_header(extent_no, size); page->get_header(val.m_extent_no, val.m_extent_size);
ndbrequire(header->m_table != RNIL); ndbrequire(header->m_table != RNIL);
Uint32 page_no_in_extent = (key->m_page_no - data_off) % size; Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
/** /**
* Toggle word * Toggle word
*/ */
@ -1767,18 +1755,10 @@ Tsman::restart_undo_page_free_bits(Signal* signal,
file_key.m_file_no = key->m_file_no; file_key.m_file_no = key->m_file_no;
ndbrequire(m_file_hash.find(file_ptr, file_key)); ndbrequire(m_file_hash.find(file_ptr, file_key));
Uint32 size = file_ptr.p->m_extent_size; struct req val = lookup_extent(key->m_page_no, file_ptr.p);
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
Uint32 extent = (key->m_page_no - data_off) / size + per_page;
Uint32 page_no = extent / per_page;
Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = page_no; preq.m_page.m_page_no = val.m_extent_page_no;
preq.m_page.m_file_no = key->m_file_no; preq.m_page.m_file_no = key->m_file_no;
/** /**
@ -1793,7 +1773,7 @@ Tsman::restart_undo_page_free_bits(Signal* signal,
File_formats::Datafile::Extent_page* page = File_formats::Datafile::Extent_page* page =
(File_formats::Datafile::Extent_page*)ptr_p; (File_formats::Datafile::Extent_page*)ptr_p;
File_formats::Datafile::Extent_header* header = File_formats::Datafile::Extent_header* header =
page->get_header(extent_no, size); page->get_header(val.m_extent_no, val.m_extent_size);
Uint64 lsn = 0; Uint64 lsn = 0;
lsn += page->m_page_header.m_page_lsn_hi; lsn <<= 32; lsn += page->m_page_header.m_page_lsn_hi; lsn <<= 32;
@ -1816,7 +1796,7 @@ Tsman::restart_undo_page_free_bits(Signal* signal,
return 0; return 0;
} }
Uint32 page_no_in_extent = (key->m_page_no - data_off) % size; Uint32 page_no_in_extent = calc_page_no_in_extent(key->m_page_no, &val);
Uint32 src = header->get_free_bits(page_no_in_extent); Uint32 src = header->get_free_bits(page_no_in_extent);
ndbrequire(header->m_table == tableId); ndbrequire(header->m_table == tableId);
@ -1862,17 +1842,11 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
Datafile file_key; Datafile file_key;
file_key.m_file_no = req.key.m_file_no; file_key.m_file_no = req.key.m_file_no;
ndbrequire(m_file_hash.find(file_ptr, file_key)); ndbrequire(m_file_hash.find(file_ptr, file_key));
Uint32 size = file_ptr.p->m_extent_size; struct req val = lookup_extent(req.key.m_page_no, file_ptr.p);
Uint32 data_off = file_ptr.p->m_online.m_offset_data_pages;
Uint32 eh_words = File_formats::Datafile::extent_header_words(size);
Uint32 per_page = File_formats::Datafile::EXTENT_PAGE_WORDS/eh_words;
Uint32 extent = (req.key.m_page_no - data_off) / size;
Uint32 extent_no = extent % per_page;
Page_cache_client::Request preq; Page_cache_client::Request preq;
preq.m_page.m_page_no = 1 /* zero */ + extent / per_page; preq.m_page.m_page_no = val.m_extent_page_no;
preq.m_page.m_file_no = req.key.m_file_no; preq.m_page.m_file_no = req.key.m_file_no;
Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE; Uint32 SZ= File_formats::Datafile::EXTENT_HEADER_BITMASK_BITS_PER_PAGE;
@ -1891,11 +1865,11 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
File_formats::Datafile::Extent_page* page = File_formats::Datafile::Extent_page* page =
(File_formats::Datafile::Extent_page*)ptr_p; (File_formats::Datafile::Extent_page*)ptr_p;
header= page->get_header(extent_no, size); header= page->get_header(val.m_extent_no, val.m_extent_size);
ndbrequire(header->m_table == req.request.table_id); ndbrequire(header->m_table == req.request.table_id);
Uint32 page_no_in_extent = (req.key.m_page_no - data_off) % size; Uint32 page_no_in_extent = calc_page_no_in_extent(req.key.m_page_no, &val);
Uint32 word = header->get_free_word_offset(page_no_in_extent); Uint32 word = header->get_free_word_offset(page_no_in_extent);
Uint32 shift = SZ * (page_no_in_extent & 7); Uint32 shift = SZ * (page_no_in_extent & 7);
@ -1912,7 +1886,7 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
* Search * Search
*/ */
Uint32 *src= header->m_page_bitmask + word; Uint32 *src= header->m_page_bitmask + word;
for(page_no= page_no_in_extent; page_no<size; page_no++) for(page_no= page_no_in_extent; page_no<val.m_extent_size; page_no++)
{ {
src_bits= (* src >> shift) & ((1 << SZ) - 1); src_bits= (* src >> shift) & ((1 << SZ) - 1);
if((src_bits & UNCOMMITTED_MASK) <= reqbits) if((src_bits & UNCOMMITTED_MASK) <= reqbits)
@ -1955,15 +1929,26 @@ Tsman::execALLOC_PAGE_REQ(Signal* signal)
found: found:
header->update_free_bits(page_no, src_bits | UNCOMMITTED_MASK); header->update_free_bits(page_no, src_bits | UNCOMMITTED_MASK);
rep->bits= (src_bits & UNCOMMITTED_MASK) >> UNCOMMITTED_SHIFT; rep->bits= (src_bits & UNCOMMITTED_MASK) >> UNCOMMITTED_SHIFT;
rep->key.m_page_no= data_off + extent * size + page_no; rep->key.m_page_no=
val.m_extent_pages + val.m_extent_no * val.m_extent_size + page_no;
rep->reply.errorCode= 0; rep->reply.errorCode= 0;
return; return;
} }
void
Tsman::execLCP_FRAG_ORD(Signal* signal)
{
jamEntry();
ndbrequire(!m_lcp_ongoing);
m_lcp_ongoing = true;
}
void void
Tsman::execEND_LCP_REQ(Signal* signal) Tsman::execEND_LCP_REQ(Signal* signal)
{ {
jamEntry(); jamEntry();
ndbrequire(m_lcp_ongoing);
m_lcp_ongoing = false;
/** /**
* Move extents from "lcp" free list to real free list * Move extents from "lcp" free list to real free list

54
storage/ndb/src/kernel/blocks/tsman.hpp
View file

@ -63,6 +63,7 @@ protected:
void execALLOC_PAGE_REQ(Signal* signal); void execALLOC_PAGE_REQ(Signal* signal);
void execLCP_FRAG_ORD(Signal*);
void execEND_LCP_REQ(Signal*); void execEND_LCP_REQ(Signal*);
void end_lcp(Signal*, Uint32 tablespace, Uint32 list, Uint32 file); void end_lcp(Signal*, Uint32 tablespace, Uint32 list, Uint32 file);
@ -108,6 +109,7 @@ public:
Uint32 m_offset_data_pages; // 1(zero) + extent header pages Uint32 m_offset_data_pages; // 1(zero) + extent header pages
Uint32 m_data_pages; Uint32 m_data_pages;
Uint32 m_used_extent_cnt; Uint32 m_used_extent_cnt;
Uint32 m_extent_headers_per_extent_page;
} m_online; } m_online;
struct { struct {
Uint32 m_senderData; Uint32 m_senderData;
@ -196,6 +198,7 @@ private:
Datafile_pool m_file_pool; Datafile_pool m_file_pool;
Tablespace_pool m_tablespace_pool; Tablespace_pool m_tablespace_pool;
bool m_lcp_ongoing;
Datafile_hash m_file_hash; Datafile_hash m_file_hash;
Tablespace_list m_tablespace_list; Tablespace_list m_tablespace_list;
Tablespace_hash m_tablespace_hash; Tablespace_hash m_tablespace_hash;
@ -226,15 +229,52 @@ private:
void release_extent_pages(Signal* signal, Ptr<Datafile> ptr); void release_extent_pages(Signal* signal, Ptr<Datafile> ptr);
void release_extent_pages_callback(Signal*, Uint32, Uint32); void release_extent_pages_callback(Signal*, Uint32, Uint32);
struct req
{
Uint32 m_extent_pages;
Uint32 m_extent_size;
Uint32 m_extent_no; // on extent page
Uint32 m_extent_page_no;
};
struct req lookup_extent(Uint32 page_no, const Datafile*) const;
Uint32 calc_page_no_in_extent(Uint32 page_no, const struct req* val) const;
}; };
inline
Tsman::req
Tsman::lookup_extent(Uint32 page_no, const Datafile * filePtrP) const
{
struct req val;
val.m_extent_size = filePtrP->m_extent_size;
val.m_extent_pages = filePtrP->m_online.m_offset_data_pages;
Uint32 per_page = filePtrP->m_online.m_extent_headers_per_extent_page;
Uint32 extent =
(page_no - val.m_extent_pages) / val.m_extent_size + per_page;
val.m_extent_page_no = extent / per_page;
val.m_extent_no = extent % per_page;
return val;
}
inline
Uint32
Tsman::calc_page_no_in_extent(Uint32 page_no, const Tsman::req* val) const
{
return (page_no - val->m_extent_pages) % val->m_extent_size;
}
class Tablespace_client class Tablespace_client
{ {
public:
Tsman * m_tsman; Tsman * m_tsman;
Signal* m_signal; Signal* m_signal;
Uint32 m_table_id; Uint32 m_table_id;
Uint32 m_fragment_id; Uint32 m_fragment_id;
Uint32 m_tablespace_id; Uint32 m_tablespace_id;
public: public:
Tablespace_client(Signal* signal, Tsman* tsman, Tablespace_client(Signal* signal, Tsman* tsman,
Uint32 table, Uint32 fragment, Uint32 tablespaceId) { Uint32 table, Uint32 fragment, Uint32 tablespaceId) {
@ -244,6 +284,8 @@ public:
m_fragment_id= fragment; m_fragment_id= fragment;
m_tablespace_id= tablespaceId; m_tablespace_id= tablespaceId;
} }
Tablespace_client(Signal* signal, Tsman* tsman, Local_key* key);
/** /**
* Return >0 if success, no of pages in extent, sets key * Return >0 if success, no of pages in extent, sets key
@ -274,7 +316,7 @@ public:
/** /**
* Free extent * Free extent
*/ */
int free_extent(Local_key* key); int free_extent(Local_key* key, Uint64 lsn);
/** /**
* Update page free bits * Update page free bits
@ -307,6 +349,11 @@ public:
* <0 - on error * <0 - on error
*/ */
int get_tablespace_info(CreateFilegroupImplReq* rep); int get_tablespace_info(CreateFilegroupImplReq* rep);
/**
* Update lsn of page corresponing to key
*/
int update_lsn(Local_key* key, Uint64 lsn);
}; };
#include <signaldata/Extent.hpp> #include <signaldata/Extent.hpp>
@ -351,12 +398,14 @@ Tablespace_client::alloc_page_from_extent(Local_key* key, Uint32 bits)
inline inline
int int
Tablespace_client::free_extent(Local_key* key) Tablespace_client::free_extent(Local_key* key, Uint64 lsn)
{ {
FreeExtentReq* req = (FreeExtentReq*)m_signal->theData; FreeExtentReq* req = (FreeExtentReq*)m_signal->theData;
req->request.key = *key; req->request.key = *key;
req->request.table_id = m_table_id; req->request.table_id = m_table_id;
req->request.tablespace_id = m_tablespace_id; req->request.tablespace_id = m_tablespace_id;
req->request.lsn_hi = (Uint32)(lsn >> 32);
req->request.lsn_lo = (Uint32)(lsn & 0xFFFFFFFF);
m_tsman->execFREE_EXTENT_REQ(m_signal); m_tsman->execFREE_EXTENT_REQ(m_signal);
if(req->reply.errorCode == 0){ if(req->reply.errorCode == 0){
@ -407,5 +456,4 @@ Tablespace_client::restart_undo_page_free_bits(Local_key* key,
page_lsn); page_lsn);
} }
#endif #endif