c - 套接字发送有时只在无限循环中工作 (C)

Question

我正在为 Zynq FPGA 的 DAQ 系统建立原型(prototype)。我通过以太网从服务器接收数据，使用 DMA 将其写入 FIFO，反之亦然，使用两个不同的 pthreads。然而，发送操作挂起，经过一些迭代后，它正确执行然后再次挂起，同时 recv 操作成功并填充 FIFO。

有时线程只有在某些 printf 存在或不存在时才会按预期工作，所以我暂时打印到 stderr。

对于可耻的代码感到抱歉，但我现在尝试替换几乎每一行以发现问题。

void *sender(void *params) {

    arguments *args = params;

    if (args->head == NULL) {
        fprintf(stderr, "[-] Head pointer not valid\n");
        exit(0);
    }
    if (args->tail == NULL) {
        fprintf(stderr, "[-] Tail pointer not valid\n");
        exit(0);
    }
    if (args->virtual_address == NULL) {
        fprintf(stderr, "[-] AXI DMA register pointer not valid\n");
        exit(0);
    }
    if (args->virtual_buffer == NULL) {
        fprintf(stderr, "[-] Send buffer pointer not valid\n");
        exit(0);
    }

    unsigned long int units_sent = 0;

    for (;;) {
        pthread_mutex_lock(args->lock);
        fprintf(stderr, "[*] Send   Head: %d    Tail: %d\n", *(args->head), *(args->tail));
        if (*(args->head) != *(args->tail)) {
            pthread_mutex_unlock(args->lock);
            int remaining = args->buffsize;
            int sent = 0;
            pthread_mutex_lock(args->lock);
            int src = FIFO_ADDR + (*(args->tail)) * args->buffsize;
            pthread_mutex_unlock(args->lock);
            if (args->verbose > 2) {
                fprintf(stderr, "[*] Sender: DMA is transferring data from 0x%x to 0x%x\n", src, SEND_ADDR);
            }
            unsigned int length = args->buffsize;
            unsigned int verb = args->verbose > 2 ? 1 : 0;
            pthread_mutex_lock(args->lock);
            while (remaining > 0) {
                length = remaining < MAX_TRANSF ? remaining : remaining % MAX_TRANSF;
                DMATransfer(args->virtual_address, src + sent, SEND_ADDR, length, verb);
                remaining -= args->buffsize;
                sent += remaining;
            }
            elements--;
            units_sent++;
            if (args->verbose > 2) {
                fprintf(stderr, "[*] %f elements in FIFO: %lu\n", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, elements);
                fprintf(stderr, "[*] %f DMA tranfer to buffer: %d\n", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, units_sent);
            }
            pthread_mutex_unlock(args->lock);
            remaining = args->buffsize;
            sent = 0;
            int result = 0;
            pthread_mutex_lock(args->lock);
            while (remaining > 0) {
                    result = send(args->sockfd, args->virtual_buffer + sent, remaining, 0);
                    if (result > 0) {
                        remaining -= result;
                        sent += remaining;
                    } else if (result < 0) {
                        fprintf(stderr, "[-] Error retrieving configuration from the server\n");
                    exit(0);
                    }
            }
            //memset(args->virtual_buffer, 0, args->buffsize);
            if (args->verbose > 2) {
                fprintf(stderr, "[*] %f Unit sent: %d\n", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, units_sent);
            }
            if (args->verbose > 0) {
                fprintf(stderr, "[*] Packet retrieved");
            }
            if (args->verbose > 1) {
                fprintf(stderr, " content: ");
                memdump(args->virtual_buffer, args->buffsize);
            }
            if (args->verbose > 0) {
                fprintf(stderr, "\n");
            }
            *(args->tail) = (*(args->tail) + 1) % args->fifosize;
            pthread_mutex_unlock(args->lock);
            if (args->verbose > 2) {
                fprintf(stderr, "[*] %f CPU Usage: %d\n", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, GetCPULoad());
            }
        }
        pthread_mutex_unlock(args->lock);
    }

}


