Intel TBB并行管道处理文件报错：basic_string构造无效问题排查

首先，你的错误根源主要有两个：

1. 字符串构造的无效指针问题 #

第一个过滤器在文件结束时返回NULL，而std::string的构造函数不接受空指针（NULL本质是(const char*)0），这会直接触发basic_string::_M_construct null not valid异常，导致程序崩溃。

2. TBB管道的流结束处理不当 #

当调用fc.stop()后，当前过滤器的返回值仍会被传递到下一个过滤器，而你返回的NULL会被强制转换为无效的字符串，引发后续处理错误。

另外，TBB的警告是因为你的程序抛出了异常，但链接的TBB库未编译异常支持，不过这个警告的根源还是代码中的异常，修复代码后警告会自然消失。

以下是修正后的完整代码，我会逐点说明修改：

#include <tbb/parallel_pipeline.h>
#include <tbb/flow_graph.h>
#include <fstream>
#include <vector>
#include <string>
#include <optional>
#include <stdexcept>
#include <iostream>

using namespace tbb::flow;

struct A { 
    long long time; 
    double price; 
    A(long long t, double p) : time(t), price(p) {}; 
};

long long dateToMs2(const char* date_str) {
    // 保留你的原实现逻辑
    return 0;
}

std::vector<A> parallelFile(std::string fileName) {
    std::ifstream fe(fileName);
    if (!fe.is_open()) { // 新增：检查文件是否成功打开
        throw std::runtime_error("Failed to open file: " + fileName);
    }

    std::vector<A> ord;

    parallel_pipeline(/*max_number_of_tokens=*/16,
        // 第一个过滤器：读取文件行，返回optional<string>表示是否有有效数据
        make_filter<void, std::optional<std::string>>(
            filter::serial, 
            [&fe](flow_control& fc) -> std::optional<std::string> {
                std::string line;
                if (std::getline(fe, line)) { // 读取成功则返回有效字符串
                    return line;
                } else { // 读取失败（EOF或错误），停止管道并返回空
                    fc.stop();
                    return std::nullopt;
                }
            }
        )
        & 
        // 第二个过滤器：处理每行字符串，返回optional<A>
        make_filter<std::optional<std::string>, std::optional<A>>(
            filter::parallel, 
            [](std::optional<std::string> opt_line) -> std::optional<A> {
                if (!opt_line) { // 无有效数据，直接返回空
                    return std::nullopt;
                }
                const std::string& p = *opt_line;

                // 新增：格式校验，避免substr出错
                if (p.size() < 28) {
                    throw std::runtime_error("Line too short: " + p);
                }
                auto position = p.find('"', 28);
                if (position == std::string::npos) {
                    throw std::runtime_error("Missing quote in line: " + p);
                }
                if (position + 2 >= p.size()) {
                    throw std::runtime_error("Price missing in line: " + p);
                }

                double price = std::stod(p.substr(position + 2));
                long long time = dateToMs2(p.substr(0, 26).c_str());
                return A(time, price);
            }
        )
        & 
        // 第三个过滤器：将结果存入vector
        make_filter<std::optional<A>, void>(
            filter::serial, 
            [&ord](std::optional<A> opt_a) {
                if (opt_a) { // 仅处理有效数据
                    ord.push_back(*opt_a);
                }
            }
        )
    );

    return ord;
}

int main() {
    try {
        std::vector<A> david = parallelFile("input.txt");
        // 可在此添加结果验证或输出逻辑
    } catch (const std::exception& e) {
        std::cerr << "Error: " << e.what() << std::endl;
        return 1;
    }
    return 0;
}

1. 用std::optional处理流结束

   • 第一个过滤器返回optional<string>，读取成功时返回有效字符串，失败时返回nullopt并停止管道。TBB会自动识别nullopt并终止后续处理，避免无效数据流入下一个过滤器。
   • 后续过滤器也使用optional类型，确保只处理有效数据。

2. 移除无效的NULL返回

   • 彻底避免了返回NULL给std::string的错误，从根源解决了字符串构造异常。

3. 增加错误校验

   • 检查文件是否成功打开，避免后续读取失败。
   • 在处理每行时，校验字符串长度和格式（比如引号位置），避免substr或stod抛出未捕获的异常。

4. 局部变量替代共享变量

   • 第一个过滤器中使用局部变量line存储读取的行，避免共享变量可能带来的潜在问题（虽然原代码中是串行过滤器，但局部变量更安全清晰）。

5. 异常捕获

   • 在main函数中添加异常捕获，方便调试和错误提示。

内容的提问来源于stack exchange，提问作者david.t_92