C, Creating A Dynamic Socket Thread Pool

Question

Yello,

I'm writing C server side code that designed to accept connections from remote units, receive data from them and post them into a database, these units can connect at the same time but also don't connect permanently, since they are battery powered.

After having looked about I've elected to go with a dynamic pool of pthreads that can be marked as in-use, then reused when the unit is done with them. The pool itself will also then increase the number of available threads if there is a larger number of sockets at the same time.

Basically I've done the code I think will work, however not having worked with dynamic memory before, and knowing how badly things can go wrong if you don't handle it correctly, I'm looking to have my code checked to make sure what I have will work and won't just fall over or shoot me in the foot further down the line. So any advice on if this is a good approach or if there is something wrong with would be greatly appreciated, thanks :)

Note the code section below is only for the dynamic pool. Sockets, threads and mutexs I've worked with before:

#include <pthread.h>
#include <stdbool.h>
#include <stdlib.h>

#define DEFAULT_SIZE    8
#define SIZE_INCREMENTS 8

// Structure used to hold the dynamic array
typedef struct
{
    pthread_t worker;
    bool used;
} Element;

typedef struct
{
    Element *data;
    size_t size;
} DynamicPool;

// Global function prototypes
int initDynamicPool (DynamicPool *temp);
pthread_t getFreeElement (DynamicPool *temp);
int freeElement (DynamicPool *temp, pthread_t element);
int freePool (DynamicPool *temp);

// Local function prototypes
int resizePool (DynamicPool *temp);

// Create a new dynamic array
int initDynamicPool (DynamicPool *temp)
{
    if (temp->size == 0)
    {
        temp->size = DEFAULT_SIZE;
        temp->data = (Element*) malloc (temp->size * sizeof (Element));

        // Assigns defaults to new elements
        int t;
        for (t = 0; t < temp->size; t++)
        {
            //temp->data[t].worker = NULL;
            temp->data[t].used = false;
        }

        return temp->size;
    }
    else
    {
        return -1;
    }
}

// Get first free element
pthread_t getFreeElement (DynamicPool *temp)
{
    if (temp->size > 0)
    {
        // Search the array for any element that isn't used
        int t;
        for (t = 0; t < temp->size; t++)
        {
            if (!temp->data[t].used)
            {
                temp->data[t].used = true;
                return temp->data[t].worker;
            }
        }

        // If no worker found, increase the size and send the first new element
        resizePool(temp);
        temp->data[t].used = true;
        return temp->data[t].worker;
    }
    else
    {
        return -1;
    }
}

// Resize array
int resizePool (DynamicPool *temp)
{
    if (temp->size > 0)
    {
        int old_size = temp->size;
        temp->size += SIZE_INCREMENTS;
        temp->data = (Element*) realloc (temp->data, temp->size * sizeof (Element));

        // Assigns defaults to new elements
        int t;
        for (t = old_size; t < temp->size; t++)
        {
            //temp->data[t].worker = NULL;
            temp->data[t].used = false;
        }

        return temp->size;
    }
    else
    {
        return -1;
    }
}

// Free element
int freeElement (DynamicPool *temp, pthread_t element)
{
    if (temp->size > 0)
    {
        // Search the pool for the selected element
        int t;
        for (t = 0; t < temp->size; t++)
        {
            // If we find the element, null the thread and make it as unused
            if (temp->data[t].worker == element)
            {
                //temp->data[t].worker = NULL;
                temp->data[t].used = false;
                break;
            }
        }

        // If we can't find the element error out
        if (t >= temp->size)
        {
            return -2;
        }
        else
        {
            return t;
        }
    }
    else
    {
        return -1;
    }
}

// Free the pool
int freePool (DynamicPool *temp)
{
    if (temp->size > 0)
    {
        // Free the memory then reset all values incase re-initializion is intended
        free (temp->data);
        temp->size = 0;
        temp->data = NULL;

        return temp->size;
    }
    else
    {
        return -1;
    }
}

Answer 1

Your code is OK, but you have to ensure that all calls to DynamicPool are synchronized. I assume you call that API from different threads. Also the counters and other fields that can be modified by multiple threads must be volatile.

typedef struct
{
    volatile Element *data;/* Force compiler always load/store the value */
    volatile size_t size;  /* Force compiler always load/store the value */
    pthread_mutex_t mutex; /* Protect size  and data */
} DynamicPool;

int initDynamicPool (DynamicPool *temp)
{
   ...
   pthread_mutex_init(&temp->mutex);
   ...
}

pthread_t getFreeElement (DynamicPool *temp)
{
    pthread_mutex_lock(&temp->mutex);  /* */
    ...
    pthread_mutex_unlock(&temp->mutex); /* always unlock before return */
    return res;  /* return value must be computed before unlocking */
}

And so on...