int main(int argc, char *argv[]) {

        if (argc < 3 || argc > 5) {
                fprintf(stderr, "\nUsage: DAQTest [IP address] [fifo size]\nExample: DAQTest 192.168.1.81 64\n\n");
        fprintf(stderr, "Optional flags:    -v  Verbose (print operations)\n");
        fprintf(stderr, "       -vv Very verbose (also print data content)\n");
        fprintf(stderr, "       -vvv    Extremely verbose (also print DMA info)\n\n");
                exit(0);
        }
    if (isValidIpAddress(argv[1]) == 1) {
        fprintf(stderr, "[-] Invalid ip address\n");
        exit(0);
    }
        int fifosize = atoi(argv[2]);
        if (fifosize < 0 || fifosize > 8192) {
                fprintf(stderr, "[-] Invalid fifo size\n");
                exit(0);
        }

    char verbose = 0;
    if (argc == 4) {
        if (strcmp(argv[3], "-v") == 0) {
            verbose = 1;
        } else if (strcmp(argv[3], "-vv") == 0) {
            verbose = 2;
        } else if (strcmp(argv[3], "-vvv") == 0) {
            verbose = 3;
        } else {
            fprintf(stderr, "[-] Unwanted parameter\n");
            exit(0);
        }
    }

        struct sockaddr_in servaddr, cli;

        int sockfd = socket(AF_INET, SOCK_STREAM, 0);
        if (-1 == sockfd) {
                fprintf(stderr, "[-] Socket creation failed\n");
                exit(0);
        }
        bzero(&servaddr, sizeof(servaddr));

        servaddr.sin_family = AF_INET;
        servaddr.sin_addr.s_addr = inet_addr(argv[1]);
        servaddr.sin_port = htons(PORT);

    /*
    struct timeval tv;
    tv.tv_sec = TIMEOUT;
    tv.tv_usec = 0;
    setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof tv);
    */

        if (connect(sockfd, (SA * ) & servaddr, sizeof(servaddr)) != 0) {
                fprintf(stderr, "[-] Connection failed\n");
                exit(0);
        }

    fprintf(stderr, "[+] Connected to the server\n");

    int buffsize = 0;
    char* recv_buffer = (char*)&buffsize;
    int remaining = sizeof(int);
    int received = 0;
    int result = 0;
    while (remaining > 0) {
            result = recv(sockfd, recv_buffer + received, remaining, 0);
            if (result > 0) {
                remaining -= result;
                received += result;
            } else if (result == 0) {
                fprintf(stderr, "[-] Remote side closed his end of the connection before all data was received\n");
                exit(0);
            } else if (result < 0) {
                fprintf(stderr, "[-] Error retrieving configuration from the server\n");
            exit(0);
            }
    }

    //fprintf(stderr, "[*] Page size: %ld\n", sysconf(_SC_PAGE_SIZE));

        int dh = open("/dev/mem", O_RDWR | O_SYNC);

        unsigned int *virtual_address = mmap(NULL, 65535, PROT_READ | PROT_WRITE, MAP_SHARED, dh, VIRT_ADDR);
    unsigned int *virtual_sendbuff = mmap(NULL, buffsize, PROT_READ | PROT_WRITE, MAP_SHARED, dh, SEND_ADDR);
    unsigned int *virtual_recvbuff = mmap(NULL, buffsize, PROT_READ | PROT_WRITE, MAP_SHARED, dh, RECV_ADDR);
    unsigned int *virtual_fifo = mmap(NULL, (fifosize + 1) * buffsize, PROT_READ | PROT_WRITE, MAP_SHARED, dh, FIFO_ADDR);
    if (virtual_address == MAP_FAILED) {
        fprintf(stderr, "[-] AXI DMA registers mmap failed\n");
    }
    if (virtual_sendbuff == MAP_FAILED) {
        fprintf(stderr, "[-] Send buffer mmap failed\n");
    }
    if (virtual_sendbuff == MAP_FAILED) {
        fprintf(stderr, "[-] Send buffer mmap failed\n");
    }
    if (virtual_recvbuff == MAP_FAILED) {
        fprintf(stderr, "[-] Receiver buffer mmap failed\n");
    }
    if (virtual_fifo == MAP_FAILED) {
        fprintf(stderr, "[-] Fifo mmap failed\n");
    }
    memset(virtual_address, 0, buffsize);
    memset(virtual_sendbuff, 0, buffsize);
    memset(virtual_recvbuff, 0, buffsize);
    memset(virtual_fifo, 0, buffsize);
    int head = 0, tail = 0;

    pthread_t sendth, recvth;
    pthread_mutex_t lock;
    pthread_mutex_init(&lock, NULL);

    arguments send_args;
    send_args.virtual_address = virtual_address;
    send_args.virtual_buffer = virtual_sendbuff;
    send_args.head = &head;
    send_args.tail = &tail;
    send_args.buffsize = buffsize;
    send_args.fifosize = fifosize;
    send_args.sockfd = sockfd;
    send_args.lock = &lock;
    send_args.verbose = verbose;

    arguments recv_args;
    recv_args.virtual_address = virtual_address;
    recv_args.virtual_buffer = virtual_recvbuff;
    recv_args.head = &head;
    recv_args.tail = &tail;
    recv_args.buffsize = buffsize;
    recv_args.fifosize = fifosize;
    recv_args.sockfd = sockfd;
    recv_args.lock = &lock;
    recv_args.verbose = verbose;

    start_time = clock();

    if (pthread_create(&sendth, NULL, sender, &send_args)) {
        fprintf(stderr, "[-] Error creating sender thread\n");
        exit(0);
    }
    if (pthread_create(&recvth, NULL, receiver, &recv_args)) {
        fprintf(stderr, "[-] Error creating receiver thread\n");
        exit(0);
    }

    if (pthread_join(sendth, NULL)) {
        fprintf(stderr, "[-] Error joining sender thread\n");
        exit(0);
    }
    if (pthread_join(recvth, NULL)) {
        fprintf(stderr, "[-] Error joining receiver thread\n");
        exit(0);
    }

    pthread_mutex_destroy(&lock);
        close(sockfd);

    fprintf(stderr, "[+] Exit\n");

    return 0;

}

实际输出显示来自 recv 线程的打印，而发送线程打印

[*] Send Head: %d Tail: %d\n

几乎总是，有时它必须做什么。

Answer 1

在 sender 函数中，您有这样的代码(大大缩短了)

if (*(args->head) != *(args->tail)) {
    // ...

    pthread_mutex_unlock(args->lock);
    if (args->verbose > 2) {
        fprintf(stderr, "[*] %f CPU Usage: %d\n", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, GetCPULoad());
    }
}
pthread_mutex_unlock(args->lock);

这将导致 pthread_mutex_unlock(args->lock) 将在同一个锁上被调用两次，中间没有锁定。

例如可以看出this POSIX reference for pthread_mutex_unlock尝试解锁不属于您的线程的互斥锁(例如解锁已经解锁的互斥锁，或解锁另一个线程持有的互斥锁)会导致未定义的行为。

这需要以某种方式避免，例如将最后一个解锁调用放在 else 子句中:

if (*(args->head) != *(args->tail)) {
    // ...

    pthread_mutex_unlock(args->lock);
    if (args->verbose > 2) {
        fprintf(stderr, "[*] %f CPU Usage: %d\n", ((double)(clock() - start_time)) / CLOCKS_PER_SEC, GetCPULoad());
    }
}
else {
    pthread_mutex_unlock(args->lock);
}