>
在之前的一篇文章 中，介绍了assembleResponse函数(位于instance.cpp第224行)，它会根据op操作枚举类型来调用相应的crud操作，枚举类型定义如下：
enum
Operations {
opReply
=
1
,
/*
reply. responseTo is set.
*/
dbMsg
=
1000
,
/*
generic msg command followed by a string
*/
dbUpdate
=
2001
,
/*
update object
*/
dbInsert
=
2002
,
//
dbGetByOID = 2003,
dbQuery
=
2004
,
dbGetMore
=
2005
,
dbDelete
=
2006
,
dbKillCursors
=
2007
};
可以看到dbInsert = 2002 为插入操作枚举值，下面我们看一下assembleResponse在确定是插入操作时调用的方法，如下：
assembleResponse( Message
&
m, DbResponse
&
dbresponse,
const
SockAddr
&
client ) {
…..
try
{
if
( op
==
dbInsert ) {
//
添加记录操作
receivedInsert(m, currentOp);
}
else
if
( op
==
dbUpdate ) {
//
更新记录
receivedUpdate(m, currentOp);
}
else
if
( op
==
dbDelete ) {
//
删除记录
receivedDelete(m, currentOp);
}
else
if
( op
==
dbKillCursors ) {
//
删除Cursors（游标）对象
currentOp.ensureStarted();
logThreshold
=
10
;
ss
<<
“
killcursors
“
;
receivedKillCursors(m);
}
else
{
mongo::log()
<<
“
operation isn’t supported:
“
<<
op
<<
endl;
currentOp.done();
log
=
true
;
}
}
…..
}
}
从上面代码可以看出，系统在确定dbInsert操作时，调用了receivedInsert()方法（位于instance.cpp文件第570行），下面是该方法的定义：
void
receivedInsert(Message
&
m, CurOp
&
op) {
DbMessage d(m);
//
初始化数据库格式的消息
const
char
*
ns
=
d.getns();
//
获取名空间，用于接下来insert数据
assert(
*
ns);
uassert(
10058
,
“
not master
“
, isMasterNs( ns ) );
op.debug().str
<<
ns;
writelock lk(ns);
//
声明写锁
if
( handlePossibleShardedMessage( m ,
0
) )
//
查看是不是sharding信息，如果是则处理
return
;
Client::Context ctx(ns);
int
n
=
0
;
while
( d.moreJSObjs() ) {
//
循环获取当前消息体中的BSONObj数据（数据库记录）
BSONObj js
=
d.nextJsObj();
uassert(
10059
,
“
object to insert too large
“
, js.objsize()
<=
BSONObjMaxUserSize);
{
//
声明BSONObj迭代器，以查看里面元素是否有更新操作，如set inc push pull 等
BSONObjIterator i( js );
while
( i.more() ) {
BSONElement e
=
i.next();
uassert(
13511
,
“
object to insert can’t have $ modifiers
“
, e.fieldName()[
0
]
!=
‘
$
‘
);
}
}
//
插入记录操作，god = false用于标识当前BSONObj对象为有效数据
theDataFileMgr.insertWithObjMod(ns, js,
false
);
logOp(
“
i
“
, ns, js);
//
日志操作，包括master状态下及sharding分片情况
if
(
++
n
%
4
==
0
) {
//
在插入一些数据后，进行持久化操作，有关持久化部分参见我的这篇文章
//
http://www.cnblogs.com/daizhj/archive/2011/03/21/1990344.html
getDur().commitIfNeeded();
}
}
globalOpCounters.incInsertInWriteLock(n);
//
在写锁环境下添加已插入记录数（n）,锁采用InterlockedIncrement实现数的原子性
}
上面的方法中，主要是在“写锁”环境下执行插入数据操作，并且在插入记录之前进行简单的数据对象检查，如长度和插入数据是否被修改，以确保数据的最终有效性。
最终上面代码会调用 insertWithObjMod()方法（位于pdfile.cpp 文件第1432行），该方法定义如下：
DiskLoc DataFileMgr::insertWithObjMod(
const
char
*
ns, BSONObj
&
o,
bool
god) {
DiskLoc loc
=
insert( ns, o.objdata(), o.objsize(), god );
if
(
!
loc.isNull() )
//
判断返回记录地址是否为空（记录是否插入成功）
o
=
BSONObj( loc.rec() );
//
如有效，则用记录地地址上的记录(record类型指针)绑定到o上
return
loc;
}
该方法只是一个对插入操作及返回结果的封装，其中ns为数据对象的名空间，o就是要插入的数据对象（BSONObj），god用于标识当前BSONObj 对象是否为有效数据（false=有效），这里之所以要传入god这个参数，是因为在接下来的insert方法里同时支持添加名空间(及索引)和插入记录 操作（都会不断调用该方法），而在添加名空间时god=true。
下面我们看一下insert方法（pdfile.cpp 第1467行），因为其内容较长，请详见注释：
DiskLoc DataFileMgr::insert(
const
char
*
ns,
const
void
*
obuf,
int
len,
bool
god,
const
BSONElement
&
writeId,
bool
mayAddIndex) {
bool
wouldAddIndex
=
false
;
massert(
10093
,
“
cannot insert into reserved $ collection
“
, god
||
isANormalNSName( ns ) );
uassert(
10094
, str::stream()
<<
“
invalid ns:
“
<<
ns , isValidNS( ns ) );
const
char
*
sys
=
strstr(ns,
“
system.
“
);
if
( sys ) {
//
对插入记录的ns进行判断，是否要插入保留的数据库名(system)，如是则停止执行其它代码
uassert(
10095
,
“
attempt to insert in reserved database name ‘system’
“
, sys
!=
ns);
if
( strstr(ns,
“
.system.
“
) ) {
//
later:check for dba-type permissions here if have that at some point separate
if
( strstr(ns,
“
.system.indexes
“
) )
//
判断是否创建索引
wouldAddIndex
=
true
;
else
if
( legalClientSystemNS( ns ,
true
) )
;
else
if
(
!
god ) {
//
表示obuf有数据，但这就意味着要向system下插入数据（把system当成数据表了）
out
()
<<
“
ERROR: attempt to insert in system namespace
“
<<
ns
<<
endl;
return
DiskLoc();
}
}
else
sys
=
0
;
}
bool
addIndex
=
wouldAddIndex
&&
mayAddIndex;
//
判断是否需要添加索引
NamespaceDetails
*
d
=
nsdetails(ns);
//
获取ns的详细信息
if
( d
==
0
) {
addNewNamespaceToCatalog(ns);
//
向system catalog添加新的名空间，它会再次调用当前insert()方法
/*
todo: shouldn’t be in the namespace catalog until after the allocations here work.
also if this is an addIndex, those checks should happen before this!
*/
//
创建第一个数据库文件.
cc().database()
->
allocExtent(ns, Extent::initialSize(len),
false
);
d
=
nsdetails(ns);
if
(
!
god )
ensureIdIndexForNewNs(ns);
}
d
->
paddingFits();
NamespaceDetails
*
tableToIndex
=
0
;
string
tabletoidxns;
BSONObj fixedIndexObject;
if
( addIndex ) {
assert( obuf );
BSONObj io((
const
char
*
) obuf);
//
做索引准备工作，这里并不真正创建索引，只是进行参数检查，以及索引是否已存在等
if
(
!
prepareToBuildIndex(io, god, tabletoidxns, tableToIndex, fixedIndexObject ) )
return
DiskLoc();
if
(
!
fixedIndexObject.isEmpty() ) {
obuf
=
fixedIndexObject.objdata();
len
=
fixedIndexObject.objsize();
}
}
const
BSONElement
*
newId
=
&
writeId;
int
addID
=
0
;
if
(
!
god ) {
//
检查对象 是否有_id字段，没有则添加
//
Note that btree buckets which we insert aren’t BSONObj’s, but in that case god==true.
BSONObj io((
const
char
*
) obuf);
BSONElement idField
=
io.getField(
“
_id
“
);
uassert(
10099
,
“
_id cannot be an array
“
, idField.type()
!=
Array );
if
( idField.eoo()
/*
判断是否是结束元素
*/
&&
!
wouldAddIndex
&&
strstr(ns,
“
.local.
“
)
==
0
) {
addID
=
len;
if
( writeId.eoo() ) {
//
初始化一个_id 随机值（因为_id可能是12 byte类型或其它类型）
idToInsert_.oid.init();
newId
=
&
idToInsert;
//
绑定初始化的_id值
}
len
+=
newId
->
size();
}
//
如果io对象中有时间戳元素时，并用当前时间进行更新
BSONElementManipulator::lookForTimestamps( io );
}
//
兼容旧的数据文件
DiskLoc extentLoc;
int
lenWHdr
=
len
+
Record::HeaderSize;
lenWHdr
=
(
int
) (lenWHdr
*
d
->
paddingFactor);
if
( lenWHdr
==
0
) {
assert( d
->
paddingFactor
==
0
);
*
getDur().writing(
&
d
->
paddingFactor)
=
1.0
;
lenWHdr
=
len
+
Record::HeaderSize;
}
//
在对新的对象分配空间前检查数据是否会造成索引冲突（唯一索引）
//
capped标识是否是固定大小的集合类型，这种类型下系统会自动将过于陈旧的数据remove掉
//
注：此cap与nosql中常说的cap无太大关联
//
nosql cap即：一致性，有效性，分区容忍性
//
参见这篇文章:
http://blog.nosqlfan.com/html/1112.html
,
//
http://blog.nosqlfan.com/html/96.html
)
if
( d
->
nIndexes
&&
d
->
capped
&&
!
god ) {
checkNoIndexConflicts( d, BSONObj( reinterpret_cast
<
const
char
*>
( obuf ) ) );
}
DiskLoc loc
=
d
->
alloc(ns, lenWHdr, extentLoc);
//
为当前记录分配空间namespace.cpp __stdAlloc方法
if
( loc.isNull() ) {
//
如果分配失效
if
( d
->
capped
==
0
) {
//
cap大小未增加，即
log(
1
)
<<
“
allocating new extent for
“
<<
ns
<<
“
padding:
“
<<
d
->
paddingFactor
<<
“
lenWHdr:
“
<<
lenWHdr
<<
endl;
//
尝试从空闲空间列表中分配空间
cc().database()
->
allocExtent(ns, Extent::followupSize(lenWHdr, d
->
lastExtentSize),
false
);
//
尝试再次为当前记录分配空间
loc
=
d
->
alloc(ns, lenWHdr, extentLoc);
if
( loc.isNull() ) {
log()
<<
“
WARNING: alloc() failed after allocating new extent. lenWHdr:
“
<<
lenWHdr
<<
“
last extent size:
“
<<
d
->
lastExtentSize
<<
“
; trying again/n
“
;
for
(
int
zzz
=
0
; zzz
<
10
&&
lenWHdr
>
d
->
lastExtentSize; zzz
++
) {
//
最多尝试循环10次分配空间
log()
<<
“
try #
“
<<
zzz
<<
endl;
cc().database()
->
allocExtent(ns, Extent::followupSize(len, d
->
lastExtentSize),
false
);
loc
=
d
->
alloc(ns, lenWHdr, extentLoc);
if
(
!
loc.isNull() )
break
;
}
}
}
if
( loc.isNull() ) {
//
最终未分配空间给对象
log()
<<
“
insert: couldn’t alloc space for object ns:
“
<<
ns
<<
“
capped:
“
<<
d
->
capped
<<
endl;
assert(d
->
capped);
return
DiskLoc();
}
}
Record
*
r
=
loc.rec();
{
assert( r
->
lengthWithHeaders
>=
lenWHdr );
r
=
(Record
*
) getDur().writingPtr(r, lenWHdr);
//
持久化插入记录信息
if
( addID ) {
/*
a little effort was made here to avoid a double copy when we add an ID
*/
((
int
&
)
*
r
->
data)
=
*
((
int
*
) obuf)
+
newId
->
size();
memcpy(r
->
data
+
4
, newId
->
rawdata(), newId
->
size());
//
拷贝_id字段到指定记录内存空间
memcpy(r
->
data
+
4
+
newId
->
size(), ((
char
*
)obuf)
+
4
, addID
–
4
);
//
拷贝数据到指定内存空间
}
else
{
if
( obuf )
memcpy(r
->
data, obuf, len);
//
直接拷贝数据到记录字段r
}
}
{
Extent
*
e
=
r
->
myExtent(loc);
if
( e
->
lastRecord.isNull() ) {
//
如果未尾记录为空，本人理解：即之前未插入过记录
Extent::FL
*
fl
=
getDur().writing(e
->
fl());
fl
->
firstRecord
=
fl
->
lastRecord
=
loc;
r
->
prevOfs
=
r
->
nextOfs
=
DiskLoc::NullOfs;
}
else
{
Record
*
oldlast
=
e
->
lastRecord.rec();
//
否则将新记录添加到最后一条记录的后面
r
->
prevOfs
=
e
->
lastRecord.getOfs();
r
->
nextOfs
=
DiskLoc::NullOfs;
getDur().writingInt(oldlast
->
nextOfs)
=
loc.getOfs();
getDur().writingDiskLoc(e
->
lastRecord)
=
loc;
}
}
/*
持久化操作并更新相应统计信息
*/
{
NamespaceDetails::Stats
*
s
=
getDur().writing(
&
d
->
stats);
s
->
datasize
+=
r
->
netLength();
s
->
nrecords
++
;
}
//
在god时会清空stats信息，同时会添加一个 btree bucket（占据存储空间）
if
(
!
god )
NamespaceDetailsTransient::get_w( ns ).notifyOfWriteOp();
//
在写操作时清空缓存，优化查询优化
if
( tableToIndex ) {
uassert(
13143
,
“
can’t create index on system.indexes
“
, tabletoidxns.find(
“
.system.indexes
“
)
==
string
::npos );
BSONObj info
=
loc.obj();
bool
background
=
info[
“
background
“
].trueValue();
if
( background
&&
cc().isSyncThread() ) {
/*
don’t do background indexing on slaves. there are nuances. this could be added later but requires more code.
*/
log()
<<
“
info: indexing in foreground on this replica; was a background index build on the primary
“
<<
endl;
background
=
false
;
}
int
idxNo
=
tableToIndex
->
nIndexes;
IndexDetails
&
idx
=
tableToIndex
->
addIndex(tabletoidxns.c_str(),
!
background);
//
清空临时缓存信息; 同时递增索引数量
getDur().writingDiskLoc(idx.info)
=
loc;
try
{
buildAnIndex(tabletoidxns, tableToIndex, idx, idxNo, background);
//
创建索引
}
catch
( DBException
&
e ) {
//
保存异常信息，并执行dropIndexes
LastError
*
le
=
lastError.
get
();
int
savecode
=
0
;
string
saveerrmsg;
if
( le ) {
savecode
=
le
->
code;
saveerrmsg
=
le
->
msg;
}
else
{
savecode
=
e.getCode();
saveerrmsg
=
e.what();
}
//
回滚索引操作(drop索引)
string
name
=
idx.indexName();
BSONObjBuilder b;
string
errmsg;
bool
ok
=
dropIndexes(tableToIndex, tabletoidxns.c_str(), name.c_str(), errmsg, b,
true
);
if
(
!
ok ) {
log()
<<
“
failed to drop index after a unique key error building it:
“
<<
errmsg
<<
‘
‘
<<
tabletoidxns
<<
‘
‘
<<
name
<<
endl;
}
assert( le
&&
!
saveerrmsg.empty() );
raiseError(savecode,saveerrmsg.c_str());
throw
;
}
}
/*
将记录数据添加到索引信息（btree）中
*/
if
( d
->
nIndexes ) {
try
{
BSONObj obj(r
->
data);
indexRecord(d, obj, loc);
}
catch
( AssertionException
&
e ) {
//
_id index 键值重复
if
( tableToIndex
||
d
->
capped ) {
massert(
12583
,
“
unexpected index insertion failure on capped collection
“
,
!
d
->
capped );
string
s
=
e.toString();
s
+=
“
: on addIndex/capped – collection and its index will not match
“
;
uassert_nothrow(s.c_str());
error()
<<
s
<<
endl;
}
else
{
//
回滚上述操作
_deleteRecord(d, ns, r, loc);
throw
;
}
}
}
//
out() << ” inserted at loc:” << hex << loc.getOfs() << ” lenwhdr:” << hex << lenWHdr << dec << ‘ ‘ << ns << endl;
return
loc;
}
正如之前所说，该方法会完成添加名空间，添加索引，添加数据记录（memcpy调用）。其中名空间的添加方法addNewNamespaceToCatalog 比较简单，下面主要介绍一下索引的创建过程，这里分为了两步：
1.创建索引树（b树）
2.将数据（主要是地址）添加到索引（树）中
先看一下创建索引过程：
static
void
buildAnIndex(
string
ns, NamespaceDetails
*
d, IndexDetails
&
idx,
int
idxNo,
bool
background) {
tlog()
<<
“
building new index on
“
<<
idx.keyPattern()
<<
“
for
“
<<
ns
<<
( background
?
“
background
“
:
“”
)
<<
endl;
Timer t;
unsigned
long
long
n;
if
( background ) {
log(
2
)
<<
“
buildAnIndex: background=true/n
“
;
}
assert(
!
BackgroundOperation::inProgForNs(ns.c_str()) );
//
should have been checked earlier, better not be…
assert( d
->
indexBuildInProgress
==
0
);
assertInWriteLock();
RecoverableIndexState recoverable( d );
if
( inDBRepair
||
!
background ) {
//
当数据库在repair时或非后台工作方式下
n
=
fastBuildIndex(ns.c_str(), d, idx, idxNo);
//
创建索引
assert(
!
idx.head.isNull() );
}
else
{
BackgroundIndexBuildJob j(ns.c_str());
//
以后台方式创建索引
n
=
j.go(ns, d, idx, idxNo);
}
tlog()
<<
“
done for
“
<<
n
<<
“
records
“
<<
t.millis()
/
1000.0
<<
“
secs
“
<<
endl;
}
创建索引方法会要据创建方式（是否是后台线程等），使用不同的方法，这里主要讲解非后台方式，也就是上面的fastBuildIndex方法（pdfile.cpp第1101行），其定义如下（内容详见注释）：
unsigned
long
long
fastBuildIndex(
const
char
*
ns, NamespaceDetails
*
d, IndexDetails
&
idx,
int
idxNo) {
CurOp
*
op
=
cc().curop();
//
设置当前操作指针，用于设置操作信息
Timer t;
tlog(
1
)
<<
“
fastBuildIndex
“
<<
ns
<<
“
idxNo:
“
<<
idxNo
<<
‘
‘
<<
idx.info.obj().toString()
<<
endl;
bool
dupsAllowed
=
!
idx.unique();
bool
dropDups
=
idx.dropDups()
||
inDBRepair;
BSONObj order
=
idx.keyPattern();
getDur().writingDiskLoc(idx.head).Null();
if
( logLevel
>
1
) printMemInfo(
“
before index start
“
);
/*
获取并排序所有键值 —–
*/
unsigned
long
long
n
=
0
;
shared_ptr
<
Cursor
>
c
=
theDataFileMgr.findAll(ns);
BSONObjExternalSorter sorter(order);
sorter.hintNumObjects( d
->
stats.nrecords );
unsigned
long
long
nkeys
=
0
;
ProgressMeterHolder pm( op
->
setMessage(
“
index: (1/3) external sort
“
, d
->
stats.nrecords ,
10
) );
while
( c
->
ok() ) {
BSONObj o
=
c
->
current();
DiskLoc loc
=
c
->
currLoc();
BSONObjSetDefaultOrder keys;
idx.getKeysFromObject(o, keys);
//
从对象中获取键值信息
int
k
=
0
;
for
( BSONObjSetDefaultOrder::iterator i
=
keys.begin(); i
!=
keys.end(); i
++
) {
if
(
++
k
==
2
) {
//
是否是多键索引
d
->
setIndexIsMultikey(idxNo);
}
sorter.add(
*
i, loc);
//
向排序器添加键值和记录位置信息
nkeys
++
;
}
c
->
advance();
n
++
;
pm.hit();
if
( logLevel
>
1
&&
n
%
10000
==
0
) {
printMemInfo(
“
/t iterating objects
“
);
}
};
pm.finished();
if
( logLevel
>
1
) printMemInfo(
“
before final sort
“
);
sorter.sort();
if
( logLevel
>
1
) printMemInfo(
“
after final sort
“
);
log(t.seconds()
>
5
?
0
:
1
)
<<
“
/t external sort used :
“
<<
sorter.numFiles()
<<
“
files
“
<<
“
in
“
<<
t.seconds()
<<
“
secs
“
<<
endl;
list
<
DiskLoc
>
dupsToDrop;
/*
创建索引
*/
{
BtreeBuilder btBuilder(dupsAllowed, idx);
//
实例化b树索引对象
//
BSONObj keyLast;
auto_ptr
<
BSONObjExternalSorter::Iterator
>
i
=
sorter.iterator();
//
初始化迭代器用于下面遍历
assert( pm
==
op
->
setMessage(
“
index: (2/3) btree bottom up
“
, nkeys ,
10
) );
while
( i
->
more() ) {
RARELY killCurrentOp.checkForInterrupt();
//
检查冲突如shutdown或kill指令
BSONObjExternalSorter::Data d
=
i
->
next();
try
{
btBuilder.addKey(d.first, d.second);
//
向b树索引对象中添加索引键值和记录位置信息
}
catch
( AssertionException
&
e ) {
if
( dupsAllowed ) {
//
unknow exception??
throw
;
}
if
( e.interrupted() )
throw
;
if
(
!
dropDups )
throw
;
/*
we could queue these on disk, but normally there are very few dups, so instead we
keep in ram and have a limit.
*/
dupsToDrop.push_back(d.second);
uassert(
10092
,
“
too may dups on index build with dropDups=true
“
, dupsToDrop.size()
<
1000000
);
}
pm.hit();
}
pm.finished();
op
->
setMessage(
“
index: (3/3) btree-middle
“
);
log(t.seconds()
>
10
?
0
:
1
)
<<
“
/t done building bottom layer, going to commit
“
<<
endl;
btBuilder.commit();
//
提交创建索引操作，该方法会完成最终构造Btree索引操作
wassert( btBuilder.getn()
==
nkeys
||
dropDups );
}
log(
1
)
<<
“
/t fastBuildIndex dupsToDrop:
“
<<
dupsToDrop.size()
<<
endl;
//
删除索引中已出现的重复记录
for
( list
<
DiskLoc
>
::iterator i
=
dupsToDrop.begin(); i
!=
dupsToDrop.end(); i
++
)
theDataFileMgr.deleteRecord( ns, i
->
rec(),
*
i,
false
,
true
);
return
n;
}
上面方法主要对要创建的索引信息进行提取，并封装到一个BtreeBuilder中，顾名思义，该对象用于进行b树的创建(因为索引也是一个b树),当信息收集排序完成后，就开始创建索引，如下：
btree.cpp 1842行
void
BtreeBuilder::commit() {
buildNextLevel(first);
committed
=
true
;
}
void
BtreeBuilder::buildNextLevel(DiskLoc loc) {
int
levels
=
1
;
while
(
1
) {
if
( loc.btree()
->
tempNext().isNull() ) {
//
在当前层级上只有一个 bucket
getDur().writingDiskLoc(idx.head)
=
loc;
break
;
}
levels
++
;
DiskLoc upLoc
=
BtreeBucket::addBucket(idx);
//
添加bucket并实例化上一层DiskLoc
DiskLoc upStart
=
upLoc;
BtreeBucket
*
up
=
upLoc.btreemod();
//
获取上一层的bucket指针
DiskLoc xloc
=
loc;
while
(
!
xloc.isNull() ) {
RARELY {
getDur().commitIfNeeded();
b
=
cur.btreemod();
up
=
upLoc.btreemod();
}
BtreeBucket
*
x
=
xloc.btreemod();
BSONObj k;
DiskLoc r;
x
->
popBack(r,k);
//
弹出当前bucket中最右边的键
bool
keepX
=
( x
->
n
!=
0
);
//
当前bucket中元素个数是否为0
DiskLoc keepLoc
=
keepX
?
xloc : x
->
nextChild;
//
压入上面弹出的最右边的键值，该键值为当前up（bucket）中最大值
if
(
!
up
->
_pushBack(r, k, ordering, keepLoc) )
{
//
当前 bucket 已满，则新创建一个addBucket
DiskLoc n
=
BtreeBucket::addBucket(idx);
up
->
tempNext()
=
n;
upLoc
=
n;
up
=
upLoc.btreemod();
up
->
pushBack(r, k, ordering, keepLoc);
}
DiskLoc nextLoc
=
x
->
tempNext();
//
get next in chain at current level
if
( keepX ) {
//
表示当前结点非顶层结点，则设置它的父结点
x
->
parent
=
upLoc;
}
else
{
if
(
!
x
->
nextChild.isNull() )
x
->
nextChild.btreemod()
->
parent
=
upLoc;
x
->
deallocBucket( xloc, idx );
//
删除xloc bucket
}
xloc
=
nextLoc;
//
指向当前层的下个元素
}
loc
=
upStart;
//
升级当前结点
mayCommitProgressDurably();
}
if
( levels
>
1
)
log(
2
)
<<
“
btree levels:
“
<<
levels
<<
endl;
}
上面的buildNextLevel方法自下而上根据之前抽取的键值逐层构造一个b树。这里有一个问题需要注意一下，因为mongodb使用 bucket来作为b树中的一个层次结点或叶子结点容器（如下图），bucket最大尺寸为8192字节,c。有关b树索引的文章可以参见这篇文章 ：，
mongodb目前关于B树索引的文档 ：http://blog.nosqlfan.com/html/758.html
当初始化了b树索引及空间信息之后，下面就会将数据绑定到相应信息结点上了，也就是DataFileMgr::insert方法(pdfile.cpp文件)的如下代码：
/*
将记录数据添加到索引信息（btree）中
*/
if
( d
->
nIndexes ) {
try
{
BSONObj obj(r
->
data);
indexRecord(d, obj, loc);
}
……
}
上面的indexRecord方法会将键值和数据（包括存储位置）添加到索引中(其中参数d包括之前创建的B树索引信息), 该方法定义如下（pdfile.cpp 第1355行）:
/*
将键值和数据（包括存储位置）添加到索引中
*/
static
void
indexRecord(NamespaceDetails
*
d, BSONObj obj, DiskLoc loc) {
int
n
=
d
->
nIndexesBeingBuilt();
//
获取已（及正在）构建的索引数
for
(
int
i
=
0
; i
<
n; i
++
) {
try
{
bool
unique
=
d
->
idx(i).unique();
//
内联函数（inline）:将索引和记录相关信息初始化到btree中
_indexRecord(d, i
/*
索引顺序位
*/
, obj, loc,
/*
dupsAllowed
*/
!
unique);
}
catch
( DBException
&
) {
/*
如果发生异常，则进行回滚操作
note <= i (not < i) is important here as the index we were just attempted
may be multikey and require some cleanup.
*/
for
(
int
j
=
0
; j
<=
i; j
++
) {
try
{
_unindexRecord(d
->
idx(j), obj, loc,
false
);
}
catch
(…) {
log(
3
)
<<
“
unindex fails on rollback after unique failure/n
“
;
}
}
throw
;
}
}
}
上面的_indexRecord为内联函数（pdfile.cpp）（inline关键字参见C++说明），该参数声明如下：
static
inline
void
_indexRecord(NamespaceDetails
*
d,
int
idxNo, BSONObj
&
obj, DiskLoc recordLoc,
bool
dupsAllowed) {
IndexDetails
&
idx
=
d
->
idx(idxNo);
//
BSONObjSetDefaultOrder keys;
idx.getKeysFromObject(obj, keys);
//
从对象信息中获取键属性信息
BSONObj order
=
idx.keyPattern();
Ordering ordering
=
Ordering::make(order);
//
初始化排序方式用于下面传参
int
n
=
0
;
for
( BSONObjSetDefaultOrder::iterator i
=
keys.begin(); i
!=
keys.end(); i
++
) {
if
(
++
n
==
2
) {
d
->
setIndexIsMultikey(idxNo);
//
设置多键值索引
}
assert(
!
recordLoc.isNull() );
try
{
idx.head
/*
DiskLoc
*/
.btree()
/*
BtreeBucket
*/
->
bt_insert(idx.head, recordLoc,
//
执行向btree中添加记录和绑定索引信息的操作
*
i, ordering, dupsAllowed, idx);
}
catch
(AssertionException
&
e) {
if
( e.getCode()
==
10287
&&
idxNo
==
d
->
nIndexes ) {
DEV log()
<<
“
info: caught key already in index on bg indexing (ok)
“
<<
endl;
continue
;
}
if
(
!
dupsAllowed ) {
//
重复键值异常
throw
;
}
problem()
<<
“
caught assertion _indexRecord
“
<<
idx.indexNamespace()
<<
endl;
}
}
}
上面方法最终会执行b树插入方法bt_insert（btree.cpp文件1622行），如下（详情见注释）：
int
BtreeBucket::bt_insert(
const
DiskLoc thisLoc,
const
DiskLoc recordLoc,
const
BSONObj
&
key,
const
Ordering
&
order,
bool
dupsAllowed,
IndexDetails
&
idx,
bool
toplevel)
const
{
if
( toplevel ) {
//
如果是顶级节点（如果是通过构造索引方式调用 ，则toplevel=true）
//
判断键值是否过界（因为其会存储在system.indexs中），其中：KeyMax = 8192 / 10 .mongodb开发团队可能会在更高版本中扩大该值
if
( key.objsize()
>
KeyMax ) {
problem()
<<
“
Btree::insert: key too large to index, skipping
“
<<
idx.indexNamespace()
<<
‘
‘
<<
key.objsize()
<<
‘
‘
<<
key.toString()
<<
endl;
return
3
;
}
}
//
执行添加操作
int
x
=
_insert(thisLoc, recordLoc, key, order, dupsAllowed, DiskLoc(), DiskLoc(), idx);
assertValid( order );
//
assert排序方式是否有效
return
x;
}
上面代码紧接着会调用btree.cpp文件的内部方法_insert（btree.cpp文件 1554行）：
int
BtreeBucket::_insert(
const
DiskLoc thisLoc,
const
DiskLoc recordLoc,
const
BSONObj
&
key,
const
Ordering
&
order,
bool
dupsAllowed,
const
DiskLoc lChild,
const
DiskLoc rChild, IndexDetails
&
idx)
const
{
if
( key.objsize()
>
KeyMax ) {
problem()
<<
“
ERROR: key too large len:
“
<<
key.objsize()
<<
“
max:
“
<<
KeyMax
<<
‘
‘
<<
key.objsize()
<<
‘
‘
<<
idx.indexNamespace()
<<
endl;
return
2
;
}
assert( key.objsize()
>
0
);
int
pos;
//
在btree bucket中使用二分查询，查看键值是否已在所索引信息中
bool
found
=
find(idx, key, recordLoc, order, pos
/*
返回该索引信息所在或应该在的位置
*/
,
!
dupsAllowed);
if
( insert_debug ) {
out
()
<<
“
“
<<
thisLoc.toString()
<<
‘
.
‘
<<
“
_insert
“
<<
key.toString()
<<
‘
/
‘
<<
recordLoc.toString()
<<
“
l:
“
<<
lChild.toString()
<<
“
r:
“
<<
rChild.toString()
<<
endl;
out
()
<<
“
found:
“
<<
found
<<
“
pos:
“
<<
pos
<<
“
n:
“
<<
n
<<
endl;
}
if
( found ) {
const
_KeyNode
&
kn
=
k(pos);
//
获取指定磁盘位置的节点信息，_KeyNode
if
( kn.isUnused() ) {
//
查看已存在的键结点是否已使用
log(
4
)
<<
“
btree _insert: reusing unused key
“
<<
endl;
massert(
10285
,
“
_insert: reuse key but lchild is not null
“
, lChild.isNull());
massert(
10286
,
“
_insert: reuse key but rchild is not null
“
, rChild.isNull());
kn.writing().setUsed();
return
0
;
}
DEV {
log()
<<
“
_insert(): key already exists in index (ok for background:true)/n
“
;
log()
<<
“
“
<<
idx.indexNamespace()
<<
“
thisLoc:
“
<<
thisLoc.toString()
<<
‘
/n
‘
;
log()
<<
“
“
<<
key.toString()
<<
‘
/n
‘
;
log()
<<
“
“
<<
“
recordLoc:
“
<<
recordLoc.toString()
<<
“
pos:
“
<<
pos
<<
endl;
log()
<<
“
old l r:
“
<<
childForPos(pos).toString()
<<
‘
‘
<<
childForPos(pos
+
1
).toString()
<<
endl;
log()
<<
“
new l r:
“
<<
lChild.toString()
<<
‘
‘
<<
rChild.toString()
<<
endl;
}
alreadyInIndex();
//
提示键值结点已在索引中，不必再创建，并抛出异常
}
DEBUGGING
out
()
<<
“
TEMP: key:
“
<<
key.toString()
<<
endl;
DiskLoc child
=
childForPos(pos);
//
查询当前pos的子结点信息，以寻找插入位置
if
( insert_debug )
out
()
<<
“
getChild(
“
<<
pos
<<
“
):
“
<<
child.toString()
<<
endl;
if
( child.isNull()
||
!
rChild.isNull()
/*
在当前buckets中插入，即 ‘internal’ 插入
*/
) {
insertHere(thisLoc, pos, recordLoc, key, order, lChild, rChild, idx);
//
在当前buckets中插入
return
0
;
}
//
如果有子结点，则在子结点上执行插入操作
return
child.btree()
->
bt_insert(child, recordLoc, key, order, dupsAllowed, idx,
/*
toplevel
*/
false
);
}
上面_insert方法首先会使用二分法查找要插入的记录是否已存在于索引中，同时会返回一个插入点（pos），如不存在则会进一步在插入点位置查看找 元素以决定是在当前bucket中插入，还是在当前pos位置的（右）子结点(bucket)上插入（这会再次递归调用上面的bt_insert方法）， 这里我们假定在当前bucket插入，则会执行insertHere方法（btree.cpp文件1183行），它的定义如下：
/*
*
* insert a key in this bucket, splitting if necessary.
* @keypos – where to insert the key in range 0..n. 0=make leftmost, n=make rightmost.
* NOTE this function may free some data, and as a result the value passed for keypos may
* be invalid after calling insertHere()
*/
void
BtreeBucket::insertHere(
const
DiskLoc thisLoc,
int
keypos,
const
DiskLoc recordLoc,
const
BSONObj
&
key,
const
Ordering
&
order,
const
DiskLoc lchild,
const
DiskLoc rchild, IndexDetails
&
idx)
const
{
if
( insert_debug )
out
()
<<
“
“
<<
thisLoc.toString()
<<
“
.insertHere
“
<<
key.toString()
<<
‘
/
‘
<<
recordLoc.toString()
<<
‘
‘
<<
lchild.toString()
<<
‘
‘
<<
rchild.toString()
<<
“
keypos:
“
<<
keypos
<<
endl;
DiskLoc oldLoc
=
thisLoc;
//
根据keypos插入相应位置并将数据memcpy到内存指定位置
if
(
!
basicInsert(thisLoc, keypos, recordLoc, key, order) ) {
//
如果插入无效，表示当前bucket已满，则分割记录并放到新创建的bucket中
thisLoc.btreemod()
->
split(thisLoc, keypos, recordLoc, key, order, lchild, rchild, idx);
return
;
}
{
//
持久化当前thisLoc的结点信息并根据插入位置（是否最后一个key），来更新当前thisLoc（及后面key结点）的子结点信息
const
_KeyNode
*
_kn
=
&
k(keypos);
_KeyNode
*
kn
=
(_KeyNode
*
) getDur().alreadyDeclared((_KeyNode
*
) _kn);
//
already declared intent in basicInsert()
if
( keypos
+
1
==
n ) {
//
n为pack（打包后）存储的记录数，这里”判断等于n”表示为最后(last)一个key
if
( nextChild
!=
lchild ) {
//
如果是最后元素，那么”当前最高键值的右子结点应该与要插入的左子结点相同
out
()
<<
“
ERROR nextChild != lchild
“
<<
endl;
out
()
<<
“
thisLoc:
“
<<
thisLoc.toString()
<<
‘
‘
<<
idx.indexNamespace()
<<
endl;
out
()
<<
“
keyPos:
“
<<
keypos
<<
“
n:
“
<<
n
<<
endl;
out
()
<<
“
nextChild:
“
<<
nextChild.toString()
<<
“
lchild:
“
<<
lchild.toString()
<<
endl;
out
()
<<
“
recordLoc:
“
<<
recordLoc.toString()
<<
“
rchild:
“
<<
rchild.toString()
<<
endl;
out
()
<<
“
key:
“
<<
key.toString()
<<
endl;
dump();
assert(
false
);
}
kn
->
prevChildBucket
=
nextChild;
//
“当前最高键值的右子结点”绑定到持久化结点的左子结点
assert( kn
->
prevChildBucket
==
lchild );
nextChild.writing()
=
rchild;
//
持久化”当前最高键值的右子结点”，并将“要插入结点”的右子结点绑定到
if
(
!
rchild.isNull() )
//
如果有右子结点，则更新右子结点的父结点信息为当前thisLoc
rchild.btree()
->
parent.writing()
=
thisLoc;
}
else
{
//
如果keypos位置不是最后一个
kn
->
prevChildBucket
=
lchild;
//
将左子结点绑定到keypos位置结点的左子结点上
if
( k(keypos
+
1
).prevChildBucket
!=
lchild ) {
//
这时左子结点应该与下一个元素的左子结点相同
out
()
<<
“
ERROR k(keypos+1).prevChildBucket != lchild
“
<<
endl;
out
()
<<
“
thisLoc:
“
<<
thisLoc.toString()
<<
‘
‘
<<
idx.indexNamespace()
<<
endl;
out
()
<<
“
keyPos:
“
<<
keypos
<<
“
n:
“
<<
n
<<
endl;
out
()
<<
“
k(keypos+1).pcb:
“
<<
k(keypos
+
1
).prevChildBucket.toString()
<<
“
lchild:
“
<<
lchild.toString()
<<
endl;
out
()
<<
“
recordLoc:
“
<<
recordLoc.toString()
<<
“
rchild:
“
<<
rchild.toString()
<<
endl;
out
()
<<
“
key:
“
<<
key.toString()
<<
endl;
dump();
assert(
false
);
}
const
DiskLoc
*
pc
=
&
k(keypos
+
1
).prevChildBucket;
//
获取keypos后面元素的左子结点信息
*
getDur().alreadyDeclared((DiskLoc
*
) pc)
=
rchild;
//
将右子结点绑定到下一个元素（keypos+1）的左子结点上declared in basicInsert()
if
(
!
rchild.isNull() )
//
如果有右子结点，则更新右子结点的父结点信息为当前thisLoc
rchild.btree()
->
parent.writing()
=
thisLoc;
}
return
;
}
}
该方法中会调用一个叫basicInsert的方法，它主要会在当前bucket中指定位置（keypos）添加记录信息，同时持久化该结点信息，如下：
//
tree.cpp 1183
bool
BucketBasics::basicInsert(
const
DiskLoc thisLoc,
int
&
keypos,
const
DiskLoc recordLoc,
const
BSONObj
&
key,
const
Ordering
&
order)
const
{
assert( keypos
>=
0
&&
keypos
<=
n );
//
判断bucket剩余的空间是否满足当前数据需要的存储空间
int
bytesNeeded
=
key.objsize()
+
sizeof
(_KeyNode);
if
( bytesNeeded
>
emptySize ) {
_pack(thisLoc, order, keypos);
//
如不够用，进行一次整理打包操作，以为bucket中整理更多空间
if
( bytesNeeded
>
emptySize )
//
如还不够用，则返回
return
false
;
}
BucketBasics
*
b;
//
声明Bucket管理对象指针，该对象提供了Bucket存储管理的基本操作和属性，如insert,_pack等
{
const
char
*
p
=
(
const
char
*
)
&
k(keypos);
const
char
*
q
=
(
const
char
*
)
&
k(n
+
1
);
//
declare that we will write to [k(keypos),k(n)]
//
todo: this writes a medium amount to the journal. we may want to add a verb “shift” to the redo log so
//
we can log a very small amount.
b
=
(BucketBasics
*
) getDur().writingAtOffset((
void
*
)
this
, p
–
(
char
*
)
this
, q
–
p);
//
如已有3个结点，目前要插到第三个结点之间，则对每三个元素进行迁移，
//
e.g. n==3, keypos==2
//
1 4 9
//
->
//
1 4 _ 9
for
(
int
j
=
n; j
>
keypos; j
—
)
//
make room
b
->
k(j)
=
b
->
k(j
–
1
);
}
getDur().declareWriteIntent(
&
b
->
emptySize,
12
);
//
[b->emptySize..b->n] is 12 bytes and we are going to write those
b
->
emptySize
-=
sizeof
(_KeyNode);
//
将当前bucket中的剩余空闲空间减少
b
->
n
++
;
//
已有结点数加1
_KeyNode
&
kn
=
b
->
k(keypos);
kn.prevChildBucket.Null();
//
设置当前结点的左子结点为空
kn.recordLoc
=
recordLoc;
//
绑定结点记录信息
kn.setKeyDataOfs((
short
) b
->
_alloc(key.objsize()) );
//
设置结点数据偏移信息
char
*
p
=
b
->
dataAt(kn.keyDataOfs());
//
实例化指向磁盘数据(journal文件)位置（含偏移量）的指针
getDur().declareWriteIntent(p, key.objsize());
//
持久化结点数据信息
memcpy(p, key.objdata(), key.objsize());
//
将当前结点信息复制到p指向的地址空间
return
true
;
}
如果上面方法调用失效，则意味着当前 bucket中已有可用空间插入新记录，这时系统会调用 split（btree.cpp文件 1240行）方法来进行bucket分割，以创建新的bucket并将信息塞入其中，如下：
void
BtreeBucket::split(
const
DiskLoc thisLoc,
int
keypos,
const
DiskLoc recordLoc,
const
BSONObj
&
key,
const
Ordering
&
order,
const
DiskLoc lchild,
const
DiskLoc rchild, IndexDetails
&
idx) {
assertWritable();
if
( split_debug )
out
()
<<
“
“
<<
thisLoc.toString()
<<
“
.split
“
<<
endl;
int
split
=
splitPos( keypos );
//
找到要迁移的数据位置
DiskLoc rLoc
=
addBucket(idx);
//
添加一个新的BtreeBucket
BtreeBucket
*
r
=
rLoc.btreemod();
if
( split_debug )
out
()
<<
“
split:
“
<<
split
<<
‘
‘
<<
keyNode(split).key.toString()
<<
“
n:
“
<<
n
<<
endl;
for
(
int
i
=
split
+
1
; i
<
n; i
++
) {
KeyNode kn
=
keyNode(i);
r
->
pushBack(kn.recordLoc, kn.key, order, kn.prevChildBucket);
//
向新bucket中迁移过剩数据
}
r
->
nextChild
=
nextChild;
//
绑定新bucket的右子结点
r
->
assertValid( order );
if
( split_debug )
out
()
<<
“
new rLoc:
“
<<
rLoc.toString()
<<
endl;
r
=
0
;
rLoc.btree()
->
fixParentPtrs(rLoc);
//
设置当前bucket树的父指针信息
{
KeyNode splitkey
=
keyNode(split);
//
获取内存中分割点位置所存储的数据
nextChild
=
splitkey.prevChildBucket;
//
提升splitkey 键，它的子结点将会是 thisLoc (l) 和 rLoc (r)
if
( split_debug ) {
out
()
<<
“
splitkey key:
“
<<
splitkey.key.toString()
<<
endl;
}
//
将 splitkey 提升为父结点
if
( parent.isNull() ) {
//
如果无父结点时，则创建一个，并将
DiskLoc L
=
addBucket(idx);
BtreeBucket
*
p
=
L.btreemod();
p
->
pushBack(splitkey.recordLoc, splitkey.key, order, thisLoc);
p
->
nextChild
=
rLoc;
//
将分割的bucket为了当前
p
->
assertValid( order );
parent
=
idx.head.writing()
=
L;
//
将splitkey 提升为父结点
if
( split_debug )
out
()
<<
“
we were root, making new root:
“
<<
hex
<<
parent.getOfs()
<<
dec
<<
endl;
rLoc.btree()
->
parent.writing()
=
parent;
}
else
{
//
set this before calling _insert – if it splits it will do fixParent() logic and change the value.
rLoc.btree()
->
parent.writing()
=
parent;
if
( split_debug )
out
()
<<
“
promoting splitkey key
“
<<
splitkey.key.toString()
<<
endl;
//
提升splitkey键，它的左子结点 thisLoc， 右子点rLoc
parent.btree()
->
_insert(parent, splitkey.recordLoc, splitkey.key, order,
/*
dupsallowed
*/
true
, thisLoc, rLoc, idx);
}
}
int
newpos
=
keypos;
//
打包压缩数据(pack,移除无用数据)，以提供更多空间
truncateTo(split, order, newpos);
//
note this may trash splitkey.key. thus we had to promote it before finishing up here.
//
add our new key, there is room now
{
if
( keypos
<=
split ) {
//
如果还有空间存储新键
if
( split_debug )
out
()
<<
“
keypos<split, insertHere() the new key
“
<<
endl;
insertHere(thisLoc, newpos, recordLoc, key, order, lchild, rchild, idx);
//
再次向当前bucket中添加记录
}
else
{
//
如压缩之后依旧无可用空间，则向新创建的bucket中添加节点
int
kp
=
keypos
–
split
–
1
;
assert(kp
>=
0
);
rLoc.btree()
->
insertHere(rLoc, kp, recordLoc, key, order, lchild, rchild, idx);
}
}
if
( split_debug )
out
()
<<
“
split end
“
<<
hex
<<
thisLoc.getOfs()
<<
dec
<<
endl;
}
好了，今天的内容到这里就告一段落了，在接下来的文章中，将会介绍客户端发起Delete操作时，Mongodb的执行流程和相应实现部分。
原文链接:http://www.cnblogs.com/daizhj/archive/2011/03/30/1999699.html
作者: daizhj, 代震军
微博: http://t.sina.com.cn/daizhj
Tags: mongodb,c++,btree
