Boost C++ Libraries: Ticket #13043: Serialized MPI doesn't properly handle cancellation of a request

michael.williams@… — Tue, 23 May 2017 15:38:01 GMT

The following changes seem to fix it, with analogous changes to the array version. I guarantee nothing, draw your own conclusions, of course.

template<typename T>
optional<status>
request::handle_serialized_irecv(request* self, request_action action)
{
  typedef detail::serialized_irecv_data<T> data_t;
  shared_ptr<data_t> data = static_pointer_cast<data_t>(self->m_data);
  if (action == ra_wait) {
    status stat;
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Wait for the count message to complete
      BOOST_MPI_CHECK_RESULT(MPI_Wait,
                             (self->m_requests, &stat.m_status));
	  if (stat.cancelled())
	  {
		  return stat;
	  }
      // Resize our buffer and get ready to receive its data
      data->ia.resize(data->count);
      BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                             (data->ia.address(), data->ia.size(), MPI_PACKED,
                              stat.source(), stat.tag(),
                              MPI_Comm(data->comm), self->m_requests + 1));
    }
    // Wait until we have received the entire message
    BOOST_MPI_CHECK_RESULT(MPI_Wait,
                           (self->m_requests + 1, &stat.m_status));
    data->deserialize(stat);
    return stat;
  } else if (action == ra_test) {
    status stat;
    int flag = 0;
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Check if the count message has completed
      BOOST_MPI_CHECK_RESULT(MPI_Test,
                             (self->m_requests, &flag, &stat.m_status));
	  if (stat.cancelled())
	  {
		  return stat;
	  }
      if (flag) {
        // Resize our buffer and get ready to receive its data
        data->ia.resize(data->count);
        BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                               (data->ia.address(), data->ia.size(),MPI_PACKED,
                                stat.source(), stat.tag(),
                                MPI_Comm(data->comm), self->m_requests + 1));
      } else
        return optional<status>(); // We have not finished yet
    }
    // Check if we have received the message data
    BOOST_MPI_CHECK_RESULT(MPI_Test,
                           (self->m_requests + 1, &flag, &stat.m_status));
    if (flag) {
      data->deserialize(stat);
      return stat;
    } else
      return optional<status>();
  }
  else if (action == ra_cancel)
  {
	  status stat;
	  BOOST_MPI_CHECK_RESULT(MPI_Cancel, (&self->m_requests[0]));
	  return optional<status>();
  }
  else
  {
    return optional<status>();
  }
}

michael.williams@… — Tue, 23 May 2017 18:39:40 GMT

I was close, there was a mistake in it.

template<typename T>
optional<status>
request::handle_serialized_irecv(request* self, request_action action)
{
  typedef detail::serialized_irecv_data<T> data_t;
  shared_ptr<data_t> data = static_pointer_cast<data_t>(self->m_data);
  if (action == ra_wait) {
	status stat;
	  if (self->m_requests[0] == MPI_REQUEST_NULL)
	  {
		  return optional<status>(); //done
	  }
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Wait for the count message to complete
      BOOST_MPI_CHECK_RESULT(MPI_Wait,
                             (self->m_requests, &stat.m_status));
	  if (stat.cancelled())
	  {
		  return stat;
	  }
      // Resize our buffer and get ready to receive its data
      data->ia.resize(data->count);
      BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                             (data->ia.address(), data->ia.size(), MPI_PACKED,
                              stat.source(), stat.tag(),
                              MPI_Comm(data->comm), self->m_requests + 1));
    }
    // Wait until we have received the entire message
    BOOST_MPI_CHECK_RESULT(MPI_Wait,
                           (self->m_requests + 1, &stat.m_status));
    data->deserialize(stat);
    return stat;
  } else if (action == ra_test) {
    status stat;
    int flag = 0;
	if (self->m_requests[0] == MPI_REQUEST_NULL)
	{
		return optional<status>(); //done
	}
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Check if the count message has completed
      BOOST_MPI_CHECK_RESULT(MPI_Test,
                             (self->m_requests, &flag, &stat.m_status));
	  if (stat.cancelled())
	  {
		  return stat;
	  }
      if (flag) {
        // Resize our buffer and get ready to receive its data
        data->ia.resize(data->count);
        BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                               (data->ia.address(), data->ia.size(),MPI_PACKED,
                                stat.source(), stat.tag(),
                                MPI_Comm(data->comm), self->m_requests + 1));
      } else
        return optional<status>(); // We have not finished yet
    }
    // Check if we have received the message data
    BOOST_MPI_CHECK_RESULT(MPI_Test,
                           (self->m_requests + 1, &flag, &stat.m_status));
    if (flag) {
      data->deserialize(stat);
      return stat;
    } else
      return optional<status>();
  }
  else if (action == ra_cancel)
  {
	  status stat;
	  BOOST_MPI_CHECK_RESULT(MPI_Cancel, (&self->m_requests[0]));
	  return optional<status>();
  }
  else
  {
    return optional<status>();
  }
}

michael.williams@… — Tue, 23 May 2017 20:09:37 GMT

Still had a mistake. Don't worry, this is the last one, till I find a mistake in it.

