[Xapian-discuss] Threaded test (in C++) to reproduce our database problems

[Richard Boulton]

This is a multi-part message in MIME format.
--------------030706000907010606080908
Content-Type: text/plain; charset=us-ascii; format=flowed
Content-Transfer-Encoding: 7bit

For the record, running a test with helgrind (a skin for valgrind to 
detect race conditions) throws up a large number of race conditions, 
mainly involving strings.  Since the tests don't obviously share any 
string data, I assume this is either something happening behind the 
scenes in the string class, or a false report from helgrind (which occur 
a lot).  If you still have problems, it would certainly be worth looking 
into these reports further.

My amended test case which I've been running with helgrind is attached 
(I put a mutex round the output lines to remove a large number of 
obvious false positives).

Eric Ridge wrote:
> As an aside, is it necessary to compile Xapian with -pthread or
> -lpthread or -D_REENTRANT or -D_THREAD_SAFE or something?  My
> thinking is no, but I'm just a novice when it comes to pthreads.

It depends on your system really, but I would say that it would be a 
good idea to define both _REENTRANT and _THREAD_SAFE if compiling for 
use in a multithreaded environment.  I doubt that doing so will cause 
any harm, at least.

Actually, it occurs to me that this could be the reason that Olly saw 
"Interrupted system call" errors failing to be caught - if the relevant 
file isn't compiled to use thread-local errno values, a race condition 
could cause the wrong error condition to be reported.

-- 
Richard

--------------030706000907010606080908
Content-Type: text/x-c++;
 name="xapian_threads1.cpp"
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="xapian_threads1.cpp"

#include <xapian.h>
#include <pthread.h>
#include <unistd.h>

#define DATABASE "/tmp/xapian_threads.db"
#define MAX_THREADS 3

using namespace std;

pthread_t _writer;
pthread_t _readers[MAX_THREADS];
pthread_mutex_t output_mutex;

void msg(const char * msg) {
    pthread_mutex_lock(&output_mutex);
    fprintf(stderr, "%s\n", msg);
    pthread_mutex_unlock(&output_mutex);
}

void* do_writes(void *) {
    unlink(DATABASE"/db_lock");
    Xapian::WritableDatabase db = Xapian::Auto::open(DATABASE, Xapian::DB_CREATE_OR_OVERWRITE);
    char *rnd = (char *) malloc(255);
    memset(rnd, 0, 255);
    long cnt = 0;

    try {
        // endlessly add a 2000 term document to the database
        while (true) {
            Xapian::Document doc;
            int i=0;

            for (; i<1000; i++) {
                doc.add_posting("random", i);

                sprintf(rnd, "%d", rand());
                doc.add_posting(rnd, i);
            }

            db.add_document(doc);

            if (++cnt % 10 == 0) {
                // flush every 10 documents
                db.flush();
                msg("Writer:  flush()");
            }
        }
    } catch (Xapian::Error &err) {
        msg((string("Writer:  ERROR=") + err.get_msg()).c_str());
    }
    msg("Writer thread unexpectedly exited...");
    pthread_exit(NULL);
}

void* do_reads(void *) {
    msg("Reader: started");
    try {
        // create the reader database once
        // and continue to query it.
        Xapian::Database db = Xapian::Auto::open(DATABASE);

        while (true) {
            // NOTE:  Creating the database here causes the test
            // to fail too, but in different and mysterious ways
            Xapian::Enquire enq(db);
            Xapian::Query query("random");
            Xapian::MSet set;

            enq.set_query(query);
            set = enq.get_mset(0, 2500);

            // walk the results to exercize the mset iterator
            for (Xapian::MSetIterator i = set.begin(); i != set.end(); ++i);
        }
    } catch (Xapian::Error &err) {
        msg((string("Reader:  ERROR=") + err.get_msg()).c_str());
    }

    pthread_exit(NULL);
}

void setup_writer_thread() {
    pthread_create(&_writer, NULL, do_writes, NULL);
}

void setup_reader_threads() {
    int i = 0;

    for (; i<MAX_THREADS; i++) {
        int success = pthread_create(&(_readers[i]), NULL, do_reads, NULL) == 0;

        if (!success) {
            msg("setup_reader_threads:  Could not create thread");
        }
    }
}

int main(int argv, char **argc) {
    int i = 0;

    pthread_mutex_init(&output_mutex, NULL);

    setup_writer_thread();
    // let the writer thread get going before 
    // we start a bunch of readers
    sleep(2);   


    setup_reader_threads();
    msg("Readers started");

    // sit and wait for everything to finish
    // in a perfect world, they'll never finish
    pthread_join(_writer, NULL);	
    for (; i<MAX_THREADS; i++) {
        pthread_join(_readers[i], NULL);
    }

    pthread_mutex_destroy(&output_mutex);
    return 0;
}


--------------030706000907010606080908--