尝试使用 FIFO 发送消息时出错

如何解决尝试使用 FIFO 发送消息时出错

我是这种编程的新手，所以如果这是一个转储问题，我很抱歉。我正在尝试做一个非常简单的任务，但我似乎不明白出了什么问题。

我有一个创建多个子进程的父进程，并且通过使用 FIFO 我想向所有子进程发送消息（例如“hi”），进程接收到消息，但是有无论我做什么，都会出现错误，而且似乎无法找到问题所在。

这是父级的主要功能：

ERR_CONNECTION_REFUSED localhost refused to connect.

这是我保存每个进程信息的类：

int main(int argc,char *argv[])
{
  int num_monitors,buf_size; //command line arguments
  num_monitors = stoi(argv[1]);
  buf_size = stoi(argv[2]);

  // Structures to store monitor info
  Monitors *m_info = new Monitors[num_monitors]; // Stores monitor pid & pipe ends

  create_n_monitors(m_info,num_monitors,buf_size,input_dir_path);
  sleep(1); // making sure all pipes get created

  for(int i=0; i<num_monitors; i++) // opening the write end now that the monitors have been created
  {
    m_info[i].write_fd = open(m_info[i].write_p,O_WRONLY | O_NONBLOCK);
    if (m_info[i].write_fd == -1){perror("open @ 27 main parent");exit(1);}
  }

  for(int i=0; i<num_monitors; i++)
    send_message(m_info[i].write_fd,(char*)"hi",buf_size);

  delete [] m_info;
  return 0;
}

这是我如何创建进程和管道（FIFO）：

class Monitors
{
public:
    pid_t m_pid;                // monitors's PID
    char read_p[32];         // write monitor - read parent pipe name 
    char write_p[32];        // read monitor - write parent pipe name
    int read_fd;                // file descriptor for read fifo of monitor
    int write_fd;               // file descriptor for write fifo of monitor
    Monitors();
    ~Monitors();
};

最后是进程的主要功能：

    void create_n_monitors(Monitors *m_info,int num_monitors,int buf_size,char *input_dir)
    {
      create_unique_fifo(true,NULL,0);  // creates a fifo file
    
      for (int i = 0; i < num_monitors; ++i) // create num monitors
        create_monitor(m_info,i,input_dir);
    }
    
    /* ========================================================================= */
    
    // Create a monitor and it's named fifos. Store it's info in <m_info[index]>.
    void create_monitor(Monitors *m_info,int index,char *input_dir)
    {
      create_unique_fifo(false,m_info,index);  // Create fifos
    
      pid_t pid = fork();
      
      if(pid == -1) {
        perror("fork");
        exit(1);
      }
      else if ( pid == 0) { // we are in the child monitor,read_p(read parent) : monitor's write end of the fifo
                            // write_p(write parent): monitor's read end
        char buf_size_str[15];
        sprintf(buf_size_str,"%d",buf_size); // buf_size must be a char*
        
        execl("./Monitor","Monitor",buf_size_str,m_info[index].read_p,m_info[index].write_p,(char * )NULL);
        perror("execl");
        exit(1);
      }
    
      //else
      m_info[index].m_pid = pid;  // Store it's pid
    }
    
    /* ========================================================================= */
    
    // If <setup> is true,create a directory to store fifos.
    void create_unique_fifo(bool setup,Monitors *m_info,int index)
    {
      static char fifo_name[32];
      static int counter = 0;
      
      if (setup == true)
      {
        char dir_path[] = "named_fifos";
        if (access(dir_path,F_OK) == 0)  // If dir already exists (due to abnormal previous termination,eg: SIGKILL)
          delete_flat_dir(dir_path);      // completely remove it
    
        if (mkdir(dir_path,0777) == -1){perror("mkdir @ unique_fifo");exit(1);}
        sprintf(fifo_name,"named_fifos/f");
        return;
      }
        struct stat stat_temp;
    
      // Create a unique name (e.g named_fifos1R,named_fifos6W )
        sprintf(m_info[index].read_p,"%s%d%c",fifo_name,counter,'R');
    
      // Create fifos
        if(stat(m_info[index].read_p,&stat_temp) == -1){
            if (mkfifo(m_info[index].read_p,0666) < 0 ){perror("mkfifo @ unique_fifo");exit(1);}
        }   
    
        m_info[index].read_fd = open(m_info[index].read_p,O_RDONLY | O_NONBLOCK);
      if (m_info[index].read_fd == -1) {perror("open @ 73 setup_monitors");exit(1);}
    
        sprintf(m_info[index].write_p,'W');
    
      ++counter; // counter used for pipe names
    }
    
    /* ========================================================================= */
    
    // Remove a flat directory and its contents.
    void delete_flat_dir(char *init_flat_path)
    {
      char flat_path[32];
      strcpy(flat_path,init_flat_path);
    
      DIR *dir = opendir(flat_path);
      if (dir == NULL){perror("opendir @ delete_flat_dir"); exit(1);}
    
      struct dirent *entry;
      while ((entry = readdir(dir)) != NULL)  // Delete contents/files
      {
        char *f_name = entry->d_name;
        if (!strcmp(f_name,".") || !strcmp(f_name,".."))
          continue;
    
        char f_path[32];
        snprintf(f_path,32,"%s/%s",flat_path,f_name);  // Remove file
        if (remove(f_path) == -1){perror("remove @ delete_flat_dir"); exit(1);}
      }
    
      // Remove dir
      if (closedir(dir) == -1){perror("closedir 2 @ delete_flat_dir"); exit(1);}
      if (rmdir(flat_path) == -1){perror("rmdir @ delete_flat_dir"); exit(1);}  
    }

