sqlite1.0.2事务分析

在parse.y语法中定义
84行 cmd ::= BEGIN trans_opt onconf(R).  {sqliteBeginTransaction(pParse,R);}
此句的意思是:在解析中碰到BEGIN trans_opt onconf(R). 句式时,
采用sqliteBeginTransaction(pParse,R);函数替换执行。

sqliteBeginTransaction(pParse,R)函数定义在build.c中的2094行

定义如下

void sqliteBeginTransaction(Parse *pParse, int onError)
{
 
sqlite *db;

 
if( pParse==0 || (db=pParse->db)==0 || db->aDb[0].pBt==0 ) return;
 
if( pParse->nErr || sqlite_malloc_failed ) return;
 
if( sqliteAuthCheck(pParse, SQLITE_TRANSACTION, "BEGIN", 0, 0) ) return;
 
if( db->flags & SQLITE_InTrans )
{
    
sqliteErrorMsg(pParse, "cannot start a transaction within a transaction");
    
return;
 
}
 
sqliteBeginWriteOperation(pParse, 0, 0);
 
db->flags |= SQLITE_InTrans;
 
db->onError = onError;

}

主要关注一个变量 一个函数
db->flags 变量
sqliteBeginWriteOperation(pParse, 0, 0);
 函数
也在build.c下的 2182 行
void sqliteBeginWriteOperation(Parse *pParse, int setCheckpoint, int iDb)
{
 

Vdbe *v;
 
sqlite *db = pParse->db;
 
if( DbHasProperty(db, iDb, DB_Locked) ) return;
 
v = sqliteGetVdbe(pParse);
  if( v==0 ) return;
 
if( !db->aDb[iDb].inTrans )
{
    
sqliteVdbeAddOp(v, OP_Transaction, iDb, 0);
    
DbSetProperty(db, iDb, DB_Locked);
    
sqliteCodeVerifySchema(pParse, iDb);
    
if( iDb!=1 )
{
      
sqliteBeginWriteOperation(pParse, setCheckpoint, 1);
    
}
 
}else if( setCheckpoint )
{
    
sqliteVdbeAddOp(v, OP_Checkpoint, iDb, 0);
    
DbSetProperty(db, iDb, DB_Locked);
 
}


}

关注的函数
v = sqliteGetVdbe(pParse); 此函数是获取执行的虚拟机
sqliteVdbeAddOp(v, OP_Transaction, iDb, 0);
 此处是在虚拟机中新增一个事务开始的处理
sqliteCodeVerifySchema(pParse, iDb);
 判断在执行之前是否已经被更新  

sqliteVdbeAddOp(v, OP_Checkpoint, iDb, 0);
 此处是在虚拟机中设置一个提交点

sqliteVdbeAddOp这个函数的作用是在v这个虚拟机中新增OP_Transaction,这种类型的操作


void sqliteCodeVerifySchema(Parse *pParse, int iDb)
{
 
sqlite *db = pParse->db;
 
Vdbe *v = sqliteGetVdbe(pParse);
 
assert( iDb>=0 && iDbnDb );
 
assert( db->aDb[iDb].pBt!=0 );
 
if( iDb!=1 && !DbHasProperty(db, iDb, DB_Cookie) )
{
    
sqliteVdbeAddOp(v, OP_VerifyCookie, iDb, db->aDb[iDb].schema_cookie);
    
DbSetProperty(db, iDb, DB_Cookie);
 
}

}



针对OP_Transaction的操作关键函数是sqliteBtreeBeginTrans

