Sockets, pthread, pipe, and fork -- a messaging server

ggravarr · 06-27-2005, 11:19 AM

I'm writing a server that takes connections over a socket

Code:

int raw_tcp_listen(int port)
{
  int sock;
  if((sock=socket(AF_INET,SOCK_STREAM,0))<0)
    {perror("socket err");return -1;}
#ifdef FULL_DEBUG
  else
    printf("socket\n");
#endif

  struct sockaddr_in sin;
  memset(&sin,0,sizeof(sin));
  sin.sin_addr.s_addr=INADDR_ANY;
  sin.sin_family=AF_INET;
  sin.sin_port=htons(port);

  int RAval=1;
  if(setsockopt(sock,SOL_SOCKET,SO_REUSEADDR,&RAval,sizeof(RAval)) != 0)
    {perror("reuse set err");return -1;}
#ifdef FULL_DEBUG
  else
    printf("reuse addr set\n");
#endif

  linger ling;
  ling.l_onoff=0;
  ling.l_linger=1;
  if (setsockopt(sock,SOL_SOCKET,SO_LINGER,&ling,sizeof(ling)) != 0)
    {perror("linger set err");return -1;}
#ifdef FULL_DEBUG
  else
    printf("linger set\n");
#endif

  if(bind(sock,(struct sockaddr *)&sin,sizeof(sin))!=0)
    {perror("bind err");return -1;}
#ifdef FULL_DEBUG
  else
    printf("bind\n");
#endif

#ifdef FULL_DEBUG
  printf("listening to socket\n");
#endif
  if (listen(sock,2) != 0)
    {perror("listen err");return -1;}
#ifdef FULL_DEBUG
  else
    printf("listen\n");
#endif

  return(sock);
}

typedef struct {
  int err;
  int port;
} l_raw_t;

void *listening_function_raw(void *ptr)
{
  l_raw_t ipt = *(l_raw_t*)(ptr);

  int id =-1;
  int err = ipt.err;
  int port=ipt.port;

  while ((err) && (keep_on)) {
    int l_sd=raw_tcp_listen(port);
    if (l_sd > 0) {
      struct sockaddr_in csa;
      memset(&csa,0,sizeof(csa));
      socklen_t clen=sizeof(csa);
      int sd = accept(l_sd, (struct sockaddr*)&csa,&clen);

      if (close(l_sd) != 0)
        {perror("l_socket close");err=0;}

      if (sd < 0) {
        {perror("accept");err=0;}
      } else {
        printf ("accepted\n");
      }

      if (getpeername(sd,(struct sockaddr*)&csa,&clen) != 0)
        {perror("get ip");err=0;}

      char *c_ip = inet_ntoa(csa.sin_addr);
      printf("%s\n",c_ip);
      if (err) {
        int test=1;

        if (
            //make sure I know whois
            (strcmp("xxx.xxx.xxx.xxx",c_ip) != 0)
           )
          test=0;

        if (test) {

          pthread_mutex_lock(&con_cnt_mutex);
          id=con_cnt;
          ++con_cnt;
          pthread_mutex_unlock(&con_cnt_mutex);

#ifdef FULL_DEBUG
          printf("raw %i connected\n",id);

#endif

          printf("%s=raw %i\n",c_ip,id);

          pthread_mutex_lock(&raw_table_mutex);

          raw_table_t raw_add;

          raw_add.sd=sd;
          raw_add.id=id;
          raw_add.state=1;

          raw_table.push_back(raw_add);

          pthread_mutex_unlock(&raw_table_mutex);

          pthread_attr_t io_attr;
          pthread_attr_init(&io_attr);
          pthread_attr_setdetachstate(&io_attr,PTHREAD_CREATE_DETACHED);

          pthread_t tid;
          pthread_create
            (&tid, &io_attr,&io_function_raw, (void *) &id );
        }
      }
       err=1;
    } else sleep(1);
  }

  pthread_exit(NULL);
}

and then splits a thread off to run a command....

Code:

