How to Optimize Client/Server with C/C++ and Boost Asio

I have two applications which work like a TCP client/server.

First application is the client which uses OpenCV to detect and send commands via TCP to the Server which controls a mobile robot.

My applications work well if I'm in my developing computer, but when I test it in real world with my robot, i realize that I have some delays with the data exchanged between client and server. This happens because the computer where iItest the applications is a little bit slow compared to my developing computer which is faster and give no problems. In real world case, server doesn't receive packets from client in real time so it execute the operations with a delay.

So, the problem is when the client loose the detection and send commands to the server in order to stop it. The server receives packets with a delay so when clients sends stop (heading = 0, distance = 0, nodetection) server doesn't receive the command immediately because it is receivin开发者_运维知识库g previous command packets and so it stop only after few meters.

I'd like to find a solution in order to stop immediately the server and discard all the packages about the moving information because they are useless if the robot has to stop. In order to stop the robot I send a nodetecting package which unfortunately is not received in real time so the robot continue to move for a while. (I'm doing this test on the same machine, so I connect on localhost)

At the moment, client uses this code:

while (key_mode!='q')
    {
        //wait and error processing

        context.WaitAnyUpdateAll();

        // obtain al the metadata image,depthmap and scene

        Mat frame = getImageFromKinect();

        // do detection and tracking

        switch(mode)

        {
..
                               case ROBOT_CONTROL:

                 {

                 // Connect to the server

                 using boost::asio::ip::tcp;

                             boost::asio::io_service io_service;

                                 tcp::resolver resolver(io_service);

                     tcp::resolver::query query(tcp::v4(), server, boost::lexical_cast<string>(porta));

                                 tcp::resolver::iterator iterator = resolver.resolve(query);

                                 tcp::socket s(io_service); 

                                 try

                 {

                                 s.connect(*iterator);

                 }

                                 catch (boost::system::system_error const& e)

                 {

                                 std::cout << "Warning: could not connect to the server\n" << e.what() << "\nPossible Solution: try to check is Server is UP\n" << std::endl;

                 }

..
..
 float delta = heading - last_heading;

                     if (!is_equal(delta, 0.0)){ 

                    // heading_data = send_heading + token + boost::lexical_cast<std::string>(delta);

                    // heading_length = strlen(heading_data.c_str());

                     try

                     {

                        // boost::asio::write(s, boost::asio::buffer(heading_data, heading_length));  

                     }

                     catch (boost::system::system_error const& e)

                     {

                         std::cout << "Warning: could not send commands : " << e.what() << std::endl;

                     }



                     }

                                         last_heading = heading; // store current for next subtraction

                     #endif

                     #if 1

                                         heading_scalato = heading / 3.0;

                     heading_data = send_heading + token + boost::lexical_cast<std::string>(heading_scalato);

                     heading_length = strlen(heading_data.c_str());

                         try

                     {

                         boost::asio::write(s, boost::asio::buffer(heading_data, heading_length));  

                     }

                     catch (boost::system::system_error const& e)

                     {

                         std::cout << "Warning: could not send commands : " << e.what() << std::endl;

                     }

                                         #endif



                                        distance_data = send_distance + token + boost::lexical_cast<std::string>(distance);





                    distance_length = strlen(distance_data.c_str());



                    try

                    {

                    boost::asio::write(s, boost::asio::buffer(distance_data, distance_length));

                    }

                    catch (boost::system::system_error const& e)

                    {

                    std::cout << "Warning: could not connect : " << e.what() << std::endl;

                    }
..
..
// if it has to stop:

else

                {

                    // stop rover

                    //control.setHeading(0.0);

                    //control.setDistance(0.0);

                                        float heading = 0.0;

                                        float distance = 0.0;

                                        heading_data = send_heading + token + boost::lexical_cast<std::string>(heading);

                                        distance_data = send_distance + token + boost::lexical_cast<std::string>(distance);

                                        heading_length = heading_data.size();//strlen(heading_data.c_str());

                    distance_length = strlen(distance_data.c_str());



                        try

                    {

                                        boost::asio::write(s, boost::asio::buffer(heading_data, heading_length));

                                        boost::asio::write(s, boost::asio::buffer(distance_data, distance_length));

                    }

                    catch (boost::system::system_error const& e)

                    {

                        std::cout << "Warning: could not send commands : " << e.what() << std::endl;

                    }





                    // write info on image

                    char text[100];

                    sprintf(text,"ROBOT CONTROL: No detection");

                    putText(hogResultFrame,text,Point(4,89),FONT_HERSHEY_PLAIN,1,Scalar(0,0,0));

                    putText(hogResultFrame,text,Point(5,90),FONT_HERSHEY_PLAIN,1,Scalar(100,100,255));

                                        nodetection_length = nodetection.size();

                    try

                    {

                                        boost::asio::write(s, boost::asio::buffer(nodetection, nodetection_length));

                    }

                    catch (boost::system::system_error const& e)

                    {

                        std::cout << "Warning: could not send commands : " << e.what() << std::endl;

                    }

In server, i use:

void* runThread(void*)
  {

        while(Aria::getRunning())
        {
            if(start_routine){

                if(temp_heading < 0.0){
                printf("\n\nStarting Discovering routine, then sleeping 3 seconds.\a\n\n");
                robot.setRotVel(5.0);
                ArUtil::sleep(3000);
        temp_heading = -1;
        }           
            else if(temp_heading >= 0.0) {
        printf("\n\nStarting Clockwise Discovering routine, then sleeping 3 seconds.\a\n\n");
                robot.setRotVel(-5.0);
                ArUtil::sleep(3000);
        temp_heading = 1;
                } 

                } 


                if( !flag_heading && !flag_distance)
            {               

                myMutex.lock();
                temp_heading=m_heading;
                temp_distance=m_distance;
                myMutex.unlock();

                if (is_equal(temp_heading, 0.0)){
                    robot.setRotVel(0.0);
                    }
                else robot.setRotVel(-ArMath::radToDeg(temp_heading));

                if(temp_distance <= distanza_minima || is_equal(temp_distance, 0.0))
                    robot.setVel(0.0);
                else
                    robot.setVel(float(temp_distance/20));

                printf("runThread:: heading= %f distance = %f rob_vel = %f rob_rot_vel = %f\n",ArMath::radToDeg(temp_heading),temp_distance, robot.getVel(),robot.getRotVel());

                flag_heading = true;
                flag_distance = true;
                start_routine = false;
            }
                ArUtil::sleep(100);

        }
  }
DataLine GetValueFromLine(const std::string& sData) {

  std::string sName, sInteger;
  std::stringstream ss;
  DataLine Result;

  size_t sz = sData.find('@');

  sName = sData.substr(0,sz); // Just in case you need it later

  Result.sName = sName;

  sInteger = sData.substr(sz + 1,sData.length() - sz);

  ss.str(sInteger);

  ss >> Result.nNumber;

  if (ss.fail()) {

    // something went wrong, probably not an integer
  }

  return Result;
}
void session(socket_ptr sock)
{
  try
  {
    for (;;)
    {
      char data[max_length];

      boost::system::error_code error;
      size_t length = sock->read_some(boost::asio::buffer(data), error);
      data[length] = 0;
      if (error == boost::asio::error::eof)
        break; // Connection closed cleanly by peer.
      else if (error)
        throw boost::system::system_error(error); // Some other error.

      output = GetValueFromLine(data);



      std::cout << "*******************\n";
      comando = output.sName;
      valore = output.nNumber;
      if (output.sName == "nodetection"){
      start_routine = true;
      std::cout << "\nSto ricevendo: " << output.sName;
      }
      else if (output.sName == "heading"){
      start_routine = false;
      control.setHeading(output.nNumber);
      std::cout << "\nSto ricevendo: " << output.sName << "e heading: " << output.nNumber;
      }                                
      else if (output.sName == "distance"){
      start_routine = false;
      control.setDistance(output.nNumber);
      std::cout << "\nSto ricevendo: " << output.sName << "e distance: " << output.nNumber;
      }                             




     // boost::asio::write(*sock, boost::asio::buffer(data, length));

    }
  }
  catch (std::exception& e)
  {
    std::cerr << "Exception in thread: " << e.what() << "\n";
  }
}

void server(boost::asio::io_service& io_service, short port)
{
  tcp::acceptor a(io_service, tcp::endpoint(tcp::v4(), port));
  for (;;)
  {
    socket_ptr sock(new tcp::socket(io_service));
    a.accept(*sock);
    boost::thread t(boost::bind(session, sock));

  }
}

int main(int argc, char **argv)
{
         // control server initialitation..
         ....

          boost::asio::io_service io_service;  
          server(io_service, porta);

  return 0;                                    
}

I was thinking to force the client to close the TCP connection when it reaches a no detecting condition in order to force the server to reject the pending packets, but how I can do this? How to destroy s pointer in boost?

Are there any other solutions? If I close the connection, does the server reject the pending packets?

As I understand your problem, you have a message that you intend your application to abandon all current processing and flush the input queue. However, because the application is busy receiving and processing previous messages, it does not receive the abandon and flush message until all previous messages are processed - which makes abandon and flush a no operation

IMHO you need to design and code a multithreaded application.

One thread, as light weight as possible, reads the incoming messages as fast as possible and quickly checks for the abandon and flush message. If the message is OK, then it is added to a queue and the next message is checked.

The second thread pulls the messages from the queue where the messages are stored by the first thread and processes them, perhaps taking a long time to do so. From time time it checks for an abandon and flush signal from the first thread.

Another approach to consider: the application that sends the messages maintains the queue. When the application which receives the messages finishes processing a message, it sends a request for the next message. The sender only sends messages when requested. If the abandon and flush condition arises, the application sending the messages looks after this. The application receiving the messages only has to deal with one at a time. This approach greatly simplify the message receiver, at the cost of complexity in the sending application, a more elaborate communications protocol and, possibly, a reduction in maximum throughput.