C 读取套接字偶尔发生变化

Question

所以我有一个网络服务器，它 fork 成子进程，这些子进程将通过多线程系统处理多个客户端。

但是有一个问题。有时，accept() 返回错误的文件描述符!使用 ab 也测试 Web 服务器，有时我会收到该错误。因此，假设前 500 个请求处理得很好，但是文件描述符错误 - 套接字 fd 突然看起来像一个真正的大整数(不再是有效的整数)。

我只关闭它一次 - 在程序结束时，所以它突然关闭是没有意义的 - gdb 断点保证我它正确退出。

不过，我还有另一个问题，这似乎是子进程创建的线程“随机”退出。我知道它不是随机的，但我找不到它存在的原因，因为没有任何内容输出到屏幕上。我说它是“随机的”，因为 helgrind 告诉我退出线程仍然持有两个锁(我在互斥体中有一个互斥体(并且需要它) - 我也尝试过仅使用一个互斥体，但问题仍然存在)。是否有可能导致线程关闭的原因？

我可以提供代码，但它有点广泛，并且有葡萄牙语注释，变量/函数名称也有葡萄牙语。

更新

这是我的代码片段:

void *gere_processo(void*pst)
{
    process_struct pstruct = *(process_struct*)pst;

    int fork_id;
    int fd[2];

    if(pipe(fd)==-1)
    {
        perror("Erro a criar pipe: ");
        exit(-1);
    }

    // Começa a 0 antes de fazer fork!
    nthreads = 0;

    mainpid = getpid();

    printf("\nEntering thread %d\n\n", (unsigned int)pthread_self());

    //assim nao temos que usar mux mais abaixo quando fazemos uso destes valores na entrada da funcao
    pthread_mutex_lock(&processofilhomux);
    processofilho++;
    int localprocessofilho = processofilho;
    pthread_mutex_unlock(&processofilhomux);

    pthread_mutex_lock(&processopaimux);
    processopai++;
    int localprocessopai = processopai; 
    pthread_mutex_unlock(&processopaimux);

    fork_id=fork();
    if (fork_id==-1)
    {
        perror("Erro a criar filho:");
        exit(-1);
    }
    else if(fork_id==0)
    {
        if(signal(SIGUSR1, func_forcereadconfig) == SIG_ERR)
            printf("Erro no sinal SIGUSR1");
        if(signal(SIGINT, func_forcedsigint) == SIG_ERR)
            printf("Erro no sinal SIGINT");

        /***********************************************************************
        * Codigo do Filho
        * Espera clientes e cria uma thread para cada um.
        * Sempre que um cliente é criado, envia uma mensagem para o pai com o numero de clientes activos.
        * Sempre que um cliente é terminado, envia uma mensagem para o pai com o numero de clientes activos.
        *************************************************************************/

        int i, active_n=0;
        char address[BUFFSIZE];
        int msgsock;

        char nbuffer[50];

        pthread_attr_t attr;
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_JOINABLE);

        thread_s threadtable[MAXCLIENTSPERFORK];
        int a;
        for(a=0;a<MAXCLIENTSPERFORK;a++)
        {
            threadtable[a].thread = 0;
            threadtable[a].active = false;
            threadtable[a].t_buffer = NULL;
        }

        // Tempo inicial
        // Se passado X segundos nao existirem pedidos, o fork sai (desde que nao seja a primeiro child)
        struct timeval begin, now;
        gettimeofday(&begin, NULL);

        int timeout = 3;
        double timediff;

        while(cicle)
        {
            gettimeofday(&now , NULL);

            // Tempo que passou desde que iniciamos o timer
            timediff = (now.tv_sec - begin.tv_sec) + 1e-6 * (now.tv_usec - begin.tv_usec);

            //printf("\nServer (2) Socket: %d\n", pstruct.recsock);

            pthread_mutex_lock(&nthreadsmux);

            if(localprocessofilho == 2)
            {
                printf("\n[%d] - Tempo que passou: %lf - %d\n", localprocessofilho, timediff, nthreads);
            }

            // Condicoes para execuçao:
            // - nthreads = 0 e timeout < timediff (queremos tentar ate existir um timeout)
            // - nthreads < MAXCLIENTSPERFORK mas maior que zero
            // - nthreads < MAXCLIENTSPERFORK e processo filho é o primeiro (corre sempre)
            if((nthreads == 0 && timeout > timediff) || nthreads < MAXCLIENTSPERFORK && nthreads > 0 || (nthreads < MAXCLIENTSPERFORK && localprocessofilho == 1))
            {
                pthread_mutex_unlock(&nthreadsmux);

                // Espera um cliente
                printf("\tProcesso %d espera de clientes\n", localprocessofilho);
                if((msgsock = espera_pedido(pstruct.recsock, address)) < 0)
                    continue;

                pthread_mutex_lock(&athreadmux);

                // Figure out our thread number
                int a;
                for(a=0;a<MAXCLIENTSPERFORK;a++)
                {
                    if(threadtable[a].active == false)
                    {
                        if(threadtable[a].thread != 0)
                            pthread_join(threadtable[a].thread, NULL); // Temos que garantir que acaba...antes de a usar
                        break;
                    }
                }

                threadtable[a].t_buffer = malloc(sizeof(thread_buffer));
                if(threadtable[a].t_buffer == NULL)
                {
                    perror("\nErro (0) ao alocar estrutura config:");
                    exit(-1);
                }

                threadtable[a].t_buffer->msg = malloc(sizeof(char)*BUFFSIZE);
                if(threadtable[a].t_buffer->msg == NULL)
                {
                    perror("\nErro (0) ao alocar estrutura config:");
                    exit(-1);
                }

                sprintf(threadtable[a].t_buffer->msg, "%s", address);

                threadtable[a].t_buffer->sock = msgsock;
                threadtable[a].t_buffer->conf = conf;
                threadtable[a].active = true;

                // localtime é Non-threadsafe portanto temos que correr fora das threads criadas para cada pedido
                time_t timer = time(NULL);
                threadtable[a].t_buffer->t = *localtime(&timer);
                threadtable[a].t_buffer->timer = timer;

                // Cria nova threads
                if (pthread_create(&threadtable[a].thread, &attr, thread_func_pedido, (void*)&threadtable[a]) != 0)
                {
                    printf("\n\nERROR: %d", errno);
                    perror("Erro a criar thread: ");
                    close(msgsock);
                    free(threadtable[a].t_buffer);

                    printf("\n\n\n");
                    sleep(10);

                    continue;
                }

                pthread_mutex_unlock(&athreadmux);


                pthread_mutex_lock(&nthreadsmux);
                memset(nbuffer, '\0', sizeof(nbuffer));
                sprintf(nbuffer, "%d", nthreads);
                if(nthreads >= MAXCLIENTSPERFORK)
                {
                    printf("\nReached MAXCLIENTSPERFORK in Child %d\n", localprocessofilho);
                }
                pthread_mutex_unlock(&nthreadsmux);

                //printf("\nProcesso %d enviar para o pai: %s\n", localprocessofilho, nbuffer);
                write(fd[WRITE], nbuffer, (strlen(nbuffer)+1));

                // Reset ao timer
                gettimeofday(&begin, NULL);
            }
            else if(nthreads == 0 && localprocessofilho > 1)
            {
                pthread_mutex_unlock(&nthreadsmux);
                printf("\n00Reached 0 in Child %d\n", localprocessofilho);
                break;
            }
            else
            {
                pthread_mutex_unlock(&nthreadsmux);
            }
        }

        for(a=0;a<MAXCLIENTSPERFORK;a++)
        {
            if(threadtable[a].active != false)
            {
                pthread_join(threadtable[a].thread, NULL);
            }
        }

        printf("\nSAI FILHO\n\n");
        sleep(10);
        _exit(0);
    }
    else
    {
        /***********************************************************************
        * Codigo do Pai
        * O pai apenas lê o que o filho envia pelo pipe.
        * O filho envia constantemente o numero de threads activas.
        * Quando n atinge zero, o processo mata-se a si próprio o pai espera pela sua morte.
        * Quando n atingoe MAXCLIENTSPERTHREAD o processo mete wantnewfork a TRUE.
        *************************************************************************/
        close(fd[WRITE]);

        char buf[10];
        pid_t result;
        int status;
        // Verifica o estado do child (se == 0 entao esta a correr)
        // Se este for o processo pai, nao é suposto sairmos, NUNCA (a menos que haja um sinal)
        while((result = waitpid(fork_id, &status, WNOHANG)) == 0 || localprocessopai == 1)
        {
            int retn = read(fd[READ], buf, 10);
            if(retn > 0)
            {
                int n = atoi(buf);
                //printf("\n--- PAI: %d |\n", n);

                if(n == 0 && localprocessopai > 1)
                {
                    int status = 0;
                    wait(&status);
                    printf("\n%d A sair...\n", localprocessopai);
                    break;
                }
                else if(n == MAXCLIENTSPERFORK)
                {
                    printf("\n\n\nQUEREMOS UM NOVO PROCESSO - %d!!!\n\n\n", n);
                    pthread_mutex_lock(&processomux);
                    wantnewfork = true;
                    pthread_mutex_unlock(&processomux);
                }
            }
        }

        free(pst);

        // Está na hora de sair daqui
        int ret;
        printf("\n\n\n%d SAIR DA THREAD - %d - %d - %d\n\n", (unsigned int)pthread_self(), localprocessopai, result, WIFEXITED(status));
        pthread_exit(&ret);
    }
}