void *io_function_raw(void *ptr)
{
  int id= *(int *)ptr;
  int found=0;

  pthread_mutex_lock(&raw_table_mutex);

  for (int s=0;s < raw_table.size();s++)
    if (raw_table.at(s) == id) {
      raw_table.at(s).tid=pthread_self();
      found=1;
    }

  pthread_mutex_unlock(&raw_table_mutex);

  if (found) {
    while (keep_on) {
      int sd;
      int good=0;

      pthread_mutex_lock(&raw_table_mutex);

      for (int s=0;s < raw_table.size();s++)
        if (raw_table.at(s) == id) {
          sd=raw_table.at(s).sd;
          if (raw_table.at(s).active())
            good=1;
          break;
        }

      pthread_mutex_unlock(&raw_table_mutex);

      if (!good) {
#ifdef FULL_DEBUG
        printf("raw %i read error: R0062\n",id);
#endif
        break;
      }

#ifdef FULL_DEBUG
      printf("raw %i waiting for command\n",id);
#endif
      int err=0;

      char buf[100000]="";
      //read here
      err=read(sd,buf,99999);
      if (err > 0) {
#ifdef FULL_DEBUG
        printf("raw %i got command string len: %i\n",id,strlen(buf));
#endif

        pthread_mutex_lock(&exec_cnt_mutex);
        int eid = exec_cnt;
        ++exec_cnt;
        pthread_mutex_unlock(&exec_cnt_mutex);

        pthread_mutex_lock(&exec_table_mutex);

        exec_table_t exec_add;

        exec_add.id=eid;
        exec_add.cid=id;
        //strcpy(exec_add.cmdstring,strdup(buf));
        exec_add.cmdstring = strdup(buf);
        exec_add.cmeth=raw_w_type;
        exec_add.state=1;

        exec_table.push_back(exec_add);

        pthread_mutex_unlock(&exec_table_mutex);

        pthread_attr_t exec_attr;
        pthread_attr_init(&exec_attr);
        pthread_attr_setdetachstate(&exec_attr,PTHREAD_CREATE_DETACHED);

        pthread_t scmd;
        int scmd_tid;
        scmd_tid = pthread_create
          (&scmd,&exec_attr,&execute_command_string,(void *) &eid);

#ifdef FULL_DEBUG
        printf("ran exec\n");
#endif
      } else {
#ifdef FULL_DEBUG
        printf("raw %i read error: R0065\n",id);
#endif

        pthread_mutex_lock(&raw_table_mutex);

        for (int s=0;s < raw_table.size();s++)
          if (raw_table.at(s) == id) {
            raw_table.at(s).state=0;
            break;
          }

        pthread_mutex_unlock(&raw_table_mutex);

        break;
      }
    }
  }
  RAW_end(id);

  pthread_exit(NULL);
}

///function that selects command based on input
runs a the function for the command itself no thread out of here

and that all works fine.....

now, what I want to do is be able to set up a command to run specific things off of the CLI.... since I can't expect all the code I want to run to drop in source instantly(if ever)

and what I want to do for that... is in the command's thread forck off and exec

which I have working....

