Many hyperlinks are disabled.
Use anonymous login to enable hyperlinks.

Overview
Comment:Fix a bug occuring when the last connection to disconnect from a database is not also the last to write to it.
Downloads: Tarball | ZIP archive
Timelines: family | ancestors | descendants | both | multi-process
Files: files | file ages | folders
SHA1: 75de95787f4893950743f97160d08d32ae31e447
User & Date: dan 2012-09-03 19:26:03.589
Context
2012-09-04
19:42
Add real inter-process locks and shared memory in place of the stubs. Currently requires activation using LSM_CONFIG_MULTIPLE_PROCESSES. check-in: d37b353a55 user: dan tags: multi-process
2012-09-03
19:26
Fix a bug occuring when the last connection to disconnect from a database is not also the last to write to it. check-in: 75de95787f user: dan tags: multi-process
17:57
Add a missing unlock(WRITER) call to lsmCheckpointWrite(). check-in: e97d02f068 user: dan tags: multi-process
Changes
Unified Diff Ignore Whitespace Patch
Changes to src/lsmInt.h.
︙︙
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506

*/
int lsmTreeNew(lsm_env *, int (*)(void *, int, void *, int), Tree **ppTree);
void lsmTreeRelease(lsm_env *, Tree *);
void lsmTreeClear(lsm_db *);
void lsmTreeInit(lsm_db *);

int lsmTreeSize(lsm_db *);
int lsmTreeIsEmpty(lsm_db *);
int lsmTreeEndTransaction(lsm_db *pDb, int bCommit);
int lsmTreeBeginTransaction(lsm_db *pDb);
int lsmTreeLoadHeader(lsm_db *pDb);

int lsmTreeInsert(lsm_db *pDb, void *pKey, int nKey, void *pVal, int nVal);
void lsmTreeRollback(lsm_db *pDb, TreeMark *pMark);
void lsmTreeMark(lsm_db *pDb, TreeMark *pMark);








<








492
493
494
495
496
497
498

499
500
501
502
503
504
505

*/
int lsmTreeNew(lsm_env *, int (*)(void *, int, void *, int), Tree **ppTree);
void lsmTreeRelease(lsm_env *, Tree *);
void lsmTreeClear(lsm_db *);
void lsmTreeInit(lsm_db *);

int lsmTreeSize(lsm_db *);

int lsmTreeEndTransaction(lsm_db *pDb, int bCommit);
int lsmTreeBeginTransaction(lsm_db *pDb);
int lsmTreeLoadHeader(lsm_db *pDb);

int lsmTreeInsert(lsm_db *pDb, void *pKey, int nKey, void *pVal, int nVal);
void lsmTreeRollback(lsm_db *pDb, TreeMark *pMark);
void lsmTreeMark(lsm_db *pDb, TreeMark *pMark);
︙︙
Changes to src/lsm_shared.c.
︙︙
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

