Introduce Fixed-Capacity Queues

[csegarragonz]

In this PR I fix a memory management problem with our queue implementation under very specific workloads stressing MPI.

The symptoms observed were a progressive increase in the memory footprint of the workers as the workload was being executed, which would revert back to normal after the experiment finished (so no memory leaks). The increase was so accute that it could drive worker processes into OOMing, and crashing.

The problem, as hinted, had to do with queues. Even though elements were popped from the queue, memory would not be returned to the OS and (for a reason I am still not sure about) the pointed-to memory (i.e. MPI messages) would not be made available until queues were cleared. It is clear how this can end up very badly if we are sending (i) many many messages, and (ii) very big messages.

The solution is, either implementing a queue that shrinks in size, or using a fixed-size queue. I have opted for the latter as it seemed easier. Note that, for the particular case of MPI, we never have many elements in the queue.

I use an off-the-shelve fixed-capacity single-producer single-consumer queue. The only drawback is that this queue does not support (as it comes) returning a reference to the first element in the queue, thus calls to MPI_Probe will now fail. The fix is easy, but given that probe is not used in any experiment, we don't fix it.

Lastly, only version 1.0.5 of the library is available in Conan. In this version, TSAN complains about data races in the queue. Said warnings are fixed in 1.0.6. I have opened a PR in conan center to bump the version there, once it is accepted we'll bump the dependency here as well.

[Shillaker]

We should really add tests for things like this, even though it's implicitly tested in the MPI tests, they probably don't check all the edge cases or do a very intense stress test. You can look at the tests for the existing queue and try and adapt them with new sections.

It makes me feel uneasy adding an extra locking layer on top of a lock-free queue class and essentially making it a locking queue. Couldn't the lock-free message queue here just be a std::list?

I think we should look at in-proc 0MQ pair sockets for this use-case as they do all the queueing and blocking for us, and also enforce the single-producer-single-consumer semantics.

[Shillaker]

I've added inproc PAIR sockets in this PR: #206

Would be good to see if these can work in place of the queues. We can also generate the in-proc label based on the world ID and rank pair so don't need to have a map lookup.

[Shillaker]

I'm not sure this test is really doing what it claims to do, it only shows that an unspecified error is thrown if we enqueue above the capacity. While this is a valid test, it could be improved by checking that the exception thrown is actually a timeout exception.

More importantly, we still need a test that checks that the queue blocks and eventually does return if it's full and a dequeue happens sufficiently quickly. Without this, there's some important locking and condition variable logic that's not checked (and we know from experience that this is where concurrency bugs hide).

I realise this is an edge case, but it's still important to test.

[csegarragonz]

See here

[Shillaker]

Nice, just one small test change.

[Shillaker]

I think these tests could still do a better job of exercising the await-to-enqueue logic.

Something like this is fairly certain to exercise it:

    FixedCapIntQueue q(2);
    int nMessages = 100;

    // Fast producer
    bool producerSuccess = false;
    std::thread producerThread([&q, nMessages, &producerSuccess] {
        for(int i = 0; i < nMessages; i++) {
            SLEEP_MS(1);
            q.enqueue(i);
        }
        
        producerSuccess = true;
    };
    
    // Slow consumer
    bool consumerSuccess = false;
    std::thread consumerThread([&latch, nMessages, &consumerSuccess] {
        for(int i = 0; i < nMessages; i++) {
            SLEEP_MS(100);
            int res = q.dequeue();            
            if(res != i) {
                return;
            }
        }
        
        consumerSuccess = true;
    });

    if (producerThread.joinable()) {
        producerThread.join();
    }

    if (consumerThread.joinable()) {
        consumerThread.join();
    }
    
    REQUIRE(producerSuccess);
    REQUIRE(consumerSuccess);

[csegarragonz]

👍 I add this test but halve the sleeping time in the slower thread so the test doesn't take too long to run.