改进了merge，开始第二阶段过程的工作，需要clean一下代码

Co-authored-by: Copilot <copilot@github.com>

src/sort/phase_1.cpp

@@ -68,5 +68,8 @@ void phase1Pipeline() {
 
     PROF_G_END(mid_all);
 
+
+    
+
 #endif
 }

src/sort/phase_1.h

@@ -99,9 +99,19 @@ struct ThreadUncompressWrap {
     }
 };
 
+struct BlockBams {
+    DataBuffer blockBuf; // 解压的block放在这里
+    BamPtrArr bamPtrArr; // 解析后的bam数据放在这里
+
+    void Clear() {
+        blockBuf.Clear();
+        bamPtrArr.Clear();
+    }
+};
+
 // 用于合并压缩的数据结构
 struct MergeCompressData {
-    vector<DataBuffer> blockDataArr; // 待压缩的数据
+    vector<BlockBams> blockDataArr;     // 待压缩的数据
     vector<DataBuffer> compressDataArr; // 压缩后的数据
 
     void Resize(int blockNum) {

src/sort/phase_1_compress.cpp

@@ -26,9 +26,15 @@
 static void mtCompressBlock(void* data, long idx, int tid) {
     Phase1PipelineArg& p = *(Phase1PipelineArg*)data;
     MergeCompressData& mergeCompressData = p.mergeCompressData[p.compressOrder % p.COMPRESS_BUF_NUM];
-    auto& blockData = mergeCompressData.blockDataArr[idx];
+    auto& bams = mergeCompressData.blockDataArr[idx].bamPtrArr;
+    auto& blockData = mergeCompressData.blockDataArr[idx].blockBuf;
     auto& compressData = mergeCompressData.compressDataArr[idx];
 
+    for (int i = 0; i < bams.Size(); ++i) {
+        const OneBam* bp = bams.arr[i];
+        blockData.MemCopy(p.uncompressData.dataBuf + bp->offset, bp->wholeBamLen);
+    }
+
     // bgzfCompress(void* _dst, size_t* dlen, const void* src, size_t slen, int level)
     compressData.ReAllocMem(SINGLE_BLOCK_SIZE);  // 压缩后的block数据不会超过单个block的大小
     compressData.curLen = SINGLE_BLOCK_SIZE;

src/sort/phase_1_sort.cpp

@@ -153,7 +153,8 @@ bool doMergeSort(Phase1PipelineArg& p, Phase1BamHeap<BamGreaterThan>& heap) {
             mergeCompressData.blockDataArr[mergedBlockNum].Clear(); // 清理，为添加bam数据做准备
             bamBytes = 0;
         }
-        mergeCompressData.blockDataArr[mergedBlockNum].MemCopy(p.uncompressData.dataBuf + bam->offset, bam->wholeBamLen);
+        // mergeCompressData.blockDataArr[mergedBlockNum].MemCopy(p.uncompressData.dataBuf + bam->offset, bam->wholeBamLen);
+        mergeCompressData.blockDataArr[mergedBlockNum].bamPtrArr.Add(bam);
         bamBytes += bam->wholeBamLen;  // for test
         heap.Pop();
     }
@@ -161,7 +162,7 @@ bool doMergeSort(Phase1PipelineArg& p, Phase1BamHeap<BamGreaterThan>& heap) {
         // 都处理完
         finish = true;
     }
-    spdlog::info("mergedBlockNum: {}, bamBytes: {}", mergedBlockNum, bamBytes);
+    // spdlog::info("mergedBlockNum: {}, bamBytes: {}", mergedBlockNum, bamBytes);
 
     return finish;
 }

src/sort/phase_1_uncompress.cpp

@@ -381,8 +381,10 @@ static void doMemCopy(Phase1PipelineArg& p) {
 #endif
 
         // 开启排序并写入中间文件
+        PROF_G_BEG(after_full);
         phase1Sort(&p);
         phase1MergeCompress(&p);
+        PROF_G_END(after_full);
 
         p.uncompressData.NextRound();
         p.allBams.Clear();
@@ -390,7 +392,7 @@ static void doMemCopy(Phase1PipelineArg& p) {
 
     PROF_G_BEG(mem_copy);
 
-    p.allBams.Add(uncompressWrap.GetTotalBamNum());
+    p.allBams.AddSize(uncompressWrap.GetTotalBamNum());
     p.bamNum += uncompressWrap.GetTotalBamNum();
     p.blockNum += uncompressWrap.GetTotalBlockNum();
 

src/sort/phase_1_write.cpp

@@ -24,8 +24,10 @@
 #include "util/profiling.h"
 
 static void doWrite(Phase1PipelineArg& p) {
+    PROF_G_BEG(write_mid);
     DataBuffer& compressBuf = p.compressBuf[p.compressOrder % p.COMPRESS_BUF_NUM];
     fwrite(compressBuf.data, 1, compressBuf.curLen, p.midFilePtr);
+    PROF_G_END(write_mid);
 }
 
 void* phase1Write(void* data) {

src/sort/phase_2.h

@@ -13,3 +13,11 @@
 */
 
 #pragma once
+
+struct Phase2File {
+    FILE* fp;
+    // 双buffer
+    // 当前读入的buffer指针df 
+    // ReadBuffer
+
+};

src/sort/sort.h

@@ -68,6 +68,12 @@ struct UncompressBlockBuffer {
         lastEndPos = 0;
     }
 
+    uint8_t *HandOverBuf() {
+        uint8_t *handOverBuf = dataBuf;
+        dataBuf = nullptr; // 交出buf的所有权，外部需要负责释放内存
+        return handOverBuf;
+    }
+
     void NextRound() {
         usedBufSize -= lastEndPos;
         memcpy(dataBuf, dataBuf + lastEndPos, usedBufSize);
@@ -95,7 +101,13 @@ struct FastVector {
 #endif
         }
     }
-    void Add(size_t num) {
+
+    void Add(const T& item) {
+        T &newItem = Add();
+        newItem = item;
+    }
+
+    void AddSize(size_t num) {
         if (curIdx + num > arr.size()) {
             arr.resize(curIdx + num);
         }
@@ -383,6 +395,7 @@ struct MergeSortData {
 };
 
 typedef FastVector<OneBam> BamArr;
+typedef FastVector<const OneBam*> BamPtrArr;
 typedef FastVector<OneBlock> BlockArr;
 
 
@@ -413,9 +426,11 @@ struct FirstPipeArg {
     ReadBuffer readData[READ_BUF_NUM]; // 用来读如数据，双缓冲
     UncompressData uncompressData[UNCOMPRESS_BUF_NUM]; // 每个线程内保留一些自己的数据，比如解压缩的数据
     UncompressBlockBuffer unCompblockDataBuf; // 所有线程共用一个，串行往这里添加解压后的block数据
+    
     vector<const OneBam*> sortedBamArr;
     vector<ArrayInterval> taskArr;
     vector<DataBuffer> threadBuf;
+    
     MergeSortData mergeSortData;
 
     FirstPipeArg()

src/util/profiling.cpp

@@ -58,6 +58,7 @@ int displayProfiling(int nthread) {
     PRINT_GP(mem_copy);
     PRINT_GP(parse_block);
     PRINT_GP(uncompress);
+    PRINT_GP(after_full);
     PRINT_GP(sort);
     PRINT_GP(merge);
     PRINT_GP(compress);

src/util/profiling.h

@@ -71,6 +71,7 @@ enum {
     GP_mem_copy,
     GP_parse_block,
     GP_uncompress,
+    GP_after_full,
     GP_sort,
     GP_merge,
     GP_compress,