case OP_Transaction:
{
 
int busy = 1;
 
int i = pOp->p1;
 
assert( i>=0 && inDb );
 
if( db->aDb[i].inTrans ) break;
 
while( db->aDb[i].pBt!=0 && busy )
{
    
rc = sqliteBtreeBeginTrans(db->aDb[i].pBt);
    
switch( rc )
{
      
case SQLITE_BUSY:
{
        
if( db->xBusyCallback==0 )
{
          
p->pc = pc;
          
p->undoTransOnError = 1;
          
p->rc = SQLITE_BUSY;
          
return SQLITE_BUSY;
        
}else if( (*db->xBusyCallback)(db->pBusyArg, "", busy++)==0 )
{
          
sqliteSetString(&p->zErrMsg, sqlite_error_string(rc), (char*)0);
          
busy = 0;
       
}
        
break;
      
}
      
case SQLITE_READONLY:
{
        
rc = SQLITE_OK;
        /* Fall thru into the next case */
      
}
      
case SQLITE_OK:
{
        
p->inTempTrans = 0;
        
busy = 0;
        
break;
      
}
      
default:
{
        
goto abort_due_to_error;
      
}
    
}
 
}
 
db->aDb[i].inTrans = 1;
 
p->undoTransOnError = 1;
 
break;

}

针对OP_Checkpoint关键函数是sqliteBtreeBeginCkpt
case OP_Checkpoint:
{
 
int i = pOp->p1;
 
if( i>=0 && inDb && db->aDb[i].pBt && db->aDb[i].inTrans==1 )
{
    
rc = sqliteBtreeBeginCkpt(db->aDb[i].pBt);
    
if( rc==SQLITE_OK ) db->aDb[i].inTrans = 2;
 
}
 
break;

}
针对OP_VerifyCookie关键函数是sqliteBtreeGetMeta

case OP_VerifyCookie: {
 
int aMeta[SQLITE_N_BTREE_META];
 
assert( pOp->p1>=0 && pOp->p1nDb );
 
rc = sqliteBtreeGetMeta(db->aDb[pOp->p1].pBt, aMeta);
 
if( rc==SQLITE_OK && aMeta[1]!=pOp->p2 )
{
    
sqliteSetString(&p->zErrMsg, "database schema has changed", (char*)0);
    
rc = SQLITE_SCHEMA;
  }
 
break;

}


在btree.h de  105 行的宏定义中

#define sqliteBtreeBeginTrans(pBt)         (btOps(pBt)->BeginTrans(pBt))

#define sqliteBtreeBeginCkpt(pBt)          (btOps(pBt)->BeginCkpt(pBt))

#define sqliteBtreeGetMeta(pBt, aMeta)     (btOps(pBt)->GetMeta(pBt, aMeta))

btOps这个结构体在 btree.h 的40行

struct BtOps {
    
int (*Close)(Btree*);
    
int (*SetCacheSize)(Btree*, int);
    
int (*SetSafetyLevel)(Btree*, int);
    
int (*BeginTrans)(Btree*);
    
int (*Commit)(Btree*);
    
int (*Rollback)(Btree*);
    
int (*BeginCkpt)(Btree*);
    
int (*CommitCkpt)(Btree*);
    
int (*RollbackCkpt)(Btree*);
    
int (*CreateTable)(Btree*, int*);
    
int (*CreateIndex)(Btree*, int*);
    
int (*DropTable)(Btree*, int);
    
int (*ClearTable)(Btree*, int);
    
int (*Cursor)(Btree*, int iTable, int wrFlag, BtCursor **ppCur);
    
int (*GetMeta)(Btree*, int*);
    
int (*UpdateMeta)(Btree*, int*);
    
char *(*IntegrityCheck)(Btree*, int*, int);
    
const char *(*GetFilename)(Btree*);
    
int (*Copyfile)(Btree*,Btree*);

#ifdef SQLITE_TEST
    
int (*PageDump)(Btree*, int, int);
    
struct Pager *(*Pager)(Btree*);

#endif

};
初始化的部分在btree.c的3538行
static BtOps sqliteBtreeOps = {
    
fileBtreeClose,
    
fileBtreeSetCacheSize,
    
fileBtreeSetSafetyLevel,
    
fileBtreeBeginTrans,
    
fileBtreeCommit,
    
fileBtreeRollback,
    
fileBtreeBeginCkpt,
    
fileBtreeCommitCkpt,
    
fileBtreeRollbackCkpt,
    
fileBtreeCreateTable,
    
fileBtreeCreateTable,  
/* Really sqliteBtreeCreateIndex() */
    
fileBtreeDropTable,
    
fileBtreeClearTable,
    
fileBtreeCursor,
    
fileBtreeGetMeta,
    
fileBtreeUpdateMeta,
    
fileBtreeIntegrityCheck,
    
fileBtreeGetFilename,
    
fileBtreeCopyFile,

#ifdef SQLITE_TEST
    
fileBtreePageDump,
    
fileBtreePager

#endif

};