template<typename T>
optional<status>
request::handle_serialized_irecv(request* self, request_action action)
{
  typedef detail::serialized_irecv_data<T> data_t;
  shared_ptr<data_t> data = static_pointer_cast<data_t>(self->m_data);
  if (action == ra_wait) {
	  status stat;
	  if (self->m_requests[0] == MPI_REQUEST_NULL)
	  {
		  return optional<status>(); //done
	  }
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Wait for the count message to complete
      BOOST_MPI_CHECK_RESULT(MPI_Wait,
                             (self->m_requests, &stat.m_status));
	  if (stat.cancelled())
	  {
		  return stat;
	  }
      // Resize our buffer and get ready to receive its data
      data->ia.resize(data->count);
      BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                             (data->ia.address(), data->ia.size(), MPI_PACKED,
                              stat.source(), stat.tag(),
                              MPI_Comm(data->comm), self->m_requests + 1));
    }
    // Wait until we have received the entire message
    BOOST_MPI_CHECK_RESULT(MPI_Wait,
                           (self->m_requests + 1, &stat.m_status));
    data->deserialize(stat);
    return stat;
  } else if (action == ra_test) {
    status stat;
    int flag = 0;
	if (self->m_requests[0] == MPI_REQUEST_NULL)
	{
		return optional<status>(); //done
	}
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Check if the count message has completed
      BOOST_MPI_CHECK_RESULT(MPI_Test,
                             (self->m_requests, &flag, &stat.m_status));
      if (flag) {
		if (stat.cancelled())
		{
			return stat;
		}
        // Resize our buffer and get ready to receive its data
        data->ia.resize(data->count);
        BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                               (data->ia.address(), data->ia.size(),MPI_PACKED,
                                stat.source(), stat.tag(),
                                MPI_Comm(data->comm), self->m_requests + 1));
      } else
        return optional<status>(); // We have not finished yet
    }
    // Check if we have received the message data
    BOOST_MPI_CHECK_RESULT(MPI_Test,
                           (self->m_requests + 1, &flag, &stat.m_status));
    if (flag) {
      data->deserialize(stat);
      return stat;
    } else
      return optional<status>();
  }
  else if (action == ra_cancel)
  {
	  status stat;
	  BOOST_MPI_CHECK_RESULT(MPI_Cancel, (&self->m_requests[0]));
	  return optional<status>();
  }
  else
  {
    return optional<status>();
  }
}

michael.williams@… — Mon, 05 Jun 2017 13:24:20 GMT

I was wrong. It's simpler than all that. Sorry.

template<typename T>
optional<status>
request::handle_serialized_irecv(request* self, request_action action)
{
  typedef detail::serialized_irecv_data<T> data_t;
  shared_ptr<data_t> data = static_pointer_cast<data_t>(self->m_data);
  if (action == ra_wait) {
	  status stat;
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Wait for the count message to complete
      BOOST_MPI_CHECK_RESULT(MPI_Wait,
                             (self->m_requests, &stat.m_status));
	  if (stat.cancelled())
	  {
		  return stat;
	  }
      // Resize our buffer and get ready to receive its data
      data->ia.resize(data->count);
      BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                             (data->ia.address(), data->ia.size(), MPI_PACKED,
                              stat.source(), stat.tag(),
                              MPI_Comm(data->comm), self->m_requests + 1));
    }
    // Wait until we have received the entire message
    BOOST_MPI_CHECK_RESULT(MPI_Wait,
                           (self->m_requests + 1, &stat.m_status));
	if (stat.cancelled())
	{
		return stat;
	}
    data->deserialize(stat);
    return stat;
  } else if (action == ra_test) {
    status stat;
    int flag = 0;
    if (self->m_requests[1] == MPI_REQUEST_NULL) {
      // Check if the count message has completed
      BOOST_MPI_CHECK_RESULT(MPI_Test,
                             (self->m_requests, &flag, &stat.m_status));
      if (flag) {
		if (stat.cancelled())
		{
			return stat;
		}
        // Resize our buffer and get ready to receive its data
        data->ia.resize(data->count);
        BOOST_MPI_CHECK_RESULT(MPI_Irecv,
                               (data->ia.address(), data->ia.size(),MPI_PACKED,
                                stat.source(), stat.tag(),
                                MPI_Comm(data->comm), self->m_requests + 1));
      } else
        return optional<status>(); // We have not finished yet
    }
    // Check if we have received the message data
    BOOST_MPI_CHECK_RESULT(MPI_Test,
                           (self->m_requests + 1, &flag, &stat.m_status));
    if (flag) {
      if (stat.cancelled())
	  {
		  return stat;
	  }
      data->deserialize(stat);
      return stat;
    } else
      return optional<status>();
  }
  else if (action == ra_cancel)
  {
	  status stat;
	  if (self->m_requests[0] != MPI_REQUEST_NULL)
	  {
		  BOOST_MPI_CHECK_RESULT(MPI_Cancel, (&self->m_requests[0]));
	  }
	  if (self->m_requests[1] != MPI_REQUEST_NULL)
	  {
		  BOOST_MPI_CHECK_RESULT(MPI_Cancel, (&self->m_requests[0]));
	  }
	  return optional<status>();
  }
  else
  {
    return optional<status>();
  }
}