Many hyperlinks are disabled.
Use anonymous login
to enable hyperlinks.
Overview
| Comment: | In the sorter, only use large memory allocations if scratch memory has not been configured. Add #ifdefs to disable unused code when SQLITE_MAX_WORKER_THREADS is zero. Other sorter changes in support of testability. |
|---|---|
| Downloads: | Tarball | ZIP archive | SQL archive |
| Timelines: | family | ancestors | descendants | both | threads |
| Files: | files | file ages | folders |
| SHA1: |
d7e2b0d9cb099eda3341bc934bedff9f |
| User & Date: | drh 2014-05-12 15:30:00 |
Context
|
2014-05-15
| ||
| 16:56 | Use #ifdef to omit code that is not used when SQLITE_MAX_WORKER_THREADS is zero. check-in: 2e8d287d user: drh tags: threads | |
|
2014-05-12
| ||
| 15:30 | In the sorter, only use large memory allocations if scratch memory has not been configured. Add #ifdefs to disable unused code when SQLITE_MAX_WORKER_THREADS is zero. Other sorter changes in support of testability. check-in: d7e2b0d9 user: drh tags: threads | |
|
2014-05-09
| ||
| 15:00 | Merge the latest trunk changes into the threads branch. check-in: 9ac8f1e7 user: dan tags: threads | |
Changes
Changes to src/main.c.
512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 |
case SQLITE_CONFIG_WIN32_HEAPSIZE: {
sqlite3GlobalConfig.nHeap = va_arg(ap, int);
break;
}
#endif
case SQLITE_CONFIG_WORKER_THREADS: {
int n = va_arg(ap, int);
if( n>SQLITE_MAX_WORKER_THREADS ) n = SQLITE_MAX_WORKER_THREADS;
if( n>=0 ) sqlite3GlobalConfig.nWorker = n;
break;
}
default: {
rc = SQLITE_ERROR;
break;
}
|
> > |
512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 |
case SQLITE_CONFIG_WIN32_HEAPSIZE: {
sqlite3GlobalConfig.nHeap = va_arg(ap, int);
break;
}
#endif
case SQLITE_CONFIG_WORKER_THREADS: {
#if SQLITE_MAX_WORKER_THREADS>0
int n = va_arg(ap, int);
if( n>SQLITE_MAX_WORKER_THREADS ) n = SQLITE_MAX_WORKER_THREADS;
if( n>=0 ) sqlite3GlobalConfig.nWorker = n;
#endif
break;
}
default: {
rc = SQLITE_ERROR;
break;
}
|
Changes to src/vdbesort.c.
445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 ... 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 ... 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 .... 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 .... 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 .... 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 .... 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 |
** Free all memory belonging to the PmaReader object passed as the second
** argument. All structure fields are set to zero before returning.
*/
static void vdbePmaReaderClear(PmaReader *pIter){
sqlite3_free(pIter->aAlloc);
sqlite3_free(pIter->aBuffer);
if( pIter->aMap ) sqlite3OsUnfetch(pIter->pFile, 0, pIter->aMap);
if( pIter->pIncr ) vdbeIncrFree(pIter->pIncr);
memset(pIter, 0, sizeof(PmaReader));
}
/*
** Read nByte bytes of data from the stream of data iterated by object p.
** If successful, set *ppOut to point to a buffer containing the data
** and return SQLITE_OK. Otherwise, if an error occurs, return an SQLite
................................................................................
int i; /* Used to iterate through aTask[] */
int mxCache; /* Cache size */
VdbeSorter *pSorter; /* The new sorter */
KeyInfo *pKeyInfo; /* Copy of pCsr->pKeyInfo with db==0 */
int szKeyInfo; /* Size of pCsr->pKeyInfo in bytes */
int sz; /* Size of pSorter in bytes */
int rc = SQLITE_OK;
int nWorker = (sqlite3GlobalConfig.bCoreMutex?sqlite3GlobalConfig.nWorker:0);
assert( pCsr->pKeyInfo && pCsr->pBt==0 );
szKeyInfo = sizeof(KeyInfo) + (pCsr->pKeyInfo->nField-1)*sizeof(CollSeq*);
sz = sizeof(VdbeSorter) + nWorker * sizeof(SortSubtask);
pSorter = (VdbeSorter*)sqlite3DbMallocZero(db, sz + szKeyInfo);
pCsr->pSorter = pSorter;
................................................................................
if( !sqlite3TempInMemory(db) ){
pSorter->mnPmaSize = SORTER_MIN_WORKING * pgsz;
mxCache = db->aDb[0].pSchema->cache_size;
if( mxCache<SORTER_MIN_WORKING ) mxCache = SORTER_MIN_WORKING;
pSorter->mxPmaSize = mxCache * pgsz;
/* If the application is using memsys3 or memsys5, use a separate
** allocation for each sort-key in memory. Otherwise, use a single big
** allocation at pSorter->aMemory for all sort-keys. */
if( sqlite3GlobalConfig.pHeap==0 ){
assert( pSorter->iMemory==0 );
pSorter->nMemory = pgsz;
pSorter->list.aMemory = (u8*)sqlite3Malloc(pgsz);
if( !pSorter->list.aMemory ) rc = SQLITE_NOMEM;
}
}
}
................................................................................
rc = vdbeIncrInitMerger(pTask, pIncr->pMerger, eMode);
/* Set up the required files for pIncr. A multi-theaded IncrMerge object
** requires two temp files to itself, whereas a single-threaded object
** only requires a region of pTask->file2. */
if( rc==SQLITE_OK ){
int mxSz = pIncr->mxSz;
if( pIncr->bUseThread==0 ){
if( pTask->file2.pFd==0 ){
assert( pTask->file2.iEof>0 );
rc = vdbeSorterOpenTempFile(db, pTask->file2.iEof, &pTask->file2.pFd);
pTask->file2.iEof = 0;
}
if( rc==SQLITE_OK ){
pIncr->aFile[1].pFd = pTask->file2.pFd;
pIncr->iStartOff = pTask->file2.iEof;
pTask->file2.iEof += mxSz;
}
}else{
rc = vdbeSorterOpenTempFile(db, mxSz, &pIncr->aFile[0].pFd);
if( rc==SQLITE_OK ){
rc = vdbeSorterOpenTempFile(db, mxSz, &pIncr->aFile[1].pFd);
}
}
}
if( rc==SQLITE_OK && pIncr->bUseThread ){
/* Use the current thread to populate aFile[1], even though this
** iterator is multi-threaded. The reason being that this function
** is already running in background thread pIncr->pTask->thread. */
assert( eMode==INCRINIT_ROOT || eMode==INCRINIT_TASK );
rc = vdbeIncrPopulate(pIncr);
}
if( rc==SQLITE_OK && eMode!=INCRINIT_TASK ){
rc = vdbePmaReaderNext(pIter);
}
}
return rc;
}
................................................................................
** of *ppOut is undefined.
*/
static int vdbeSorterMergeTreeBuild(VdbeSorter *pSorter, MergeEngine **ppOut){
MergeEngine *pMain = 0;
int rc = SQLITE_OK;
int iTask;
/* If the sorter uses more than one task, then create the top-level
** MergeEngine here. This MergeEngine will read data from exactly
** one PmaReader per sub-task. */
assert( pSorter->bUseThreads || pSorter->nTask==1 );
if( pSorter->nTask>1 ){
pMain = vdbeMergeEngineNew(pSorter->nTask);
if( pMain==0 ) rc = SQLITE_NOMEM;
}
for(iTask=0; iTask<pSorter->nTask && rc==SQLITE_OK; iTask++){
SortSubtask *pTask = &pSorter->aTask[iTask];
if( pTask->nPMA ){
MergeEngine *pRoot = 0; /* Root node of tree for this task */
int nDepth = vdbeSorterTreeDepth(pTask->nPMA);
i64 iReadOff = 0;
................................................................................
int rc; /* Return code */
assert( pSorter->bUsePMA || (pSorter->pReader==0 && pSorter->pMerger==0) );
if( pSorter->bUsePMA ){
assert( pSorter->pReader==0 || pSorter->pMerger==0 );
assert( pSorter->bUseThreads==0 || pSorter->pReader );
assert( pSorter->bUseThreads==1 || pSorter->pMerger );
if( pSorter->bUseThreads ){
rc = vdbePmaReaderNext(pSorter->pReader);
*pbEof = (pSorter->pReader->pFile==0);
}else{
rc = vdbeSorterNext(&pSorter->aTask[0], pSorter->pMerger, pbEof);
}
}else{
SorterRecord *pFree = pSorter->list.pList;
pSorter->list.pList = pFree->u.pNext;
pFree->u.pNext = 0;
if( pSorter->list.aMemory==0 ) vdbeSorterRecordFree(db, pFree);
................................................................................
*/
static void *vdbeSorterRowkey(
const VdbeSorter *pSorter, /* Sorter object */
int *pnKey /* OUT: Size of current key in bytes */
){
void *pKey;
if( pSorter->bUsePMA ){
PmaReader *pReader = (pSorter->bUseThreads ?
pSorter->pReader : &pSorter->pMerger->aIter[pSorter->pMerger->aTree[1]]
);
*pnKey = pReader->nKey;
pKey = pReader->aKey;
}else{
*pnKey = pSorter->list.pList->nVal;
pKey = SRVAL(pSorter->list.pList);
}
return pKey;
|
| > > > > | | | > > > | > | > > > > > > > < < < < < > > > > > | > > > > | > > > > | < > |
445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 ... 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 ... 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 .... 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 .... 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 .... 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 .... 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 |
** Free all memory belonging to the PmaReader object passed as the second
** argument. All structure fields are set to zero before returning.
*/
static void vdbePmaReaderClear(PmaReader *pIter){
sqlite3_free(pIter->aAlloc);
sqlite3_free(pIter->aBuffer);
if( pIter->aMap ) sqlite3OsUnfetch(pIter->pFile, 0, pIter->aMap);
vdbeIncrFree(pIter->pIncr);
memset(pIter, 0, sizeof(PmaReader));
}
/*
** Read nByte bytes of data from the stream of data iterated by object p.
** If successful, set *ppOut to point to a buffer containing the data
** and return SQLITE_OK. Otherwise, if an error occurs, return an SQLite
................................................................................
int i; /* Used to iterate through aTask[] */
int mxCache; /* Cache size */
VdbeSorter *pSorter; /* The new sorter */
KeyInfo *pKeyInfo; /* Copy of pCsr->pKeyInfo with db==0 */
int szKeyInfo; /* Size of pCsr->pKeyInfo in bytes */
int sz; /* Size of pSorter in bytes */
int rc = SQLITE_OK;
#if SQLITE_MAX_WORKER_THREADS==0
const int nWorker = 0;
#else
int nWorker = (sqlite3GlobalConfig.bCoreMutex?sqlite3GlobalConfig.nWorker:0);
#endif
assert( pCsr->pKeyInfo && pCsr->pBt==0 );
szKeyInfo = sizeof(KeyInfo) + (pCsr->pKeyInfo->nField-1)*sizeof(CollSeq*);
sz = sizeof(VdbeSorter) + nWorker * sizeof(SortSubtask);
pSorter = (VdbeSorter*)sqlite3DbMallocZero(db, sz + szKeyInfo);
pCsr->pSorter = pSorter;
................................................................................
if( !sqlite3TempInMemory(db) ){
pSorter->mnPmaSize = SORTER_MIN_WORKING * pgsz;
mxCache = db->aDb[0].pSchema->cache_size;
if( mxCache<SORTER_MIN_WORKING ) mxCache = SORTER_MIN_WORKING;
pSorter->mxPmaSize = mxCache * pgsz;
/* If the application has not configure scratch memory using
** SQLITE_CONFIG_SCRATCH then we assume it is OK to do large memory
** allocations. If scratch memory has been configured, then assume
** large memory allocations should be avoided to prevent heap
** fragmentation.
*/
if( sqlite3GlobalConfig.pScratch==0 ){
assert( pSorter->iMemory==0 );
pSorter->nMemory = pgsz;
pSorter->list.aMemory = (u8*)sqlite3Malloc(pgsz);
if( !pSorter->list.aMemory ) rc = SQLITE_NOMEM;
}
}
}
................................................................................
rc = vdbeIncrInitMerger(pTask, pIncr->pMerger, eMode);
/* Set up the required files for pIncr. A multi-theaded IncrMerge object
** requires two temp files to itself, whereas a single-threaded object
** only requires a region of pTask->file2. */
if( rc==SQLITE_OK ){
int mxSz = pIncr->mxSz;
#if SQLITE_MAX_WORKER_THREADS>0
if( pIncr->bUseThread ){
rc = vdbeSorterOpenTempFile(db, mxSz, &pIncr->aFile[0].pFd);
if( rc==SQLITE_OK ){
rc = vdbeSorterOpenTempFile(db, mxSz, &pIncr->aFile[1].pFd);
}
}else
#endif
/*if( !pIncr->bUseThread )*/{
if( pTask->file2.pFd==0 ){
assert( pTask->file2.iEof>0 );
rc = vdbeSorterOpenTempFile(db, pTask->file2.iEof, &pTask->file2.pFd);
pTask->file2.iEof = 0;
}
if( rc==SQLITE_OK ){
pIncr->aFile[1].pFd = pTask->file2.pFd;
pIncr->iStartOff = pTask->file2.iEof;
pTask->file2.iEof += mxSz;
}
}
}
#if SQLITE_MAX_WORKER_THREADS>0
if( rc==SQLITE_OK && pIncr->bUseThread ){
/* Use the current thread to populate aFile[1], even though this
** iterator is multi-threaded. The reason being that this function
** is already running in background thread pIncr->pTask->thread. */
assert( eMode==INCRINIT_ROOT || eMode==INCRINIT_TASK );
rc = vdbeIncrPopulate(pIncr);
}
#endif
if( rc==SQLITE_OK && eMode!=INCRINIT_TASK ){
rc = vdbePmaReaderNext(pIter);
}
}
return rc;
}
................................................................................
** of *ppOut is undefined.
*/
static int vdbeSorterMergeTreeBuild(VdbeSorter *pSorter, MergeEngine **ppOut){
MergeEngine *pMain = 0;
int rc = SQLITE_OK;
int iTask;
#if SQLITE_MAX_WORKER_THREADS>0
/* If the sorter uses more than one task, then create the top-level
** MergeEngine here. This MergeEngine will read data from exactly
** one PmaReader per sub-task. */
assert( pSorter->bUseThreads || pSorter->nTask==1 );
if( pSorter->nTask>1 ){
pMain = vdbeMergeEngineNew(pSorter->nTask);
if( pMain==0 ) rc = SQLITE_NOMEM;
}
#endif
for(iTask=0; iTask<pSorter->nTask && rc==SQLITE_OK; iTask++){
SortSubtask *pTask = &pSorter->aTask[iTask];
if( pTask->nPMA ){
MergeEngine *pRoot = 0; /* Root node of tree for this task */
int nDepth = vdbeSorterTreeDepth(pTask->nPMA);
i64 iReadOff = 0;
................................................................................
int rc; /* Return code */
assert( pSorter->bUsePMA || (pSorter->pReader==0 && pSorter->pMerger==0) );
if( pSorter->bUsePMA ){
assert( pSorter->pReader==0 || pSorter->pMerger==0 );
assert( pSorter->bUseThreads==0 || pSorter->pReader );
assert( pSorter->bUseThreads==1 || pSorter->pMerger );
#if SQLITE_MAX_WORKER_THREADS>0
if( pSorter->bUseThreads ){
rc = vdbePmaReaderNext(pSorter->pReader);
*pbEof = (pSorter->pReader->pFile==0);
}else
#endif
/*if( !pSorter->bUseThreads )*/ {
rc = vdbeSorterNext(&pSorter->aTask[0], pSorter->pMerger, pbEof);
}
}else{
SorterRecord *pFree = pSorter->list.pList;
pSorter->list.pList = pFree->u.pNext;
pFree->u.pNext = 0;
if( pSorter->list.aMemory==0 ) vdbeSorterRecordFree(db, pFree);
................................................................................
*/
static void *vdbeSorterRowkey(
const VdbeSorter *pSorter, /* Sorter object */
int *pnKey /* OUT: Size of current key in bytes */
){
void *pKey;
if( pSorter->bUsePMA ){
PmaReader *pReader;
#if SQLITE_MAX_WORKER_THREADS>0
if( pSorter->bUseThreads ){
pReader = pSorter->pReader;
}else
#endif
/*if( !pSorter->bUseThreads )*/{
pReader = &pSorter->pMerger->aIter[pSorter->pMerger->aTree[1]];
}
*pnKey = pReader->nKey;
pKey = pReader->aKey;
}else{
*pnKey = pSorter->list.pList->nVal;
pKey = SRVAL(pSorter->list.pList);
}
return pKey;
|