同时回滚的时候的初始化在btree_rb.c的1446行
static BtOps sqliteRbtreeOps = {
    
(int(*)(Btree*)) memRbtreeClose,
    
(int(*)(Btree*,int)) memRbtreeSetCacheSize,
    
(int(*)(Btree*,int)) memRbtreeSetSafetyLevel,
    
(int(*)(Btree*)) memRbtreeBeginTrans,
    
(int(*)(Btree*)) memRbtreeCommit,
    
(int(*)(Btree*)) memRbtreeRollback,
    
(int(*)(Btree*)) memRbtreeBeginCkpt,
    
(int(*)(Btree*)) memRbtreeCommitCkpt,
    
(int(*)(Btree*)) memRbtreeRollbackCkpt,
    
(int(*)(Btree*,int*)) memRbtreeCreateTable,
    
(int(*)(Btree*,int*)) memRbtreeCreateTable,
    
(int(*)(Btree*,int)) memRbtreeDropTable,
    
(int(*)(Btree*,int)) memRbtreeClearTable,
    
(int(*)(Btree*,int,int,BtCursor**)) memRbtreeCursor,
    
(int(*)(Btree*,int*)) memRbtreeGetMeta,
    
(int(*)(Btree*,int*)) memRbtreeUpdateMeta,
    
(char*(*)(Btree*,int*,int)) memRbtreeIntegrityCheck,
    
(const char*(*)(Btree*)) memRbtreeGetFilename,
    
(int(*)(Btree*,Btree*)) memRbtreeCopyFile,


#ifdef SQLITE_TEST
    
(int(*)(Btree*,int,int)) memRbtreePageDump,
    
(struct Pager*(*)(Btree*)) memRbtreePager

#endif

};

分析得出btree.c是处理磁盘文件,btree_rb.c是处理缓存页数据的

分析fileBtreeBeginTrans函数
static int fileBtreeBeginTrans(Btree *pBt){
 
int rc;
 
if( pBt->inTrans ) return SQLITE_ERROR;
 
if( pBt->readOnly ) return SQLITE_READONLY;
 
if( pBt->page1==0 ){
    
rc = lockBtree(pBt);
    
if( rc!=SQLITE_OK ){
      return rc;
    }
 
}
 
rc = sqlitepager_begin(pBt->page1);
 
if( rc==SQLITE_OK ){
    rc = newDatabase(pBt);
  }
 
if( rc==SQLITE_OK ){
    pBt->inTrans = 1;
    pBt->inCkpt = 0;
  }
else{
    unlockBtreeIfUnused(pBt);
  }
 
return rc;

}
关键函数是sqlitepager_begin定义在pager.c的


