특정 fd에 이벤트 발생시 rts에 시그널 전달

www.joinc.co.kr 에 있는 RTS와 스레드 풀 문서를 보고 실습을 했습니다.

#include <pthread.h>
#include <iostream>
#include <signal.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <vector>

#include <netinet/in.h>
#include <arpa/inet.h>

#include <map>

#ifndef __USE_GNU
#define __USE_GNU
#endif
#include <fcntl.h>

using namespace std;

pthread_mutex_t mutex_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t sync_cond  = PTHREAD_COND_INITIALIZER;

int gpid;
typedef struct _fd_sig
{
    int signum;
    int pid;
} fd_sig;

/*
 * 1
 * 쓰레드당 처리중인 소켓의 정보를 유지하기 
 * 위한 자료구조. 
 * key   : 처리중인 소켓의 수 
 * value : 쓰레드(RTS 정보)
 */
multimap<int, fd_sig> pool_list;
multimap<int, fd_sig>::iterator mi; 

// RTS overflow가 발생했을 때 실행되는 핸들러
void do_sigio(int signo)
{
    printf("SIGIO : RTS signal queue overflow\n");    
}

/* 
 * 기본 시그널 핸들로 초기화 및 등록
 * 여기에서는 RTS overflow의 처리를 위한 시그널 
 * 핸들러를 등록한다.  
 */
void init_signal_handler()
{
    struct sigaction sigact;
    sigemptyset(&sigact.sa_mask);
    sigact.sa_flags        = SA_SIGINFO;
    sigact.sa_restorer    = NULL;
    sigact.sa_handler    = do_sigio;
    if (sigaction(SIGIO, &sigact, NULL) < 0)
    {
        perror("sigaction SIGIO ");
        exit(0);
    }

    return ;
}

/*
 * 2
 * 인자로 주어지는 소켓지정자 fd가 RTS시그널을
 * 발생하도록 설정한다. 
 * 발생시키는 RTS시그널 번호는 sig_num에 의해서 
 * 결정된다. 
 */
int setup_sigio(int fd, int sig_num, int pid)
{
    if (fcntl(fd, F_SETFL, O_RDWR|O_NONBLOCK|O_ASYNC) < 0)
    {
        perror("fcntl NONBLOCK ");
        return -1;
    }
    if (fcntl(fd, F_SETSIG, SIGRTMIN+sig_num) < 0)
    {
        perror("fcntl SETSIG "); 
        return -1;
    }

    // 인자로 주어진 파일지정자 fd에서 이벤트가 발생할 경우  
    // pid를 가지는 쓰레드로 RTS SIGRTMIN+sig_num 시그널이 전달된다.  
    if (fcntl(fd, F_SETOWN, pid) < 0)
    {
        perror("fcntl SETOWN ");
        return -1;
    }
    return 0;
}

/*
 * 3 
 * 듣기 소켓을 만들고 
 * 연결을 기다린다. 
 * 만약 연결이 들어온다면 pool_list자료구조를 통해서 
 * 가장 적은 소켓을 처리하는 쓰레드를 알아오고 
 * 그 쓰레드에 해당되는 RTS시그널 번호로 RTS시그널을 
 * 발생하도록 소켓을 설정한다.  
 */