Here are the functions used for the communication of the processes and parent: 

// Sends <message> to file descriptor <fd>
void send_message(int fd,char *message,int buf_size)
{
  int length = strlen(message);
  char buffer[10];
  sprintf(buffer,"%d@",length);
    
  write(fd,buffer,9);      // sending the number of bytes reader is about to read
  write(fd,message,length); // sending the message itself
}

char *read_message(int read_end_fd,int buf_size)
{
  char buffer[10];
  int fifo_buffer_size = buf_size;
  read(read_end_fd,9);
  char * tok = strtok(buffer,"@");
  int length = atoi(tok); // how many characters will be received
  char * input_read = new char[length + 1];

  char * str = input_read;
  int bytes_read = 0,total_bytes = 0; // We might need to read less or more bytes
  fifo_buffer_size = length < fifo_buffer_size ? length : fifo_buffer_size; // // than <buf_size>

  while(total_bytes < length)
  {
    str += bytes_read;      // move str pointer
    bytes_read = read(read_end_fd,str,fifo_buffer_size); //and read the next <buf_size> characters
    total_bytes += bytes_read;  // adding them to the total amount of bytes read altogether

    if((total_bytes + fifo_buffer_size) > length)
        fifo_buffer_size = length - total_bytes; // reading exactly the amount that's left
  }

  input_read[length] = '\0';
  return input_read; 
}

我正在使用 valgrind 调试器并在 num_monitors=5 和 buf_size=2 时得到以下结果：

int create_unique_fifo(char* read_fifo)
{
  int read_fd;
    struct stat stat_temp;

  // Create fifos
    if(stat(read_fifo,&stat_temp) == -1){
        if (mkfifo(read_fifo,0666) < 0 ){perror("mkfifo @ unique_fifo 37");exit(1);}
    }   

    read_fd = open(read_fifo,O_RDONLY);  // Open named pipe for reading
  if (read_fd == -1){perror("open @ monitor.cpp 1"); exit(1);}
  return read_fd;
}


int main(int argc,char* argv[])
{
  int buf_size = stoi(argv[1]);  // Process command line args
  char read_fifo[100],write_fifo[100];
  strcpy(write_fifo,argv[2]); // parent read - monitor write
  strcpy(read_fifo,argv[3]);  // parent write - monitor read

  int read_fd,write_fd;
  read_fd = create_unique_fifo(read_fifo);

  write_fd = open(write_fifo,O_WRONLY | O_NONBLOCK);  // Open named pipe for writing
  if (write_fd == -1){perror("open @ monitor.cpp 2"); exit(1);}

  char* message;
  message = read_message(read_fd,buf_size);
  cout << "2:" << message << endl;
  return 0;
}

知道为什么会发生这种情况吗？我在创建进程后添加了 sleep(1) 以确保及时创建所有先进先出，但仍然出现此错误...任何帮助将不胜感激

解决方法

在 send_message 中，您总是从 buffer 写入 9 个字节，但并非所有这些字节都已写入值。这是因为填充 sprintf 的 buffer 只写入前几个位置，可能少至 3 个。

解决方案是要么将它们全部初始化为 0

char buffer[10] = 0;

或者只写字符串中的字节数。从 sprintf 返回的值很容易知道这一点。

auto buf_len = sprintf(buffer,"%d@",length);
write(fd,buffer,buf_len);