c - 使用路由器标识作为简单客户端/服务器程序的退出条件

Question

我有两个使用 CZMQ 的独立 C 程序。一个是客户端，一个是服务器。服务器将数据从一个文件传输到客户端，客户端将接收到的数据写入一个名为“testdata_out”的新文件。一旦客户端从服务器接收到最后的数据 block ，客户端就会退出。

问题是服务器坐在那里，等待发送更多的数据给客户端，但是客户端程序已经完全退出了。我查看了 GDB，服务器卡在了

zframe_t *identity = zframe_recv( router );

我已经使用 fprintf() 完成了详细的日志记录以确认是这种情况。

我正在使用 ZMQ File transfer model 3作为我的示例 - 我复制了确切的代码并对其进行了测试以确保它也能正常工作。

基本上我在寻找想法 - 如果身份验证不起作用，我可以使用什么作为退出条件？ 我能否以某种方式发送一个中止信号，因为我无法编辑 ZeroMQ 函数“zframe_recv”？或者，如果有人可以看到我的代码或设置有什么问题，那么允许我使用 ID 验证作为退出条件的修复会很棒。

这是服务器函数:

int zmq_send_internal_file( struct arg_struct *args )
{
   struct arg_struct *input = (struct arg_struct *)args;
   int node_num = input->node;
   const char *fp = input->file_name;
   int chunkNum = 0;
   FILE *file_to_xfer = fopen( fp, "r" );
   zctx_t *ctx = zctx_new( );
   void *router = zsocket_new (ctx, ZMQ_ROUTER);

   //two parts per msg so HWM is size PIPELINE * 2
   zsocket_set_hwm (router, PIPELINE * 2);
   char *complete_address = "tcp://127.0.0.1:6000";
   if( 0 == zsocket_connect (router, complete_address ) )
   {
       printf("logging here\n");
   }
   else
   {
       printf("logging here\n");
   }
   while (true)
   {
        //what if we don't check the sender identity as a breaking condition?
        //what if we check chunk size as the breaking condition instead? - this causes program to exit too early
       //first frame in each message is the sender identity
       zframe_t *identity = zframe_recv( router );
       if (!identity)
       {
           printf("no identity, breaking.\n");
           break; //if we can't find an id, we're all set
       }
       //second frame is 'fetch' command
       char *command = zstr_recv (router);
       assert (streq (command, "fetch"));
       free (command);

       //third frame is chunk offset in file
       char *offset_str = zstr_recv (router);
       size_t offset = atoi (offset_str);
       free (offset_str);

       fprintf( fp_clnt_log, "%s : zmq:frame 4\n", time_str );
       //fourth frame is max chunk size
       char *chunksz_str = zstr_recv (router);
       size_t chunksz = atoi (chunksz_str);
       free (chunksz_str);

       //read chunk of data from file
       fseek (file_to_xfer, offset, SEEK_SET);
       byte *data = malloc (chunksz);
       assert (data);

       //send resulting chunk to client
       size_t size = fread (data, 1, chunksz, file_to_xfer);
       zframe_t *chunk = zframe_new (data, size);
       zframe_send (&identity, router, ZFRAME_MORE);
       zframe_send (&chunk, router, 0);
       chunkNum++;
   }
   if ( 0 == fclose( file_to_xfer ) )
   {
      printf("file_to_xfer closed successfully. \n");
   }
   else
   {
      printf("failed to close file_to_xfer. Exiting.\n");
      return -1;
   }
      return 0;
}

这是客户端函数:

void
zmq_receive_file()
{
   FILE *fp_out = fopen ("/usr/vos/bin/testdata_out","w");
   if ( NULL == fp_out )
   {
       printf( "fp_out pointer is NULL.\n");
   }
       return;
   }

   zctx_t *ctx = zctx_new ();
   void *dealer = zsocket_new (ctx, ZMQ_DEALER);
   zsocket_bind (dealer, "tcp://*:6000");
   //up to PIPELINE number of chunks can be in transit
   size_t credit = PIPELINE;
   size_t total = 0; //total bytes received
   size_t chunks = 0; //total chunks received
   size_t offset = 0; //offset of next chunk request
   int chunkNum = 0;

   while (true)
   {
       while (credit) //ask for next chunks until credit is 0
       {
           printf("%s : ft_server has credit, requesting next chunk...\n", time_str);
           zstr_sendm (dealer, "fetch");
           zstr_sendfm (dealer, "%ld", offset);
           chunkNum++;
           chunks++;
       }

       zframe_t *chunk = zframe_recv (dealer);
       if( !chunk )
       {
           break; //shut down and quit
       }
       //if there IS a chunk, write it to testdata_out
       else
       {
           size_t bytes_in_frame = zframe_size( chunk );
           char *data = zframe_data( chunk );
           fwrite( data, bytes_in_frame, 1, fp_out );
       }

       time_str = asctime( timeinfo );
       chunks++;
       credit++;
       size_t size = zframe_size (chunk);
       total+=size;

       if (size < CHUNK_SIZE)
       {
           printf ( "%s : Chunk smaller than CHUNK_SIZE, all done.\n");
           break; //last chunk received, exit

//program exits and leaves server hanging and waiting for an identity, so it never exits
       }
   }
   fclose( fp_out );
}

Answer 1

您的服务器中的读取是阻塞的，即它只会在实际收到消息时返回。

这里有几个选项来获得你想要的行为:

1. 让客户端在完成时发送“关闭”消息(很像您发送“获取”消息)，然后在处理第二帧时向服务器添加逻辑，以便它在收到此消息时跳出循环关机消息。
2. 更改服务器以执行非阻塞接收调用(检查 zpoller 和 zloop)并正确处理消息缺失。