High Fibre Programming

Like most websites, glogster.com periodically sends out an e-newsletter to its user-base.    This is done by a simple PHP script that opens a socket to a mail server and goes through the user database, sending an email to each one.    This approach has worked fine until recently, the number of users has gone up dramatically and now it takes an infeasibly long time to send emails to all our users.

After optimising the script as much as possible, the only alternative option was to run many instances of the script simultaneously.   PHP doesn’t support multithreading, and although PHP does support POSIX style process control ( http://www.php.net/manual/en/book.pcntl.php ) I decided to create a wrapper in C that would handle multiple instances of PHP.   (I haven’t found anything technically wrong with PHP’s process control, I just needed something stable, reliable and potentially able to mulitask other scripts such as Python).

How it works

The wrapper I made (see below for code listing) is fairly simple and works by creating a series of threads, each of which forks and creates a PHP instance (replacing the child process) that runs the emailing script.   The thread from the parent process waits for the PHP process to finish and then ends itself.   Once a thread ends, the main thread creates another one, thereby maintaining a constant number of threads.   Signal handling is handled by a separate thread and can either tell the main thread to stop creating new threads and thereby safely shutdown the program or directly terminate all child processes and thus abruptly end the program.

Almost done

So now I had a way of running any number of PHP processes simultaneously and was almost ready to send emails at lightning speed, however one problem remained.   Once an email has been sent to a user, the database is updated so that I know which user has received which email and thus prevent me from sending the same user the same email more than once.   The problem is; what if one instance of the PHP script reads a user, and before it has time to send them the email and update the database, another PHP instance reads the same user and also sends them the email?   One way to solve this would be through locking the database table before reading and updating each row, however this would slow the database down which is not ideal.

The solution I came up with was to limit the number of simultaneous PHP instances to 8 and have the wrapper send each PHP process a unique instance ID from 0 to 7.   When selecting users from the database, the script looked at the 3 least significant bits (LSBs) of the IDs in the recipients table and selected only those rows whose 3 LSBs equated to the instance ID passed from the wrapper.

SELECT *,
(
	if( (`id` | 1) = `id`, 1, 0) +
	if( (`id` | 2) = `id`, 2, 0) +
	if( (`id` | 4) = `id`, 4, 0)
) AS `IID`
FROM `recipients`
WHERE `sent` = 0
HAVING `IID` = :IID
LIMIT :OFFSET, :BATCH_SIZE;

Just bind the instance ID passed from the wrapper into the query and each instance will now select recipients that are unique to that ID – no two concurrent instances can select the same recipients. Just don’t forget to update each row once each email has been sent.

Finishing touches

We’re almost there, just one more little feature is required and the system will be perfect (sort of). If a thread process has nothing to do then it will return immediately, so we end up with a situation when thread processes are constantly created and destroyed which is not the most efficient use of resources. The solution was to check the exit status of the PHP process. If zero then all is ok, the thread ends and is available to be restarted immediately as normal. If however, the exit status is non-zero then the thread is marked as “sleeping” and cannot be restarted by the main thread for another 30 seconds.

If the PHP script doesn’t find any rows in the database to process then it returns 250 (a non-zero, non-reserved exit status) and so the calling thread in the wrapper sleeps for 30 seconds before trying again to see if there are any new items in the database to process.

This has the added advantage that if anything goes wrong in the PHP script, such as a fatal error, then the thread will sleep and you won’t end up with perpetuate thread cycling.

Code listing

Here is the code for the wrapper. The code should be fairly straightforward, with the above notes and the inline comments you should be able to figure out what’s going on – just don’t forget to compile with the pthread library. Comments are of course very welcome!

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <errno.h>
#include <sys/wait.h>

#define NUM_THREADS 8
#define THREAD_STACK_HEADROOM 1048576

pthread_mutex_t mutexSignal;
int slots[NUM_THREADS];
int handledSignal = -1;

void* signalHandler(void* arg);
void waitForThreads();
void killall();
void *task(void *i);

/**
 * Each thread will do this
 *
 */
void *task(void *i)
{
    sigset_t signalSet;
    int iid;
    char *tid;
    int stat_loc;

    iid = (int)i;

    /* Say hello and show the thread number */
    printf("Thread %d: starting\n", iid);

    /* Spawn a child to run the program. */
    pid_t pid=fork();

    switch (pid)
    {
        case 0:
            setsid();
            sigfillset(&signalSet);
            pthread_sigmask(SIG_UNBLOCK, &signalSet, NULL );

            tid = malloc(11*sizeof(char));

            sprintf(tid, "tid=%d", iid);

            char *argv[]={"php", "./sleep.php", tid, NULL};

            execv("/usr/bin/php",argv);
            free(tid);
            exit(EXIT_FAILURE); /* only if execv fails */
        case -1:
            printf("Thread %d: Fork failed\n", iid);
            slots[iid] = 0;
            break;
        default:
            /*  parent process */
            slots[iid] = pid;

            if (pid == waitpid(pid,&stat_loc,WUNTRACED)) /* wait for child to exit */
            {
                printf("Thread %d: Child finished with exit code: %d", iid, WEXITSTATUS(stat_loc));

                if (WEXITSTATUS(stat_loc) != 0)
                {
                    /* Sleep this thread, set available time to now+30s */
                    printf(" Going to sleep\n");
                    slots[iid] = -1 * (time(0) + 30);
                }
                else
                {
                    /* Free this thread slot */
                    printf(" Returning to pool\n");
                    slots[iid] = 0;
                }
            }
            else
            {
                printf("Thread %d: Bad exit\n", iid);
                slots[iid] = 0;
            }
    }

    printf("Thread %d: Finished\n", iid);

    /* Terminate the thread */
    pthread_exit((void*) i);
}

void killall()
{
    int kv,i;

    for (i=0; i<NUM_THREADS;i++)
    {
        kv = kill(slots[i], SIGINT);
        printf("Killed %d: %s\n", slots[i], kv ==0?"ok":"fail");
    }
}

void waitForThreads()
{
    int i, stoppedThreads;

    printf("Waiting for threads to finish\n");

    while(1)
    {
        stoppedThreads = 0;

        for (i=0; i<NUM_THREADS;i++)
        {
            /* Check if thread has finished or is sleeping */
            if (slots[i] == 0 || slots[i] < -1)
            {
                stoppedThreads++;
            }
        }

        if (stoppedThreads == NUM_THREADS)
        {
            break;
        }
    }
}

void* signalHandler( void* arg )
{
    sigset_t signal_set;
    int sig;

    while (1)
    {
        /* Wait for any and all signals */
        sigfillset(&signal_set);

        sigwait(&signal_set, &sig);

        /* When we get this far, we've caught a signal */

        switch(sig)
        {
            /* SIGQUIT */
            case SIGTERM:
            case SIGQUIT:
                pthread_mutex_lock(&mutexSignal);
                handledSignal = SIGQUIT;
                pthread_mutex_unlock(&mutexSignal);
                break;

            /* SIGINT */
            case SIGINT:
                pthread_mutex_lock(&mutexSignal);
                handledSignal = SIGINT;
                pthread_mutex_unlock(&mutexSignal);
                break;

            /* other signals */
            default:
                pthread_mutex_lock(&mutexSignal);
                handledSignal = 0;
                pthread_mutex_unlock(&mutexSignal);
                break;
        }
    }

    return (void*)0;
}

/**
 * Main
 *
 */
int main()
{
    int rc, i;
    int running = 1;
    size_t stacksize;
    sigset_t signalSet;
    pthread_t threadSignalHandler;
    pthread_t thread[NUM_THREADS];
    pthread_attr_t attr;

    /* Initialise the signal mutex lock */
    pthread_mutex_init(&mutexSignal, NULL);

    /* Initialise and set thread attributes */
    pthread_attr_init(&attr);
    //pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    /* Ensure each thread has a given stack size - portability */
    pthread_attr_getstacksize(&attr, &stacksize);
    printf("Default stack size = %li\n", (long) stacksize);

        /* Determine and set a new stack size */
        stacksize = sizeof(sigset_t) + THREAD_STACK_HEADROOM;
        printf("Amount of stack calculated per thread = %li\n", (long) stacksize);
        pthread_attr_setstacksize (&attr, stacksize);

    /* block all signals */
    sigfillset(&signalSet);
    pthread_sigmask(SIG_BLOCK, &signalSet, NULL );

    /* create the signal handling thread */
    pthread_create(&threadSignalHandler, NULL, signalHandler, NULL);

    while (running > 0)
    {
        /* Find an empty slot */
        for (i=0; i<NUM_THREADS;i++)
        {
            if (slots[i] == 0)  /* Thread is currently not doing anything, so let's give it something to do */
            {
                /* Set this slot as used, -1 is a temporary value before being replaced by the PID of the forked process (prevents race hazards) */
                slots[i] = -1;

                /* This slot is spare */
                printf("Main: creating thread %d\n", i);

                /* Create a new thread */
                rc = pthread_create(&thread[i], &attr, task, (void *)i);

                if (rc)
                {
                    printf("ERROR: return code from pthread_create() is %d\n", rc);
                    exit(1);
                }

                break;
            }
            else if (slots[i] < -1)     /* Thread is sleeping, see if we should wake it up */
            {
                if (time(0) > (-1 * slots[i]) )
                {
                    /* Thread has slept long enough so let's wake it up */
                    slots[i] = 0;
                }
            }
        }

        pthread_mutex_lock(&mutexSignal);

        switch ( handledSignal )
        {
            case -1:
                break;

            case 0:
                /* The case for signals we're not interested in */
                handledSignal = -1;
                break;

            case SIGTERM:
            case SIGQUIT:
                printf("Main: SIGQUIT\n");
                handledSignal = -1;
                running = -1;
                killall();
                break;

            case SIGINT:
                printf("Main: SIGINT\n" );
                handledSignal = -1;
                running = -1;
                break;
        }

        pthread_mutex_unlock(&mutexSignal);

        /* Sleep for a bit - all this thread creation is hard work! */
        usleep(100000);
    }

    waitForThreads();

    printf("Bye\n");
}

And here’s a sample PHP script (but you can of course have the wrapper call anything).

<?php

list($junk, $tid) = explode("=", $argv[1], 2);

printf("PHP: Hello from thread %d\n", $tid);

sleep(mt_rand(4,14));

// Non-zero exit status if the calling thread should sleep
// Chosen 250 because it's not a reserved status
// NOTE: PHP returns 255 on fatal error
exit(250);

Exercises left to the reader

The above code is not really intended as production-ready, (although I have actually used it in production) and there are still plenty of loose ends that could do with tidying up. One thing you might want to do is detach the child PHP processes from the parent process’ output sockets, thereby effectively sending them to the background. For inspiration take a look at this basic daemonising program: A simple daemon in C.

Another nice addition might be to allow the child process to be specified in command line arguments to the wrapper. It would be useful to specify the sleep time for threads and also the maximum number of concurrent threads – this way you wouldn’t have to recompile each time you changed something.

If you find this useful or interesting then comments would be greatly welcomed!

Update: I have created an updated article which describes a multithreaded approach in Java Spinning command line cursor in Java

I think I must be really bored this morning, I can’t believe I’m actually blogging this, but it might be useful for someone, who knows.   Anyway, I am currently writing a program that sits and does stuff for a very long time, and I need a way to nicely indicate the program is still running and doing its stuff.   So I have created a little method that prints a spinning cursor on the command line.   The implementation would of course have to be multithreaded; one thread to do stuff, the other to spin the cursor – if there’s time I’ll update it to a more complete solution, but for now here’s the spin implementation:

public class Spin
{
    public static void main(String[] args) throws InterruptedException
    {
        String[] phases = {"|", "/", "-", "\\"};

        System.out.printf("Spinning... |");

        while (true)
        {
            for (String phase : phases)
            {
                System.out.printf("\b"+phase);
                Thread.sleep(100);
            }
        }
    }
}

And here it is in PHP.

<?php

    $phases = array("|", "/", "-", "\\");

    printf("Spinning... |");

    while (1)
    {
        foreach ($phases AS $phase)
        {
            printf('%s%s', chr(8), $phase);
            usleep(100000); // Replace this with one iteration of doing stuff
        }
    }

Of course PHP doesn’t support threading so you’d have to call each iteration of the “doing stuff” loop inside the spin loop which would mean you’d get a bit of a jumpy spinning cursor but I think most people could live with that.

A daemon is a process which runs in the background of your computer, periodically carrying out a specific task. The following is an example of a simple daemon written in C. It works by forking to create a child process, the parent then terminates but the child carries on in the background – entering a continuous loop of doing a task and then sleeping. The child process is of course an identical copy of the parent so care must be taken to close all file descriptors, thus detaching the child completely from the calling process. The deamonised process is controlled by sending it signals which it can catch and take action accordingly. In the example below, the process is terminated by sending SIGINT or SIGTERM, but you can of course add in your own handling – for example to re-read config data on SIGHUP.

#include <stdio.h>
#include <signal.h>
#include <syslog.h>
#include <errno.h>
#include <stdlib.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>

#define DAEMON_NAME "simpledaemon"

    void daemonShutdown();
    void signal_handler(int sig);
    void daemonize(char *rundir, char *pidfile);

    int pidFilehandle;

    void signal_handler(int sig)
    {
        switch(sig)
        {
            case SIGHUP:
                syslog(LOG_WARNING, "Received SIGHUP signal.");
                break;
            case SIGINT:
            case SIGTERM:
                syslog(LOG_INFO, "Daemon exiting");
                daemonShutdown();
                exit(EXIT_SUCCESS);
                break;
            default:
                syslog(LOG_WARNING, "Unhandled signal %s", strsignal(sig));
                break;
        }
    }

    void daemonShutdown()
    {
        close(pidFilehandle);
    }

    void daemonize(char *rundir, char *pidfile)
    {
        int pid, sid, i;
        char str[10];
        struct sigaction newSigAction;
        sigset_t newSigSet;

        /* Check if parent process id is set */
        if (getppid() == 1)
        {
            /* PPID exists, therefore we are already a daemon */
            return;
        }

        /* Set signal mask - signals we want to block */
        sigemptyset(&newSigSet);
        sigaddset(&newSigSet, SIGCHLD);  /* ignore child - i.e. we don't need to wait for it */
        sigaddset(&newSigSet, SIGTSTP);  /* ignore Tty stop signals */
        sigaddset(&newSigSet, SIGTTOU);  /* ignore Tty background writes */
        sigaddset(&newSigSet, SIGTTIN);  /* ignore Tty background reads */
        sigprocmask(SIG_BLOCK, &newSigSet, NULL);   /* Block the above specified signals */

        /* Set up a signal handler */
        newSigAction.sa_handler = signal_handler;
        sigemptyset(&newSigAction.sa_mask);
        newSigAction.sa_flags = 0;

            /* Signals to handle */
            sigaction(SIGHUP, &newSigAction, NULL);     /* catch hangup signal */
            sigaction(SIGTERM, &newSigAction, NULL);    /* catch term signal */
            sigaction(SIGINT, &newSigAction, NULL);     /* catch interrupt signal */

        /* Fork*/
        pid = fork();

        if (pid < 0)
        {
            /* Could not fork */
            exit(EXIT_FAILURE);
        }

        if (pid > 0)
        {
            /* Child created ok, so exit parent process */
            printf("Child process created: %d\n", pid);
            exit(EXIT_SUCCESS);
        }

        /* Child continues */

        umask(027); /* Set file permissions 750 */

        /* Get a new process group */
        sid = setsid();

        if (sid < 0)
        {
            exit(EXIT_FAILURE);
        }

        /* close all descriptors */
        for (i = getdtablesize(); i >= 0; --i)
        {
            close(i);
        }

        /* Route I/O connections */
        close(STDIN_FILENO);
        close(STDOUT_FILENO);
        close(STDERR_FILENO);

        chdir(rundir); /* change running directory */

        /* Ensure only one copy */
        pidFilehandle = open(pidfile, O_RDWR|O_CREAT, 0600);

        if (pidFilehandle == -1 )
        {
            /* Couldn't open lock file */
            syslog(LOG_INFO, "Could not open PID lock file %s, exiting", pidfile);
            exit(EXIT_FAILURE);
        }

        /* Try to lock file */
        if (lockf(pidFilehandle,F_TLOCK,0) == -1)
        {
            /* Couldn't get lock on lock file */
            syslog(LOG_INFO, "Could not lock PID lock file %s, exiting", pidfile);
            exit(EXIT_FAILURE);
        }

        /* Get and format PID */
        sprintf(str,"%d\n",getpid());

        /* write pid to lockfile */
        write(pidFilehandle, str, strlen(str));
    }

    int main()
    {
        /* Debug logging
        setlogmask(LOG_UPTO(LOG_DEBUG));
        openlog(DAEMON_NAME, LOG_CONS, LOG_USER);
        */

        /* Logging */
        setlogmask(LOG_UPTO(LOG_INFO));
        openlog(DAEMON_NAME, LOG_CONS | LOG_PERROR, LOG_USER);

        syslog(LOG_INFO, "Daemon starting up");

        /* Deamonize */
        daemonize("/tmp/", "/tmp/daemon.pid");

        syslog(LOG_INFO, "Daemon running");

        while (1)
        {
            syslog(LOG_INFO, "daemon says hello");

            sleep(1);
        }
    }

The above example is only simple but it serves as a good starting point to create your own daemon. In my next article I will demonstrate how it can be extended to periodically call a PHP script.

If I find enough time I will also demonstrate how it can be extended from being a single-threaded daemon to a multi-threaded daemon, maintaining a continuous pool of x threads, each doing a given task.

The other day a Ruby developer was extolling the virtues of Ruby on Rails to me, in particular the use of “acts_as_” to make a class magically include all the behaviour from another. It isn’t inheritance, the relationship isn’t “is a” but rather “has a” – it’s a sort of transparent aggregation with all the methods and properties of the aggregated object magically available through the containing object. At least that’s what it looks like, from my somewhat limited RoR experience.

Personally, I would rather achieve this functionality by using the Strategy Pattern – it requires more boiler-plate code but I think the end result is much clearer.

Anyway, not to be outdone, I set about implementing RoR’s acts_as in PHP.   My implementation works by using a class called Aggregatable which allows all child classes to aggregate, or “act_as”, other standard classes.   I’ll start with an example with two simple classes; Binman and Postman which each do different jobs.

Here are the classes:

class Binman
{
    public function collectRubbish()
    {
        printf("Collecting rubbish\n");
    }
}

class Postman
{
    public function deliverLetter()
    {
        printf("Delivering a letter\n");
    }
}

We can use the classes like this:

$binman = new Binman();
$binman->collectRubbish();    // Prints "Collecting rubbish"

$postman = new Postman();
$postman->deliverLetter();    // Prints "Delivering a letter"

So far so good, but what if I wanted a CasualWorker class to be able to both collect rubbish and deliver letters?   I could of course copy the methods from the Binman class and the Postman class into the CasualWorker class, but this would be rather inflexible – suppose if we needed to change the Binman behaviour we would have to update both the Binman and CasualWorker classes.   Instead we make the CasualWorker class a child of the Aggregatable class and aggregate the Binman and Postman classes.   That might sound a bit complex so I’ll illustrate with an example.

Here’s the CasualWorker class:

class CasualWorker extends Aggregatable
{
    public function __construct()
    {
        $this->aggregate("Binman");
        $this->aggregate("Postman");
    }
}

The CasualWorker class can now do everything that the Binman and Postman classes can:

$casualWorker = new CasualWorker();
$casualWorker->collectRubbish();    // Prints "Collecting rubbish"
$casualWorker->deliverLetter();    // Prints "Delivering a letter"

Great!   It works!   But how?

The Aggregatable class contains an array of instances of all the classes it has been asked to aggregate.   It also makes use of PHP’s magic method, __call(), so that when client code invokes a method that is not defined by it or the child class it is extending, it loops through the objects it is aggregating to see if any of those define the method.   If an object is found that defines the requested method then it is run and the result returned – just as if the invoked class defined the method.

class Aggregatable
{
    /**
     * Store of aggregated objects
     *
     */
    private $aggregated = array();

    /**
     * Aggregates objects
     *
     * @param string Classname
     */
    protected function aggregate($class)
    {
        // Check an instance of this class has not already been aggregated
        if (array_key_exists($class, $this->aggregated))
        {
            throw new Exception(sprintf("Class already aggregated: %s", $class));
        }

        // Add a new instance of the class to the store
        $this->aggregated[$class] = new $class();
    }

    /**
     * Magic method - catch calls to undefined methods
     *
     * @param String method name
     * @param array Arguments
     */
    public function __call($method, $arguments)
    {
        // Loop through the aggregated objects
        foreach ($this->aggregated as $subject)
        {
            // Check each object to see if it defines the method
            if (method_exists($subject, (string) $method))
            {
                // Object defines the requested method, so call it and return the result
                return call_user_func_array(array($subject, (string) $method), $arguments);
            }
        }

        throw new Exception(sprintf("Method not found: %s", $method));
    }
}

What about properties?

I’ve kept the example above fairly simple but there’s no reason why it couldn’t be extended to cover properties using the other Magic Functions __set(), __get() and __unset().   In fact, in the example in the download (see below) I’ve implemented basic support for properties.

Conclusion

Like I’ve said before, I’m not a big fan of this sort of “magic”.   It seems to me to break the golden rule of object oriented programmed, as mentioned in the Gang Of Four book; “program to an interface, not an implementation”.   The interface the class provides is defined by the public properties and methods available, or even better, from the Interfaces it implements.   By adding this sort of run-time dynamism, the client code can never be sure what the class can do, instanceof won’t help so you’re left with either making extra functions like bool isAggregating(classname) or reflection.   Admittedly magic is nice, and even if a isAggregating() method were to solve all the problems, I would strongly advise to stay away from such temptations.

To quote Matt Zandstra,

Magic is nice but clarity is much nicer.

Downloads

aggregation.zip
Here’s a zip file containing all the classes mentioned above plus support for properties.   It’s perhaps worth mentioning that this really isn’t production-ready, should you ever really want to use such a thing it will need a few tweaks first – namely how to handle adding two classes which define methods of the same name – which one has priority?   No doubt you’ll also want some inspection methods such as isAggregating() as previously mentioned.