Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

HPCC-32766 Decompress partition points to roxiemem in global sort #19178

Merged
merged 1 commit into from
Oct 3, 2024
+82 −22
Merged

HPCC-32766 Decompress partition points to roxiemem in global sort #19178

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 63 additions & 21 deletions thorlcr/msort/tsortm.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,6 +248,7 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
Linked<IThorRowInterfaces> rowif;
Linked<IThorRowInterfaces> auxrowif;
Linked<IThorRowInterfaces> keyIf;
Owned<roxiemem::IVariableRowHeap> partitionHeap;

int AddSlave(ICommunicator *comm,rank_t rank,SocketEndpoint &endpoint,mptag_t mpTagRPC)
{
Expand Down Expand Up @@ -522,9 +523,10 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
CriticalSection &asect;
unsigned averagesamples;
rowcount_t averagerecspernode;
roxiemem::IVariableRowHeap *heap;
public:
casyncfor1(CSortMaster &_owner, NodeArray &_slaves, CThorExpandingRowArray &_sample, unsigned _averagesamples, rowcount_t _averagerecspernode, CriticalSection &_asect)
: owner(_owner), slaves(_slaves), sample(_sample), asect(_asect)
casyncfor1(CSortMaster &_owner, NodeArray &_slaves, CThorExpandingRowArray &_sample, unsigned _averagesamples, rowcount_t _averagerecspernode, CriticalSection &_asect, roxiemem::IVariableRowHeap *_heap)
: owner(_owner), slaves(_slaves), sample(_sample), asect(_asect), heap(_heap)
{
averagesamples = _averagesamples;
averagerecspernode = _averagerecspernode;
Expand All @@ -537,13 +539,28 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
if (slavesamples)
{
size32_t samplebufsize;
void *samplebuf=NULL;
slave.GetMultiNthRow(slavesamples, samplebufsize, samplebuf);
MemoryBuffer mb;
fastLZDecompressToBuffer(mb, samplebuf);
free(samplebuf);
OwnedMalloc<void> samplebuf;
void *p;
slave.GetMultiNthRow(slavesamples, samplebufsize, p);
samplebuf.setown(p);
size32_t expandedSz;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If a similar pattern appears anywhere else, then this code could be extracted into a class, and commoned up with the code below. Not worth it otherwise.

const void *expandedPtr = nullptr;
MemoryBuffer expandedSampleMb;
OwnedConstThorRow expandedSample;
if (heap)
{
expandedSample.setown(fastLZDecompressToRoxieMem(*heap, samplebuf, expandedSz));
expandedPtr = expandedSample;
}
else
{
fastLZDecompressToBuffer(expandedSampleMb, samplebuf);
expandedSz = expandedSampleMb.length();
expandedPtr = expandedSampleMb.toByteArray();
}
samplebuf.clear();
CriticalBlock block(asect);
CThorStreamDeserializerSource d(mb.length(), mb.toByteArray());
CThorStreamDeserializerSource d(expandedSz, expandedPtr);
while (!d.eos())
{
RtlDynamicRowBuilder rowBuilder(owner.keyIf->queryRowAllocator());
Expand All @@ -552,7 +569,7 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
}
}
}
} afor1(*this, slaves,sample,averagesamples,averagerecspernode,asect);
} afor1(*this, slaves,sample,averagesamples,averagerecspernode,asect,partitionHeap);
afor1.For(numnodes, 20, true);
#ifdef TRACE_PARTITION2
{
Expand Down Expand Up @@ -655,7 +672,6 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
return splitMap.getClear();
}


