c - 从文件中读取异步 I/O

Question

我最近对多个项目有了想法，这些项目都涉及从文件中读取 IP 地址。由于它们都应该能够处理大量主机，因此我尝试实现多线程或创建套接字池并从中选择()-ing以实现某种形式的并发以获得更好的性能.在很多情况下，从文件中读取似乎是提高性能的瓶颈。按照我的理解，使用 fgets 或类似文件读取文件是一种同步的阻塞操作。因此，即使我成功地实现了一个异步连接到多个主机的客户端，该操作仍然是同步的，因为我一次只能从一个文件中读取一个地址。

    /* partially pseudo code */

/* getaddrinfo() stuff here */

while(fgets(ip, sizeof(ip), file) {
FD_ZERO(&readfds);
/* create n sockets here in a for loop */
for (i = 0; i < socket_num; i++) {
    if (newfd > fd[i]) newfd = fd[i];
    FD_SET(fd[i], &readfds);
}

/* here's where I think I should connect n sockets to n addresses from file
 * but I'm only getting one IP at a time from file, so I'm not sure how to connect to 
 * n addresses at once with fgets
 */

for (j = 0; j < socket_num; j++) {
        if ((connect(socket, ai->ai_addr, ai->ai_addrlen)) == -1)
        // error
        else { 
            freeaddrinfo(ai);       
        FD_SET(socket, &master);
            fdmax = socket;
            if (select(socket+1, &master, NULL, NULL, &tv) == -1);
        // error        
            if ((recvd = read(socket, banner, RECVD)) <= 0)
        // error
            if (FD_ISSET(socket, &master))
        // print success
        }
    /* clear sets and close sockets and stuff */
}

我已经在评论中指出了我的问题，但只是为了澄清:我不确定如何在从文件读取的多个目标服务器上执行异步 I/O 操作，因为从文件读取条目似乎是严格同步的.我在多线程方面遇到过类似的问题，但成功率略高。

    void *function_passed_to_pthread_create(void *opts) 
    { 
        while(fgets(ip_addr, sizeof(ip_addr), opts->file) {
            /* speak to ip_addr and get response */
    }
}

main()
{
    /* necessary stuff */
    for (i = 0; i < thread_num; i++) {
        pthread_create(&tasks, NULL, above_function, opts)
    }
    for (j = 0; j < thread_num; j++)
        /* join threads */
    return 0;
}

这似乎可行，但由于多个线程都在处理同一个文件，因此结果并不总是准确的。我想这是因为多个线程可能同时处理文件中的相同地址。

我考虑过将文件中的所有条目加载到数组/内存中，但如果文件特别大，我想这可能会导致内存问题。最重要的是，我不确定这样做是否有意义。

作为最后的说明；如果我正在读取的文件恰好是一个具有大量 IP 的特别大的文件，那么我认为这两种解决方案都无法很好地扩展。不过，使用 C 语言一切皆有可能，所以我想有一些方法可以实现我所希望的。

总结这篇文章；我想找到一种在从文件读取条目时使用异步 I/O 或多线程来提高客户端应用程序性能的方法。

Answer 1

一些人在他们的评论中暗示了一个很好的解决方案，但可能值得更详细地说明。 完整 解决方案包含很多细节并且代码非常复杂，因此我将使用伪代码来解释我的建议。

您在这里遇到的实际上是经典生产者/消费者问题的变体:您有一个生成数据的事物，以及许多试图使用该数据的事物。在您的情况下，它必须是生成该数据的“单一事物”，因为源文件每行的长度是未知的:您不能只向前跳“n”个字节并以某种方式在下一个IP。一次只能有一个参与者将读取指针移向 \n 的下一个未知位置，因此根据定义，您只有一个生产者。

一般有3种攻击方式:

• 解决方案 A 涉及让每个线程从共享文件缓冲区中拉出更多，并在每次最后一次读取时启动异步(非阻塞)读取阅读完成。要使这个解决方案正确，会遇到一大堆令人头疼的问题，因为它对文件系统和正在执行的工作之间的时间差异非常敏感:如果文件读取速度很慢，所有工作人员都会停止等待文件。如果工作人员速度慢，文件读取器将停止或填满内存以等待他们使用数据。此解决方案可能是绝对 最快的解决方案，但它也是非常难以正确处理大量警告的同步代码。除非您是线程专家(或者非常巧妙地滥用 epoll_wait())，否则您可能不想走这条路。

• 解决方案 B 有一个“主”线程，负责读取文件，并用它读取的数据填充某种线程安全队列，每个队列条目有一个 IP 地址(一个字符串)。每个工作线程都尽可能快地使用队列条目，查询远程服务器，然后请求另一个队列条目。这需要一点小心才能正确，但通常比解决方案 A 安全得多，尤其是当您使用其他人的队列实现时。

• 解决方案 C 非常玄妙，但您不应该立即忽略它，具体取决于您在做什么。该解决方案仅涉及使用类似 Un*x sed 命令(请参阅 Get a range of lines from a file given the start and end line numbers )的东西将您的源文件预先分割成一堆“ block 状”源文件——比如说，20 个。然后，您只需使用 & 并行运行一个非常简单的单线程程序的 20 个副本，每个副本位于文件的不同“切片”上。结合一个小的 shell 脚本来实现自动化，这可能是满足很多需求的“足够好”的解决方案。

---

让我们仔细看看解决方案 B — 具有线程安全队列的主线程。我要作弊并假设您可以构造一个工作队列实现(如果不能，有 StackOverflow 文章介绍如何使用 pthreads 实现线程安全队列:pthread synchronized blocking queue)。

在伪代码中，这个解决方案是这样的:

main()
{
    /* Create a queue. */
    queue = create_queue();

    /* Kick off the master thread to read the file, and give it the queue. */
    master_thread = pthread_create(master, queue);

    /* Kick off a bunch of workers with access to the queue. */
    for (i = 0; i < 20; i++) {
        worker_thread[i] = pthread_create(worker, queue);
    }

    /* Wait for everybody to finish. */
    pthread_join(master_thread);
    for (i = 0; i < 20; i++) {
        pthread_join(worker_thread[i]);
    }
}

void master(queue q)
{
    FILE *fp = fopen("ips.txt", "r");
    char buffer[BIGGER_THAN_ANY_IP];

    /* Inhale the file as fast as we can, and push each line we
       read onto the queue. */
    while (fgets(fp, buffer) != NULL) {
        char *next_ip = strdup(buffer);
        enqueue(q, next_ip);
    }

    /* Add some final messages in the queue to let the workers
       know that we're out of data.  There are *much* better ways
       of notifying them that we're "done", but in this case,
       pushing a bunch of NULLs equal to the number of threads is
       simple and probably good enough. */
    for (i = 0; i < 20; i++) {
        enqueue(q, NULL);
    }
}

void worker(queue q)
{
    char *ip;

    /* Inhale messages off the queue as fast as we can until
       we get a "NULL", which means that it's time to stop.
       The call to dequeue() *must* block if there's nothing
       in the queue; the call should only return NULL if the
       queue actually had NULL pushed into it. */
    while ((ip = dequeue(q)) != NULL) {

        /* Insert code to actually do the work here. */
        connect_and_send_and_receive_to(ip);
    }
}

在实际实现中有很多注意事项和细节(例如:我们如何实现队列、环形缓冲区或链表？如果文本不全是 IP 怎么办？如果字符缓冲区不大怎么办够了吗？多少线程就够了？我们如何处理文件或网络错误？malloc 性能会成为瓶颈吗？如果队列太大怎么办？我们可以更好地重叠网络 I/O 吗？)。

但是，抛开注意事项和细节不谈，我上面提供的伪代码是一个足够好的起点，您可以将其扩展为一个可行的解决方案。