Occasionally Publish() blocks forever causing service disruption

[aks-zero]

Hi

We are using the pubsub client library to send messages from a C++ application to pubsub topics in GCP and the functionality is working fine most of the time.

We are aware of the fact that the Publish() function is synchronous in nature and blocks until there is a final result. It blocks and finally invokes the callback function (via the future object returned by Publish()) in a separate thread.

Most of the time it returns immediately as message is dispatched successfully. Sometimes it takes up to few secs (4 sec max as configured) due to attempts on failure. But there was an occasion when the Publish() did not return and blocked forever. Because of the application thread that is responsible for publishing messages was deadlocked and we had ultimately to restart the application which caused massive service failure.

I am curious to know if anyone else came across this situation? Is there a solution to this problem? Is there a setting to return after few seconds in case of failure?

We are already using the RetryPolicyOption and BackoffPolicyOption as below while setting up connection during startup.

    options.set<pubsub::RetryPolicyOption>(pubsub::LimitedTimeRetryPolicy(max_retry_time).clone());
    options.set<pubsub::BackoffPolicyOption>(pubsub::ExponentialBackoffPolicy(initial_delay, final_delay, scale).clone()); 
    
    connection = pubsub::MakePublisherConnection(pubsubTopic, options);

Regards,
Ajay

[coryan]

We are aware of the fact that the Publish() function is synchronous in nature and blocks until there is a final result.

I assume you are referring to:

https://cloud.google.com/cpp/docs/reference/pubsub/latest/classgoogle_1_1cloud_1_1pubsub_1_1Publisher#classgoogle_1_1cloud_1_1pubsub_1_1Publisher_1adcaba4f8339e31d6d296eab026164a4a

It should not be synchronous. Normally control should return to the application as soon as the message is queued. The callback function is invoked when the response for a batch of messages is acked by the service.

The operation may block if the queue is full (or becomes full). In that case this option may be useful:

https://cloud.google.com/cpp/docs/reference/pubsub/latest/structgoogle_1_1cloud_1_1pubsub_1_1FullPublisherActionOption
https://cloud.google.com/cpp/docs/reference/pubsub/latest/namespacegoogle_1_1cloud_1_1pubsub_1ab3f4e687e783e30ae0d046c8f10ac14c

The default is to block until there is room in the queue for this message.

Because of the application thread that is responsible for publishing messages was deadlocked and we had ultimately to restart the application which caused massive service failure.

Ugh, I am sorry to hear that.

Is there a solution to this problem? Is there a setting to return after few seconds in case of failure?

See above. You can configure the client to reject messages that will fill the queue, or to simply ignore full queues, though this is only suitable if you don't care about message ordering.

There is always the possibility that there is a bug in the library. But we may need more details to help troubleshoot this, such as the batch settings, the retry settings, the size of the message that cause the problem, etc.

[aks-zero]

Hi Coryan,

Thank you so much for your reviewing my question and also for quick reply.

It should not be synchronous. Normally control should return to the application as soon as the message is queued. The callback function is invoked when the response for a batch of messages is acked by the service.

That's good to know. But then we may be doing something wrong. As you mentioned the Publish() function queues the message and returns immediately giving the control back to application. But when we call future.then(handler) from application, that's where it blocks, waits until it is the future object is satisfied (i.e. callback is invoked). If that didn't block, then we wouldn't run into this issue. Asynchronous behavior is highly desired for us. We tried by looking at the examples, couldn't find a way.

The code looks something like below. Please check and let me know if there is a better way.

    pubsub::Publisher publisher = pubsub::Publisher(connection);

    // constructs a string object from the vector, required by MessageBuilder 
    std::string data(avro_data.begin(), avro_data.end());
    pubsub::Message message = pubsub::MessageBuilder{}.SetData(data).SetAttribute("eventName", std::string(eventName)).Build();

    result = publisher.Publish(message);
    // Returns a future object that becomes satisfied when the message is published or on a unrecoverable error.
    // On success, the future is satisfied with the server-assigned ID of the message.
    // A callable function is invoked when the future is ready. The function might be called immediately, e.g., if the future is ready.

    // Callback function to check the final result of publish operation. Gets executed in a separate thread by the client library.
    auto handler = [topic, eventName](future<StatusOr<std::string>> ret) {
        auto pubStatus = ret.get();
        // Returns a Status object if it failed or message_id if was successful
        if (!pubStatus) {
            Status error = pubStatus.status();
            LOG_ERROR("pubsub_send_message: Failed to publish message. event: %s, error_message: %s, error_code: %s, error_info: %s\n",
                    eventName, error.message().c_str(), StatusCodeToString(error.code()).c_str(), error.error_info().reason().c_str());
        } else {
            std::string messageId = pubStatus.value();
            LOG_INFO("pubsub_send_message: Successfully published message. event: %s, topicName: %s, messageId: %s\n", 
                    eventName, topic.c_str(), messageId.c_str());
        }
    };

    auto done = result.then(handler);
    // Waits until the shared state becomes ready, then retrieves the value stored in the shared state.
    done.get();

FullPublisherActionOption suggested by you seems interesting and could be helpful to us in this situation.

