Home

Shared memory between processes

This example demonstrates how to use QtSharedMemory to access a block of shared memory between processes.

factory.h:

#ifndef FACTORY_H
#define FACTORY_H
#include <qptrlist.h>
#include <qtsharedmemory.h>
#include <qobject.h>

class Worker {
public:
    inline Worker(const QString &key, int count) { k = key; cnt = count; }
    void start();

private:
    QString k;
    int cnt;
};

class Factory : public QObject
{
    Q_OBJECT
public:
    Factory(const QString &argv0, int workers);
    ~Factory();

    void startCounting();

public slots:
    void decreaseProcessCounter();
    void checkStatus();

private:
    int processCounter;
    int nworkers;
    int lastValue;
    QtSharedMemory sharedMemory;
    QString argv0;
};

#endif

factory.cpp:

#include "factory.h"
#include <qstringlist.h>
#include <qprocess.h>
#include <qtimer.h>
#include <qtsharedmemory.h>
#include <qapplication.h>
#include <stdlib.h>

/*
    Constructs a factory with \a nworkers workers, each working in its
    own process. Each worker gets a share of the job of counting to
    1000000. Adjusting the number of workers may change the overall
    time spent to reach the number 1000000.

    All counting is synchronous across processes.
*/
Factory::Factory(const QString &argv0, int workers)
    : sharedMemory("Shared memory segment used for counting")
{
    nworkers = workers;
    this->argv0 = argv0;
}

/*
    Creates the shared memory segment and starts the workers. Waits
    for all workers to detach from the segment, then destroys it.
*/
void Factory::startCounting()
{
    if (sharedMemory.exists())
        sharedMemory.destroy(QtSharedMemory::ForceNoWait);

    if (!sharedMemory.create(16)) {
        qFatal("Unable to create shared memory: %s",
               sharedMemory.errorString().latin1());
        return;
    }

    qDebug("Starting count to 1000000 using %i processes", nworkers);

    processCounter = nworkers;
    int cnt = 1000000;
    int share = cnt / nworkers;
    for (int i = 0; i < nworkers; ++i) {
        if (i == nworkers - 1) share = cnt;

        QStringList args;
        args << argv0;
        args << "0";
        args << QString::number(share);
        QProcess *proc = new QProcess(args, this);
        connect(proc, SIGNAL(processExited()), SLOT(decreaseProcessCounter()));
        proc->start();

        cnt -= share;
    }

    lastValue = 0;

    QTimer::singleShot(0, this, SLOT(checkStatus()));
}

void Factory::checkStatus()
{
    // wait until the threads have started.
    if (processCounter > 0) {
        if (!sharedMemory.lock() || !sharedMemory.attach()) {
            qDebug("The factory was unable to attach to the segment: %s",
                   sharedMemory.errorString().latin1());
            return;
        }

        int value = *(int *)sharedMemory.data();
        qDebug("Progress: %7i/%7i | %3i%% | %7i/sec", value, 1000000, value * 100 / 1000000,
               (value - lastValue));
        lastValue = value;

        if (!sharedMemory.detach()) {
            qDebug("The factory was unable to detach from the segment: %s",
                   sharedMemory.errorString().latin1());
            sharedMemory.unlock();
            return;
        }

        sharedMemory.unlock();

        QTimer::singleShot(1000, this, SLOT(checkStatus()));
    } else {
        if (sharedMemory.lock() && sharedMemory.attach()) {
            int value = *(int *)sharedMemory.data();
            qDebug("Progress: %7i/%7i | %3i%% | %7i/sec", value, 1000000, value * 100 / 1000000,
                   (value - lastValue));
            lastValue = value;

            sharedMemory.detach();
            sharedMemory.unlock();
        }
        if (!sharedMemory.destroy())
            qDebug("Failed to destroy the segment: %s",
                   sharedMemory.errorString().latin1());
        qApp->quit();
    }
}

void Factory::decreaseProcessCounter()
{
    --processCounter;
}

/*
    Waits until all threads have finished, then destructs the factory.
*/
Factory::~Factory()
{
}

/*
    A worker increments the first integer of the shared memory segment
    by one, cnt times. It does this in a locked region to avoid race
    conditions which are very likely to occur here.
*/
void Worker::start()
{
    // Initialize and attach to the shared memory segment
    QtSharedMemory mem(k);
    if (!mem.attach()) {
        qDebug("A worker failed to attach: %s", mem.errorString().latin1());
        exit(0);
    }

    // Loop cnt times, incrementing the first int value in the shared
    // memory segment in a locked region.
    for (int i = 0; i < cnt; ++i) {
        if (!mem.lock()) {
            qDebug("A worker failed to lock the segment: %s",
                   mem.errorString().latin1());
            mem.unlock();
            break;
        }

        // A sequence of operations that is likely to cause
        // unsynchronized access to break.
        volatile int *array = (int *) mem.data();
        volatile int a = array[0];
        ++a;
        array[0] = a;

        mem.unlock();
    }

    if (!mem.detach())
        qDebug("A worker failed to detach: %s", mem.errorString().latin1());
}

main.cpp:

#include "factory.h"
#include <qapplication.h>

int main(int argc, char *argv[])
{
    QApplication app(argc, argv, false);

    if (argc < 2) {
        qDebug("Usage: %s <nprocs>", argv[0]);
        qDebug("Counts to 1000000 using nprocs processes");
        return 1;
    }

    if (argc == 2) {
        Factory *f = new Factory(argv[0], QString(argv[1]).toInt());
        f->startCounting();
        return app.exec();
    } else {
        Worker w("Shared memory segment used for counting",
                 QString(argv[2]).toInt());
        w.start();
        qDebug("Worker is done!");
        return 0;
    }
};

threadcounter.pro:

TEMPLATE = app
CONFIG -= moc
CONFIG += thread debug console
INCLUDEPATH += .

include(../../src/qtsharedmemory.pri)

# Input
SOURCES += main.cpp factory.cpp
HEADERS += factory.h