| #include <stdio.h> |
| #include <stdlib.h> |
| #include <string.h> |
| |
| #include "adc.h" |
| #include "macrodef.h" |
| #include "protots.h" |
| |
| #ifdef UNIX |
| #include <errno.h> |
| #endif |
| |
| extern int32 computeChecksum(ADC_VIEW_CNTL *avp,treeNode *t,uint64 *ordern); |
| extern int32 WriteViewToDiskCS(ADC_VIEW_CNTL *avp,treeNode *t,uint64 *ordern); |
| |
| int32 ReadWholeInputData(ADC_VIEW_CNTL *avp, FILE *inpf){ |
| uint32 iRec = 0; |
| uint32 inpBufferLineSize, inpBufferPace, inpRecSize, ib = 0; |
| |
| FSEEK(inpf, 0L, SEEK_SET); |
| inpRecSize = 8*avp->nm+4*avp->nTopDims; |
| inpBufferLineSize = inpRecSize; |
| if (inpBufferLineSize%8) inpBufferLineSize += 4; |
| inpBufferPace = inpBufferLineSize/4; |
| |
| while(fread(&avp->inpDataBuffer[ib], inpRecSize, 1, inpf)){ |
| iRec++; |
| ib += inpBufferPace; |
| } |
| avp->nRowsToRead = iRec; |
| FSEEK(inpf, 0L, SEEK_SET); |
| |
| if(avp->nInputRecs != iRec){ |
| fprintf(stderr, " ReadWholeInputData(): wrong input data reading.\n"); |
| return ADC_INTERNAL_ERROR; |
| } |
| return ADC_OK; |
| } |
| int32 ComputeMemoryFittedView (ADC_VIEW_CNTL *avp){ |
| uint32 iRec = 0; |
| uint32 viewBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| uint32 inpBufferLineSize, inpBufferPace, inpRecSize,ib; |
| uint64 ordern=0; |
| #ifdef VIEW_FILE_OUTPUT |
| uint32 retCode; |
| #endif |
| |
| FSEEK(avp->viewFile, 0L, SEEK_END); |
| inpRecSize = 8*avp->nm+4*avp->nTopDims; |
| inpBufferLineSize = inpRecSize; |
| if (inpBufferLineSize%8) inpBufferLineSize += 4; |
| inpBufferPace = inpBufferLineSize/4; |
| |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| |
| ib=0; |
| for ( iRec = 1; iRec <= avp->nRowsToRead; iRec++ ){ |
| SelectToView( &avp->inpDataBuffer[ib], avp->selection, viewBuf, |
| avp->nd, avp->nm, avp->nv ); |
| ib += inpBufferPace; |
| TreeInsert(avp->tree, viewBuf); |
| if(avp->tree->memoryIsFull){ |
| fprintf(stderr, "ComputeMemoryFittedView(): Not enough memory.\n"); |
| return 1; |
| } |
| } |
| |
| #ifdef VIEW_FILE_OUTPUT |
| if( retCode = WriteViewToDiskCS(avp, avp->tree->root.left,&ordern) ){ |
| fprintf(stderr, "ComputeMemoryFittedView() Write error is occured.\n"); |
| return retCode; |
| } |
| #else |
| computeChecksum(avp,avp->tree->root.left,&ordern); |
| #endif |
| |
| avp->nViewRows = avp->tree->count; |
| avp->totalOfViewRows += avp->nViewRows; |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| return ADC_OK; |
| } |
| |
| int32 SharedSortAggregate(ADC_VIEW_CNTL *avp){ |
| int32 retCode; |
| uint32 iRec = 0; |
| uint32 attrs[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| uint32 currBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| int64 chunkOffset = 0; |
| int64 inpfOffset; |
| uint32 nPart = 0; |
| uint32 prevV; |
| uint32 currV; |
| uint32 total = 0; |
| unsigned char *ib; |
| uint32 ibsize = SSA_BUFFER_SIZE; |
| uint32 nib; |
| uint32 iib; |
| uint32 nreg; |
| uint32 nlst; |
| uint32 nsgs; |
| uint32 ncur; |
| uint32 ibOffset = 0; |
| uint64 ordern=0; |
| |
| ib = (unsigned char*) malloc(ibsize); |
| if (!ib){ |
| fprintf(stderr,"SharedSortAggregate: memory allocation failed\n"); |
| return ADC_MEMORY_ALLOCATION_FAILURE; |
| } |
| |
| nib = ibsize/avp->inpRecSize; |
| nsgs = avp->nRowsToRead/nib; |
| |
| if (nsgs == 0){ |
| nreg = avp->nRowsToRead; |
| nlst = nreg; |
| nsgs = 1; |
| }else{ |
| nreg = nib; |
| if (avp->nRowsToRead%nib) { |
| nsgs++; |
| nlst = avp->nRowsToRead%nib; |
| }else{ |
| nlst = nreg; |
| } |
| } |
| |
| avp->nViewRows = 0; |
| for( iib = 1; iib <= nsgs; iib++ ){ |
| if(iib > 1) FSEEK(avp->viewFile, inpfOffset, SEEK_SET); |
| if( iib == nsgs ) ncur = nlst; else ncur = nreg; |
| |
| fread(ib, ncur*avp->inpRecSize, 1, avp->viewFile); |
| inpfOffset = ftell(avp->viewFile); |
| |
| for( ibOffset = 0, iRec = 1; iRec <= ncur; iRec++ ){ |
| memcpy(attrs, &ib[ibOffset], avp->inpRecSize); |
| ibOffset += avp->inpRecSize; |
| SelectToView(attrs, avp->selection, currBuf, avp->nd, avp->nm, avp->nv); |
| currV = currBuf[2*avp->nm]; |
| |
| if(iib == 1 && iRec == 1){ |
| prevV = currV; |
| nPart = 1; |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| TreeInsert(avp->tree, currBuf); |
| }else{ |
| if (currV == prevV){ |
| nPart++; |
| TreeInsert (avp->tree, currBuf); |
| if (avp->tree->memoryIsFull){ |
| avp->chunksParams[avp->numberOfChunks].curChunkNum = |
| avp->tree->count; |
| avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset; |
| (avp->numberOfChunks)++; |
| if(avp->numberOfChunks >= MAX_NUM_OF_CHUNKS){ |
| fprintf(stderr,"Too many chunks were created.\n"); |
| exit(1); |
| } |
| chunkOffset += (uint64)(avp->tree->count*avp->outRecSize); |
| retCode=WriteChunkToDisk(avp->outRecSize, avp->fileOfChunks, |
| avp->tree->root.left, avp->logf); |
| if(retCode!=ADC_OK){ |
| fprintf(stderr,"SharedSortAggregate: Write error occured.\n"); |
| return retCode; |
| } |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| } /* memoryIsFull */ |
| }else{ |
| if(avp->numberOfChunks && avp->tree->count!=0){ |
| avp->chunksParams[avp->numberOfChunks].curChunkNum = |
| avp->tree->count; |
| avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset; |
| (avp->numberOfChunks)++; |
| chunkOffset += |
| (uint64)(avp->tree->count*(4*avp->nv + 8*avp->nm)); |
| retCode=WriteChunkToDisk( avp->outRecSize, avp->fileOfChunks, |
| avp->tree->root.left, avp->logf); |
| if(retCode!=ADC_OK){ |
| fprintf(stderr,"SharedSortAggregate: Write error occured.\n"); |
| return retCode; |
| } |
| } |
| FSEEK(avp->viewFile, 0L, SEEK_END); |
| if(!avp->numberOfChunks){ |
| avp->nViewRows += avp->tree->count; |
| retCode = WriteViewToDiskCS(avp, avp->tree->root.left,&ordern); |
| if(retCode!=ADC_OK){ |
| fprintf(stderr, |
| "SharedSortAggregate: Write error occured.\n"); |
| return retCode; |
| } |
| }else{ |
| retCode=MultiWayMerge(avp); |
| if(retCode!=ADC_OK) { |
| fprintf(stderr,"SharedSortAggregate.MultiWayMerge: failed.\n"); |
| return retCode; |
| } |
| } |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| TreeInsert(avp->tree, currBuf); |
| total += nPart; |
| nPart = 1; |
| } |
| } |
| prevV = currV; |
| } /* iRec */ |
| } /* iib */ |
| |
| if(avp->numberOfChunks && avp->tree->count!=0) { |
| avp->chunksParams[avp->numberOfChunks].curChunkNum = avp->tree->count; |
| avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset; |
| (avp->numberOfChunks)++; |
| chunkOffset += (uint64)(avp->tree->count*(4*avp->nv + 8*avp->nm)); |
| retCode=WriteChunkToDisk(avp->outRecSize, avp->fileOfChunks, |
| avp->tree->root.left, avp->logf); |
| if(retCode!=ADC_OK){ |
| fprintf(stderr,"SharedSortAggregate: Write error occured.\n"); |
| return retCode; |
| } |
| } |
| FSEEK(avp->viewFile, 0L, SEEK_END); |
| if(!avp->numberOfChunks){ |
| avp->nViewRows += avp->tree->count; |
| if( retCode = WriteViewToDiskCS(avp, avp->tree->root.left,&ordern)){ |
| fprintf(stderr, "SharedSortAggregate: Write error occured.\n"); |
| return retCode; |
| } |
| }else{ |
| retCode=MultiWayMerge(avp); |
| if(retCode!=ADC_OK) { |
| fprintf(stderr,"SharedSortAggregate.MultiWayMerge failed.\n"); |
| return retCode; |
| } |
| } |
| FSEEK(avp->fileOfChunks, 0L, SEEK_SET); |
| |
| total += nPart; |
| avp->totalOfViewRows += avp->nViewRows; |
| if(ib) free(ib); |
| return ADC_OK; |
| } |
| int32 PrefixedAggregate(ADC_VIEW_CNTL *avp, FILE *iof){ |
| uint32 i; |
| uint32 iRec = 0; |
| uint32 attrs[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| uint32 aggrBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| uint32 currBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| uint32 prevBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| int64 *aggrmp; |
| int64 *currmp; |
| int32 compRes; |
| uint32 nOut = 0; |
| uint32 mpOffset = 0; |
| uint32 nOutBufRecs; |
| uint32 nViewRows = 0; |
| int64 inpfOffset; |
| |
| aggrmp = (int64*) &aggrBuf[0]; |
| currmp = (int64*) &currBuf[0]; |
| |
| for(i = 0; i < 2*avp->nm+avp->nv; i++){prevBuf[i] = 0; aggrBuf[i] = 0;} |
| nOutBufRecs = avp->memoryLimit/avp->outRecSize; |
| |
| for(iRec = 1; iRec <= avp->nRowsToRead; iRec++ ){ |
| fread(attrs, avp->inpRecSize, 1, iof); |
| SelectToView(attrs, avp->selection, currBuf, avp->nd, avp->nm, avp->nv); |
| if (iRec == 1) memcpy(aggrBuf, currBuf, avp->outRecSize); |
| else{ |
| compRes = KeyComp( &currBuf[2*avp->nm], &prevBuf[2*avp->nm], avp->nv); |
| |
| switch(compRes){ |
| case 1: |
| memcpy(&avp->memPool[mpOffset], aggrBuf, avp->outRecSize); |
| mpOffset += avp->outRecSize; |
| nOut++; |
| for ( i = 0; i < avp->nm; i++ ){ |
| avp->mSums[i] += aggrmp[i]; |
| avp->checksums[i] += nOut*aggrmp[i]%measbound; |
| } |
| memcpy(aggrBuf, currBuf, avp->outRecSize); |
| break; |
| case 0: |
| for ( i = 0; i < avp->nm; i++ ) aggrmp[i] += currmp[i]; |
| break; |
| case -1: |
| fprintf(stderr,"PrefixedAggregate: wrong parent view order.\n"); |
| exit(1); |
| break; |
| default: |
| fprintf(stderr,"PrefixedAggregate: wrong KeyComp() result.\n"); |
| exit(1); |
| break; |
| } |
| |
| if (nOut == nOutBufRecs){ |
| inpfOffset = ftell(iof); |
| FSEEK(iof, 0L, SEEK_END); |
| WriteToFile(avp->memPool, nOut*avp->outRecSize, 1, iof, stderr); |
| FSEEK(iof, inpfOffset, SEEK_SET); |
| mpOffset = 0; |
| nViewRows += nOut; |
| nOut = 0; |
| } |
| } |
| memcpy(prevBuf, currBuf, avp->outRecSize); |
| } |
| memcpy(&avp->memPool[mpOffset], aggrBuf, avp->outRecSize); |
| nOut++; |
| for ( i = 0; i < avp->nm; i++ ){ |
| avp->mSums[i] += aggrmp[i]; |
| avp->checksums[i] += nOut*aggrmp[i]%measbound; |
| } |
| FSEEK(iof, 0L, SEEK_END); |
| WriteToFile(avp->memPool, nOut*avp->outRecSize, 1, iof, stderr); |
| avp->nViewRows = nViewRows+nOut; |
| avp->totalOfViewRows += avp->nViewRows; |
| return ADC_OK; |
| } |
| int32 RunFormation (ADC_VIEW_CNTL *avp, FILE *inpf){ |
| uint32 iRec = 0; |
| uint32 viewBuf[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| uint32 attrs[MAX_VIEW_ROW_SIZE_IN_INTS]; |
| int64 chunkOffset = 0; |
| |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| |
| for(iRec = 1; iRec <= avp->nRowsToRead; iRec++ ){ |
| fread(attrs, avp->inpRecSize, 1, inpf); |
| SelectToView(attrs, avp->selection, viewBuf, avp->nd, avp->nm, avp->nv); |
| TreeInsert(avp->tree, viewBuf); |
| |
| if(avp->tree->memoryIsFull) { |
| avp->chunksParams[avp->numberOfChunks].curChunkNum = avp->tree->count; |
| avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset; |
| (avp->numberOfChunks)++; |
| if (avp->numberOfChunks >= MAX_NUM_OF_CHUNKS) { |
| fprintf(stderr, "RunFormation: Too many chunks were created.\n"); |
| return ADC_INTERNAL_ERROR; |
| } |
| chunkOffset += (uint64)(avp->tree->count*avp->outRecSize); |
| if(WriteChunkToDisk( avp->outRecSize, avp->fileOfChunks, |
| avp->tree->root.left, avp->logf )){ |
| fprintf(stderr, |
| "RunFormation.WriteChunkToDisk: Write error is occured.\n"); |
| return ADC_WRITE_FAILED; |
| } |
| InitializeTree(avp->tree, avp->nv, avp->nm); |
| } |
| } /* Insertion ... */ |
| if(avp->numberOfChunks && avp->tree->count!=0) { |
| avp->chunksParams[avp->numberOfChunks].curChunkNum = avp->tree->count; |
| avp->chunksParams[avp->numberOfChunks].chunkOffset = chunkOffset; |
| (avp->numberOfChunks)++; |
| chunkOffset += (uint64)(avp->tree->count*(4*avp->nv + 8*avp->nm)); |
| if(WriteChunkToDisk(avp->outRecSize, avp->fileOfChunks, |
| avp->tree->root.left, avp->logf)){ |
| fprintf(stderr, |
| "RunFormation(.WriteChunkToDisk: Write error is occured.\n"); |
| return ADC_WRITE_FAILED; |
| } |
| } |
| FSEEK(avp->viewFile, 0L, SEEK_END); |
| return ADC_OK; |
| } |
| void SeekAndReadNextSubChunk( uint32 multiChunkBuffer[], |
| uint32 k, |
| FILE *inFile, |
| uint32 chunkRecSize, |
| uint64 inFileOffs, |
| uint32 subChunkNum){ |
| int64 ret; |
| |
| ret = FSEEK(inFile, inFileOffs, SEEK_SET); |
| if (ret < 0){ |
| fprintf(stderr,"SeekAndReadNextSubChunk.fseek() < 0 "); |
| exit(1); |
| } |
| fread(&multiChunkBuffer[k], chunkRecSize*subChunkNum, 1, inFile); |
| } |
| void ReadSubChunk( |
| uint32 chunkRecSize, |
| uint32 *multiChunkBuffer, |
| uint32 mwBufRecSizeInInt, |
| uint32 iChunk, |
| uint32 regSubChunkSize, |
| CHUNKS *chunks, |
| FILE *fileOfChunks |
| ){ |
| if (chunks[iChunk].curChunkNum > 0){ |
| if(chunks[iChunk].curChunkNum < regSubChunkSize){ |
| SeekAndReadNextSubChunk(multiChunkBuffer, |
| (iChunk*regSubChunkSize + |
| (regSubChunkSize-chunks[iChunk].curChunkNum))* |
| mwBufRecSizeInInt, |
| fileOfChunks, |
| chunkRecSize, |
| chunks[iChunk].chunkOffset, |
| chunks[iChunk].curChunkNum); |
| chunks[iChunk].posSubChunk=regSubChunkSize-chunks[iChunk].curChunkNum; |
| chunks[iChunk].curSubChunk=chunks[iChunk].curChunkNum; |
| chunks[iChunk].curChunkNum=0; |
| chunks[iChunk].chunkOffset=-1; |
| }else{ |
| SeekAndReadNextSubChunk(multiChunkBuffer, |
| iChunk*regSubChunkSize*mwBufRecSizeInInt, |
| fileOfChunks, |
| chunkRecSize, |
| chunks[iChunk].chunkOffset, |
| regSubChunkSize); |
| chunks[iChunk].posSubChunk = 0; |
| chunks[iChunk].curSubChunk = regSubChunkSize; |
| chunks[iChunk].curChunkNum -= regSubChunkSize; |
| chunks[iChunk].chunkOffset += regSubChunkSize * chunkRecSize; |
| } |
| } |
| } |
| int32 MultiWayMerge(ADC_VIEW_CNTL *avp){ |
| uint32 outputBuffer[OUTPUT_BUFFER_SIZE]; |
| uint32 r_buf [OUTPUT_BUFFER_SIZE]; |
| uint32 min_r_buf [OUTPUT_BUFFER_SIZE]; |
| uint32 first_one; |
| uint32 i; |
| uint32 iChunk; |
| uint32 min_r_chunk; |
| uint32 sPos; |
| uint32 iPos; |
| uint32 numEmptyBufs; |
| uint32 numEmptyRuns; |
| uint32 mwBufRecSizeInInt; |
| uint32 chunkRecSize; |
| uint32 *multiChunkBuffer; |
| uint32 regSubChunkSize; |
| int32 compRes; |
| int64 *m_min_r_buf; |
| int64 *m_outputBuffer; |
| |
| FSEEK(avp->fileOfChunks, 0L, SEEK_SET); |
| |
| multiChunkBuffer = (uint32*) &avp->memPool[0]; |
| first_one = 1; |
| avp->nViewRows = 0; |
| |
| chunkRecSize = avp->outRecSize; |
| mwBufRecSizeInInt = chunkRecSize/4; |
| m_min_r_buf = (int64*)&min_r_buf[0]; |
| m_outputBuffer = (int64*)&outputBuffer[0]; |
| |
| mwBufRecSizeInInt = chunkRecSize/4; |
| regSubChunkSize = (avp->memoryLimit/avp->numberOfChunks)/chunkRecSize; |
| |
| if (regSubChunkSize==0) { |
| fprintf(stderr, |
| "MultiWayMerge: Not enough memory to run the external sort\n"); |
| return ADC_INTERNAL_ERROR; |
| } |
| multiChunkBuffer = (uint32*) &avp->memPool[0]; |
| |
| for(i = 0; i < avp->numberOfChunks; i++ ){ |
| ReadSubChunk( |
| chunkRecSize, |
| multiChunkBuffer, |
| mwBufRecSizeInInt, |
| i, |
| regSubChunkSize, |
| avp->chunksParams, |
| avp->fileOfChunks |
| ); |
| } |
| while(1){ |
| for(iChunk = 0;iChunk<avp->numberOfChunks;iChunk++){ |
| if (avp->chunksParams[iChunk].curSubChunk > 0){ |
| sPos = iChunk*regSubChunkSize*mwBufRecSizeInInt; |
| iPos = sPos+mwBufRecSizeInInt*avp->chunksParams[iChunk].posSubChunk; |
| memcpy(&min_r_buf[0], &multiChunkBuffer[iPos], avp->outRecSize); |
| min_r_chunk = iChunk; |
| break; |
| } |
| } |
| for ( iChunk = min_r_chunk; iChunk < avp->numberOfChunks; iChunk++ ){ |
| uint32 iPos; |
| |
| if (avp->chunksParams[iChunk].curSubChunk > 0){ |
| iPos = mwBufRecSizeInInt*(iChunk*regSubChunkSize+ |
| avp->chunksParams[iChunk].posSubChunk); |
| memcpy(&r_buf[0],&multiChunkBuffer[iPos],avp->outRecSize); |
| |
| compRes=KeyComp(&r_buf[2*avp->nm],&min_r_buf[2*avp->nm],avp->nv); |
| if(compRes < 0) { |
| memcpy(&min_r_buf[0], &r_buf[0], avp->outRecSize); |
| min_r_chunk = iChunk; |
| } |
| } |
| } |
| /* Step forward */ |
| if(avp->chunksParams[min_r_chunk].curSubChunk != 0){ |
| avp->chunksParams[min_r_chunk].curSubChunk--; |
| avp->chunksParams[min_r_chunk].posSubChunk++; |
| } |
| |
| /* Aggreagation if a duplicate is encountered */ |
| if(first_one){ |
| memcpy( &outputBuffer[0], &min_r_buf[0], avp->outRecSize); |
| first_one = 0; |
| }else{ |
| compRes = KeyComp( &outputBuffer[2*avp->nm], |
| &min_r_buf[2*avp->nm], avp->nv ); |
| if(!compRes){ |
| for(i = 0; i < avp->nm; i++ ){ |
| m_outputBuffer[i] += m_min_r_buf[i]; |
| } |
| }else{ |
| WriteToFile(outputBuffer,avp->outRecSize,1,avp->viewFile,stderr); |
| avp->nViewRows++; |
| for(i=0;i<avp->nm;i++){ |
| avp->mSums[i]+=m_outputBuffer[i]; |
| avp->checksums[i] += avp->nViewRows*m_outputBuffer[i]%measbound; |
| } |
| memcpy( &outputBuffer[0], &min_r_buf[0], avp->outRecSize ); |
| } |
| } |
| |
| for(numEmptyBufs = 0, |
| numEmptyRuns = 0, i = 0; i < avp->numberOfChunks; i++ ){ |
| if (avp->chunksParams[i].curSubChunk == 0) numEmptyBufs++; |
| if (avp->chunksParams[i].curChunkNum == 0) numEmptyRuns++; |
| } |
| if( numEmptyBufs == avp->numberOfChunks |
| &&numEmptyRuns == avp->numberOfChunks) break; |
| |
| if(avp->chunksParams[min_r_chunk].curSubChunk == 0) { |
| ReadSubChunk( |
| chunkRecSize, |
| multiChunkBuffer, |
| mwBufRecSizeInInt, |
| min_r_chunk, |
| regSubChunkSize, |
| avp->chunksParams, |
| avp->fileOfChunks); |
| } |
| } /* while(1) */ |
| |
| WriteToFile( outputBuffer, avp->outRecSize, 1, avp->viewFile, stderr); |
| avp->nViewRows++; |
| for(i = 0; i < avp->nm; i++ ){ |
| avp->mSums[i] += m_outputBuffer[i]; |
| avp->checksums[i] += avp->nViewRows*m_outputBuffer[i]%measbound; |
| } |
| |
| avp->totalOfViewRows += avp->nViewRows; |
| return ADC_OK; |
| } |
| void SelectToView( uint32 * ib, uint32 *ix, uint32 *viewBuf, |
| uint32 nd, uint32 nm, uint32 nv ){ |
| uint32 i, j; |
| for ( j = 0, i = 0; i < nv; i++ ) viewBuf[2*nm+j++] = ib[2*nm+ix[i]-1]; |
| memcpy(&viewBuf[0], &ib[0], MSR_FSZ*nm); |
| } |
| FILE * AdcFileOpen(const char *fileName, const char *mode){ |
| FILE *fr; |
| if ((fr = (FILE*) fopen(fileName, mode))==NULL) |
| fprintf(stderr, "AdcFileOpen: Cannot open the file %s errno = %d\n", |
| fileName, errno); |
| return fr; |
| } |
| void AdcFileName(char *adcFileName, const char *adcName, |
| const char *fileName, uint32 taskNumber){ |
| sprintf(adcFileName, "%s.%s.%d",adcName,fileName,taskNumber); |
| } |
| ADC_VIEW_CNTL * NewAdcViewCntl(ADC_VIEW_PARS *adcpp, uint32 pnum){ |
| ADC_VIEW_CNTL *adccntl; |
| uint32 i, j, k; |
| #ifdef IN_CORE |
| uint32 ux; |
| #endif |
| char id[8+1]; |
| |
| adccntl = (ADC_VIEW_CNTL *) malloc(sizeof(ADC_VIEW_CNTL)); |
| if (adccntl==NULL) return NULL; |
| |
| adccntl->ndid = adcpp->ndid; |
| adccntl->taskNumber = pnum; |
| adccntl->retCode = 0; |
| adccntl->swapIt = 0; |
| strcpy(adccntl->adcName, adcpp->adcName); |
| adccntl->nTopDims = adcpp->nd; |
| adccntl->nd = adcpp->nd; |
| adccntl->nm = adcpp->nm; |
| adccntl->nInputRecs = adcpp->nInputRecs; |
| adccntl->inpRecSize = GetRecSize(adccntl->nd,adccntl->nm); |
| adccntl->outRecSize = GetRecSize(adccntl->nv,adccntl->nm); |
| adccntl->accViewFileOffset = 0; |
| adccntl->totalViewFileSize = 0; |
| adccntl->numberOfMadeViews = 0; |
| adccntl->numberOfViewsMadeFromInput = 0; |
| adccntl->numberOfPrefixedGroupbys = 0; |
| adccntl->numberOfSharedSortGroupbys = 0; |
| adccntl->totalOfViewRows = 0; |
| adccntl->memoryLimit = adcpp->memoryLimit; |
| adccntl->nTasks = adcpp->nTasks; |
| strcpy(adccntl->inpFileName, adcpp->adcInpFileName); |
| sprintf(id, ".%d", adcpp->ndid); |
| |
| AdcFileName(adccntl->adcLogFileName, |
| adccntl->adcName, "logf", adccntl->taskNumber); |
| strcat(adccntl->adcLogFileName, id); |
| adccntl->logf = AdcFileOpen(adccntl->adcLogFileName, "w"); |
| |
| AdcFileName(adccntl->inpFileName, adccntl->adcName, "dat", adcpp->ndid); |
| adccntl->inpf = AdcFileOpen(adccntl->inpFileName, "rb"); |
| if(!adccntl->inpf){ |
| adccntl->retCode = ADC_FILE_OPEN_FAILURE; |
| return(adccntl); |
| } |
| |
| AdcFileName(adccntl->viewFileName, adccntl->adcName, |
| "view.dat", adccntl->taskNumber); |
| strcat(adccntl->viewFileName, id); |
| adccntl->viewFile = AdcFileOpen(adccntl->viewFileName, "wb+"); |
| |
| AdcFileName(adccntl->chunksFileName, adccntl->adcName, |
| "chunks.dat", adccntl->taskNumber); |
| strcat(adccntl->chunksFileName, id); |
| adccntl->fileOfChunks = AdcFileOpen(adccntl->chunksFileName,"wb+"); |
| |
| AdcFileName(adccntl->groupbyFileName, adccntl->adcName, |
| "groupby.dat", adccntl->taskNumber); |
| strcat(adccntl->groupbyFileName, id); |
| adccntl->groupbyFile = AdcFileOpen(adccntl->groupbyFileName,"wb+"); |
| |
| AdcFileName(adccntl->adcViewSizesFileName, adccntl->adcName, |
| "view.sz", adcpp->ndid); |
| adccntl->adcViewSizesFile = AdcFileOpen(adccntl->adcViewSizesFileName,"r"); |
| if(!adccntl->adcViewSizesFile){ |
| adccntl->retCode = ADC_FILE_OPEN_FAILURE; |
| return(adccntl); |
| } |
| |
| AdcFileName(adccntl->viewSizesFileName, adccntl->adcName, |
| "viewsz.dat", adccntl->taskNumber); |
| strcat(adccntl->viewSizesFileName, id); |
| adccntl->viewSizesFile = AdcFileOpen(adccntl->viewSizesFileName, "wb+"); |
| |
| adccntl->chunksParams = (CHUNKS*) malloc(MAX_NUM_OF_CHUNKS*sizeof(CHUNKS)); |
| if(adccntl->chunksParams==NULL){ |
| fprintf(adccntl->logf,"NewAdcViewCntl: Cannot allocate 'chunksParsms'\n"); |
| adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; |
| return(adccntl); |
| } |
| adccntl->memPool = (unsigned char*) malloc(adccntl->memoryLimit); |
| if(adccntl->memPool == NULL ){ |
| fprintf(adccntl->logf, |
| "NewAdcViewCntl: Cannot allocate 'main memory pool'\n"); |
| adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; |
| return(adccntl); |
| } |
| |
| #ifdef IN_CORE |
| /* add a condition to allocate this memory buffer, THIS is IMPORTANT */ |
| ux = 4*adccntl->nTopDims + 8*adccntl->nm; |
| if (adccntl->nTopDims%8) ux += 4; |
| adccntl->inpDataBuffer = (uint32*) malloc(adccntl->nInputRecs*ux); |
| if(adccntl->inpDataBuffer == NULL ){ |
| fprintf(adccntl->logf, |
| "NewAdcViewCntl: Cannot allocate 'input data buffer'\n"); |
| adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; |
| return(adccntl); |
| } |
| #endif |
| adccntl->numberOfChunks = 0; |
| |
| for ( i = 0; i < adccntl->nm; i++ ){ |
| adccntl->mSums[i] = 0; |
| adccntl->checksums[i] = 0; |
| adccntl->totchs[i] = 0; |
| } |
| adccntl->tree = CreateEmptyTree(adccntl->nd, adccntl->nm, |
| adccntl->memoryLimit, adccntl->memPool); |
| if(!adccntl->tree){ |
| fprintf(adccntl->logf,"\nNewAdcViewCntl.CreateEmptyTree failed.\n"); |
| adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; |
| return(adccntl); |
| } |
| |
| adccntl->nv = adcpp->nd; /* default */ |
| for ( i = 0; i < adccntl->nv; i++ ) adccntl->selection[i]=i+1; |
| |
| adccntl->nViewLimit = (1<<adcpp->nd)-1; |
| adccntl->jpp=(JOB_POOL *) malloc((adccntl->nViewLimit+1)*sizeof(JOB_POOL)); |
| if ( adccntl->jpp == NULL){ |
| fprintf(adccntl->logf, |
| "\n Not enough space to allocate %ld byte for a job pool.", |
| (long)(adccntl->nViewLimit+1)*sizeof(JOB_POOL)); |
| adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; |
| return(adccntl); |
| } |
| adccntl->lpp = (LAYER * ) malloc( (adcpp->nd+1)*sizeof(LAYER)); |
| if ( adccntl->lpp == NULL){ |
| fprintf(adccntl->logf, |
| "\n Not enough space to allocate %ld byte for a layer reference array.", |
| (long)(adcpp->nd+1)*sizeof(LAYER)); |
| adccntl->retCode = ADC_MEMORY_ALLOCATION_FAILURE; |
| return(adccntl); |
| } |
| |
| for ( j = 1, i = 1; i <= adcpp->nd; i++ ) { |
| k = NumOfCombsFromNbyK ( adcpp->nd, i ); |
| adccntl->lpp[i].layerIndex = j; |
| j += k; |
| adccntl->lpp[i].layerQuantityLimit = k; |
| adccntl->lpp[i].layerCurrentPopulation = 0; |
| } |
| |
| JobPoolInit ( adccntl->jpp, (adccntl->nViewLimit+1), adcpp->nd ); |
| |
| fprintf(adccntl->logf,"\nMeaning of the log file colums is as follows:\n"); |
| fprintf(adccntl->logf, |
| "Row Number | Groupby | View Size | Measure Sums | Number of Chunks\n"); |
| |
| adccntl->verificationFailed = 1; |
| return adccntl; |
| } |
| void InitAdcViewCntl(ADC_VIEW_CNTL *adccntl, |
| uint32 nSelectedDims, |
| uint32 *selection, |
| uint32 fromParent ){ |
| uint32 i; |
| |
| adccntl->nv = nSelectedDims; |
| |
| for (i = 0; i < adccntl->nm; i++ ) adccntl->mSums[i] = 0; |
| for (i = 0; i < adccntl->nv; i++ ) adccntl->selection[i] = selection[i]; |
| |
| adccntl->outRecSize = GetRecSize(adccntl->nv,adccntl->nm); |
| adccntl->numberOfChunks = 0; |
| adccntl->fromParent = fromParent; |
| adccntl->nViewRows = 0; |
| |
| if(fromParent){ |
| adccntl->nd = adccntl->smallestParentLevel; |
| FSEEK(adccntl->viewFile, adccntl->viewOffset, SEEK_SET); |
| adccntl->nRowsToRead = adccntl->nParentViewRows; |
| }else{ |
| adccntl->nd = adccntl->nTopDims; |
| adccntl->nRowsToRead = adccntl->nInputRecs; |
| } |
| adccntl->inpRecSize = GetRecSize(adccntl->nd,adccntl->nm); |
| adccntl->outRecSize = GetRecSize(adccntl->nv,adccntl->nm); |
| } |
| int32 CloseAdcView(ADC_VIEW_CNTL *adccntl){ |
| if (adccntl->inpf) fclose(adccntl->inpf); |
| if (adccntl->viewFile) fclose(adccntl->viewFile); |
| if (adccntl->fileOfChunks) fclose(adccntl->fileOfChunks); |
| if (adccntl->groupbyFile) fclose(adccntl->groupbyFile); |
| if (adccntl->adcViewSizesFile) fclose(adccntl->adcViewSizesFile); |
| if (adccntl->viewSizesFile) fclose(adccntl->viewSizesFile); |
| |
| if (DeleteOneFile(adccntl->chunksFileName)) |
| return ADC_FILE_DELETE_FAILURE; |
| if (DeleteOneFile(adccntl->viewSizesFileName)) |
| return ADC_FILE_DELETE_FAILURE; |
| |
| if (DeleteOneFile(adccntl->groupbyFileName)) |
| return ADC_FILE_DELETE_FAILURE; |
| |
| if (adccntl->chunksParams){ |
| free(adccntl->chunksParams); |
| adccntl->chunksParams=NULL; |
| } |
| if (adccntl->memPool){ free(adccntl->memPool); adccntl->memPool=NULL;} |
| if (adccntl->jpp){ free(adccntl->jpp); adccntl->jpp=NULL; } |
| if (adccntl->lpp){ free(adccntl->lpp); adccntl->lpp=NULL; } |
| |
| if (adccntl->logf) fclose(adccntl->logf); |
| free(adccntl); |
| return ADC_OK; |
| } |
| void AdcCntlLog(ADC_VIEW_CNTL *adccntlp){ |
| fprintf(adccntlp->logf," memoryLimit = %20d\n", |
| adccntlp->memoryLimit); |
| fprintf(adccntlp->logf," treeNodeSize = %20d\n", |
| adccntlp->tree->treeNodeSize); |
| fprintf(adccntlp->logf," treeMemoryLimit = %20d\n", |
| adccntlp->tree->memoryLimit); |
| fprintf(adccntlp->logf," nNodesLimit = %20d\n", |
| adccntlp->tree->nNodesLimit); |
| fprintf(adccntlp->logf,"freeNodeCounter = %20d\n", |
| adccntlp->tree->freeNodeCounter); |
| fprintf(adccntlp->logf," nViewRows = %20d\n", |
| adccntlp->nViewRows); |
| } |
| int32 ViewSizesVerification(ADC_VIEW_CNTL *adccntlp){ |
| char inps[MAX_PARAM_LINE_SIZE]; |
| char msg[64]; |
| uint32 *viewCounts; |
| uint32 selection_viewSize[2]; |
| uint32 sz; |
| uint32 sel[64]; |
| uint32 i; |
| uint32 k; |
| uint64 tx; |
| uint32 iTx; |
| |
| viewCounts = (uint32 *) &adccntlp->memPool[0]; |
| for ( i = 0; i <= adccntlp->nViewLimit; i++) viewCounts[i] = 0; |
| |
| FSEEK(adccntlp->viewSizesFile, 0L, SEEK_SET); |
| FSEEK(adccntlp->adcViewSizesFile, 0L, SEEK_SET); |
| |
| while(fread(selection_viewSize, 8, 1, adccntlp->viewSizesFile)){ |
| viewCounts[selection_viewSize[0]] = selection_viewSize[1]; |
| } |
| k = 0; |
| while ( fscanf(adccntlp->adcViewSizesFile, "%s", inps) != EOF ){ |
| if ( strcmp(inps, "Selection:") == 0 ) { |
| while ( fscanf(adccntlp->adcViewSizesFile, "%s", inps)) { |
| if ( strcmp(inps, "View") == 0 ) break; |
| sel[k++] = atoi(inps); |
| } |
| } |
| |
| if ( strcmp(inps, "Size:") == 0 ) { |
| fscanf(adccntlp->adcViewSizesFile, "%s", inps); |
| sz = atoi(inps); |
| CreateBinTuple(&tx, sel, k); |
| iTx = (int32)(tx>>(64-adccntlp->nTopDims)); |
| adccntlp->verificationFailed = 0; |
| if (!adccntlp->numberOfMadeViews) adccntlp->verificationFailed = 1; |
| |
| if ( viewCounts[iTx] != 0){ |
| if (viewCounts[iTx] != sz) { |
| if (viewCounts[iTx] != adccntlp->nInputRecs){ |
| fprintf(adccntlp->logf, |
| "A view size is wrong: genSz=%d calcSz=%d\n", |
| sz, viewCounts[iTx]); |
| adccntlp->verificationFailed = 1; |
| return ADC_VERIFICATION_FAILED; |
| } |
| } |
| } |
| k = 0; |
| } |
| } /* of while() */ |
| |
| fprintf(adccntlp->logf, |
| "\n\nMeaning of the log file colums is as follows:\n"); |
| fprintf(adccntlp->logf, |
| "Row Number | Groupby | View Size | Measure Sums | Number of Chunks\n"); |
| |
| if (!adccntlp->verificationFailed) |
| strcpy(msg, "Verification=passed"); |
| else strcpy(msg, "Verification=failed"); |
| FSEEK(adccntlp->logf, 0L, SEEK_SET); |
| fprintf(adccntlp->logf, msg); |
| FSEEK(adccntlp->logf, 0L, SEEK_END); |
| FSEEK(adccntlp->viewSizesFile, 0L, SEEK_SET); |
| return ADC_OK; |
| } |
| int32 ComputeGivenGroupbys(ADC_VIEW_CNTL *adccntlp){ |
| int32 retCode; |
| uint32 i; |
| uint64 binRepTuple; |
| uint32 ut32; |
| uint32 nViews = 0; |
| uint32 nSelectedDims; |
| uint32 smp; |
| #ifdef IN_CORE |
| uint32 firstView = 1; |
| #endif |
| uint32 selection_viewsize[2]; |
| char ttout[16]; |
| |
| while (fread(&binRepTuple, 8, 1, adccntlp->groupbyFile )){ |
| for(i = 0; i < adccntlp->nm; i++) adccntlp->checksums[i]=0; |
| nViews++; |
| swap8(&binRepTuple); |
| |
| GetRegTupleFromBin64(binRepTuple, adccntlp->selection, |
| adccntlp->nTopDims, &nSelectedDims); |
| ut32 = (uint32)(binRepTuple>>(64-adccntlp->nTopDims)); |
| selection_viewsize[0] = ut32; |
| ut32 <<= (32-adccntlp->nTopDims); |
| adccntlp->groupby = ut32; |
| #ifndef IN_CORE |
| smp = GetParent(adccntlp, ut32); |
| #endif |
| #ifdef IN_CORE |
| if (firstView) { |
| firstView = 0; |
| if(ReadWholeInputData(adccntlp, adccntlp->inpf)) { |
| fprintf(stderr, "ReadWholeInputData failed.\n"); |
| return ADC_INTERNAL_ERROR; |
| } |
| } |
| smp = noneParent; |
| #endif |
| |
| if (smp != noneParent) |
| GetRegTupleFromParent(binRepTuple, |
| adccntlp->parBinRepTuple, |
| adccntlp->selection, |
| adccntlp->nTopDims); |
| InitAdcViewCntl(adccntlp, nSelectedDims, |
| adccntlp->selection, (smp == noneParent)?0:1); |
| #ifdef IN_CORE |
| if(retCode = ComputeMemoryFittedView(adccntlp)) { |
| fprintf(stderr, "ComputeMemoryFittedView failed.\n"); |
| return retCode; |
| } |
| #else |
| #ifdef OPTIMIZATION |
| if (smp == prefixedParent){ |
| if (retCode = PrefixedAggregate(adccntlp, adccntlp->viewFile)) { |
| fprintf(stderr, |
| "ComputeGivenGroupbys.PrefixedAggregate failed.\n"); |
| return retCode; |
| } |
| adccntlp->numberOfPrefixedGroupbys++; |
| }else if (smp == sharedSortParent) { |
| if (retCode = SharedSortAggregate(adccntlp)) { |
| fprintf(stderr, |
| "ComputeGivenGroupbys.SharedSortAggregate failed.\n"); |
| return retCode; |
| } |
| adccntlp->numberOfSharedSortGroupbys++; |
| }else |
| #endif /* OPTIMIZATION */ |
| { |
| if( smp != noneParent ) { |
| retCode = RunFormation(adccntlp, adccntlp->viewFile); |
| if(retCode!=ADC_OK){ |
| fprintf(stderr, |
| "ComputrGivenGroupbys.RunFormation failed.\n"); |
| return retCode; |
| } |
| }else{ |
| if ((retCode=RunFormation (adccntlp, adccntlp->inpf)) != ADC_OK){ |
| fprintf(stderr, |
| "ComputrGivenGroupbys.RunFormation failed.\n"); |
| return retCode; |
| } |
| adccntlp->numberOfViewsMadeFromInput++; |
| } |
| if(!adccntlp->numberOfChunks){ |
| uint64 ordern=0; |
| adccntlp->nViewRows = adccntlp->tree->count; |
| adccntlp->totalOfViewRows += adccntlp->nViewRows; |
| retCode=WriteViewToDiskCS(adccntlp,adccntlp->tree->root.left,&ordern); |
| if(retCode!=ADC_OK){ |
| fprintf(stderr, |
| "ComputeGivenGroupbys.WriteViewToDisk: Write error.\n"); |
| return ADC_WRITE_FAILED; |
| } |
| }else { |
| retCode=MultiWayMerge(adccntlp); |
| if(retCode!=ADC_OK) { |
| fprintf(stderr,"ComputeGivenGroupbys.MultiWayMerge failed.\n"); |
| return retCode; |
| } |
| } |
| } |
| |
| JobPoolUpdate(adccntlp); |
| |
| adccntlp->accViewFileOffset += |
| (int64)(adccntlp->nViewRows*adccntlp->outRecSize); |
| FSEEK(adccntlp->fileOfChunks, 0L, SEEK_SET); |
| FSEEK(adccntlp->inpf, 0L, SEEK_SET); |
| #endif /* IN_CORE */ |
| for( i = 0; i < adccntlp->nm; i++) |
| adccntlp->totchs[i]+=adccntlp->checksums[i]; |
| selection_viewsize[1] = adccntlp->nViewRows; |
| fwrite(selection_viewsize, 8, 1, adccntlp->viewSizesFile); |
| adccntlp->totalViewFileSize += |
| adccntlp->outRecSize*adccntlp->nViewRows; |
| sprintf(ttout, "%7d ", nViews); |
| WriteOne32Tuple(ttout, adccntlp->groupby, |
| adccntlp->nTopDims, adccntlp->logf); |
| fprintf(adccntlp->logf, " | %15d | ", adccntlp->nViewRows); |
| for ( i = 0; i < adccntlp->nm; i++ ){ |
| fprintf(adccntlp->logf, " %20lld", adccntlp->checksums[i]); |
| } |
| fprintf(adccntlp->logf, " | %5d", adccntlp->numberOfChunks); |
| } |
| adccntlp->numberOfMadeViews = nViews; |
| if(ViewSizesVerification(adccntlp)) return ADC_VERIFICATION_FAILED; |
| return ADC_OK; |
| } |