void *accept_listener(void *data)
{
    int server_sockfd, cli_sockfd;    
    fd_sig lfd_sig;
    int count;
    sigset_t set;
    int client_sockfd;
    socklen_t clilen;
    int ret;

    int signum = *((int *)data);
    struct sockaddr_in serveraddr, clientaddr;
    struct siginfo si;

    if ((server_sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        perror("socket error ");
        exit(0);
    }

    bzero(&serveraddr, sizeof(serveraddr));
    serveraddr.sin_family        = AF_INET;
    serveraddr.sin_addr.s_addr    = htonl(INADDR_ANY);
    serveraddr.sin_port            = htons(1234);

    if (bind (server_sockfd, (struct sockaddr *)&serveraddr, 
            sizeof(serveraddr)) == -1)
    {
        perror("bind error ");
        exit(0);
    }

    if (listen(server_sockfd, 5) == -1)
    {
        perror("listen error ");
        exit(0);
    }
    if (setup_sigio(server_sockfd, signum, getpid()) == -1) 
    {
        printf("sigio error\n");
        exit(0);
    }

    sigemptyset(&set);
    sigaddset(&set, SIGRTMIN+signum);
    sigprocmask(SIG_BLOCK, &set, NULL);

    pthread_mutex_lock(&mutex_lock);
    pthread_cond_signal(&sync_cond);
    pthread_mutex_unlock(&mutex_lock);

    while(1)
    {
        int ret;
        ret = sigwaitinfo(&set, &si);
        if (ret == SIGRTMIN+signum)
        {
            if (si.si_fd == server_sockfd) 
            {
                cli_sockfd = accept(server_sockfd, 
                                (struct sockaddr *)&clientaddr,
                                &clilen);
                if(cli_sockfd < 0)
                {
                    printf("Accept error\n");
                    continue;
                }
                mi = pool_list.begin();
                lfd_sig = mi->second;
                count = mi->first+1;

                /*
                 * 연결소켓에 대해서 SIGRTMIN+signum RTS를 발생하도록 
                 * 설정한다. 
                 * fcntl()을 위해서 pid를 넘기는걸 주목하기 바란다.  
                 */
                cout << "Accept " << cli_sockfd << " : " 
                    << mi->second.signum << " : " 
                    << mi->second.pid << endl;
                setup_sigio(cli_sockfd, mi->second.signum, mi->second.pid);          ---------->>>>>>여리서 호출시 F_SETOWN 에러~!!!
                pool_list.erase(mi);
                pool_list.insert(pair<int, fd_sig>(count, lfd_sig));
            }
            else
            {
            }
        }
    }
}

/*
 * 클라이언트와 데이터를 주고 받을 쓰레드 함수이다. 
 * 클라이언트로 부터 읽은 데이터를 반향(echo)한다. 
 */
void *jecho(void *rts_num)
{

    int signum = *((int *)rts_num);
    int ret; 
    socklen_t clen;
    char buf[256];
    struct sockaddr_in cname;
    int n;

    // 쓰레드의 PID를 얻어온다.     
    gpid = getpid(); 

    sigset_t set;

    sigemptyset(&set);
    sigaddset(&set, SIGRTMIN+signum);
    pthread_sigmask(SIG_BLOCK, &set, NULL);

    pthread_mutex_lock(&mutex_lock);
    pthread_cond_signal(&sync_cond);
    pthread_mutex_unlock(&mutex_lock);

    struct siginfo si;
    fd_sig lfd_sig;
    int count;
    int lpid;

    while(1)
    {
        clen = sizeof(cname);
        ret = sigwaitinfo(&set, &si);
        memset(buf,0x00, 256); 
        if (ret == SIGRTMIN+signum)
        {
            if ((n = read (si.si_fd, buf, 255)) <= 0)
            {
                printf("read error \n");
                close(si.si_fd);
                mi = pool_list.begin();
                while(mi != pool_list.end())
                {
                    if (mi->second.signum == signum)
                    {
                        lfd_sig = mi->second;
                        count = mi->first - 1;
                        pool_list.erase(mi);
                        pool_list.insert(pair<int, fd_sig>(count, lfd_sig));
                    }
                    *mi++;
                }
            }    
            else
            {
                getsockname(si.si_fd, (struct sockaddr *)&cname, &clen);
                printf("%s(%d) : %s", inet_ntoa(cname.sin_addr), signum, buf);
                write(si.si_fd, buf, strlen(buf));
            }
        }
        else
        {
        }
    }
}

int main(int argc, char **argv)
{
    struct siginfo si;
    int status;
    int k;
    fd_sig lfd_sig;
    sigset_t set;
    unsigned int i;

    if (argc !=2 )
    {
        printf("Usage : ./rts_th [thread num]\n");
        exit(1);
    }
    int thread_num = atoi(argv[1]);
    vector<void *(*)(void *)> thread_list;
    vector<pthread_t> tident(thread_num);
    pthread_attr_t myattr;
    init_signal_handler();

    sigemptyset(&set);
    sigaddset(&set, SIGRTMIN);
    sigprocmask(SIG_BLOCK,&set, NULL);

    thread_list.push_back(accept_listener);
    for (i = 0; i < thread_num; i++) 
    {    
        thread_list.push_back(jecho);
    }

    /*
     * 쓰레드를 생성한다.  
     * 첫번째 쓰레드는 accept()전용 쓰레드이며
     * 이후 생성되는 쓰레드가 클라이언트 통신전용 쓰레드이다. 
     * 쓰레드를 생성할때 넘어가는 인자 K는 쓰레드가 기다릴 
     * RTS 시그널 번호이다. 
     * 각 쓰레드는 SIGRTSMIN+k 번호를 가지는 RTS를 기다리게된다.  
     */    
    for (i = 0, k = 1; i < thread_list.size(); i++, k++)
    {
        /*
         * 메인 쓰레드와 생성되는 쓰레드간에 정확한 데이터 
         * 전달이 필요하므로 뮤텍스와 조건변수를 이용해서  
         * 쓰레드 동기화를 시켜준다.
         */
        pthread_mutex_lock(&mutex_lock);
        pthread_create(&tident[i], NULL, thread_list[i], (void *)&k);

        /*
         * 쓰레드 자료구조
         * 각 쓰레드에서 처리하는 RTS번호와 처리중인 소켓의 갯수를
         * 유지한다. 
         */
        lfd_sig.signum = k;  
        pthread_cond_wait(&sync_cond, &mutex_lock);
        lfd_sig.pid    = gpid;
        if (i !=0)
            pool_list.insert(pair<int, fd_sig>(0, lfd_sig));
        pthread_mutex_unlock(&mutex_lock);
    }

    cout << "Thread Join " << endl;
    for (i = 0; i < thread_list.size(); i++)
    {
        pthread_join(tident[i], (void **)&status);
    }
    return 1;
}
		

실행을 하면 fd에 이벤트 발생시 특정 rts로 시그널을 발생 시키도록 하는 구문에서요...

/*
** 인자로 주어진 fd에서 이벤트 발생시 RTS SIGRTMIN+sig_num에 시그널 전달
*/
if (fcntl(fd, F_SETOWN, pid) < 0)
{
_debug(_fl, "fcntl SETOWN \n");
return -1;
}

이부분에서 계속 에러가 납니다.

server_fd를 셋팅할때는 에러가 안나는데 client가 접근하여 accept하고 나서
셋팅하려면 에러가 납니다.

fd값과 pid값도 확인해 본 결과 제대로 됐는데
왜 에러가 나는지 모르겠습니다.

에러 메세지는 "Operation not permitted" 이렇게 나옵니다...

그런데 데이터를 쏘면 데이터는 받더군요..ㅡ.ㅡa

그런데 한번에 여러개의 패킷을 보내면 위와 같은 에러를 내면서 중간중간 패킷을 받지 못합니다..

왜 그러는지 알려주세요