void *thread_func_pedido(void * threadstruct)
{
    thread_s *tstmp = (thread_s*)threadstruct;
    thread_buffer t = *(*tstmp).t_buffer;

    char buffer[BUFSIZE];

    char ver;

    pthread_mutex_lock(&nthreadsmux);
    nthreads++;
    pthread_mutex_unlock(&nthreadsmux);

    /*Leitura do Pedido*/
    int totalread;
    if((totalread = recv(t.sock, buffer, BUFSIZE,0)) <= 0)
    {
        printf("\nTotal read: %d\n", totalread);
        perror("Erro lendo a request: ");

        wait(10);

        close(t.sock);

        pthread_mutex_lock(&nthreadsmux);
        if(nthreads > 0)
            nthreads--;
        pthread_mutex_unlock(&nthreadsmux);

        int ret;
        pthread_exit(&ret);
    }

    req_data pedido;
    pedido = init_data();

    /* Processamento do Pedido */
    if(buffer == NULL)
    {
        pedido.errorcode=400;
    }
    else
    {
        pedido.request = convert_get(buffer,&pedido.errorcode,&ver);

        if(pedido.errorcode==200)
        {
            if(special_request(t.sock, pedido.request,ver) == false)
            {
                pedido.errorcode = imprime_ficheiro(t.sock, pedido.request, ver, t.conf->httpdocs, t.conf->cgibin);
            }
        }
    }

    if (pedido.errorcode != 200)
    {
        imprime_erro(t.sock, pedido.errorcode, ver, pedido.request);
    }

    /*--- Armazenamento do Pedido ---*/
    pedido = gen_data(pedido.errorcode, (char*)t.msg, pedido.request, tstmp->t_buffer->t, tstmp->t_buffer->timer);

    pthread_mutex_lock( &mux );
    FILE* statfile = (FILE*) stat_init();
    stat_armazena_req(pedido,statfile);
    fclose(statfile);
    pthread_mutex_unlock( &mux );

    pedido=free_data(pedido);

    //printf("\nClosing socket %d\n", t.sock);
    close(t.sock);

    pthread_mutex_lock(&nthreadsmux);
    pthread_mutex_lock(&processofilhomux);
    if(nthreads > 0)
        nthreads--;
    //printf("\n [%d] nthreads reduced to %d\n",processofilho, nthreads);
    pthread_mutex_unlock(&processofilhomux);
    pthread_mutex_unlock(&nthreadsmux);

    // Free the thread buffer, we're going to realloc it later
    free(tstmp->t_buffer->msg);
    free(tstmp->t_buffer);
    tstmp->t_buffer = NULL;

    pthread_mutex_lock(&athreadmux);
    tstmp->active = false;
    pthread_mutex_unlock(&athreadmux);

    int ret;
    pthread_exit(&ret);
}


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


    // Mascara de Sinais 
    if(signal(SIGTERM, func_sigint) == SIG_ERR)
        printf("Erro no sinal SIGCONT");

    if(signal(SIGINT, func_sigint) == SIG_ERR)
        printf("Erro no sinal SIGINT");

    if(signal(SIGUSR1, func_readconfig) == SIG_ERR)
        printf("Erro no sinal SIGUSR1");


    // Socket
    sock = cria_socket(PORT);

    conf = read_config("./www.config");
    if(conf == NULL)
    {
        conf = (config*)malloc(sizeof(config));
        if(conf == NULL)
        {
            perror("\nErro (2) ao alocar estrutura config:");
            exit(-1);
        }

        // Set defaults
        sprintf(conf->httpdocs, DOCUMENT_ROOT);
        sprintf(conf->cgibin, CGI_ROOT);
    }

    /*
        Inicialmente é criado um processo que tratará de X threads/clientes
        Após cada cliente ser tratado, a respectiva thread é destruída
        Depois, caso a lista de threads esteja vazia, o processo mata-se a si próprio, informado o pai do sucedido.
        Só é criado um novo processo caso não existam processos livres (i.e. se houver muitos clientes concorrentes)
        O processo pai cria uma thread para cada processo filho criado,
        de forma a que estas consigam interagir com o mesmo através de um pipe
    */
    while (cicle)
    {
        //Aparentemente o ultimo processo ficou cheio, portanto temos que criar um novo
        pthread_mutex_lock(&processomux);
        if(wantnewfork)
        {
            pthread_mutex_unlock(&processomux);

            pthread_mutex_lock(&processopaimux);
            printf("\n-----------PROCESSO NUM: %d\n", processopai+1);
            pthread_mutex_unlock(&processopaimux);

            pthread_t forkthread;
            pthread_attr_t attr;
            pthread_attr_init(&attr);
            pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);


            process_struct *pstruct = malloc(sizeof(process_struct));
            if(pstruct == NULL)
            {
                perror("\nErro (0) ao alocar estrutura config:");
                exit(-1);
            }

            pstruct->recsock = sock;
            pstruct->conf = conf;

            //printf("\nServer (1) Socket: %d\n", pstruct->recsock);

            if (pthread_create(&forkthread, &attr, gere_processo, (void *)pstruct) != 0)
            {
                perror("Erro a criar thread gere_processo: ");
                exit(-1);
            }

            pthread_mutex_lock(&processomux);
            wantnewfork = false;
            pthread_mutex_unlock(&processomux);
        }
        else
            pthread_mutex_unlock(&processomux);

        usleep(10000);
    }

    close(sock);
    free(conf);

    printf("Sai Pai\n");
    return 0;
}

