Home

Shared memory between threads

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

factory.h:

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

class Worker : public QThread {
public:
    inline Worker(int count) { cnt = count; }
    void run();
private:
    int cnt;
};

class Factory
{
public:
    Factory(int workers);

    void startCounting();
private:
    QPtrList<Worker> workers;
    QtSharedMemory sharedMemory;
};

#endif

factory.cpp:

#include "factory.h"
#include <qtsharedmemory.h>
#ifdef Q_OS_UNIX
#include <unistd.h>
#endif

static void qt_sleep(int seconds)
{
#ifdef Q_OS_WIN32
        Sleep(seconds * 1000);
#else
        sleep(seconds);
#endif
}

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

    All counting is synchronous across threads.
*/
Factory::Factory(int nworkers)
    : sharedMemory("Shared memory segment used for counting")
{
    int cnt = 1000000;
    int share = cnt / nworkers;
    for (int i = 0; i < nworkers; ++i) {
        if (i == nworkers - 1) share = cnt;
        workers.append(new Worker(share));
        cnt -= share;
    }
    workers.setAutoDelete(true);
}

/*
    Creates the shared memory segment and starts the workers. Waits
    for all workers to detach from the segment, then destroys it.
*/
void Factory::startCounting()
{
    // Create the segment
    if (sharedMemory.exists())
        sharedMemory.destroy(QtSharedMemory::ForceNoWait);
    if (!sharedMemory.create(16)) {
        qFatal("Unable to allocate shared memory: %s",
               sharedMemory.errorString().latin1());
        return;
    }

    // Start all the threads
    qDebug("Starting count to 1000000 using %i threads", workers.count());
    QPtrListIterator<Worker> it(workers);
    while (it.current()) {
        (*it)->start();
        ++it;
    }

    // Loop until an error occurs, or until all threads are done
    int lastValue = 0;
    bool allDone;
    do {
        allDone = true;
        QPtrListIterator<Worker> it(workers);
        while (it.current()) {
            if (!(*it)->finished()) {
                allDone = false;
                break;
            }
            ++it;
        }

        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;
        } else if (sharedMemory.error() != QtSharedMemory::OutOfResources) {
            qDebug("The factory was unable to attach to the segment: %s",
                   sharedMemory.errorString().latin1());
            allDone = true;
        }

        sharedMemory.detach();
        sharedMemory.unlock();
        if (!allDone)
            qt_sleep(1);
    } while (!allDone);

    // Destroy the segment
    if (!sharedMemory.destroy()) {
        qDebug("Failed to destroy the segment: %s",
               sharedMemory.errorString().latin1());
    }
}

/*
    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::run()
{
    // Initialize and attach to the shared memory segment
    QtSharedMemory mem("Shared memory segment used for counting");

    // 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());
            break;
        }

        if (!mem.attach()) {
            if (mem.error() == QtSharedMemory::OutOfResources) {
                // out of resources
                qt_sleep(1);
                continue;
            }

            qDebug("A worker failed to attach to the segment: %s",
                   mem.errorString().latin1());
            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.detach();
        mem.unlock();
    }
}

main.cpp:

#include "factory.h"

int main(int argc, char *argv[])
{
    int threads = argc > 1 ? QString(argv[1]).toInt() : 0;
    if (argc < 2 || threads < 1) {
        qDebug("Usage: %s <nthreads>", argv[0]);
        qDebug("Counts to 1000000 using nthreads (at least 1) threads");
        return 1;
    }

    if (threads > 7) {
        qDebug("*** WARNING: Certain operating systems allow only a limited number");
        qDebug("             of shared memory attachments per process. See also");
        qDebug("             the QtSharedMemory::attach() documentation.");
    }

    Factory f(threads);
    f.startCounting();

    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