C 中带有套接字的 POSIX 线程

Question

#include "cs241.c"

#define THREAD_COUNT 10

int list_s;
int connections[THREAD_COUNT];
char space[THREAD_COUNT];

int done = 0;

pthread_mutex_t muxlock = PTHREAD_MUTEX_INITIALIZER;

int *numbers;
int numbers_count;

void *listener(void *arg) {
    int n = *(int *) arg;

    FILE *f = fdopen(connections[n], "r");
    if (f == NULL)
        printf("Could not open file\n");
    char *line = NULL;
    size_t *len = malloc(sizeof(int));
    while(getline(&line, len, f) != -1) {
        printf("%s", line);
        if (strcmp("END", line) == 0) {
                            pthread_mutex_lock(&muxlock);
            done = 1;
            pthread_mutex_unlock(&muxlock);                     
        }
    }

    space[n] = 't';
    fclose(f);
    free(len);
    close(connections[n]);
    return NULL;
}

void initialize() {
    int n;
    for (n = 0; n < THREAD_COUNT; n++) {
        space[n] = 't';
    }
}

int check() {
    int index;
    for (index = 0; index < THREAD_COUNT; index++) {
        if (space[index] == 't') {
            space[index] = 'f';
            return index;
        }
    }
    return 0;
}   

int main(int argc, char *argv[]) {
    int port = 0;
    int binder;
    int lis;
    int i = 0;
    int *j = malloc(sizeof(int));
    initialize();
    pthread_t threads[THREAD_COUNT];

    if ((argc != 2) || (sscanf(argv[1], "%d", &port) != 1)) {
        fprintf(stderr, "Usage: %s [PORT]\n", argv[0]);
        exit(1);
    }

    if (port < 1024) {
        fprintf(stderr, "Port must be greater than 1024\n");
        exit(1);
    }

    // set the initial conditions for the numbers array.
    numbers = malloc(sizeof(int));
    numbers_count = 0;

    struct sockaddr_in servaddr; // socket address structure
    // set all bytes in socket address structure to zero, and fill in the relevant data members
    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
    servaddr.sin_port = htons(port);

    list_s = socket(AF_INET, SOCK_STREAM, 0);
    if (list_s < 0) {
        printf("Could not create socket\n");
        exit(EXIT_FAILURE);
    }
    binder = bind(list_s, (struct sockaddr *) &servaddr, sizeof(servaddr));
    if (binder < 0) {
        printf("Could not bind socket\n");
        exit(EXIT_FAILURE);
    }
    lis = listen(list_s, SOMAXCONN);
    if (lis < 0) {
        printf("Could not listen on socket\n");
        exit(EXIT_FAILURE);
    }
    SET_NON_BLOCKING(list_s);
    while (done != 1) {
        connections[i] = accept(list_s, NULL, NULL);
        if (connections[i] < 0) {
            if ((errno == EAGAIN) || (errno == EWOULDBLOCK))
                continue;
            printf("Could not accept connection\n");
            exit(EXIT_FAILURE);
        }
        i = check();
        *j = i;
                    SET_BLOCKING(list_s);
        pthread_create(&threads[i], NULL, listener, j);
    }

    // Verify the array.
    //verify(numbers, numbers_count);

    free(j);
    close(list_s);

    exit(EXIT_SUCCESS);

}

所以我的 main() 中有一个 while 循环，它应该在 'done' = 1 时退出。这是由 listener() 在收到 'END' 时设置的。第一个问题是，当我在第一次迭代中发送“END”时，while 循环不会退出，只有在发送了另一个“END”之后。

我在另一个文件中有两个宏 SET_NON_BLOCKING 和 SET_BLOCKING 用于解除阻塞和阻塞套接字，因此如果没有连接，它会等待连接。下一个问题是当我不使用这些宏时，listener() 中的 getline() 无法读取从流中输出的所有内容。当我使用它们时，它根本无法打开流。

我认为一些问题在于将“j”传递给线程，因为当第二个线程启动时，它会在第一个线程可以读取之前覆盖“j”。但是，我已经在它上面呆了几天，却无处可去。任何帮助将不胜感激。谢谢！

另外我想问一下我的socket阻塞和线程锁定是否在正确的地方？

Answer 1

您可能必须发送 END 两次，因为您的主线程在生成侦听器线程后立即阻塞在 accept() 上。由于它在accept() 上被阻止，它看不到done == 1。

您可以通过保持非阻塞模式来解决此问题。如果你这样做，你可能想睡觉以避免在一个紧密的循环中旋转。另一种选择是发送一个信号来唤醒接受，使其设置 EINTR。

在将连接索引传递给侦听器线程方面，您可能是正确的，您在线程之间存在覆盖值的竞争。由于int 需要的字节数与void * 相同或更少，因此您实际上可以直接传递int。例如：

pthread_create(&threads[i], NULL, listener, (void *)i);

然后在监听器中：

int n = (int)arg;

这是一种黑客行为。更完整的解决方案是使用 malloc 为每个线程分配一个单独的参数结构。侦听器线程将负责释放参数。

struct params *p = malloc(sizeof(struct params));
p.index = i;

pthread_create(&threads[i], NULL, listener, p);

然后在监听器中：

struct params *p = args;
if (p == NULL) {
  // handle error
  return NULL;
}
int n = p->index;
free(p);

Answer 2

您管理connections[] 和space[] 数组的方式似乎有很多问题：

• 您接受到connections[i] 的连接，但您传递给listener 线程的i 是check() 返回的那个——这可能是一个完全不同的i。
• 每个线程都被传递给线程参数的相同地址 - j 只分配一次，并且每次创建线程时都使用相同的分配。如果两个或更多线程几乎同时启动，您将丢失应该传递给其中一些线程的索引。
• 当它完成时，listener 将连接标记为可用（我假设这就是space[n] = 't'; 的意图），然后使用connections[n] 关闭连接。这些都不是通过任何同步来执行的。

其他一些cmets：

• 你到底为什么要动态分配len？
• 在main() 的while 循环中对done 的检查应该受到互斥锁的保护
• 我不确定传输数据的确切协议是什么，但由于您使用的是getline()，我假设它是一组以换行符结尾的行。在这种情况下，请记住 getline() 文档中的这一点信息：“缓冲区以空值结尾并包括换行符，如果找到的话”。如果 "END" 行包含换行符，strcmp("END",line) 将不会返回 0。