Passing MySQL fetched rows to thread pool in C

Googlebot 0
Viewed 78

I want to process data fetched from the MySQL database simultaneously. I pass the data to each thread process (no need for thread-safe consideration; rows are processed independently in each thread):

#include <mysql.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <pthread.h>
#include "thpool.h" // https://github.com/Pithikos/C-Thread-Pool

#define THREADS 10

struct fparam
{
  int id;
  char *data;
};

void process(void *arg)
{
  struct fparam *args = arg;
  // Processing ID and Data here
  printf("%d - %s\n", args->id, args->data);
}

int main(int argc, char **argv)
{
  threadpool thpool = thpool_init(THREADS);

  // MySQL connection

  MYSQL_RES *result = mysql_store_result(con);

  int num_fields = mysql_num_fields(result);
  struct fparam items[100]; // 100 is for the representation

  MYSQL_ROW row;
  int i = 0;
  while ((row = mysql_fetch_row(result)))
  {
    items[i].id = atoi(row[0]);
    items[i].data = row[1];
    thpool_add_work(thpool, process, (void *)(&items[i]));
    i++;
  }

  mysql_free_result(result);
  mysql_close(con);

  thpool_wait(thpool);
  thpool_destroy(thpool);

  exit(0);
}

When there are many rows, items gets too big to fit in memory (not just heap).

How can I limit the number of rows stored in memory and delete them when they have been processed?

I think a key issue that we do not know if process function is faster or fetching the rows from the database.

c
2 Answers

Use a queue, a list where you add items at one end and take them out of the other.

You can write your own; a linked list can be used as a queue adding items to one end and removing them from the other. Or use an existing implementation such as the one provided by GLib.

Schwern0

You do not need to create new queue in your scenario as thpool_init(THREADS) already provides you one and thpool_add_work feeds this internal queue and it grows on fetching. If fetching from database is fast but processing is slow you have to limit fetching new rows to reasonable rate so they will fit into memory. Looking into documentation of "thpool.h" there is this function thpool_num_threads_working(threadpool). It will return number of working threads so as you defined THREADS in simplest form you would want to fetch new row whenever there is at least one idle thread available (Something like while(thpool_num_threads_working(thpool) < THREADS)).

Taking into account performance reasons you should think about prefetching some rows to have your data already there when any thread will finish its job to be able to feed it without waiting. How many of these rows can be there waiting on processing it depends on memory available. It is even more important to consider this when fetching can be quite time consuming but processing is fast.

The way you use items[i] also creates an issue here because i can go to infinity overflowing items[100] array. If we put there very strong assumption that threads are finishing their jobs in the same order as started you could just reset i in such way that items specific indexes would be reused for new rows (as kind of circular buffer). Unfortunately right now I am afraid C-Thread-Pool is not supporting identification which specific thread has finished its job (and which corresponding data are no longer needed). If You need to be 100% safe here i would consider two solutions possible. Extending C-Thread-Pool with thread status vs its job id verification or processing rows in batches in such way you feed all threads (one job per thread to queue) and then wait till all of them will finish their job to feed them all together again with next batch of rows to process.

And remember to check for errors when using thpool_add_work and thpool_init.

WojciechS0
Related