170
171
172
173
174
175
176

  int rc;

  /* Block for an exclusive lock on DMS1. This lock serializes all calls
  ** to doDbConnect() and doDbDisconnect() across all processes.  */
  rc = lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_EXCL, 1);
  if( rc==LSM_OK ){

    /* Try an exclusive lock on DMS2. If successful, this is the first and 
    ** only connection to the database. In this case initialize the 
    ** shared-memory and run log file recovery.  */
    rc = lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_EXCL, 0);
    if( rc==LSM_OK ){
      /* Flush the in-memory tree, if required. If there is data to flush,
      ** this will create a new client snapshot in Database.pClient. The
      ** checkpoint (serialization) of this snapshot may be written to disk
      ** by the following block.  */
      if( 0==lsmTreeIsEmpty(pDb) ){

        rc = lsmFlushToDisk(pDb);
      }

      /* Write a checkpoint to disk. */
      if( rc==LSM_OK ){
        rc = lsmCheckpointWrite(pDb);
      }








|
|
|






|
>








153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

  int rc;

  /* Block for an exclusive lock on DMS1. This lock serializes all calls
  ** to doDbConnect() and doDbDisconnect() across all processes.  */
  rc = lsmShmLock(pDb, LSM_LOCK_DMS1, LSM_LOCK_EXCL, 1);
  if( rc==LSM_OK ){

    /* Try an exclusive lock on DMS2. If successful, this is the last
    ** connection to the database. In this case flush the contents of the
    ** in-memory tree to disk and write a checkpoint.  */
    rc = lsmShmLock(pDb, LSM_LOCK_DMS2, LSM_LOCK_EXCL, 0);
    if( rc==LSM_OK ){
      /* Flush the in-memory tree, if required. If there is data to flush,
      ** this will create a new client snapshot in Database.pClient. The
      ** checkpoint (serialization) of this snapshot may be written to disk
      ** by the following block.  */
      rc = lsmTreeLoadHeader(pDb);
      if( rc==LSM_OK && lsmTreeSize(pDb)>0 ){
        rc = lsmFlushToDisk(pDb);
      }

      /* Write a checkpoint to disk. */
      if( rc==LSM_OK ){
        rc = lsmCheckpointWrite(pDb);
      }
︙︙
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689


  /* If there is no read-transaction open, open one now. */
  rc = lsmBeginReadTrans(pDb);

  /* Attempt to take the WRITER lock */
  if( rc==LSM_OK ){
    rc = lsmShmLock(pDb, LSM_LOCK_WRITER, LSM_LOCK_EXCL, 0);
  assert( rc!=LSM_BUSY );
  }

  /* If the previous writer failed mid-transaction, run emergency rollback. */
  if( rc==LSM_OK && pShm->bWriter ){
    /* TODO: This! */
    assert( 0 );
    rc = LSM_CORRUPT_BKPT;








<








676
677
678
679
680
681
682

683
684
685
686
687
688
689


  /* If there is no read-transaction open, open one now. */
  rc = lsmBeginReadTrans(pDb);

  /* Attempt to take the WRITER lock */
  if( rc==LSM_OK ){
    rc = lsmShmLock(pDb, LSM_LOCK_WRITER, LSM_LOCK_EXCL, 0);

  }

  /* If the previous writer failed mid-transaction, run emergency rollback. */
  if( rc==LSM_OK && pShm->bWriter ){
    /* TODO: This! */
    assert( 0 );
    rc = LSM_CORRUPT_BKPT;
︙︙
Changes to src/lsm_tree.c.
︙︙
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100

** Return, in bytes, the amount of memory currently used by the tree 
** structure.
*/
int lsmTreeSize(lsm_db *pDb){
  return pDb->treehdr.nByte;
}

/*
** Return true if the tree is empty. Otherwise false.
**
** The caller is responsible for ensuring that it has exclusive access
** to the Tree structure for this call.
*/
int lsmTreeIsEmpty(lsm_db *pDb){
  /* TODO: This is not right in a true multi-process system (due to
  ** race conditions)... */
  return (pDb->pShmhdr->hdr1.iRoot==0);
}

/*
** Open a cursor on the in-memory tree pTree.
*/
int lsmTreeCursorNew(lsm_db *pDb, TreeCursor **ppCsr){
  TreeCursor *pCsr;
  *ppCsr = pCsr = lsmMalloc(pDb->pEnv, sizeof(TreeCursor));
  if( pCsr ){








<
<
<
<
<
<
<
<
<
<
<
<








1075
1076
1077
1078
1079
1080
1081












1082
1083
1084
1085
1086
1087
1088

** Return, in bytes, the amount of memory currently used by the tree 
** structure.
*/
int lsmTreeSize(lsm_db *pDb){
  return pDb->treehdr.nByte;
}













/*
** Open a cursor on the in-memory tree pTree.
*/
int lsmTreeCursorNew(lsm_db *pDb, TreeCursor **ppCsr){
  TreeCursor *pCsr;
  *ppCsr = pCsr = lsmMalloc(pDb->pEnv, sizeof(TreeCursor));
  if( pCsr ){
︙︙