int sqlitepager_begin(void *pData){
 
PgHdr *pPg = DATA_TO_PGHDR(pData);
 
Pager *pPager = pPg->pPager;
 
int rc = SQLITE_OK;
 
assert( pPg->nRef>0 );
 
assert( pPager->state!=SQLITE_UNLOCK );
 
if( pPager->state==SQLITE_READLOCK ){
    
assert( pPager->aInJournal==0 );
    
rc = sqliteOsWriteLock(&pPager->fd);
    
if( rc!=SQLITE_OK ){
      return rc;
    }
    
pPager->state = SQLITE_WRITELOCK;
    
pPager->dirtyFile = 0;
    
TRACE1("TRANSACTION\n");
    
if( pPager->useJournal && !pPager->tempFile ){
      rc = pager_open_journal(pPager);
    }
 
}
 
return rc;

}
把锁写入数据文件中sqliteOsWriteLock-通过调用os.c函数实现
同时打开日志文件pager_open_journal函数在pager.c的1528行
static int pager_open_journal(Pager *pPager){
 
int rc;
 
assert( pPager->state==SQLITE_WRITELOCK );
 
assert( pPager->journalOpen==0 );
 
assert( pPager->useJournal );
 
sqlitepager_pagecount(pPager);
 
pPager->aInJournal = sqliteMalloc( pPager->dbSize/8 + 1 );
 
if( pPager->aInJournal==0 )
{
    
sqliteOsReadLock(&pPager->fd);
    
pPager->state = SQLITE_READLOCK;
    
return SQLITE_NOMEM;
 
}
 
rc = sqliteOsOpenExclusive(pPager->zJournal, &pPager->jfd,pPager->tempFile);
 
if( rc!=SQLITE_OK )
{
    
sqliteFree(pPager->aInJournal);
    
pPager->aInJournal = 0;
    
sqliteOsReadLock(&pPager->fd);
    
pPager->state = SQLITE_READLOCK;
    
return SQLITE_CANTOPEN;
 
}
 
sqliteOsOpenDirectory(pPager->zDirectory, &pPager->jfd);
 
pPager->journalOpen = 1;
 
pPager->journalStarted = 0;
 
pPager->needSync = 0;
 
pPager->alwaysRollback = 0;
 
pPager->nRec = 0;
 
if( pPager->errMask!=0 ){
    rc = pager_errcode(pPager);
    return rc;
  }
 
pPager->origDbSize = pPager->dbSize;
 
if( journal_format==JOURNAL_FORMAT_3 )
{
    
rc = sqliteOsWrite(&pPager->jfd, aJournalMagic3, sizeof(aJournalMagic3));
    
if( rc==SQLITE_OK )
{
      
rc = write32bits(&pPager->jfd, pPager->noSync ? 0xffffffff : 0);
    
}
    
if( rc==SQLITE_OK )
{
      
pPager->cksumInit = (u32)sqliteRandomInteger();
      
rc = write32bits(&pPager->jfd, pPager->cksumInit);
    
}
 
}else if( journal_format==JOURNAL_FORMAT_2 )
{
    
rc = sqliteOsWrite(&pPager->jfd, aJournalMagic2, sizeof(aJournalMagic2));
 
}else{
    
assert( journal_format==JOURNAL_FORMAT_1 );
    
rc = sqliteOsWrite(&pPager->jfd, aJournalMagic1, sizeof(aJournalMagic1));
 
}
 
if( rc==SQLITE_OK ){
    rc = write32bits(&pPager->jfd, pPager->dbSize);
  }
 
if( pPager->ckptAutoopen && rc==SQLITE_OK ){
    rc = sqlitepager_ckpt_begin(pPager);
  }
 
if( rc!=SQLITE_OK )
{
    
rc = pager_unwritelock(pPager);
    
if( rc==SQLITE_OK ){
      rc = SQLITE_FULL;
    }
 
}
 
return rc;  

}
这个函数就是向日志文件中写入信息的

事务启动之后的update操作处理流程如下,update的操作只是在虚拟机中加入适当的命令
在parse.y 中定义的update如下

//
cmd ::= UPDATE orconf(R) nm(X) dbnm(D) SET setlist(Y) where_opt(Z).
    {sqliteUpdate(pParse,sqliteSrcListAppend(0,&X,&D),Y,Z,R);}

sqliteUpdate 这个函数在update.c中，update.c这个文件就实现了这么一个函数

