Bulk HTTP request queue consumer

Question

Today I had to write a small tool to help me send HTTP requests in bulk. Rabbit was overloading my server, so I decided to change my consumers to buffer the contents of the request, before sending. After changing my API, I did this:

#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <ev.h>
#include <cpr/cpr.h>

typedef std::unordered_map<std::string, std::ostringstream> HttpBuffer;
const char* RabbitQueue = getenv("RABBIT_NAME");
char QueueAddress[128];

int main()
{
    sprintf(QueueAddress, "amqp://%s:%s", getenv("RABBIT_HOST"), getenv("RABBIT_PORT"));

    int MaxThreads = std::thread::hardware_concurrency();
    std::vector<std::thread> threads;

    for (int i = 0; i < MaxThreads; i++) {
        threads.push_back(std::thread(RabbitThread));
    }

    for (std::thread &thread : threads) {
        thread.join();
    }

    threads.clear();
}

void RabbitThread()
{
    struct ev_loop *loop = ev_loop_new(0);
    AMQP::LibEvHandler handler(loop);
    AMQP::TcpConnection connection(&handler, AMQP::Address(QueueAddress));
    AMQP::TcpChannel channel(&connection);

    AMQP::MessageCallback onMessage = [&channel](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered) {
        HttpBuffer[message.replyTo()] << message.message().c_str() << ",";
        if (HttpBuffer[message.replyTo()].tellp() < 300000) {
            channel.ack(deliveryTag);
            return;
        }

        cpr::PostCallback(handle_response, cpr::Url{"http://localhost:8888"}, cpr::Body{HttpBuffer[message.replyTo()].str()});
        HttpBuffer[message.replyTo()].str("");

        channel.ack(deliveryTag);
    };

    channel.declareQueue(RabbitQueue);
    channel.bindQueue("default", RabbitQueue, "default");
    channel.consume(RabbitQueue).onReceived(onMessage);

    ev_run(loop);
    ev_loop_destroy(loop);
}

What do you guys think? How can I improve this one?

Answer 1

I don't know rabbit and have only small nit-picks regarding C++.

emplace

You might consider constructing threads in place instead of moving them to vector.

Instead of:

threads.push_back(std::thread(RabbitThread));

I would try:

threads.emplace_back(RabbitThread);

clear()

Is threads.clear(); in main() necessary?

struct and null pointer

This looks like C-style struct keyword usage. It is very rare to see in C++ (not saying it is wrong).

struct ev_loop *loop = ev_loop_new(0);

This might be also ok and looks more familiar

ev_loop *loop = ev_loop_new(0);

One more thing and this is a wild guess. Is this zero really a NULL pointer?

struct ev_loop *loop = ev_loop_new(0);

This would be more C++11 way:

struct ev_loop *loop = ev_loop_new(nullptr);

hardware_concurrency()

I really have no idea where lies bottleneck in you use case but I suspect it is not CPU bound so maybe you might utilize even much more threads.

int MaxThreads = std::thread::hardware_concurrency();

Answer 2

I've never used the Rabbit libraries, so I'll only comment on the C++.

1. #include what you use

   I had to add

   #include <unordered_map>
   #include <stdlib.h>
   #include <string.h>
   #include <thread>
   #include <vector>
   

   before I could compile. Don't omit the essentials from code reviews!

2. Error handling

   Did you leave out all the error handling to simplify the review? If so, please tell us this in your question. Otherwise, it looks like you've not thought about it at all!

3. sprintf

   sprintf(QueueAddress, "amqp://%s:%s", getenv("RABBIT_HOST"), getenv("RABBIT_PORT"));
   

   There's no bounds checking here. You could use snprintf(QueueAddress, sizeof QueueAddress, ...), but I'd lean more to building a std::string (with + or with a std::ostringstream), and use its c_str() if you need a C-style string:

   const auto QueueAddress = std::string("amqp://") + getenv("RABBIT_HOST") + ':' + getenv("RABBIT_PORT");
   ....
   auto address = AMQP::Address(QueueAddress.c_str());
   

4. threads.push_back

   Generally, we prefer emplace_back() when creating new objects in a vector. This saves us from having to think whether the class has an efficient move assignment operator.

5. threads.clear()

   I'm not sure about clearing the vector immediately before it goes out of scope. I can see that it demonstrates a desire to clean up correctly - but perhaps it shows a lack of confidence in scoped variables. It's obviously good practice in garbage-collected environments, but in C++ it just adds noise to the program.

6. magic constants

   What's that 300000 doing in the middle of the program? I can't tell the significance of it, or how it was chosen; it looks like a policy choice mixed in with the implementation. I'd suggest creating a named constant so that we can understand why that test is there. I think it's the minimum buffered data to trigger a reply, so I'll call it BUFFER_THRESHOLD.

7. naming

   Naming standards can always be contentious, and if you have existing conventions, then you should stick with them. But I find the use of PascalCase for variables jarring - C++ code generally uses snake_case or camelCase here. I note that the Rabbit classes use PascalCase for classes, and camelCase for members, so I recommend doing likewise. Especially as your single-word variables (threads, channel, loop) are not capitalised.

   This actually is an issue here, as you create a type alias HttpBuffer but then refer to a variable of that name. It's not clear whether the typedef should be a declaration instead, or whether you meant to use the type later to declare a local. It appears that you meant the former, but a consistent naming scheme would have helped.

8. HttpBuffer[message.replyTo()]

   It's probably worth keeping a local for the result of this call in onMessage, as it's used four times:

   AMQP::MessageCallback onMessage = [&channel](const AMQP::Message &message, uint64_t deliveryTag, bool /*unused*/) {
       auto& reply = HttpBuffer[message.replyTo()];
       reply << message.message().c_str() << ",";
       if (reply.tellp() < BUFFER_THRESHOLD) {
           channel.ack(deliveryTag);
           return;
       }
   
       cpr::PostCallback(handle_response, cpr::Url{"http://localhost:8888"}, cpr::Body{reply.str()});
       reply.str("");
   
       channel.ack(deliveryTag);
   };
   

   (And what's the type of message.message()? If it's a std::string, then no need to convert to a C string for operator<<.

9. inverted condition

   In the callback, you have condition of the form:

   if (c) {
       foo();
       return;
   }
   ...;
   foo();
   }
   

   It's probably clearer to use the opposite condition to decide whether to take action:

   if (!c) {
       ...;
   }
   foo();
   }
   

   This gives you

   AMQP::MessageCallback onMessage = [&channel](const AMQP::Message &message, uint64_t deliveryTag, bool /*unused*/) {
       const auto& reply_to = message.replyTo();
       auto& reply = HttpBuffer[message.replyTo()];
       reply << message.message().c_str() << ",";
   
       if (reply.tellp() >= BUFFER_THRESHOLD) {
           cpr::PostCallback(handle_response, cpr::Url{"http://localhost:8888"}, cpr::Body{reply.str()});
           reply.str("");
       }
   
       channel.ack(deliveryTag);
   };
   

10. Final flushing

    It's not clear to me whether one or more threads may still have unflushed data at program exit. I can't see any code to ensure that all replies get transmitted.