rowcount_t *CalcPartition(bool logging)
{
CriticalBlock block(ECFcrit);
Expand Down Expand Up @@ -727,31 +743,51 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
Semaphore *nextsem;
unsigned numsplits;
IThorRowInterfaces *keyIf;
roxiemem::IVariableRowHeap *heap;
public:
casyncfor2(NodeArray &_slaves, CThorExpandingRowArray &_totmid, unsigned _numsplits, Semaphore *_nextsem, IThorRowInterfaces *_keyIf)
: slaves(_slaves), totmid(_totmid), keyIf(_keyIf)
casyncfor2(NodeArray &_slaves, CThorExpandingRowArray &_totmid, unsigned _numsplits, Semaphore *_nextsem, IThorRowInterfaces *_keyIf, roxiemem::IVariableRowHeap *_heap)
: slaves(_slaves), totmid(_totmid), keyIf(_keyIf), heap(_heap)
{
nextsem = _nextsem;
numsplits = _numsplits;
}
void Do(unsigned i)
{
CSortNode &slave = slaves.item(i);
void *p = NULL;
size32_t retlen=0;
if (slave.numrecs!=0)
slave.GetMultiMidPointStop(retlen,p);
OwnedMalloc<void> midPoints;
if (slave.numrecs!=0)
{
// if the rows this SORT is based are large, these midpoints (nodes-1) could be large (but compressed)
// could consider serializing and deserializnig them in chunks..
void *p;
slave.GetMultiMidPointStop(retlen, p);
midPoints.setown(p);
}
if (i)
nextsem[i-1].wait();
try
{
unsigned base = totmid.ordinality();
if (p)
if (midPoints)
{
MemoryBuffer mb;
fastLZDecompressToBuffer(mb, p);
free(p);
CThorStreamDeserializerSource d(mb.length(), mb.toByteArray());
size32_t expandedSz;
const void *expandedPtr = nullptr;
MemoryBuffer expandedMidPointsMb;
OwnedConstThorRow expandedMidPoints;
if (heap)
{
expandedMidPoints.setown(fastLZDecompressToRoxieMem(*heap, midPoints, expandedSz));
expandedPtr = expandedMidPoints;
}
else
{
fastLZDecompressToBuffer(expandedMidPointsMb, midPoints);
expandedSz = expandedMidPointsMb.length();
expandedPtr = expandedMidPointsMb.toByteArray();
}
midPoints.clear();
CThorStreamDeserializerSource d(expandedSz, expandedPtr);
while (!d.eos())
{
RtlDynamicRowBuilder rowBuilder(keyIf->queryRowAllocator());
Expand All @@ -777,7 +813,7 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
}
nextsem[i].signal();
}
} afor2(slaves, totmid, numsplits, nextsem, keyIf);
} afor2(slaves, totmid, numsplits, nextsem, keyIf, partitionHeap);
afor2.For(numnodes, 20);

delete [] nextsem;
Expand Down Expand Up @@ -1096,6 +1132,12 @@ class CSortMaster : public IThorSorterMaster, public CSimpleInterface
}
ActPrintLog(activity, "Sort: canoptimizenullcolumns=%s, usepartitionrow=%s, betweensort=%s skewWarning=%f skewError=%f minisortthreshold=%" I64F "d",canoptimizenullcolumns?"true":"false",usepartitionrow?"true":"false",betweensort?"true":"false",skewWarning,skewError,(__int64)minisortthreshold);
assertex(partitioninfo);

constexpr byte AT_SortPartition=1;
activity_id actId = createCompoundActSeqId(activity->queryId(), AT_SortPartition);
if (activity->getOptBool(THOROPT_ROXIEMEM_GLOBALSORT_PARTITION, true))
partitionHeap.setown(activity->queryRowManager()->createVariableRowHeap(actId | ACTIVITY_FLAG_ISREGISTERED, roxiemem::RoxieHeapFlags::RHFvariable));

maxdeviance = _maxdeviance;
unsigned i;
bool overflowed = false;
Expand Down
18 changes: 17 additions & 1 deletion thorlcr/thorutil/thmem.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2656,4 +2656,20 @@ IOutputMetaData *createOutputMetaDataWithChildRow(IEngineRowAllocator *childAllo
return new COutputMetaWithChildRow(childAllocator, extraSz);
}


void *fastLZDecompressToRoxieMem(roxiemem::IVariableRowHeap &heap, const void * src, size32_t &expsz)
{
size32_t *sz = (size32_t *)src;
expsz = *(sz++);
size32_t cmpsz = *(sz++);
memsize_t capacity;
void *o = heap.allocate(expsz, capacity);
if (cmpsz!=expsz)
{
size32_t written = fastlz_decompress(sz,cmpsz,o,expsz);
if (written!=expsz)
throw MakeStringException(0, "fastLZDecompressToBuffer - corrupt data(1) %d %d",written,expsz);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

picky: Should really be %u for unsigned values.

}
else
memcpy_iflen(o,sz,expsz);
return o;
}
1 change: 1 addition & 0 deletions thorlcr/thorutil/thmem.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -573,6 +573,7 @@ extern graph_decl IThorRowLoader *createThorRowLoader(CActivityBase &activity, I
extern graph_decl IThorRowCollector *createThorRowCollector(CActivityBase &activity, IThorRowInterfaces *rowIf, ICompare *iCompare=NULL, StableSortFlag stableSort=stableSort_none, RowCollectorSpillFlags diskMemMix=rc_mixed, unsigned spillPriority=SPILL_PRIORITY_DEFAULT, EmptyRowSemantics emptyRowSemantics=ers_forbidden);


extern graph_decl void *fastLZDecompressToRoxieMem(roxiemem::IVariableRowHeap &heap, const void * src, size32_t &expsz);


class CSDSServerStatus;
Expand Down
1 change: 1 addition & 0 deletions thorlcr/thorutil/thormisc.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@
#define THOROPT_LOOKAHEAD_WRITEAHEADK "readAheadWriteAheadK" // Splitter spilling write ahead size (K) (default = 2MB)
#define THOROPT_LOOKAHEAD_COMPRESSIONTOTALK "readAheadCompressionTotalK" // Splitter total compression buffer size (shared between writer and readers) (K) (default = 3MB)
#define THOROPT_LOOKAHEAD_TEMPFILE_GRANULARITY "readAheadTempFileGranularity" // Splitter temp file granularity (default = 1GB)
#define THOROPT_ROXIEMEM_GLOBALSORT_PARTITION "useRoxieMemGlobalSortPartition" // Use roxiemem for global sort partitioning (default = true)

#define INITIAL_SELFJOIN_MATCH_WARNING_LEVEL 20000 // max of row matches before selfjoin emits warning

Expand Down
Loading