更新

我有一个名为 sock 的全局变量，它保存 fd。 pstruct 还有一个名为 revsock 的成员，它与 sock 具有相同的值。

显然，recsock 发生了变化(valgrind 不会提示)，但 sock 保持不变。

free(pst) 可能是它的原因 - 尽管我不确定它为什么执行(它不应该用于主进程 - 但这可以解释随机退出，并且会导致随机的错误文件描述符问题在不是第二个线程的线程上退出)。

我不明白为什么 free(pst) 会导致这种情况...... child 也应该有自己的 pst 结构。但如果我评论 free(pst) 我就不再遇到错误的文件描述符问题。尽管如此，free(pst) 应该在那里，它是“随机线程退出”导致它在不应该运行的时候运行。

Answer 1

如果表示套接字文件描述符的整数突然更改，并且您的程序没有故意更改它，那么您将遇到以下两个错误之一:

• 您正在使用线程，并且没有锁定线程对共享内存的访问。这可能会导致诸如在将指针设置为值之前使用指针之类的情况。如果该指针指向其中包含 fd 的内存，则可能会覆盖它。

• 您遇到了缓冲区溢出。在某个地方，您的程序正在将错误数据写入保存文件描述符的内存中。

为了解决这个问题，我建议使用调试器和硬件观察点来告诉您文件描述符的值何时发生变化。