Regards,
Ajay

[coryan]

The .then() call is non-blocking too (mostly see below), the call that is definitely blocking is .get().

    auto done = result.then(handler);
    // Waits until the shared state becomes ready, then retrieves the value stored in the shared state.
    done.get();

Do you need to call .get() here? It does not look like you need to. I would write something like:

void logResult(StatusOr<std::string> id, std::string const& topic, std:string const& eventName) {
  if (!id) {
    auto error = std::move(id).status();
    LOG_ERROR("pubsub_send_message: Failed to publish message. event: %s, error_message: %s, error_code: %s, error_info: %s\n",
        eventName.c_str(), error.message().c_str(), StatusCodeToString(error.code()).c_str(), error.error_info().reason().c_str());
    return;
  }
  LOG_INFO("pubsub_send_message: Successfully published message. event: %s, topicName: %s, messageId: %s\n", 
            eventName.c_str(), topic.c_str(), id->c_str());
}
... ... ...
  (void)publisher.Publish(std::move(message)).then([topic, name = std::string(eventName)](auto f) {
    logResult(f.get(), topic, name);
  });

That will publish the message (without blocking), then attach a callback (without blocking), and then ignore the result of attaching the callback (that is a google::cloud::future<void>). The callback will be invoked asynchronously, in a separate thread, when the message is published.

x.then(cb) might "block" if x is already satisfied. In this case the callback cb is called as part of the .then() call. In practice, that would be extremely rare, and your callback is not that long anyway.

[morwin1]

Dear Coryan

Thanks for your help on this topic! The code you suggested did indeed get rid of the blocking behaviour.
I have one more question, however: In my callback function, I would like to modify a datastructure, in our case a map, when an error occurs. However, I get the following error message when I try this:

AutomaticallyCreatedBackgroundThreads::Shutdown: Resource deadlock avoided (/usr/local/src/google-cloud-cpp/google/cloud/internal/background_threads_impl.cc:56)

I introduce a map and a mutex

static unordered_map<std::string, std::shared_ptr<pubsub::PublisherConnection>> pubsubConnections;
static std::shared_mutex pubsubConnectionsMutex;

and the callback I currently have is:

void logResult(google::cloud::StatusOr<std::string> id, std::string const& topic, std::string const& eventName) {
    if (!id) { // in case of error retrieve error status and log error
        auto error = std::move(id).status();
        LOG_ERROR("pubsub_send_message: Failed to publish message. event: %s, error_message: %s, error_code: %s, error_info: %s\n",
            eventName.c_str(), error.message().c_str(), StatusCodeToString(error.code()).c_str(), error.error_info().reason().c_str());
        // Delete the existing connection so that next publish operation will establish a new connection.
        auto it = pubsubConnections.find(topic);
        if (it != pubsubConnections.end()) {
            LOG_INFO("pubsub_send_message: Deleting client connection to the pusub service. topicName: %s\n", topic.c_str());
            std::unique_lock<std::shared_mutex> uniqueLock(pubsubConnectionsMutex);
            it->second.reset();
            pubsubConnections.erase(it);
            uniqueLock.unlock();
        }
        return;
    }
    // success
    LOG_INFO("pubsub_send_message: Successfully published message. event: %s, topicName: %s, messageId: %s\n", 
        eventName.c_str(), topic.c_str(), id->c_str());
}

    (void)publisher.Publish(std::move(message)).then([topic, name = std::string(eventName)](auto f) {
        logResult(f.get(), topic, name);
    });
}

As far as I understand, the error message indicates that a resource deadlock was detected and avoided during the shutdown of automatically created background threads in the Google Cloud C++ client library. Is there anyway to circumvent this problem?

Thanks in advance and best regards, Moritz

[coryan]

This is more complicated, but not incredibly so. In your callback you are deleting a PublisherConnection:

            std::unique_lock<std::shared_mutex> uniqueLock(pubsubConnectionsMutex);
            it->second.reset();                    // <<<< calls destructor
            pubsubConnections.erase(it);
            uniqueLock.unlock();

When you delete the publisher connection it cleans up all the resources it created. That includes joining background threads. One of the background threads is the thread calling the destructor. That would deadlock because a thread cannot join itself.

You need to avoid this somehow, either:

1. Use a different thread to delete the PublisherConnection, or
2. Provide your own threads to the PublisherConnection so it won't have any need to join them.

The second approach is (probably) easier to implement, and is covered in some detail here:

https://cloud.google.com/cpp/docs/background-threads#providing_your_own_thread_pool

---

PS: You probably need to start the critical section earlier, when you start reading the pubsubConnections data structure, and I would probably delete the object outside the critical section, e.g.:

            std::shared_ptr<pubsub::PublisherConnection> tmp;
            {
              std::unique_lock<std::shared_mutex> uniqueLock(pubsubConnectionsMutex);
              auto it = pubsubConnections.find(topic);
              if (it == pubsubConnections.end()) return;
              tmp.swap(it->second);
              pubsubConnections.erase(it);
            }
            return;

[morwin1]

Thanks for the quick reply! I will go ahead and try this! Much appreciated