Code:

      int cli_stdout[2];

      if (pipe(cli_stdout) == -1) {
         std::cerr << cli_cmd.name << ": pipe error" << std::endl;
         cli_cmd.deactivate();
      }

      cli_pid=fork();
      switch (cli_pid) {
       case -1:
         std::cerr << cli_cmd.name << ": fork error" << std::endl;
         break;
       case 0: //child
         if (close(cli_stdout[0]) == -1)
           std::cerr << cli_cmd.name << ": c00 error" << std::endl;

         if (dup2(cli_stdout[1], STDOUT_FILENO) == -1)
           std::cerr << cli_cmd.name << ": dup error" << std::endl;

         if (close(cli_stdout[1]) == -1)
           std::cerr << cli_cmd.name << ": c10 error" << std::endl;

         if (execvp(cmd_name,cmd_arr) == -1)  {
         std::cerr << cli_cmd.name << ": exec error" << std::endl;
         cli_cmd.deactivate();
         _exit(42);
         } else _exit(45);
         break;
       default: //parent
         if (close(cli_stdout[1]) == -1)
           std::cerr << cli_cmd.name << ": c11 error" << std::endl;

         cli_cmd << "started\n";
           cli_cmd.send();
         cli_cmd.empty();

         while (cli_cmd.active()) {
            char cli_iobuf[500]="";

            if (read(cli_stdout[0],cli_iobuf,500) <= 0) {
                 cli_cmd.deactivate();
                 std::cerr << cli_cmd.name << ": read error" << std::endl;
              } else {
                 cli_cmd << cli_iobuf;
                 cli_cmd.send();
                 cli_cmd.empty();
              }
         }
         if (close(cli_stdout[0]) == -1)
           std::cerr << cli_cmd.name << ": c01 close error" << std::endl;
         break;
      }

      if (kill(cli_pid,SIGKILL) != 0)
        std::cerr << cli_cmd.name << ": kill error" << std::endl;

      cli_cmd.clear();
      std::cout << cli_cmd.name << ": ended" << std::endl;
   }
}

the only problem is.... that once the fork is open it only accepts one more socket and then I start getting bind err: Address already in use

once the fork ends one way or another (exit or kill) the socket is able to be attached to again....

I assume this is due to something in the fork environment cloning.... I just haven't hacked it quite that deep....

and was hoping someone out ther could help either waylay or at least guide me in that process.

and, yeah... I know I kinda just gave sporadic pieces of the code, but this is already too long.... let me know if I missed something

oh yeah... and I've taken the exec completely out and just done for () cout << stuff; exit.... that didn't change it.... so I don't think it's that.

cppkid · 06-28-2005, 04:54 AM

Hi buddy.
I try to read your code but i didn't understand that fully. I think the problem is because when you do the fork() after creating a socket. I think it should give bind error, because fork() makes another copy of that socket and now two processes have two seperate sockets but are trying to bind to a single address or port.
So if you want to have multiple connections, i think the solution is somewhere in server_sockets.

Wish this will help you.

ggravarr · 06-28-2005, 01:45 PM

yeah.... I'm pretty sure that's what's happening too... I thought that wouldn't be an issue because the socket accepting and listening happens in a different pthread than the fork is trying to run and all the socket-bind-listen-accept methods are local to their threaded functions....

Ok, that's true... but the command thread is a child of the listening thread...

I have a couple different socket types listening. I tried running the fork in one and then connecting on another.... the socket bind issues happen on both... so I guess it's not inheriting from the local variables of the calling function of the parent thread....

I'm going to try and design a pthread_atfork handler that can stop this... but, since the listens happen inside a different thread, and you can't pass variables to atforks(with good reason), its going to be difficult to get the handler to know what its looking for... I guess I could just make the listening socket global... but I highly dislike unnecissary global variables... which kinda confuses me how fork could be cloning something that I can't get an atfork to see.... but, I can't really profess to know all the detailed inner workings of linux process cloning.

if that doesn't work.... I guess I'm going to try and build my own fork method out of a clone call... and if that doesn't work..... ..... ..... I'm lost again.

so, what are these server_sockets of which you speak.... were you just saying I should look at the configuration of my socket listeners(seems quite a good idea really -- I'm doing reuseaddr w/o linger, but all I've really tried other than that is to set O_NONBLOCK on my fd).... or is server_sockets some socketing method of which I was previously(and I guess still am cause google ain't got it) unaware of.

ggravarr · 06-28-2005, 02:39 PM

Ok, so all I had to do was make the listening sockets fd global and put a close atfork.... makes sense.... I guess I was kind of hoping that the scope was a little thinner... but that would mean more proc at time of fork... and deviation from standards.... so, I just didn't think about it... anyway, it was in the end a simple fix(as most of the right ones are).... thanks.... It was in fact helpful... I was focusing far too much on the event that was causing the problem and not enough at the objects that were failing...