主要流程如下

开始扫描数据
pWInfo = sqliteWhereBegin(pParse, pTabList, pWhere, 1, 0);
 
if( pWInfo==0 ) goto update_cleanup;

 
记录列表的操作类型
sqliteVdbeAddOp(v, OP_ListWrite, 0, 0);

 
结束扫描
sqliteWhereEnd(pWInfo);
记录update的总数
if( db->flags & SQLITE_CountRows && !pParse->trigStack )
{
    
sqliteVdbeAddOp(v, OP_Integer, 0, 0);
 
}
将总数入栈,在虚拟机中通过操作此数来遍历记录集合
if( db->flags & SQLITE_CountRows && !pParse->trigStack)
{
    
sqliteVdbeAddOp(v, OP_AddImm, 1, 0);
 
}

遍历数据的时候将循环操作放入虚拟机，循环操作的原理如下:
5     Rewind        0      10                                       
6     Column        0      0                                        
............................                                     
8     Callback      2      0
............................                                       
10     Next          0      6                                        

第一次运行的时候执行rewind命令，到一个Btree数的first地方去
然后是处理每个字段，Callback负责弹出栈顶的两个元素，
然后是处理元素的命令，
next就是一个循环将命令操作转向6。

sqliteWhereBegin这个函数负责遍历数据，这个函数在where.c的349行
在693行的地方

for(i=0; inSrc; i++)
{

/* 在不使用索引的情况下 */
int start;

      
brk = pLevel->brk = sqliteVdbeMakeLabel(v);
      
cont = pLevel->cont = sqliteVdbeMakeLabel(v);
      
sqliteVdbeAddOp(v, OP_Rewind, iCur, brk);
      
start = sqliteVdbeCurrentAddr(v);
      
pLevel->op = OP_Next;
      
pLevel->p1 = iCur;
      
pLevel->p2 = start;
      
haveKey = 0;

}
这部分操作就是在虚拟机中建立循环的命令,循环的位置寻找在 iCur 这个树中。

iCur 就是虚拟机用来存取表中数据的游标的序号，start-就是在这个游标中这条记录的位置

sqliteVdbeAddOp(v, OP_ListWrite, 0, 0);

  这个操作才是记录刚才遍历的记录都是进行update操作

操作过程如下

首先把记录读取出来
 sqliteVdbeAddOp(v, OP_ListRewind, 0, 0);
      
 addr = sqliteVdbeAddOp(v, OP_ListRead, 0, 0);
      
 sqliteVdbeAddOp(v, OP_Dup, 0, 0);
 sqliteVdbeAddOp(v, OP_NotExists, iCur, addr);
然后是计算新值(分情况处理)

sqliteExprCode(pParse, pRecnoExpr);
      
sqliteVdbeAddOp(v, OP_Column, iCur, i);
sqliteExprCode(pParse, pChanges->a[j].pExpr);

然后是删除旧值
sqliteVdbeAddOp(v, OP_Delete, iCur, 0);
然后是 将新值插入
sqliteCompleteInsertion(pParse, pTab, iCur, aIdxUsed, chngRecno, 1, -1);


sqliteCompleteInsertion这个函数在 insert.c的881行

定位位置
sqliteVdbeAddOp(v, OP_IdxPut, base+i+1, 0);
生成记录
sqliteVdbeAddOp(v, OP_MakeRecord, pTab->nCol, 0);
在索引位置插入记录到btree游标中(索引用来定位，记录数据插入进去，这里进入的只是缓存)
sqliteVdbeAddOp(v, OP_PutIntKey, base, pParse->trigStack?0:1);
弹出命令并且销毁掉
sqliteVdbeAddOp(v, OP_Pop, 1, 0);


这个地方只是将update遗留下来的空间释放掉
addr-这个变量指向的是循环开始地址
先跳转到循环开始地址
sqliteVdbeAddOp(v, OP_Goto, 0, addr);
 
sqliteVdbeCurrentAddr-指向的是下一条记录的地址，sqliteVdbeChangeP2-负责将下一条记录放入parse v 中去
sqliteVdbeChangeP2(v, addr, sqliteVdbeCurrentAddr(v));
 
sqliteVdbeAddOp-这个函数增加一个结束的命令
sqliteVdbeAddOp(v, OP_ListReset, 0, 0);



在parse.y中 ，表示语法分析结束，结束的时候调用sqliteExec函数

cmdx ::= cmd.           { sqliteExec(pParse); }

sqliteExec这个函数在build.c的78行
这个函数调用sqliteVdbeExec执行虚拟机

sqliteVdbeExec这个函数定义在 vdbe.c的482行，负责执行虚拟机
执行的大致流程如下

/* 这个循环的控制变量时 rc 在循环内部控制 此变量 */
for(pc=p->pc; rc==SQLITE_OK; pc++)
{
  /* 循环获取下一条命令*/
  pOp = &p->aOp[pc];
  /* 负责处理每一条命令 */
  switch( pOp->opcode )
  {
    case  OP_Rewind:{}
    case  OP_Callback:{}
    case  OP_OP_ListWrite:{}
    case  OP_Next:{}

  }

}

到这个地方表示结束
vdbe_halt:
 
if( rc )
{
    
p->rc = rc;
    
rc = SQLITE_ERROR;
 
}
else
{
    
rc = SQLITE_DONE;
 
}
 
p->magic = VDBE_MAGIC_HALT;
 
return rc;


在parse.y中 88行 定义
cmd ::= COMMIT trans_opt.      {sqliteCommitTransaction(pParse);}

cmd ::= END trans_opt.         {sqliteCommitTransaction(pParse);}

cmd ::= ROLLBACK trans_opt.    {sqliteRollbackTransaction(pParse);}

也就是说最后一定需要 commit事务或者是rollback事务

sqliteCommitTransaction函数在build.c的2112行
主要是通过调用sqliteEndWriteOperation函数实现的


sqliteEndWriteOperation定义在build.c的2211行，
这个函数关键操作时 sqliteVdbeAddOp(v, OP_Commit, 0, 0);
在虚拟机中新增commit命令


OP_Commit操作在虚拟机中的执行如下：

rc = sqliteBtreeCommit(db->aDb[i].pBt);

sqliteCommitInternalChanges(db);


#define sqliteBtreeCommit(pBt)             (btOps(pBt)->Commit(pBt))

对应的是 fileBtreeCommit 函数，定义在btree.c中的 904行

sqlitepager_commit(pBt->pPager);

unlockBtreeIfUnused(pBt);

sqlitepager_commit函数定义在 pager.c的1876行

pPg = pager_get_all_dirty_pages(pPager);

rc = pager_write_pagelist(pPg);

rc = pager_unwritelock(pPager);

static PgHdr *pager_get_all_dirty_pages(Pager *pPager)
{
 
PgHdr *p, *pList;
 
pList = 0;
 
for(p=pPager->pAll; p; p=p->pNextAll)
{
    
if( p->dirty )
{
      
p->pDirty = pList;
      
pList = p;
    
}
 
}
 
return pList;

}


static int pager_write_pagelist(PgHdr *pList)
{
 
Pager *pPager;
 
int rc;

 
if( pList==0 ) return SQLITE_OK;
 
pPager = pList->pPager;
 
while( pList )
{
    
assert( pList->dirty );
    
sqliteOsSeek(&pPager->fd, (pList->pgno-1)*(off_t)SQLITE_PAGE_SIZE);
    
rc = sqliteOsWrite(&pPager->fd, PGHDR_TO_DATA(pList), SQLITE_PAGE_SIZE);
    
if( rc ) return rc;
    
pList->dirty = 0;
    
pList = pList->pDirty;
 
}
 
return SQLITE_OK;

}